ref(cogs): No longer send cogs data to Kafka (#3953)

We can track COGS now via statsd metrics, there is no longer a need to
emit COGS to Kafka. This also means we have proper COGS for PoP Relays
through statsd.

In a follow-up we can remove the Kafka topic configuration from the
deployment.

CHANGELOG.md

@@ -17,6 +17,10 @@
 - Record too long discard reason for session replays. ([#3950](https://github.com/getsentry/relay/pull/3950))
 - Add `EnvelopeStore` trait and implement `DiskUsage` for tracking disk usage. ([#3925](https://github.com/getsentry/relay/pull/3925))
 
+**Internal**:
+
+- No longer send COGS data to dedicated Kafka topic. ([#3953](https://github.com/getsentry/relay/pull/3953))
+
 ## 24.8.0
 
 **Bug Fixes**:

Cargo.toml

@@ -148,7 +148,6 @@ sentry-core = "0.32.2"
 sentry-kafka-schemas = { version = "0.1.86", default-features = false }
 sentry-release-parser = { version = "1.3.2", default-features = false }
 sentry-types = "0.32.2"
-sentry_usage_accountant = { version = "0.1.0", default-features = false }
 semver = "1.0.23"
 serde = { version = "1.0.159", features = ["derive", "rc"] }
 serde-transcode = "1.1.1"

relay-config/src/config.rs

@@ -1505,16 +1505,6 @@ pub struct Cogs {
     ///
     /// Defaults to `false`.
     enabled: bool,
-    /// Granularity of the COGS measurements.
-    ///
-    /// Measurements are aggregated based on the granularity in seconds.
-    ///
-    /// Aggregated measurements are always flushed at the end of their
-    /// aggregation window, which means the granularity also controls the flush
-    /// interval.
-    ///
-    /// Defaults to `60` (1 minute).
-    granularity_secs: u64,
     /// Maximium amount of COGS measurements allowed to backlog.
     ///
     /// Any additional COGS measurements recorded will be dropped.
@@ -1533,7 +1523,6 @@ impl Default for Cogs {
     fn default() -> Self {
         Self {
             enabled: false,
-            granularity_secs: 60,
             max_queue_size: 10_000,
             relay_resource_id: "relay_service".to_owned(),
         }
@@ -2511,11 +2500,6 @@ impl Config {
         self.values.cogs.enabled
     }
 
-    /// Granularity for COGS measurements.
-    pub fn cogs_granularity(&self) -> Duration {
-        Duration::from_secs(self.values.cogs.granularity_secs)
-    }
-
     /// Maximum amount of COGS measurements buffered in memory.
     pub fn cogs_max_queue_size(&self) -> u64 {
         self.values.cogs.max_queue_size

relay-kafka/src/config.rs

@@ -49,8 +49,6 @@ pub enum KafkaTopic {
     Spans,
     /// Summary for metrics collected during a span.
     MetricsSummaries,
-    /// COGS measurements topic.
-    Cogs,
     /// Feedback events topic.
     Feedback,
 }
@@ -60,7 +58,7 @@ impl KafkaTopic {
     /// It will have to be adjusted if the new variants are added.
     pub fn iter() -> std::slice::Iter<'static, Self> {
         use KafkaTopic::*;
-        static TOPICS: [KafkaTopic; 15] = [
+        static TOPICS: [KafkaTopic; 14] = [
             Events,
             Attachments,
             Transactions,
@@ -74,7 +72,6 @@ impl KafkaTopic {
             Monitors,
             Spans,
             MetricsSummaries,
-            Cogs,
             Feedback,
         ];
         TOPICS.iter()
@@ -137,7 +134,6 @@ define_topic_assignments! {
     monitors: (KafkaTopic::Monitors, "ingest-monitors", "Monitor check-ins."),
     spans: (KafkaTopic::Spans, "snuba-spans", "Standalone spans without a transaction."),
     metrics_summaries: (KafkaTopic::MetricsSummaries, "snuba-metrics-summaries", "Summary for metrics collected during a span."),
-    cogs: (KafkaTopic::Cogs, "shared-resources-usage", "COGS measurements."),
     feedback: (KafkaTopic::Feedback, "ingest-feedback-events", "Feedback events topic."),
 }
 

relay-server/Cargo.toml

@@ -100,7 +100,6 @@ reqwest = { workspace = true, features = [
 ] }
 rmp-serde = { workspace = true }
 rust-embed = { workspace = true, optional = true }
-sentry_usage_accountant = { workspace = true, default-features = false }
 serde = { workspace = true }
 serde_bytes = { workspace = true }
 serde_json = { workspace = true }

relay-server/src/service.rs

@@ -217,11 +217,7 @@ impl ServiceState {
             })
             .transpose()?;
 
-        let cogs = CogsService::new(
-            &config,
-            #[cfg(feature = "processing")]
-            store.clone(),
-        );
+        let cogs = CogsService::new(&config);
         let cogs = Cogs::new(CogsServiceRecorder::new(&config, cogs.start()));
 
         EnvelopeProcessorService::new(

relay-server/src/services/cogs.rs

@@ -2,11 +2,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
 
 use relay_cogs::{CogsMeasurement, CogsRecorder, ResourceId};
 use relay_config::Config;
-use relay_system::{Addr, Controller, FromMessage, Interface, Service};
-use sentry_usage_accountant::{Producer, UsageAccountant, UsageUnit};
+use relay_system::{Addr, FromMessage, Interface, Service};
 
-#[cfg(feature = "processing")]
-use crate::services::store::{Store, StoreCogs};
 use crate::statsd::RelayCounters;
 
 pub struct CogsReport(CogsMeasurement);
@@ -21,52 +18,15 @@ impl FromMessage<CogsMeasurement> for CogsReport {
     }
 }
 
-/// Relay [accountant producer](Producer) which produces to a dynamic
-/// upstram.
-enum RelayProducer {
-    /// Produces to Kafka via the store service.
-    #[cfg(feature = "processing")]
-    Store(Addr<Store>),
-    /// Discards all measurements.
-    Noop,
-}
-
-impl Producer for RelayProducer {
-    type Error = std::convert::Infallible;
-
-    fn send(&mut self, _message: Vec<u8>) -> Result<(), Self::Error> {
-        match self {
-            #[cfg(feature = "processing")]
-            Self::Store(addr) => addr.send(StoreCogs(_message)),
-            Self::Noop => {}
-        }
-
-        Ok(())
-    }
-}
-
 /// Service implementing the [`CogsReport`] interface.
 pub struct CogsService {
     relay_resource_id: String,
-    usage_accountant: UsageAccountant<RelayProducer>,
 }
 
 impl CogsService {
-    pub fn new(config: &Config, #[cfg(feature = "processing")] store: Option<Addr<Store>>) -> Self {
-        #[cfg(feature = "processing")]
-        let producer = store
-            .map(RelayProducer::Store)
-            .filter(|_| config.cogs_enabled())
-            .unwrap_or(RelayProducer::Noop);
-
-        #[cfg(not(feature = "processing"))]
-        let producer = RelayProducer::Noop;
-
-        let granularity = chrono::Duration::from_std(config.cogs_granularity()).ok();
-
+    pub fn new(config: &Config) -> Self {
         Self {
             relay_resource_id: config.cogs_relay_resource_id().to_owned(),
-            usage_accountant: UsageAccountant::new(producer, granularity),
         }
     }
 
@@ -77,28 +37,17 @@ impl CogsService {
             ResourceId::Relay => &self.relay_resource_id,
         };
 
-        let (amount, unit) = match measurement.value {
-            relay_cogs::Value::Time(duration) => (
-                duration.as_micros().try_into().unwrap_or(u64::MAX),
-                UsageUnit::Milliseconds,
-            ),
+        let amount = match measurement.value {
+            relay_cogs::Value::Time(duration) => {
+                duration.as_micros().try_into().unwrap_or(i64::MAX)
+            }
         };
 
         relay_statsd::metric!(
-            counter(RelayCounters::CogsUsage) += amount as i64,
+            counter(RelayCounters::CogsUsage) += amount,
             resource_id = resource_id,
             app_feature = measurement.feature.as_str()
         );
-
-        self.usage_accountant
-            .record(resource_id, measurement.feature.as_str(), amount, unit)
-            .unwrap_or_else(|err| match err {});
-    }
-
-    fn handle_shutdown(&mut self) {
-        self.usage_accountant
-            .flush()
-            .unwrap_or_else(|err| match err {});
     }
 }
 
@@ -107,17 +56,8 @@ impl Service for CogsService {
 
     fn spawn_handler(mut self, mut rx: relay_system::Receiver<Self::Interface>) {
         tokio::spawn(async move {
-            let mut shutdown = Controller::shutdown_handle();
-
-            loop {
-                tokio::select! {
-                    biased;
-
-                    Some(message) = rx.recv() => self.handle_report(message),
-                    _ = shutdown.notified() => self.handle_shutdown(),
-
-                    else => break,
-                }
+            while let Some(message) = rx.recv().await {
+                self.handle_report(message);
             }
         });
     }

relay-server/src/services/store.rs

@@ -90,15 +90,11 @@ pub struct StoreMetrics {
     pub retention: u16,
 }
 
-#[derive(Debug)]
-pub struct StoreCogs(pub Vec<u8>);
-
 /// Service interface for the [`StoreEnvelope`] message.
 #[derive(Debug)]
 pub enum Store {
     Envelope(StoreEnvelope),
     Metrics(StoreMetrics),
-    Cogs(StoreCogs),
 }
 
 impl Store {
@@ -107,7 +103,6 @@ impl Store {
         match self {
             Store::Envelope(_) => "envelope",
             Store::Metrics(_) => "metrics",
-            Store::Cogs(_) => "cogs",
         }
     }
 }
@@ -130,14 +125,6 @@ impl FromMessage<StoreMetrics> for Store {
     }
 }
 
-impl FromMessage<StoreCogs> for Store {
-    type Response = NoResponse;
-
-    fn from_message(message: StoreCogs, _: ()) -> Self {
-        Self::Cogs(message)
-    }
-}
-
 /// Service implementing the [`Store`] interface.
 pub struct StoreService {
     workers: WorkerGroup,
@@ -173,7 +160,6 @@ impl StoreService {
             match message {
                 Store::Envelope(message) => self.handle_store_envelope(message),
                 Store::Metrics(message) => self.handle_store_metrics(message),
-                Store::Cogs(message) => self.handle_store_cogs(message),
             }
         })
     }
@@ -444,16 +430,6 @@ impl StoreService {
         }
     }
 
-    fn handle_store_cogs(&self, StoreCogs(payload): StoreCogs) {
-        let message = KafkaMessage::Cogs(CogsKafkaMessage(payload));
-        if let Err(error) = self.produce(KafkaTopic::Cogs, message) {
-            relay_log::error!(
-                error = &error as &dyn std::error::Error,
-                "failed to store cogs measurement"
-            );
-        }
-    }
-
     fn create_metric_message<'a>(
         &self,
         organization_id: u64,
@@ -1449,10 +1425,6 @@ struct ProfileChunkKafkaMessage {
     payload: Bytes,
 }
 
-#[derive(Debug, Serialize)]
-#[serde(transparent)]
-struct CogsKafkaMessage(Vec<u8>);
-
 /// An enum over all possible ingest messages.
 #[derive(Debug, Serialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
@@ -1479,7 +1451,6 @@ enum KafkaMessage<'a> {
         message: SpanKafkaMessage<'a>,
     },
     MetricsSummary(MetricsSummaryKafkaMessage<'a>),
-    Cogs(CogsKafkaMessage),
     ProfileChunk(ProfileChunkKafkaMessage),
 }
 
@@ -1505,7 +1476,6 @@ impl Message for KafkaMessage<'_> {
             KafkaMessage::CheckIn(_) => "check_in",
             KafkaMessage::Span { .. } => "span",
             KafkaMessage::MetricsSummary(_) => "metrics_summary",
-            KafkaMessage::Cogs(_) => "cogs",
             KafkaMessage::ProfileChunk(_) => "profile_chunk",
         }
     }
@@ -1530,7 +1500,6 @@ impl Message for KafkaMessage<'_> {
             | Self::Span { .. }
             | Self::ReplayRecordingNotChunked(_)
             | Self::MetricsSummary(_)
-            | Self::Cogs(_)
             | Self::ProfileChunk(_) => Uuid::nil(),
 
             // TODO(ja): Determine a partitioning key
@@ -1584,8 +1553,6 @@ impl Message for KafkaMessage<'_> {
                 .map(Cow::Owned)
                 .map_err(ClientError::InvalidJson),
 
-            KafkaMessage::Cogs(CogsKafkaMessage(payload)) => Ok(payload.into()),
-
             _ => rmp_serde::to_vec_named(&self)
                 .map(Cow::Owned)
                 .map_err(ClientError::InvalidMsgPack),

tests/integration/conftest.py

@@ -36,7 +36,6 @@
     spans_consumer,
     profiles_consumer,
     metrics_summaries_consumer,
-    cogs_consumer,
     feedback_consumer,
 )