Issue 2678: Span PoC - direction/feedback

Cargo.toml

@@ -33,7 +33,6 @@ httpdate = "1.0"
 httparse = "1.5.1"
 h2 = { version = "0.3.3", optional = true }
 itoa = "0.4.1"
-tracing = { version = "0.1", default-features = false, features = ["std"] }
 pin-project-lite = "0.2.4"
 tower-service = "0.3"
 tokio = { version = "1", features = ["sync"] }
@@ -42,7 +41,18 @@ want = "0.3"
 # Optional
 
 libc = { version = "0.2", optional = true }
+log = { version = "0.4.14", optional = true }
+opentelemetry = { version = "0.16.0", features = ["rt-tokio", "trace"], optional = true }
+opentelemetry-jaeger = { version = "0.15.0", features = ["reqwest_collector_client","rt-tokio"], optional = true }
+serde_json = { version = "1.0", optional = true }
 socket2 = { version = "0.4", optional = true }
+reqwest = {version = "0.11", optional = true }
+tracing = { version = "0.1", default-features = false, features = ["std", "log"], optional = true }
+tracing-core = { version = "0.1", default-features = false, optional = true }
+tracing-log = { version = "0.1", default-features = false, optional = true }
+tracing-opentelemetry = { version = "0.16", optional = true }
+tracing-subscriber = { version = "0.3", features = ["std", "env-filter"], optional = true }
+tracing-tree = { version = "0.1", default-features = false, optional = true }
 
 [dev-dependencies]
 futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
@@ -80,9 +90,10 @@ full = [
     "client",
     "http1",
     "http2",
+    "layers",
+    "runtime",
     "server",
     "stream",
-    "runtime",
 ]
 
 # HTTP versions
@@ -108,6 +119,21 @@ tcp = [
     "tokio/time",
 ]
 
+# Tracing layers
+layers = [
+    "log",
+    "opentelemetry",
+    "opentelemetry-jaeger",
+    "serde_json",
+    "reqwest",
+    "tracing",
+    "tracing-core",
+    "tracing-log",
+    "tracing-opentelemetry",
+    "tracing-subscriber",
+    "tracing-tree",
+    ]
+
 # C-API support (currently unstable (no semver))
 ffi = ["libc"]
 
@@ -140,11 +166,40 @@ name = "client_json"
 path = "examples/client_json.rs"
 required-features = ["full"]
 
+[[example]]
+name = "client_json_tracing_json"
+path = "examples/client_json_tracing_json.rs"
+required-features = ["full",
+    "tracing/max_level_trace",
+    ]
+
+[[example]]
+name = "client_json_tracing_off"
+path = "examples/client_json_tracing_off.rs"
+required-features = ["full",
+    "tracing/max_level_off",
+    "log/max_level_off",
+    ]
+
+[[example]]
+name = "client_json_tracing_otel"
+path = "examples/client_json_tracing_otel.rs"
+required-features = ["full",
+    "tracing/max_level_trace",
+    ]
+
 [[example]]
 name = "echo"
 path = "examples/echo.rs"
 required-features = ["full"]
 
+[[example]]
+name = "echo_tracing_otel"
+path = "examples/echo_tracing_otel.rs"
+required-features = ["full",
+    "tracing/max_level_trace",
+    ]
+
 [[example]]
 name = "gateway"
 path = "examples/gateway.rs"

examples/client_json_tracing_json.rs

@@ -0,0 +1,64 @@
+#![deny(warnings)]
+#![warn(rust_2018_idioms)]
+
+// Statically compile tracing events and spans - "compile out" tracing code.
+//
+// Usage:
+//
+// $ cargo run --features="full tracing/max_level_info" --example client_json_tracing
+// ...
+// Running `target/debug/examples/client_json_tracing`
+
+// etc.
+
+use hyper::body::Buf;
+use hyper::Client;
+use hyper::JsonLayer;
+use serde::Deserialize;
+use tracing::info;
+use tracing_subscriber::prelude::*;
+
+// A simple type alias so as to DRY.
+type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
+
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Set up `tracing-subscriber` to process tracing data.
+    tracing_subscriber::registry().with(JsonLayer).init();
+
+    // Log a `tracing` "event".
+    info!(status = true, answer = 42, message = "first event");
+
+    let url = "http://jsonplaceholder.typicode.com/users".parse().unwrap();
+    let users = fetch_json(url).await?;
+    // print users
+    println!("users: {:#?}", users);
+
+    // print the sum of ids
+    let sum = users.iter().fold(0, |acc, user| acc + user.id);
+    println!("sum of ids: {}", sum);
+    Ok(())
+}
+
+async fn fetch_json(url: hyper::Uri) -> Result<Vec<User>> {
+    let client = Client::new();
+
+    // Fetch the url...
+    let res = client.get(url).await?;
+
+    // asynchronously aggregate the chunks of the body
+    let body = hyper::body::aggregate(res).await?;
+
+    // try to parse as json with serde_json
+    let users = serde_json::from_reader(body.reader())?;
+
+    Ok(users)
+}
+
+#[derive(Deserialize, Debug)]
+struct User {
+    id: i32,
+    #[allow(unused)]
+    name: String,
+}

examples/client_json_tracing_off.rs

