From 7f78142a315270a25ea878582c98eae881d80f9c Mon Sep 17 00:00:00 2001 From: Igor Aleksanov Date: Thu, 17 Sep 2026 10:51:09 +0400 Subject: [PATCH] feat: Make opentelemetry optional --- .github/workflows/main.yml | 2 + tarpc/Cargo.toml | 15 +++- tarpc/src/context.rs | 65 +++++++------- tarpc/src/lib.rs | 2 + tarpc/src/otel.rs | 144 ++++++++++++++++++++++++++++++ tarpc/src/trace.rs | 64 +------------ tarpc/tests/service_functional.rs | 38 ++++++++ 7 files changed, 232 insertions(+), 98 deletions(-) create mode 100644 tarpc/src/otel.rs diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index bfb542f85..6646e1378 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -21,6 +21,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: + opentelemetry: ["", "--no-default-features"] serde: ["--features serde1", ""] tokio: ["--features tokio1", ""] serde-transport: ["--features serde-transport", ""] @@ -46,6 +47,7 @@ jobs: - uses: dtolnay/rust-toolchain@stable - run: > cargo test --manifest-path tarpc/Cargo.toml + ${{ matrix.opentelemetry }} ${{ matrix.serde }} ${{ matrix.tokio }} ${{ matrix.serde-transport }} ${{ matrix.serde-transport-json }} ${{ matrix.serde-transport-bincode }} ${{ matrix.tcp }} ${{ matrix.unix }} diff --git a/tarpc/Cargo.toml b/tarpc/Cargo.toml index 5fa719764..f22f1aaf5 100644 --- a/tarpc/Cargo.toml +++ b/tarpc/Cargo.toml @@ -17,7 +17,13 @@ readme = "README.md" description = "An RPC framework for Rust with a focus on ease of use." [features] -default = [] +default = ["opentelemetry"] + +opentelemetry = [ + "dep:opentelemetry", + "dep:opentelemetry-semantic-conventions", + "dep:tracing-opentelemetry", +] serde1 = ["tarpc-plugins/serde1", "serde", "serde/derive", "serde/rc"] tokio1 = ["tokio/rt"] @@ -28,6 +34,7 @@ tcp = ["tokio/net"] unix = ["tokio/net"] full = [ + "opentelemetry", "serde1", "tokio1", "serde-transport", @@ -57,9 +64,9 @@ tracing = { version = "0.1", default-features = false, features = [ "attributes", "log", ] } -tracing-opentelemetry = { version = "0.33", default-features = false } -opentelemetry = { version = "0.32", default-features = false } -opentelemetry-semantic-conventions = "0.32" +tracing-opentelemetry = { optional = true, version = "0.33", default-features = false } +opentelemetry = { optional = true, version = "0.32", default-features = false } +opentelemetry-semantic-conventions = { optional = true, version = "0.32" } [dev-dependencies] anyhow = "1.0" diff --git a/tarpc/src/context.rs b/tarpc/src/context.rs index fec15ac14..1ff949dc0 100644 --- a/tarpc/src/context.rs +++ b/tarpc/src/context.rs @@ -8,13 +8,11 @@ //! client to server and is used by the server to enforce response deadlines. use crate::trace::{self, TraceId}; -use opentelemetry::trace::TraceContextExt; use static_assertions::assert_impl_all; use std::{ convert::TryFrom, time::{Duration, Instant}, }; -use tracing_opentelemetry::OpenTelemetrySpanExt; /// A request context that carries request-scoped information like deadlines and trace information. /// It is sent from client to server and is used by the server to enforce response deadlines. @@ -102,15 +100,6 @@ pub fn current() -> Context { Context::current() } -#[derive(Clone)] -struct Deadline(Instant); - -impl Default for Deadline { - fn default() -> Self { - Self(ten_seconds_from_now()) - } -} - impl Context { /// Returns the context for the current request, or a default Context if no request is active. pub fn current() -> Self { @@ -118,12 +107,7 @@ impl Context { Self { trace_context: trace::Context::try_from(&span) .unwrap_or_else(|_| trace::Context::default()), - deadline: span - .context() - .get::() - .cloned() - .unwrap_or_default() - .0, + deadline: span.deadline().unwrap_or_else(ten_seconds_from_now), } } @@ -135,26 +119,43 @@ impl Context { /// An extension trait for [`tracing::Span`] for propagating tarpc Contexts. pub(crate) trait SpanExt { + fn trace_context(&self) -> Result; + fn deadline(&self) -> Option; /// Sets the given context on this span. Newly-created spans will be children of the given /// context's trace context. fn set_context(&self, context: &Context); } +#[cfg(not(feature = "opentelemetry"))] impl SpanExt for tracing::Span { - fn set_context(&self, context: &Context) { - // Explicitly ignore the returned result because it either means that the span has - // already started, or the Otel layer is not present, so we don't mind if the result - // is an error we silently ignore. - let _ = self.set_parent( - opentelemetry::Context::new() - .with_remote_span_context(opentelemetry::trace::SpanContext::new( - opentelemetry::trace::TraceId::from(context.trace_context.trace_id), - opentelemetry::trace::SpanId::from(context.trace_context.span_id), - opentelemetry::trace::TraceFlags::from(context.trace_context.sampling_decision), - true, - opentelemetry::trace::TraceState::default(), - )) - .with_value(Deadline(context.deadline)), - ); + fn trace_context(&self) -> Result { + Err(trace::NoActiveSpan) + } + + fn deadline(&self) -> Option { + None + } + + fn set_context(&self, _: &Context) { + // Plain tracing spans cannot store a distributed context; the request still carries it. + } +} + +#[cfg(test)] +mod tests { + use super::current; + use crate::trace; + use std::time::{Duration, Instant}; + + #[test] + fn current_without_subscriber() { + tracing::subscriber::with_default(tracing::subscriber::NoSubscriber::default(), || { + let earliest_deadline = Instant::now() + Duration::from_secs(10); + let context = current(); + let latest_deadline = Instant::now() + Duration::from_secs(10); + + assert_eq!(context.trace_context, trace::Context::default()); + assert!((earliest_deadline..=latest_deadline).contains(&context.deadline)); + }); } } diff --git a/tarpc/src/lib.rs b/tarpc/src/lib.rs index 7e1944305..4a97f46d6 100644 --- a/tarpc/src/lib.rs +++ b/tarpc/src/lib.rs @@ -244,6 +244,8 @@ pub use tarpc_plugins::service; pub(crate) mod cancellations; pub mod client; pub mod context; +#[cfg(feature = "opentelemetry")] +mod otel; pub mod server; pub mod transport; pub(crate) mod util; diff --git a/tarpc/src/otel.rs b/tarpc/src/otel.rs new file mode 100644 index 000000000..e72f8c168 --- /dev/null +++ b/tarpc/src/otel.rs @@ -0,0 +1,144 @@ +// Copyright 2026 Google LLC +// +// Use of this source code is governed by an MIT-style +// license that can be found in the LICENSE file or at +// https://opensource.org/licenses/MIT. + +use crate::{ + context::{self, SpanExt}, + trace::{Context, NoActiveSpan, SamplingDecision, SpanId, TraceId}, +}; +use opentelemetry::trace::TraceContextExt; +use std::time::Instant; +use tracing_opentelemetry::OpenTelemetrySpanExt; + +// The deadline travels with the OpenTelemetry context but is interpreted by tarpc. +struct Deadline(Instant); + +impl SpanExt for tracing::Span { + fn trace_context(&self) -> Result { + let context = self.context(); + if context.has_active_span() { + Ok(Context::from(context.span())) + } else { + Err(NoActiveSpan) + } + } + + fn deadline(&self) -> Option { + self.context().get::().map(|deadline| deadline.0) + } + + fn set_context(&self, context: &context::Context) { + // Explicitly ignore the returned result because it either means that the span has + // already started, or the Otel layer is not present, so we don't mind if the result + // is an error we silently ignore. + let _ = self.set_parent( + opentelemetry::Context::new() + .with_remote_span_context(opentelemetry::trace::SpanContext::new( + opentelemetry::trace::TraceId::from(context.trace_context.trace_id), + opentelemetry::trace::SpanId::from(context.trace_context.span_id), + opentelemetry::trace::TraceFlags::from(context.trace_context.sampling_decision), + true, + opentelemetry::trace::TraceState::default(), + )) + .with_value(Deadline(context.deadline)), + ); + } +} + +impl From for TraceId { + fn from(trace_id: opentelemetry::trace::TraceId) -> Self { + Self::from(u128::from_be_bytes(trace_id.to_bytes())) + } +} + +impl From for opentelemetry::trace::TraceId { + fn from(trace_id: TraceId) -> Self { + Self::from_bytes(u128::from(trace_id).to_be_bytes()) + } +} + +impl From for SpanId { + fn from(span_id: opentelemetry::trace::SpanId) -> Self { + Self::from(u64::from_be_bytes(span_id.to_bytes())) + } +} + +impl From for opentelemetry::trace::SpanId { + fn from(span_id: SpanId) -> Self { + Self::from_bytes(u64::from(span_id).to_be_bytes()) + } +} + +impl From> for Context { + fn from(span: opentelemetry::trace::SpanRef<'_>) -> Self { + let otel_ctx = span.span_context(); + Self { + trace_id: TraceId::from(otel_ctx.trace_id()), + span_id: SpanId::from(otel_ctx.span_id()), + sampling_decision: SamplingDecision::from(otel_ctx), + } + } +} + +impl From for opentelemetry::trace::TraceFlags { + fn from(decision: SamplingDecision) -> Self { + match decision { + SamplingDecision::Sampled => opentelemetry::trace::TraceFlags::SAMPLED, + SamplingDecision::Unsampled => opentelemetry::trace::TraceFlags::default(), + } + } +} + +impl From<&opentelemetry::trace::SpanContext> for SamplingDecision { + fn from(context: &opentelemetry::trace::SpanContext) -> Self { + if context.is_sampled() { + SamplingDecision::Sampled + } else { + SamplingDecision::Unsampled + } + } +} + +#[cfg(test)] +mod tests { + use super::SpanExt; + use crate::{context, trace}; + use opentelemetry::trace::TracerProvider as _; + use std::time::{Duration, Instant}; + use tracing_subscriber::layer::SubscriberExt; + + #[test] + fn current_inherits_request_context() { + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder().build(); + let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("tarpc-test"))); + + tracing::subscriber::with_default(subscriber, || { + let mut parent = context::current(); + parent.deadline = Instant::now() + Duration::from_secs(30); + parent.trace_context = trace::Context { + trace_id: 42.into(), + span_id: 7.into(), + sampling_decision: trace::SamplingDecision::Sampled, + }; + + let span = tracing::info_span!("RPC"); + span.set_context(&parent); + let _entered = span.enter(); + let current = context::current(); + + assert_eq!(current.deadline, parent.deadline); + assert_eq!( + current.trace_context.trace_id, + parent.trace_context.trace_id + ); + assert_eq!( + current.trace_context.sampling_decision, + parent.trace_context.sampling_decision + ); + assert!(!current.trace_context.span_id.is_none()); + }); + } +} diff --git a/tarpc/src/trace.rs b/tarpc/src/trace.rs index d78f960af..ee35a6948 100644 --- a/tarpc/src/trace.rs +++ b/tarpc/src/trace.rs @@ -16,14 +16,13 @@ //! This crate's design is based on [opencensus //! tracing](https://opencensus.io/core-concepts/tracing/). -use opentelemetry::trace::TraceContextExt; +use crate::context::SpanExt; use rand::{Rng, RngExt}; use std::{ convert::TryFrom, fmt::{self, Formatter}, num::{NonZeroU64, NonZeroU128}, }; -use tracing_opentelemetry::OpenTelemetrySpanExt; /// A context for tracing the execution of processes, distributed or otherwise. /// @@ -137,70 +136,11 @@ impl From for SpanId { } } -impl From for TraceId { - fn from(trace_id: opentelemetry::trace::TraceId) -> Self { - Self::from(u128::from_be_bytes(trace_id.to_bytes())) - } -} - -impl From for opentelemetry::trace::TraceId { - fn from(trace_id: TraceId) -> Self { - Self::from_bytes(u128::from(trace_id).to_be_bytes()) - } -} - -impl From for SpanId { - fn from(span_id: opentelemetry::trace::SpanId) -> Self { - Self::from(u64::from_be_bytes(span_id.to_bytes())) - } -} - -impl From for opentelemetry::trace::SpanId { - fn from(span_id: SpanId) -> Self { - Self::from_bytes(u64::from(span_id).to_be_bytes()) - } -} - impl TryFrom<&tracing::Span> for Context { type Error = NoActiveSpan; fn try_from(span: &tracing::Span) -> Result { - let context = span.context(); - if context.has_active_span() { - Ok(Self::from(context.span())) - } else { - Err(NoActiveSpan) - } - } -} - -impl From> for Context { - fn from(span: opentelemetry::trace::SpanRef<'_>) -> Self { - let otel_ctx = span.span_context(); - Self { - trace_id: TraceId::from(otel_ctx.trace_id()), - span_id: SpanId::from(otel_ctx.span_id()), - sampling_decision: SamplingDecision::from(otel_ctx), - } - } -} - -impl From for opentelemetry::trace::TraceFlags { - fn from(decision: SamplingDecision) -> Self { - match decision { - SamplingDecision::Sampled => opentelemetry::trace::TraceFlags::SAMPLED, - SamplingDecision::Unsampled => opentelemetry::trace::TraceFlags::default(), - } - } -} - -impl From<&opentelemetry::trace::SpanContext> for SamplingDecision { - fn from(context: &opentelemetry::trace::SpanContext) -> Self { - if context.is_sampled() { - SamplingDecision::Sampled - } else { - SamplingDecision::Unsampled - } + span.trace_context() } } diff --git a/tarpc/tests/service_functional.rs b/tarpc/tests/service_functional.rs index 06542b43b..683745664 100644 --- a/tarpc/tests/service_functional.rs +++ b/tarpc/tests/service_functional.rs @@ -8,6 +8,7 @@ use tarpc::{ client::{self}, context, server::{BaseChannel, Channel, incoming::Incoming}, + trace, transport::channel, }; use tokio::join; @@ -44,6 +45,43 @@ async fn sequential() { assert_eq!(client.call(context::current(), 1).await.unwrap(), 2); } +#[tokio::test] +async fn propagates_explicit_request_context() { + let (tx, rx) = channel::unbounded(); + let client = client::new(client::Config::default(), tx).spawn(); + tokio::spawn( + BaseChannel::with_defaults(rx) + .execute(tarpc::server::serve( + |context, _: ()| async move { Ok(context) }, + )) + .for_each(|response| response), + ); + + // Explicit contexts must propagate even when no OpenTelemetry subscriber is installed. + let mut context = context::current(); + context.deadline = Instant::now() + Duration::from_secs(30); + context.trace_context = trace::Context { + trace_id: 42.into(), + span_id: 7.into(), + sampling_decision: trace::SamplingDecision::Sampled, + }; + let received = client + .call(context, ()) + .await + .expect("the server echoes the request context"); + + assert_eq!(received.deadline, context.deadline); + assert_eq!( + received.trace_context.trace_id, + context.trace_context.trace_id + ); + assert_eq!( + received.trace_context.sampling_decision, + context.trace_context.sampling_decision + ); + assert!(!received.trace_context.span_id.is_none()); +} + #[tokio::test] async fn dropped_channel_aborts_in_flight_requests() -> anyhow::Result<()> { #[tarpc_plugins::service]