Merge branch 'main' into add-shutdown-with-timeout-for-metric-exporter

Author: mohammadVatandoost

.github/workflows/ci.yml

@@ -106,7 +106,7 @@ jobs:
     continue-on-error: true # Prevent sudden announcement of a new advisory from failing ci
     steps:
       - uses: actions/checkout@v4
-      - uses: EmbarkStudios/cargo-deny-action@v1
+      - uses: EmbarkStudios/cargo-deny-action@v2
         with:
           command: check advisories
   docs:
@@ -153,3 +153,18 @@ jobs:
           CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
         with:
           fail_ci_if_error: true
+  cargo-machete:
+    continue-on-error: true
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: true
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: stable
+      - uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-machete
+      - name: cargo machete
+        run: cargo machete

README.md

@@ -179,16 +179,16 @@ you're more than welcome to participate!
 
 ### Maintainers
 
-* [Cijo Thomas](https://github.com/cijothomas)
+* [Cijo Thomas](https://github.com/cijothomas), Microsoft
 * [Harold Dost](https://github.com/hdost)
-* [Lalit Kumar Bhasin](https://github.com/lalitb)
-* [Utkarsh Umesan Pillai](https://github.com/utpilla)
+* [Lalit Kumar Bhasin](https://github.com/lalitb), Microsoft
+* [Utkarsh Umesan Pillai](https://github.com/utpilla), Microsoft
 * [Zhongyang Wu](https://github.com/TommyCpp)
 
 ### Approvers
 
-* [Shaun Cox](https://github.com/shaun-cox)
-* [Scott Gerring](https://github.com/scottgerring)
+* [Shaun Cox](https://github.com/shaun-cox), Microsoft
+* [Scott Gerring](https://github.com/scottgerring), Datadog
 
 ### Emeritus
 

deny.toml

@@ -1,14 +1,7 @@
-exclude=[
-    "actix-http",
-    "actix-http-tracing",
-    "actix-udp",
-    "actix-udp-example",
-    "tracing-grpc",
-    "http"
-]
+[graph]
+exclude=[]
 
 [licenses]
-unlicensed = "deny"
 allow = [
     "MIT",
     "Apache-2.0",
@@ -29,5 +22,5 @@ license-files = [
 ]
 
 [advisories]
-unmaintained = "allow"
+unmaintained = "none"
 yanked = "allow"

docs/adr/001_error_handling.md

@@ -171,3 +171,5 @@ Note that at the time of writing, there is no instance we have identified within
 We will use [thiserror](https://docs.rs/thiserror/latest/thiserror/) by default to implement Rust's [error trait](https://doc.rust-lang.org/core/error/trait.Error.html).
 This keeps our code clean, and as it does not appear in our interface, we can choose to replace any particular usage with a hand-rolled implementation should we need to.
 
+### 6. Don't use `#[non_exhaustive]` by default
+If an `Error` response set is closed - if we can confidently say it is very unlikely to gain new variants in the future - we should not annotate it with `#[non_exhaustive]`. By way of example, the variants of the exporter error types described above are exhaustively documented in the OpenTelemetry Specification, and we can confidently say that we do not expect new variants.

examples/tracing-grpc/Cargo.toml

@@ -23,3 +23,8 @@ tonic = { workspace = true, features = ["server", "codegen", "channel", "prost"]
 
 [build-dependencies]
 tonic-build = { workspace = true }
+
+[package.metadata.cargo-machete]
+ignored = [
+    "prost" # needed for `tonic-build`
+]

examples/tracing-http-propagator/Cargo.toml

@@ -23,5 +23,8 @@ tokio = { workspace = true, features = ["full"] }
 opentelemetry = { path = "../../opentelemetry" }
 opentelemetry_sdk = { path = "../../opentelemetry-sdk" }
 opentelemetry-http = { path = "../../opentelemetry-http" }
-opentelemetry-stdout = { workspace = true, features = ["trace"] }
+opentelemetry-stdout = { workspace = true, features = ["trace", "logs"] }
 opentelemetry-semantic-conventions = { path = "../../opentelemetry-semantic-conventions" }
+opentelemetry-appender-tracing = { workspace = true }
+tracing = { workspace = true, features = ["std"]}
+tracing-subscriber = { workspace = true, features = ["env-filter","registry", "std", "fmt"] }

examples/tracing-http-propagator/src/client.rs

@@ -3,21 +3,42 @@ use hyper_util::{client::legacy::Client, rt::TokioExecutor};
 use opentelemetry::{
     global,
     trace::{SpanKind, TraceContextExt, Tracer},
-    Context, KeyValue,
+    Context,
 };
+use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use opentelemetry_http::{Bytes, HeaderInjector};
-use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::SdkTracerProvider};
-use opentelemetry_stdout::SpanExporter;
+use opentelemetry_sdk::{
+    logs::SdkLoggerProvider, propagation::TraceContextPropagator, trace::SdkTracerProvider,
+};
+use opentelemetry_stdout::{LogExporter, SpanExporter};
+use tracing::info;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
-fn init_tracer() {
+fn init_tracer() -> SdkTracerProvider {
     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.
     let provider = SdkTracerProvider::builder()
         .with_simple_exporter(SpanExporter::default())
         .build();
 
-    global::set_tracer_provider(provider);
+    global::set_tracer_provider(provider.clone());
+    provider
+}
+
+fn init_logs() -> SdkLoggerProvider {
+    // Setup tracerprovider with stdout exporter
+    // that prints the spans to stdout.
+    let logger_provider = SdkLoggerProvider::builder()
+        .with_simple_exporter(LogExporter::default())
+        .build();
+    let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider);
+    tracing_subscriber::registry()
+        .with(otel_layer)
+        .with(tracing_subscriber::filter::LevelFilter::INFO)
+        .init();
+
+    logger_provider
 }
 
 async fn send_request(
@@ -37,21 +58,22 @@ async fn send_request(
     global::get_text_map_propagator(|propagator| {
         propagator.inject_context(&cx, &mut HeaderInjector(req.headers_mut().unwrap()))
     });
+    req.headers_mut()
+        .unwrap()
+        .insert("baggage", "is_synthetic=true".parse().unwrap());
     let res = client
         .request(req.body(Full::new(Bytes::from(body_content.to_string())))?)
         .await?;
 
-    cx.span().add_event(
-        "Got response!",
-        vec![KeyValue::new("status", res.status().to_string())],
-    );
+    info!(name: "ResponseReceived", status = res.status().to_string(), message = "Response received");
 
     Ok(())
 }
 
 #[tokio::main]
 async fn main() -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
-    init_tracer();
+    let tracer_provider = init_tracer();
+    let logger_provider = init_logs();
 
     send_request(
         "http://127.0.0.1:3000/health",
@@ -66,5 +88,11 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error + Send + Sy
     )
     .await?;
 
+    tracer_provider
+        .shutdown()
+        .expect("Shutdown provider failed");
+    logger_provider
+        .shutdown()
+        .expect("Shutdown provider failed");
     Ok(())
 }

examples/tracing-http-propagator/src/server.rs

@@ -2,16 +2,27 @@ use http_body_util::{combinators::BoxBody, BodyExt, Full};
 use hyper::{body::Incoming, service::service_fn, Request, Response, StatusCode};
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use opentelemetry::{
+    baggage::BaggageExt,
     global::{self, BoxedTracer},
+    logs::LogRecord,
+    propagation::TextMapCompositePropagator,
     trace::{FutureExt, Span, SpanKind, TraceContextExt, Tracer},
-    Context, KeyValue,
+    Context, InstrumentationScope, KeyValue,
 };
+use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use opentelemetry_http::{Bytes, HeaderExtractor};
-use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::SdkTracerProvider};
+use opentelemetry_sdk::{
+    error::OTelSdkResult,
+    logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider},
+    propagation::{BaggagePropagator, TraceContextPropagator},
+    trace::{SdkTracerProvider, SpanProcessor},
+};
 use opentelemetry_semantic_conventions::trace;
-use opentelemetry_stdout::SpanExporter;
+use opentelemetry_stdout::{LogExporter, SpanExporter};
 use std::{convert::Infallible, net::SocketAddr, sync::OnceLock};
 use tokio::net::TcpListener;
+use tracing::info;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 fn get_tracer() -> &'static BoxedTracer {
     static TRACER: OnceLock<BoxedTracer> = OnceLock::new();
@@ -30,11 +41,11 @@ async fn handle_health_check(
     _req: Request<Incoming>,
 ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Infallible> {
     let tracer = get_tracer();
-    let mut span = tracer
+    let _span = tracer
         .span_builder("health_check")
         .with_kind(SpanKind::Internal)
         .start(tracer);
-    span.add_event("Health check accessed", vec![]);
+    info!(name: "health_check", message = "Health check endpoint hit");
 
     let res = Response::new(
         Full::new(Bytes::from_static(b"Server is up and running!"))
@@ -50,11 +61,11 @@ async fn handle_echo(
     req: Request<Incoming>,
 ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Infallible> {
     let tracer = get_tracer();
-    let mut span = tracer
+    let _span = tracer
         .span_builder("echo")
         .with_kind(SpanKind::Internal)
         .start(tracer);
-    span.add_event("Echoing back the request", vec![]);
+    info!(name = "echo", message = "Echo endpoint hit");
 
     let res = Response::new(req.into_body().boxed());
 
@@ -69,14 +80,14 @@ async fn router(
     let response = {
         // Create a span parenting the remote client span.
         let tracer = get_tracer();
-        let mut span = tracer
+        let span = tracer
             .span_builder("router")
             .with_kind(SpanKind::Server)
             .start_with_context(tracer, &parent_cx);
 
-        span.add_event("dispatching request", vec![]);
+        info!(name = "router", message = "Dispatching request");
 
-        let cx = Context::default().with_span(span);
+        let cx = parent_cx.with_span(span);
         match (req.method(), req.uri().path()) {
             (&hyper::Method::GET, "/health") => handle_health_check(req).with_context(cx).await,
             (&hyper::Method::GET, "/echo") => handle_echo(req).with_context(cx).await,
@@ -93,24 +104,90 @@ async fn router(
     response
 }
 
+/// A custom log processor that enriches LogRecords with baggage attributes.
+/// Baggage information is not added automatically without this processor.
+#[derive(Debug)]
+struct EnrichWithBaggageLogProcessor;
+impl LogProcessor for EnrichWithBaggageLogProcessor {
+    fn emit(&self, data: &mut SdkLogRecord, _instrumentation: &InstrumentationScope) {
+        Context::map_current(|cx| {
+            for (kk, vv) in cx.baggage().iter() {
+                data.add_attribute(kk.clone(), vv.0.clone());
+            }
+        });
+    }
+
+    fn force_flush(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        Ok(())
+    }
+}
+
+/// A custom span processor that enriches spans with baggage attributes. Baggage
+/// information is not added automatically without this processor.
+#[derive(Debug)]
+struct EnrichWithBaggageSpanProcessor;
+impl SpanProcessor for EnrichWithBaggageSpanProcessor {
+    fn force_flush(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn on_start(&self, span: &mut opentelemetry_sdk::trace::Span, cx: &Context) {
+        for (kk, vv) in cx.baggage().iter() {
+            span.set_attribute(KeyValue::new(kk.clone(), vv.0.clone()));
+        }
+    }
+
+    fn on_end(&self, _span: opentelemetry_sdk::trace::SpanData) {}
+}
+
 fn init_tracer() -> SdkTracerProvider {
-    global::set_text_map_propagator(TraceContextPropagator::new());
+    let baggage_propagator = BaggagePropagator::new();
+    let trace_context_propagator = TraceContextPropagator::new();
+    let composite_propagator = TextMapCompositePropagator::new(vec![
+        Box::new(baggage_propagator),
+        Box::new(trace_context_propagator),
+    ]);
+
+    global::set_text_map_propagator(composite_propagator);
 
     // Setup tracerprovider with stdout exporter
     // that prints the spans to stdout.
     let provider = SdkTracerProvider::builder()
+        .with_span_processor(EnrichWithBaggageSpanProcessor)
         .with_simple_exporter(SpanExporter::default())
         .build();
 
     global::set_tracer_provider(provider.clone());
     provider
 }
 
+fn init_logs() -> SdkLoggerProvider {
+    // Setup tracerprovider with stdout exporter
+    // that prints the spans to stdout.
+    let logger_provider = SdkLoggerProvider::builder()
+        .with_log_processor(EnrichWithBaggageLogProcessor)
+        .with_simple_exporter(LogExporter::default())
+        .build();
+    let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider);
+    tracing_subscriber::registry().with(otel_layer).init();
+
+    logger_provider
+}
+
 #[tokio::main]
 async fn main() {
     use hyper_util::server::conn::auto::Builder;
 
     let provider = init_tracer();
+    let logger_provider = init_logs();
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
     let listener = TcpListener::bind(addr).await.unwrap();
 
@@ -124,4 +201,7 @@ async fn main() {
     }
 
     provider.shutdown().expect("Shutdown provider failed");
+    logger_provider
+        .shutdown()
+        .expect("Shutdown provider failed");
 }

opentelemetry-http/Cargo.toml

@@ -28,5 +28,10 @@ reqwest = { workspace = true, features = ["blocking"], optional = true }
 tokio = { workspace = true, features = ["time"], optional = true }
 tracing = {workspace = true, optional = true}
 
+[package.metadata.cargo-machete]
+ignored = [
+    "tracing" # needed for `internal-logs`
+]
+
 [lints]
 workspace = true

opentelemetry-jaeger-propagator/Cargo.toml

@@ -32,5 +32,10 @@ opentelemetry = { features = ["testing"], path = "../opentelemetry" }
 default = ["internal-logs"]
 internal-logs = ["tracing"]
 
+[package.metadata.cargo-machete]
+ignored = [
+    "tracing" # needed for `internal-logs`
+]
+
 [lints]
 workspace = true