diff --git a/Cargo.lock b/Cargo.lock index 94cc7b3df6..d178930a4a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4388,6 +4388,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 0cdd32ade0..d7e1c0b8e4 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 41381f9a40..fd138aae5b 100644 --- a/crates/openshell-cli/src/tls.rs +++ b/crates/openshell-cli/src/tls.rs @@ -5,6 +5,7 @@ use miette::{IntoDiagnostic, Result, WrapErr}; pub use openshell_core::replica_routing::{ owner_replica, retry_unrouted, routed_to, with_owner_hint, }; +use openshell_otel::EnvTraceContextInterceptor; use openshell_sdk::{AuthConfig, ClientConfig, ClientIdentity, OpenShellClient}; use rustls::{ RootCertStore, @@ -16,7 +17,32 @@ use std::path::PathBuf; 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 { @@ -313,9 +339,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. @@ -323,14 +350,71 @@ 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, + )) } #[cfg(test)] mod tests { - use super::{TlsOptions, sdk_client_config}; - use openshell_sdk::AuthConfig; + use super::{GatewayInterceptor, TlsOptions, sdk_client_config}; + use crate::TEST_ENV_LOCK; + 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 sdk_config_carries_cli_mtls_materials() { diff --git a/crates/openshell-otel/Cargo.toml b/crates/openshell-otel/Cargo.toml index c6cd0111c9..a857018b91 100644 --- a/crates/openshell-otel/Cargo.toml +++ b/crates/openshell-otel/Cargo.toml @@ -11,31 +11,31 @@ license.workspace = true repository.workspace = true [dependencies] -async-trait = "0.1" -bytes = { workspace = true } -futures = { workspace = true } +async-trait = { version = "0.1", optional = true } +bytes = { workspace = true, optional = true } +futures = { workspace = true, optional = true } http = { workspace = true } -http-body-util = { workspace = true } -hyper-rustls = { version = "0.27", default-features = false, features = ["http1", "logging", "tls12"] } -hyper-util = { workspace = true, features = ["client-legacy", "http1", "tokio"] } -metrics = { workspace = true } -openshell-crypto = { workspace = true } -openshell-ocsf = { path = "../openshell-ocsf" } +http-body-util = { workspace = true, optional = true } +hyper-rustls = { version = "0.27", default-features = false, optional = true, features = ["http1", "logging", "tls12"] } +hyper-util = { workspace = true, optional = true, features = ["client-legacy", "http1", "tokio"] } +metrics = { workspace = true, optional = true } +openshell-crypto = { workspace = true, optional = true } +openshell-ocsf = { path = "../openshell-ocsf", optional = true } opentelemetry = { workspace = true } -opentelemetry_sdk = { workspace = true, features = ["logs"] } -opentelemetry-http = { workspace = true } -opentelemetry-otlp = { workspace = true, features = ["gzip-http", "http-proto", "logs"] } -rustls = { workspace = true } -rustls-native-certs = { workspace = true } -serde_json = { workspace = true } -thiserror = { workspace = true } -tokio = { workspace = true } +opentelemetry_sdk = { workspace = true } +opentelemetry-http = { workspace = true, optional = true } +opentelemetry-otlp = { workspace = true, optional = true, features = ["gzip-http", "http-proto", "logs"] } +rustls = { workspace = true, optional = true } +rustls-native-certs = { 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 } -zeroize = { workspace = true } +tower-http = { workspace = true, optional = true } +zeroize = { workspace = true, optional = true } [dev-dependencies] openshell-otel-test-support = { path = "../openshell-otel-test-support" } @@ -44,6 +44,37 @@ opentelemetry_sdk = { workspace = true, features = ["testing"] } prost = { workspace = true } tempfile = "3" 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:async-trait", + "dep:bytes", + "dep:futures", + "dep:http-body-util", + "dep:hyper-rustls", + "dep:hyper-util", + "dep:metrics", + "dep:openshell-crypto", + "dep:openshell-ocsf", + "dep:opentelemetry-http", + "dep:opentelemetry-otlp", + "dep:rustls", + "dep:rustls-native-certs", + "dep:serde_json", + "dep:thiserror", + "dep:tokio", + "dep:tower-http", + "dep:tracing-subscriber", + "dep:zeroize", + "opentelemetry_sdk/logs", +] [lints] workspace = true diff --git a/crates/openshell-otel/src/lib.rs b/crates/openshell-otel/src/lib.rs index 1f9de6bf17..32077a7f93 100644 --- a/crates/openshell-otel/src/lib.rs +++ b/crates/openshell-otel/src/lib.rs @@ -2,573 +2,79 @@ // 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 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 correlation; -pub use correlation::current_ocsf_trace_correlation; +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 correlation; +#[cfg(feature = "export")] mod driver; +#[cfg(feature = "export")] mod grpc; +#[cfg(feature = "export")] mod headers; +#[cfg(feature = "export")] mod logs; +#[cfg(feature = "export")] mod ocsf; -mod propagation; +#[cfg(feature = "export")] +mod setup; +#[cfg(feature = "export")] mod tls; +#[cfg(feature = "export")] mod transport; #[cfg(test)] #[path = "../../../tests/support/environment.rs"] mod test_environment; +#[cfg(feature = "export")] +pub use correlation::current_ocsf_trace_correlation; + +#[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 headers::{encode_otlp_headers, merge_otlp_headers, parse_otlp_headers}; +#[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, -}; +#[cfg(feature = "export")] pub use tls::{MAX_TLS_FILE_BYTES, OtlpTls, TlsError, TlsFileError}; - +#[cfg(feature = "export")] pub use transport::{ CONNECT_TIMEOUT, OtlpHttpClient, TCP_KEEPALIVE_IDLE, TCP_KEEPALIVE_INTERVAL, TCP_KEEPALIVE_RETRIES, signal_url, }; -use opentelemetry::KeyValue; -use opentelemetry::trace::TracerProvider as _; -pub use opentelemetry_otlp::LogExporter; -use opentelemetry_otlp::tonic_types::metadata::MetadataMap; -use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithHttpConfig, 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; - -/// Export supervisor spans at INFO by default, or DEBUG and TRACE when enabled. -/// Dependency spans remain at INFO. -#[must_use] -pub fn supervisor_span_filter(log_level: &str) -> tracing_subscriber::EnvFilter { - use tracing_subscriber::filter::LevelFilter; - let level = log_level - .parse::() - .unwrap_or(LevelFilter::INFO) - .max(LevelFilter::INFO); - tracing_subscriber::EnvFilter::new(format!("info,openshell={level}")) -} - -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), -} - -/// OTLP transport to a collector. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub enum Protocol { - #[default] - Grpc, - HttpProtobuf, -} - -/// Inputs for OTLP 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, -} - -/// How exports reach a collector: the transport, request headers, and TLS. -#[derive(Clone, Default, PartialEq, Eq)] -pub struct CollectorTransport { - pub protocol: Protocol, - pub headers: http::HeaderMap, - pub tls: OtlpTls, -} - -impl std::fmt::Debug for CollectorTransport { - fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter - .debug_struct("CollectorTransport") - .field("protocol", &self.protocol) - .field("header_names", &self.headers.keys().collect::>()) - .field("tls", &self.tls) - .finish() - } -} - -/// 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), - - #[error("invalid OTLP endpoint {endpoint:?}: {source}")] - InvalidSignalUrl { - endpoint: String, - source: http::Error, - }, - - #[error("OTLP TLS settings need an https:// endpoint, not {endpoint:?}")] - TlsWithoutHttps { endpoint: String }, - - #[error("invalid OTLP TLS settings: {0}")] - Tls(#[from] TlsError), - - #[error("invalid OTLP TLS files: {0}")] - TlsFiles(#[from] TlsFileError), - - #[error("OTLP/HTTP export needs a tokio runtime")] - NoRuntime, -} - -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. -/// -/// Every exporter that reads an operator-supplied endpoint uses this, so a -/// value accepted for one signal is accepted for all of them. -pub 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 collector endpoint and the resource that every signal -/// exported to it describes. -#[derive(Clone)] -pub struct OtlpTarget { - endpoint: String, - tls: bool, - protocol: Protocol, - headers: http::HeaderMap, - tls_settings: OtlpTls, - resource: Resource, -} - -impl std::fmt::Debug for OtlpTarget { - fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter - .debug_struct("OtlpTarget") - .field("endpoint", &self.endpoint) - .field("tls", &self.tls) - .field("protocol", &self.protocol) - .field("header_names", &self.headers.keys().collect::>()) - .field("tls_settings", &self.tls_settings) - .field("resource", &self.resource) - .finish() - } -} - -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), - protocol: Protocol::Grpc, - headers: http::HeaderMap::new(), - tls_settings: OtlpTls::default(), - resource: resource_for(config), - }) - } - - /// Export over `protocol` instead of gRPC. - #[must_use] - pub fn with_protocol(mut self, protocol: Protocol) -> Self { - self.protocol = protocol; - self - } - - /// Send `headers` with every export request. Headers from the process - /// environment (`OTEL_EXPORTER_OTLP_*HEADERS`) still apply and replace - /// any of the same name. - #[must_use] - pub fn with_headers(mut self, headers: http::HeaderMap) -> Self { - self.headers = headers; - self - } - - /// Reach the collector the way `transport` describes. - #[must_use] - pub fn with_transport(self, transport: &CollectorTransport) -> Self { - self.with_protocol(transport.protocol) - .with_headers(transport.headers.clone()) - .with_tls(transport.tls.clone()) - } - - /// Verify an `https://` collector and authenticate to it with `tls`. - #[must_use] - pub fn with_tls(mut self, tls: OtlpTls) -> Self { - self.tls_settings = tls; - self - } - - /// The URL of `signal_path` under the endpoint, for OTLP/HTTP export. - pub(crate) fn http_signal_url(&self, signal_path: &str) -> Result { - signal_url(&self.endpoint, signal_path) - .map(|url| url.to_string()) - .map_err(|source| SetupError::InvalidSignalUrl { - endpoint: self.endpoint.clone(), - source, - }) - } - - /// 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()) - .with_metadata(MetadataMap::from_headers(self.headers.clone())); - if self.tls { - builder.with_tls_config(self.tls_settings.grpc_config()) - } else { - builder - } - } - - /// Point an OTLP/HTTP exporter builder at `signal_path` under this - /// target. - pub(crate) fn configure_http( - &self, - builder: B, - signal_path: &str, - timeout_var: &str, - ) -> Result - where - B: WithExportConfig + WithHttpConfig, - { - // The SDK percent-decodes these values again. - let headers = self - .headers - .iter() - .filter_map(|(name, value)| { - Some((name.to_string(), value.to_str().ok()?.replace('%', "%25"))) - }) - .collect(); - Ok(builder - .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) - .with_endpoint(self.http_signal_url(signal_path)?) - .with_http_client(OtlpHttpClient::new( - self.tls.then_some(&self.tls_settings), - transport::export_timeout(timeout_var), - )?) - .with_headers(headers)) - } - - #[must_use] - pub fn endpoint(&self) -> &str { - &self.endpoint - } - - #[must_use] - pub fn resource(&self) -> &Resource { - &self.resource - } - - /// Fails when TLS settings are given for a plaintext endpoint. - pub(crate) fn check_tls(&self) -> Result<(), SetupError> { - if self.tls || self.tls_settings.is_default() { - Ok(()) - } else { - Err(SetupError::TlsWithoutHttps { - endpoint: self.endpoint.clone(), - }) - } - } - - /// Build a trace provider that exports spans to this target in batches. - pub fn tracer_provider(&self) -> Result { - self.check_tls()?; - let exporter = match self.protocol { - Protocol::Grpc => self - .configure(SpanExporter::builder().with_tonic()) - .build()?, - Protocol::HttpProtobuf => self - .configure_http( - SpanExporter::builder().with_http(), - "v1/traces", - "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT", - )? - .build()?, - }; - Ok(SdkTracerProvider::builder() - .with_batch_exporter(exporter) - .with_resource(self.resource.clone()) - .build()) - } -} - -/// Build an OTLP 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, ()> { @@ -576,265 +82,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 supervisor_span_filter_raises_only_openshell_targets() { - let directives = |log_level: &str| supervisor_span_filter(log_level).to_string(); - assert_eq!(directives("warn"), "openshell=info,info"); - assert_eq!(directives("info"), "openshell=info,info"); - assert_eq!(directives("debug"), "openshell=debug,info"); - assert_eq!(directives("info,h2=debug"), "openshell=info,info"); - } - - #[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 tls_settings_need_an_https_endpoint() { - let pki = openshell_otel_test_support::TestPki::generate(); - let tls = OtlpTls::from_pem(Some(pki.ca_pem.as_bytes()), None).unwrap(); - for protocol in [Protocol::Grpc, Protocol::HttpProtobuf] { - let target = |endpoint| { - OtlpTarget::new(&OtlpExportConfig { - endpoint, - service_name: ServiceName::Fixed("test-service"), - service_version: None, - resource_attributes: Vec::new(), - }) - .unwrap() - .with_protocol(protocol) - .with_tls(tls.clone()) - }; - assert!(matches!( - target("http://collector:4317").tracer_provider(), - Err(SetupError::TlsWithoutHttps { .. }) - )); - assert!(matches!( - target("http://collector:4317").log_exporter(), - Err(SetupError::TlsWithoutHttps { .. }) - )); - assert!(target("https://collector:4317").tracer_provider().is_ok()); - } - } - - #[test] - fn debug_output_names_headers_without_their_values() { - let target = OtlpTarget::new(&OtlpExportConfig { - endpoint: "http://collector:4317", - service_name: ServiceName::Fixed("test-service"), - service_version: None, - resource_attributes: Vec::new(), - }) - .unwrap() - .with_headers(parse_otlp_headers("authorization=Bearer%20secret-token")); - - let debug = format!("{target:?}"); - assert!(debug.contains("authorization"), "{debug}"); - assert!(!debug.contains("secret-token"), "{debug}"); - } - - #[tokio::test(flavor = "multi_thread")] - async fn trace_provider_sends_headers_as_grpc_metadata() { - test_environment::Environment::new() - .remove("OTEL_EXPORTER_OTLP_HEADERS") - .remove("OTEL_EXPORTER_OTLP_TRACES_HEADERS") - .remove("OTEL_EXPORTER_OTLP_LOGS_HEADERS") - .run_async(async { - use opentelemetry::trace::Tracer as _; - - let collector = openshell_otel_test_support::OtlpTestServer::start().await; - let provider = OtlpTarget::new(&OtlpExportConfig { - endpoint: collector.endpoint(), - service_name: ServiceName::Fixed("test-service"), - service_version: None, - resource_attributes: Vec::new(), - }) - .unwrap() - .with_headers(parse_otlp_headers("authorization=Bearer%20token")) - .tracer_provider() - .unwrap(); - provider.tracer("test").in_span("grpc-span", |_| {}); - let flush = tokio::task::spawn_blocking(move || provider.force_flush()); - - collector.wait_for_export().await; - flush.await.unwrap().expect("flush succeeds"); - assert_eq!(collector.received().trace_authorizations, ["Bearer token"]); - }) - .await; - } - - #[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..1bf8407e59 --- /dev/null +++ b/crates/openshell-otel/src/setup.rs @@ -0,0 +1,798 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OTLP export setup: resources, providers, and tracing layers. +//! +//! Compiled only with the `export` feature; see the crate-level documentation. + +use crate::tls::{OtlpTls, TlsError, TlsFileError}; +use crate::transport::{self, OtlpHttpClient, signal_url}; +use opentelemetry::KeyValue; +use opentelemetry::trace::TracerProvider as _; +pub use opentelemetry_otlp::LogExporter; +use opentelemetry_otlp::tonic_types::metadata::MetadataMap; +use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithHttpConfig, 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; + +/// Export supervisor spans at INFO by default, or DEBUG and TRACE when enabled. +/// Dependency spans remain at INFO. +#[must_use] +pub fn supervisor_span_filter(log_level: &str) -> tracing_subscriber::EnvFilter { + use tracing_subscriber::filter::LevelFilter; + let level = log_level + .parse::() + .unwrap_or(LevelFilter::INFO) + .max(LevelFilter::INFO); + tracing_subscriber::EnvFilter::new(format!("info,openshell={level}")) +} + +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), +} + +/// OTLP transport to a collector. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum Protocol { + #[default] + Grpc, + HttpProtobuf, +} + +/// Inputs for OTLP 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, +} + +/// How exports reach a collector: the transport, request headers, and TLS. +#[derive(Clone, Default, PartialEq, Eq)] +pub struct CollectorTransport { + pub protocol: Protocol, + pub headers: http::HeaderMap, + pub tls: OtlpTls, +} + +impl std::fmt::Debug for CollectorTransport { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("CollectorTransport") + .field("protocol", &self.protocol) + .field("header_names", &self.headers.keys().collect::>()) + .field("tls", &self.tls) + .finish() + } +} + +/// 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), + + #[error("invalid OTLP endpoint {endpoint:?}: {source}")] + InvalidSignalUrl { + endpoint: String, + source: http::Error, + }, + + #[error("OTLP TLS settings need an https:// endpoint, not {endpoint:?}")] + TlsWithoutHttps { endpoint: String }, + + #[error("invalid OTLP TLS settings: {0}")] + Tls(#[from] TlsError), + + #[error("invalid OTLP TLS files: {0}")] + TlsFiles(#[from] TlsFileError), + + #[error("OTLP/HTTP export needs a tokio runtime")] + NoRuntime, +} + +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. +/// +/// Every exporter that reads an operator-supplied endpoint uses this, so a +/// value accepted for one signal is accepted for all of them. +pub 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 collector endpoint and the resource that every signal +/// exported to it describes. +#[derive(Clone)] +pub struct OtlpTarget { + pub(crate) endpoint: String, + pub(crate) tls: bool, + pub(crate) protocol: Protocol, + pub(crate) headers: http::HeaderMap, + pub(crate) tls_settings: OtlpTls, + pub(crate) resource: Resource, +} + +impl std::fmt::Debug for OtlpTarget { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("OtlpTarget") + .field("endpoint", &self.endpoint) + .field("tls", &self.tls) + .field("protocol", &self.protocol) + .field("header_names", &self.headers.keys().collect::>()) + .field("tls_settings", &self.tls_settings) + .field("resource", &self.resource) + .finish() + } +} + +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), + protocol: Protocol::Grpc, + headers: http::HeaderMap::new(), + tls_settings: OtlpTls::default(), + resource: resource_for(config), + }) + } + + /// Export over `protocol` instead of gRPC. + #[must_use] + pub fn with_protocol(mut self, protocol: Protocol) -> Self { + self.protocol = protocol; + self + } + + /// Send `headers` with every export request. Headers from the process + /// environment (`OTEL_EXPORTER_OTLP_*HEADERS`) still apply and replace + /// any of the same name. + #[must_use] + pub fn with_headers(mut self, headers: http::HeaderMap) -> Self { + self.headers = headers; + self + } + + /// Reach the collector the way `transport` describes. + #[must_use] + pub fn with_transport(self, transport: &CollectorTransport) -> Self { + self.with_protocol(transport.protocol) + .with_headers(transport.headers.clone()) + .with_tls(transport.tls.clone()) + } + + /// Verify an `https://` collector and authenticate to it with `tls`. + #[must_use] + pub fn with_tls(mut self, tls: OtlpTls) -> Self { + self.tls_settings = tls; + self + } + + /// The URL of `signal_path` under the endpoint, for OTLP/HTTP export. + pub(crate) fn http_signal_url(&self, signal_path: &str) -> Result { + signal_url(&self.endpoint, signal_path) + .map(|url| url.to_string()) + .map_err(|source| SetupError::InvalidSignalUrl { + endpoint: self.endpoint.clone(), + source, + }) + } + + /// 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()) + .with_metadata(MetadataMap::from_headers(self.headers.clone())); + if self.tls { + builder.with_tls_config(self.tls_settings.grpc_config()) + } else { + builder + } + } + + /// Point an OTLP/HTTP exporter builder at `signal_path` under this + /// target. + pub(crate) fn configure_http( + &self, + builder: B, + signal_path: &str, + timeout_var: &str, + ) -> Result + where + B: WithExportConfig + WithHttpConfig, + { + // The SDK percent-decodes these values again. + let headers = self + .headers + .iter() + .filter_map(|(name, value)| { + Some((name.to_string(), value.to_str().ok()?.replace('%', "%25"))) + }) + .collect(); + Ok(builder + .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) + .with_endpoint(self.http_signal_url(signal_path)?) + .with_http_client(OtlpHttpClient::new( + self.tls.then_some(&self.tls_settings), + transport::export_timeout(timeout_var), + )?) + .with_headers(headers)) + } + + #[must_use] + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + #[must_use] + pub fn resource(&self) -> &Resource { + &self.resource + } + + /// Fails when TLS settings are given for a plaintext endpoint. + pub(crate) fn check_tls(&self) -> Result<(), SetupError> { + if self.tls || self.tls_settings.is_default() { + Ok(()) + } else { + Err(SetupError::TlsWithoutHttps { + endpoint: self.endpoint.clone(), + }) + } + } + + /// Build a trace provider that exports spans to this target in batches. + pub fn tracer_provider(&self) -> Result { + self.check_tls()?; + let exporter = match self.protocol { + Protocol::Grpc => self + .configure(SpanExporter::builder().with_tonic()) + .build()?, + Protocol::HttpProtobuf => self + .configure_http( + SpanExporter::builder().with_http(), + "v1/traces", + "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT", + )? + .build()?, + }; + Ok(SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_resource(self.resource.clone()) + .build()) + } +} + +/// Build an OTLP 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::*; + use crate::parse_otlp_headers; + + #[test] + fn supervisor_span_filter_raises_only_openshell_targets() { + let directives = |log_level: &str| supervisor_span_filter(log_level).to_string(); + assert_eq!(directives("warn"), "openshell=info,info"); + assert_eq!(directives("info"), "openshell=info,info"); + assert_eq!(directives("debug"), "openshell=debug,info"); + assert_eq!(directives("info,h2=debug"), "openshell=info,info"); + } + + #[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 tls_settings_need_an_https_endpoint() { + let pki = openshell_otel_test_support::TestPki::generate(); + let tls = OtlpTls::from_pem(Some(pki.ca_pem.as_bytes()), None).unwrap(); + for protocol in [Protocol::Grpc, Protocol::HttpProtobuf] { + let target = |endpoint| { + OtlpTarget::new(&OtlpExportConfig { + endpoint, + service_name: ServiceName::Fixed("test-service"), + service_version: None, + resource_attributes: Vec::new(), + }) + .unwrap() + .with_protocol(protocol) + .with_tls(tls.clone()) + }; + assert!(matches!( + target("http://collector:4317").tracer_provider(), + Err(SetupError::TlsWithoutHttps { .. }) + )); + assert!(matches!( + target("http://collector:4317").log_exporter(), + Err(SetupError::TlsWithoutHttps { .. }) + )); + assert!(target("https://collector:4317").tracer_provider().is_ok()); + } + } + + #[test] + fn debug_output_names_headers_without_their_values() { + let target = OtlpTarget::new(&OtlpExportConfig { + endpoint: "http://collector:4317", + service_name: ServiceName::Fixed("test-service"), + service_version: None, + resource_attributes: Vec::new(), + }) + .unwrap() + .with_headers(parse_otlp_headers("authorization=Bearer%20secret-token")); + + let debug = format!("{target:?}"); + assert!(debug.contains("authorization"), "{debug}"); + assert!(!debug.contains("secret-token"), "{debug}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn trace_provider_sends_headers_as_grpc_metadata() { + crate::test_environment::Environment::new() + .remove("OTEL_EXPORTER_OTLP_HEADERS") + .remove("OTEL_EXPORTER_OTLP_TRACES_HEADERS") + .remove("OTEL_EXPORTER_OTLP_LOGS_HEADERS") + .run_async(async { + use opentelemetry::trace::Tracer as _; + + let collector = openshell_otel_test_support::OtlpTestServer::start().await; + let provider = OtlpTarget::new(&OtlpExportConfig { + endpoint: collector.endpoint(), + service_name: ServiceName::Fixed("test-service"), + service_version: None, + resource_attributes: Vec::new(), + }) + .unwrap() + .with_headers(parse_otlp_headers("authorization=Bearer%20token")) + .tracer_provider() + .unwrap(); + provider.tracer("test").in_span("grpc-span", |_| {}); + let flush = tokio::task::spawn_blocking(move || provider.force_flush()); + + collector.wait_for_export().await; + flush.await.unwrap().expect("flush succeeds"); + assert_eq!(collector.received().trace_authorizations, ["Bearer token"]); + }) + .await; + } + + #[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 1f03916ef2..8241ac49b5 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -89,6 +89,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 @@ -1770,6 +1780,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 0cf8ca3c27..3cbdba70a3 100644 --- a/docs/observability/traces.mdx +++ b/docs/observability/traces.mdx @@ -71,6 +71,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