feat(server): Track outcome source

py/CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 0.5.10
+
+- Set default transaction name (#576)
+- Apply clock drift correction based on received_at (#580, #582)
+- Add AWS Security Scanner to web crawlers (#577)
+- Do not default transactions to level error (#585)
+- Update `sentry-release-parser` to 0.6.0 (#590)
+- Add schema for success metrics (failed and errored processing) (#593)
+
 ## 0.5.9
 
 - PII: Make and/or selectors specific.

relay-cabi/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "relay-cabi"
-version = "0.5.9"
+version = "0.5.10"
 authors = ["Sentry <oss@sentry.io>"]
 homepage = "https://getsentry.github.io/relay/"
 repository = "https://github.com/getsentry/relay"

relay-common/src/macros.rs

@@ -62,39 +62,46 @@ macro_rules! tryf {
     };
 }
 
-/// An alternative to a `move` closure.
+/// A cloning alternative to a `move` closure.
 ///
-/// When one needs to use a closure with move semantics one often needs to clone and
-/// move some of the free variables. This macro automates the process of cloning and moving
-/// variables.
+/// When one needs to use a closure with move semantics one often needs to clone and move some of
+/// the free variables. This macro automates the process of cloning and moving variables.
 ///
 /// The following code:
-/// ```compile_fail
-/// let arg1 = v1.clone()
-/// let arg2 = v2.clone()
 ///
-/// let result = some_function( move || f(arg1, arg2)})
 /// ```
+/// # use std::sync::{Arc, Mutex};
+/// let shared = Arc::new(Mutex::new(0));
+///
+/// let cloned = shared.clone();
+/// std::thread::spawn(move || {
+///     *cloned.lock().unwrap() = 42
+/// }).join();
+///
+/// assert_eq!(*shared.lock().unwrap(), 42);
+/// ```
+///
 /// Can be rewritten in a cleaner way by using the `clone!` macro like so:
 ///
-/// ```compile_fail
-/// let result = some_function( clone! { v1, v2, || f(v1,v2)})
+/// ```
+/// # use std::sync::{Arc, Mutex};
+/// use relay_common::clone;
+///
+/// let shared = Arc::new(Mutex::new(0));
+/// std::thread::spawn(clone!(shared, || {
+///     *shared.lock().unwrap() = 42
+/// })).join();
+///
+/// assert_eq!(*shared.lock().unwrap(), 42);
 /// ```
 #[macro_export]
 macro_rules! clone {
-    (@param _) => ( _ );
-    (@param ()) => (());
-    (@param $x:ident) => ( $x );
-    ($($n:ident),+ , || $body:expr) => (
-        {
-            $( let $n = $n.clone(); )+
-            move || $body
-        }
-    );
-    ($($n:ident),+ , |$($p:tt),+| $body:expr) => (
-        {
-            $( let $n = $n.clone(); )+
-            move |$(clone!(@param $p),)+| $body
-        }
-    );
+    ($($n:ident ,)+ || $body:expr) => {{
+        $( let $n = $n.clone(); )+
+        move || $body
+    }};
+    ($($n:ident ,)+ |$($p:pat),+| $body:expr) => {{
+        $( let $n = $n.clone(); )+
+        move |$($p),+| $body
+    }};
 }

relay-config/src/config.rs

@@ -230,6 +230,8 @@ pub struct OverridableConfig {
     pub secret_key: Option<String>,
     /// The public key of the relay
     pub public_key: Option<String>,
+    /// Outcome source
+    pub outcome_source: Option<String>,
 }
 
 /// The relay credentials
@@ -351,9 +353,6 @@ pub struct Relay {
     pub tls_identity_path: Option<PathBuf>,
     /// Password for the PKCS12 archive.
     pub tls_identity_password: Option<String>,
-    /// Controls whether outcomes will be emitted when processing is disabled.
-    /// Processing relays always emit outcomes (for backwards compatibility).
-    pub emit_outcomes: bool,
 }
 
 impl Default for Relay {
@@ -366,7 +365,6 @@ impl Default for Relay {
             tls_port: None,
             tls_identity_path: None,
             tls_identity_password: None,
-            emit_outcomes: false,
         }
     }
 }
@@ -710,6 +708,35 @@ impl Default for Processing {
     }
 }
 
+/// 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,
+    /// The maximum number of outcomes that are batched before being sent
+    /// via http to the upstream (only applies to non processing relays)
+    pub max_outcome_batch_size: usize,
+    /// The maximum time interval (in milliseconds) that an outcome may be batched
+    /// via http to the upstream (only applies to non processing relays)
+    pub max_outcome_interval_millsec: u64,
+    /// Defines the source string registered in the outcomes originating from
+    /// this Relay ( typically something like the region and or the layer )
+    pub source: Option<String>,
+}
+
+impl Default for Outcomes {
+    fn default() -> Self {
+        Outcomes {
+            emit_outcomes: false,
+            max_outcome_batch_size: 1000,
+            max_outcome_interval_millsec: 500,
+            source: None,
+        }
+    }
+}
+
 /// Minimal version of a config for dumping out.
 #[derive(Serialize, Deserialize, Debug, Default)]
 pub struct MinimalConfig {
@@ -757,6 +784,8 @@ struct ConfigValues {
     sentry: Sentry,
     #[serde(default)]
     processing: Processing,
+    #[serde(default)]
+    outcomes: Outcomes,
 }
 
 impl ConfigObject for ConfigValues {
@@ -811,7 +840,7 @@ impl Config {
     /// command line parameters)
     pub fn apply_override(
         &mut self,
-        overrides: OverridableConfig,
+        mut overrides: OverridableConfig,
     ) -> Result<&mut Self, ConfigError> {
         let relay = &mut self.values.relay;
 
@@ -838,7 +867,7 @@ impl Config {
                 "true" | "1" => processing.enabled = true,
                 "false" | "0" | "" => processing.enabled = false,
                 _ => {
-                    return Err(ConfigError::new(ConfigErrorKind::InvalidValue).field("processing"))
+                    return Err(ConfigError::new(ConfigErrorKind::InvalidValue).field("processing"));
                 }
             }
         }
@@ -886,6 +915,10 @@ impl Config {
         } else {
             None
         };
+        let mut outcomes = &mut self.values.outcomes;
+        if overrides.outcome_source.is_some() {
+            outcomes.source = overrides.outcome_source.take();
+        }
 
         if let Some(credentials) = &mut self.credentials {
             //we have existing credentials we may override some entries
@@ -1044,7 +1077,22 @@ impl Config {
 
     /// Returns the emit_outcomes flag
     pub fn emit_outcomes(&self) -> bool {
-        self.values.relay.emit_outcomes
+        self.values.outcomes.emit_outcomes
+    }
+
+    /// Returns the maximum number of outcomes that are batched before being sent
+    pub fn max_outcome_batch_size(&self) -> usize {
+        self.values.outcomes.max_outcome_batch_size
+    }
+
+    /// Returns the maximum interval that an outcome may be batched
+    pub fn max_outcome_interval(&self) -> Duration {
+        Duration::from_millis(self.values.outcomes.max_outcome_interval_millsec)
+    }
+
+    /// The originating source of the outcome
+    pub fn outcome_source(&self) -> Option<&str> {
+        self.values.outcomes.source.as_deref()
     }
 
     /// Returns the log level.

relay-general/src/protocol/span.rs

@@ -1,6 +1,6 @@
 use chrono::{DateTime, Utc};
 
-use crate::protocol::{OperationType, SpanId, TraceId};
+use crate::protocol::{OperationType, SpanId, SpanStatus, TraceId};
 use crate::types::{Annotated, Object, Value};
 
 #[derive(Clone, Debug, Default, PartialEq, Empty, FromValue, ToValue, ProcessValue)]
@@ -34,6 +34,9 @@ pub struct Span {
     #[metastructure(required = "true")]
     pub trace_id: Annotated<TraceId>,
 
+    /// The status of a span
+    pub status: Annotated<SpanStatus>,
+
     // TODO remove retain when the api stabilizes
     /// Additional arbitrary fields for forwards compatibility.
     #[metastructure(additional_properties, retain = "true", pii = "maybe")]
@@ -53,7 +56,8 @@ mod tests {
   "description": "desc",
   "op": "operation",
   "span_id": "fa90fdead5f74052",
-  "trace_id": "4c79f60c11214eb38604f4ae0781bfb2"
+  "trace_id": "4c79f60c11214eb38604f4ae0781bfb2",
+  "status": "ok"
 }"#;
 
         let span = Annotated::new(Span {
@@ -63,6 +67,7 @@ mod tests {
             op: Annotated::new("operation".to_owned()),
             trace_id: Annotated::new(TraceId("4c79f60c11214eb38604f4ae0781bfb2".into())),
             span_id: Annotated::new(SpanId("fa90fdead5f74052".into())),
+            status: Annotated::new(SpanStatus::Ok),
             ..Default::default()
         });
         assert_eq_str!(json, span.to_json_pretty().unwrap());

relay-quotas/Cargo.toml

@@ -11,7 +11,7 @@ publish = false
 
 [features]
 default = []
-rate-limiter = [
+redis = [
     "failure",
     "log",
     "relay-redis/impl",

relay-quotas/src/legacy.rs

@@ -8,7 +8,7 @@
 use serde::{ser::SerializeSeq, Deserialize, Deserializer, Serialize, Serializer};
 use smallvec::smallvec;
 
-use crate::types::{DataCategory, Quota, QuotaScope, ReasonCode};
+use crate::quota::{DataCategory, Quota, QuotaScope, ReasonCode};
 
 /// Legacy format of the `Quota` type.
 #[derive(Deserialize, Serialize)]

relay-quotas/src/lib.rs

@@ -2,13 +2,20 @@
 
 #![warn(missing_docs)]
 
-mod types;
-pub use self::types::*;
+/// The default timeout to apply when a scope is fully rejected. This
+/// typically happens for disabled keys, projects, or organizations.
+const REJECT_ALL_SECS: u64 = 60;
+
+mod quota;
+mod rate_limit;
+
+pub use self::quota::*;
+pub use self::rate_limit::*;
 
 #[cfg(feature = "legacy")]
 pub mod legacy;
 
-#[cfg(feature = "rate-limiter")]
-mod rate_limiter;
-#[cfg(feature = "rate-limiter")]
-pub use self::rate_limiter::*;
+#[cfg(feature = "redis")]
+mod redis;
+#[cfg(feature = "redis")]
+pub use self::redis::*;