feat(otlp): process incoming span links (#4590)

Take incoming span links and convert them into Sentry's span links
(introduced in #4486), attaching them to the Sentry span.

Note that these links are not currently being written to the
`SpanKafkaMessage` struct, so they don't actually affect anything
downstream yet.

Author: mjq

relay-spans/src/span.rs

@@ -1,16 +1,18 @@
+use std::collections::BTreeMap;
 use std::str::FromStr;
 
 use chrono::{TimeZone, Utc};
 use opentelemetry_proto::tonic::common::v1::any_value::Value as OtelValue;
+use opentelemetry_proto::tonic::trace::v1::span::Link as OtelLink;
 
 use crate::otel_trace::{
     status::StatusCode as OtelStatusCode, Span as OtelSpan, SpanFlags as OtelSpanFlags,
 };
 use crate::status_codes;
 use relay_event_schema::protocol::{
-    EventId, Span as EventSpan, SpanData, SpanId, SpanStatus, Timestamp, TraceId,
+    EventId, Span as EventSpan, SpanData, SpanId, SpanLink, SpanStatus, Timestamp, TraceId,
 };
-use relay_protocol::{Annotated, FromValue, Object};
+use relay_protocol::{Annotated, FromValue, Object, Value};
 
 /// convert_from_otel_to_sentry_status returns a status as defined by Sentry based on the OTel status.
 fn convert_from_otel_to_sentry_status(
@@ -96,6 +98,7 @@ pub fn otel_to_sentry_span(otel_span: OtelSpan) -> EventSpan {
         kind,
         attributes,
         status,
+        links,
         ..
     } = otel_span;
 
@@ -165,26 +168,8 @@ pub fn otel_to_sentry_span(otel_span: OtelSpan) -> EventSpan {
                 }
                 _ => {
                     let key = attribute.key;
-                    match value {
-                        OtelValue::ArrayValue(_) => {}
-                        OtelValue::BoolValue(v) => {
-                            data.insert(key, Annotated::new(v.into()));
-                        }
-                        OtelValue::BytesValue(v) => {
-                            if let Ok(v) = String::from_utf8(v) {
-                                data.insert(key, Annotated::new(v.into()));
-                            }
-                        }
-                        OtelValue::DoubleValue(v) => {
-                            data.insert(key, Annotated::new(v.into()));
-                        }
-                        OtelValue::IntValue(v) => {
-                            data.insert(key, Annotated::new(v.into()));
-                        }
-                        OtelValue::KvlistValue(_) => {}
-                        OtelValue::StringValue(v) => {
-                            data.insert(key, Annotated::new(v.into()));
-                        }
+                    if let Some(v) = otel_to_sentry_value(value) {
+                        data.insert(key, Annotated::new(v));
                     }
                 }
             }
@@ -199,6 +184,11 @@ pub fn otel_to_sentry_span(otel_span: OtelSpan) -> EventSpan {
         description = description.or(Some(format!("{http_method} {http_route}")));
     }
 
+    let sentry_links: Vec<Annotated<SpanLink>> = links
+        .into_iter()
+        .map(|link| otel_to_sentry_link(link).into())
+        .collect();
+
     EventSpan {
         op: op.into(),
         description: description.into(),
@@ -221,10 +211,49 @@ pub fn otel_to_sentry_span(otel_span: OtelSpan) -> EventSpan {
         timestamp: Timestamp(end_timestamp).into(),
         trace_id: TraceId(trace_id).into(),
         platform: platform.into(),
+        links: sentry_links.into(),
         ..Default::default()
     }
 }
 
+fn otel_to_sentry_link(otel_link: OtelLink) -> SpanLink {
+    // See the W3C trace context specification:
+    // <https://www.w3.org/TR/trace-context-2/#sampled-flag>
+    const W3C_TRACE_CONTEXT_SAMPLED: u32 = 1 << 0;
+
+    let attributes = BTreeMap::from_iter(otel_link.attributes.into_iter().flat_map(|kv| {
+        kv.value
+            .and_then(|v| v.value)
+            .and_then(otel_to_sentry_value)
+            .map(|v| (kv.key, v.into()))
+    }))
+    .into();
+
+    SpanLink {
+        trace_id: TraceId(hex::encode(otel_link.trace_id)).into(),
+        span_id: SpanId(hex::encode(otel_link.span_id)).into(),
+        sampled: (otel_link.flags & W3C_TRACE_CONTEXT_SAMPLED != 0).into(),
+        attributes,
+        // The parent span ID is not available over OTLP.
+        parent_span_id: Annotated::empty(),
+        other: Default::default(),
+    }
+}
+
+fn otel_to_sentry_value(value: OtelValue) -> Option<Value> {
+    match value {
+        OtelValue::BoolValue(v) => Some(Value::Bool(v)),
+        OtelValue::DoubleValue(v) => Some(Value::F64(v)),
+        OtelValue::IntValue(v) => Some(Value::I64(v)),
+        OtelValue::StringValue(v) => Some(Value::String(v)),
+        OtelValue::BytesValue(v) => {
+            String::from_utf8(v).map_or(None, |str| Some(Value::String(str)))
+        }
+        OtelValue::ArrayValue(_) => None,
+        OtelValue::KvlistValue(_) => None,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -707,7 +736,7 @@ mod tests {
                     ),
                 },
             },
-            links: ~,
+            links: [],
             sentry_tags: ~,
             received: ~,
             measurements: ~,
@@ -756,4 +785,76 @@ mod tests {
             Some("fa90fdead5f74052")
         );
     }
+
+    #[test]
+    fn parse_link() {
+        let json = r#"{
+            "links": [
+                {
+                    "traceId": "4c79f60c11214eb38604f4ae0781bfb2",
+                    "spanId": "fa90fdead5f74052",
+                    "attributes": [
+                        {
+                            "key": "str_key",
+                            "value": {
+                                "stringValue": "str_value"
+                            }
+                        },
+                        {
+                            "key": "bool_key",
+                            "value": {
+                                "boolValue": true
+                            }
+                        },
+                        {
+                            "key": "int_key",
+                            "value": {
+                                "intValue": "123"
+                            }
+                        },
+                        {
+                            "key": "double_key",
+                            "value": {
+                                "doubleValue": 1.23
+                            }
+                        }
+                    ],
+                    "flags": 1
+                }
+            ]
+        }"#;
+        let otel_span: OtelSpan = serde_json::from_str(json).unwrap();
+        let event_span: EventSpan = otel_to_sentry_span(otel_span);
+        let annotated_span: Annotated<EventSpan> = Annotated::new(event_span);
+        assert_eq!(
+            get_path!(annotated_span.links[0].trace_id),
+            Some(&Annotated::new(TraceId(
+                "4c79f60c11214eb38604f4ae0781bfb2".into()
+            )))
+        );
+        assert_eq!(
+            get_path!(annotated_span.links[0].span_id),
+            Some(&Annotated::new(SpanId("fa90fdead5f74052".into())))
+        );
+        assert_eq!(
+            get_path!(annotated_span.links[0].attributes["str_key"]),
+            Some(&Annotated::new(Value::String("str_value".into())))
+        );
+        assert_eq!(
+            get_path!(annotated_span.links[0].attributes["bool_key"]),
+            Some(&Annotated::new(Value::Bool(true)))
+        );
+        assert_eq!(
+            get_path!(annotated_span.links[0].attributes["int_key"]),
+            Some(&Annotated::new(Value::I64(123)))
+        );
+        assert_eq!(
+            get_path!(annotated_span.links[0].attributes["double_key"]),
+            Some(&Annotated::new(Value::F64(1.23)))
+        );
+        assert_eq!(
+            get_path!(annotated_span.links[0].sampled),
+            Some(&Annotated::new(true))
+        );
+    }
 }