feat(ourlogs): Allow log ingestion behind a flag

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## Unreleased
+- Add data categories for LogItem and LogByte. ([#4448](https://github.com/getsentry/relay/pull/4455))
+
 ## 25.1.0
 
 **Internal**

Cargo.toml

@@ -151,7 +151,7 @@ reqwest = "0.12.9"
 rmp-serde = "1.3.0"
 sentry = "0.34.0"
 sentry-core = "0.34.0"
-sentry-kafka-schemas = { version = "0.1.122", default-features = false }
+sentry-kafka-schemas = { version = "0.1.129", default-features = false }
 sentry-release-parser = { version = "1.3.2", default-features = false }
 sentry-types = "0.34.0"
 semver = "1.0.23"

py/sentry_relay/consts.py

@@ -31,8 +31,8 @@ class DataCategory(IntEnum):
     REPLAY_VIDEO = 20
     UPTIME = 21
     ATTACHMENT_ITEM = 22
-    LOG_COUNT = 23
-    LOG_BYTES = 24
+    LOG_ITEM = 23
+    LOG_BYTE = 24
     UNKNOWN = -1
     # end generated
 

relay-base-schema/src/data_category.rs

@@ -92,15 +92,15 @@ pub enum DataCategory {
     Uptime = 21,
     /// Counts the number of individual attachments, as opposed to the number of bytes in an attachment.
     AttachmentItem = 22,
-    /// LogCount
+    /// LogItem
     ///
     /// This is the category for logs for which we store the count log events for users for measuring
     /// missing breadcrumbs, and count of logs for rate limiting purposes.
-    LogCount = 23,
-    /// LogBytes
+    LogItem = 23,
+    /// LogByte
     ///
     /// This is the category for logs for which we store log event total bytes for users.
-    LogBytes = 24,
+    LogByte = 24,
     //
     // IMPORTANT: After adding a new entry to DataCategory, go to the `relay-cabi` subfolder and run
     // `make header` to regenerate the C-binding. This allows using the data category from Python.
@@ -129,8 +129,8 @@ impl DataCategory {
             "transaction_indexed" => Self::TransactionIndexed,
             "monitor" => Self::Monitor,
             "span" => Self::Span,
-            "log_count" => Self::LogCount,
-            "log_bytes" => Self::LogBytes,
+            "log_item" => Self::LogItem,
+            "log_byte" => Self::LogByte,
             "monitor_seat" => Self::MonitorSeat,
             "feedback" => Self::UserReportV2,
             "user_report_v2" => Self::UserReportV2,
@@ -163,8 +163,8 @@ impl DataCategory {
             Self::TransactionIndexed => "transaction_indexed",
             Self::Monitor => "monitor",
             Self::Span => "span",
-            Self::LogCount => "log_count",
-            Self::LogBytes => "log_bytes",
+            Self::LogItem => "log_item",
+            Self::LogByte => "log_byte",
             Self::MonitorSeat => "monitor_seat",
             Self::UserReportV2 => "feedback",
             Self::MetricBucket => "metric_bucket",

relay-cabi/include/relay.h

@@ -148,13 +148,13 @@ enum RelayDataCategory {
    * This is the category for logs for which we store the count log events for users for measuring
    * missing breadcrumbs, and count of logs for rate limiting purposes.
    */
-  RELAY_DATA_CATEGORY_LOG_COUNT = 23,
+  RELAY_DATA_CATEGORY_LOG_ITEM = 23,
   /**
    * LogBytes
    *
    * This is the category for logs for which we store log event total bytes for users.
    */
-  RELAY_DATA_CATEGORY_LOG_BYTES = 24,
+  RELAY_DATA_CATEGORY_LOG_BYTE = 24,
   /**
    * Any other data category not known by this Relay.
    */

relay-event-schema/src/processor/traits.rs

@@ -113,6 +113,7 @@ pub trait Processor: Sized {
     process_method!(process_trace_context, crate::protocol::TraceContext);
     process_method!(process_native_image_path, crate::protocol::NativeImagePath);
     process_method!(process_contexts, crate::protocol::Contexts);
+    process_method!(process_attribute_value, crate::protocol::AttributeValue);
 
     fn process_other(
         &mut self,

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

@@ -1,4 +1,8 @@
-use relay_protocol::{Annotated, Empty, FromValue, IntoValue, Object, Value};
+use relay_protocol::{
+    Annotated, Empty, Error, FromValue, IntoValue, Object, SkipSerialization, Value,
+};
+
+use serde::ser::SerializeMap;
 
 use crate::processor::ProcessValue;
 use crate::protocol::{SpanId, TraceId};
@@ -27,32 +31,161 @@ pub struct OurLog {
     pub trace_flags: Annotated<f64>,
 
     /// This is the original string representation of the severity as it is known at the source
-    #[metastructure(required = false, max_chars = 32, pii = "maybe", trim = false)]
+    #[metastructure(required = false, max_chars = 32, pii = "true", trim = false)]
     pub severity_text: Annotated<String>,
 
     /// Numerical representation of the severity level
     #[metastructure(required = false)]
     pub severity_number: Annotated<i64>,
 
     /// Log body.
-    #[metastructure(required = true, pii = "maybe", trim = false)]
+    #[metastructure(required = true, pii = "true", trim = false)]
     pub body: Annotated<String>,
 
     /// Arbitrary attributes on a log.
-    #[metastructure(pii = "maybe", trim = false)]
+    #[metastructure(pii = "true", trim = false)]
     pub attributes: Annotated<Object<AttributeValue>>,
 
     /// Additional arbitrary fields for forwards compatibility.
     #[metastructure(additional_properties, retain = true, pii = "maybe", trim = false)]
     pub other: Object<Value>,
 }
 
-#[derive(Debug, Clone, Default, PartialEq, Empty, FromValue, IntoValue, ProcessValue)]
-pub struct AttributeValue {
-    pub string_value: Annotated<Value>,
-    pub int_value: Annotated<Value>,
-    pub double_value: Annotated<Value>,
-    pub bool_value: Annotated<Value>,
+#[derive(Debug, Clone, PartialEq, ProcessValue)]
+pub enum AttributeValue {
+    #[metastructure(field = "string_value", pii = "true")]
+    StringValue(String),
+    #[metastructure(field = "int_value", pii = "true")]
+    IntValue(i64),
+    #[metastructure(field = "double_value", pii = "true")]
+    DoubleValue(f64),
+    #[metastructure(field = "bool_value", pii = "true")]
+    BoolValue(bool),
+    /// Any other unknown attribute value.
+    ///
+    /// This exists to ensure other attribute values such as array and object can be added in the future.
+    Unknown(String),
+}
+
+impl IntoValue for AttributeValue {
+    fn into_value(self) -> Value {
+        let mut map = Object::new();
+        match self {
+            AttributeValue::StringValue(v) => {
+                map.insert("string_value".to_string(), Annotated::new(Value::String(v)));
+            }
+            AttributeValue::IntValue(v) => {
+                map.insert("int_value".to_string(), Annotated::new(Value::I64(v)));
+            }
+            AttributeValue::DoubleValue(v) => {
+                map.insert("double_value".to_string(), Annotated::new(Value::F64(v)));
+            }
+            AttributeValue::BoolValue(v) => {
+                map.insert("bool_value".to_string(), Annotated::new(Value::Bool(v)));
+            }
+            AttributeValue::Unknown(v) => {
+                map.insert("unknown".to_string(), Annotated::new(Value::String(v)));
+            }
+        }
+        Value::Object(map)
+    }
+
+    fn serialize_payload<S>(&self, s: S, _behavior: SkipSerialization) -> Result<S::Ok, S::Error>
+    where
+        Self: Sized,
+        S: serde::Serializer,
+    {
+        let mut map = s.serialize_map(None)?;
+        match self {
+            AttributeValue::StringValue(v) => {
+                map.serialize_entry("string_value", v)?;
+            }
+            AttributeValue::IntValue(v) => {
+                map.serialize_entry("int_value", v)?;
+            }
+            AttributeValue::DoubleValue(v) => {
+                map.serialize_entry("double_value", v)?;
+            }
+            AttributeValue::BoolValue(v) => {
+                map.serialize_entry("bool_value", v)?;
+            }
+            AttributeValue::Unknown(v) => {
+                map.serialize_entry("unknown", v)?;
+            }
+        }
+        map.end()
+    }
+}
+
+impl AttributeValue {
+    /// Returns the string representation of this attribute value.
+    pub fn as_str(&self) -> String {
+        match self {
+            AttributeValue::StringValue(s) => s.clone(),
+            AttributeValue::IntValue(s) => s.to_string(),
+            AttributeValue::DoubleValue(s) => s.to_string(),
+            AttributeValue::BoolValue(s) => s.to_string(),
+            AttributeValue::Unknown(s) => s.clone(),
+        }
+    }
+    pub fn string_value(&self) -> Option<&String> {
+        match self {
+            AttributeValue::StringValue(s) => Some(s),
+            _ => None,
+        }
+    }
+    pub fn int_value(&self) -> Option<i64> {
+        match self {
+            AttributeValue::IntValue(i) => Some(*i),
+            _ => None,
+        }
+    }
+    pub fn double_value(&self) -> Option<f64> {
+        match self {
+            AttributeValue::DoubleValue(d) => Some(*d),
+            _ => None,
+        }
+    }
+    pub fn bool_value(&self) -> Option<bool> {
+        match self {
+            AttributeValue::BoolValue(b) => Some(*b),
+            _ => None,
+        }
+    }
+}
+
+impl Empty for AttributeValue {
+    #[inline]
+    fn is_empty(&self) -> bool {
+        matches!(self, Self::Unknown(_))
+    }
+}
+
+impl FromValue for AttributeValue {
+    fn from_value(value: Annotated<Value>) -> Annotated<Self> {
+        match value {
+            Annotated(Some(Value::String(value)), meta) => {
+                Annotated(Some(AttributeValue::StringValue(value)), meta)
+            }
+            Annotated(Some(Value::I64(value)), meta) => {
+                Annotated(Some(AttributeValue::IntValue(value)), meta)
+            }
+            Annotated(Some(Value::F64(value)), meta) => {
+                Annotated(Some(AttributeValue::DoubleValue(value)), meta)
+            }
+            Annotated(Some(Value::Bool(value)), meta) => {
+                Annotated(Some(AttributeValue::BoolValue(value)), meta)
+            }
+            Annotated(Some(value), mut meta) => {
+                meta.add_error(Error::expected(
+                    "a valid attribute value (string, int, double, bool)",
+                ));
+                meta.set_original_value(Some(value));
+                Annotated(None, meta)
+            }
+            Annotated(None, meta) => Annotated(None, meta),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -64,55 +197,43 @@ mod tests {
         let json = r#"{
   "timestamp_nanos": 1544712660300000000,
   "observed_timestamp_nanos": 1544712660300000000,
-  "severity_number": 10,
-  "severity_text": "Information",
   "trace_id": "5b8efff798038103d269b633813fc60c",
   "span_id": "eee19b7ec3c1b174",
+  "severity_text": "Information",
+  "severity_number": 10,
   "body": "Example log record",
   "attributes": {
-    "string.attribute": {
-      "string_value": "some string"
-    },
     "boolean.attribute": {
       "bool_value": true
     },
+    "double.attribute": {
+      "double_value": 637.704
+    },
     "int.attribute": {
       "int_value": 10
     },
-    "double.attribute": {
-      "double_value": 637.704
+    "string.attribute": {
+      "string_value": "some string"
     }
   }
 }"#;
 
         let mut attributes = Object::new();
         attributes.insert(
             "string.attribute".into(),
-            Annotated::new(AttributeValue {
-                string_value: Annotated::new(Value::String("some string".into())),
-                ..Default::default()
-            }),
+            Annotated::new(AttributeValue::StringValue("some string".into())),
         );
         attributes.insert(
             "boolean.attribute".into(),
-            Annotated::new(AttributeValue {
-                bool_value: Annotated::new(Value::Bool(true)),
-                ..Default::default()
-            }),
+            Annotated::new(AttributeValue::BoolValue(true)),
         );
         attributes.insert(
             "int.attribute".into(),
-            Annotated::new(AttributeValue {
-                int_value: Annotated::new(Value::I64(10)),
-                ..Default::default()
-            }),
+            Annotated::new(AttributeValue::IntValue(10)),
         );
         attributes.insert(
             "double.attribute".into(),
-            Annotated::new(AttributeValue {
-                double_value: Annotated::new(Value::F64(637.704)),
-                ..Default::default()
-            }),
+            Annotated::new(AttributeValue::DoubleValue(637.704)),
         );
 
         let log = Annotated::new(OurLog {
@@ -127,12 +248,6 @@ mod tests {
             ..Default::default()
         });
 
-        let expected: serde_json::Value = serde_json::from_str(json).unwrap();
-        let actual: serde_json::Value =
-            serde_json::from_str(&log.to_json_pretty().unwrap()).unwrap();
-        assert_eq!(expected, actual);
-
-        let log_from_string = Annotated::<OurLog>::from_json(json).unwrap();
-        assert_eq!(log, log_from_string);
+        assert_eq!(json, log.to_json_pretty().unwrap());
     }
 }

relay-ourlogs/src/lib.rs

@@ -1,4 +1,4 @@
-//! Structs and functions needed to ingest OpenTelemetry spans.
+//! Structs and functions needed to ingest OpenTelemetry logs.
 
 #![warn(missing_docs)]
 #![doc(