feat(eap): Add support for native scalar arrays

Author: Dav1dde

relay-event-normalization/src/eap/mod.rs

@@ -66,6 +66,12 @@ pub fn normalize_attribute_types(attributes: &mut Annotated<Attributes>) {
             (Annotated(Some(Double), _), Annotated(Some(Value::U64(_)), _)) => (),
             (Annotated(Some(Double), _), Annotated(Some(Value::F64(_)), _)) => (),
             (Annotated(Some(String), _), Annotated(Some(Value::String(_)), _)) => (),
+            (Annotated(Some(Array), _), Annotated(Some(Value::Array(arr)), _)) => {
+                if !is_supported_array(arr) {
+                    let _ = attribute.value_mut().take();
+                    attribute.meta_mut().add_error(ErrorKind::InvalidData);
+                }
+            }
             // Note: currently the mapping to Kafka requires that invalid or unknown combinations
             // of types and values are removed from the mapping.
             //
@@ -90,6 +96,43 @@ pub fn normalize_attribute_types(attributes: &mut Annotated<Attributes>) {
     }
 }
 
+/// Returns `true` if the passed array is an array we currently support.
+///
+/// Currently all arrays must be homogeneous types.
+fn is_supported_array(arr: &[Annotated<Value>]) -> bool {
+    let mut iter = arr.iter();
+
+    let Some(first) = iter.next() else {
+        // Empty arrays are supported.
+        return true;
+    };
+
+    let item = iter.try_fold(first, |prev, current| {
+        let r = match (prev.value(), current.value()) {
+            (None, None) => prev,
+            (None, Some(_)) => current,
+            (Some(_), None) => prev,
+            (Some(Value::String(_)), Some(Value::String(_))) => prev,
+            (Some(Value::Bool(_)), Some(Value::Bool(_))) => prev,
+            (
+                Some(Value::I64(_) | Value::U64(_) | Value::F64(_)),
+                Some(Value::I64(_) | Value::U64(_) | Value::F64(_)),
+            ) => prev,
+            // Everything else is unsupported.
+            //
+            // This includes nested arrays, nested objects and mixed arrays for now.
+            (Some(_), Some(_)) => return None,
+        };
+
+        Some(r)
+    });
+
+    matches!(
+        item.and_then(|v| v.value()),
+        Some(Value::String(_) | Value::Bool(_) | Value::I64(_) | Value::U64(_) | Value::F64(_))
+    )
+}
+
 /// Adds the `received` time to the attributes.
 pub fn normalize_received(attributes: &mut Annotated<Attributes>, received: DateTime<Utc>) {
     attributes
@@ -582,13 +625,33 @@ mod tests {
             },
             "missing_value": {
                 "type": "string"
+            },
+            "supported_array_string": {
+                "type": "array",
+                "value": ["foo", "bar"]
+            },
+            "supported_array_double": {
+                "type": "array",
+                "value": [3, 3.0, 3]
+            },
+            "unsupported_array_mixed": {
+                "type": "array",
+                "value": ["foo", 1.0]
+            },
+            "unsupported_array_object": {
+                "type": "array",
+                "value": [{}]
+            },
+            "unsupported_array_in_array": {
+                "type": "array",
+                "value": [[]]
             }
         }"#;
 
         let mut attributes = Annotated::<Attributes>::from_json(json).unwrap();
         normalize_attribute_types(&mut attributes);
 
