diff --git a/Cargo.lock b/Cargo.lock index f8675b9c65..16e633ab1d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4327,6 +4327,7 @@ dependencies = [ "openshell-bootstrap", "openshell-core", "openshell-crypto", + "openshell-otel", "openshell-policy", "openshell-providers", "openshell-sdk", diff --git a/crates/openshell-cli/Cargo.toml b/crates/openshell-cli/Cargo.toml index bc7266e4a4..1c33c90319 100644 --- a/crates/openshell-cli/Cargo.toml +++ b/crates/openshell-cli/Cargo.toml @@ -18,6 +18,7 @@ path = "src/main.rs" openshell-crypto = { workspace = true } openshell-bootstrap = { path = "../openshell-bootstrap" } openshell-core = { path = "../openshell-core", default-features = false } +openshell-otel = { path = "../openshell-otel", default-features = false } openshell-policy = { path = "../openshell-policy" } openshell-providers = { path = "../openshell-providers" } openshell-sdk = { path = "../openshell-sdk" } diff --git a/crates/openshell-cli/src/tls.rs b/crates/openshell-cli/src/tls.rs index 6dceb395de..2224094fdb 100644 --- a/crates/openshell-cli/src/tls.rs +++ b/crates/openshell-cli/src/tls.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use miette::{IntoDiagnostic, Result, WrapErr}; +use openshell_otel::EnvTraceContextInterceptor; use openshell_sdk::{AuthConfig, ClientConfig, ClientIdentity, OpenShellClient}; use rustls::{ RootCertStore, @@ -14,7 +15,32 @@ use tonic::metadata::AsciiMetadataValue; use tonic::transport::Channel; /// Concrete gRPC client type used by all commands. -pub type GrpcClient = openshell_sdk::raw::AuthedGrpcClient; +pub type GrpcClient = openshell_sdk::raw::GrpcClient< + tonic::service::interceptor::InterceptedService, +>; + +/// Interceptor stack applied to gateway requests from clients built by +/// [`grpc_client`]. +/// +/// Runs authentication header injection first, then passive W3C trace-context +/// forwarding so a CI pipeline's `TRACEPARENT` extends into the gateway's +/// spans. tonic allows a single interceptor per channel, so the two concerns +/// are composed here. +#[derive(Clone)] +pub struct GatewayInterceptor { + auth: openshell_sdk::EdgeAuthInterceptor, + trace: EnvTraceContextInterceptor, +} + +impl tonic::service::Interceptor for GatewayInterceptor { + fn call( + &mut self, + request: tonic::Request<()>, + ) -> std::result::Result, tonic::Status> { + let request = self.auth.call(request)?; + self.trace.call(request) + } +} #[derive(Clone, Debug, Default)] pub struct TlsOptions { @@ -305,9 +331,10 @@ pub async fn build_channel(server: &str, tls: &TlsOptions) -> Result { /// Build the SDK client and return its raw gRPC escape hatch. /// -/// When `tls.edge_token` is set, the returned client is wrapped with an -/// interceptor that injects authentication headers on every request. -/// Otherwise, standard mTLS is used (interceptor is a no-op). +/// The returned client carries a [`GatewayInterceptor`] that injects +/// authentication headers (when a token is set) and passively forwards the +/// `TRACEPARENT`/`TRACESTATE` trace context from the environment on every +/// request. With no token and no trace context, the interceptor is a no-op. pub async fn grpc_client(server: &str, tls: &TlsOptions) -> Result { let config = sdk_client_config(server, tls)?; // Keep the SDK's transport state machine out of every command future. @@ -315,7 +342,17 @@ pub async fn grpc_client(server: &str, tls: &TlsOptions) -> Result { let client = Box::pin(OpenShellClient::connect(config)) .await .into_diagnostic()?; - Ok(client.raw_grpc()) + // tonic allows a single interceptor per channel: compose the SDK's auth + // interceptor (sharing its live bearer slot) with the CLI's passive + // trace-context forwarding. + let interceptor = GatewayInterceptor { + auth: client.auth_interceptor(), + trace: EnvTraceContextInterceptor::from_env(), + }; + Ok(openshell_sdk::raw::GrpcClient::with_interceptor( + client.channel(), + interceptor, + )) } /// Gateway replica that owns the sandbox named in `response`, when it says. @@ -365,13 +402,61 @@ pub fn retry_unrouted(replica: Option<&AsciiMetadataValue>, status: &tonic::Stat #[cfg(test)] mod tests { use super::{ - TlsOptions, owner_replica, retry_unrouted, routed_to, sdk_client_config, with_owner_hint, + GatewayInterceptor, TlsOptions, owner_replica, retry_unrouted, routed_to, + sdk_client_config, with_owner_hint, }; + use crate::TEST_ENV_LOCK; use openshell_core::replica_routing::{ OWNER_REPLICA_HEADER, OWNER_REQUEST_HEADER, ROUTE_REPLICA_HEADER, }; - use openshell_sdk::AuthConfig; + use openshell_otel::EnvTraceContextInterceptor; + use openshell_sdk::{AuthConfig, EdgeAuthInterceptor}; use std::io::Write; + use tonic::service::Interceptor as _; + + #[test] + fn gateway_interceptor_composes_auth_and_trace_forwarding() { + let _guard = TEST_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + temp_env::with_vars( + [ + ("TRACEPARENT", Some(traceparent)), + ("TRACESTATE", Some("vendor=value")), + ], + || { + let mut interceptor = GatewayInterceptor { + auth: EdgeAuthInterceptor::new(Some("test-token"), None).unwrap(), + trace: EnvTraceContextInterceptor::from_env(), + }; + + let request = interceptor.call(tonic::Request::new(())).unwrap(); + + assert_eq!( + request + .metadata() + .get("authorization") + .and_then(|value| value.to_str().ok()), + Some("Bearer test-token") + ); + assert_eq!( + request + .metadata() + .get("traceparent") + .and_then(|value| value.to_str().ok()), + Some(traceparent) + ); + assert_eq!( + request + .metadata() + .get("tracestate") + .and_then(|value| value.to_str().ok()), + Some("vendor=value") + ); + }, + ); + } #[test] fn owner_hint_is_requested_explicitly() { diff --git a/crates/openshell-otel/Cargo.toml b/crates/openshell-otel/Cargo.toml index 3a85893cfb..ec188d67b0 100644 --- a/crates/openshell-otel/Cargo.toml +++ b/crates/openshell-otel/Cargo.toml @@ -11,25 +11,47 @@ license.workspace = true repository.workspace = true [dependencies] -futures = { workspace = true } +futures = { workspace = true, optional = true } http = { workspace = true } -metrics = { workspace = true } -openshell-ocsf = { path = "../openshell-ocsf" } +metrics = { workspace = true, optional = true } +openshell-ocsf = { path = "../openshell-ocsf", optional = true } opentelemetry = { workspace = true } -opentelemetry_sdk = { workspace = true, features = ["logs"] } -opentelemetry-otlp = { workspace = true, features = ["logs"] } -serde_json = { workspace = true } -thiserror = { workspace = true } -tokio = { workspace = true } +opentelemetry_sdk = { workspace = true } +opentelemetry-otlp = { workspace = true, optional = true } +serde_json = { workspace = true, optional = true } +thiserror = { workspace = true, optional = true } +tokio = { workspace = true, optional = true } tracing = { workspace = true } tracing-opentelemetry = { workspace = true } -tracing-subscriber = { workspace = true } +tracing-subscriber = { workspace = true, optional = true } tonic = { workspace = true } -tower-http = { workspace = true } +tower-http = { workspace = true, optional = true } [dev-dependencies] opentelemetry_sdk = { workspace = true, features = ["testing"] } tokio = { workspace = true, features = ["test-util"] } +tracing-subscriber = { workspace = true } + +[features] +default = ["export"] +# OTLP/gRPC export: tracer providers, tracing layers, compute-driver and +# gRPC tracing machinery, and the OCSF log bridge. Services that report to +# a collector keep the default; clients that only forward trace context +# (e.g. the CLI) opt out with `default-features = false` so the exporter +# stack stays out of their binaries. +export = [ + "dep:futures", + "dep:metrics", + "dep:openshell-ocsf", + "dep:opentelemetry-otlp", + "dep:serde_json", + "dep:thiserror", + "dep:tokio", + "dep:tower-http", + "dep:tracing-subscriber", + "opentelemetry_sdk/logs", + "opentelemetry-otlp/logs", +] [lints] workspace = true diff --git a/crates/openshell-otel/src/lib.rs b/crates/openshell-otel/src/lib.rs index 914bbea861..554e229d2f 100644 --- a/crates/openshell-otel/src/lib.rs +++ b/crates/openshell-otel/src/lib.rs @@ -2,384 +2,54 @@ // SPDX-License-Identifier: Apache-2.0 //! Shared OpenTelemetry trace and log export support for `OpenShell` services. +//! +//! The crate has two tiers: +//! +//! - Trace-context propagation, always compiled in: W3C `traceparent` / +//! `tracestate` extraction and injection for HTTP headers, tonic metadata, +//! and process environments. +//! - The `export` feature (on by default): OTLP/gRPC tracer providers, +//! tracing layers, compute-driver and gRPC tracing machinery, and the OCSF +//! log bridge. Services that report to a collector keep the default; +//! clients that only forward trace context (e.g. the CLI) opt out with +//! `default-features = false` to keep the exporter stack out of their +//! binaries. +mod propagation; + +pub use propagation::{ + EnvTraceContextInterceptor, HeaderMapExtractor, MetadataMapInjector, TRACEPARENT_ENV, + TRACESTATE_ENV, TraceContextInterceptor, current_trace_context_carrier, + current_trace_context_environment, set_parent_from_environment, set_parent_from_trace_context, + trace_context_environment, +}; + +#[cfg(feature = "export")] mod driver; +#[cfg(feature = "export")] mod grpc; +#[cfg(feature = "export")] mod logs; +#[cfg(feature = "export")] mod ocsf; -mod propagation; +#[cfg(feature = "export")] +mod setup; +#[cfg(feature = "export")] pub use driver::{ BoxGrpcStream, ComputeDriverTracing, DriverTracingConfig, DriverTracingHandle, IN_PROCESS_COMPUTE_DRIVER_TARGET, InProcessRpcTracer, install_driver_tracing, }; - +#[cfg(feature = "export")] pub use grpc::{ COMPUTE_DRIVER_RPC_SERVICE, ComputeDriverRpc, ComputeDriverRpcSpan, RecordGrpcFailure, RecordGrpcStatus, TracedGrpcStream, compute_driver_rpc_layer, compute_driver_rpc_operation, grpc_status_code_name, record_grpc_status, rpc, }; +#[cfg(feature = "export")] pub use ocsf::{OcsfOtlp, OcsfOtlpLayer}; -pub use propagation::{ - HeaderMapExtractor, MetadataMapInjector, TRACEPARENT_ENV, TRACESTATE_ENV, - TraceContextInterceptor, current_trace_context_carrier, current_trace_context_environment, - set_parent_from_environment, set_parent_from_trace_context, trace_context_environment, -}; - -use opentelemetry::KeyValue; -use opentelemetry::trace::TracerProvider as _; -pub use opentelemetry_otlp::LogExporter; -use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig; -use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithTonicConfig}; -use opentelemetry_sdk::Resource; -use opentelemetry_sdk::trace::SdkTracer; -pub use opentelemetry_sdk::trace::SdkTracerProvider; -use tracing::Subscriber; -use tracing_opentelemetry::OpenTelemetryLayer; -use tracing_subscriber::Layer as _; -use tracing_subscriber::layer::{Context, Filter}; -use tracing_subscriber::registry::LookupSpan; - -const SDK_UNKNOWN_SERVICE_PREFIX: &str = "unknown_service"; - -/// Build the resource attributes shared by gateway and compute-driver providers. -pub fn gateway_resource_attributes( - gateway_name: Option<&str>, - compute_driver: Option<&str>, -) -> Vec { - let mut attributes = Vec::new(); - if let Some(name) = gateway_name.map(str::trim).filter(|name| !name.is_empty()) { - attributes.push(KeyValue::new("openshell.gateway.name", name.to_string())); - } - if let Some(driver) = compute_driver - .map(str::trim) - .filter(|driver| !driver.is_empty()) - { - attributes.push(KeyValue::new( - "openshell.gateway.compute_driver", - driver.to_string(), - )); - } - attributes -} - -/// Mark `span` as failed. -/// -/// The field must be declared on the span at creation because `tracing` drops -/// records for fields a span does not have. -pub fn mark_error(span: &tracing::Span) { - span.record("otel.status_code", "ERROR"); -} - -/// Marks the current span when an instrumented operation returns an error. -pub fn record_error_result(result: Result) -> Result { - if result.is_err() { - mark_error(&tracing::Span::current()); - } - result -} - -/// Marks an instrumented function's span when it exits before returning success. -/// -/// Create the guard at the start of the instrumented function and return the -/// successful result through [`Self::finish`]. An early `?` or error return -/// drops the unfinished guard and marks the captured span as failed. -#[must_use] -pub struct ErrorStatusGuard { - span: tracing::Span, - finished: bool, -} - -impl ErrorStatusGuard { - /// Captures the current instrumented span. - pub fn current() -> Self { - Self { - span: tracing::Span::current(), - finished: false, - } - } - - /// Returns `result`, marking this guard complete when it is successful. - pub fn finish(mut self, result: Result) -> Result { - self.finished = result.is_ok(); - result - } -} - -impl Drop for ErrorStatusGuard { - fn drop(&mut self) { - if !self.finished { - mark_error(&self.span); - } - } -} - -/// How a process chooses its OpenTelemetry `service.name`. -#[derive(Debug, Clone, Copy)] -pub enum ServiceName<'a> { - /// Always use this name, overriding `OTEL_SERVICE_NAME`. - Fixed(&'a str), - /// Use `OTEL_SERVICE_NAME` when set, otherwise use this default. - EnvironmentOr(&'a str), -} - -/// Inputs for OTLP/gRPC export. -#[derive(Debug, Clone)] -pub struct OtlpExportConfig<'a> { - pub endpoint: &'a str, - pub service_name: ServiceName<'a>, - pub service_version: Option<&'a str>, - pub resource_attributes: Vec, -} - -/// Failure to set up OTLP export. -#[derive(Debug, thiserror::Error)] -pub enum SetupError { - #[error("OTLP endpoint is empty")] - EmptyEndpoint, - - #[error("invalid OTLP endpoint {endpoint:?}: {source}")] - InvalidEndpoint { - endpoint: String, - source: http::uri::InvalidUri, - }, - - #[error("failed to build the OTLP exporter: {0}")] - Exporter(#[from] opentelemetry_otlp::ExporterBuildError), -} - -fn resource_attributes(config: &OtlpExportConfig<'_>) -> Vec { - let mut attributes = config.resource_attributes.clone(); - if let Some(version) = config - .service_version - .map(str::trim) - .filter(|version| !version.is_empty()) - { - attributes.push(KeyValue::new("service.version", version.to_string())); - } - attributes -} - -/// Build the OpenTelemetry resource for an export configuration. -#[must_use] -pub fn resource_for(config: &OtlpExportConfig<'_>) -> Resource { - let attributes = resource_attributes(config); - match config.service_name { - ServiceName::Fixed(name) => Resource::builder() - .with_service_name(name.trim().to_string()) - .with_attributes(attributes) - .build(), - ServiceName::EnvironmentOr(default) => { - let detected = Resource::builder() - .with_attributes(attributes.clone()) - .build(); - if detected - .get(&opentelemetry::Key::from_static_str("service.name")) - .is_some_and(|value| !value.to_string().starts_with(SDK_UNKNOWN_SERVICE_PREFIX)) - { - detected - } else { - Resource::builder() - .with_service_name(default.trim().to_string()) - .with_attributes(attributes) - .build() - } - } - } -} - -/// Trim and validate an OTLP endpoint before handing it to an exporter. -fn validated_endpoint(endpoint: &str) -> Result<(&str, http::Uri), SetupError> { - let endpoint = endpoint.trim(); - if endpoint.is_empty() { - return Err(SetupError::EmptyEndpoint); - } - let uri = endpoint - .parse::() - .map_err(|source| SetupError::InvalidEndpoint { - endpoint: endpoint.to_string(), - source, - })?; - Ok((endpoint, uri)) -} - -/// A validated OTLP/gRPC collector endpoint and the resource that every -/// signal exported to it describes. -#[derive(Debug, Clone)] -pub struct OtlpTarget { - endpoint: String, - tls: bool, - resource: Resource, -} - -impl OtlpTarget { - /// Validate the endpoint and build the resource. - pub fn new(config: &OtlpExportConfig<'_>) -> Result { - let (endpoint, uri) = validated_endpoint(config.endpoint)?; - Ok(Self { - endpoint: endpoint.to_string(), - tls: uri.scheme() == Some(&http::uri::Scheme::HTTPS), - resource: resource_for(config), - }) - } - - /// Point an exporter builder at this target, over TLS for `https://`. - pub(crate) fn configure(&self, builder: B) -> B - where - B: WithExportConfig + WithTonicConfig, - { - let builder = builder.with_endpoint(self.endpoint.clone()); - if self.tls { - // Tonic trusts nothing by default, so enable the platform roots. - builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots()) - } else { - builder - } - } - - #[must_use] - pub fn endpoint(&self) -> &str { - &self.endpoint - } - - #[must_use] - pub fn resource(&self) -> &Resource { - &self.resource - } - - /// Build a trace provider that exports spans to this target in batches. - pub fn tracer_provider(&self) -> Result { - let exporter = self - .configure(SpanExporter::builder().with_tonic()) - .build()?; - Ok(SdkTracerProvider::builder() - .with_batch_exporter(exporter) - .with_resource(self.resource.clone()) - .build()) - } -} - -/// Build an OTLP/gRPC trace provider. -pub fn build_provider(config: &OtlpExportConfig<'_>) -> Result { - OtlpTarget::new(config)?.tracer_provider() -} - -/// Build the provider for an optional OTLP configuration. -/// -/// Telemetry setup failures disable export and remain available for the caller -/// to report after its tracing subscriber is installed. -#[must_use] -pub fn provider_for( - config: Option>, -) -> (Option, Option) { - match config.as_ref().map(build_provider) { - None => (None, None), - Some(Ok(provider)) => (Some(provider), None), - Some(Err(error)) => (None, Some(error)), - } -} - -/// Filtered OpenTelemetry layer returned by [`layer`]. -pub type OtlpLayer = tracing_subscriber::filter::Filtered< - OpenTelemetryLayer, - tracing_subscriber::filter::FilterFn, - S, ->; - -pub type TargetOtlpLayer = - tracing_subscriber::filter::Filtered, TargetPrefixFilter, S>; - -#[derive(Debug, Clone)] -pub struct TargetPrefixFilter { - prefixes: Vec<&'static str>, - include: bool, -} - -impl Filter for TargetPrefixFilter { - fn enabled(&self, metadata: &tracing::Metadata<'_>, _ctx: &Context<'_, S>) -> bool { - let matches_prefix = self - .prefixes - .iter() - .any(|prefix| metadata.target().starts_with(prefix)); - metadata.is_span() - && !metadata.target().starts_with("opentelemetry") - && (matches_prefix == self.include) - } -} - -/// Build a tracing layer that exports spans and excludes exporter callsites. -pub fn layer(provider: &SdkTracerProvider, instrumentation_scope: &'static str) -> OtlpLayer -where - S: Subscriber + for<'span> LookupSpan<'span>, -{ - tracing_opentelemetry::layer() - .with_tracer(provider.tracer(instrumentation_scope)) - .with_filter(tracing_subscriber::filter::filter_fn(|metadata| { - metadata.is_span() && !metadata.target().starts_with("opentelemetry") - })) -} - -/// Build a tracing layer that exports only spans from `target_prefix`. -pub fn layer_for_target_prefix( - provider: &SdkTracerProvider, - instrumentation_scope: &'static str, - target_prefix: &'static str, -) -> TargetOtlpLayer -where - S: Subscriber + for<'span> LookupSpan<'span>, -{ - layer_for_target_prefixes(provider, instrumentation_scope, [target_prefix]) -} - -/// Build a tracing layer that exports only spans from `target_prefixes`. -pub fn layer_for_target_prefixes( - provider: &SdkTracerProvider, - instrumentation_scope: &'static str, - target_prefixes: impl IntoIterator, -) -> TargetOtlpLayer -where - S: Subscriber + for<'span> LookupSpan<'span>, -{ - tracing_opentelemetry::layer() - .with_tracer(provider.tracer(instrumentation_scope)) - .with_filter(TargetPrefixFilter { - prefixes: target_prefixes.into_iter().collect(), - include: true, - }) -} - -/// Build a tracing layer that excludes spans from `target_prefix`. -/// -/// `None` excludes no application spans. -pub fn layer_excluding_target_prefix( - provider: &SdkTracerProvider, - instrumentation_scope: &'static str, - target_prefix: Option<&'static str>, -) -> TargetOtlpLayer -where - S: Subscriber + for<'span> LookupSpan<'span>, -{ - layer_excluding_target_prefixes(provider, instrumentation_scope, target_prefix) -} - -/// Build a tracing layer that excludes spans from `target_prefixes`. -/// -/// An empty iterator excludes no application spans. -pub fn layer_excluding_target_prefixes( - provider: &SdkTracerProvider, - instrumentation_scope: &'static str, - target_prefixes: impl IntoIterator, -) -> TargetOtlpLayer -where - S: Subscriber + for<'span> LookupSpan<'span>, -{ - tracing_opentelemetry::layer() - .with_tracer(provider.tracer(instrumentation_scope)) - .with_filter(TargetPrefixFilter { - prefixes: target_prefixes.into_iter().collect(), - include: false, - }) -} +#[cfg(feature = "export")] +pub use setup::*; #[cfg(test)] pub(crate) fn test_lock() -> std::sync::MutexGuard<'static, ()> { @@ -387,182 +57,3 @@ pub(crate) fn test_lock() -> std::sync::MutexGuard<'static, ()> { LOCK.lock() .unwrap_or_else(std::sync::PoisonError::into_inner) } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn error_status_guard_marks_only_unfinished_results() { - use tracing_subscriber::layer::SubscriberExt as _; - - let _tracing_lock = test_lock(); - let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); - let provider = SdkTracerProvider::builder() - .with_simple_exporter(exporter.clone()) - .build(); - let subscriber = tracing_subscriber::registry().with(layer(&provider, "guard-test")); - - tracing::subscriber::with_default(subscriber, || { - let failed = - tracing::info_span!("failed-operation", otel.status_code = tracing::field::Empty); - { - let _entered = failed.enter(); - drop(ErrorStatusGuard::current()); - } - drop(failed); - - let succeeded = tracing::info_span!( - "successful-operation", - otel.status_code = tracing::field::Empty - ); - { - let _entered = succeeded.enter(); - ErrorStatusGuard::current().finish(Ok::<_, ()>(())).unwrap(); - } - drop(succeeded); - }); - - let spans = exporter.get_finished_spans().unwrap(); - let failed = spans - .iter() - .find(|span| span.name == "failed-operation") - .unwrap(); - assert!(matches!( - failed.status, - opentelemetry::trace::Status::Error { .. } - )); - let succeeded = spans - .iter() - .find(|span| span.name == "successful-operation") - .unwrap(); - assert_eq!(succeeded.status, opentelemetry::trace::Status::Unset); - } - - #[test] - fn target_scoped_layers_partition_in_process_driver_spans() { - use tracing_subscriber::layer::SubscriberExt as _; - - let _tracing_lock = test_lock(); - let gateway_exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); - let gateway_provider = SdkTracerProvider::builder() - .with_simple_exporter(gateway_exporter.clone()) - .build(); - let driver_exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); - let driver_provider = SdkTracerProvider::builder() - .with_simple_exporter(driver_exporter.clone()) - .build(); - let subscriber = tracing_subscriber::registry() - .with(layer_excluding_target_prefix( - &gateway_provider, - "gateway", - Some("driver_target"), - )) - .with(layer_for_target_prefix( - &driver_provider, - "driver", - "driver_target", - )); - - tracing::subscriber::with_default(subscriber, || { - let gateway = tracing::info_span!(target: "gateway_target", "gateway.span"); - let _entered = gateway.enter(); - drop(tracing::info_span!(target: "driver_target::operation", "driver.span")); - }); - - gateway_provider.force_flush().unwrap(); - driver_provider.force_flush().unwrap(); - let gateway_spans = gateway_exporter.get_finished_spans().unwrap(); - let driver_spans = driver_exporter.get_finished_spans().unwrap(); - assert_eq!( - gateway_spans - .iter() - .map(|span| span.name.as_ref()) - .collect::>(), - ["gateway.span"] - ); - assert_eq!( - driver_spans - .iter() - .map(|span| span.name.as_ref()) - .collect::>(), - ["driver.span"] - ); - assert_eq!( - driver_spans[0].span_context.trace_id(), - gateway_spans[0].span_context.trace_id() - ); - assert_eq!( - driver_spans[0].parent_span_id, - gateway_spans[0].span_context.span_id() - ); - } - - #[test] - fn excluding_no_target_prefix_exports_all_application_spans() { - use tracing_subscriber::layer::SubscriberExt as _; - - let _tracing_lock = test_lock(); - let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); - let provider = SdkTracerProvider::builder() - .with_simple_exporter(exporter.clone()) - .build(); - let subscriber = tracing_subscriber::registry() - .with(layer_excluding_target_prefix(&provider, "gateway", None)); - - tracing::subscriber::with_default(subscriber, || { - drop(tracing::info_span!(target: "\0driver_target", "nul-prefixed.span")); - drop(tracing::info_span!(target: "gateway_target", "gateway.span")); - }); - - provider.force_flush().unwrap(); - let spans = exporter.get_finished_spans().unwrap(); - assert_eq!(spans.len(), 2); - assert!(spans.iter().any(|span| span.name == "nul-prefixed.span")); - assert!(spans.iter().any(|span| span.name == "gateway.span")); - } - - #[test] - fn resource_uses_fixed_service_identity_and_custom_attributes() { - let resource = resource_for(&OtlpExportConfig { - endpoint: "http://127.0.0.1:4317", - service_name: ServiceName::Fixed("openshell-driver-vm"), - service_version: Some("1.2.3"), - resource_attributes: vec![KeyValue::new("openshell.gateway.name", "vm-dev")], - }); - - assert_eq!( - resource - .get(&opentelemetry::Key::from_static_str("service.name")) - .map(|value| value.to_string()), - Some("openshell-driver-vm".to_string()) - ); - assert_eq!( - resource - .get(&opentelemetry::Key::from_static_str("service.version")) - .map(|value| value.to_string()), - Some("1.2.3".to_string()) - ); - assert_eq!( - resource - .get(&opentelemetry::Key::from_static_str( - "openshell.gateway.name", - )) - .map(|value| value.to_string()), - Some("vm-dev".to_string()) - ); - } - - #[tokio::test] - async fn malformed_endpoint_disables_export_with_a_reportable_error() { - let (provider, error) = provider_for(Some(OtlpExportConfig { - endpoint: "definitely not a url", - service_name: ServiceName::Fixed("test-service"), - service_version: None, - resource_attributes: Vec::new(), - })); - - assert!(provider.is_none()); - assert!(matches!(error, Some(SetupError::InvalidEndpoint { .. }))); - } -} diff --git a/crates/openshell-otel/src/propagation.rs b/crates/openshell-otel/src/propagation.rs index 02f2bdb120..5bb7ba5ecc 100644 --- a/crates/openshell-otel/src/propagation.rs +++ b/crates/openshell-otel/src/propagation.rs @@ -6,6 +6,7 @@ use std::collections::BTreeMap; use http::HeaderMap; +use opentelemetry::Context; use opentelemetry::propagation::{Extractor, Injector, TextMapPropagator}; use opentelemetry_sdk::propagation::TraceContextPropagator; use tracing_opentelemetry::OpenTelemetrySpanExt as _; @@ -112,12 +113,16 @@ pub fn current_trace_context_environment() -> Vec<(&'static str, String)> { /// Parent `span` under the trace context in `TRACEPARENT` and `TRACESTATE`. /// +/// Reads the environment the same way [`EnvTraceContextInterceptor`] does: +/// both spellings are accepted, whitespace is trimmed, and blank values are +/// treated as unset. +/// /// Leaves `span` unchanged when the environment carries no valid context. pub fn set_parent_from_environment(span: &tracing::Span) { set_parent_from_trace_context( span, - std::env::var(TRACEPARENT_ENV).ok().as_deref(), - std::env::var(TRACESTATE_ENV).ok().as_deref(), + read_env(TRACEPARENT_ENV).as_deref(), + read_env(TRACESTATE_ENV).as_deref(), ); } @@ -177,22 +182,142 @@ impl tonic::service::Interceptor for TraceContextInterceptor { } } +/// Reads W3C trace-context propagation fields from process environment +/// variables (`TRACEPARENT` and `TRACESTATE`). +/// +/// OpenTelemetry integrations in CI systems (for example the Jenkins +/// OpenTelemetry plugin or an OpenTelemetry GitHub Action) export these +/// variables to propagate the pipeline trace to child processes. +/// Both upper-case and lower-case spellings are accepted; blank values are +/// treated as unset. +#[derive(Debug, Clone, Default)] +struct EnvTraceContext { + traceparent: Option, + tracestate: Option, +} + +impl EnvTraceContext { + fn from_env() -> Self { + Self { + traceparent: read_env(TRACEPARENT_ENV), + tracestate: read_env(TRACESTATE_ENV), + } + } +} + +/// Read one environment variable, trimming whitespace and discarding blanks. +fn read_one(name: &str) -> Option { + std::env::var(name) + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +/// Read an environment variable by its upper-case name, falling back to the +/// lower-case spelling, trimming whitespace and discarding blank values. +/// +/// Each spelling is normalized before the fallback, so a blank upper-case +/// value never shadows a valid lower-case one. +fn read_env(upper: &str) -> Option { + read_one(upper).or_else(|| read_one(&upper.to_ascii_lowercase())) +} + +impl Extractor for EnvTraceContext { + fn get(&self, key: &str) -> Option<&str> { + match key { + "traceparent" => self.traceparent.as_deref(), + "tracestate" => self.tracestate.as_deref(), + _ => None, + } + } + + fn keys(&self) -> Vec<&str> { + let mut keys = Vec::new(); + if self.traceparent.is_some() { + keys.push("traceparent"); + } + if self.tracestate.is_some() { + keys.push("tracestate"); + } + keys + } +} + +/// Passively forwards a W3C trace context captured from the process +/// environment onto outbound tonic requests. +/// +/// Unlike [`TraceContextInterceptor`], this does not require an active +/// OpenTelemetry span or an `SdkTracerProvider`; it reads `TRACEPARENT` +/// (and optionally `TRACESTATE`) once at construction and injects them as +/// gRPC metadata. This lets short-lived clients (e.g. the CLI on a CI runner) +/// extend a pipeline trace to the gateway without any collector configuration. +/// +/// When `TRACEPARENT` is unset or invalid the captured context carries no +/// valid span, so injection is a no-op. +#[derive(Debug, Clone)] +pub struct EnvTraceContextInterceptor { + context: Context, +} + +impl EnvTraceContextInterceptor { + /// Capture the W3C trace context from the process environment. + #[must_use] + pub fn from_env() -> Self { + let env = EnvTraceContext::from_env(); + let context = context_from(&env); + if env.traceparent.is_some() && !span_context_is_valid(&context) { + tracing::debug!("TRACEPARENT is set but invalid; not forwarding trace context"); + } + Self { context } + } +} + +/// Extract a [`Context`] from an [`Extractor`] using a fresh base context, so +/// the result depends only on the extractor and not on any ambient context. +fn context_from(extractor: &impl Extractor) -> Context { + TraceContextPropagator::new().extract_with_context(&Context::new(), extractor) +} + +fn span_context_is_valid(context: &Context) -> bool { + use opentelemetry::trace::TraceContextExt as _; + + context.span().span_context().is_valid() +} + +impl tonic::service::Interceptor for EnvTraceContextInterceptor { + fn call( + &mut self, + mut request: tonic::Request<()>, + ) -> Result, tonic::Status> { + TraceContextPropagator::new().inject_context( + &self.context, + &mut MetadataMapInjector::new(request.metadata_mut()), + ); + Ok(request) + } +} + #[cfg(test)] mod tests { use super::*; + use tonic::service::Interceptor as _; fn env_test_exporter() -> ( opentelemetry_sdk::trace::SdkTracerProvider, opentelemetry_sdk::trace::InMemorySpanExporter, impl tracing::Subscriber + Send + Sync, ) { + use opentelemetry::trace::TracerProvider as _; use tracing_subscriber::layer::SubscriberExt as _; let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() .with_simple_exporter(exporter.clone()) .build(); - let subscriber = tracing_subscriber::registry().with(crate::layer(&provider, "env-test")); + // Built directly from `tracing_opentelemetry` so the test runs even + // without the `export` feature. + let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("env-test"))); (provider, exporter, subscriber) } @@ -263,6 +388,7 @@ mod tests { #[test] fn invalid_environment_carrier_leaves_the_span_unparented() { + use opentelemetry::trace::TracerProvider as _; use tracing_subscriber::layer::SubscriberExt as _; let _tracing_lock = crate::test_lock(); @@ -270,7 +396,8 @@ mod tests { let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() .with_simple_exporter(exporter.clone()) .build(); - let subscriber = tracing_subscriber::registry().with(crate::layer(&provider, "env-test")); + let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("env-test"))); tracing::subscriber::with_default(subscriber, || { let span = tracing::info_span!("orphan"); @@ -285,12 +412,14 @@ mod tests { #[test] fn interceptor_adds_traceparent_only_inside_an_exported_span() { + use opentelemetry::trace::TracerProvider as _; use tonic::service::Interceptor as _; use tracing_subscriber::layer::SubscriberExt as _; let _tracing_lock = crate::test_lock(); let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder().build(); - let subscriber = tracing_subscriber::registry().with(crate::layer(&provider, "rpc-test")); + let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("rpc-test"))); let (outside, inside) = tracing::subscriber::with_default(subscriber, || { let outside = TraceContextInterceptor @@ -330,4 +459,147 @@ mod tests { Some("value") ); } + + fn interceptor_with(extractor: EnvTraceContext) -> EnvTraceContextInterceptor { + EnvTraceContextInterceptor { + context: context_from(&extractor), + } + } + + /// Restores snapshotted environment variables when dropped, so a failed + /// assertion cannot leak them into later tests sharing the process. + struct EnvRestore(Vec<(&'static str, Option)>); + + impl EnvRestore { + fn snapshot(names: &[&'static str]) -> Self { + Self( + names + .iter() + .map(|name| (*name, std::env::var(name).ok())) + .collect(), + ) + } + } + + impl Drop for EnvRestore { + #[allow(unsafe_code)] + fn drop(&mut self) { + for (name, value) in &self.0 { + unsafe { + match value { + Some(value) => std::env::set_var(name, value), + None => std::env::remove_var(name), + } + } + } + } + } + + #[test] + #[allow(unsafe_code)] + fn env_trace_context_from_env_trims_and_skips_blanks() { + let _tracing_lock = crate::test_lock(); + let _restore = EnvRestore::snapshot(&[TRACEPARENT_ENV, TRACESTATE_ENV]); + let traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + + // Upper-case spelling with surrounding whitespace. + unsafe { + std::env::set_var(TRACEPARENT_ENV, format!(" {traceparent} ")); + std::env::set_var(TRACESTATE_ENV, " vendor=value "); + } + let padded = EnvTraceContext::from_env(); + assert_eq!(padded.traceparent.as_deref(), Some(traceparent)); + assert_eq!(padded.tracestate.as_deref(), Some("vendor=value")); + + // Blank values are treated as unset. + unsafe { + std::env::set_var(TRACEPARENT_ENV, " "); + std::env::set_var(TRACESTATE_ENV, ""); + } + let blank = EnvTraceContext::from_env(); + assert_eq!(blank.traceparent, None); + assert_eq!(blank.tracestate, None); + } + + /// Windows treats environment names case-insensitively, so the lower-case + /// fallback cannot be exercised there: `TRACEPARENT` and `traceparent` + /// are the same variable. + #[test] + #[cfg(not(windows))] + #[allow(unsafe_code)] + fn env_trace_context_from_env_reads_lower_case_spelling() { + let _tracing_lock = crate::test_lock(); + let _restore = + EnvRestore::snapshot(&[TRACEPARENT_ENV, "traceparent", TRACESTATE_ENV, "tracestate"]); + let traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + + // Lower-case-only spelling. + unsafe { + std::env::remove_var(TRACEPARENT_ENV); + std::env::remove_var(TRACESTATE_ENV); + std::env::set_var("traceparent", traceparent); + std::env::set_var("tracestate", "vendor=value"); + } + let lower = EnvTraceContext::from_env(); + assert_eq!(lower.traceparent.as_deref(), Some(traceparent)); + assert_eq!(lower.tracestate.as_deref(), Some("vendor=value")); + + // A blank upper-case value must not shadow a valid lower-case one. + unsafe { + std::env::set_var(TRACEPARENT_ENV, " "); + std::env::remove_var("tracestate"); + } + let shadowed = EnvTraceContext::from_env(); + assert_eq!(shadowed.traceparent.as_deref(), Some(traceparent)); + assert_eq!(shadowed.tracestate, None); + } + + #[test] + fn env_interceptor_forwards_valid_traceparent() { + let traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + let mut interceptor = interceptor_with(EnvTraceContext { + traceparent: Some(traceparent.to_string()), + tracestate: Some("vendor=value".to_string()), + }); + + let request = interceptor.call(tonic::Request::new(())).unwrap(); + + assert_eq!( + request + .metadata() + .get("traceparent") + .and_then(|value| value.to_str().ok()), + Some(traceparent) + ); + assert_eq!( + request + .metadata() + .get("tracestate") + .and_then(|value| value.to_str().ok()), + Some("vendor=value") + ); + } + + #[test] + fn env_interceptor_is_noop_without_traceparent() { + let mut interceptor = interceptor_with(EnvTraceContext::default()); + + let request = interceptor.call(tonic::Request::new(())).unwrap(); + + assert!(request.metadata().get("traceparent").is_none()); + assert!(request.metadata().get("tracestate").is_none()); + } + + #[test] + fn env_interceptor_is_noop_for_invalid_traceparent() { + let mut interceptor = interceptor_with(EnvTraceContext { + traceparent: Some("not-a-valid-traceparent".to_string()), + tracestate: Some("vendor=value".to_string()), + }); + + let request = interceptor.call(tonic::Request::new(())).unwrap(); + + assert!(request.metadata().get("traceparent").is_none()); + assert!(request.metadata().get("tracestate").is_none()); + } } diff --git a/crates/openshell-otel/src/setup.rs b/crates/openshell-otel/src/setup.rs new file mode 100644 index 0000000000..9d2b2ed6ab --- /dev/null +++ b/crates/openshell-otel/src/setup.rs @@ -0,0 +1,540 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OTLP/gRPC export setup: resources, providers, and tracing layers. +//! +//! Compiled only with the `export` feature; see the crate-level documentation. + +use opentelemetry::KeyValue; +use opentelemetry::trace::TracerProvider as _; +pub use opentelemetry_otlp::LogExporter; +use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig; +use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithTonicConfig}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::trace::SdkTracer; +pub use opentelemetry_sdk::trace::SdkTracerProvider; +use tracing::Subscriber; +use tracing_opentelemetry::OpenTelemetryLayer; +use tracing_subscriber::Layer as _; +use tracing_subscriber::layer::{Context, Filter}; +use tracing_subscriber::registry::LookupSpan; + +const SDK_UNKNOWN_SERVICE_PREFIX: &str = "unknown_service"; + +/// Build the resource attributes shared by gateway and compute-driver providers. +pub fn gateway_resource_attributes( + gateway_name: Option<&str>, + compute_driver: Option<&str>, +) -> Vec { + let mut attributes = Vec::new(); + if let Some(name) = gateway_name.map(str::trim).filter(|name| !name.is_empty()) { + attributes.push(KeyValue::new("openshell.gateway.name", name.to_string())); + } + if let Some(driver) = compute_driver + .map(str::trim) + .filter(|driver| !driver.is_empty()) + { + attributes.push(KeyValue::new( + "openshell.gateway.compute_driver", + driver.to_string(), + )); + } + attributes +} + +/// Mark `span` as failed. +/// +/// The field must be declared on the span at creation because `tracing` drops +/// records for fields a span does not have. +pub fn mark_error(span: &tracing::Span) { + span.record("otel.status_code", "ERROR"); +} + +/// Marks the current span when an instrumented operation returns an error. +pub fn record_error_result(result: Result) -> Result { + if result.is_err() { + mark_error(&tracing::Span::current()); + } + result +} + +/// Marks an instrumented function's span when it exits before returning success. +/// +/// Create the guard at the start of the instrumented function and return the +/// successful result through [`Self::finish`]. An early `?` or error return +/// drops the unfinished guard and marks the captured span as failed. +#[must_use] +pub struct ErrorStatusGuard { + span: tracing::Span, + finished: bool, +} + +impl ErrorStatusGuard { + /// Captures the current instrumented span. + pub fn current() -> Self { + Self { + span: tracing::Span::current(), + finished: false, + } + } + + /// Returns `result`, marking this guard complete when it is successful. + pub fn finish(mut self, result: Result) -> Result { + self.finished = result.is_ok(); + result + } +} + +impl Drop for ErrorStatusGuard { + fn drop(&mut self) { + if !self.finished { + mark_error(&self.span); + } + } +} + +/// How a process chooses its OpenTelemetry `service.name`. +#[derive(Debug, Clone, Copy)] +pub enum ServiceName<'a> { + /// Always use this name, overriding `OTEL_SERVICE_NAME`. + Fixed(&'a str), + /// Use `OTEL_SERVICE_NAME` when set, otherwise use this default. + EnvironmentOr(&'a str), +} + +/// Inputs for OTLP/gRPC export. +#[derive(Debug, Clone)] +pub struct OtlpExportConfig<'a> { + pub endpoint: &'a str, + pub service_name: ServiceName<'a>, + pub service_version: Option<&'a str>, + pub resource_attributes: Vec, +} + +/// Failure to set up OTLP export. +#[derive(Debug, thiserror::Error)] +pub enum SetupError { + #[error("OTLP endpoint is empty")] + EmptyEndpoint, + + #[error("invalid OTLP endpoint {endpoint:?}: {source}")] + InvalidEndpoint { + endpoint: String, + source: http::uri::InvalidUri, + }, + + #[error("failed to build the OTLP exporter: {0}")] + Exporter(#[from] opentelemetry_otlp::ExporterBuildError), +} + +fn resource_attributes(config: &OtlpExportConfig<'_>) -> Vec { + let mut attributes = config.resource_attributes.clone(); + if let Some(version) = config + .service_version + .map(str::trim) + .filter(|version| !version.is_empty()) + { + attributes.push(KeyValue::new("service.version", version.to_string())); + } + attributes +} + +/// Build the OpenTelemetry resource for an export configuration. +#[must_use] +pub fn resource_for(config: &OtlpExportConfig<'_>) -> Resource { + let attributes = resource_attributes(config); + match config.service_name { + ServiceName::Fixed(name) => Resource::builder() + .with_service_name(name.trim().to_string()) + .with_attributes(attributes) + .build(), + ServiceName::EnvironmentOr(default) => { + let detected = Resource::builder() + .with_attributes(attributes.clone()) + .build(); + if detected + .get(&opentelemetry::Key::from_static_str("service.name")) + .is_some_and(|value| !value.to_string().starts_with(SDK_UNKNOWN_SERVICE_PREFIX)) + { + detected + } else { + Resource::builder() + .with_service_name(default.trim().to_string()) + .with_attributes(attributes) + .build() + } + } + } +} + +/// Trim and validate an OTLP endpoint before handing it to an exporter. +fn validated_endpoint(endpoint: &str) -> Result<(&str, http::Uri), SetupError> { + let endpoint = endpoint.trim(); + if endpoint.is_empty() { + return Err(SetupError::EmptyEndpoint); + } + let uri = endpoint + .parse::() + .map_err(|source| SetupError::InvalidEndpoint { + endpoint: endpoint.to_string(), + source, + })?; + Ok((endpoint, uri)) +} + +/// A validated OTLP/gRPC collector endpoint and the resource that every +/// signal exported to it describes. +#[derive(Debug, Clone)] +pub struct OtlpTarget { + endpoint: String, + tls: bool, + resource: Resource, +} + +impl OtlpTarget { + /// Validate the endpoint and build the resource. + pub fn new(config: &OtlpExportConfig<'_>) -> Result { + let (endpoint, uri) = validated_endpoint(config.endpoint)?; + Ok(Self { + endpoint: endpoint.to_string(), + tls: uri.scheme() == Some(&http::uri::Scheme::HTTPS), + resource: resource_for(config), + }) + } + + /// Point an exporter builder at this target, over TLS for `https://`. + pub(crate) fn configure(&self, builder: B) -> B + where + B: WithExportConfig + WithTonicConfig, + { + let builder = builder.with_endpoint(self.endpoint.clone()); + if self.tls { + // Tonic trusts nothing by default, so enable the platform roots. + builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots()) + } else { + builder + } + } + + #[must_use] + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + #[must_use] + pub fn resource(&self) -> &Resource { + &self.resource + } + + /// Build a trace provider that exports spans to this target in batches. + pub fn tracer_provider(&self) -> Result { + let exporter = self + .configure(SpanExporter::builder().with_tonic()) + .build()?; + Ok(SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_resource(self.resource.clone()) + .build()) + } +} + +/// Build an OTLP/gRPC trace provider. +pub fn build_provider(config: &OtlpExportConfig<'_>) -> Result { + OtlpTarget::new(config)?.tracer_provider() +} + +/// Build the provider for an optional OTLP configuration. +/// +/// Telemetry setup failures disable export and remain available for the caller +/// to report after its tracing subscriber is installed. +#[must_use] +pub fn provider_for( + config: Option>, +) -> (Option, Option) { + match config.as_ref().map(build_provider) { + None => (None, None), + Some(Ok(provider)) => (Some(provider), None), + Some(Err(error)) => (None, Some(error)), + } +} + +/// Filtered OpenTelemetry layer returned by [`layer`]. +pub type OtlpLayer = tracing_subscriber::filter::Filtered< + OpenTelemetryLayer, + tracing_subscriber::filter::FilterFn, + S, +>; + +pub type TargetOtlpLayer = + tracing_subscriber::filter::Filtered, TargetPrefixFilter, S>; + +#[derive(Debug, Clone)] +pub struct TargetPrefixFilter { + prefixes: Vec<&'static str>, + include: bool, +} + +impl Filter for TargetPrefixFilter { + fn enabled(&self, metadata: &tracing::Metadata<'_>, _ctx: &Context<'_, S>) -> bool { + let matches_prefix = self + .prefixes + .iter() + .any(|prefix| metadata.target().starts_with(prefix)); + metadata.is_span() + && !metadata.target().starts_with("opentelemetry") + && (matches_prefix == self.include) + } +} + +/// Build a tracing layer that exports spans and excludes exporter callsites. +pub fn layer(provider: &SdkTracerProvider, instrumentation_scope: &'static str) -> OtlpLayer +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + tracing_opentelemetry::layer() + .with_tracer(provider.tracer(instrumentation_scope)) + .with_filter(tracing_subscriber::filter::filter_fn(|metadata| { + metadata.is_span() && !metadata.target().starts_with("opentelemetry") + })) +} + +/// Build a tracing layer that exports only spans from `target_prefix`. +pub fn layer_for_target_prefix( + provider: &SdkTracerProvider, + instrumentation_scope: &'static str, + target_prefix: &'static str, +) -> TargetOtlpLayer +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + layer_for_target_prefixes(provider, instrumentation_scope, [target_prefix]) +} + +/// Build a tracing layer that exports only spans from `target_prefixes`. +pub fn layer_for_target_prefixes( + provider: &SdkTracerProvider, + instrumentation_scope: &'static str, + target_prefixes: impl IntoIterator, +) -> TargetOtlpLayer +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + tracing_opentelemetry::layer() + .with_tracer(provider.tracer(instrumentation_scope)) + .with_filter(TargetPrefixFilter { + prefixes: target_prefixes.into_iter().collect(), + include: true, + }) +} + +/// Build a tracing layer that excludes spans from `target_prefix`. +/// +/// `None` excludes no application spans. +pub fn layer_excluding_target_prefix( + provider: &SdkTracerProvider, + instrumentation_scope: &'static str, + target_prefix: Option<&'static str>, +) -> TargetOtlpLayer +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + layer_excluding_target_prefixes(provider, instrumentation_scope, target_prefix) +} + +/// Build a tracing layer that excludes spans from `target_prefixes`. +/// +/// An empty iterator excludes no application spans. +pub fn layer_excluding_target_prefixes( + provider: &SdkTracerProvider, + instrumentation_scope: &'static str, + target_prefixes: impl IntoIterator, +) -> TargetOtlpLayer +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + tracing_opentelemetry::layer() + .with_tracer(provider.tracer(instrumentation_scope)) + .with_filter(TargetPrefixFilter { + prefixes: target_prefixes.into_iter().collect(), + include: false, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn error_status_guard_marks_only_unfinished_results() { + use tracing_subscriber::layer::SubscriberExt as _; + + let _tracing_lock = crate::test_lock(); + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry().with(layer(&provider, "guard-test")); + + tracing::subscriber::with_default(subscriber, || { + let failed = + tracing::info_span!("failed-operation", otel.status_code = tracing::field::Empty); + { + let _entered = failed.enter(); + drop(ErrorStatusGuard::current()); + } + drop(failed); + + let succeeded = tracing::info_span!( + "successful-operation", + otel.status_code = tracing::field::Empty + ); + { + let _entered = succeeded.enter(); + ErrorStatusGuard::current().finish(Ok::<_, ()>(())).unwrap(); + } + drop(succeeded); + }); + + let spans = exporter.get_finished_spans().unwrap(); + let failed = spans + .iter() + .find(|span| span.name == "failed-operation") + .unwrap(); + assert!(matches!( + failed.status, + opentelemetry::trace::Status::Error { .. } + )); + let succeeded = spans + .iter() + .find(|span| span.name == "successful-operation") + .unwrap(); + assert_eq!(succeeded.status, opentelemetry::trace::Status::Unset); + } + + #[test] + fn target_scoped_layers_partition_in_process_driver_spans() { + use tracing_subscriber::layer::SubscriberExt as _; + + let _tracing_lock = crate::test_lock(); + let gateway_exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let gateway_provider = SdkTracerProvider::builder() + .with_simple_exporter(gateway_exporter.clone()) + .build(); + let driver_exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let driver_provider = SdkTracerProvider::builder() + .with_simple_exporter(driver_exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(layer_excluding_target_prefix( + &gateway_provider, + "gateway", + Some("driver_target"), + )) + .with(layer_for_target_prefix( + &driver_provider, + "driver", + "driver_target", + )); + + tracing::subscriber::with_default(subscriber, || { + let gateway = tracing::info_span!(target: "gateway_target", "gateway.span"); + let _entered = gateway.enter(); + drop(tracing::info_span!(target: "driver_target::operation", "driver.span")); + }); + + gateway_provider.force_flush().unwrap(); + driver_provider.force_flush().unwrap(); + let gateway_spans = gateway_exporter.get_finished_spans().unwrap(); + let driver_spans = driver_exporter.get_finished_spans().unwrap(); + assert_eq!( + gateway_spans + .iter() + .map(|span| span.name.as_ref()) + .collect::>(), + ["gateway.span"] + ); + assert_eq!( + driver_spans + .iter() + .map(|span| span.name.as_ref()) + .collect::>(), + ["driver.span"] + ); + assert_eq!( + driver_spans[0].span_context.trace_id(), + gateway_spans[0].span_context.trace_id() + ); + assert_eq!( + driver_spans[0].parent_span_id, + gateway_spans[0].span_context.span_id() + ); + } + + #[test] + fn excluding_no_target_prefix_exports_all_application_spans() { + use tracing_subscriber::layer::SubscriberExt as _; + + let _tracing_lock = crate::test_lock(); + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(layer_excluding_target_prefix(&provider, "gateway", None)); + + tracing::subscriber::with_default(subscriber, || { + drop(tracing::info_span!(target: "\0driver_target", "nul-prefixed.span")); + drop(tracing::info_span!(target: "gateway_target", "gateway.span")); + }); + + provider.force_flush().unwrap(); + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 2); + assert!(spans.iter().any(|span| span.name == "nul-prefixed.span")); + assert!(spans.iter().any(|span| span.name == "gateway.span")); + } + + #[test] + fn resource_uses_fixed_service_identity_and_custom_attributes() { + let resource = resource_for(&OtlpExportConfig { + endpoint: "http://127.0.0.1:4317", + service_name: ServiceName::Fixed("openshell-driver-vm"), + service_version: Some("1.2.3"), + resource_attributes: vec![KeyValue::new("openshell.gateway.name", "vm-dev")], + }); + + assert_eq!( + resource + .get(&opentelemetry::Key::from_static_str("service.name")) + .map(|value| value.to_string()), + Some("openshell-driver-vm".to_string()) + ); + assert_eq!( + resource + .get(&opentelemetry::Key::from_static_str("service.version")) + .map(|value| value.to_string()), + Some("1.2.3".to_string()) + ); + assert_eq!( + resource + .get(&opentelemetry::Key::from_static_str( + "openshell.gateway.name", + )) + .map(|value| value.to_string()), + Some("vm-dev".to_string()) + ); + } + + #[tokio::test] + async fn malformed_endpoint_disables_export_with_a_reportable_error() { + let (provider, error) = provider_for(Some(OtlpExportConfig { + endpoint: "definitely not a url", + service_name: ServiceName::Fixed("test-service"), + service_version: None, + resource_attributes: Vec::new(), + })); + + assert!(provider.is_none()); + assert!(matches!(error, Some(SetupError::InvalidEndpoint { .. }))); + } +} diff --git a/crates/openshell-sdk/src/client.rs b/crates/openshell-sdk/src/client.rs index d0904245bc..7da715f890 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -83,6 +83,16 @@ impl OpenShellClient { self.channel.clone() } + /// Clone of the auth interceptor backing [`OpenShellClient::raw_grpc`]. + /// + /// tonic allows a single interceptor per channel, so callers that need + /// to stack another concern (e.g. trace-context forwarding) on top of + /// authentication can compose around this clone — it shares the same + /// live bearer slot, so OIDC refresh still reaches the composed stack. + pub fn auth_interceptor(&self) -> EdgeAuthInterceptor { + self.interceptor.clone() + } + /// Authenticated gRPC client for the main `OpenShell` service. /// /// Use this when the curated surface below doesn't expose the RPC or @@ -1700,6 +1710,31 @@ mod tests { ); } + #[tokio::test] + async fn auth_interceptor_clone_sees_refreshed_token() { + use tonic::service::Interceptor as _; + + // The clone handed out by `auth_interceptor()` must share the live + // bearer slot: callers compose other concerns (e.g. the CLI's + // trace-forwarding gateway interceptor) around it, and a forced OIDC + // refresh must reach those already-built stacks. + let calls = Arc::new(AtomicUsize::new(0)); + let client = oidc_client_with_refresher(Arc::clone(&calls)); + let mut interceptor = client.auth_interceptor(); + + assert!(client.force_refresh().await.unwrap()); + + let request = interceptor.call(tonic::Request::new(())).unwrap(); + assert_eq!( + request + .metadata() + .get("authorization") + .and_then(|value| value.to_str().ok()), + Some("Bearer token-1"), + "refreshed token must reach the cloned interceptor" + ); + } + #[test] fn store_bearer_rejects_malformed_token_and_keeps_previous() { let slot: BearerSlot = Arc::new(RwLock::new(None)); diff --git a/docs/observability/traces.mdx b/docs/observability/traces.mdx index ef8bd6f5f8..30f5c52a61 100644 --- a/docs/observability/traces.mdx +++ b/docs/observability/traces.mdx @@ -69,6 +69,8 @@ Give each installation that reports to the same collector its own name, includin The gateway continues a valid W3C `traceparent` sent by the caller, so a client's trace can include the gateway's spans for its requests. Without one, the gateway starts a new trace. +The CLI forwards `TRACEPARENT` and `TRACESTATE` from its environment on gateway requests from its standard commands. A CI pipeline instrumented with an OpenTelemetry integration that exports these variables (for example the Jenkins OpenTelemetry plugin or an OpenTelemetry GitHub Action) therefore parents the gateway's spans for the commands it runs, with no collector or provider configuration on the CI runner. + The gateway also passes its trace context to compute drivers, so driver spans appear in the same trace as the gateway request that caused them. Built-in drivers export their spans through the gateway's process. Managed external drivers receive the trace endpoint, gateway name, and compute driver from the gateway. Drivers that you run yourself own their telemetry configuration. ## Next Steps