feat(outcomes): Emit outcomes as client reports [INGEST-247]

relay-config/src/config.rs

@@ -10,6 +10,7 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use failure::{Backtrace, Context, Fail};
+use serde::de::{Unexpected, Visitor};
 use serde::{de::DeserializeOwned, Deserialize, Deserializer, Serialize, Serializer};
 
 use relay_auth::{generate_key_pair, generate_relay_id, PublicKey, RelayId, SecretKey};
@@ -983,13 +984,94 @@ impl Default for OutcomeAggregatorConfig {
     }
 }
 
+/// Determines how to emit outcomes.
+/// For compatibility reasons, this can either be true, false or AsClientReports
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+
+pub enum EmitOutcomes {
+    /// Do not emit any outcomes
+    None,
+    /// Emit outcomes as client reports
+    AsClientReports,
+    /// Emit outcomes as outcomes
+    AsOutcomes,
+}
+
+impl EmitOutcomes {
+    /// Returns true if outcomes are not emitted at all.
+    pub fn is_false(&self) -> bool {
+        matches!(self, EmitOutcomes::None)
+    }
+
+    /// Returns true of outcomes are emitted via http, kafka, or client reports.
+    pub fn any(&self) -> bool {
+        !self.is_false()
+    }
+}
+
+impl Serialize for EmitOutcomes {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        // For compatibility, serialize None and AsOutcomes as booleans.
+        match self {
+            Self::None => serializer.serialize_bool(false),
+            Self::AsClientReports => serializer.serialize_str("as_client_reports"),
+            Self::AsOutcomes => serializer.serialize_bool(true),
+        }
+    }
+}
+
+struct EmitOutcomesVisitor {}
+
+impl<'de> Visitor<'de> for EmitOutcomesVisitor {
+    type Value = EmitOutcomes;
+
+    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+        formatter.write_str("true, false, or 'as_client_reports'")
+    }
+
+    fn visit_bool<E>(self, v: bool) -> Result<Self::Value, E>
+    where
+        E: serde::de::Error,
+    {
+        Ok(if v {
+            EmitOutcomes::AsOutcomes
+        } else {
+            EmitOutcomes::None
+        })
+    }
+
+    fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+    where
+        E: serde::de::Error,
+    {
+        if v == "as_client_reports" {
+            Ok(EmitOutcomes::AsClientReports)
+        } else {
+            Err(E::invalid_value(Unexpected::Str(v), &"as_client_reports"))
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for EmitOutcomes {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        deserializer.deserialize_any(EmitOutcomesVisitor {})
+    }
+}
+
 /// Outcome generation specific configuration values.
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(default)]
 pub struct Outcomes {
     /// Controls whether outcomes will be emitted when processing is disabled.
     /// Processing relays always emit outcomes (for backwards compatibility).
-    pub emit_outcomes: bool,
+    /// Can take the following values: false, "as_client_reports", true
+    pub emit_outcomes: EmitOutcomes,
     /// Controls wheather client reported outcomes should be emitted.
     pub emit_client_outcomes: bool,
     /// The maximum number of outcomes that are batched before being sent
@@ -1008,7 +1090,7 @@ pub struct Outcomes {
 impl Default for Outcomes {
     fn default() -> Self {
         Outcomes {
-            emit_outcomes: false,
+            emit_outcomes: EmitOutcomes::None,
             emit_client_outcomes: true,
             batch_size: 1000,
             batch_interval: 500,
@@ -1487,8 +1569,11 @@ impl Config {
     ///
     /// This is `true` either if `outcomes.emit_outcomes` is explicitly enabled, or if this Relay is
     /// in processing mode.
-    pub fn emit_outcomes(&self) -> bool {
-        self.values.outcomes.emit_outcomes || self.values.processing.enabled
+    pub fn emit_outcomes(&self) -> EmitOutcomes {
+        if self.processing_enabled() {
+            return EmitOutcomes::AsOutcomes;
+        }
+        self.values.outcomes.emit_outcomes
     }
 
     /// Returns whether this Relay should emit client outcomes
@@ -1856,4 +1941,25 @@ cache:
         assert_eq!(values.cache.envelope_buffer_size, 1_000_000);
         assert_eq!(values.cache.envelope_expiry, 1800);
     }
+
+    #[test]
+    fn test_emit_outcomes() {
+        for (serialized, deserialized) in &[
+            ("true", EmitOutcomes::AsOutcomes),
+            ("false", EmitOutcomes::None),
+            ("\"as_client_reports\"", EmitOutcomes::AsClientReports),
+        ] {
+            let value: EmitOutcomes = serde_json::from_str(serialized).unwrap();
+            assert_eq!(value, *deserialized);
+            assert_eq!(serde_json::to_string(&value).unwrap(), *serialized);
+        }
+    }
+
+    #[test]
+    fn test_emit_outcomes_invalid() {
+        assert!(matches!(
+            serde_json::from_str::<EmitOutcomes>("asdf"),
+            Err(_)
+        ));
+    }
 }

relay-general/benches/benchmarks.rs

@@ -91,6 +91,7 @@ fn bench_store_processor(c: &mut Criterion) {
         sent_at: None,
         received_at: None,
         breakdowns: None,
+        ..Default::default()
     };
 
     let mut processor = StoreProcessor::new(config, None);

relay-general/src/protocol/client_report.rs

@@ -8,12 +8,22 @@ pub struct DiscardedEvent {
     pub quantity: u32,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
 pub struct ClientReport {
     /// The timestamp of when the report was created.
     pub timestamp: Option<UnixTimestamp>,
-    /// Discard reason counters.
+    /// Counters for events discarded by the client.
     pub discarded_events: Vec<DiscardedEvent>,
+    /// Counters for events rate limited by a relay configured to emit outcomes as client reports
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub _server_rate_limited_events: Vec<DiscardedEvent>,
+    /// Counters for events filtered by a relay configured to emit outcomes as client reports
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub _server_filtered_events: Vec<DiscardedEvent>,
+    /// Counters for events filtered by a sampling rule,
+    /// by a relay configured to emit outcomes as client reports
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub _server_filtered_sampling_events: Vec<DiscardedEvent>,
 }
 
 impl ClientReport {
@@ -39,6 +49,9 @@ mod tests {
   "discarded_events": [
     {"reason": "foo_reason", "category": "error", "quantity": 42},
     {"reason": "foo_reason", "category": "transaction", "quantity": 23}
+  ],
+  "_server_rate_limited_events" : [
+      {"reason": "bar_reason", "category": "session", "quantity": 456}
   ]
 }"#;
 
@@ -55,6 +68,13 @@ mod tests {
       "category": "transaction",
       "quantity": 23
     }
+  ],
+  "_server_rate_limited_events": [
+    {
+      "reason": "bar_reason",
+      "category": "session",
+      "quantity": 456
+    }
   ]
 }"#;
 
@@ -72,6 +92,13 @@ mod tests {
                     quantity: 23,
                 },
             ],
+            _server_rate_limited_events: vec![DiscardedEvent {
+                reason: "bar_reason".into(),
+                category: DataCategory::Session,
+                quantity: 456,
+            }],
+
+            ..Default::default()
         };
 
         let parsed = ClientReport::parse(json.as_bytes()).unwrap();

relay-general/src/protocol/mod.rs

@@ -33,7 +33,7 @@ pub use sentry_release_parser::{validate_environment, validate_release};
 
 pub use self::breadcrumb::Breadcrumb;
 pub use self::breakdowns::Breakdowns;
-pub use self::client_report::ClientReport;
+pub use self::client_report::{ClientReport, DiscardedEvent};
 pub use self::clientsdk::{ClientSdkInfo, ClientSdkPackage};
 pub use self::constants::VALID_PLATFORMS;
 pub use self::contexts::{

relay-metrics/benches/aggregator.rs

@@ -131,6 +131,7 @@ fn bench_insert_and_flush(c: &mut Criterion) {
         bucket_interval: 1000,
         initial_delay: 0,
         debounce_delay: 0,
+        ..Default::default()
     };
 
     let flush_receiver = TestReceiver.start().recipient();