-        insta::assert_json_snapshot!(SerializableAnnotated(&attributes), @r###"
+        insta::assert_json_snapshot!(SerializableAnnotated(&attributes), @r#"
         {
           "double_with_i64": {
             "type": "double",
@@ -597,7 +660,25 @@ mod tests {
           "invalid_int_from_invalid_string": null,
           "missing_type": null,
           "missing_value": null,
+          "supported_array_double": {
+            "type": "array",
+            "value": [
+              3,
+              3.0,
+              3
+            ]
+          },
+          "supported_array_string": {
+            "type": "array",
+            "value": [
+              "foo",
+              "bar"
+            ]
+          },
           "unknown_type": null,
+          "unsupported_array_in_array": null,
+          "unsupported_array_mixed": null,
+          "unsupported_array_object": null,
           "valid_bool": {
             "type": "boolean",
             "value": true
@@ -673,6 +754,27 @@ mod tests {
                 }
               }
             },
+            "unsupported_array_in_array": {
+              "": {
+                "err": [
+                  "invalid_data"
+                ]
+              }
+            },
+            "unsupported_array_mixed": {
+              "": {
+                "err": [
+                  "invalid_data"
+                ]
+              }
+            },
+            "unsupported_array_object": {
+              "": {
+                "err": [
+                  "invalid_data"
+                ]
+              }
+            },
             "valid_int_from_string": {
               "": {
                 "err": [
@@ -686,7 +788,7 @@ mod tests {
             }
           }
         }
-        "###);
+        "#);
     }
 
     #[test]

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

@@ -120,10 +120,29 @@ pub fn attribute_pii_from_conventions(state: &ProcessingState) -> Pii {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum AttributeType {
+    /// A boolean.
+    ///
+    /// The respective value must be of type [`Value::Bool`].
     Boolean,
+    /// An integer type.
+    ///
+    /// The respective value must be of type [`Value::I64`] or [`Value::U64`].
     Integer,
+    /// A floating point/double type.
+    ///
+    /// The respective value must be of type [`Value::F64`].
     Double,
+    /// A string type.
+    ///
+    /// The respective value must be of type [`Value::String`].
     String,
+    /// A string type.
+    ///
+    /// The respective value must be of type [`Value::Array`].
+    Array,
+    /// An unknown type.
+    ///
+    /// Kept for forward compatibility.
     Unknown(String),
 }
 
@@ -136,6 +155,7 @@ impl AttributeType {
             Self::Integer => "integer",
             Self::Double => "double",
             Self::String => "string",
+            Self::Array => "array",
             Self::Unknown(value) => value,
         }
     }
@@ -158,6 +178,7 @@ impl From<String> for AttributeType {
             "integer" => Self::Integer,
             "double" => Self::Double,
             "string" => Self::String,
+            "array" => Self::Array,
             _ => Self::Unknown(value),
         }
     }

relay-otel/src/lib.rs

@@ -37,31 +37,26 @@ pub fn otel_value_to_attribute(otel_value: OtelValue) -> Option<Attribute> {
             (AttributeType::String, Value::String(s))
         }
         OtelValue::ArrayValue(array) => {
-            // Filter out nested arrays and key-value lists for safety.
-            // This is not usually allowed by the OTLP protocol, but we filter
-            // these values out before serializing for robustness.
-            let safe_values: Vec<serde_json::Value> = array
+            let values: Vec<Annotated<Value>> = array
                 .values
                 .into_iter()
-                .filter_map(|v| match v.value? {
-                    OtelValue::StringValue(s) => Some(serde_json::Value::String(s)),
-                    OtelValue::BoolValue(b) => Some(serde_json::Value::Bool(b)),
-                    OtelValue::IntValue(i) => {
-                        Some(serde_json::Value::Number(serde_json::Number::from(i)))
-                    }
-                    OtelValue::DoubleValue(d) => {
-                        serde_json::Number::from_f64(d).map(serde_json::Value::Number)
-                    }
-                    OtelValue::BytesValue(bytes) => {
-                        String::from_utf8(bytes).ok().map(serde_json::Value::String)
-                    }
-                    // Skip nested complex types for safety
-                    OtelValue::ArrayValue(_) | OtelValue::KvlistValue(_) => None,
+                .filter_map(|v| {
+                    Some(match v.value? {
+                        OtelValue::StringValue(s) => Value::String(s),
+                        OtelValue::BoolValue(b) => Value::Bool(b),
+                        OtelValue::IntValue(i) => Value::I64(i),
+                        OtelValue::DoubleValue(d) => Value::F64(d),
+                        OtelValue::BytesValue(bytes) => {
+                            Value::String(String::from_utf8(bytes).ok()?)
+                        }
+                        // Currently not supported.
+                        OtelValue::ArrayValue(_) | OtelValue::KvlistValue(_) => return None,
+                    })
                 })
+                .map(Annotated::new)
                 .collect();
 
-            let json = serde_json::to_string(&safe_values).ok()?;
-            (AttributeType::String, Value::String(json))
+            (AttributeType::Array, Value::Array(values))
         }
         OtelValue::KvlistValue(kvlist) => {
             // Convert key-value list to JSON object and serialize as string.

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

@@ -2,15 +2,15 @@ use std::collections::HashMap;
 
 use chrono::{DateTime, Utc};
 use relay_event_schema::protocol::{Attributes, OurLog, OurLogLevel, SpanId};
-use relay_protocol::{Annotated, IntoValue, Value};
+use relay_protocol::Annotated;
 use relay_quotas::Scoping;
 use sentry_protos::snuba::v1::{AnyValue, TraceItem, TraceItemType, any_value};
 use uuid::Uuid;
 
 use crate::envelope::WithHeader;
 use crate::processing::logs::{Error, Result};
-use crate::processing::utils::store::{AttributeMeta, extract_meta_attributes, proto_timestamp};
-use crate::processing::{Counted, Retention};
+use crate::processing::utils::store::{extract_meta_attributes, proto_timestamp};
+use crate::processing::{self, Counted, Retention};
 use crate::services::outcome::DiscardReason;
 use crate::services::store::StoreTraceItem;
 
@@ -114,42 +114,7 @@ fn attributes(
     // +N, one for each field attribute added and some extra for potential meta.
     result.reserve(attributes.0.len() + 5 + 3);
 
-    for (name, attribute) in attributes {
-        let meta = AttributeMeta {
-            meta: IntoValue::extract_meta_tree(&attribute),
-        };
-        if let Some(meta) = meta.to_any_value() {
-            result.insert(format!("sentry._meta.fields.attributes.{name}"), meta);
-        }
-
-        let value = attribute
-            .into_value()
-            .and_then(|v| v.value.value.into_value());
-
-        let Some(value) = value else {
-            // Meta has already been handled, no value -> skip.
-            // There are also no current plans to handle `null` in EAP.
-            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) });
-    }
+    processing::utils::store::convert_attributes_into(&mut result, attributes);
 
     let FieldAttributes {
         level,

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

@@ -4,17 +4,15 @@ use chrono::{DateTime, Utc};
 use prost_types::Timestamp;
 use relay_base_schema::metrics::MetricUnit;
 use relay_event_schema::protocol::{Attributes, MetricType, SpanId, TraceMetric};
-use relay_protocol::{Annotated, IntoValue, Value};
+use relay_protocol::{Annotated, Value};
 use relay_quotas::Scoping;
 use sentry_protos::snuba::v1::{AnyValue, TraceItem, TraceItemType, any_value};
 use uuid::Uuid;
 
 use crate::envelope::WithHeader;
 use crate::processing::trace_metrics::{Error, Result};
-use crate::processing::utils::store::{
-    AttributeMeta, extract_client_sample_rate, extract_meta_attributes,
-};
-use crate::processing::{Counted, Retention};
+use crate::processing::utils::store::{extract_client_sample_rate, extract_meta_attributes};
+use crate::processing::{self, Counted, Retention};
 use crate::services::outcome::DiscardReason;
 use crate::services::store::StoreTraceItem;
 
@@ -114,40 +112,9 @@ fn attributes(
     fields: FieldAttributes,
 ) -> HashMap<String, AnyValue> {
     let mut result = meta;
-    result.reserve(attributes.0.len() + 5);
+    result.reserve(attributes.0.len() + 15);
 
-    for (name, attribute) in attributes {
-        let meta = AttributeMeta {
-            meta: IntoValue::extract_meta_tree(&attribute),
-        };
-        if let Some(meta) = meta.to_any_value() {
-            result.insert(format!("sentry._meta.fields.attributes.{name}"), meta);
-        }
-
-        let value = attribute
-            .into_value()
-            .and_then(|v| v.value.value.into_value());
-
-        let Some(value) = value else {
-            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)),
-            Value::Array(_) | Value::Object(_) => {
-                debug_assert!(false, "unsupported trace metric value");
-                None
-            }
-        }) else {
-            continue;
-        };
-
-        result.insert(name, AnyValue { value: Some(value) });
-    }
+    processing::utils::store::convert_attributes_into(&mut result, attributes);
 
     let FieldAttributes {
         metric_name,