implement w3c trace context response propagation (#998)

Co-authored-by: Cijo Thomas <cithomas@microsoft.com>

Cargo.toml

@@ -31,6 +31,7 @@ members = [
     "examples/grpc",
     "examples/http",
     "examples/hyper-prometheus",
+    "examples/traceresponse",
     "examples/tracing-grpc",
     "examples/jaeger-remote-sampler",
     "examples/zipkin",

examples/traceresponse/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "traceresponse"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[[bin]] # Bin to run the http server
+name = "http-server"
+path = "src/server.rs"
+doc = false
+
+[[bin]] # Bin to run the client
+name = "http-client"
+path = "src/client.rs"
+doc = false
+
+[dependencies]
+hyper = { version = "0.14", features = ["full"] }
+tokio = { version = "1.0", features = ["full"] }
+opentelemetry = { path = "../../opentelemetry" }
+opentelemetry-http = { path = "../../opentelemetry-http" }
+opentelemetry-contrib = { path = "../../opentelemetry-contrib" }

examples/traceresponse/README.md

@@ -0,0 +1,28 @@
+# HTTP Example
+
+This is a simple example using [hyper] that demonstrates tracing http request
+from client to server, and from the server back to the client using the
+[W3C Trace Context Response] header. The example shows key aspects of tracing
+such as:
+
+- Root Span (on Client)
+- Child Span from a Remote Parent (on Server)
+- SpanContext Propagation (from Client to Server)
+- SpanContext Propagation (from Server to Client)
+- Span Events
+- Span Attributes
+
+[hyper]: https://hyper.rs/
+[W3C Trace Context Response]: https://w3c.github.io/trace-context/#traceresponse-header
+
+## Usage
+
+```shell
+# Run server
+$ cargo run --bin http-server
+
+# In another tab, run client
+$ cargo run --bin http-client
+
+# The spans should be visible in stdout in the order that they were exported.
+```

examples/traceresponse/src/client.rs

@@ -0,0 +1,71 @@
+use hyper::http::HeaderValue;
+use hyper::{body::Body, Client};
+use opentelemetry::global;
+use opentelemetry::propagation::TextMapPropagator;
+use opentelemetry::sdk::export::trace::stdout;
+use opentelemetry::sdk::{
+    propagation::TraceContextPropagator,
+    trace::{self, Sampler},
+};
+use opentelemetry::trace::SpanKind;
+use opentelemetry::{
+    trace::{TraceContextExt, Tracer},
+    Context, KeyValue,
+};
+use opentelemetry_contrib::trace::propagator::trace_context_response::TraceContextResponsePropagator;
+use opentelemetry_http::{HeaderExtractor, HeaderInjector};
+
+fn init_tracer() -> impl Tracer {
+    global::set_text_map_propagator(TraceContextPropagator::new());
+    // Install stdout exporter pipeline to be able to retrieve the collected spans.
+    // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
+    // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
+    stdout::new_pipeline()
+        .with_trace_config(trace::config().with_sampler(Sampler::AlwaysOn))
+        .install_simple()
+}
+
+#[tokio::main]
+async fn main() -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
+    let _tracer = init_tracer();
+
+    let client = Client::new();
+    let tracer = global::tracer("example/client");
+    let span = tracer
+        .span_builder("say hello")
+        .with_kind(SpanKind::Client)
+        .start(&tracer);
+    let cx = Context::current_with_span(span);
+
+    let mut req = hyper::Request::builder().uri("http://127.0.0.1:3000");
+    global::get_text_map_propagator(|propagator| {
+        propagator.inject_context(&cx, &mut HeaderInjector(req.headers_mut().unwrap()))
+    });
+    let res = client.request(req.body(Body::from("Hello!"))?).await?;
+
+    let response_propagator: &dyn TextMapPropagator = &TraceContextResponsePropagator::new();
+
+    let response_cx =
+        response_propagator.extract_with_context(&cx, &HeaderExtractor(res.headers()));
+
+    let response_span = response_cx.span();
+
+    cx.span().add_event(
+        "Got response!".to_string(),
+        vec![
+            KeyValue::new("status", res.status().to_string()),
+            KeyValue::new(
+                "traceresponse",
+                res.headers()
+                    .get("traceresponse")
+                    .unwrap_or(&HeaderValue::from_static(""))
+                    .to_str()
+                    .unwrap()
+                    .to_string(),
+            ),
+            KeyValue::new("child_sampled", response_span.span_context().is_sampled()),
+        ],
+    );
+
+    Ok(())
+}

examples/traceresponse/src/server.rs

@@ -0,0 +1,67 @@
+use hyper::service::{make_service_fn, service_fn};
+use hyper::{Body, Request, Response, Server};
+use opentelemetry::propagation::TextMapPropagator;
+use opentelemetry::trace::{SpanKind, TraceContextExt};
+use opentelemetry::Context;
+use opentelemetry::{
+    global,
+    sdk::export::trace::stdout,
+    sdk::{
+        propagation::TraceContextPropagator,
+        trace::{self, Sampler},
+    },
+    trace::Tracer,
+};
+use opentelemetry_contrib::trace::propagator::trace_context_response::TraceContextResponsePropagator;
+use opentelemetry_http::{HeaderExtractor, HeaderInjector};
+use std::{convert::Infallible, net::SocketAddr};
+
+async fn handle(req: Request<Body>) -> Result<Response<Body>, Infallible> {
+    let parent_cx = global::get_text_map_propagator(|propagator| {
+        propagator.extract(&HeaderExtractor(req.headers()))
+    });
+    let _cx_guard = parent_cx.attach();
+
+    let tracer = global::tracer("example/server");
+    let span = tracer
+        .span_builder("say hello")
+        .with_kind(SpanKind::Server)
+        .start(&tracer);
+
+    let cx = Context::current_with_span(span);
+
+    cx.span().add_event("handling this...", Vec::new());
+
+    let mut res = Response::new("Hello, World!".into());
+
+    let response_propagator: &dyn TextMapPropagator = &TraceContextResponsePropagator::new();
+    response_propagator.inject_context(&cx, &mut HeaderInjector(res.headers_mut()));
+
+    Ok(res)
+}
+
+fn init_tracer() -> impl Tracer {
+    global::set_text_map_propagator(TraceContextPropagator::new());
+
+    // Install stdout exporter pipeline to be able to retrieve the collected spans.
+    // For the demonstration, use `Sampler::AlwaysOn` sampler to sample all traces. In a production
+    // application, use `Sampler::ParentBased` or `Sampler::TraceIdRatioBased` with a desired ratio.
+    stdout::new_pipeline()
+        .with_trace_config(trace::config().with_sampler(Sampler::AlwaysOn))
+        .install_simple()
+}
+
+#[tokio::main]
+async fn main() {
+    let _tracer = init_tracer();
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
+
+    let make_svc = make_service_fn(|_conn| async { Ok::<_, Infallible>(service_fn(handle)) });
+
+    let server = Server::bind(&addr).serve(make_svc);
+
+    println!("Listening on {addr}");
+    if let Err(e) = server.await {
+        eprintln!("server error: {e}");
+    }
+}

opentelemetry-contrib/Cargo.toml

@@ -32,7 +32,9 @@ async-std = { version = "1.10", optional = true }
 async-trait = { version = "0.1", optional = true }
 base64 = { version = "0.13", optional = true }
 futures = { version = "0.3", optional = true }
+once_cell = "1.17.1"
 opentelemetry = { version = "0.18", path = "../opentelemetry", features = ["trace"] }
+opentelemetry_api = { version = "0.18", path = "../opentelemetry-api" }
 serde_json = { version = "1", optional = true }
 tokio = { version = "1.0", features = ["fs", "io-util"], optional = true }
 

opentelemetry-contrib/src/trace/propagator/mod.rs

@@ -9,3 +9,4 @@
 //!
 //! This module also provides relative types for those propagators.
 pub mod binary;
+pub mod trace_context_response;