ref(metrics): Remodel metric submission flow

getsentry · Jul 25, 2024 · 5445632 · 5445632
1 parent 1523351
commit 5445632
Show file tree

Hide file tree

Showing 9 changed files with 309 additions and 284 deletions.
diff --git a/relay-server/src/endpoints/common.rs b/relay-server/src/endpoints/common.rs
@@ -11,8 +11,10 @@ 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, ProcessProjectMetrics, ProcessingGroup,
+};
+use crate::services::project_cache::{CheckEnvelope, ProcessMetrics, ValidateEnvelope};
 use crate::statsd::{RelayCounters, RelayHistograms};
 use crate::utils::{self, ApiErrorResponse, FormDataIter, ManagedEnvelope, MultipartError};
 
@@ -273,9 +275,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(),

diff --git a/relay-server/src/service.rs b/relay-server/src/service.rs
@@ -240,6 +240,7 @@ impl ServiceState {
  test_store: test_store.clone(),
  #[cfg(feature = "processing")]
  store_forwarder: store.clone(),
+ aggregator: aggregator.clone(),
  },
  metric_outcomes.clone(),
  )