ref(metrics): Remodel metric submission flow (#3867)

Remove all the load of processing metrics from the project cache and
move it to the processor.

We slowly moved more and more logic into the project cache, which over
time became quite expensive to execute. Especially for heavily batched
metrics. To alleviate this problem, the project cache only enriches
messages with project information and dispatches based on this
information instead of also dealing with filtering/validating metrics
and rate limiting metrics.
Validation and rate limits are now enforced in the Processor.

The old flow for Metrics:
```mermaid
sequenceDiagram
    Endpoint->>Processor: ProcessMetrics
    Processor->>Project Cache: AddMetricBuckets
    activate Project Cache
    Note right of Project Cache: [check_buckets]
    Project Cache->>Aggregator: MergeBuckets
    deactivate Project Cache
    Aggregator->>Project Cache: FlushBuckets
    activate Project Cache
    Note right of Project Cache: [check_buckets]
    Project Cache->>Processor: EncodeMetrics
    deactivate Project Cache
    Processor->>Store: SendRequest
```

The new flow:
```mermaid

sequenceDiagram
    participant Endpoint
    participant Processor
    participant Project Cache
    participant Aggregator
    participant Store
    Endpoint->>Project Cache: ProcessMetrics
    Project Cache->>Processor: ProcessProjectMetrics
    activate Processor
    Note right of Processor: [check_buckets]
    Processor->>Aggregator: MergeBuckets
    deactivate Processor
    Aggregator->>Project Cache: FlushBuckets
    Project Cache->>Processor: EncodeMetrics
    activate Processor
    Note right of Processor: [check_buckets]
    Processor->>Store: SendRequest
    deactivate Processor
```

Batched metrics just have a slightly adjusted flow, they go from the
endpoint to the processor for parsing, then move forward with the same
`ProcessMetrics` flow.

relay-server/src/endpoints/common.rs

@@ -11,8 +11,8 @@ use serde::Deserialize;
 use crate::envelope::{AttachmentType, Envelope, EnvelopeError, Item, ItemType, Items};
 use crate::service::ServiceState;
 use crate::services::outcome::{DiscardReason, Outcome};
-use crate::services::processor::{ProcessMetricMeta, ProcessMetrics, ProcessingGroup};
-use crate::services::project_cache::{CheckEnvelope, ValidateEnvelope};
+use crate::services::processor::{MetricData, ProcessMetricMeta, ProcessingGroup};
+use crate::services::project_cache::{CheckEnvelope, ProcessMetrics, ValidateEnvelope};
 use crate::statsd::{RelayCounters, RelayHistograms};
 use crate::utils::{self, ApiErrorResponse, FormDataIter, ManagedEnvelope, MultipartError};
 
@@ -273,9 +273,9 @@ fn queue_envelope(
 
  if !metric_items.is_empty() {
  relay_log::trace!("sending metrics into processing queue");
- state.processor().send(ProcessMetrics {
- items: metric_items.into_vec(),
- start_time: envelope.meta().start_time(),
+ state.project_cache().send(ProcessMetrics {
+ data: MetricData::Raw(metric_items.into_vec()),
+ start_time: envelope.meta().start_time().into(),
  sent_at: envelope.sent_at(),
  project_key: envelope.meta().public_key(),
  source: envelope.meta().into(),

relay-server/src/service.rs

@@ -242,6 +242,7 @@ impl ServiceState {
  test_store: test_store.clone(),
  #[cfg(feature = "processing")]
  store_forwarder: store.clone(),
+ aggregator: aggregator.clone(),
  },
  metric_outcomes.clone(),
  )
@@ -263,7 +264,6 @@ impl ServiceState {
  MemoryChecker::new(memory_stat.clone(), config.clone()),
  envelope_buffer.clone(),
  project_cache_services,
- metric_outcomes,
  redis_pool.clone(),
  )
  .spawn_handler(project_cache_rx);