feat(inbound filters): Enable filtering sessions

CHANGELOG.md

@@ -7,6 +7,7 @@
 - Add logic to extract event json from userdata in prosperodumps. ([#4755](https://github.com/getsentry/relay/pull/4755)
 - Add browser name/version to logs. ([#4757](https://github.com/getsentry/relay/pull/4757))
 - Accept standalone spans in the V2 format. This feature is still highly experimental! ([#4771](https://github.com/getsentry/relay/pull/4771))
+- Enable filtering sessions by IP address, release, and user agent. ([#4745](https://github.com/getsentry/relay/pull/4745))
 
 **Internal**:
 

relay-event-schema/src/protocol/session.rs

@@ -2,6 +2,7 @@ use std::fmt::{self, Display};
 use std::time::SystemTime;
 
 use chrono::{DateTime, Utc};
+use relay_protocol::Getter;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
@@ -269,6 +270,14 @@ impl SessionLike for SessionUpdate {
     }
 }
 
+// Dummy implementation of `Getter` to satisfy the bound of `should_filter`.
+// We don't actually want to use `get_value` at this time.`
+impl Getter for SessionUpdate {
+    fn get_value(&self, _path: &str) -> Option<relay_protocol::Val<'_>> {
+        None
+    }
+}
+
 #[allow(clippy::trivially_copy_pass_by_ref)]
 fn is_zero(val: &u32) -> bool {
     *val == 0
@@ -353,6 +362,14 @@ impl SessionAggregates {
     }
 }
 
+// Dummy implementation of `Getter` to satisfy the bound of `should_filter`.
+// We don't actually want to use `get_value` at this time.`
+impl Getter for SessionAggregates {
+    fn get_value(&self, _path: &str) -> Option<relay_protocol::Val<'_>> {
+        None
+    }
+}
+
 #[cfg(test)]
 mod tests {
 

relay-filter/src/interface.rs

@@ -3,7 +3,8 @@
 use url::Url;
 
 use relay_event_schema::protocol::{
-    Csp, Event, EventType, Exception, LogEntry, Replay, Span, Values,
+    Csp, Event, EventType, Exception, LogEntry, Replay, SessionAggregates, SessionUpdate, Span,
+    Values,
 };
 
 /// A data item to which filters can be applied.
@@ -176,3 +177,85 @@ impl Filterable for Span {
         None
     }
 }
+
+impl Filterable for SessionUpdate {
+    fn csp(&self) -> Option<&Csp> {
+        None
+    }
+
+    fn exceptions(&self) -> Option<&Values<Exception>> {
+        None
+    }
+
+    fn ip_addr(&self) -> Option<&str> {
+        self.attributes
+            .ip_address
+            .as_ref()
+            .map(|addr| addr.as_str())
+    }
+
+    fn logentry(&self) -> Option<&LogEntry> {
+        None
+    }
+
+    fn release(&self) -> Option<&str> {
+        Some(&self.attributes.release)
+    }
+
+    fn transaction(&self) -> Option<&str> {
+        None
+    }
+
+    fn url(&self) -> Option<Url> {
+        None
+    }
+
+    fn user_agent(&self) -> Option<&str> {
+        self.attributes.user_agent.as_deref()
+    }
+
+    fn header(&self, _header_name: &str) -> Option<&str> {
+        None
+    }
+}
+
+impl Filterable for SessionAggregates {
+    fn csp(&self) -> Option<&Csp> {
+        None
+    }
+
+    fn exceptions(&self) -> Option<&Values<Exception>> {
+        None
+    }
+
+    fn ip_addr(&self) -> Option<&str> {
+        self.attributes
+            .ip_address
+            .as_ref()
+            .map(|addr| addr.as_str())
+    }
+
+    fn logentry(&self) -> Option<&LogEntry> {
+        None
+    }
+
+    fn release(&self) -> Option<&str> {
+        Some(&self.attributes.release)
+    }
+
+    fn transaction(&self) -> Option<&str> {
+        None
+    }
+
+    fn url(&self) -> Option<Url> {
+        None
+    }
+
+    fn user_agent(&self) -> Option<&str> {
+        self.attributes.user_agent.as_deref()
+    }
+
+    fn header(&self, _header_name: &str) -> Option<&str> {
+        None
+    }
+}

relay-filter/src/lib.rs

@@ -44,6 +44,10 @@ pub use interface::Filterable;
 ///
 /// If the event should be filtered, the `Err` returned contains a filter reason.
 /// The reason is the message returned by the first filter that didn't pass.
+///
+/// The `client_ip` parameter is the "client IP" extracted from the envelope. It's
+/// used for client IP filtering and should not be confused with a "user IP" that may
+/// be contained in the item, which is used for localhost filtering.
 pub fn should_filter<F: Filterable + Getter>(
     item: &F,
     client_ip: Option<IpAddr>,

relay-server/src/services/processor.rs

@@ -2010,24 +2010,26 @@ impl EnvelopeProcessorService {
     async fn process_sessions(
         &self,
         managed_envelope: &mut TypedEnvelope<SessionGroup>,
-        project_info: Arc<ProjectInfo>,
-        rate_limits: Arc<RateLimits>,
+        config: &Config,
+        project_info: &ProjectInfo,
+        rate_limits: &RateLimits,
     ) -> Result<Option<ProcessingExtractedMetrics>, ProcessingError> {
         let mut extracted_metrics = ProcessingExtractedMetrics::new();
 
         session::process(
             managed_envelope,
+            &self.inner.global_config.current(),
+            config,
             &mut extracted_metrics,
-            &project_info,
-            &self.inner.config,
+            project_info,
         );
 
         self.enforce_quotas(
             managed_envelope,
             Annotated::empty(),
             &mut extracted_metrics,
-            &project_info,
-            &rate_limits,
+            project_info,
+            rate_limits,
         )
         .await?;
 
@@ -2287,7 +2289,12 @@ impl EnvelopeProcessorService {
                     reservoir_counters
                 )
             }
-            ProcessingGroup::Session => run!(process_sessions, project_info, rate_limits),
+            ProcessingGroup::Session => run!(
+                process_sessions,
+                &self.inner.config.clone(),
+                &project_info,
+                &rate_limits
+            ),
             ProcessingGroup::Standalone => run!(
                 process_standalone,
                 self.inner.config.clone(),