Process and add span pointers to the Lambda execution span's meta under _dd.span_links

Author: nhulston

bottlecap/src/lifecycle/invocation/processor.rs

@@ -359,6 +359,7 @@ impl Processor {
                 header_tags,
                 vec![traces],
                 body_size,
+                Some(self.inferrer.span_pointers.clone()),
             );
 
             if let Err(e) = trace_agent_tx.send(send_data).await {

bottlecap/src/traces/trace_agent.rs

@@ -258,6 +258,7 @@ impl TraceAgent {
             tracer_header_tags,
             traces,
             body_size,
+            None,
         );
 
         // send trace payload to our trace flusher

bottlecap/src/traces/trace_processor.rs

@@ -21,6 +21,8 @@ use datadog_trace_obfuscation::obfuscate::obfuscate_span;
 use datadog_trace_protobuf::pb::Span;
 use datadog_trace_utils::trace_utils::SendData;
 use datadog_trace_utils::trace_utils::{self};
+use serde_json::{json, Value};
+use crate::span_pointers::SpanPointer;
 
 #[derive(Clone)]
 #[allow(clippy::module_name_repetitions)]
@@ -32,6 +34,7 @@ pub struct ServerlessTraceProcessor {
 struct ChunkProcessor {
     obfuscation_config: Arc<obfuscation_config::ObfuscationConfig>,
     tags_provider: Arc<provider::Provider>,
+    span_pointers: Option<Vec<SpanPointer>>,
 }
 
 impl TraceChunkProcessor for ChunkProcessor {
@@ -48,6 +51,32 @@ impl TraceChunkProcessor for ChunkProcessor {
                 }
             }
 
+            if span.name == "aws.lambda" {
+                if let Some(span_pointers) = &self.span_pointers {
+                    if !span_pointers.is_empty() {
+                        let span_links: Vec<Value> = span_pointers
+                            .iter()
+                            .map(|sp| {
+                                json!({
+                            "attributes": {
+                                "link.kind": "span-pointer",
+                                "ptr.dir": "u",
+                                "ptr.hash": sp.hash,
+                                "ptr.kind": sp.kind,
+                            },
+                            "span_id": "0",
+                            "trace_id": "0"
+                        })
+                            })
+                            .collect();
+
+                        if let Ok(span_links_json) = serde_json::to_string(&span_links) {
+                            span.meta.insert("_dd.span_links".to_string(), span_links_json);
+                        }
+                    }
+                }
+            }
+
             self.tags_provider.get_tags_map().iter().for_each(|(k, v)| {
                 span.meta.insert(k.clone(), v.clone());
             });
@@ -116,6 +145,7 @@ pub trait TraceProcessor {
         header_tags: tracer_header_tags::TracerHeaderTags,
         traces: Vec<Vec<pb::Span>>,
         body_size: usize,
+        span_pointers: Option<Vec<SpanPointer>>,
     ) -> SendData;
 }
 
@@ -127,13 +157,15 @@ impl TraceProcessor for ServerlessTraceProcessor {
         header_tags: tracer_header_tags::TracerHeaderTags,
         traces: Vec<Vec<pb::Span>>,
         body_size: usize,
+        span_pointers: Option<Vec<SpanPointer>>,
     ) -> SendData {
         let payload = trace_utils::collect_trace_chunks(
             V07(traces),
             &header_tags,
             &mut ChunkProcessor {
                 obfuscation_config: self.obfuscation_config.clone(),
                 tags_provider: tags_provider.clone(),
+                span_pointers,
             },
             true,
         );
@@ -270,7 +302,7 @@ mod tests {
         let config = create_test_config();
         let tags_provider = create_tags_provider(config.clone());
         let tracer_payload =
-            trace_processor.process_traces(config, tags_provider.clone(), header_tags, traces, 100);
+            trace_processor.process_traces(config, tags_provider.clone(), header_tags, traces, 100, None);
 
         let expected_tracer_payload = pb::TracerPayload {
             container_id: "33".to_string(),