feat(otlp): duplicate resource and instrumentation attrs into spans (#…

…4533)

OTLP trace payloads contain a nested structure wherein resources contain
instrumentation scopes, and instrumentation scopes contain spans. The
attributes set on the resource or instrumentation scope apply to all
contained spans, and are useful to query and aggregate on at the span
level. (For example, resource attributes conventionally contain the
service name and version of the service emitting spans).

When converting incoming OTLP traces into individual `OtelSpan` items,
denormalize these incoming extra attributes onto each span. This will
make the data available in Sentry for viewing, searching, and
aggregation. Prefix each attribute name with a prefix indicating its
source so that attributes with the same name from different sources
don't clobber each other.

(We might change our mind on whether these attributes should be prefixed
or not - that's fine as no one is using the OTLP endpoint or `OtlpSpan`
envelope item type yet).

In addition to an `attributes` dictionary, instrumentation scopes also
have `name` and `version` string properties. Copy those to spans too.

CHANGELOG.md

@@ -15,6 +15,7 @@
 - Track an utilization metric for internal services. ([#4501](https://github.com/getsentry/relay/pull/4501))
 - Add new `relay-threading` crate with asynchronous thread pool. ([#4500](https://github.com/getsentry/relay/pull/4500))
 - Expose additional metrics through the internal relay metric endpoint. ([#4511](https://github.com/getsentry/relay/pull/4511))
+- Write resource and instrumentation scope attributes as span attributes during OTLP ingestion. ([#4533](https://github.com/getsentry/relay/pull/4533))
 
 ## 25.2.0
 

relay-server/Cargo.toml

@@ -56,6 +56,7 @@ mime = { workspace = true }
 minidump = { workspace = true, optional = true }
 multer = { workspace = true }
 once_cell = { workspace = true }
+opentelemetry-proto = { workspace = true }
 papaya = { workspace = true }
 pin-project-lite = { workspace = true }
 priority-queue = { workspace = true }

relay-server/src/services/processor/span.rs

@@ -2,6 +2,8 @@
 
 use std::sync::Arc;
 
+use opentelemetry_proto::tonic::common::v1::any_value::Value;
+use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue};
 use prost::Message;
 use relay_dynamic_config::Feature;
 use relay_event_normalization::span::tag_extraction;
@@ -62,10 +64,43 @@ fn convert_traces_data(item: Item, managed_envelope: &mut TypedEnvelope<SpanGrou
             return;
         }
     };
-    for resource in traces_data.resource_spans {
-        for scope in resource.scope_spans {
-            for span in scope.spans {
-                // TODO: resources and scopes contain attributes, should denormalize into spans?
+    for resource_spans in traces_data.resource_spans {
+        for scope_spans in resource_spans.scope_spans {
+            for mut span in scope_spans.spans {
+                // Denormalize instrumentation scope and resource attributes into every span.
+                if let Some(ref scope) = scope_spans.scope {
+                    if !scope.name.is_empty() {
+                        span.attributes.push(KeyValue {
+                            key: "instrumentation.name".to_owned(),
+                            value: Some(AnyValue {
+                                value: Some(Value::StringValue(scope.name.clone())),
+                            }),
+                        })
+                    }
+                    if !scope.version.is_empty() {
+                        span.attributes.push(KeyValue {
+                            key: "instrumentation.version".to_owned(),
+                            value: Some(AnyValue {
+                                value: Some(Value::StringValue(scope.version.clone())),
+                            }),
+                        })
+                    }
+                    scope.attributes.iter().for_each(|a| {
+                        span.attributes.push(KeyValue {
+                            key: format!("instrumentation.{}", a.key),
+                            value: a.value.clone(),
+                        });
+                    });
+                }
+                if let Some(ref resource) = resource_spans.resource {
+                    resource.attributes.iter().for_each(|a| {
+                        span.attributes.push(KeyValue {
+                            key: format!("resource.{}", a.key),
+                            value: a.value.clone(),
+                        });
+                    });
+                }
+
                 let Ok(payload) = serde_json::to_vec(&span) else {
                     track_invalid(managed_envelope, DiscardReason::Internal);
                     continue;
@@ -119,3 +154,140 @@ pub fn extract_transaction_span(
 
     spans.into_iter().next().and_then(Annotated::into_value)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use super::*;
+    use crate::services::processor::ProcessingGroup;
+    use crate::utils::{ManagedEnvelope, TypedEnvelope};
+    use crate::Envelope;
+    use bytes::Bytes;
+    use relay_spans::otel_trace::Span as OtelSpan;
+    use relay_system::Addr;
+
+    #[test]
+    fn attribute_denormalization() {
+        // Construct an OTLP trace payload with:
+        // - a resource with one attribute, containing:
+        // - an instrumentation scope with one attribute, containing:
+        // - a span with one attribute
+        let traces_data = r#"
+        {
+            "resourceSpans": [
+                {
+                    "resource": {
+                        "attributes": [
+                            {
+                                "key": "resource_key",
+                                "value": {
+                                    "stringValue": "resource_value"
+                                }
+                            }
+                        ]
+                    },
+                    "scopeSpans": [
+                        {
+                            "scope": {
+                                "name": "test_instrumentation",
+                                "version": "0.0.1",
+                                "attributes": [
+                                    {
+                                        "key": "scope_key",
+                                        "value": {
+                                            "stringValue": "scope_value"
+                                        }
+                                    }
+                                ]
+                            },
+                            "spans": [
+                                {
+                                    "attributes": [
+                                        {
+                                            "key": "span_key",
+                                            "value": {
+                                                "stringValue": "span_value"
+                                            }
+                                        }
+                                    ]
+                                }
+                            ]
+                        }
+                    ]
+                }
+            ]
+        }
+        "#;
+
+        // Build an envelope containing the OTLP trace data.
+        let bytes =
+            Bytes::from(r#"{"dsn":"https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42"}"#);
+        let envelope = Envelope::parse_bytes(bytes).unwrap();
+        let (test_store, _) = Addr::custom();
+        let (outcome_aggregator, _) = Addr::custom();
+        let managed_envelope = ManagedEnvelope::new(
+            envelope,
+            outcome_aggregator,
+            test_store,
+            ProcessingGroup::Span,
+        );
+        let mut typed_envelope: TypedEnvelope<_> = managed_envelope.try_into().unwrap();
+        let mut item = Item::new(ItemType::OtelTracesData);
+        item.set_payload(ContentType::Json, traces_data);
+        typed_envelope.envelope_mut().add_item(item.clone());
+
+        // Convert the OTLP trace data into `OtelSpan` item(s).
+        convert_traces_data(item, &mut typed_envelope);
+
+        // Assert that the attributes from the resource and instrumentation
+        // scope were copied.
+        let item = typed_envelope
+            .envelope()
+            .items()
+            .find(|i| *i.ty() == ItemType::OtelSpan)
+            .expect("converted span missing from envelope");
+        let attributes = serde_json::from_slice::<OtelSpan>(&item.payload())
+            .expect("unable to deserialize otel span")
+            .attributes
+            .into_iter()
+            .map(|kv| (kv.key, kv.value.unwrap()))
+            .collect::<BTreeMap<_, _>>();
+        let attribute_value = |key: &str| -> String {
+            match attributes
+                .get(key)
+                .unwrap_or_else(|| panic!("attribute {} missing", key))
+                .to_owned()
+                .value
+            {
+                Some(Value::StringValue(str)) => str,
+                _ => panic!("attribute {} not a string", key),
+            }
+        };
+        assert_eq!(
+            attribute_value("span_key"),
+            "span_value".to_owned(),
+            "original span attribute should be present"
+        );
+        assert_eq!(
+            attribute_value("instrumentation.name"),
+            "test_instrumentation".to_owned(),
+            "instrumentation name should be in attributes"
+        );
+        assert_eq!(
+            attribute_value("instrumentation.version"),
+            "0.0.1".to_owned(),
+            "instrumentation version should be in attributes"
+        );
+        assert_eq!(
+            attribute_value("resource.resource_key"),
+            "resource_value".to_owned(),
+            "resource attribute should be copied with prefix"
+        );
+        assert_eq!(
+            attribute_value("instrumentation.scope_key"),
+            "scope_value".to_owned(),
+            "instruementation scope attribute should be copied with prefix"
+        );
+    }
+}