ref(server): Forward logs without extra serialization to Store

Author: Dav1dde

CHANGELOG.md

@@ -14,9 +14,10 @@
 - Update opentelemetry-proto and sentry-protos dependencies. ([#4847](https://github.com/getsentry/relay/pull/4847))
 - Take into account more types of tokens when doing AI cost calculation. ([#4840](https://github.com/getsentry/relay/pull/4840))
 - Use the `FiniteF64` type for measurements. ([#4828](https://github.com/getsentry/relay/pull/4828))
-- Derive a `sentry.description` attribute for V2 spans ([#4832](https://github.com/getsentry/relay/pull/4832))
+- Derive a `sentry.description` attribute for V2 spans. ([#4832](https://github.com/getsentry/relay/pull/4832))
+- Forward logs to Kafka directly instead of serialized as envelope. ([#4875](https://github.com/getsentry/relay/pull/4875))
 - Consider `gen_ai` also as AI span op prefix. ([#4859](https://github.com/getsentry/relay/pull/4859))
-- Change pii scrubbing on some AI attributes to optional ([#4860](https://github.com/getsentry/relay/pull/4860))
+- Change pii scrubbing on some AI attributes to optional. ([#4860](https://github.com/getsentry/relay/pull/4860))
 - Conditionally set `total_cost` and `total_tokens` attributes on AI spans. ([#4868](https://github.com/getsentry/relay/pull/4868))
 
 ## 25.6.1

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

@@ -162,7 +162,7 @@ impl IntoValue for AttributeType {
 
 /// Wrapper struct around a collection of attributes with some convenience methods.
 #[derive(Debug, Clone, Default, PartialEq, Empty, FromValue, IntoValue, ProcessValue)]
-pub struct Attributes(Object<Attribute>);
+pub struct Attributes(pub Object<Attribute>);
 
 impl Attributes {
     /// Creates an empty collection of attributes.

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

@@ -817,6 +817,16 @@ impl Timestamp {
     }
 }
 
+impl From<Timestamp> for uuid::Timestamp {
+    fn from(value: Timestamp) -> Self {
+        uuid::Timestamp::from_unix(
+            uuid::NoContext,
+            u64::try_from(value.0.timestamp()).unwrap_or(0),
+            value.0.timestamp_subsec_nanos(),
+        )
+    }
+}
+
 impl ProcessValue for Timestamp {
     #[inline]
     fn value_type(&self) -> EnumSet<ValueType> {

relay-server/src/processing/logs/mod.rs

@@ -18,6 +18,7 @@ use crate::utils::ManagedEnvelope;
 
 mod filter;
 mod process;
+mod store;
 mod validate;
 
 /// Temporary byte count of a single log.
@@ -126,7 +127,7 @@ impl processing::Processor for LogsProcessor {
         self.limiter.enforce_quotas(&mut logs, ctx).await?;
 
         if ctx.is_processing() {
-            let mut logs = process::expand(logs);
+            let mut logs = process::expand(logs, ctx);
             process::process(&mut logs, ctx);
 
             Ok(Output::just(LogOutput::Processed(logs)))
@@ -141,6 +142,7 @@ impl processing::Processor for LogsProcessor {
     clippy::large_enum_variant,
     reason = "until we have a better solution to combat the excessive growth of Item, see #4819"
 )]
+#[derive(Debug)]
 pub enum LogOutput {
     NotProcessed(Managed<SerializedLogs>),
     Processed(Managed<ExpandedLogs>),
@@ -159,9 +161,43 @@ impl Forward for LogOutput {
 
         Ok(logs.map(|logs, _| logs.serialize_envelope()))
     }
+
+    #[cfg(feature = "processing")]
+    fn forward_store(
+        self,
+        s: &relay_system::Addr<crate::services::store::Store>,
+    ) -> Result<(), Rejected<()>> {
+        let logs = match self {
+            LogOutput::NotProcessed(logs) => {
+                return Err(logs.internal_error(
+                    "logs must be processed before they can be forwarded to the store",
+                ));
+            }
+            LogOutput::Processed(logs) => logs,
+        };
+
+        let scoping = logs.scoping();
+        let received_at = logs.received_at();
+
+        let (logs, retention) = logs.split_with_context(|logs| (logs.logs, logs.retention));
+        let ctx = store::Context {
+            scoping,
+            received_at,
+            retention,
+        };
+
+        for log in logs {
+            if let Ok(log) = log.try_map(|log, _| store::convert(log, &ctx)) {
+                s.send(log)
+            };
+        }
+
+        Ok(())
+    }
 }
 
 /// Logs in their serialized state, as transported in an envelope.
+#[derive(Debug)]
 pub struct SerializedLogs {
     /// Original envelope headers.
     headers: EnvelopeHeaders,
@@ -241,9 +277,12 @@ impl RateLimited for Managed<SerializedLogs> {
 }
 
 /// Logs which have been parsed and expanded from their serialized state.
+#[derive(Debug)]
 pub struct ExpandedLogs {
     /// Original envelope headers.
     headers: EnvelopeHeaders,
+    /// Retention in days.
+    retention: Option<u16>,
     /// Expanded and parsed logs.
     logs: ContainerItems<OurLog>,
 }

relay-server/src/processing/logs/process.rs

@@ -15,7 +15,7 @@ use crate::processing::logs::{Error, ExpandedLogs, Result, SerializedLogs};
 use crate::processing::{Context, Managed};
 use crate::services::outcome::DiscardReason;
 
-pub fn expand(logs: Managed<SerializedLogs>) -> Managed<ExpandedLogs> {
+pub fn expand(logs: Managed<SerializedLogs>, ctx: Context<'_>) -> Managed<ExpandedLogs> {
     let received_at = logs.received_at();
     logs.map(|logs, records| {
         let mut all_logs = SmallVec::with_capacity(logs.count());
@@ -38,6 +38,7 @@ pub fn expand(logs: Managed<SerializedLogs>) -> Managed<ExpandedLogs> {
 
         ExpandedLogs {
             headers: logs.headers,
+            retention: ctx.project_info.config.event_retention,
             logs: all_logs,
         }
     })

relay-server/src/processing/logs/store.rs

@@ -0,0 +1,110 @@
+use std::collections::HashMap;
+
+use chrono::{DateTime, Utc};
+use prost_types::Timestamp;
+use relay_event_schema::protocol::{Attributes, OurLog};
+use relay_protocol::{Annotated, Value};
+use relay_quotas::Scoping;
+use sentry_protos::snuba::v1::{AnyValue, TraceItem, TraceItemType, any_value};
+use uuid::Uuid;
+
+use crate::constants::DEFAULT_EVENT_RETENTION;
+use crate::processing::Counted;
+use crate::processing::logs::{Error, Result};
+use crate::services::outcome::DiscardReason;
+use crate::services::store::StoreLog;
+
+macro_rules! required {
+    ($value:expr) => {{
+        match $value {
+            Annotated(Some(value), _) => value,
+            Annotated(None, meta) => {
+                relay_log::debug!(
+                    "dropping log because of missing required field {} with meta {meta:?}",
+                    stringify!($value),
+                );
+                return Err(Error::Invalid(DiscardReason::InvalidLog));
+            }
+        }
+    }};
+}
+
+/// Context parameters for [`convert`].
+#[derive(Debug, Clone, Copy)]
+pub struct Context {
+    /// Received time.
+    pub received_at: DateTime<Utc>,
+    /// Item scoping.
+    pub scoping: Scoping,
+    /// Storage retention in days.
+    pub retention: Option<u16>,
+}
+
+pub fn convert(log: Annotated<OurLog>, ctx: &Context) -> Result<StoreLog> {
+    let quantities = log.quantities();
+
+    let log = required!(log);
+    let timestamp = required!(log.timestamp);
+    let attrs = log.attributes.0.unwrap_or_default();
+
+    let trace_item = TraceItem {
+        item_type: TraceItemType::Log.into(),
+        organization_id: ctx.scoping.organization_id.value(),
+        project_id: ctx.scoping.project_id.value(),
+        received: Some(ts(ctx.received_at)),
+        retention_days: ctx.retention.unwrap_or(DEFAULT_EVENT_RETENTION).into(),
+        timestamp: Some(ts(timestamp.0)),
+        trace_id: required!(log.trace_id).to_string(),
+        item_id: Uuid::new_v7(timestamp.into()).as_bytes().to_vec(),
+        attributes: attributes(attrs),
+        client_sample_rate: 1.0,
+        server_sample_rate: 1.0,
+    };
+
+    Ok(StoreLog {
+        trace_item,
+        quantities,
+    })
+}
+
+fn ts(dt: DateTime<Utc>) -> Timestamp {
+    Timestamp {
+        seconds: dt.timestamp(),
+        nanos: i32::try_from(dt.timestamp_subsec_nanos()).unwrap_or(0),
+    }
+}
+
+fn attributes(attributes: Attributes) -> HashMap<String, AnyValue> {
+    let mut result = HashMap::with_capacity(attributes.0.len());
+
+    for (name, attribute) in attributes {
+        let value = attribute
+            .into_value()
+            .and_then(|v| v.value.value.into_value());
+
+        let Some(value) = value else {
+            // Emit `_meta` attributes here with #4804.
+            continue;
+        };
+
+        let Some(value) = (match value {
+            Value::Bool(v) => Some(any_value::Value::BoolValue(v)),
+            Value::I64(v) => Some(any_value::Value::IntValue(v)),
+            Value::U64(v) => i64::try_from(v).ok().map(any_value::Value::IntValue),
+            Value::F64(v) => Some(any_value::Value::DoubleValue(v)),
+            Value::String(v) => Some(any_value::Value::StringValue(v)),
+            // These cases do not happen, as they are not valid attributes
+            // and they should have been filtered out before already.
+            Value::Array(_) | Value::Object(_) => {
+                debug_assert!(false, "unsupported log value");
+                None
+            }
+        }) else {
+            continue;
+        };
+
+        result.insert(name, AnyValue { value: Some(value) });
+    }
+
+    result
+}