diff --git a/Cargo.lock b/Cargo.lock index 1964b5e421..edb2e8f62e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4874,6 +4874,7 @@ dependencies = [ "opentelemetry_sdk", "prost", "serde_json", + "tempfile", "thiserror 2.0.20", "tokio", "tonic", diff --git a/crates/openshell-core/src/jwt.rs b/crates/openshell-core/src/jwt.rs index f79490f832..b4e8b4b75c 100644 --- a/crates/openshell-core/src/jwt.rs +++ b/crates/openshell-core/src/jwt.rs @@ -406,10 +406,18 @@ mod session { /// Never delivered to the workload. Missing only in older bundles. #[serde(default, skip_serializing_if = "Option::is_none")] pub ssh_host_private_key: Option, + /// Headers for the supervisor's trace collector, present only when + /// the gateway has them. Never delivered to the workload. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub traces_otlp_headers: Option, /// Headers for the OCSF OTLP collector, present only when the gateway /// has them. Never delivered to the workload. #[serde(default, skip_serializing_if = "Option::is_none")] pub ocsf_otlp_headers: Option, + /// Headers for the agent trace relay's collector, present only when + /// the gateway configures them. Never delivered to the workload. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_otlp_headers: Option, } /// Gateway-created authentication input trusted by a compute driver. @@ -496,7 +504,9 @@ mod session { .field("sandbox_token", &"[REDACTED]") .field("sandbox_expires_at", &self.sandbox_expires_at) .field("ssh_host_private_key", &self.ssh_host_private_key) + .field("traces_otlp_headers", &self.traces_otlp_headers) .field("ocsf_otlp_headers", &self.ocsf_otlp_headers) + .field("agent_otlp_headers", &self.agent_otlp_headers) .finish() } } @@ -1625,7 +1635,9 @@ mod tests { sandbox_token: pair.sandbox.token, sandbox_expires_at: pair.sandbox.expires_at, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }; bundle.validate().expect("non-expiring auth bundle"); } @@ -1764,9 +1776,15 @@ mod tests { ssh_host_private_key: Some(SecretSshHostKey::new( "private-ssh-host-key".to_string(), )), + traces_otlp_headers: Some(SecretOtlpHeaders::new( + "authorization=traces-token".to_string(), + )), ocsf_otlp_headers: Some(SecretOtlpHeaders::new( "authorization=collector-token".to_string(), )), + agent_otlp_headers: Some(SecretOtlpHeaders::new( + "authorization=agent-token".to_string(), + )), }; let encoded = serde_json::to_vec(&bundle).expect("serialize auth bundle"); @@ -1787,22 +1805,36 @@ mod tests { assert!(!debug.contains(bundle.sandbox_token.expose_secret())); assert!(!debug.contains("private-ssh-host-key")); assert!(!debug.contains("collector-token")); - assert_eq!(debug.matches("[REDACTED]").count(), 4); + assert!(!debug.contains("agent-token")); + assert!(!debug.contains("traces-token")); + assert_eq!(debug.matches("[REDACTED]").count(), 6); assert_eq!( decoded.ssh_host_private_key.unwrap().expose_secret(), "private-ssh-host-key" ); + assert_eq!( + decoded.traces_otlp_headers.unwrap().expose_secret(), + "authorization=traces-token" + ); assert_eq!( decoded.ocsf_otlp_headers.unwrap().expose_secret(), "authorization=collector-token" ); + assert_eq!( + decoded.agent_otlp_headers.unwrap().expose_secret(), + "authorization=agent-token" + ); let mut legacy = serde_json::to_value(&bundle).unwrap(); let fields = legacy.as_object_mut().unwrap(); fields.remove("ssh_host_private_key"); + fields.remove("traces_otlp_headers"); fields.remove("ocsf_otlp_headers"); + fields.remove("agent_otlp_headers"); let legacy: SupervisorAuthBundle = serde_json::from_value(legacy).unwrap(); assert!(legacy.ssh_host_private_key.is_none()); + assert!(legacy.traces_otlp_headers.is_none()); assert!(legacy.ocsf_otlp_headers.is_none()); + assert!(legacy.agent_otlp_headers.is_none()); } #[derive(Serialize)] diff --git a/crates/openshell-driver-docker/src/tests.rs b/crates/openshell-driver-docker/src/tests.rs index 125cb1d56a..174339acc0 100644 --- a/crates/openshell-driver-docker/src/tests.rs +++ b/crates/openshell-driver-docker/src/tests.rs @@ -64,7 +64,9 @@ fn test_launch_authentication() -> Vec { sandbox_token: SecretJwt::parse("sandbox.token.value").unwrap(), sandbox_expires_at: i64::MAX, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: "gateway-test".to_string(), verification_keys: vec![SessionVerificationKey { diff --git a/crates/openshell-driver-podman/src/driver.rs b/crates/openshell-driver-podman/src/driver.rs index 867c09a38c..cc9c80747e 100644 --- a/crates/openshell-driver-podman/src/driver.rs +++ b/crates/openshell-driver-podman/src/driver.rs @@ -2663,7 +2663,9 @@ mod tests { sandbox_token: SecretJwt::parse("sandbox.token.value").unwrap(), sandbox_expires_at: i64::MAX, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: "gateway-test".to_string(), verification_keys: vec![SessionVerificationKey { diff --git a/crates/openshell-driver-podman/src/isolation.rs b/crates/openshell-driver-podman/src/isolation.rs index 6da06a5773..7349b9083f 100644 --- a/crates/openshell-driver-podman/src/isolation.rs +++ b/crates/openshell-driver-podman/src/isolation.rs @@ -471,7 +471,9 @@ mod tests { sandbox_token: SecretJwt::parse("sandbox.token.value").unwrap(), sandbox_expires_at: i64::MAX, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: "gateway-test".to_string(), verification_keys: vec![SessionVerificationKey { diff --git a/crates/openshell-driver-vm/src/driver.rs b/crates/openshell-driver-vm/src/driver.rs index 0a0cdf30fd..d40e8e472d 100644 --- a/crates/openshell-driver-vm/src/driver.rs +++ b/crates/openshell-driver-vm/src/driver.rs @@ -11242,7 +11242,9 @@ mod tests { sandbox_token: SecretJwt::parse(format!("sandbox-{label}")).expect("sandbox token"), sandbox_expires_at: 1, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: "gateway-a".to_string(), verification_keys: vec![SessionVerificationKey { diff --git a/crates/openshell-gateway/src/lib.rs b/crates/openshell-gateway/src/lib.rs index 26758f85bc..47caea8fce 100644 --- a/crates/openshell-gateway/src/lib.rs +++ b/crates/openshell-gateway/src/lib.rs @@ -749,3 +749,7 @@ mod tests { assert!(::supports_network_supervisor_trust(&VmFactory)); } } + +#[cfg(test)] +#[path = "../../../tests/support/environment.rs"] +mod test_environment; diff --git a/crates/openshell-gateway/src/vm.rs b/crates/openshell-gateway/src/vm.rs index fcfc8ffe83..b861bbe840 100644 --- a/crates/openshell-gateway/src/vm.rs +++ b/crates/openshell-gateway/src/vm.rs @@ -729,6 +729,11 @@ fn append_otlp_args(command: &mut Command, otlp_config: Option<&OtlpConfig>, gat sandbox_env::OTLP_AGENT_ENDPOINT_FLAG, sandbox_env::OTLP_AGENT_PROTOCOL_FLAG, ); + // The driver's own trace exporter reads this like any OTLP SDK; keeping + // it out of the arguments keeps the headers out of process listings. + if let Some(headers) = openshell_server::gateway_trace_headers(otlp_config) { + command.env("OTEL_EXPORTER_OTLP_TRACES_HEADERS", headers.expose_secret()); + } // The driver names its own spans with the gateway name, and supervisors // name their spans, OCSF records, and relayed traces with it. if !otlp.is_empty() { @@ -833,7 +838,8 @@ mod tests { use openshell_core::sandbox_env::OtlpProtocol; use openshell_core::{NetworkSupervisorTrustBundle, UpstreamProxyConfig}; use openshell_server::config_file::{ - AgentOtlpConfig, OcsfOtlpConfig, OtlpCollectorConfig, OtlpConfig, OtlpTracesConfig, + AgentOtlpConfig, OcsfOtlpConfig, OtlpCollectorConfig, OtlpConfig, OtlpHeaders, + OtlpTracesConfig, }; use std::os::unix::fs::PermissionsExt; use std::os::unix::net::UnixListener as StdUnixListener; @@ -876,6 +882,36 @@ mod tests { .expect("an explicit bootstrap image is valid"); } + #[test] + fn vm_driver_receives_trace_headers_in_its_environment_only() { + crate::test_environment::Environment::new() + .remove("OTEL_EXPORTER_OTLP_HEADERS") + .remove("OTEL_EXPORTER_OTLP_TRACES_HEADERS") + .run(|| { + let otlp: OtlpConfig = toml::from_str( + "endpoint = 'http://collector.internal:4318'\n\ + headers = { authorization = 'Bearer secret-token' }\n", + ) + .unwrap(); + let mut command = tokio::process::Command::new("openshell-driver-vm"); + append_otlp_args(&mut command, Some(&otlp), "production-us-west"); + + let env = command + .as_std() + .get_envs() + .find(|(name, _)| *name == "OTEL_EXPORTER_OTLP_TRACES_HEADERS") + .and_then(|(_, value)| value) + .map(|value| value.to_string_lossy().into_owned()); + assert_eq!(env.as_deref(), Some("authorization=Bearer%20secret-token")); + assert!( + command + .as_std() + .get_args() + .all(|arg| !arg.to_string_lossy().contains("secret-token")) + ); + }); + } + #[test] fn vm_driver_command_includes_gateway_otlp_configuration() { let mut command = tokio::process::Command::new("openshell-driver-vm"); @@ -887,6 +923,8 @@ mod tests { collector: OtlpCollectorConfig { endpoint: "http://collector.internal:4317".to_string(), protocol: OtlpProtocol::Grpc, + headers: OtlpHeaders::default(), + headers_file: None, }, }), ocsf: None, @@ -919,6 +957,8 @@ mod tests { collector: OtlpCollectorConfig { endpoint: "http://collector.internal:4318".to_string(), protocol: OtlpProtocol::HttpProtobuf, + headers: OtlpHeaders::default(), + headers_file: None, }, }), ocsf: None, @@ -954,6 +994,8 @@ mod tests { collector: OtlpCollectorConfig { endpoint: "http://siem.internal:4318".to_string(), protocol: OtlpProtocol::HttpProtobuf, + headers: OtlpHeaders::default(), + headers_file: None, }, schema_version: Some("2.0".into()), }), @@ -993,6 +1035,8 @@ mod tests { collector: OtlpCollectorConfig { endpoint: "http://agents.internal:4318".to_string(), protocol: OtlpProtocol::HttpProtobuf, + headers: OtlpHeaders::default(), + headers_file: None, }, }), }), diff --git a/crates/openshell-otel-test-support/src/lib.rs b/crates/openshell-otel-test-support/src/lib.rs index ea34843649..f97fcd20e5 100644 --- a/crates/openshell-otel-test-support/src/lib.rs +++ b/crates/openshell-otel-test-support/src/lib.rs @@ -109,6 +109,7 @@ pub async fn assert_compute_driver_tracing( let (provider, error) = descriptor.provider_for( Some(collector.endpoint()), openshell_otel::Protocol::Grpc, + &::http::HeaderMap::new(), service_version, Some("test-gateway"), Some(descriptor.compute_driver()), @@ -142,6 +143,8 @@ pub struct ReceivedTraces { pub spans: Vec, /// String bodies of exported log records. pub log_bodies: Vec, + /// `authorization` metadata of each trace export request. + pub trace_authorizations: Vec, /// `authorization` metadata of each log export request. pub log_authorizations: Vec, pub service_names: Vec, @@ -216,6 +219,15 @@ impl TraceService for Collector { { received.encodings.push(encoding.to_string()); } + if let Some(authorization) = request + .metadata() + .get("authorization") + .and_then(|value| value.to_str().ok()) + { + received + .trace_authorizations + .push(authorization.to_string()); + } for resource_span in request.into_inner().resource_spans { let attributes = received.record_resource(resource_span.resource); for scope_span in resource_span.scope_spans { diff --git a/crates/openshell-otel/Cargo.toml b/crates/openshell-otel/Cargo.toml index b174f65871..e8e1333d01 100644 --- a/crates/openshell-otel/Cargo.toml +++ b/crates/openshell-otel/Cargo.toml @@ -37,6 +37,7 @@ openshell-otel-test-support = { path = "../openshell-otel-test-support" } opentelemetry-proto = { workspace = true, features = ["logs"] } opentelemetry_sdk = { workspace = true, features = ["testing"] } prost = { workspace = true } +tempfile = "3" tokio = { workspace = true, features = ["test-util"] } [lints] diff --git a/crates/openshell-otel/src/driver.rs b/crates/openshell-otel/src/driver.rs index b5e9cca555..e99c819152 100644 --- a/crates/openshell-otel/src/driver.rs +++ b/crates/openshell-otel/src/driver.rs @@ -74,6 +74,7 @@ impl ComputeDriverTracing { self, endpoint: Option<&str>, protocol: crate::Protocol, + headers: &http::HeaderMap, service_version: &'static str, gateway_name: Option<&str>, compute_driver: Option<&str>, @@ -89,6 +90,7 @@ impl ComputeDriverTracing { ), })? .with_protocol(protocol) + .with_headers(headers.clone()) .tracer_provider() }); match provider { @@ -225,6 +227,7 @@ pub fn install_driver_tracing( let (provider, setup_error) = descriptor.provider_for( config.endpoint, config.protocol, + &http::HeaderMap::new(), config.service_version, config.gateway_name, Some(descriptor.compute_driver()), @@ -259,25 +262,35 @@ mod tests { } #[tokio::test(flavor = "multi_thread")] - async fn driver_provider_exports_over_the_configured_protocol() { - use opentelemetry::trace::{Tracer as _, TracerProvider as _}; - - const TRACING: ComputeDriverTracing = crate::compute_driver_tracing!(); - let mut collector = openshell_otel_test_support::OtlpHttpTestServer::start().await; - let (provider, error) = TRACING.provider_for( - Some(collector.endpoint()), - crate::Protocol::HttpProtobuf, - "1.2.3", - Some("test-gateway"), - Some("docker"), - ); - assert!(error.is_none(), "unexpected setup error: {error:?}"); - let provider = provider.expect("provider"); - provider.tracer("test").in_span("driver-span", |_| {}); - let flush = tokio::task::spawn_blocking(move || provider.force_flush()); - - assert_eq!(collector.next_export().await.path, "/v1/traces"); - flush.await.unwrap().expect("flush succeeds"); + async fn driver_provider_exports_with_the_configured_protocol_and_headers() { + 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 _, TracerProvider as _}; + + const TRACING: ComputeDriverTracing = crate::compute_driver_tracing!(); + let mut collector = openshell_otel_test_support::OtlpHttpTestServer::start().await; + let (provider, error) = TRACING.provider_for( + Some(collector.endpoint()), + crate::Protocol::HttpProtobuf, + &crate::parse_otlp_headers("authorization=driver"), + "1.2.3", + Some("test-gateway"), + Some("docker"), + ); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + let provider = provider.expect("provider"); + provider.tracer("test").in_span("driver-span", |_| {}); + let flush = tokio::task::spawn_blocking(move || provider.force_flush()); + + let export = collector.next_export().await; + assert_eq!(export.path, "/v1/traces"); + assert_eq!(export.headers["authorization"], "driver"); + flush.await.unwrap().expect("flush succeeds"); + }) + .await; } #[tokio::test] diff --git a/crates/openshell-otel/src/headers.rs b/crates/openshell-otel/src/headers.rs new file mode 100644 index 0000000000..f6dfe2c253 --- /dev/null +++ b/crates/openshell-otel/src/headers.rs @@ -0,0 +1,136 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Collector request headers in the `OTEL_EXPORTER_OTLP_HEADERS` format. + +use std::fmt::Write as _; + +use http::{HeaderMap, HeaderName, HeaderValue}; + +/// Parse headers the way the SDK parses `OTEL_EXPORTER_OTLP_HEADERS`: +/// comma-separated `key=value` pairs with percent-encoded values. Entries +/// that are empty or not valid HTTP headers are skipped. +#[must_use] +pub fn parse_otlp_headers(headers: &str) -> HeaderMap { + let mut map = HeaderMap::new(); + for (key, value) in headers.split_terminator(',').filter_map(|entry| { + let (key, value) = entry.trim().split_once('=')?; + let (key, value) = (key.trim(), value.trim()); + let value = percent_decode(value).unwrap_or_else(|| value.to_string()); + (!key.is_empty() && !value.is_empty()).then_some((key.to_string(), value)) + }) { + if let (Ok(key), Ok(value)) = ( + HeaderName::try_from(key), + HeaderValue::try_from(value.as_str()), + ) { + map.insert(key, value); + } + } + map +} + +/// Encode `headers` for [`parse_otlp_headers`], percent-encoding every value +/// byte outside the URI unreserved set. +#[must_use] +pub fn encode_otlp_headers(headers: &HeaderMap) -> String { + let mut encoded = String::new(); + for (key, value) in headers { + if !encoded.is_empty() { + encoded.push(','); + } + encoded.push_str(key.as_str()); + encoded.push('='); + for &byte in value.as_bytes() { + if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) { + encoded.push(char::from(byte)); + } else { + let _ = write!(encoded, "%{byte:02X}"); + } + } + } + encoded +} + +/// `headers` with each header in `overrides` replacing any of the same name. +#[must_use] +pub fn merge_otlp_headers(mut headers: HeaderMap, overrides: HeaderMap) -> HeaderMap { + headers.extend(overrides); + headers +} + +fn percent_decode(value: &str) -> Option { + let bytes = value.as_bytes(); + let mut decoded = Vec::with_capacity(bytes.len()); + let mut index = 0; + while index < bytes.len() { + if bytes[index] == b'%' { + let hex = std::str::from_utf8(bytes.get(index + 1..index + 3)?).ok()?; + decoded.push(u8::from_str_radix(hex, 16).ok()?); + index += 3; + } else { + decoded.push(bytes[index]); + index += 1; + } + } + String::from_utf8(decoded).ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn map(entries: &[(&'static str, &'static str)]) -> HeaderMap { + entries + .iter() + .map(|(key, value)| { + ( + HeaderName::from_static(key), + HeaderValue::from_static(value), + ) + }) + .collect() + } + + #[test] + fn headers_parse_like_the_sdk_environment_variable() { + let headers = parse_otlp_headers( + "authorization=Bearer%20secret, x-tenant = blue ,=empty-key,no-value=,bad", + ); + assert_eq!(headers["authorization"], "Bearer secret"); + assert_eq!(headers["x-tenant"], "blue"); + assert_eq!(headers.len(), 2); + } + + #[test] + fn malformed_percent_encoding_is_kept_verbatim() { + assert_eq!(parse_otlp_headers("token=abc%zz")["token"], "abc%zz"); + } + + #[test] + fn encoded_headers_parse_back_to_the_same_values() { + let mut headers = map(&[("x-mlflow-experiment-id", "0")]); + headers.insert( + "authorization", + HeaderValue::from_str("Bearer a,b=c%d").unwrap(), + ); + let encoded = encode_otlp_headers(&headers); + assert!(!encoded.contains(' '), "{encoded}"); + assert_eq!(parse_otlp_headers(&encoded), headers); + } + + #[test] + fn overrides_replace_headers_of_the_same_name() { + let merged = merge_otlp_headers( + map(&[("authorization", "toml"), ("x-tenant", "blue")]), + map(&[("authorization", "env"), ("x-extra", "1")]), + ); + assert_eq!( + merged, + map(&[ + ("authorization", "env"), + ("x-tenant", "blue"), + ("x-extra", "1") + ]) + ); + } +} diff --git a/crates/openshell-otel/src/lib.rs b/crates/openshell-otel/src/lib.rs index 47332df7f5..244f5e7511 100644 --- a/crates/openshell-otel/src/lib.rs +++ b/crates/openshell-otel/src/lib.rs @@ -5,11 +5,16 @@ mod driver; mod grpc; +mod headers; mod logs; mod ocsf; mod propagation; mod transport; +#[cfg(test)] +#[path = "../../../tests/support/environment.rs"] +mod test_environment; + pub use driver::{ BoxGrpcStream, ComputeDriverTracing, DriverTracingConfig, DriverTracingHandle, IN_PROCESS_COMPUTE_DRIVER_TARGET, InProcessRpcTracer, install_driver_tracing, @@ -20,6 +25,7 @@ pub use grpc::{ RecordGrpcStatus, TracedGrpcStream, compute_driver_rpc_layer, compute_driver_rpc_operation, grpc_status_code_name, record_grpc_status, rpc, }; +pub use headers::{encode_otlp_headers, merge_otlp_headers, parse_otlp_headers}; pub use ocsf::{OcsfOtlp, OcsfOtlpLayer}; pub use propagation::{ HeaderMapExtractor, MetadataMapInjector, TRACEPARENT_ENV, TRACESTATE_ENV, @@ -34,6 +40,7 @@ pub use transport::{ use opentelemetry::KeyValue; use opentelemetry::trace::TracerProvider as _; pub use opentelemetry_otlp::LogExporter; +use opentelemetry_otlp::tonic_types::metadata::MetadataMap; use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig; use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; use opentelemetry_sdk::Resource; @@ -233,14 +240,28 @@ pub fn validated_endpoint(endpoint: &str) -> Result<(&str, http::Uri), SetupErro /// A validated OTLP collector endpoint and the resource that every signal /// exported to it describes. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct OtlpTarget { endpoint: String, tls: bool, protocol: Protocol, + headers: http::HeaderMap, 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("resource", &self.resource) + .finish() + } +} + impl OtlpTarget { /// Validate the endpoint and build the resource. pub fn new(config: &OtlpExportConfig<'_>) -> Result { @@ -249,6 +270,7 @@ impl OtlpTarget { endpoint: endpoint.to_string(), tls: uri.scheme() == Some(&http::uri::Scheme::HTTPS), protocol: Protocol::Grpc, + headers: http::HeaderMap::new(), resource: resource_for(config), }) } @@ -260,6 +282,15 @@ impl OtlpTarget { 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 + } + /// The URL of `signal_path` under the endpoint, for OTLP/HTTP export. pub(crate) fn http_signal_url(&self, signal_path: &str) -> Result { if self.tls { @@ -278,7 +309,9 @@ impl OtlpTarget { where B: WithExportConfig + WithTonicConfig, { - let builder = builder.with_endpoint(self.endpoint.clone()); + let builder = builder + .with_endpoint(self.endpoint.clone()) + .with_metadata(MetadataMap::from_headers(self.headers.clone())); if self.tls { // Tonic trusts nothing by default, so enable the platform roots. builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots()) @@ -287,6 +320,32 @@ impl OtlpTarget { } } + /// 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(transport::export_timeout(timeout_var))?) + .with_headers(headers)) + } + #[must_use] pub fn endpoint(&self) -> &str { &self.endpoint @@ -303,13 +362,12 @@ impl OtlpTarget { Protocol::Grpc => self .configure(SpanExporter::builder().with_tonic()) .build()?, - Protocol::HttpProtobuf => SpanExporter::builder() - .with_http() - .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) - .with_endpoint(self.http_signal_url("v1/traces")?) - .with_http_client(OtlpHttpClient::new(transport::export_timeout( + Protocol::HttpProtobuf => self + .configure_http( + SpanExporter::builder().with_http(), + "v1/traces", "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT", - ))?) + )? .build()?, }; Ok(SdkTracerProvider::builder() @@ -613,6 +671,52 @@ mod tests { ); } + #[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 { diff --git a/crates/openshell-otel/src/logs.rs b/crates/openshell-otel/src/logs.rs index 09eab68446..e99755531e 100644 --- a/crates/openshell-otel/src/logs.rs +++ b/crates/openshell-otel/src/logs.rs @@ -3,91 +3,29 @@ //! OTLP log export support. -use http::{HeaderMap, HeaderName, HeaderValue}; -use opentelemetry_otlp::tonic_types::metadata::MetadataMap; -use opentelemetry_otlp::{LogExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; +use opentelemetry_otlp::LogExporter; -use crate::{OtlpHttpClient, OtlpTarget, Protocol, SetupError}; +use crate::{OtlpTarget, Protocol, SetupError}; impl OtlpTarget { /// Build a log exporter for this target. Callers attach /// [`Self::resource`] to it. pub fn log_exporter(&self) -> Result { - self.log_exporter_with_headers(None) - } - - /// Build a log exporter that sends `headers`, in the - /// `OTEL_EXPORTER_OTLP_HEADERS` format, with every request. Headers from - /// the process environment still apply and take precedence. - pub fn log_exporter_with_headers( - &self, - headers: Option<&str>, - ) -> Result { - let headers = headers.map(parse_headers).unwrap_or_default(); match self.protocol { Protocol::Grpc => Ok(self .configure(LogExporter::builder().with_tonic()) - .with_metadata(MetadataMap::from_headers(headers)) .build()?), - Protocol::HttpProtobuf => Ok(LogExporter::builder() - .with_http() - .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) - .with_endpoint(self.http_signal_url("v1/logs")?) - .with_http_client(OtlpHttpClient::new(crate::transport::export_timeout( + Protocol::HttpProtobuf => Ok(self + .configure_http( + LogExporter::builder().with_http(), + "v1/logs", "OTEL_EXPORTER_OTLP_LOGS_TIMEOUT", - ))?) - // The SDK percent-decodes these values again. - .with_headers( - headers - .iter() - .filter_map(|(name, value)| { - Some((name.to_string(), value.to_str().ok()?.replace('%', "%25"))) - }) - .collect(), - ) + )? .build()?), } } } -/// Parse headers the way the SDK parses `OTEL_EXPORTER_OTLP_HEADERS`: -/// comma-separated `key=value` pairs with percent-encoded values. Entries -/// that are empty or not valid headers are skipped. -fn parse_headers(headers: &str) -> HeaderMap { - let mut map = HeaderMap::new(); - for (key, value) in headers.split_terminator(',').filter_map(|entry| { - let (key, value) = entry.trim().split_once('=')?; - let (key, value) = (key.trim(), value.trim()); - let value = percent_decode(value).unwrap_or_else(|| value.to_string()); - (!key.is_empty() && !value.is_empty()).then_some((key.to_string(), value)) - }) { - if let (Ok(key), Ok(value)) = ( - HeaderName::try_from(key), - HeaderValue::try_from(value.as_str()), - ) { - map.insert(key, value); - } - } - map -} - -fn percent_decode(value: &str) -> Option { - let bytes = value.as_bytes(); - let mut decoded = Vec::with_capacity(bytes.len()); - let mut index = 0; - while index < bytes.len() { - if bytes[index] == b'%' { - let hex = std::str::from_utf8(bytes.get(index + 1..index + 3)?).ok()?; - decoded.push(u8::from_str_radix(hex, 16).ok()?); - index += 3; - } else { - decoded.push(bytes[index]); - index += 1; - } - } - String::from_utf8(decoded).ok() -} - #[cfg(test)] mod tests { use crate::{OtlpExportConfig, OtlpTarget, ServiceName, SetupError}; @@ -117,25 +55,6 @@ mod tests { )); } - #[test] - fn headers_parse_like_the_sdk_environment_variable() { - let metadata = super::parse_headers( - "authorization=Bearer%20secret, x-tenant = blue ,=empty-key,no-value=,bad", - ); - assert_eq!( - metadata.get("authorization").unwrap().to_str().unwrap(), - "Bearer secret" - ); - assert_eq!(metadata.get("x-tenant").unwrap().to_str().unwrap(), "blue"); - assert_eq!(metadata.len(), 2); - } - - #[test] - fn malformed_percent_encoding_is_kept_verbatim() { - let metadata = super::parse_headers("token=abc%zz"); - assert_eq!(metadata.get("token").unwrap().to_str().unwrap(), "abc%zz"); - } - #[tokio::test] async fn every_signal_shares_the_target_resource() { let target = OtlpTarget::new(&config("http://127.0.0.1:4317")).unwrap(); diff --git a/crates/openshell-otel/src/transport.rs b/crates/openshell-otel/src/transport.rs index c14e6c10f1..bb9fc3680f 100644 --- a/crates/openshell-otel/src/transport.rs +++ b/crates/openshell-otel/src/transport.rs @@ -306,7 +306,7 @@ mod tests { ); assert!( matches!( - http_target(endpoint).log_exporter_with_headers(None), + http_target(endpoint).log_exporter(), Err(SetupError::HttpsUnsupported) ), "{endpoint}" @@ -317,12 +317,16 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn trace_provider_exports_over_http_from_the_batch_thread() { let mut collector = OtlpHttpTestServer::start().await; - let provider = http_target(collector.endpoint()).tracer_provider().unwrap(); + let provider = http_target(collector.endpoint()) + .with_headers(crate::parse_otlp_headers("x-mlflow-experiment-id=7")) + .tracer_provider() + .unwrap(); provider.tracer("test").in_span("http-span", |_| {}); let flush = tokio::task::spawn_blocking(move || provider.force_flush()); let export = collector.next_export().await; assert_eq!(export.method, http::Method::POST); + assert_eq!(export.headers["x-mlflow-experiment-id"], "7"); assert_eq!(export.path, "/v1/traces"); assert_eq!( export.headers[http::header::CONTENT_TYPE], @@ -336,27 +340,37 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn log_exporter_sends_headers_over_http_from_a_blocking_thread() { - let mut collector = OtlpHttpTestServer::start().await; - let exporter = http_target(collector.endpoint()) - .log_exporter_with_headers(Some("authorization=Bearer%20token,x-token=abc%252F")) - .unwrap(); - let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder() - .with_simple_exporter(exporter) - .build(); - let emit = tokio::task::spawn_blocking(move || { - let logger = provider.logger("test"); - let mut record = logger.create_log_record(); - record.set_body(AnyValue::from("ocsf-event")); - logger.emit(record); - }); - - let export = collector.next_export().await; - assert_eq!(export.path, "/v1/logs"); - assert_eq!(export.headers["authorization"], "Bearer token"); - assert_eq!(export.headers["x-token"], "abc%2F", "decoded exactly once"); - let request = ExportLogsServiceRequest::decode(export.body).unwrap(); - let record = &request.resource_logs[0].scope_logs[0].log_records[0]; - assert!(format!("{:?}", record.body).contains("ocsf-event")); - emit.await.unwrap(); + 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 { + let mut collector = OtlpHttpTestServer::start().await; + let exporter = http_target(collector.endpoint()) + .with_headers(crate::parse_otlp_headers( + "authorization=Bearer%20token,x-token=abc%252F", + )) + .log_exporter() + .unwrap(); + let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder() + .with_simple_exporter(exporter) + .build(); + let emit = tokio::task::spawn_blocking(move || { + let logger = provider.logger("test"); + let mut record = logger.create_log_record(); + record.set_body(AnyValue::from("ocsf-event")); + logger.emit(record); + }); + + let export = collector.next_export().await; + assert_eq!(export.path, "/v1/logs"); + assert_eq!(export.headers["authorization"], "Bearer token"); + assert_eq!(export.headers["x-token"], "abc%2F", "decoded exactly once"); + let request = ExportLogsServiceRequest::decode(export.body).unwrap(); + let record = &request.resource_logs[0].scope_logs[0].log_records[0]; + assert!(format!("{:?}", record.body).contains("ocsf-event")); + emit.await.unwrap(); + }) + .await; } } diff --git a/crates/openshell-server/src/auth/sandbox_jwt.rs b/crates/openshell-server/src/auth/sandbox_jwt.rs index 778936ec8f..7ecffef1d2 100644 --- a/crates/openshell-server/src/auth/sandbox_jwt.rs +++ b/crates/openshell-server/src/auth/sandbox_jwt.rs @@ -265,7 +265,9 @@ impl SandboxSessionJwtAuthority { sandbox_token: pair.sandbox.token, sandbox_expires_at: pair.sandbox.expires_at, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: self.gateway_id.clone(), verification_keys: self.verification_keys.clone(), diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2039ef4f7f..39a45f0b6e 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -935,7 +935,7 @@ pub struct ComputeRuntime { Arc>>>, restart_notify: Arc, ssh_identities: Arc>, - supervisor_ocsf_headers: Arc>, + supervisor_otlp_headers: Arc>, /// Desired assignments of the shared supervisor host, when the gateway /// coordinates one for the in-process driver. supervisor_host: Option, @@ -1051,7 +1051,7 @@ impl ComputeRuntime { restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), ssh_identities: Arc::new(OnceLock::new()), - supervisor_ocsf_headers: Arc::new(OnceLock::new()), + supervisor_otlp_headers: Arc::new(OnceLock::new()), supervisor_host: None, service_route_index: Arc::new(OnceLock::new()), }) @@ -1096,13 +1096,11 @@ impl ComputeRuntime { let _ = self.service_route_index.set(index); } - pub(crate) fn configure_supervisor_ocsf_headers( + pub(crate) fn configure_supervisor_otlp_headers( &self, - headers: Option, + headers: crate::otel_tracing::SupervisorOtlpHeaders, ) { - if let Some(headers) = headers { - let _ = self.supervisor_ocsf_headers.set(headers); - } + let _ = self.supervisor_otlp_headers.set(headers); } pub(crate) async fn prepare_ssh_identity( @@ -1131,7 +1129,9 @@ impl ComputeRuntime { let mut with_identity = sandbox.clone(); self.prepare_ssh_identity(&mut with_identity, &mut authentication) .await?; - authentication.supervisor.ocsf_otlp_headers = self.supervisor_ocsf_headers.get().cloned(); + if let Some(headers) = self.supervisor_otlp_headers.get() { + headers.apply(&mut authentication.supervisor); + } serialize_launch_authentication(authentication) } @@ -8621,7 +8621,7 @@ pub fn new_test_runtime_with_driver( restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), ssh_identities: Arc::new(OnceLock::new()), - supervisor_ocsf_headers: Arc::new(OnceLock::new()), + supervisor_otlp_headers: Arc::new(OnceLock::new()), supervisor_host: None, service_route_index: Arc::new(OnceLock::new()), } @@ -9824,7 +9824,7 @@ mod tests { restart_authority: Arc::new(OnceLock::new()), restart_notify: Arc::new(Notify::new()), ssh_identities: Arc::new(OnceLock::new()), - supervisor_ocsf_headers: Arc::new(OnceLock::new()), + supervisor_otlp_headers: Arc::new(OnceLock::new()), supervisor_host: None, service_route_index: Arc::new(OnceLock::new()), } @@ -13633,17 +13633,20 @@ mod tests { } #[tokio::test] - async fn launch_bundles_carry_supervisor_ocsf_headers() { + async fn launch_bundles_carry_each_supervisor_exports_headers() { + use openshell_core::jwt::SecretOtlpHeaders; + let driver = ControlledDriver::new(); let runtime = test_runtime(driver.clone()).await; - runtime.configure_supervisor_ocsf_headers(Some( - openshell_core::jwt::SecretOtlpHeaders::new( - "authorization=Bearer%20secret".to_string(), - ), - )); + let secret = |value: &str| Some(SecretOtlpHeaders::new(value.to_string())); + runtime.configure_supervisor_otlp_headers(crate::otel_tracing::SupervisorOtlpHeaders { + traces: secret("authorization=traces"), + ocsf: secret("authorization=Bearer%20secret"), + agent: secret("x-mlflow-experiment-id=0"), + }); let sandbox = sandbox_record( - "sb-ocsf-headers", - "sandbox-ocsf-headers", + "sb-otlp-headers", + "sandbox-otlp-headers", SandboxPhase::Stopped, ); runtime.store.put_message(&sandbox).await.unwrap(); @@ -13660,13 +13663,24 @@ mod tests { let authentications = driver.start_authentications(); let authentication: openshell_core::jwt::SandboxLaunchAuthentication = serde_json::from_slice(&authentications[0]).unwrap(); - assert_eq!( - authentication - .supervisor - .ocsf_otlp_headers + let exposed = |headers: &Option| { + headers .as_ref() - .map(openshell_core::jwt::SecretOtlpHeaders::expose_secret), - Some("authorization=Bearer%20secret") + .map(SecretOtlpHeaders::expose_secret) + .map(str::to_string) + }; + let bundle = &authentication.supervisor; + assert_eq!( + [ + exposed(&bundle.traces_otlp_headers), + exposed(&bundle.ocsf_otlp_headers), + exposed(&bundle.agent_otlp_headers), + ], + [ + Some("authorization=traces".to_string()), + Some("authorization=Bearer%20secret".to_string()), + Some("x-mlflow-experiment-id=0".to_string()), + ] ); } diff --git a/crates/openshell-server/src/config_file.rs b/crates/openshell-server/src/config_file.rs index 2c74c19360..4615f778bd 100644 --- a/crates/openshell-server/src/config_file.rs +++ b/crates/openshell-server/src/config_file.rs @@ -416,6 +416,252 @@ pub struct OtlpCollectorConfig { /// Transport to the collector. Defaults to gRPC. #[serde(default, skip_serializing_if = "is_grpc")] pub protocol: OtlpProtocol, + + /// Headers sent with every export request, such as an authorization + /// token. + #[serde(default, skip_serializing_if = "OtlpHeaders::is_unserialized")] + pub headers: OtlpHeaders, + + /// File the headers are read from, one `name=value` per line. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub headers_file: Option, +} + +/// Headers the exporters set themselves, or that HTTP/2 forbids. +const RESERVED_HEADERS: &[&str] = &[ + "connection", + "content-encoding", + "content-length", + "content-type", + "host", + "keep-alive", + "proxy-connection", + "te", + "trailer", + "transfer-encoding", + "upgrade", +]; + +/// Most bytes read from a `headers_file`. +const MAX_HEADERS_FILE_BYTES: u64 = 64 * 1024; + +/// Collector request headers, validated as HTTP headers when the config is +/// parsed. Headers read from a `headers_file` are not serialized. +#[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(from = "BTreeMap", into = "BTreeMap")] +pub struct OtlpHeaders { + headers: BTreeMap, + from_file: bool, +} + +impl OtlpHeaders { + /// The headers of table `table`, from its `headers` or its + /// `headers_file`. + fn resolve( + table: &str, + headers: Option>, + file: Option<&Path>, + ) -> Result { + let Some(path) = file else { + return Self::validated( + &format!("{table}.headers"), + headers + .unwrap_or_default() + .into_iter() + .map(|(name, value)| (name, value, HeaderSource::Table)), + ); + }; + if headers.is_some() { + return Err(format!("{table} sets both headers and headers_file")); + } + let field = format!("{table}.headers_file"); + let headers = Self::validated(&field, read_headers_file(&field, path)?)?; + Ok(Self { + from_file: true, + ..headers + }) + } + + /// Accepts headers that every exporter sends the same way: valid names, + /// sent once each, with non-empty, visible-ASCII values. Errors never + /// quote a value or a name that failed validation, and cite a file's + /// line rather than its header names. + fn validated( + field: &str, + headers: impl IntoIterator, + ) -> Result { + let mut seen = std::collections::BTreeSet::new(); + let mut validated = BTreeMap::new(); + for (name, value, source) in headers { + let Ok(header) = http::HeaderName::try_from(name.as_str()) else { + return Err(match source { + HeaderSource::Table => format!("{field}: invalid header name"), + HeaderSource::Line(line) => { + format!("{field}: line {line} has an invalid header name") + } + }); + }; + let header = header.as_str(); + if !seen.insert(header.to_string()) { + return Err(match source { + HeaderSource::Table => { + format!("{field}: header {header:?} is set more than once") + } + HeaderSource::Line(line) => { + format!("{field}: line {line} repeats an earlier header") + } + }); + } + if RESERVED_HEADERS.contains(&header) { + return Err(format!("{field}: header {header:?} is set by the exporter")); + } + if header.starts_with("grpc-") { + return Err(match source { + HeaderSource::Table => { + format!("{field}: header {header:?} is set by the exporter") + } + HeaderSource::Line(line) => { + format!("{field}: line {line} sets a header that the exporter sets") + } + }); + } + if header.ends_with("-bin") { + return Err(match source { + HeaderSource::Table => { + format!("{field}: binary header {header:?} is not supported") + } + HeaderSource::Line(line) => { + format!("{field}: line {line} sets a binary header, which is not supported") + } + }); + } + if value.is_empty() { + return Err(match source { + HeaderSource::Table => format!("{field}: empty value for header {header:?}"), + HeaderSource::Line(line) => format!("{field}: line {line} has an empty value"), + }); + } + if !value + .bytes() + .all(|byte| byte == b' ' || byte.is_ascii_graphic()) + { + return Err(match source { + HeaderSource::Table => format!("{field}: invalid value for header {header:?}"), + HeaderSource::Line(line) => { + format!("{field}: line {line} has an invalid value") + } + }); + } + validated.insert(name, value); + } + Ok(Self { + headers: validated, + from_file: false, + }) + } + + pub fn is_empty(&self) -> bool { + self.headers.is_empty() + } + + /// Whether serialization omits these headers: there are none, or they + /// came from a `headers_file`. + fn is_unserialized(&self) -> bool { + self.from_file || self.is_empty() + } + + pub fn header_map(&self) -> http::HeaderMap { + self.headers + .iter() + .filter_map(|(name, value)| { + Some(( + http::HeaderName::try_from(name.as_str()).ok()?, + http::HeaderValue::try_from(value.as_str()).ok()?, + )) + }) + .collect() + } +} + +#[cfg(test)] +impl From<[(&str, &str); N]> for OtlpHeaders { + fn from(headers: [(&str, &str); N]) -> Self { + Self::from( + headers + .into_iter() + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect::>(), + ) + } +} + +impl From> for OtlpHeaders { + fn from(headers: BTreeMap) -> Self { + Self { + headers, + from_file: false, + } + } +} + +impl From for BTreeMap { + fn from(headers: OtlpHeaders) -> Self { + headers.headers + } +} + +/// Where a header came from, for validation errors. +#[derive(Clone, Copy)] +enum HeaderSource { + Table, + Line(usize), +} + +/// Reads one `name=value` header per line, skipping blank lines. +fn read_headers_file( + field: &str, + path: &Path, +) -> Result, String> { + let mut contents = Vec::new(); + std::fs::File::open(path) + .and_then(|file| { + file.take(MAX_HEADERS_FILE_BYTES + 1) + .read_to_end(&mut contents) + }) + .map_err(|error| format!("{field}: cannot read {}: {error}", path.display()))?; + if contents.len() as u64 > MAX_HEADERS_FILE_BYTES { + return Err(format!( + "{field}: {} is larger than {MAX_HEADERS_FILE_BYTES} bytes", + path.display() + )); + } + let contents = String::from_utf8(contents) + .map_err(|_| format!("{field}: {} is not UTF-8", path.display()))?; + let mut headers = Vec::new(); + for (index, line) in contents.lines().enumerate() { + let line = line.trim(); + if line.is_empty() { + continue; + } + let (name, value) = line + .split_once('=') + .ok_or_else(|| format!("{field}: line {} is not name=value", index + 1))?; + headers.push(( + name.trim().to_string(), + value.trim().to_string(), + HeaderSource::Line(index + 1), + )); + } + Ok(headers) +} + +impl std::fmt::Debug for OtlpHeaders { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_map() + .entries(self.headers.keys().map(|name| (name, "[REDACTED]"))) + .finish() + } } impl OtlpCollectorConfig { @@ -487,6 +733,10 @@ struct RawOtlpConfig { #[serde(default)] protocol: Option, #[serde(default)] + headers: Option>, + #[serde(default)] + headers_file: Option, + #[serde(default)] service_name: Option, #[serde(default)] traces: Option, @@ -505,6 +755,10 @@ struct RawAgentOtlp { endpoint: Option, #[serde(default)] protocol: Option, + #[serde(default)] + headers: Option>, + #[serde(default)] + headers_file: Option, } #[derive(Deserialize)] @@ -514,6 +768,10 @@ struct RawOtlpTraces { endpoint: Option, #[serde(default)] protocol: Option, + #[serde(default)] + headers: Option>, + #[serde(default)] + headers_file: Option, } #[derive(Deserialize)] @@ -526,50 +784,89 @@ struct RawOcsfOtlp { #[serde(default)] protocol: Option, #[serde(default)] + headers: Option>, + #[serde(default)] + headers_file: Option, + #[serde(default)] schema_version: Option, } +/// The collector keys every OTLP table accepts. +struct RawCollector { + endpoint: Option, + protocol: Option, + headers: Option>, + headers_file: Option, +} + +macro_rules! raw_collector { + ($raw:expr) => { + RawCollector { + endpoint: $raw.endpoint, + protocol: $raw.protocol, + headers: $raw.headers, + headers_file: $raw.headers_file, + } + }; +} + +impl RawCollector { + /// The first key set that needs the table's own endpoint. + fn first_key_set(&self) -> Option<&'static str> { + [ + ("protocol", self.protocol.is_some()), + ("headers", self.headers.is_some()), + ("headers_file", self.headers_file.is_some()), + ] + .into_iter() + .find_map(|(key, set)| set.then_some(key)) + } + + /// The collector for table `otlp.{table}`: its own when it sets an + /// endpoint, otherwise the whole `shared` one. + fn resolve( + mut self, + table: &str, + shared: Option<&OtlpCollectorConfig>, + ) -> Result { + let Some(endpoint) = self.endpoint.take() else { + if let Some(key) = self.first_key_set() { + return Err(format!( + "{table} sets {key} without its own endpoint; a table without an \ + endpoint inherits the whole otlp collector" + )); + } + return shared.cloned().ok_or_else(|| { + format!("{table} needs an endpoint, or otlp.endpoint to default to") + }); + }; + OtlpCollectorConfig { + endpoint, + protocol: self.protocol.unwrap_or_default(), + headers: OtlpHeaders::resolve(table, self.headers, self.headers_file.as_deref())?, + headers_file: self.headers_file, + } + .checked(table) + } +} + impl TryFrom for OtlpConfig { type Error = String; fn try_from(raw: RawOtlpConfig) -> Result { - if raw.endpoint.is_none() && raw.protocol.is_some() { - return Err( - "otlp sets protocol without its own endpoint; a table with its own \ - endpoint takes nothing from otlp" - .to_string(), - ); - } - let shared = raw - .endpoint - .clone() - .map(|endpoint| { - OtlpCollectorConfig { - endpoint, - protocol: raw.protocol.unwrap_or_default(), - } - .checked("otlp") - }) - .transpose()?; - let collector_for = - |table: &str, endpoint: Option, protocol: Option| match ( - endpoint, protocol, - ) { - (Some(endpoint), protocol) => OtlpCollectorConfig { - endpoint, - protocol: protocol.unwrap_or_default(), - } - .checked(&format!("otlp.{table}")), - (None, Some(_)) => Err(format!( - "otlp.{table} sets protocol without its own endpoint; a table without \ - an endpoint inherits the whole otlp collector" - )), - (None, None) => shared.clone().ok_or_else(|| { - format!("otlp.{table} needs an endpoint, or otlp.endpoint to default to") - }), - }; + let shared = raw_collector!(raw); + let shared = match (&shared.endpoint, shared.first_key_set()) { + (Some(_), _) => Some(shared.resolve("otlp", None)?), + (None, Some(key)) => { + return Err(format!( + "otlp sets {key} without its own endpoint; a table with its own \ + endpoint takes nothing from otlp" + )); + } + (None, None) => None, + }; let traces = match raw.traces { - Some(traces) => Some(collector_for("traces", traces.endpoint, traces.protocol)?), + Some(traces) => Some(raw_collector!(traces).resolve("otlp.traces", shared.as_ref())?), None => shared.clone(), } .map(|collector| OtlpTracesConfig { collector }); @@ -578,8 +875,8 @@ impl TryFrom for OtlpConfig { .filter(|ocsf| ocsf.enabled) .map(|ocsf| { Ok::<_, String>(OcsfOtlpConfig { - collector: collector_for("ocsf", ocsf.endpoint, ocsf.protocol)?, schema_version: ocsf.schema_version, + collector: raw_collector!(ocsf).resolve("otlp.ocsf", shared.as_ref())?, }) }) .transpose()?; @@ -588,7 +885,7 @@ impl TryFrom for OtlpConfig { .filter(|agent| agent.enabled) .map(|agent| { Ok::<_, String>(AgentOtlpConfig { - collector: collector_for("agent", agent.endpoint, agent.protocol)?, + collector: raw_collector!(agent).resolve("otlp.agent", shared.as_ref())?, }) }) .transpose()?; @@ -839,12 +1136,10 @@ pub enum ConfigFileError { }, #[error("gateway config path '{}' is not a regular file ({kind})", path.display())] NonRegular { path: PathBuf, kind: &'static str }, - #[error("failed to parse gateway config file '{}': {source}", path.display())] - Parse { - path: PathBuf, - #[source] - source: Box, - }, + /// Names the location and the parser's message but never quotes the + /// file, which can hold credentials such as collector headers. + #[error("failed to parse gateway config file '{}': {detail}", path.display())] + Parse { path: PathBuf, detail: String }, #[error( "gateway config schema version is required; add `[openshell]` and `version = {SCHEMA_VERSION}`" )] @@ -982,13 +1277,53 @@ fn read_config_contents(path: &Path) -> Result { Ok(contents) } +/// `error`'s message, prefixed with its line and column in `contents`. +fn parse_error_detail(contents: &str, error: &toml::de::Error) -> String { + let message = without_quoted_strings(error.message().trim_end()); + let Some(span) = error.span() else { + return message; + }; + let before = &contents[..span.start.min(contents.len())]; + let line = before.matches('\n').count() + 1; + let line_start = before.rfind('\n').map_or(0, |newline| newline + 1); + let column = before[line_start..].chars().count() + 1; + format!("line {line}, column {column}: {message}") +} + +/// Replaces each `string "..."` that serde quotes in a type error with `a +/// string`, since the quoted value may be a credential. +fn without_quoted_strings(message: &str) -> String { + const QUOTED: &str = "string \""; + let mut result = String::with_capacity(message.len()); + let mut rest = message; + while let Some(start) = rest.find(QUOTED) { + result.push_str(&rest[..start]); + result.push_str("a string"); + let quoted = &rest[start + QUOTED.len()..]; + let mut escaped = false; + let end = quoted.char_indices().find_map(|(index, character)| { + if escaped { + escaped = false; + } else if character == '\\' { + escaped = true; + } else if character == '"' { + return Some(index + 1); + } + None + }); + rest = end.map_or("", |end| "ed[end..]); + } + result.push_str(rest); + result +} + fn parse_and_validate(path: &Path, contents: &str) -> Result { if contents.trim().is_empty() { return Err(ConfigFileError::MissingVersion); } - let file: ConfigFile = toml::from_str(contents).map_err(|source| ConfigFileError::Parse { + let file: ConfigFile = toml::from_str(contents).map_err(|error| ConfigFileError::Parse { path: path.to_path_buf(), - source: Box::new(source), + detail: parse_error_detail(contents, &error), })?; match file.openshell.version { @@ -1601,6 +1936,8 @@ service_name = "openshell-gateway-dev" collector: OtlpCollectorConfig { endpoint: "http://127.0.0.1:4317".into(), protocol: OtlpProtocol::Grpc, + headers: OtlpHeaders::default(), + headers_file: None, }, schema_version: None, }) @@ -1621,6 +1958,8 @@ service_name = "openshell-gateway-dev" collector: OtlpCollectorConfig { endpoint: "http://siem:4317".into(), protocol: OtlpProtocol::Grpc, + headers: OtlpHeaders::default(), + headers_file: None, }, schema_version: None, }) @@ -1808,13 +2147,261 @@ service_name = "openshell-gateway-dev" } } + #[test] + fn otlp_lane_headers_follow_the_collector_they_belong_to() { + let headers = |otlp: &Option<_>| { + otlp.as_ref() + .map(|collector: &OtlpCollectorConfig| collector.headers.header_map()) + .unwrap_or_default() + }; + let otlp = otlp( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4318'\n\ + headers = { authorization = 'Bearer shared', X-Tenant = 'blue' }\n\n\ + [openshell.gateway.otlp.ocsf]\nenabled = true\n\n\ + [openshell.gateway.otlp.agent]\nenabled = true\nendpoint = 'http://mlflow:5000'\n\ + headers = { x-mlflow-experiment-id = '0' }\n", + ); + let traces = headers(&otlp.traces.map(|traces| traces.collector)); + assert_eq!(traces["authorization"], "Bearer shared"); + assert_eq!(traces["x-tenant"], "blue"); + assert_eq!(headers(&otlp.ocsf.map(|ocsf| ocsf.collector)), traces); + let agent = headers(&otlp.agent.map(|agent| agent.collector)); + assert_eq!( + agent.len(), + 1, + "a lane with its own endpoint inherits no headers" + ); + assert_eq!(agent["x-mlflow-experiment-id"], "0"); + } + + #[test] + fn otlp_lane_headers_need_their_own_endpoint() { + for table in [ + "[openshell.gateway.otlp.traces]\nheaders = { a = 'b' }\n", + "[openshell.gateway.otlp.ocsf]\nenabled = true\nheaders = { a = 'b' }\n", + "[openshell.gateway.otlp.agent]\nenabled = true\nheaders = { a = 'b' }\n", + "[openshell.gateway.otlp.traces]\nheaders_file = '/run/otlp/headers'\n", + ] { + let tmp = write_tmp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\n\n{table}" + )); + let error = load(tmp.path()).expect_err("headers without endpoint"); + assert!( + format!("{error:#}").contains("without its own endpoint"), + "{table}: {error:#}" + ); + } + let tmp = write_tmp( + "[openshell.gateway.otlp]\nheaders = { authorization = 'Bearer t' }\n\n\ + [openshell.gateway.otlp.traces]\nendpoint = 'http://traces:4317'\n", + ); + let error = load(tmp.path()).expect_err("shared headers without a shared endpoint"); + assert!( + format!("{error:#}").contains("otlp sets headers without its own endpoint"), + "{error:#}" + ); + } + + #[test] + fn otlp_headers_must_be_valid_http_headers() { + for (headers, problem) in [ + ( + "{ x-empty = '' }", + "otlp.headers: empty value for header \"x-empty\"", + ), + ( + "{ x-text = 'SECRETVAL-café' }", + "otlp.headers: invalid value for header \"x-text\"", + ), + ( + "{ Authorization = 'SECRETVAL-1', authorization = 'SECRETVAL-2' }", + "otlp.headers: header \"authorization\" is set more than once", + ), + ( + "{ x-data-bin = 'SECRETVAL' }", + "otlp.headers: binary header \"x-data-bin\"", + ), + ( + "{ Content-Type = 'SECRETVAL' }", + "otlp.headers: header \"content-type\" is set by the exporter", + ), + ( + "{ grpc-timeout = 'SECRETVAL' }", + "otlp.headers: header \"grpc-timeout\" is set by the exporter", + ), + ( + "{ 'Bearer SECRETNAME' = 'SECRETVAL' }", + "otlp.headers: invalid header name", + ), + ( + "{ authorization = \"SECRETVAL\\nbreak\" }", + "otlp.headers: invalid value for header \"authorization\"", + ), + ] { + let tmp = write_tmp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\nheaders = {headers}\n" + )); + let error = format!("{:#}", load(tmp.path()).expect_err("invalid header")); + assert!(error.contains(problem), "{headers}: {error}"); + for secret in ["SECRETVAL", "SECRETNAME", "break"] { + assert!(!error.contains(secret), "{headers}: {error}"); + } + } + } + + fn headers_file(contents: &[u8]) -> tempfile::NamedTempFile { + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), contents).unwrap(); + file + } + + #[test] + fn otlp_headers_file_supplies_the_collector_headers() { + let file = headers_file(b"authorization = Bearer file-token\n\nx-tenant=blue\n"); + let path = file.path().display(); + let otlp = otlp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\nheaders_file = '{path}'\n\n\ + [openshell.gateway.otlp.agent]\nenabled = true\nendpoint = 'http://agents:4318'\n\ + headers_file = '{path}'\n" + )); + let expected = + OtlpHeaders::from([("authorization", "Bearer file-token"), ("x-tenant", "blue")]) + .header_map(); + let traces = otlp.traces.as_ref().unwrap(); + assert_eq!(traces.collector.headers.header_map(), expected); + assert_eq!( + otlp.agent.as_ref().unwrap().collector.headers.header_map(), + expected + ); + + let encoded = toml::to_string(&otlp).unwrap(); + assert!(encoded.contains("headers_file"), "{encoded}"); + assert!(!encoded.contains("file-token"), "{encoded}"); + } + + #[test] + fn otlp_headers_file_excludes_inline_headers() { + let file = headers_file(b"authorization=Bearer t\n"); + let tmp = write_tmp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\n\ + headers = {{ x-tenant = 'blue' }}\nheaders_file = '{}'\n", + file.path().display() + )); + let error = load(tmp.path()).expect_err("headers and headers_file"); + assert!( + format!("{error:#}").contains("otlp sets both headers and headers_file"), + "{error:#}" + ); + } + + #[test] + fn otlp_headers_file_is_validated_like_inline_headers() { + let too_large = format!("x-big={}\n", "a".repeat(64 * 1024)); + for (contents, problem) in [ + ( + b"authorization: Bearer SECRETVAL\n".as_slice(), + "otlp.headers_file: line 1 is not name=value", + ), + ( + b"Authorization: Basic SECRETVAL==\n".as_slice(), + "otlp.headers_file: line 1 has an invalid header name", + ), + ( + b"x-ok=1\nSECRETVAL=\n".as_slice(), + "otlp.headers_file: line 2 has an empty value", + ), + ( + b"a=SECRETVAL-1\nA=SECRETVAL-2\n".as_slice(), + "otlp.headers_file: line 2 repeats an earlier header", + ), + ( + b"content-type=SECRETVAL\n".as_slice(), + "otlp.headers_file: header \"content-type\" is set by the exporter", + ), + ( + b"SECRETVAL-bin=AAAA\n".as_slice(), + "otlp.headers_file: line 1 sets a binary header", + ), + ( + b"x-text=SECRETVAL-caf\xc3\xa9\n".as_slice(), + "otlp.headers_file: line 1 has an invalid value", + ), + (b"x-bytes=SECRETVAL\xff\n", "is not UTF-8"), + (too_large.as_bytes(), "is larger than 65536 bytes"), + ] { + let file = headers_file(contents); + let tmp = write_tmp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\nheaders_file = '{}'\n", + file.path().display() + )); + let error = format!("{:#}", load(tmp.path()).expect_err(problem)); + assert!(error.contains(problem), "{error}"); + assert!(error.contains("otlp.headers_file: "), "{error}"); + assert!(!error.contains("SECRETVAL"), "{error}"); + } + + let tmp = write_tmp( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\n\ + headers_file = '/nonexistent/otlp-headers'\n", + ); + let error = format!("{:#}", load(tmp.path()).expect_err("missing file")); + assert!( + error.contains("otlp.headers_file: cannot read /nonexistent/otlp-headers"), + "{error}" + ); + } + + #[test] + fn parse_errors_locate_the_problem_without_quoting_the_file() { + for (headers, expected) in [ + ( + "{ authorization = 'Bearer SECRETVAL', x-tenant = 42 }", + ["line 6", "expected a string"], + ), + ( + "'authorization=Bearer%20SECRETVAL'", + ["line 6", "invalid type: a string, expected a map"], + ), + ( + "{ x-tenant = 'café', x-n = 42 }", + ["line 6, column 38", "expected a string"], + ), + ] { + let tmp = write_tmp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\nheaders = {headers}\n" + )); + let error = load(tmp.path()).expect_err("an invalid headers value"); + let mut report = error.to_string(); + let mut source = std::error::Error::source(&error); + while let Some(error) = source { + report = format!("{report}\n{error}"); + source = error.source(); + } + for expected in expected { + assert!(report.contains(expected), "{headers}: {report}"); + } + assert!(!report.contains("SECRETVAL"), "{headers}: {report}"); + } + } + + #[test] + fn otlp_header_values_are_redacted_from_debug_output() { + let otlp = otlp( + "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\n\ + headers = { authorization = 'Bearer secret-token' }\n", + ); + let debug = format!("{otlp:?}"); + assert!(debug.contains("authorization"), "{debug}"); + assert!(!debug.contains("secret-token"), "{debug}"); + } + #[test] fn otlp_config_round_trips() { let tmp = write_tmp( "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\nservice_name = 'gw'\n\n\ [openshell.gateway.otlp.ocsf]\nenabled = true\n\n\ [openshell.gateway.otlp.agent]\nenabled = true\nendpoint = 'http://agents:4318'\n\ - protocol = 'http/protobuf'\n", + protocol = 'http/protobuf'\nheaders = { x-mlflow-experiment-id = '0' }\n", ); let file = load(tmp.path()).unwrap(); let encoded = toml::to_string(&file).unwrap(); @@ -1875,7 +2462,9 @@ policy_validation_failure_mode = "keep_old" "#, ); let error = load(tmp.path()).expect_err("unknown posture must fail TOML validation"); - assert!(error.to_string().contains("policy_validation_failure_mode")); + let error = error.to_string(); + assert!(error.contains("line 6, column 34"), "{error}"); + assert!(error.contains("unknown variant `keep_old`"), "{error}"); } #[test] diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 12ca9223ea..d64c395426 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -292,6 +292,8 @@ fn spawn_supervisor_token_refresh( } pub use auth::audit::ConnectionSource; pub use multiplex::{MultiplexService, MultiplexedService}; +pub use otel_tracing::gateway_trace_headers; +use otel_tracing::supervisor_otlp_headers; pub use persistence::Store; use sandbox_index::SandboxIndex; use sandbox_watch::SandboxWatchBus; @@ -738,7 +740,7 @@ pub(crate) async fn run_server( shutdown_rx.clone(), ) .await?; - compute.configure_supervisor_ocsf_headers(otel_tracing::supervisor_ocsf_headers( + compute.configure_supervisor_otlp_headers(supervisor_otlp_headers( config_file .as_ref() .and_then(|file| file.openshell.gateway.otlp.as_ref()), diff --git a/crates/openshell-server/src/otel_tracing.rs b/crates/openshell-server/src/otel_tracing.rs index 426c283f83..61a1aac441 100644 --- a/crates/openshell-server/src/otel_tracing.rs +++ b/crates/openshell-server/src/otel_tracing.rs @@ -122,6 +122,9 @@ pub struct OtlpExport { /// Validated collector that supervisors relay workload traces to. The /// gateway itself exports nothing for this signal. agent: Option>, + /// Headers configured for the trace collector, which in-process compute + /// drivers also send. + traces_headers: http::HeaderMap, } /// One signal's exporter and the endpoint it sends to. @@ -155,8 +158,11 @@ impl OtlpExport { gateway: GatewayResourceAttributes<'_>, ) -> Result { let target = |collector: &OtlpCollectorConfig| { - OtlpTarget::new(&export_config(cfg, &collector.endpoint, gateway)) - .map(|target| target.with_protocol(collector.protocol.into())) + OtlpTarget::new(&export_config(cfg, &collector.endpoint, gateway)).map(|target| { + target + .with_protocol(collector.protocol.into()) + .with_headers(collector.headers.header_map()) + }) }; let traces = cfg .traces @@ -197,6 +203,11 @@ impl OtlpExport { traces, ocsf, agent, + traces_headers: cfg + .traces + .as_ref() + .map(|traces| traces.collector.headers.header_map()) + .unwrap_or_default(), }) } @@ -229,6 +240,11 @@ impl OtlpExport { .map_or(OtlpProtocol::Grpc, |traces| traces.protocol) } + /// The headers configured for the trace collector. + pub fn traces_headers(&self) -> http::HeaderMap { + self.traces_headers.clone() + } + /// The trace provider, when trace export is enabled. pub fn tracer_provider(&self) -> Option<&SdkTracerProvider> { self.traces.as_ref().map(|traces| &traces.exporter) @@ -258,23 +274,86 @@ impl OtlpExport { } } -/// Collector headers for supervisors to send with their OCSF exports: the -/// headers the gateway's own OCSF exporter reads from its environment, with the -/// same precedence. Supervisors receive them in their launch bundle because -/// they do not inherit the gateway's environment. -pub fn supervisor_ocsf_headers(otlp: Option<&OtlpConfig>) -> Option { - ocsf_headers_from(otlp, |name| std::env::var(name).ok()) +/// Collector headers for each supervisor export, delivered in the launch +/// bundle because supervisors do not inherit the gateway's environment. +#[derive(Clone, Debug, Default)] +pub struct SupervisorOtlpHeaders { + pub traces: Option, + pub ocsf: Option, + pub agent: Option, } -fn ocsf_headers_from( - otlp: Option<&OtlpConfig>, +impl SupervisorOtlpHeaders { + pub fn apply(&self, bundle: &mut openshell_core::jwt::SupervisorAuthBundle) { + bundle.traces_otlp_headers.clone_from(&self.traces); + bundle.ocsf_otlp_headers.clone_from(&self.ocsf); + bundle.agent_otlp_headers.clone_from(&self.agent); + } +} + +/// The headers each enabled supervisor export sends. +/// +/// Traces and the agent relay send their collector's configured headers. OCSF +/// export also sends the headers the gateway's own OCSF exporter reads from +/// its environment, which override configured ones. +pub fn supervisor_otlp_headers(otlp: Option<&OtlpConfig>) -> SupervisorOtlpHeaders { + supervisor_otlp_headers_from(otlp, |name| std::env::var(name).ok()) +} + +/// The headers the gateway's own trace exporter sends: the trace collector's +/// configured headers, overridden by the ones it reads from its environment. +pub fn gateway_trace_headers(otlp: Option<&OtlpConfig>) -> Option { + let traces = otlp?.traces.as_ref()?; + secret_headers(with_env_headers( + &traces.collector, + "OTEL_EXPORTER_OTLP_TRACES_HEADERS", + |name| std::env::var(name).ok(), + )) +} + +fn secret_headers(headers: http::HeaderMap) -> Option { + (!headers.is_empty()) + .then(|| SecretOtlpHeaders::new(openshell_otel::encode_otlp_headers(&headers))) +} + +fn with_env_headers( + collector: &OtlpCollectorConfig, + signal_var: &str, var: impl Fn(&str) -> Option, -) -> Option { - otlp?.ocsf.as_ref()?; - var("OTEL_EXPORTER_OTLP_LOGS_HEADERS") +) -> http::HeaderMap { + let env = var(signal_var) .or_else(|| var("OTEL_EXPORTER_OTLP_HEADERS")) - .filter(|headers| !headers.trim().is_empty()) - .map(SecretOtlpHeaders::new) + .map(|headers| openshell_otel::parse_otlp_headers(&headers)) + .unwrap_or_default(); + openshell_otel::merge_otlp_headers(collector.headers.header_map(), env) +} + +fn supervisor_otlp_headers_from( + otlp: Option<&OtlpConfig>, + var: impl Fn(&str) -> Option, +) -> SupervisorOtlpHeaders { + let Some(otlp) = otlp else { + return SupervisorOtlpHeaders::default(); + }; + let configured = + |collector: &OtlpCollectorConfig| secret_headers(collector.headers.header_map()); + SupervisorOtlpHeaders { + traces: otlp + .traces + .as_ref() + .and_then(|traces| configured(&traces.collector)), + ocsf: otlp.ocsf.as_ref().and_then(|ocsf| { + secret_headers(with_env_headers( + &ocsf.collector, + "OTEL_EXPORTER_OTLP_LOGS_HEADERS", + &var, + )) + }), + agent: otlp + .agent + .as_ref() + .and_then(|agent| configured(&agent.collector)), + } } /// Build the `tracing` layer that forwards spans to `provider`. @@ -605,7 +684,7 @@ pub mod test_exporter { #[cfg(test)] mod tests { use super::*; - use crate::config_file::{OcsfOtlpConfig, OtlpTracesConfig}; + use crate::config_file::{AgentOtlpConfig, OcsfOtlpConfig, OtlpHeaders, OtlpTracesConfig}; use crate::test_environment::Environment; fn config() -> OtlpConfig { @@ -619,6 +698,8 @@ mod tests { collector: OtlpCollectorConfig { endpoint: endpoint.into(), protocol: OtlpProtocol::Grpc, + headers: OtlpHeaders::default(), + headers_file: None, }, }), ocsf: None, @@ -631,6 +712,8 @@ mod tests { collector: OtlpCollectorConfig { endpoint: endpoint.into(), protocol: OtlpProtocol::Grpc, + headers: OtlpHeaders::default(), + headers_file: None, }, schema_version: None, } @@ -641,7 +724,7 @@ mod tests { } #[test] - fn supervisors_receive_the_gateway_ocsf_headers() { + fn supervisors_receive_configured_trace_headers_and_merged_ocsf_headers() { let env = |vars: &'static [(&'static str, &'static str)]| { move |name: &str| { vars.iter() @@ -649,32 +732,94 @@ mod tests { .map(|(_, value)| (*value).to_string()) } }; - let headers = |otlp: &OtlpConfig, vars| { - ocsf_headers_from(Some(otlp), env(vars)).map(|h| h.expose_secret().to_string()) + let exposed = |headers: Option| { + headers.map(|headers| headers.expose_secret().to_string()) }; let mut otlp = config(); - let both: &[(&str, &str)] = &[ - ("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "authorization=logs"), - ("OTEL_EXPORTER_OTLP_HEADERS", "authorization=all"), - ]; - assert_eq!(headers(&otlp, both), None, "OCSF export is not enabled"); + otlp.traces.as_mut().unwrap().collector.headers = + [("authorization", "toml"), ("x-tenant", "blue")].into(); + let mut ocsf = ocsf_config("http://siem:4317"); + ocsf.collector.headers = [("authorization", "toml")].into(); + otlp.ocsf = Some(ocsf); + otlp.agent = Some(AgentOtlpConfig { + collector: OtlpCollectorConfig { + endpoint: "http://mlflow:5000".into(), + protocol: OtlpProtocol::HttpProtobuf, + headers: [("x-mlflow-experiment-id", "0")].into(), + headers_file: None, + }, + }); - otlp.ocsf = Some(ocsf_config("http://127.0.0.1:4317")); - assert_eq!(headers(&otlp, both).as_deref(), Some("authorization=logs")); + let headers = supervisor_otlp_headers_from( + Some(&otlp), + env(&[ + ( + "OTEL_EXPORTER_OTLP_TRACES_HEADERS", + "authorization=traces%20env", + ), + ("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "x-extra=logs"), + ("OTEL_EXPORTER_OTLP_HEADERS", "authorization=all"), + ]), + ); assert_eq!( - headers( - &otlp, - &[("OTEL_EXPORTER_OTLP_HEADERS", "authorization=all")] - ) - .as_deref(), - Some("authorization=all") + exposed(headers.traces).as_deref(), + Some("authorization=toml,x-tenant=blue"), + "trace headers from the environment stay with the gateway" + ); + assert_eq!( + exposed(headers.ocsf).as_deref(), + Some("authorization=toml,x-extra=logs"), + "the logs variable replaces the generic one" ); assert_eq!( - headers(&otlp, &[("OTEL_EXPORTER_OTLP_LOGS_HEADERS", " ")]), - None + exposed(headers.agent).as_deref(), + Some("x-mlflow-experiment-id=0"), + "the agent lane never reads the environment" + ); + + let generic = supervisor_otlp_headers_from( + Some(&otlp), + env(&[("OTEL_EXPORTER_OTLP_HEADERS", "authorization=all")]), + ); + assert_eq!(exposed(generic.ocsf).as_deref(), Some("authorization=all")); + + let unconfigured = supervisor_otlp_headers_from( + Some(&config()), + env(&[ + ("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "authorization=traces"), + ("OTEL_EXPORTER_OTLP_HEADERS", "authorization=all"), + ("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "authorization=logs"), + ]), + ); + assert!( + unconfigured.traces.is_none(), + "environment-only trace headers add nothing to launch credentials" + ); + assert!(unconfigured.ocsf.is_none(), "OCSF export is not enabled"); + assert!(unconfigured.agent.is_none()); + assert!( + supervisor_otlp_headers_from(None, env(&[("OTEL_EXPORTER_OTLP_HEADERS", "a=b")])) + .traces + .is_none() ); - assert_eq!(headers(&otlp, &[]), None); - assert!(ocsf_headers_from(None, env(both)).is_none()); + } + + #[test] + fn gateway_trace_headers_let_the_environment_win() { + let mut otlp = config(); + otlp.traces.as_mut().unwrap().collector.headers = + [("authorization", "toml"), ("x-tenant", "blue")].into(); + Environment::new() + .set("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "authorization=env") + .remove("OTEL_EXPORTER_OTLP_HEADERS") + .run(|| { + assert_eq!( + gateway_trace_headers(Some(&otlp)) + .map(|headers| headers.expose_secret().to_string()) + .as_deref(), + Some("authorization=env,x-tenant=blue") + ); + }); } #[tokio::test] @@ -751,6 +896,8 @@ mod tests { let http = |endpoint: &str| OtlpCollectorConfig { endpoint: endpoint.into(), protocol: OtlpProtocol::HttpProtobuf, + headers: OtlpHeaders::default(), + headers_file: None, }; let mut cfg = config(); cfg.traces = Some(OtlpTracesConfig { @@ -786,38 +933,120 @@ mod tests { assert_eq!(paths, ["/v1/logs", "/v1/traces"]); } - /// Supervisors authenticate to the collector with the headers from their - /// launch bundle. - #[tokio::test] - async fn ocsf_export_sends_explicit_collector_headers() { + /// Exports one span and one OCSF event through `export`, then shuts it + /// down. + async fn export_span_and_event(export: OtlpExport) { use openshell_ocsf::{ConfigStateChangeBuilder, ocsf_emit}; + use opentelemetry::trace::{Tracer as _, TracerProvider as _}; use tracing_subscriber::prelude::*; - let collector = openshell_otel_test_support::OtlpTestServer::start().await; - let target = OtlpTarget::new(&OtlpExportConfig { - endpoint: collector.endpoint(), - service_name: ServiceName::Fixed("openshell-supervisor"), - service_version: None, - resource_attributes: Vec::new(), - }) - .unwrap(); - let export = OcsfOtlp::start( - target - .log_exporter_with_headers(Some("authorization=Bearer%20collector-token")) - .unwrap(), - target.resource().clone(), - ); + export + .tracer_provider() + .expect("trace export") + .tracer("test") + .in_span("gateway-span", |_| {}); let event = ConfigStateChangeBuilder::new(&crate::gateway_ocsf::context("sandbox-1", "sandbox")) .message("policy approved") .build(); - let subscriber = tracing_subscriber::registry().with(export.layer()); + let subscriber = + tracing_subscriber::registry().with(export.ocsf().expect("OCSF log export").layer()); tracing::subscriber::with_default(subscriber, || ocsf_emit!(event)); - export.shutdown().await; - let received = collector.shutdown().await; + tokio::task::spawn_blocking(move || { + tokio::runtime::Handle::current().block_on(export.shutdown()); + }) + .await + .unwrap(); + } - assert_eq!(received.log_bodies.len(), 1); - assert_eq!(received.log_authorizations, ["Bearer collector-token"]); + #[tokio::test(flavor = "multi_thread")] + async fn gateway_signals_send_their_configured_headers() { + Environment::new() + .remove("OTEL_EXPORTER_OTLP_HEADERS") + .remove("OTEL_EXPORTER_OTLP_TRACES_HEADERS") + .remove("OTEL_EXPORTER_OTLP_LOGS_HEADERS") + .run_async(async { + let collector = openshell_otel_test_support::OtlpTestServer::start().await; + let mut cfg = with_traces_endpoint(collector.endpoint()); + cfg.traces.as_mut().unwrap().collector.headers = + [("authorization", "Bearer traces-token")].into(); + let mut ocsf = ocsf_config(collector.endpoint()); + ocsf.collector.headers = [("authorization", "Bearer logs-token")].into(); + cfg.ocsf = Some(ocsf); + let (export, error) = + OtlpExport::start(Some(&cfg), GatewayResourceAttributes::default()); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + let export = export.expect("OTLP export"); + assert_eq!( + export.traces_headers()["authorization"], + "Bearer traces-token", + "in-process compute drivers send the trace collector's headers" + ); + + export_span_and_event(export).await; + let received = collector.shutdown().await; + + assert_eq!(received.trace_authorizations, ["Bearer traces-token"]); + assert_eq!(received.log_authorizations, ["Bearer logs-token"]); + }) + .await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn gateway_signals_let_environment_headers_win() { + Environment::new() + .set( + "OTEL_EXPORTER_OTLP_TRACES_HEADERS", + "authorization=traces-env", + ) + .set("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "authorization=logs-env") + .remove("OTEL_EXPORTER_OTLP_HEADERS") + .run_async(async { + let configured = |endpoint: &str, protocol| OtlpCollectorConfig { + endpoint: endpoint.into(), + protocol, + headers: [("authorization", "toml"), ("x-tenant", "blue")].into(), + headers_file: None, + }; + let start = |collector: OtlpCollectorConfig| { + let mut cfg = config(); + cfg.traces = Some(OtlpTracesConfig { + collector: collector.clone(), + }); + cfg.ocsf = Some(OcsfOtlpConfig { + collector, + schema_version: None, + }); + let (export, error) = + OtlpExport::start(Some(&cfg), GatewayResourceAttributes::default()); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + export.expect("OTLP export") + }; + + let grpc = openshell_otel_test_support::OtlpTestServer::start().await; + export_span_and_event(start(configured(grpc.endpoint(), OtlpProtocol::Grpc))).await; + let received = grpc.shutdown().await; + assert_eq!(received.trace_authorizations, ["traces-env"]); + assert_eq!(received.log_authorizations, ["logs-env"]); + + let mut http = openshell_otel_test_support::OtlpHttpTestServer::start().await; + export_span_and_event(start(configured( + http.endpoint(), + OtlpProtocol::HttpProtobuf, + ))) + .await; + for _ in 0..2 { + let export = http.next_export().await; + let expected = if export.path == "/v1/traces" { + "traces-env" + } else { + "logs-env" + }; + assert_eq!(export.headers["authorization"], expected, "{}", export.path); + assert_eq!(export.headers["x-tenant"], "blue", "{}", export.path); + } + }) + .await; } #[test] @@ -984,10 +1213,12 @@ mod tests { let mut bad_ocsf = config(); bad_ocsf.ocsf = Some(ocsf_config("definitely not a url")); let mut bad_agent = config(); - bad_agent.agent = Some(crate::config_file::AgentOtlpConfig { + bad_agent.agent = Some(AgentOtlpConfig { collector: OtlpCollectorConfig { endpoint: "definitely not a url".into(), protocol: OtlpProtocol::Grpc, + headers: OtlpHeaders::default(), + headers_file: None, }, }); for cfg in [ diff --git a/crates/openshell-server/src/persistence/mutation_lock_pg_tests.rs b/crates/openshell-server/src/persistence/mutation_lock_pg_tests.rs index e0b00b6ed6..c7866520f8 100644 --- a/crates/openshell-server/src/persistence/mutation_lock_pg_tests.rs +++ b/crates/openshell-server/src/persistence/mutation_lock_pg_tests.rs @@ -85,7 +85,9 @@ async fn postgres_concurrent_sandbox_creates_with_ssh_identity_do_not_exhaust_lo sandbox_token: SecretJwt::parse("sandbox-token").unwrap(), sandbox_expires_at: 0, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: "gateway".to_string(), verification_keys: Vec::new(), diff --git a/crates/openshell-server/src/ssh_identity.rs b/crates/openshell-server/src/ssh_identity.rs index 321e7286b2..7401905970 100644 --- a/crates/openshell-server/src/ssh_identity.rs +++ b/crates/openshell-server/src/ssh_identity.rs @@ -72,7 +72,9 @@ mod tests { sandbox_token: SecretJwt::parse("sandbox-token").unwrap(), sandbox_expires_at: 0, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, gateway_id: "gateway".to_string(), verification_keys: Vec::new(), diff --git a/crates/openshell-server/src/tracing_setup.rs b/crates/openshell-server/src/tracing_setup.rs index fd48c751c9..b0ac24af38 100644 --- a/crates/openshell-server/src/tracing_setup.rs +++ b/crates/openshell-server/src/tracing_setup.rs @@ -74,12 +74,16 @@ pub fn install( let driver_protocol = driver_otlp.map_or(openshell_otel::Protocol::Grpc, |otlp| { otlp.traces_protocol().into() }); + let driver_headers = driver_otlp + .map(OtlpExport::traces_headers) + .unwrap_or_default(); let (driver_tracer_provider, driver_setup_error) = driver.map_or_else( || (None, None), |descriptor| { descriptor.provider_for( driver_endpoint, driver_protocol, + &driver_headers, openshell_core::VERSION, gateway.name(), gateway.compute_driver(), diff --git a/crates/openshell-supervisor/README.md b/crates/openshell-supervisor/README.md index 813cd2b679..2c5c06cc65 100644 --- a/crates/openshell-supervisor/README.md +++ b/crates/openshell-supervisor/README.md @@ -117,7 +117,7 @@ Console JSON is installed independently of optional file appenders and uses an I ## OTLP agent trace relay -The `otlp_relay` module forwards the workload's OTLP/HTTP protobuf trace exports to the collector named by `OPENSHELL_OTLP_AGENT_ENDPOINT`, over the transport named by `OPENSHELL_OTLP_AGENT_PROTOCOL` (`grpc` when unset, or `http/protobuf`). The gateway sets those variables only when `[openshell.gateway.otlp.agent]` is enabled, so trace export alone leaves the relay inactive. Metrics and logs exports to the relay are acknowledged and discarded. The workload exports to the reserved address `192.0.0.8:4318`; the sandbox broker stages that connection to the supervisor, and the network proxy hands the stream to the relay's HTTP/1.1 receiver through the generic reserved-destination hook instead of evaluating policy. The receiver replaces the `openshell.sandbox.id` and `openshell.telemetry.source=agent` resource attributes, pushes the re-encoded batch into a bounded buffer, and a single export task forwards the encoded batches to the collector with a lazy connection. +The `otlp_relay` module forwards the workload's OTLP/HTTP protobuf trace exports to the collector named by `OPENSHELL_OTLP_AGENT_ENDPOINT`, over the transport named by `OPENSHELL_OTLP_AGENT_PROTOCOL` (`grpc` when unset, or `http/protobuf`). Each export carries the collector headers from the launch bundle's `agent_otlp_headers`; the supervisor's own span and OCSF exporters use `traces_otlp_headers` and `ocsf_otlp_headers` the same way. The gateway sets those variables only when `[openshell.gateway.otlp.agent]` is enabled, so trace export alone leaves the relay inactive. Metrics and logs exports to the relay are acknowledged and discarded. The workload exports to the reserved address `192.0.0.8:4318`; the sandbox broker stages that connection to the supervisor, and the network proxy hands the stream to the relay's HTTP/1.1 receiver through the generic reserved-destination hook instead of evaluating policy. The receiver replaces the `openshell.sandbox.id` and `openshell.telemetry.source=agent` resource attributes, pushes the re-encoded batch into a bounded buffer, and a single export task forwards the encoded batches to the collector with a lazy connection. Limits are fixed: 2 MiB per request body, at most 512 `ResourceSpans` entries, 256 KiB of encoded `Resource` fields, and 8192 resource attributes per request (each of these answers 413), a 10 second header-read timeout, 16 concurrent connections, 32 buffered batches, and a 5 second deadline per export attempt. Only the resources are decoded; spans are copied byte for byte, at most two requests at a time. Worst-case memory attributable to the agent is about 67 MiB of buffered batches, plus 32 MiB of bodies awaiting enrichment across the 16 connections, plus about 12 MiB for the two enrichments in progress, plus about 4 MiB for the batch being exported and its encoded copy. A full buffer answers 503 with `Retry-After`. Without an agent endpoint nothing starts and the proxy refuses the relay address. After the agent exits, `run_sandbox_with_backend` flushes buffered batches for up to 5 seconds before it reports the exit to the gateway, then logs a final summary of rejected and failed counts. Flushes skip the wait while the most recent export failed, and batches still pending when the relay is dropped without a flush count as failed. The supervisor tells the sandbox runtime through `AgentSpec::otlp_relay_active` whether the relay runs, and the runtime injects the generic `OTEL_EXPORTER_OTLP_*` defaults only then. diff --git a/crates/openshell-supervisor/src/backend_setup/tests.rs b/crates/openshell-supervisor/src/backend_setup/tests.rs index 4085adda2e..56ed0081d6 100644 --- a/crates/openshell-supervisor/src/backend_setup/tests.rs +++ b/crates/openshell-supervisor/src/backend_setup/tests.rs @@ -159,7 +159,9 @@ impl TestSetup { sandbox_token: SecretJwt::parse("test-sandbox-token").unwrap(), sandbox_expires_at: i64::MAX, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }, sandbox_id: "sandbox-1".to_string(), selected_name: TEST_BACKEND, diff --git a/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs b/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs index 9e2a5fec64..9fe46d58c6 100644 --- a/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs +++ b/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs @@ -271,7 +271,9 @@ async fn config_push_regression_legacy_gateway_never_observes_ready_before_workl sandbox_token: openshell_core::jwt::SecretJwt::parse("test-sandbox-token").unwrap(), sandbox_expires_at: 0, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }; let setup = HeldBackendSetup { session_id, diff --git a/crates/openshell-supervisor/src/lib.rs b/crates/openshell-supervisor/src/lib.rs index c96d7258d7..c9645d2b81 100644 --- a/crates/openshell-supervisor/src/lib.rs +++ b/crates/openshell-supervisor/src/lib.rs @@ -1298,8 +1298,12 @@ pub async fn run_sandbox_instance( // handler is installed when the first staged connection arrives, and // before attach so the agent spec can tell the boundary whether to point // workload SDKs at it. - let (mut otlp_relay, otlp_reserved_destination) = - otlp_relay::activate(otlp_agent_endpoint, otlp_agent_protocol, sandbox_id.clone()); + let (mut otlp_relay, otlp_reserved_destination) = otlp_relay::activate( + otlp_agent_endpoint, + otlp_agent_protocol, + auth_bundle.agent_otlp_headers.clone(), + sandbox_id.clone(), + ); // Construct the selected client only after shared policy and credential // setup has completed. Keep live credentials and later CA publication in @@ -4619,7 +4623,9 @@ mod tests { .expect("token"), sandbox_expires_at: 0, ssh_host_private_key: None, + traces_otlp_headers: None, ocsf_otlp_headers: None, + agent_otlp_headers: None, }; let config = SandboxInstanceConfig { command: vec!["true".to_string()], diff --git a/crates/openshell-supervisor/src/main.rs b/crates/openshell-supervisor/src/main.rs index a5f426f743..6f6d0797ed 100644 --- a/crates/openshell-supervisor/src/main.rs +++ b/crates/openshell-supervisor/src/main.rs @@ -20,6 +20,9 @@ use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::Subscriber mod logging; mod shared; +#[cfg(test)] +#[path = "../../../tests/support/environment.rs"] +mod test_environment; const DEBUG_RPC_SUBCOMMAND: &str = "debug-rpc"; const HEALTH_SUBCOMMAND: &str = "health"; @@ -497,21 +500,19 @@ fn main() -> Result<()> { ), }; // The exporters must be created inside the runtime that drives them. + let headers = bundle_otlp_headers(isolation_inputs.as_ref().map(|(_, _, auth, _, _)| auth)); let (otlp_provider, otlp_setup_error) = runtime.block_on(async { start_trace_export( args.otlp_endpoint.as_deref().map(export_config), args.otlp_protocol, + headers.traces, ) }); - let ocsf_headers = isolation_inputs - .as_ref() - .and_then(|(_, _, auth, _, _)| auth.ocsf_otlp_headers.as_ref()) - .map(openshell_core::jwt::SecretOtlpHeaders::expose_secret); let (ocsf_export, ocsf_setup_error) = runtime.block_on(async { start_ocsf_export( args.otlp_ocsf_endpoint.as_deref().map(export_config), args.otlp_ocsf_protocol, - ocsf_headers, + headers.ocsf, ) }); let otlp_layer_provider = otlp_provider.clone(); @@ -701,6 +702,27 @@ fn main() -> Result<()> { std::process::exit(exit_code?); } +/// The headers for the supervisor's own exports, from its launch bundle. +#[derive(Debug, Default, PartialEq)] +struct BundleOtlpHeaders<'a> { + traces: Option<&'a str>, + ocsf: Option<&'a str>, +} + +fn bundle_otlp_headers( + auth: Option<&openshell_core::jwt::SupervisorAuthBundle>, +) -> BundleOtlpHeaders<'_> { + fn expose(headers: &Option) -> Option<&str> { + headers + .as_ref() + .map(openshell_core::jwt::SecretOtlpHeaders::expose_secret) + } + auth.map_or_else(BundleOtlpHeaders::default, |auth| BundleOtlpHeaders { + traces: expose(&auth.traces_otlp_headers), + ocsf: expose(&auth.ocsf_otlp_headers), + }) +} + /// Start OCSF event export to an OTLP collector. /// /// A setup failure disables export and is returned for the caller to report @@ -716,9 +738,9 @@ fn start_ocsf_export( let Some(config) = config else { return (None, None); }; - let started = target(&config, protocol).and_then(|target| { + let started = target(&config, protocol, headers).and_then(|target| { Ok(openshell_otel::OcsfOtlp::start( - target.log_exporter_with_headers(headers)?, + target.log_exporter()?, target.resource().clone(), )) }); @@ -734,11 +756,12 @@ fn start_ocsf_export( fn start_trace_export( config: Option>, protocol: OtlpProtocol, + headers: Option<&str>, ) -> ( Option, Option, ) { - match config.map(|config| target(&config, protocol)?.tracer_provider()) { + match config.map(|config| target(&config, protocol, headers)?.tracer_provider()) { None => (None, None), Some(Ok(provider)) => (Some(provider), None), Some(Err(error)) => (None, Some(error)), @@ -748,8 +771,15 @@ fn start_trace_export( fn target( config: &openshell_otel::OtlpExportConfig<'_>, protocol: OtlpProtocol, + headers: Option<&str>, ) -> Result { - Ok(openshell_otel::OtlpTarget::new(config)?.with_protocol(protocol.into())) + Ok(openshell_otel::OtlpTarget::new(config)? + .with_protocol(protocol.into()) + .with_headers( + headers + .map(openshell_otel::parse_otlp_headers) + .unwrap_or_default(), + )) } /// Resource attributes identifying this supervisor's sandbox and the gateway @@ -1017,43 +1047,103 @@ mod tests { } #[tokio::test(flavor = "multi_thread")] - async fn supervisor_signals_export_over_their_configured_protocol() { - use openshell_ocsf::{ConfigStateChangeBuilder, ocsf_emit}; - use opentelemetry::trace::{Tracer as _, TracerProvider as _}; - use tracing_subscriber::prelude::*; - - let mut collector = openshell_otel_test_support::OtlpHttpTestServer::start().await; - let config = || openshell_otel::OtlpExportConfig { - endpoint: collector.endpoint(), - service_name: openshell_otel::ServiceName::Fixed(SERVICE_NAME), - service_version: None, - resource_attributes: Vec::new(), + async fn supervisor_signals_export_with_their_protocol_and_headers() { + 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 openshell_ocsf::{ConfigStateChangeBuilder, ocsf_emit}; + use opentelemetry::trace::{Tracer as _, TracerProvider as _}; + use tracing_subscriber::prelude::*; + + let mut collector = openshell_otel_test_support::OtlpHttpTestServer::start().await; + let config = || openshell_otel::OtlpExportConfig { + endpoint: collector.endpoint(), + service_name: openshell_otel::ServiceName::Fixed(SERVICE_NAME), + service_version: None, + resource_attributes: Vec::new(), + }; + let (provider, error) = start_trace_export( + Some(config()), + OtlpProtocol::HttpProtobuf, + Some("authorization=traces"), + ); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + let provider = provider.expect("trace export"); + let (ocsf, error) = start_ocsf_export( + Some(config()), + OtlpProtocol::HttpProtobuf, + Some("authorization=logs"), + ); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + let ocsf = ocsf.expect("OCSF export"); + + provider.tracer("test").in_span("supervisor-span", |_| {}); + let event = ConfigStateChangeBuilder::new(openshell_ocsf::ctx::ctx()) + .message("policy loaded") + .build(); + let subscriber = tracing_subscriber::registry().with(ocsf.layer()); + tracing::subscriber::with_default(subscriber, || ocsf_emit!(event)); + tokio::task::spawn_blocking(move || provider.force_flush()) + .await + .unwrap() + .expect("trace flush"); + ocsf.shutdown().await; + + let mut exports = [collector.next_export().await, collector.next_export().await] + .map(|export| { + let authorization = export + .headers + .get("authorization") + .map(|value| value.to_str().unwrap().to_string()); + (export.path, authorization) + }); + exports.sort(); + assert_eq!( + exports, + [ + ("/v1/logs".to_string(), Some("logs".to_string())), + ("/v1/traces".to_string(), Some("traces".to_string())), + ] + ); + }) + .await; + } + + #[test] + fn each_export_takes_its_own_headers_from_the_launch_bundle() { + use openshell_core::jwt::{ + CredentialEpoch, SecretJwt, SecretOtlpHeaders, SessionRotation, SupervisorAuthBundle, }; - let (provider, error) = start_trace_export(Some(config()), OtlpProtocol::HttpProtobuf); - assert!(error.is_none(), "unexpected setup error: {error:?}"); - let provider = provider.expect("trace export"); - let (ocsf, error) = start_ocsf_export(Some(config()), OtlpProtocol::HttpProtobuf, None); - assert!(error.is_none(), "unexpected setup error: {error:?}"); - let ocsf = ocsf.expect("OCSF export"); - - provider.tracer("test").in_span("supervisor-span", |_| {}); - let event = ConfigStateChangeBuilder::new(openshell_ocsf::ctx::ctx()) - .message("policy loaded") - .build(); - let subscriber = tracing_subscriber::registry().with(ocsf.layer()); - tracing::subscriber::with_default(subscriber, || ocsf_emit!(event)); - tokio::task::spawn_blocking(move || provider.force_flush()) - .await - .unwrap() - .expect("trace flush"); - ocsf.shutdown().await; - - let mut paths = vec![ - collector.next_export().await.path, - collector.next_export().await.path, - ]; - paths.sort(); - assert_eq!(paths, ["/v1/logs", "/v1/traces"]); + + let secret = |value: &str| Some(SecretOtlpHeaders::new(value.to_string())); + let bundle = SupervisorAuthBundle { + session_id: openshell_core::SandboxSessionId::new(), + runtime_generation: openshell_core::sandbox_generation::SandboxGenerationId::parse( + "generation-1", + ) + .unwrap(), + session_rotation: SessionRotation::new(1).unwrap(), + auth_epoch: CredentialEpoch::new(1).unwrap(), + gateway_token: SecretJwt::parse("test-gateway-token").unwrap(), + gateway_expires_at: 0, + sandbox_token: SecretJwt::parse("test-sandbox-token").unwrap(), + sandbox_expires_at: 0, + ssh_host_private_key: None, + traces_otlp_headers: secret("lane=traces"), + ocsf_otlp_headers: secret("lane=ocsf"), + agent_otlp_headers: secret("lane=agent"), + }; + + assert_eq!( + bundle_otlp_headers(Some(&bundle)), + BundleOtlpHeaders { + traces: Some("lane=traces"), + ocsf: Some("lane=ocsf"), + } + ); + assert_eq!(bundle_otlp_headers(None), BundleOtlpHeaders::default()); } #[tokio::test] diff --git a/crates/openshell-supervisor/src/otlp_relay/exporter.rs b/crates/openshell-supervisor/src/otlp_relay/exporter.rs index bc5b0e2d6c..531be544a3 100644 --- a/crates/openshell-supervisor/src/otlp_relay/exporter.rs +++ b/crates/openshell-supervisor/src/otlp_relay/exporter.rs @@ -100,41 +100,36 @@ enum Transport { pub struct Exporter { transport: Transport, endpoint: String, + headers: http::HeaderMap, } impl Exporter { - /// Prepares a lazy connection to `endpoint` over `protocol`. Fails only - /// when the endpoint is not a valid URI; no I/O happens here. - pub fn new(endpoint: &str, protocol: OtlpProtocol) -> Result { - match protocol { - OtlpProtocol::Grpc => Ok(Self::with_channel( + /// Prepares a lazy connection to `endpoint` over `protocol` that sends + /// `headers` with every export. Fails only when the endpoint is not a + /// valid URI; no I/O happens here. + pub fn new( + endpoint: &str, + protocol: OtlpProtocol, + headers: http::HeaderMap, + ) -> Result { + let transport = match protocol { + OtlpProtocol::Grpc => Transport::Grpc(Grpc::new( grpc_channel(endpoint).map_err(EndpointError::Grpc)?, - endpoint, )), - OtlpProtocol::HttpProtobuf => Ok(Self { - transport: Transport::Http { - client: Box::new( - OtlpHttpClient::new(EXPORT_TIMEOUT + CLIENT_TIMEOUT_MARGIN) - .map_err(EndpointError::Runtime)?, - ), - uri: openshell_otel::signal_url(endpoint, "v1/traces") - .map_err(EndpointError::Http)?, - }, - endpoint: endpoint.to_string(), - }), - } - } - - /// Forwards over a channel prepared elsewhere. This is the seam for a - /// shared collector transport (TLS, compression, headers) built once in - /// `openshell-otel` for the gateway, the supervisor's own spans, and this - /// relay; `endpoint` is only used for log messages. - #[must_use] - pub fn with_channel(channel: Channel, endpoint: &str) -> Self { - Self { - transport: Transport::Grpc(Grpc::new(channel)), + OtlpProtocol::HttpProtobuf => Transport::Http { + client: Box::new( + OtlpHttpClient::new(EXPORT_TIMEOUT + CLIENT_TIMEOUT_MARGIN) + .map_err(EndpointError::Runtime)?, + ), + uri: openshell_otel::signal_url(endpoint, "v1/traces") + .map_err(EndpointError::Http)?, + }, + }; + Ok(Self { + transport, endpoint: endpoint.to_string(), - } + headers, + }) } /// The collector URI this exporter targets. @@ -147,8 +142,10 @@ impl Exporter { pub async fn export(&self, batch: Bytes) -> Result<(), ExportError> { let send = async { match &self.transport { - Transport::Grpc(client) => export_grpc(client.clone(), batch).await, - Transport::Http { client, uri } => export_http(client, uri, batch).await, + Transport::Grpc(client) => export_grpc(client.clone(), &self.headers, batch).await, + Transport::Http { client, uri } => { + export_http(client, uri, &self.headers, batch).await + } } }; tokio::time::timeout(EXPORT_TIMEOUT, send) @@ -181,13 +178,18 @@ fn grpc_channel(endpoint: &str) -> Result { .connect_lazy()) } -async fn export_grpc(mut client: Grpc, batch: Bytes) -> Result<(), ExportError> { +async fn export_grpc( + mut client: Grpc, + headers: &http::HeaderMap, + batch: Bytes, +) -> Result<(), ExportError> { client.ready().await.map_err(|error| { ExportError::Status(Status::unavailable(format!( "collector channel not ready: {error}" ))) })?; let mut request = tonic::Request::new(batch); + *request.metadata_mut() = tonic::metadata::MetadataMap::from_headers(headers.clone()); request .extensions_mut() .insert(GrpcMethod::new(SERVICE, METHOD)); @@ -201,12 +203,17 @@ async fn export_grpc(mut client: Grpc, batch: Bytes) -> Result<(), Expo async fn export_http( client: &OtlpHttpClient, uri: &http::Uri, + headers: &http::HeaderMap, batch: Bytes, ) -> Result<(), ExportError> { - let request = http::Request::post(uri.clone()) - .header(http::header::CONTENT_TYPE, "application/x-protobuf") + let mut request = http::Request::post(uri.clone()) .body(batch) .map_err(|error| ExportError::Http(error.to_string()))?; + request.headers_mut().clone_from(headers); + request.headers_mut().insert( + http::header::CONTENT_TYPE, + http::HeaderValue::from_static("application/x-protobuf"), + ); let response = client .send(request) .await @@ -326,8 +333,15 @@ mod tests { #[test] fn new_rejects_an_invalid_uri() { - assert!(Exporter::new("not a uri", OtlpProtocol::Grpc).is_err()); - assert!(Exporter::new("not a uri", OtlpProtocol::HttpProtobuf).is_err()); + assert!(Exporter::new("not a uri", OtlpProtocol::Grpc, http::HeaderMap::new()).is_err()); + assert!( + Exporter::new( + "not a uri", + OtlpProtocol::HttpProtobuf, + http::HeaderMap::new() + ) + .is_err() + ); } #[tokio::test] @@ -339,7 +353,12 @@ mod tests { ] { let mut collector = OtlpHttpTestServer::start().await; let endpoint = format!("{}{suffix}", collector.endpoint()); - let exporter = Exporter::new(&endpoint, OtlpProtocol::HttpProtobuf).unwrap(); + let exporter = Exporter::new( + &endpoint, + OtlpProtocol::HttpProtobuf, + http::HeaderMap::new(), + ) + .unwrap(); let batch = Bytes::from(encoded_request("over-http", &[])); exporter @@ -361,7 +380,12 @@ mod tests { #[tokio::test] async fn http_export_fails_on_a_non_success_status() { let collector = OtlpHttpTestServer::start_with_status(http::StatusCode::FORBIDDEN).await; - let exporter = Exporter::new(collector.endpoint(), OtlpProtocol::HttpProtobuf).unwrap(); + let exporter = Exporter::new( + collector.endpoint(), + OtlpProtocol::HttpProtobuf, + http::HeaderMap::new(), + ) + .unwrap(); let error = exporter .export(Bytes::from(encoded_request("s", &[]))) .await @@ -377,7 +401,12 @@ mod tests { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let endpoint = format!("http://{}", listener.local_addr().unwrap()); drop(listener); - let exporter = Exporter::new(&endpoint, OtlpProtocol::HttpProtobuf).unwrap(); + let exporter = Exporter::new( + &endpoint, + OtlpProtocol::HttpProtobuf, + http::HeaderMap::new(), + ) + .unwrap(); let started = Instant::now(); let error = exporter @@ -402,7 +431,12 @@ mod tests { tokio::spawn(answer_once(stream)); } }); - let exporter = Exporter::new(&endpoint, OtlpProtocol::HttpProtobuf).unwrap(); + let exporter = Exporter::new( + &endpoint, + OtlpProtocol::HttpProtobuf, + http::HeaderMap::new(), + ) + .unwrap(); let started = Instant::now(); let error = exporter @@ -454,25 +488,40 @@ mod tests { } #[tokio::test] - async fn exports_over_a_channel_prepared_by_the_caller() { - let collector = OtlpTestServer::start().await; - let channel = Endpoint::from_shared(collector.endpoint().to_string()) + async fn exports_send_the_configured_headers_over_either_protocol() { + let headers = || openshell_otel::parse_otlp_headers("authorization=Bearer%20agent"); + let batch = || Bytes::from(encoded_request("with-headers", &[])); + + let grpc = OtlpTestServer::start().await; + Exporter::new(grpc.endpoint(), OtlpProtocol::Grpc, headers()) .unwrap() - .connect_lazy(); - let exporter = Exporter::with_channel(channel, collector.endpoint()); - exporter - .export(Bytes::from(encoded_request("via-channel", &[]))) + .export(batch()) .await - .expect("export over the supplied channel"); - collector.wait_for_export().await; - let received = collector.shutdown().await; - assert!(received.spans.iter().any(|span| span.name == "via-channel")); + .expect("gRPC export succeeds"); + grpc.wait_for_export().await; + assert_eq!(grpc.shutdown().await.trace_authorizations, ["Bearer agent"]); + + let mut http = OtlpHttpTestServer::start().await; + Exporter::new(http.endpoint(), OtlpProtocol::HttpProtobuf, headers()) + .unwrap() + .export(batch()) + .await + .expect("HTTP export succeeds"); + assert_eq!( + http.next_export().await.headers["authorization"], + "Bearer agent" + ); } #[tokio::test] async fn export_reaches_the_collector_with_attributes_intact() { let collector = OtlpTestServer::start().await; - let exporter = Exporter::new(collector.endpoint(), OtlpProtocol::Grpc).unwrap(); + let exporter = Exporter::new( + collector.endpoint(), + OtlpProtocol::Grpc, + http::HeaderMap::new(), + ) + .unwrap(); let batch = Bytes::from(encoded_request( "exported-span", &[("service.name", "unit-agent"), ("custom", "kept")], @@ -502,7 +551,8 @@ mod tests { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let endpoint = format!("http://{}", listener.local_addr().unwrap()); drop(listener); - let exporter = Exporter::new(&endpoint, OtlpProtocol::Grpc).unwrap(); + let exporter = + Exporter::new(&endpoint, OtlpProtocol::Grpc, http::HeaderMap::new()).unwrap(); let started = Instant::now(); let error = exporter @@ -521,7 +571,12 @@ mod tests { // error inside the export deadline rather than a 5 s timeout, and // the lazy channel is left closed for the next export to redial. let collector = SilentCollector::start().await; - let exporter = Exporter::new(&collector.endpoint, OtlpProtocol::Grpc).unwrap(); + let exporter = Exporter::new( + &collector.endpoint, + OtlpProtocol::Grpc, + http::HeaderMap::new(), + ) + .unwrap(); let started = Instant::now(); let error = exporter .export(Bytes::from(encoded_request("s", &[]))) diff --git a/crates/openshell-supervisor/src/otlp_relay/mod.rs b/crates/openshell-supervisor/src/otlp_relay/mod.rs index 281d23ac76..91cd4d4107 100644 --- a/crates/openshell-supervisor/src/otlp_relay/mod.rs +++ b/crates/openshell-supervisor/src/otlp_relay/mod.rs @@ -22,6 +22,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::{Duration, Instant}; +use openshell_core::jwt::SecretOtlpHeaders; use openshell_core::sandbox_env::OtlpProtocol; use openshell_isolation_interface::contract::BoundaryDuplexStream; use openshell_ocsf::{ConfigStateChangeBuilder, SeverityId, StateId, StatusId, ocsf_emit}; @@ -227,6 +228,8 @@ pub struct OtlpRelayConfig { pub endpoint: String, /// Transport to the collector (`OPENSHELL_OTLP_AGENT_PROTOCOL`). pub protocol: OtlpProtocol, + /// Headers sent with every export, from the launch bundle. + pub headers: Option, /// Sandbox identifier attached to every forwarded resource. pub sandbox_id: String, } @@ -327,7 +330,16 @@ pub fn start(config: OtlpRelayConfig) -> Result<(OtlpRelay, ReservedDestination) { return Err(StartError::UnsupportedScheme(scheme.to_string())); } - let exporter = Exporter::new(endpoint, config.protocol).map_err(StartError::Transport)?; + let exporter = Exporter::new( + endpoint, + config.protocol, + config + .headers + .as_ref() + .map(|headers| openshell_otel::parse_otlp_headers(headers.expose_secret())) + .unwrap_or_default(), + ) + .map_err(StartError::Transport)?; if config.sandbox_id.len() > enrichment::MAX_SANDBOX_ID_BYTES { // The attribution growth bound assumes a shorter id; batches near // the wire limit with many entries will be refused with 413. @@ -362,6 +374,7 @@ pub fn start(config: OtlpRelayConfig) -> Result<(OtlpRelay, ReservedDestination) pub fn activate( endpoint: Option, protocol: OtlpProtocol, + headers: Option, sandbox_id: Option, ) -> (Option, Option) { let Some(endpoint) = endpoint else { @@ -390,6 +403,7 @@ pub fn activate( match start(OtlpRelayConfig { endpoint: endpoint.clone(), protocol, + headers, sandbox_id, }) { Ok((relay, reserved)) => (Some(relay), Some(reserved)), @@ -727,7 +741,7 @@ mod tests { fn gated(endpoint: &str) -> (OtlpRelay, ReservedDestination, watch::Sender) { let (gate_tx, gate_rx) = watch::channel(false); - let exporter = Exporter::new(endpoint, OtlpProtocol::Grpc).unwrap(); + let exporter = Exporter::new(endpoint, OtlpProtocol::Grpc, http::HeaderMap::new()).unwrap(); let (relay, reserved) = assemble(exporter, "sb-test".into(), Some(gate_rx)); (relay, reserved, gate_tx) } @@ -749,11 +763,14 @@ mod tests { } #[tokio::test] - async fn relays_to_an_otlp_http_collector_when_configured() { + async fn relays_to_an_otlp_http_collector_with_its_headers() { let mut collector = OtlpHttpTestServer::start().await; let (_relay, reserved) = start(OtlpRelayConfig { endpoint: collector.endpoint().to_string(), protocol: OtlpProtocol::HttpProtobuf, + headers: Some(SecretOtlpHeaders::new( + "x-mlflow-experiment-id=7".to_string(), + )), sandbox_id: "sb-http".into(), }) .unwrap(); @@ -769,6 +786,7 @@ mod tests { let export = collector.next_export().await; assert_eq!(export.path, "/v1/traces"); + assert_eq!(export.headers["x-mlflow-experiment-id"], "7"); let decoded = opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest::decode( export.body, @@ -795,6 +813,7 @@ mod tests { let (relay, reserved) = start(OtlpRelayConfig { endpoint: collector.endpoint().to_string(), protocol: OtlpProtocol::Grpc, + headers: None, sandbox_id: "sb-test".into(), }) .unwrap(); @@ -864,6 +883,7 @@ mod tests { let error = start(OtlpRelayConfig { endpoint: "not a uri".into(), protocol: OtlpProtocol::Grpc, + headers: None, sandbox_id: "sb-test".into(), }) .err() @@ -873,6 +893,7 @@ mod tests { let error = start(OtlpRelayConfig { endpoint: blank.into(), protocol: OtlpProtocol::Grpc, + headers: None, sandbox_id: "sb-test".into(), }) .err() @@ -897,6 +918,7 @@ mod tests { let (relay, _reserved) = start(OtlpRelayConfig { endpoint: format!(" {}\n", collector.endpoint()), protocol: OtlpProtocol::Grpc, + headers: None, sandbox_id: "sb-test".into(), }) .expect("a padded endpoint starts the relay"); @@ -912,7 +934,7 @@ mod tests { let before = tokio::runtime::Handle::current() .metrics() .num_alive_tasks(); - let (relay, reserved) = activate(None, OtlpProtocol::Grpc, Some("sb-test".into())); + let (relay, reserved) = activate(None, OtlpProtocol::Grpc, None, Some("sb-test".into())); assert!(relay.is_none()); assert!(reserved.is_none()); let after = tokio::runtime::Handle::current() @@ -934,6 +956,7 @@ mod tests { Some(collector.endpoint().to_string()), OtlpProtocol::Grpc, None, + None, ); assert!(relay.is_none()); assert!(reserved.is_none()); @@ -954,6 +977,7 @@ mod tests { let error = start(OtlpRelayConfig { endpoint: "https://collector.example:4317".into(), protocol: OtlpProtocol::Grpc, + headers: None, sandbox_id: "sb-test".into(), }) .err() @@ -967,6 +991,7 @@ mod tests { start(OtlpRelayConfig { endpoint: "HTTP://collector.example:4317".into(), protocol: OtlpProtocol::Grpc, + headers: None, sandbox_id: "sb-test".into(), }) .is_ok() diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 20446834b9..8c326a134d 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -615,17 +615,36 @@ endpoint = "http://traces.observability.svc:4318" protocol = "http/protobuf" ``` -Each enabled signal needs an endpoint, either in its own table or from `endpoint`. A table without its own `endpoint` uses the whole `[openshell.gateway.otlp]` collector, including its `protocol`. A table with its own `endpoint` takes nothing else from `[openshell.gateway.otlp]`, so settings for one collector never reach another, and setting `protocol` in a table without an `endpoint` is a configuration error. Endpoints must be valid URIs without a query string: exporters send no query parameters, so an endpoint with one is a configuration error, and a collector that needs parameters must take them as headers. If an endpoint is malformed, the gateway logs the configuration error and continues with all OTLP export disabled. It does not connect at startup: an unreachable collector produces export failures, never a failure to serve. +Each enabled signal needs an endpoint, either in its own table or from `endpoint`. A table without its own `endpoint` uses the whole `[openshell.gateway.otlp]` collector, including its `protocol` and headers. A table with its own `endpoint` takes nothing else from `[openshell.gateway.otlp]`, so configured settings for one collector, such as its credentials, never reach another, and setting `protocol`, `headers`, or `headers_file` in a table without an `endpoint` is a configuration error. Endpoints must be valid URIs without a query string: exporters send no query parameters, so an endpoint with one is a configuration error, and a collector that needs parameters must take them as headers. If an endpoint is malformed, the gateway logs the configuration error and continues with all OTLP export disabled. It does not connect at startup: an unreachable collector produces export failures, never a failure to serve. `protocol` selects the transport to a collector: `grpc`, the default, or `http/protobuf`. With `http/protobuf`, the endpoint is a base URL and each signal is posted to its path under it, `/v1/traces` or `/v1/logs`, as with `OTEL_EXPORTER_OTLP_ENDPOINT`. HTTP/JSON is not supported. Point a `grpc` endpoint at a collector's gRPC receiver, conventionally port `4317`, and an `http/protobuf` endpoint at its HTTP receiver, conventionally port `4318`. A URI alone cannot distinguish the two, so a mismatched endpoint is accepted at startup and then fails on export. Sandbox supervisors and standalone compute drivers must come from the same release as the gateway to use `http/protobuf`: an older supervisor exports over gRPC, and an older driver rejects the protocol setting. +`headers` sets request headers sent with every export to a collector, as gRPC metadata or HTTP headers, such as an authorization token or a backend's routing header. Each name must be a valid HTTP header name set once, ignoring case; names the exporters set themselves, such as `content-type`, `host`, and `grpc-*`, and binary gRPC names ending in `-bin` are rejected. Each value must be non-empty printable ASCII; spaces are allowed. The gateway's trace and OCSF exporters also read `OTEL_EXPORTER_OTLP_TRACES_HEADERS` or `OTEL_EXPORTER_OTLP_LOGS_HEADERS`, or else `OTEL_EXPORTER_OTLP_HEADERS`, and a header set there replaces a configured header of the same name. In-process compute drivers and the VM driver send the gateway's trace headers, including those from the environment, with their own spans. A standalone compute driver sends only the headers in its own `OTEL_EXPORTER_OTLP_*HEADERS` variables. Supervisors send their trace collector's `headers` with their own spans, and the gateway's OCSF headers, including those from the environment, with their OCSF records. Trace headers from the gateway's environment do not reach supervisors, so set `headers` or `headers_file` on the trace collector when supervisor spans need them. The agent trace relay sends only its collector's `headers`, never ones from the environment, so its credentials can only be set with `headers` or `headers_file`. A changed header takes effect when the gateway restarts, and reaches a running sandbox only after that sandbox restarts. + +Put credentials in `headers_file` rather than `headers`. It reads a collector's headers from a file instead, such as a mounted Kubernetes Secret, so credentials stay out of the configuration file. The file holds one `name=value` header per line. Blank lines are skipped, and whitespace around each name and value is trimmed. The headers are validated like `headers`, the file must be UTF-8 and at most 64 KiB, and setting both `headers` and `headers_file` in one table is a configuration error. A relative path uses the gateway's working directory. The gateway reads the file at startup and fails to start if it is missing or invalid, so a changed file takes effect as a changed header does. + +Supervisors receive the gateway's headers but export to the endpoints their compute driver gives them. An external compute driver takes those endpoints from its own flags, so give it the same endpoints as the gateway, or supervisors send the gateway's credentials to a different collector. Supervisors and external compute drivers from an earlier release than the gateway reject launch credentials that carry trace or agent relay headers, so when the trace or agent collector sets `headers` or `headers_file`, upgrade them with or before the gateway. A gateway rolled back to an earlier release cannot read the stored launch credentials of sandboxes created with those headers, so delete and recreate those sandboxes after a rollback. + +Headers travel in cleartext unless the endpoint uses `https://`, which only `grpc` supports. The agent trace relay and `http/protobuf` collectors always receive them in cleartext, so do not send credentials to them over an untrusted network. + +```toml +[openshell.gateway.otlp] +endpoint = "http://otel-collector.observability.svc:4317" +headers_file = "/etc/openshell/otlp/headers" # authorization=Bearer + +[openshell.gateway.otlp.traces] +endpoint = "http://traces.observability.svc:4318" +protocol = "http/protobuf" +headers = { x-mlflow-experiment-id = "0" } +``` + Use an `https://` endpoint for TLS over `grpc`. The gateway verifies the collector certificate against the system trust roots. An `https://` endpoint with `http/protobuf` is a configuration error. The OpenTelemetry SDK logs export failures after startup. Spans in a failed batch are dropped rather than retried. `service_name` sets the gateway's `service.name` resource attribute and defaults to `openshell-gateway`. -The gateway passes the trace endpoint to compute drivers and sandbox supervisors, which export their own spans to it, so the trace collector must also be reachable from sandboxes. Spans for routine or high-volume operations use DEBUG level and are exported only when the gateway or sandbox log level is `debug`. For what is traced, span names, and resource attributes, refer to [Traces](/observability/traces). +The gateway passes the trace collector to compute drivers and sandbox supervisors, which export their own spans to it, so the trace collector must also be reachable from sandboxes. Spans for routine or high-volume operations use DEBUG level and are exported only when the gateway or sandbox log level is `debug`. For what is traced, span names, and resource attributes, refer to [Traces](/observability/traces). For Helm deployments, set `server.otlp.endpoint` to render this table. The optional `server.otlp.serviceName` value overrides the gateway service name; @@ -664,7 +683,7 @@ In-process compute drivers read their backend-specific settings from `[openshell ### OCSF Log Export -Set `enabled = true` in `[openshell.gateway.otlp.ocsf]` to export gateway OCSF events as OTLP log records. It defaults to `false`, so trace export alone never sends OCSF events. Records go to the table's `endpoint`, or to `[openshell.gateway.otlp]` `endpoint` when it is omitted, with the same resource attributes as spans, under the `openshell.ocsf` instrumentation scope. Use a separate endpoint to send OCSF events to a collector that feeds a SIEM while spans go elsewhere. Sandbox supervisors receive the same endpoint and protocol and export their own OCSF events there directly, so the endpoint must also be reachable from sandboxes. Supervisor records carry the supervisor's resource attributes, including `openshell.sandbox.id` and `openshell.gateway.name`. Supervisors also send the collector headers that the gateway uses for OCSF export, from `OTEL_EXPORTER_OTLP_LOGS_HEADERS` or else `OTEL_EXPORTER_OTLP_HEADERS`, so a collector that requires authentication accepts their records too. The gateway delivers the headers with each sandbox's launch credentials, never through the sandbox environment. Other `OTEL_*` variables do not reach supervisors. +Set `enabled = true` in `[openshell.gateway.otlp.ocsf]` to export gateway OCSF events as OTLP log records. It defaults to `false`, so trace export alone never sends OCSF events. Records go to the table's `endpoint`, or to `[openshell.gateway.otlp]` `endpoint` when it is omitted, with the same resource attributes as spans, under the `openshell.ocsf` instrumentation scope. Use a separate endpoint to send OCSF events to a collector that feeds a SIEM while spans go elsewhere. Sandbox supervisors receive the same endpoint and protocol and export their own OCSF events there directly, so the endpoint must also be reachable from sandboxes. Supervisor records carry the supervisor's resource attributes, including `openshell.sandbox.id` and `openshell.gateway.name`. Supervisors also send the collector [headers](#otlp-export) that the gateway uses for OCSF export, so a collector that requires authentication accepts their records too. The gateway delivers headers with each sandbox's launch credentials, never through the sandbox environment. Other `OTEL_*` variables do not reach supervisors. ```toml [openshell.gateway.otlp.ocsf] @@ -693,12 +712,14 @@ Metrics include `openshell_ocsf_otlp_queued_total`, `openshell_ocsf_otlp_exporte ### Agent trace relay -Set `enabled = true` in `[openshell.gateway.otlp.agent]` to have each sandbox forward the workload's own traces. It defaults to `false`, so trace export alone never changes the workload environment. Traces go to the table's collector, or to the `[openshell.gateway.otlp]` collector when it has no `endpoint`, and that collector must be reachable from sandboxes. Use `protocol = "http/protobuf"` for trace backends that accept only OTLP/HTTP. +Set `enabled = true` in `[openshell.gateway.otlp.agent]` to have each sandbox forward the workload's own traces. It defaults to `false`, so trace export alone never changes the workload environment. Traces go to the table's collector, or to the `[openshell.gateway.otlp]` collector when it has no `endpoint`, and that collector must be reachable from sandboxes. Use `protocol = "http/protobuf"` for trace backends that accept only OTLP/HTTP, and `headers` for the ones that need them. For example, MLflow accepts traces directly when each request names an experiment: ```toml [openshell.gateway.otlp.agent] enabled = true -endpoint = "http://agent-collector.observability.svc:4317" # optional +endpoint = "http://mlflow.mlops.svc:5000" +protocol = "http/protobuf" +headers = { x-mlflow-experiment-id = "0" } ``` While the relay is active, the sandbox runtime sets `OTEL_EXPORTER_OTLP_ENDPOINT=http://192.0.0.8:4318` and `OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf` in every workload process, and sets `OTEL_EXPORTER_OTLP_COMPRESSION` and `OTEL_EXPORTER_OTLP_TRACES_COMPRESSION` to `none`, so a standard OpenTelemetry SDK exports without configuration. These are lowest-precedence defaults. They are not set when the sandbox creation request or an environment passed to `openshell sandbox exec` names `OTEL_EXPORTER_OTLP_ENDPOINT` or `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, or when the image `ENV` names one for the main process, since `openshell sandbox exec` sessions do not inherit the image environment, and an endpoint the workload sets itself, for example with a shell `export`, takes precedence over them. A workload that exports elsewhere does so subject to the sandbox's network policy, and its traces carry no relay attribution. Provider credentials cannot redirect traces. While the relay is off, nothing is injected. Because the generic endpoint is set, an SDK that also exports metrics or logs sends them to the relay, which answers with an empty OTLP success and discards them. To keep metrics and logs, point those exporters elsewhere with `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` and `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`. The address is a reserved label rather than a network destination: the supervisor serves it inside the sandbox boundary and forwards accepted batches to the collector over the configured `protocol`. The collector's address is never visible inside the sandbox. @@ -709,11 +730,11 @@ Agents must export traces as OTLP/HTTP protobuf with compression disabled. The r Delivery is best-effort. The relay answers 200 once a batch is buffered and makes one export attempt per batch, so a batch is lost if the collector is unavailable at that moment. The supervisor logs the first failed export and a periodic summary of failures. Before the sandbox reports that its main process exited, and before a suspended sandbox's supervisor exits, the relay flushes buffered batches for up to 5 seconds, and skips that wait when its most recent export failed. It keeps serving `openshell sandbox exec` sessions after the main process exits and flushes once more when the sandbox is torn down. -The relayed hop does not yet support TLS or authentication to the collector; an `https` endpoint leaves the relay inactive with a logged reason. Protect the hop at the network layer, for example with a collector-side ingress policy. +The relayed hop does not yet support TLS to the collector, so headers, including credentials, travel in cleartext; an `https` endpoint leaves the relay inactive with a logged reason. Protect the hop at the network layer, for example with a collector-side ingress policy. ### Tuning -`[openshell.gateway.otlp]` decides whether and where to export. How the SDK exports is controlled by the standard OpenTelemetry environment variables, which the gateway reads through the SDK rather than mirroring as TOML keys. They apply to the gateway and its in-process compute drivers. The gateway does not pass them to sandbox supervisors, which receive only the endpoints, protocols, and the [OCSF collector headers](#ocsf-log-export). +`[openshell.gateway.otlp]` decides whether and where to export. How the SDK exports is controlled by the standard OpenTelemetry environment variables, which the gateway reads through the SDK rather than mirroring as TOML keys. They apply to the gateway and its in-process compute drivers. The VM driver also receives the trace headers, and sandbox supervisors receive headers as described for [headers](#otlp-export). No other variable reaches supervisors, which receive only the endpoints, protocols, and headers of their collectors. The gateway honors these variables: diff --git a/docs/observability/traces.mdx b/docs/observability/traces.mdx index 87854c019c..ec2277b77f 100644 --- a/docs/observability/traces.mdx +++ b/docs/observability/traces.mdx @@ -19,7 +19,7 @@ Set an OTLP collector endpoint in `gateway.toml`: endpoint = "https://backend:4317" ``` -In the Helm chart, set the table under `gatewayConfig`. The gateway passes the endpoint to the compute drivers and sandbox supervisors it starts, and supervisors connect to the collector directly, so it must be reachable from sandboxes too. For a separate trace collector, OTLP/HTTP, TLS, sampling, and the other settings, refer to [OTLP Export](/how-it-works/gateways/configuration#otlp-export). +In the Helm chart, set the table under `gatewayConfig`. The gateway passes the endpoint to the compute drivers and sandbox supervisors it starts, and supervisors connect to the collector directly, so it must be reachable from sandboxes too. For a separate trace collector, OTLP/HTTP, headers, TLS, sampling, and the other settings, refer to [OTLP Export](/how-it-works/gateways/configuration#otlp-export). ## What Is Traced