From e0c4b489346cc318a2f33e242b23ff7ec54c7d5d Mon Sep 17 00:00:00 2001 From: David Herberth Date: Thu, 18 Jun 2026 14:54:34 +0200 Subject: [PATCH 1/2] ref(outcomes): Remove outcomes aggregator, implement forwarding of metrics as client reports --- CHANGELOG.md | 1 + relay-config/src/config.rs | 27 -- relay-server/src/service.rs | 46 +-- .../src/services/outcome/aggregator.rs | 173 --------- relay-server/src/services/outcome/metric.rs | 215 ++++++++++- relay-server/src/services/outcome/mod.rs | 2 - relay-server/src/services/outcome/service.rs | 351 ++++++++---------- relay-server/src/services/processor.rs | 41 +- relay-server/src/statsd.rs | 3 - relay-server/src/utils/sleep_handle.rs | 21 +- tests/integration/test_metrics.py | 25 +- 11 files changed, 469 insertions(+), 436 deletions(-) delete mode 100644 relay-server/src/services/outcome/aggregator.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e34f8f0959a..857b222b446 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ **Internal**: - Require timestamps and verification in auth signatures. ([#6069](https://github.com/getsentry/relay/pull/6069)) +- Internally handle outcomes as metrics. ([#6107](https://github.com/getsentry/relay/pull/6107)) - Have relay generate metric billing outcomes. ([#6066](https://github.com/getsentry/relay/pull/6066)) - Update sentry-conventions to 0.12.0. diff --git a/relay-config/src/config.rs b/relay-config/src/config.rs index 9bf7b62dafa..4ae02c86468 100644 --- a/relay-config/src/config.rs +++ b/relay-config/src/config.rs @@ -1233,25 +1233,6 @@ pub enum NormalizationLevel { Full, } -/// Configuration values for the outcome aggregator -#[derive(Serialize, Deserialize, Debug)] -#[serde(default)] -pub struct OutcomeAggregatorConfig { - /// Defines the width of the buckets into which outcomes are aggregated, in seconds. - pub bucket_interval: u64, - /// Defines how often all buckets are flushed, in seconds. - pub flush_interval: u64, -} - -impl Default for OutcomeAggregatorConfig { - fn default() -> Self { - Self { - bucket_interval: 60, - flush_interval: 120, - } - } -} - /// Configuration options for objectstore's auth scheme. #[derive(Serialize, Deserialize)] pub struct ObjectstoreAuthConfig { @@ -1418,8 +1399,6 @@ pub struct Outcomes { /// Defines the source string registered in the outcomes originating from /// this Relay (typically something like the region or the layer). pub source: Option, - /// Configures the outcome aggregator. - pub aggregator: OutcomeAggregatorConfig, } impl Default for Outcomes { @@ -1429,7 +1408,6 @@ impl Default for Outcomes { batch_size: 1000, batch_interval: 500, source: None, - aggregator: OutcomeAggregatorConfig::default(), } } } @@ -2163,11 +2141,6 @@ impl Config { self.values.outcomes.source.as_deref() } - /// Returns the width of the buckets into which outcomes are aggregated, in seconds. - pub fn outcome_aggregator(&self) -> &OutcomeAggregatorConfig { - &self.values.outcomes.aggregator - } - /// Returns logging configuration. pub fn logging(&self) -> &relay_log::LogConfig { &self.values.logging diff --git a/relay-server/src/service.rs b/relay-server/src/service.rs index 0261d17aa1b..b78261e5e6a 100644 --- a/relay-server/src/service.rs +++ b/relay-server/src/service.rs @@ -18,7 +18,8 @@ use crate::services::objectstore::Objectstore; #[cfg(feature = "processing")] use crate::services::objectstore::ObjectstoreService; use crate::services::outcome::{ - OutcomeAggregator, OutcomeProducer, OutcomeProducerService, TrackOutcome, + ClientReportOutcomeProducerService, NullOutcomeProducerService, OutcomeProducerService, + TrackOutcome, }; use crate::services::processor::{ self, EnvelopeProcessor, EnvelopeProcessorService, EnvelopeProcessorServicePool, @@ -39,7 +40,7 @@ use anyhow::Result; use axum::extract::FromRequestParts; use axum::http::request::Parts; use relay_cogs::Cogs; -use relay_config::Config; +use relay_config::{Config, EmitOutcomes, RelayMode}; #[cfg(feature = "processing")] use relay_config::{RedisConfigRef, RedisConfigsRef}; #[cfg(feature = "processing")] @@ -73,7 +74,6 @@ pub enum ServiceError { #[derive(Clone, Debug)] pub struct Registry { pub health_check: Addr, - pub outcome_producer: Addr, pub outcome_aggregator: Addr, pub processor: Addr, pub relay_cache: Addr, @@ -195,19 +195,25 @@ impl ServiceState { // Create an address for the `EnvelopeProcessor`, which can be injected into the // other services. let (processor, processor_rx) = match config.relay_mode() { - relay_config::RelayMode::Proxy => channel(ProxyProcessorService::name()), - relay_config::RelayMode::Managed => channel(EnvelopeProcessorService::name()), + RelayMode::Proxy => channel(ProxyProcessorService::name()), + RelayMode::Managed => channel(EnvelopeProcessorService::name()), }; let (aggregator, aggregator_rx) = channel(RouterService::name()); - let outcome_producer = services.start(OutcomeProducerService::create( - config.clone(), - processor.clone(), - aggregator.clone(), - )?); - let outcome_aggregator = - services.start(OutcomeAggregator::new(&config, outcome_producer.clone())); + let outcome_aggregator = match config.emit_outcomes() { + EmitOutcomes::None => services.start(NullOutcomeProducerService::new()), + _ => match config.relay_mode() { + RelayMode::Proxy => services.start(ClientReportOutcomeProducerService::new( + &config, + processor.clone(), + )), + RelayMode::Managed => services.start(OutcomeProducerService::new( + Arc::clone(&config), + aggregator.clone(), + )), + }, + }; let (global_config, global_config_rx) = GlobalConfigService::new(config.clone(), upstream_relay.clone()); @@ -267,7 +273,7 @@ impl ServiceState { ); let (processor_pool, aggregator_handle, autoscaling) = match config.relay_mode() { - relay_config::RelayMode::Proxy => { + RelayMode::Proxy => { services.start_with( ProxyProcessorService::new( config.clone(), @@ -281,7 +287,7 @@ impl ServiceState { ); (None, None, None) } - relay_config::RelayMode::Managed => { + RelayMode::Managed => { let processor_pool = create_processor_pool(&config)?; let router = RouterService::new( @@ -365,7 +371,6 @@ impl ServiceState { let registry = Registry { processor, health_check, - outcome_producer, outcome_aggregator, relay_cache, global_config, @@ -426,17 +431,12 @@ impl ServiceState { &self.inner.registry.health_check } - /// Returns the address of the [`OutcomeProducer`] service. - pub fn outcome_producer(&self) -> &Addr { - &self.inner.registry.outcome_producer - } - - /// Returns the address of the [`OutcomeProducer`] service. + /// Returns the address of the [`UpstreamRelay`] service. pub fn upstream_relay(&self) -> &Addr { &self.inner.registry.upstream_relay } - /// Returns the address of the [`OutcomeProducer`] service. + /// Returns the address of the [`EnvelopeProcessor`] service. pub fn processor(&self) -> &Addr { &self.inner.registry.processor } @@ -446,7 +446,7 @@ impl ServiceState { &self.inner.registry.global_config } - /// Returns the address of the [`OutcomeProducer`] service. + /// Returns the address of the [`TrackOutcome`] service. pub fn outcome_aggregator(&self) -> &Addr { &self.inner.registry.outcome_aggregator } diff --git a/relay-server/src/services/outcome/aggregator.rs b/relay-server/src/services/outcome/aggregator.rs deleted file mode 100644 index f6be36712e2..00000000000 --- a/relay-server/src/services/outcome/aggregator.rs +++ /dev/null @@ -1,173 +0,0 @@ -//! This module contains the outcomes aggregator, which collects similar outcomes into groups -//! and flushed them periodically. - -use std::collections::HashMap; -use std::time::Duration; - -use chrono::Utc; -use relay_common::time::UnixTimestamp; -use relay_config::{Config, EmitOutcomes}; -use relay_quotas::{DataCategory, Scoping}; -use relay_statsd::metric; -use relay_system::{Addr, Controller, Service, Shutdown}; - -use crate::services::outcome::{Outcome, OutcomeProducer, TrackOutcome}; -use crate::statsd::RelayTimers; -use crate::utils::SleepHandle; - -/// Contains everything to construct a `TrackOutcome`, except quantity -#[derive(Debug, PartialEq, Eq, Hash)] -struct BucketKey { - /// The time slot for which outcomes are aggregated. timestamp = offset * bucket_interval - offset: u64, - /// Scoping of the request. - pub scoping: Scoping, - /// The outcome. - pub outcome: Outcome, - /// The event's data category. - pub category: DataCategory, -} - -/// Aggregates [`Outcome`]s into buckets and flushes them periodically. -/// -/// This service handles a single message [`TrackOutcome`]. -pub struct OutcomeAggregator { - /// Whether or not to produce outcomes. - /// - /// If `true`, all outcomes will be dropped. - disabled: bool, - /// Whether outcomes should be aggregated. - /// - /// If `false` outcomes are not aggregated and just immediately forwarded. - aggregate: bool, - /// The width of each aggregated bucket in seconds - bucket_interval: u64, - /// The number of seconds between flushes of all buckets - flush_interval: u64, - /// Mapping from bucket key to quantity. - buckets: HashMap, - /// The recipient of the aggregated outcomes - outcome_producer: Addr, - /// An optional timeout to the next scheduled flush. - flush_handle: SleepHandle, -} - -impl OutcomeAggregator { - pub fn new(config: &Config, outcome_producer: Addr) -> Self { - let disabled = matches!(config.emit_outcomes(), EmitOutcomes::None); - let aggregate = !matches!(config.emit_outcomes(), EmitOutcomes::AsOutcomes); - - Self { - disabled, - aggregate, - bucket_interval: config.outcome_aggregator().bucket_interval, - flush_interval: config.outcome_aggregator().flush_interval, - buckets: HashMap::new(), - outcome_producer, - flush_handle: SleepHandle::idle(), - } - } - - fn handle_shutdown(&mut self, message: Shutdown) { - if message.timeout.is_some() { - self.flush(); - relay_log::info!("outcome aggregator stopped"); - } - } - - fn handle_track_outcome(&mut self, msg: TrackOutcome) { - if self.disabled { - return; - } - - // This is a temporary measure until the aggregator can be entirely removed in favor of - // using metrics as the primary aggregator and forward mechanism for outcomes. - if !self.aggregate { - self.outcome_producer.send(msg); - return; - } - - let offset = msg.timestamp.timestamp() as u64 / self.bucket_interval; - - let bucket_key = BucketKey { - offset, - scoping: msg.scoping, - outcome: msg.outcome, - category: msg.category, - }; - - let counter = self.buckets.entry(bucket_key).or_insert(0); - *counter += msg.quantity; - - if self.flush_interval == 0 { - // Flush immediately. This is useful for integration tests. - self.do_flush(); - } else if self.flush_handle.is_idle() { - self.flush_handle - .set(Duration::from_secs(self.flush_interval)); - } - } - - fn do_flush(&mut self) { - self.flush_handle.reset(); - - let bucket_interval = self.bucket_interval; - let outcome_producer = self.outcome_producer.clone(); - - for (bucket_key, quantity) in self.buckets.drain() { - let BucketKey { - offset, - scoping, - outcome, - category, - } = bucket_key; - - // In case the timestamp cannot be extracted, we fallback to the current UTC `DateTime` - let timestamp = UnixTimestamp::from_secs(offset * bucket_interval) - .as_datetime() - .unwrap_or_else(Utc::now); - - let outcome = TrackOutcome { - timestamp, - scoping, - outcome, - event_id: None, - remote_addr: None, - category, - quantity, - }; - - outcome_producer.send(outcome); - } - } - - fn flush(&mut self) { - metric!(timer(RelayTimers::OutcomeAggregatorFlushTime), { - self.do_flush(); - }); - } -} - -impl Service for OutcomeAggregator { - type Interface = TrackOutcome; - - async fn run(mut self, mut rx: relay_system::Receiver) { - let mut shutdown = Controller::shutdown_handle(); - relay_log::info!("outcome aggregator started"); - - loop { - tokio::select! { - // Prioritize flush over receiving messages to prevent starving. Shutdown can be - // last since it is not vital if there are still messages in the channel. - biased; - - () = &mut self.flush_handle => self.flush(), - Some(message) = rx.recv() => self.handle_track_outcome(message), - shutdown = shutdown.notified() => self.handle_shutdown(shutdown), - else => break, - } - } - - relay_log::info!("outcome aggregator stopped"); - } -} diff --git a/relay-server/src/services/outcome/metric.rs b/relay-server/src/services/outcome/metric.rs index 1d26cb9758a..7054d22d0c5 100644 --- a/relay-server/src/services/outcome/metric.rs +++ b/relay-server/src/services/outcome/metric.rs @@ -1,8 +1,10 @@ use std::collections::BTreeMap; use std::sync::LazyLock; +use relay_base_schema::data_category::DataCategory; use relay_config::Config; -use relay_metrics::{Bucket, BucketValue, MetricName, UnixTimestamp}; +use relay_event_schema::protocol::{ClientReport, DiscardedEvent}; +use relay_metrics::{Bucket, BucketValue, MetricName, MetricNamespace, UnixTimestamp}; #[cfg(any(test, feature = "processing"))] use crate::services::outcome::OutcomeId; @@ -99,6 +101,98 @@ pub fn to_outcome_id(name: &MetricName) -> Option { }) } +/// Converts a list of outcome metrics to client reports. +/// +/// Metrics in the outcome namespace are removed from `buckets`. Outcome metrics which cannot be converted +/// to a client report are dropped. This includes outcomes which cannot be mapped to client reports +/// (e.g. [`OutcomeId::INVALID`] outcomes) as well as malformed outcome metrics. +pub fn extract_client_reports(buckets: &mut Vec) -> impl Iterator { + let outcomes = buckets.extract_if(.., |bucket| { + bucket.name.namespace() == MetricNamespace::Outcomes + }); + + let mut reports = BTreeMap::new(); + + for bucket in outcomes { + let Some((section, discarded_event)) = to_discarded_event(&bucket) else { + continue; + }; + + let report = reports + .entry(bucket.timestamp) + .or_insert_with(|| ClientReport { + timestamp: Some(bucket.timestamp), + ..Default::default() + }); + + match section { + ClientReportSection::Discarded => report.discarded_events.push(discarded_event), + ClientReportSection::RateLimited => report.rate_limited_events.push(discarded_event), + ClientReportSection::Filtered => report.filtered_events.push(discarded_event), + ClientReportSection::FilteredSampling => { + report.filtered_sampling_events.push(discarded_event) + } + } + } + + reports.into_values() +} + +#[derive(Copy, Clone)] +enum ClientReportSection { + Discarded, + RateLimited, + Filtered, + FilteredSampling, +} + +fn to_discarded_event(bucket: &Bucket) -> Option<(ClientReportSection, DiscardedEvent)> { + let reason = bucket + .tags + .get("reason") + .map(String::as_str) + .unwrap_or_default(); + + let section = match bucket.name.as_ref() { + FILTERED_MRI if reason.starts_with("Sampled:") => ClientReportSection::FilteredSampling, + FILTERED_MRI => ClientReportSection::Filtered, + RATE_LIMITED_MRI => ClientReportSection::RateLimited, + CLIENT_DISCARD_MRI => ClientReportSection::Discarded, + // Ported over the from the existing logic, ticket exists to expand it to all invalid outcomes + // . + INVALID_MRI if matches!(reason, "invalid_signature" | "missing_signature") => { + ClientReportSection::Discarded + } + _ => { + relay_log::debug!("outcome bucket '{bucket:?}' cannot be converted to client report"); + return None; + } + }; + + let category = bucket + .tags + .get("category") + .and_then(|value| value.parse::().ok()) + .and_then(|value| DataCategory::try_from(value).ok()); + + debug_assert!(category.is_some()); + debug_assert_ne!(category, Some(DataCategory::Unknown)); + + let quantity = match &bucket.value { + BucketValue::Counter(value) => value.to_f64() as u32, + _ => return None, + }; + + Some(( + section, + DiscardedEvent { + reason: reason.to_owned(), + category: category?, + quantity, + }, + )) +} + #[cfg(test)] mod tests { use relay_base_schema::data_category::DataCategory; @@ -130,6 +224,31 @@ mod tests { .unwrap() } + fn bucket( + name: &str, + timestamp: u64, + reason: Option<&str>, + category: Option, + quantity: u32, + ) -> Bucket { + let mut tags = BTreeMap::new(); + if let Some(reason) = reason { + tags.insert("reason".to_owned(), reason.to_owned()); + } + if let Some(category) = category.and_then(DataCategory::value) { + tags.insert("category".to_owned(), category.to_string()); + } + + Bucket { + name: name.into(), + value: BucketValue::Counter(quantity.into()), + timestamp: UnixTimestamp::from_secs(timestamp), + tags, + width: 0, + metadata: Default::default(), + } + } + #[test] fn test_metric_mri_valid() { let name = MetricName::from(OUTCOME_ACCEPTED_MRI); @@ -252,4 +371,98 @@ mod tests { assert_eq!(id, outcome.to_outcome_id(), "{} | {outcome:?}", bucket.name); } } + + #[test] + fn test_to_client_report() { + let mut buckets = vec![ + bucket("c:custom/foo@none", 123, None, Some(DataCategory::Error), 7), + bucket( + FILTERED_MRI, + 123, + Some("release-version"), + Some(DataCategory::Error), + 2, + ), + bucket( + FILTERED_MRI, + 123, + Some("Sampled:3000"), + Some(DataCategory::Transaction), + 3, + ), + bucket(RATE_LIMITED_MRI, 123, None, Some(DataCategory::Session), 4), + bucket( + INVALID_MRI, + 123, + Some("invalid_signature"), + Some(DataCategory::Error), + 5, + ), + bucket( + INVALID_MRI, + 123, + Some("invalid_json"), + Some(DataCategory::Error), + 6, + ), + bucket( + CLIENT_DISCARD_MRI, + 124, + Some("queue_overflow"), + Some(DataCategory::Error), + 8, + ), + ]; + + let reports = extract_client_reports(&mut buckets).collect::>(); + + assert_eq!(buckets.len(), 1); + assert_eq!(buckets[0].name.as_ref(), "c:custom/foo@none"); + + insta::assert_json_snapshot!(reports, @r#" + [ + { + "timestamp": 123, + "discarded_events": [ + { + "reason": "invalid_signature", + "category": "error", + "quantity": 5 + } + ], + "rate_limited_events": [ + { + "reason": "", + "category": "session", + "quantity": 4 + } + ], + "filtered_events": [ + { + "reason": "release-version", + "category": "error", + "quantity": 2 + } + ], + "filtered_sampling_events": [ + { + "reason": "Sampled:3000", + "category": "transaction", + "quantity": 3 + } + ] + }, + { + "timestamp": 124, + "discarded_events": [ + { + "reason": "queue_overflow", + "category": "error", + "quantity": 8 + } + ] + } + ] + "#); + } } diff --git a/relay-server/src/services/outcome/mod.rs b/relay-server/src/services/outcome/mod.rs index 0529ff31181..33106f12127 100644 --- a/relay-server/src/services/outcome/mod.rs +++ b/relay-server/src/services/outcome/mod.rs @@ -9,13 +9,11 @@ use relay_quotas::ReasonCode; use relay_sampling::config::RuleId; use relay_sampling::evaluation::MatchedRuleIds; -mod aggregator; pub mod metric; mod service; use crate::envelope::{AttachmentType, ItemType}; -pub use self::aggregator::OutcomeAggregator; pub use self::service::*; /// The numerical identifier of the outcome category (Accepted, Filtered, ...) diff --git a/relay-server/src/services/outcome/service.rs b/relay-server/src/services/outcome/service.rs index 26cb492944f..070a7e4ff47 100644 --- a/relay-server/src/services/outcome/service.rs +++ b/relay-server/src/services/outcome/service.rs @@ -5,24 +5,24 @@ //! pipeline, outcomes may not be emitted if the item is accepted. use std::collections::BTreeMap; -use std::mem; use std::net::IpAddr; use std::sync::Arc; use std::time::Duration; -use crate::services::metrics::{Aggregator, MergeBuckets}; -use crate::services::outcome::{self, DiscardReason, Outcome}; -use crate::services::processor::{EnvelopeProcessor, SubmitClientReports}; -use crate::statsd::RelayCounters; -use crate::utils::SleepHandle; use chrono::{DateTime, Utc}; -use relay_common::time::UnixTimestamp; -use relay_config::{Config, EmitOutcomes}; +use relay_config::Config; use relay_event_schema::protocol::{ClientReport, DiscardedEvent, EventId}; +use relay_metrics::{MetricNamespace, UnixTimestamp}; use relay_quotas::{DataCategory, Scoping}; use relay_statsd::metric; use relay_system::{Addr, FromMessage, Interface, NoResponse, Service}; +use crate::services::metrics::{Aggregator, MergeBuckets}; +use crate::services::outcome::{self, DiscardReason, Outcome}; +use crate::services::processor::{EnvelopeProcessor, SubmitClientReports}; +use crate::statsd::RelayCounters; +use crate::utils::SleepHandle; + /// Tracks an [`Outcome`] of an Envelope item. /// /// See the module level documentation for more information. @@ -54,124 +54,206 @@ impl FromMessage for TrackOutcome { } } -/// Outcome producer backend via HTTP as [`ClientReport`]. +/// Service implementing the [`TrackOutcome`] interface, dropping all outcomes. +#[derive(Debug)] +pub struct NullOutcomeProducerService { + _priv: (), +} + +impl NullOutcomeProducerService { + pub fn new() -> Self { + Self { _priv: () } + } +} + +impl Service for NullOutcomeProducerService { + type Interface = TrackOutcome; + + async fn run(self, mut rx: relay_system::Receiver) { + while rx.recv().await.is_some() {} + } +} + +/// Service implementing the [`TrackOutcome`] interface. +#[derive(Debug)] +pub struct OutcomeProducerService { + config: Arc, + aggregator: Addr, +} + +impl OutcomeProducerService { + pub fn new(config: Arc, aggregator: Addr) -> Self { + Self { config, aggregator } + } + + fn handle_message(&self, message: TrackOutcome) { + send_outcome_metric(&message); + self.aggregator.send(MergeBuckets { + project_key: message.scoping.project_key, + buckets: vec![outcome::metric::to_metric(&message, &self.config)], + }) + } +} + +impl Service for OutcomeProducerService { + type Interface = TrackOutcome; + + async fn run(self, mut rx: relay_system::Receiver) { + relay_log::info!("outcome producer started."); + while let Some(message) = rx.recv().await { + self.handle_message(message); + } + relay_log::info!("outcome producer stopped."); + } +} + +/// Service implementing the [`TrackOutcome`] interface only emitting [`ClientReport`]s. +/// +/// This implementation is only useful for Relay instances running in Proxy mode. +/// It is much less optimized and the implementation assumes behaviour of a Proxy mode Relay, +/// like a relatively low volume of data and only a single scoping. It will behave correctly +/// event outside these parameters, but may not perform optimally. #[derive(Debug)] -struct ClientReportOutcomeProducer { - flush_interval: Duration, - unsent_reports: BTreeMap>, +pub struct ClientReportOutcomeProducerService { + processor: Addr, + buckets: BTreeMap, + bucket_interval: u32, + flush_interval: u32, flush_handle: SleepHandle, - envelope_processor: Addr, } -impl ClientReportOutcomeProducer { - fn new(config: &Config, envelope_processor: Addr) -> Self { +impl ClientReportOutcomeProducerService { + pub fn new(config: &Config, processor: Addr) -> Self { + let agg = &config + .aggregator_config_for(MetricNamespace::Outcomes) + .aggregator; + Self { - // Use same batch interval as outcome aggregator - flush_interval: Duration::from_secs(config.outcome_aggregator().flush_interval), - unsent_reports: BTreeMap::new(), + processor, + buckets: Default::default(), + bucket_interval: agg.bucket_interval.max(1), + flush_interval: agg.initial_delay, flush_handle: SleepHandle::idle(), - envelope_processor, } } - fn flush(&mut self) { - relay_log::trace!("flushing client reports"); - self.flush_handle.reset(); + fn handle_message(&mut self, message: TrackOutcome) { + send_outcome_metric(&message); - let unsent_reports = mem::take(&mut self.unsent_reports); - for (scoping, client_reports) in unsent_reports.into_iter() { - self.envelope_processor.send(SubmitClientReports { - client_reports, - scoping, - }); - } - } + let offset = u64::try_from(message.timestamp.timestamp()).unwrap_or(0) + / u64::from(self.bucket_interval); - fn handle_message(&mut self, msg: TrackOutcome) { - let mut client_report = ClientReport { - timestamp: Some(UnixTimestamp::from_secs( - msg.timestamp.timestamp().try_into().unwrap_or(0), - )), - ..Default::default() + let bucket_key = BucketKey { + offset, + scoping: message.scoping, }; - // The outcome type determines what field to place the outcome in: - let discarded_events = match msg.outcome { - Outcome::Filtered(_) => &mut client_report.filtered_events, - Outcome::FilteredSampling(_) => &mut client_report.filtered_sampling_events, - Outcome::RateLimited(_) => &mut client_report.rate_limited_events, + let discarded_events: fn(&mut ClientReport) -> &mut _ = match message.outcome { + Outcome::Filtered(_) => |cr| &mut cr.filtered_events, + Outcome::FilteredSampling(_) => |cr| &mut cr.filtered_sampling_events, + Outcome::RateLimited(_) => |cr| &mut cr.rate_limited_events, Outcome::Invalid(DiscardReason::InvalidSignature | DiscardReason::MissingSignature) => { - &mut client_report.discarded_events + |cr| &mut cr.discarded_events } _ => { relay_log::debug!( "Outcome '{}' cannot be converted to client report", - msg.outcome + message.outcome ); return; } }; - // Now that we know where to put it, let's create a DiscardedEvent - let discarded_event = DiscardedEvent { - reason: msg.outcome.to_reason().unwrap_or_default().to_string(), - category: msg.category, - quantity: msg.quantity, - }; - discarded_events.push(discarded_event); - - self.unsent_reports - .entry(msg.scoping) - .or_default() - .push(client_report); - - if self.flush_interval == Duration::ZERO { - // Flush immediately. Useful for integration tests. - self.flush(); - } else if self.flush_handle.is_idle() { - self.flush_handle.set(self.flush_interval); + let client_report = self.buckets.entry(bucket_key).or_default(); + let discarded_events = discarded_events(client_report); + + let reason = message.outcome.to_reason().unwrap_or_default(); + let category = message.category; + + // Linear search is fine, we only have a limited amount of outcomes and volume as this + // service is only supposed to be used for proxy mode Relay. + let discaded_event = discarded_events + .iter_mut() + .find(|de| de.reason == reason && de.category == category); + + match discaded_event { + Some(discarded_event) => discarded_event.quantity += message.quantity, + None => discarded_events.push(DiscardedEvent { + reason: reason.into_owned(), + category, + quantity: message.quantity, + }), + } + + match self.flush_interval { + 0 => self.do_flush(), + v => self.flush_handle.set_if_idle(Duration::from_secs(v.into())), } } + + fn do_flush(&mut self) { + for (bucket_key, client_report) in std::mem::take(&mut self.buckets) { + let BucketKey { offset, scoping } = bucket_key; + + let timestamp = Some(offset) + // May be zero as we default timestamps out of range to 0. + .filter(|offset| *offset > 0) + .map(|offset| offset * u64::from(self.bucket_interval)) + .map(UnixTimestamp::from_secs); + + let client_report = ClientReport { + timestamp, + ..client_report + }; + + self.processor.send(SubmitClientReports { + client_reports: vec![client_report], + scoping, + }); + } + } + + fn handle_shutdown(&mut self) { + self.flush_interval = 0; + self.do_flush(); + } } -impl Service for ClientReportOutcomeProducer { +impl Service for ClientReportOutcomeProducerService { type Interface = TrackOutcome; async fn run(mut self, mut rx: relay_system::Receiver) { + let mut shutdown = relay_system::Controller::shutdown_handle(); + relay_log::info!("client report outcome producer started"); + loop { tokio::select! { - // Prioritize flush over receiving messages to prevent starving. biased; - () = &mut self.flush_handle => self.flush(), + () = &mut self.flush_handle => self.do_flush(), Some(message) = rx.recv() => self.handle_message(message), + _ = shutdown.notified() => self.handle_shutdown(), + else => break, } } + self.do_flush(); + relay_log::info!("client report outcome producer stopped"); } } -/// Produces [`Outcome`]s to a configurable backend. -/// -/// The backend is configured through the `outcomes` configuration object and can be: -/// -/// 1. Metrics -/// 2. Upstream Relay via client reports in external configuration -/// 3. (default) Disabled -#[derive(Debug)] -pub struct OutcomeProducer(TrackOutcome); - -impl Interface for OutcomeProducer {} - -impl FromMessage for OutcomeProducer { - type Response = NoResponse; - - fn from_message(message: TrackOutcome, _: ()) -> Self { - Self(message) - } +/// Contains everything to aggregate a [`ClientReport`]. +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)] +struct BucketKey { + /// The time slot for which outcomes are aggregated. + /// + /// The offset follows the formula: `timestamp = offset * bucket_interval`. + offset: u64, + /// Scoping of the outcome. + scoping: Scoping, } -fn send_outcome_metric(message: &TrackOutcome, to: &'static str) { +fn send_outcome_metric(message: &TrackOutcome) { let outcome_name = match message.outcome { Outcome::Accepted => "accepted", Outcome::Filtered(_) | Outcome::FilteredSampling(_) => "filtered", @@ -185,113 +267,10 @@ fn send_outcome_metric(message: &TrackOutcome, to: &'static str) { counter(RelayCounters::OutcomeQuantity) += message.quantity.into(), category = message.category.name(), outcome = outcome_name, - to = to, ); metric!( counter(RelayCounters::Outcomes) += 1, reason = message.outcome.to_reason().unwrap_or_default(), outcome = outcome_name, - to = to, ); } - -#[derive(Debug)] -enum OutcomeBroker { - ClientReport(Addr), - Metric(Addr), - Disabled, -} - -impl OutcomeBroker { - fn handle_message(&self, message: OutcomeProducer, config: &Config) { - relay_log::with_scope(|_| {}, || self.handle_track_outcome(message.0, config)) - } - - fn handle_track_outcome(&self, message: TrackOutcome, config: &Config) { - match self { - Self::Metric(metrics) => { - send_outcome_metric(&message, "metric"); - metrics.send(MergeBuckets { - project_key: message.scoping.project_key, - buckets: vec![outcome::metric::to_metric(&message, config)], - }) - } - Self::ClientReport(producer) => { - send_outcome_metric(&message, "client_report"); - producer.send(message); - } - Self::Disabled => (), - } - } -} - -#[derive(Debug)] -enum ProducerInner { - Metric(Addr), - ClientReport(ClientReportOutcomeProducer), - Disabled, -} - -impl ProducerInner { - fn start(self) -> OutcomeBroker { - match self { - ProducerInner::Metric(inner) => OutcomeBroker::Metric(inner), - ProducerInner::ClientReport(inner) => { - OutcomeBroker::ClientReport(inner.start_detached()) - } - ProducerInner::Disabled => OutcomeBroker::Disabled, - } - } -} - -/// Service implementing the [`OutcomeProducer`] interface. -#[derive(Debug)] -pub struct OutcomeProducerService { - config: Arc, - inner: ProducerInner, -} - -impl OutcomeProducerService { - pub fn create( - config: Arc, - envelope_processor: Addr, - metric_aggregator: Addr, - ) -> anyhow::Result { - let inner = match config.emit_outcomes() { - EmitOutcomes::AsOutcomes => { - relay_log::info!("Configured to emit outcomes via metrics"); - ProducerInner::Metric(metric_aggregator) - } - EmitOutcomes::AsClientReports => { - // We emit client reports, and we do NOT accept raw outcomes - relay_log::info!("Configured to emit outcomes as client reports"); - ProducerInner::ClientReport(ClientReportOutcomeProducer::new( - &config, - envelope_processor, - )) - } - EmitOutcomes::None => { - relay_log::info!("Configured to drop all outcomes"); - ProducerInner::Disabled - } - }; - - Ok(Self { config, inner }) - } -} - -impl Service for OutcomeProducerService { - type Interface = OutcomeProducer; - - async fn run(self, mut rx: relay_system::Receiver) { - let Self { config, inner } = self; - - let broker = inner.start(); - - relay_log::info!("OutcomeProducer started."); - while let Some(message) = rx.recv().await { - broker.handle_message(message, &config); - } - relay_log::info!("OutcomeProducer stopped."); - } -} diff --git a/relay-server/src/services/processor.rs b/relay-server/src/services/processor.rs index a62ae230173..ba588a55315 100644 --- a/relay-server/src/services/processor.rs +++ b/relay-server/src/services/processor.rs @@ -18,7 +18,7 @@ use futures::future::BoxFuture; use relay_base_schema::project::{ProjectId, ProjectKey}; use relay_cogs::{AppFeature, Cogs, FeatureWeights, ResourceId, Token}; use relay_common::time::UnixTimestamp; -use relay_config::{Config, HttpEncoding, UpstreamDescriptor}; +use relay_config::{Config, EmitOutcomes, HttpEncoding, UpstreamDescriptor}; use relay_event_normalization::{ClockDriftProcessor, GeoIpLookup}; use relay_event_schema::processor::ProcessingAction; use relay_event_schema::protocol::ClientReport; @@ -43,7 +43,7 @@ use crate::processing::{Forward as _, Output, Outputs, QuotaRateLimiter}; use crate::service::ServiceError; use crate::services::global_config::GlobalConfigHandle; use crate::services::metrics::{Aggregator, FlushBuckets, MergeBuckets, ProjectBuckets}; -use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome, TrackOutcome}; +use crate::services::outcome::{self, DiscardItemType, DiscardReason, Outcome, TrackOutcome}; use crate::services::projects::cache::ProjectCacheHandle; use crate::services::projects::project::{ProjectInfo, ProjectState}; use crate::services::upstream::{ @@ -951,6 +951,16 @@ impl EnvelopeProcessorService { scoping, } = message; + relay_log::trace!( + "sending {} client report(s) to project id {}", + client_reports.len(), + scoping.project_id + ); + + if client_reports.is_empty() { + return; + } + let upstream = self.inner.config.upstream(); let dsn = PartialDsn::outbound(&scoping, upstream); @@ -1387,6 +1397,25 @@ impl EnvelopeProcessorService { } } + /// Removes all outcome metrics from `message` and sends them as client reports. + /// + /// Returns a new [`FlushBuckets`] message, without any outcome metrics remaining. + fn encode_metrics_client_reports(&self, mut message: FlushBuckets) -> FlushBuckets { + for ProjectBuckets { + buckets, scoping, .. + } in message.buckets.values_mut() + { + let client_reports = outcome::metric::extract_client_reports(buckets).collect(); + + self.handle_submit_client_reports(SubmitClientReports { + client_reports, + scoping: *scoping, + }); + } + + message + } + async fn handle_flush_buckets(&self, mut message: FlushBuckets) { for (project_key, pb) in message.buckets.iter_mut() { let buckets = std::mem::take(&mut pb.buckets); @@ -1403,6 +1432,14 @@ impl EnvelopeProcessorService { .await; } + // Processing Relays never send outcomes as client reports, which is why this check is after + // the processing check. + if self.inner.config.emit_outcomes() == EmitOutcomes::AsClientReports { + // Remove client reports from metrics to be sent, if configured as client reports + // and send them separately. + message = self.encode_metrics_client_reports(message); + } + if self.inner.config.http_global_metrics() { self.encode_metrics_global(message) } else { diff --git a/relay-server/src/statsd.rs b/relay-server/src/statsd.rs index 12ab2692df1..d8f1573e73a 100644 --- a/relay-server/src/statsd.rs +++ b/relay-server/src/statsd.rs @@ -536,8 +536,6 @@ pub enum RelayTimers { /// - `category`: The data category of the payload. Can be one of: `event`, `transaction`, /// `security`, or `session`. TimestampDelay, - /// The time it takes the outcome aggregator to flush aggregated outcomes. - OutcomeAggregatorFlushTime, /// Time in milliseconds spent on parsing, normalizing and scrubbing replay recordings. ReplayRecordingProcessing, /// Total time spent to send a request and receive the response from upstream. @@ -668,7 +666,6 @@ impl TimerMetric for RelayTimers { RelayTimers::AttachmentScrubbing => "scrubbing.attachments.duration", RelayTimers::UpstreamRequestsDuration => "upstream.requests.duration", RelayTimers::TimestampDelay => "requests.timestamp_delay", - RelayTimers::OutcomeAggregatorFlushTime => "outcomes.aggregator.flush_time", RelayTimers::ReplayRecordingProcessing => "replay.recording.process", RelayTimers::GlobalConfigRequestDuration => "global_config.requests.duration", RelayTimers::ProcessMessageDuration => "processor.message.duration", diff --git a/relay-server/src/utils/sleep_handle.rs b/relay-server/src/utils/sleep_handle.rs index 6d7f0d3c3a0..85eeed6b161 100644 --- a/relay-server/src/utils/sleep_handle.rs +++ b/relay-server/src/utils/sleep_handle.rs @@ -3,6 +3,9 @@ use std::pin::Pin; use std::task::Poll; use std::time::Duration; +use futures::FutureExt; +use futures::future::FusedFuture; + /// A future wrapper around [`tokio::time::Sleep`]. /// /// When initialized with [`SleepHandle::idle`], this future is pending indefinitely every time it @@ -28,7 +31,14 @@ impl SleepHandle { self.0 = Some(Box::pin(tokio::time::sleep(duration))); } - /// Checks wether the internal state is currently pending indefinite. + /// Primes the internal state like [`Self::set`], only if currently [`Self::idle`]. + pub fn set_if_idle(&mut self, duration: Duration) { + if self.is_idle() { + self.set(duration) + } + } + + /// Checks whether the internal state is currently pending indefinite. pub fn is_idle(&self) -> bool { self.0.is_none() } @@ -39,7 +49,7 @@ impl Future for SleepHandle { fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { let poll = match &mut self.0 { - Some(sleep) => Pin::new(sleep).poll(cx), + Some(sleep) => sleep.poll_unpin(cx), None => Poll::Pending, }; @@ -50,3 +60,10 @@ impl Future for SleepHandle { poll } } + +impl FusedFuture for SleepHandle { + fn is_terminated(&self) -> bool { + // The handle never terminates, it may be reset or primed at any time. + false + } +} diff --git a/tests/integration/test_metrics.py b/tests/integration/test_metrics.py index 8b47bd76bf8..aa18a4f4be2 100644 --- a/tests/integration/test_metrics.py +++ b/tests/integration/test_metrics.py @@ -735,8 +735,6 @@ def test_transaction_metrics_extraction_external_relays( # Client reports. envelope = mini_sentry.get_captured_envelope() assert len(envelope.items) == 1 - envelope = mini_sentry.get_captured_envelope() - assert len(envelope.items) == 1 if send_extracted_header: _, error = mini_sentry.test_failures.get() @@ -822,7 +820,7 @@ def test_no_transaction_metrics_when_filtered(mini_sentry, relay): relay.send_transaction(project_id, tx) # The only envelopes received should be outcomes for Transaction{,Indexed}: - reports = [mini_sentry.get_client_report() for _ in range(4)] + reports = [mini_sentry.get_client_report() for _ in range(1)] filtered_events = [ outcome for report in reports for outcome in report["filtered_events"] ] @@ -1027,19 +1025,12 @@ def test_generic_metric_extraction(mini_sentry, relay): relay = relay(relay(mini_sentry, options=TEST_CONFIG), options=TEST_CONFIG) relay.send_transaction(PROJECT_ID, transaction) - # Skip client reports - envelope = mini_sentry.get_captured_envelope() - assert envelope.get_event() is None - envelope = mini_sentry.get_captured_envelope() - assert envelope.get_event() is None - - envelope = mini_sentry.get_captured_envelope() - for item in envelope.items: - # Transaction items should be sampled and not among the envelope items. - assert item.headers.get("type") != "transaction" - - item = envelope.items[0] - assert item.headers.get("type") == "metric_buckets" + while True: + envelope = mini_sentry.get_captured_envelope() + assert len(envelope.items) == 1 + item = envelope.items[0] + if item.headers.get("type") == "metric_buckets": + break metrics = metrics_without_keys( json.loads(item.get_bytes().decode()), keys={"metadata"} @@ -1324,7 +1315,7 @@ def test_histogram_outliers(mini_sentry, relay): relay.send_event(42, event) tags = {} - for _ in range(3): + for _ in range(2): envelope = mini_sentry.get_captured_envelope() for item in envelope: if item.type == "metric_buckets": From 5f7ae3ad85f93c6904ed6fba6051526b085ffef6 Mon Sep 17 00:00:00 2001 From: David Herberth Date: Mon, 22 Jun 2026 12:22:48 +0200 Subject: [PATCH 2/2] typo review --- relay-server/src/services/outcome/service.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/relay-server/src/services/outcome/service.rs b/relay-server/src/services/outcome/service.rs index 070a7e4ff47..3b53044c015 100644 --- a/relay-server/src/services/outcome/service.rs +++ b/relay-server/src/services/outcome/service.rs @@ -172,11 +172,11 @@ impl ClientReportOutcomeProducerService { // Linear search is fine, we only have a limited amount of outcomes and volume as this // service is only supposed to be used for proxy mode Relay. - let discaded_event = discarded_events + let discarded_event = discarded_events .iter_mut() .find(|de| de.reason == reason && de.category == category); - match discaded_event { + match discarded_event { Some(discarded_event) => discarded_event.quantity += message.quantity, None => discarded_events.push(DiscardedEvent { reason: reason.into_owned(),