@@ -0,0 +1,76 @@
+#![deny(warnings)]
+#![warn(rust_2018_idioms)]
+
+// Statically compile out all tracing and logging events and spans
+// - that is, "compile out" all tracing and logging code.
+//
+// Usage:
+//
+// $ cargo run --features="full tracing/max_level_off log/max_level_off" --example client_json_tracing_off
+// ...
+// Running `target/debug/examples/client_json_tracing_off`
+// users: [
+//     User {
+//         id: 1,
+//         name: "Leanne Graham",
+//     },
+//     User {
+//         id: 2,
+//         name: "Ervin Howell",
+// etc.
+
+use hyper::body::Buf;
+use hyper::Client;
+use hyper::PrintLayer;
+// use hyper::JsonLayer;
+use serde::Deserialize;
+use tracing::info;
+use tracing_subscriber::prelude::*;
+
+// A simple type alias so as to DRY.
+type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
+
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Set up `tracing-subscriber` to process tracing data.
+    // Note:
+    // We silence tracing via compile/build time features - see `Cargo.toml`.
+    // Hence, no change is required from the `client_json_tracing` example.
+    tracing_subscriber::registry().with(PrintLayer).init();
+
+    // "Log" a `tracing` "event".
+    info!(status = true, answer = 42, message = "first event");
+
+    let url = "http://jsonplaceholder.typicode.com/users".parse().unwrap();
+    let users = fetch_json(url).await?;
+    // print users
+    println!("users: {:#?}", users);
+
+    // print the sum of ids
+    let sum = users.iter().fold(0, |acc, user| acc + user.id);
+    println!("sum of ids: {}", sum);
+    Ok(())
+}
+
+async fn fetch_json(url: hyper::Uri) -> Result<Vec<User>> {
+    let client = Client::new();
+
+    // Fetch the url...
+    let res = client.get(url).await?;
+
+    // asynchronously aggregate the chunks of the body
+    let body = hyper::body::aggregate(res).await?;
+
+    // try to parse as json with serde_json
+    let users = serde_json::from_reader(body.reader())?;
+
+    Ok(users)
+}
+
+#[derive(Deserialize, Debug)]
+struct User {
+    id: i32,
+    #[allow(unused)]
+    name: String,
+}

examples/client_json_tracing_otel.rs

@@ -0,0 +1,107 @@
+#![deny(warnings)]
+#![warn(rust_2018_idioms)]
+
+// Statically compile tracing events and spans - "compile out" tracing code.
+//
+// Usage:
+//
+// $ cargo run --features="full tracing/max_level_info" --example client_otel_tracing
+// ...
+// Running `target/debug/examples/client_otel_tracing_off`
+// Hyper tracing event:
+//   level=Level(Info)
+//   target="client_otel_tracing"
+//   name="event examples/client_otel_tracing.rs:24"
+//   field=status
+//   field=answer
+//   field=message
+// etc.
+
+use hyper::body::Buf;
+use hyper::Client;
+//use hyper::OtelLayer;
+use serde::Deserialize;
+use tracing::info;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::prelude::*;
+//use tracing_subscriber::Layer;
+//use tracing_subscriber::Registry;
+
+// A simple type alias so as to DRY.
+type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // First create propagators
+    let baggage_propagator = opentelemetry::sdk::propagation::BaggagePropagator::new();
+    let trace_context_propagator = opentelemetry::sdk::propagation::TraceContextPropagator::new();
+    let jaeger_propagator = opentelemetry_jaeger::Propagator::new();
+    opentelemetry::global::set_text_map_propagator(
+        opentelemetry::sdk::propagation::TraceContextPropagator::new(),
+    );
+
+    // Second compose propagators
+    let _composite_propagator =  opentelemetry::sdk::propagation::TextMapCompositePropagator::new(vec![
+        Box::new(baggage_propagator),
+        Box::new(trace_context_propagator),
+        Box::new(jaeger_propagator),
+    ]);
+    // Third create Jaeger pipeline
+    let tracer = opentelemetry_jaeger::new_pipeline()
+        .with_service_name("client_json2")
+        .install_batch(opentelemetry::runtime::Tokio)
+        .unwrap();
+    // Initialize `tracing` using `opentelemetry-tracing` and configure stdout logging
+    tracing_subscriber::Registry::default()
+        .with(tracing_subscriber::EnvFilter::new("TRACE"))
+        .with(hyper::OtelLayer::new(tracer))
+        .with(tracing_subscriber::fmt::layer())
+        //.with(tracing_tree::HierarchicalLayer::new(2))
+        .init();
+
+    // Trace executed (async) code
+    // tracing::subscriber::with_default(subscriber, || async {
+    // Create a span and enter it, returning a guard....
+    let root_span = tracing::span!(tracing::Level::INFO, "root_span_echo").entered();
+    root_span.in_scope(|| async {
+        // Log a `tracing` "event".
+        info!(status = true, answer = 42, message = "first event");
+
+        let url = "http://jsonplaceholder.typicode.com/users".parse().unwrap();
+        let users = fetch_json(url).await.expect("Vector of user data");
+        // print users
+        println!("users: {:#?}", users);
+
+        // print the sum of ids
+        let sum = users.iter().fold(0, |acc, user| acc + user.id);
+        println!("sum of ids: {}", sum);
+
+        // ...but, it can also be exited explicitly, returning the `Span`
+        // struct:
+        //let _root_span = root_span.exit();
+    }).await;
+    opentelemetry::global::shutdown_tracer_provider();
+    Ok(())
+}
+
+async fn fetch_json(url: hyper::Uri) -> Result<Vec<User>> {
+    let client = Client::new();
+
+    // Fetch the url...
+    let res = client.get(url).await?;
+
+    // asynchronously aggregate the chunks of the body
+    let body = hyper::body::aggregate(res).await?;
+
+    // try to parse as json with serde_json
+    let users = serde_json::from_reader(body.reader())?;
+
+    Ok(users)
+}
+
+#[derive(Deserialize, Debug)]
+struct User {
+    id: i32,
+    #[allow(unused)]
+    name: String,
+}