Throwaway OTel retroactive-span diagnostics spike

sdk/cosmos/azure_data_cosmos_driver/Cargo.toml

@@ -28,6 +28,8 @@ bytes.workspace = true
 crossbeam-epoch = { workspace = true, features = ["std"] }
 futures.workspace = true
 h2 = { workspace = true, optional = true }
+opentelemetry = { workspace = true, optional = true }
+opentelemetry_sdk = { workspace = true, optional = true, features = ["testing"] }
 percent-encoding = { workspace = true, optional = true }
 rand = { workspace = true, optional = true }
 reqwest = { workspace = true, optional = true }
@@ -77,6 +79,21 @@ reqwest = [
 rustls = ["reqwest", "reqwest/rustls", "__tls"]
 native_tls = ["reqwest", "reqwest/native-tls", "__tls"]
 fault_injection = ["dep:rand"]
+# `otel_spans_spike` gates a THROWAWAY, in-crate `#[cfg(test)]` feasibility spike
+# (`diagnostics::otel_span_spike`) that reconstructs a completed `DiagnosticsContext`
+# into a BACKDATED OpenTelemetry span tree using the raw `opentelemetry` `SpanBuilder`
+# (`with_start_time`/`with_end_time`). It exists only to prove the OTel mapping in
+# DIAGNOSTICS-CONTRACT.md is feasible; it is NOT production tracing wiring and is OFF by
+# default. It pulls `opentelemetry` + `opentelemetry_sdk` (in-memory exporter) only.
+#
+# Scoping caveat (throwaway): the feature enables `dep:opentelemetry` +
+# `dep:opentelemetry_sdk` unconditionally, but the ONLY consumer is the
+# `#[cfg(all(test, feature = "otel_spans_spike"))]` module above. So a *non-test*
+# `--all-features` build (e.g. `cargo build --all-features`) compiles both crates
+# (and `opentelemetry_sdk`'s `testing` feature) into the graph with no user. That is
+# harmless for a throwaway spike, but if this ever graduates to real tracing wiring the
+# deps should move behind a non-test consumer (or the exporter behind a `dev-dependency`).
+otel_spans_spike = ["dep:opentelemetry", "dep:opentelemetry_sdk"]
 # `__internal_in_memory_emulator` exposes the in-memory query evaluator
 # (`crate::query::eval`, `crate::query::value`) used by the in-memory Cosmos DB
 # emulator. The evaluator intentionally trades full Cosmos parity for emulator

sdk/cosmos/azure_data_cosmos_driver/src/diagnostics/mod.rs

@@ -21,6 +21,12 @@
 mod diagnostics_context;
 mod proxy_configuration;
 
+/// THROWAWAY OpenTelemetry retroactive-span feasibility spike. In-crate `#[cfg(test)]`
+/// so it can build a real `DiagnosticsContext` via the `pub(crate)`
+/// `DiagnosticsContextBuilder`. Gated behind the off-by-default `otel_spans_spike` feature.
+#[cfg(all(test, feature = "otel_spans_spike"))]
+mod otel_span_spike;
+
 pub(crate) use diagnostics_context::DiagnosticsContextBuilder;
 pub use diagnostics_context::{
     DiagnosticsContext, ExecutionContext, FailedTransportShardDiagnostics, PipelineType,

sdk/cosmos/azure_data_cosmos_driver/src/diagnostics/otel_span_spike.rs

@@ -0,0 +1,223 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+//! THROWAWAY feasibility spike — **not production tracing wiring.**
+//!
+//! Behind the off-by-default `otel_spans_spike` feature, and in-crate so it can build a real
+//! [`DiagnosticsContext`](super::DiagnosticsContext) via the `pub(crate)`
+//! [`DiagnosticsContextBuilder`](super::DiagnosticsContextBuilder) — importing from
+//! `crate::diagnostics` only, with no dependency on any prototype capture engine.
+//!
+//! It proves the OpenTelemetry mapping in `DIAGNOSTICS-CONTRACT.md` is feasible: a *completed*
+//! `DiagnosticsContext` can be reconstructed into a **backdated** OTel span tree using the raw
+//! `opentelemetry` [`SpanBuilder`](opentelemetry::trace::SpanBuilder) with
+//! [`with_start_time`](opentelemetry::trace::SpanBuilder::with_start_time) for the backdated start
+//! and [`end_with_timestamp`](opentelemetry::trace::Span::end_with_timestamp) for the backdated
+//! end — timestamps the `azure_core::tracing` abstraction does not expose (it builds spans at
+//! "now"). The recorded attempts are laid onto an explicit, clearly-backdated timeline (injected
+//! durations) and the operation → attempt parent/child relationships are reconstructed, then
+//! asserted.
+
+use super::{DiagnosticsContext, DiagnosticsContextBuilder, ExecutionContext};
+use crate::driver::routing::CosmosEndpoint;
+use crate::models::{ActivityId, SubStatusCode};
+use crate::options::{DiagnosticsOptions, Region};
+use azure_core::http::StatusCode;
+
+use opentelemetry::trace::{
+    Span, SpanBuilder, SpanId, SpanKind, TraceContextExt, Tracer, TracerProvider,
+};
+use opentelemetry::{Context, KeyValue};
+use opentelemetry_sdk::trace::{
+    InMemorySpanExporterBuilder, SdkTracerProvider, SimpleSpanProcessor,
+};
+
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+/// A clearly-backdated operation start: 2020-01-01T00:00:00Z.
+const OP_START_UNIX_SECS: u64 = 1_577_836_800;
+
+/// Builds a realistic completed `DiagnosticsContext` (retry `429` -> `200`) via the
+/// `pub(crate)` builder — the same path the driver pipeline uses.
+fn completed_context() -> DiagnosticsContext {
+    use super::{PipelineType, TransportHttpVersion, TransportKind, TransportSecurity};
+
+    let mut builder = DiagnosticsContextBuilder::new(
+        ActivityId::from_string("op-activity-1".to_string()),
+        Arc::new(DiagnosticsOptions::default()),
+    );
+
+    let start = |b: &mut DiagnosticsContextBuilder, ec, region: Region, url: &str| {
+        let endpoint = CosmosEndpoint::regional(region, url::Url::parse(url).unwrap());
+        b.start_request(
+            ec,
+            PipelineType::DataPlane,
+            TransportSecurity::Secure,
+            TransportKind::Gateway,
+            TransportHttpVersion::Http11,
+            &endpoint,
+        )
+    };
+
+    let h1 = start(
+        &mut builder,
+        ExecutionContext::Initial,
+        Region::EAST_US_2,
+        "https://east/",
+    );
+    builder.complete_request(
+        h1,
+        StatusCode::TooManyRequests,
+        Some(SubStatusCode::new(3200)),
+    );
+    let h2 = start(
+        &mut builder,
+        ExecutionContext::Retry,
+        Region::WEST_US_2,
+        "https://west/",
+    );
+    builder.complete_request(h2, StatusCode::Ok, None);
+    builder.set_operation_status(StatusCode::Ok, None);
+    builder.complete()
+}
+
+#[test]
+fn reconstructs_completed_diagnostics_into_backdated_span_tree() {
+    let ctx = completed_context();
+
+    let exporter = InMemorySpanExporterBuilder::new().build();
+    let provider = SdkTracerProvider::builder()
+        .with_span_processor(SimpleSpanProcessor::new(exporter.clone()))
+        .build();
+    let tracer = provider.tracer("cosmos-diagnostics-retro-spike");
+
+    let op_start = UNIX_EPOCH + Duration::from_secs(OP_START_UNIX_SECS);
+
+    // Lay the recorded attempts out sequentially on the backdated timeline (injected durations).
+    //
+    // Spans are built off `ctx.requests()` — the *retained* attempt list — and every span-count
+    // assertion below is made against `requests.len()`, NOT `ctx.request_count()`. On `main` the two
+    // are equal, but once the retry-storm compaction (PR #4683) lands they diverge:
+    // `request_count()` reports the true pre-compaction attempt total while `requests()` is the
+    // bounded list that keeps only first+last of each collapsed run. A faithful emitter must
+    // count/iterate off the same list it emits from; it can NOT produce `request_count()` spans,
+    // because a compacted run maps to a single span + a repeat count (DIAGNOSTICS-CONTRACT.md §7.3),
+    // not one span per elided attempt. Keeping this test anchored on `requests()` means it stays
+    // correct — and stays honest about the mapping — after compaction lands.
+    let requests = ctx.requests();
+    let mut cursor = op_start;
+    let mut windows: Vec<(SystemTime, SystemTime)> = Vec::with_capacity(requests.len());
+    for (i, _req) in requests.iter().enumerate() {
+        let start = cursor;
+        let end = start + Duration::from_millis(3 + i as u64);
+        windows.push((start, end));
+        cursor = end;
+    }
+    let op_end = cursor;
+
+    // Root (operation) span, backdated.
+    let root = tracer.build(
+        SpanBuilder::from_name("Cosmos read_item")
+            .with_kind(SpanKind::Client)
+            .with_start_time(op_start)
+            .with_attributes(vec![
+                KeyValue::new(
+                    "az.client_request_id",
+                    ctx.activity_id().as_str().to_string(),
+                ),
+                KeyValue::new("db.operation", "read_item"),
+            ]),
+    );
+    let root_cx = Context::current().with_span(root);
+    let root_span_id = root_cx.span().span_context().span_id();
+
+    // Attempt (child) spans, backdated, parented to the operation root.
+    for (req, (start, end)) in requests.iter().zip(windows.iter().copied()) {
+        let mut attrs = vec![
+            KeyValue::new("server.address", req.endpoint().to_string()),
+            KeyValue::new("db.cosmosdb.request_charge", req.request_charge().value()),
+        ];
+        if let Some(id) = req.activity_id() {
+            // Note: azure_core's constant is `az.service_request.id` (a dot before `id`).
+            attrs.push(KeyValue::new(
+                "az.service_request.id",
+                id.as_str().to_string(),
+            ));
+        }
+        let mut child = tracer.build_with_context(
+            SpanBuilder::from_name("read_item attempt")
+                .with_kind(SpanKind::Client)
+                .with_start_time(start)
+                .with_attributes(attrs),
+            &root_cx,
+        );
+        child.end_with_timestamp(end);
+    }
+    root_cx.span().end_with_timestamp(op_end);
+    let _ = provider.force_flush();
+
+    // ---- assertions ----
+    let spans = exporter
+        .get_finished_spans()
+        .expect("in-memory exporter returns spans");
+    assert_eq!(
+        spans.len(),
+        requests.len() + 1,
+        "one operation root span + one span per retained attempt"
+    );
+
+    let root_data = spans
+        .iter()
+        .find(|s| s.parent_span_id == SpanId::INVALID)
+        .expect("exactly one root span (no parent)");
+    assert_eq!(root_data.span_context.span_id(), root_span_id);
+    assert_eq!(
+        root_data.start_time, op_start,
+        "root span carries the injected, backdated operation start"
+    );
+    let a_year_ago = SystemTime::now()
+        .checked_sub(Duration::from_secs(365 * 24 * 3600))
+        .expect("valid time");
+    assert!(
+        root_data.start_time < a_year_ago,
+        "root span start is far in the past (retroactive reconstruction, not 'now')"
+    );
+
+    let children: Vec<_> = spans
+        .iter()
+        .filter(|s| s.parent_span_id != SpanId::INVALID)
+        .collect();
+    assert_eq!(children.len(), requests.len());
+    for child in &children {
+        assert_eq!(
+            child.parent_span_id, root_span_id,
+            "attempt spans are children of the operation root"
+        );
+        assert!(
+            child.start_time >= op_start && child.end_time <= op_end,
+            "attempt span falls within the operation window"
+        );
+    }
+
+    // Sequential, non-overlapping layout in our reconstruction.
+    let mut ordered = children.clone();
+    ordered.sort_by_key(|s| s.start_time);
+    for pair in ordered.windows(2) {
+        assert!(
+            pair[0].end_time <= pair[1].start_time,
+            "attempts are laid out sequentially"
+        );
+    }
+
+    // Attribute alignment: operation carries the client request id, attempts the service
+    // request id (the azure_core `az.service_request.id` name, dot included).
+    assert!(root_data
+        .attributes
+        .iter()
+        .any(|kv| kv.key.as_str() == "az.client_request_id"));
+    assert!(children.iter().all(|c| c
+        .attributes
+        .iter()
+        .any(|kv| kv.key.as_str() == "az.service_request.id")));
+}