feat(spans): Add a span for the transaction itself, add Segment ID (#…

…2417)

In #2350 we set up the extraction
of spans from transactions & publishing on a dedicated spans topic.

This PR contains two fixes:

* The initial PR did not emit a span for the transaction itself, which
is conceptually also a span.
* Add a `segment_id` (= `span_id` of the transaction) and an
`is_segment` flag to every span.

relay-general/src/protocol/span.rs

@@ -1,5 +1,6 @@
 use crate::protocol::{
- JsonLenientString, OperationType, OriginType, SpanId, SpanStatus, Timestamp, TraceId,
+ Event, JsonLenientString, OperationType, OriginType, SpanId, SpanStatus, Timestamp,
+ TraceContext, TraceId,
 };
 use crate::types::{Annotated, Object, Value};
 
@@ -38,7 +39,16 @@ pub struct Span {
  #[metastructure(required = "true")]
  pub trace_id: Annotated<TraceId>,
 
- /// The status of a span
+ /// A unique identifier for a segment within a trace (8 byte hexadecimal string).
+ ///
+ /// For spans embedded in transactions, the `segment_id` is the `span_id` of the containing
+ /// transaction.
+ pub segment_id: Annotated<SpanId>,
+
+ /// Whether or not the current span is the root of the segment.
+ pub is_segment: Annotated<bool>,
+
+ /// The status of a span.
  pub status: Annotated<SpanStatus>,
 
  /// Arbitrary tags on a span, like on the top-level event.
@@ -59,6 +69,29 @@ pub struct Span {
  pub other: Object<Value>,
 }
 
+impl From<&Event> for Span {
+ fn from(event: &Event) -> Self {
+ let mut span = Self {
+ timestamp: event.timestamp.clone(),
+ start_timestamp: event.start_timestamp.clone(),
+ is_segment: Some(true).into(),
+ ..Default::default()
+ };
+
+ if let Some(trace_context) = event.context::<TraceContext>().cloned() {
+ span.exclusive_time = trace_context.exclusive_time;
+ span.op = trace_context.op;
+ span.span_id = trace_context.span_id;
+ span.parent_span_id = trace_context.parent_span_id;
+ span.trace_id = trace_context.trace_id;
+ span.segment_id = span.span_id.clone(); // a transaction is a segment
+ span.status = trace_context.status;
+ }
+
+ span
+ }
+}
+
 #[cfg(test)]
 mod tests {
  use chrono::{TimeZone, Utc};

relay-general/src/store/normalize.rs

@@ -3316,6 +3316,8 @@ mod tests {
  span_id: ~,
  parent_span_id: ~,
  trace_id: ~,
+ segment_id: ~,
+ is_segment: ~,
  status: ~,
  tags: ~,
  origin: ~,
@@ -3352,6 +3354,8 @@ mod tests {
  span_id: ~,
  parent_span_id: ~,
  trace_id: ~,
+ segment_id: ~,
+ is_segment: ~,
  status: ~,
  tags: ~,
  origin: ~,
@@ -3388,6 +3392,8 @@ mod tests {
  span_id: ~,
  parent_span_id: ~,
  trace_id: ~,
+ segment_id: ~,
+ is_segment: ~,
  status: ~,
  tags: ~,
  origin: ~,

relay-server/src/actors/processor.rs

@@ -2233,6 +2233,8 @@ impl EnvelopeProcessorService {
  #[cfg(feature = "processing")]
  fn extract_spans(&self, state: &mut ProcessEnvelopeState) {
  // For now, drop any spans submitted by the SDK.
+
+ use relay_general::protocol::Span;
  state.managed_envelope.retain_items(|item| match item.ty() {
  ItemType::Span => ItemAction::DropSilently,
  _ => ItemAction::Keep,
@@ -2251,20 +2253,38 @@ impl EnvelopeProcessorService {
  return;
  };
 
- // Extract.
- let Some(spans) = state.event.value().and_then(|e| e.spans.value()) else { return };
- for span in spans {
+ // Extract transaction as a span.
+ let Some(event) = state.event.value() else { return };
+ let transaction_span: Span = event.into();
+
+ let mut add_span = |span: Annotated<Span>| {
  let span = match span.to_json() {
  Ok(span) => span,
  Err(e) => {
  relay_log::error!(error = &e as &dyn Error, "Failed to serialize span");
- continue;
+ return;
  }
  };
  let mut item = Item::new(ItemType::Span);
  item.set_payload(ContentType::Json, span);
  state.managed_envelope.envelope_mut().add_item(item);
+ };
+
+ // Add child spans as envelope items.
+ if let Some(child_spans) = event.spans.value() {
+ for span in child_spans {
+ // HACK: clone the span to set the segment_id. This should happen
+ // as part of normalization once standalone spans reach wider adoption.
+ let mut span = span.clone();
+ let Some(inner_span) = span.value_mut() else { continue };
+ inner_span.segment_id = transaction_span.segment_id.clone();
+ inner_span.is_segment = Annotated::new(false);
+ add_span(span);
+ }
  }
+
+ // Add transaction span as an envelope item.
+ add_span(transaction_span.into());
  }
 
  /// Computes the sampling decision on the incoming event

relay-server/src/metrics_extraction/snapshots/relay_server__metrics_extraction__event__tests__extract_span_metrics_mobile.snap

@@ -1,5 +1,5 @@
 ---
-source: relay-server/src/metrics_extraction/spans/mod.rs
+source: relay-server/src/metrics_extraction/event.rs
 expression: "(&event.value().unwrap().spans, metrics)"
 ---
 (
@@ -21,6 +21,8 @@ expression: "(&event.value().unwrap().spans, metrics)"
  trace_id: TraceId(
  "ff62a8b040f340bda5d830223def1d81",
  ),
+ segment_id: ~,
+ is_segment: ~,
  status: ~,
  tags: ~,
  origin: ~,

relay-server/src/metrics_extraction/transactions/mod.rs

@@ -523,6 +523,8 @@ mod tests {
  trace_id: TraceId(
  "ff62a8b040f340bda5d830223def1d81",
  ),
+ segment_id: ~,
+ is_segment: ~,
  status: ~,
  tags: ~,
  origin: ~,

tests/integration/test_store.py

@@ -6,7 +6,7 @@
 import socket
 import threading
 import pytest
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 from requests.exceptions import HTTPError
 from flask import abort, Response
@@ -1206,21 +1206,45 @@ def test_spans(
 
  relay.send_event(project_id, event)
 
- msg = spans_consumer.get_message()
- del msg["start_time"]
- assert msg == {
+ child_span = spans_consumer.get_message()
+ del child_span["start_time"]
+ assert child_span == {
  "type": "span",
  "event_id": "cbf6960622e14a45abc1f03b2055b186",
  "project_id": 42,
  "span": {
  "description": "GET /api/0/organizations/?member=1",
+ "is_segment": False,
  "op": "http",
  "parent_span_id": "aaaaaaaaaaaaaaaa",
+ "segment_id": "968cff94913ebb07",
  "span_id": "bbbbbbbbbbbbbbbb",
  "start_timestamp": 1000.0,
  "timestamp": 3000.0,
  "trace_id": "ff62a8b040f340bda5d830223def1d81",
  },
  }
 
+ transaction_span = spans_consumer.get_message()
+ del transaction_span["start_time"]
+ assert transaction_span == {
+ "event_id": "cbf6960622e14a45abc1f03b2055b186",
+ "project_id": 42,
+ "span": {
+ "is_segment": True,
+ "op": "hi",
+ "segment_id": "968cff94913ebb07",
+ "span_id": "968cff94913ebb07",
+ "start_timestamp": datetime.fromisoformat(event["start_timestamp"])
+ .replace(tzinfo=timezone.utc)
+ .timestamp(),
+ "status": "unknown",
+ "timestamp": datetime.fromisoformat(event["timestamp"])
+ .replace(tzinfo=timezone.utc)
+ .timestamp(),
+ "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+ },
+ "type": "span",
+ }
+
  spans_consumer.assert_empty()