diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 56fb9e744..0420143c5 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -25,7 +25,7 @@ use std::thread; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use super::otel_signal::{ - MetricMarkClassification, SignalRuntimeDiagnostics, classify_metric_mark, + MetricMarkClassification, SignalRuntimeDiagnostics, classify_metric_mark, resolve_header_env, should_relog_runtime_diagnostic, }; use super::{ @@ -237,6 +237,7 @@ pub struct OpenTelemetryConfig { otel_type: OpenTelemetryType, endpoint: String, headers: HashMap, + header_env: HashMap, resource_attributes: HashMap, service_name: String, service_namespace: Option, @@ -260,6 +261,7 @@ impl OpenTelemetryConfig { otel_type: OpenTelemetryType::Full, endpoint: String::new(), headers: HashMap::new(), + header_env: HashMap::new(), resource_attributes: HashMap::new(), service_name: "unknown_service".to_string(), service_namespace: None, @@ -331,6 +333,12 @@ impl OpenTelemetryConfig { self } + /// Maps an exporter header name to the environment variable supplying its value. + pub fn with_header_env(mut self, key: impl Into, variable: impl Into) -> Self { + self.header_env.insert(key.into(), variable.into()); + self + } + #[cfg(test)] pub(crate) fn header(&self, key: &str) -> Option<&str> { self.headers.get(key).map(String::as_str) @@ -540,7 +548,7 @@ impl OpenTelemetrySubscriber { } fn new_with_runtime_diagnostics( - config: OpenTelemetryConfig, + mut config: OpenTelemetryConfig, diagnostic_field: Option, ) -> Result { if config.endpoint.trim().is_empty() { @@ -559,6 +567,8 @@ impl OpenTelemetrySubscriber { .map_err(OpenTelemetryError::InvalidMetadataPromotionPrefixes)?; reject_global_header_environment()?; validate_headers(&config.headers)?; + config.headers = resolve_header_env(&config.headers, &config.header_env)?; + validate_headers(&config.headers)?; let runtime_diagnostics = SignalRuntimeDiagnostics::new(diagnostic_field); let (provider, runtime) = build_owned_tracer_provider(config.clone(), runtime_diagnostics.clone())?; diff --git a/crates/core/src/observability/otel_logs.rs b/crates/core/src/observability/otel_logs.rs index c851fcc08..20e8ad53b 100644 --- a/crates/core/src/observability/otel_logs.rs +++ b/crates/core/src/observability/otel_logs.rs @@ -38,8 +38,8 @@ use super::otel::{ use super::otel_signal::{ MetricMarkClassification, SignalExporterRuntime, SignalRuntimeDiagnostics, build_grpc_metadata, build_in_owned_runtime, classify_metric_mark, reject_signal_header_environment, - resolve_http_signal_endpoint, should_relog_runtime_diagnostic, signal_resource, - validate_signal_headers, + resolve_header_env, resolve_http_signal_endpoint, should_relog_runtime_diagnostic, + signal_resource, validate_signal_headers, }; const DEFAULT_MAX_QUEUE_SIZE: usize = 2_048; @@ -51,6 +51,7 @@ const DEFAULT_SCHEDULED_DELAY: Duration = Duration::from_secs(1); pub struct OpenTelemetryLogConfig { endpoint: String, headers: HashMap, + header_env: HashMap, resource_attributes: HashMap, service_name: String, service_namespace: Option, @@ -72,6 +73,7 @@ impl OpenTelemetryLogConfig { Self { endpoint: endpoint.into(), headers: HashMap::new(), + header_env: HashMap::new(), resource_attributes: HashMap::new(), service_name: "unknown_service".to_string(), service_namespace: None, @@ -100,6 +102,12 @@ impl OpenTelemetryLogConfig { self } + /// Map an exporter header name to the environment variable supplying its value. + pub fn with_header_env(mut self, key: impl Into, variable: impl Into) -> Self { + self.header_env.insert(key.into(), variable.into()); + self + } + /// Add an OpenTelemetry resource attribute. pub fn with_resource_attribute( mut self, @@ -247,8 +255,10 @@ impl OpenTelemetryLogSubscriber { Self::new_with_runtime_diagnostics(config) } - fn new_with_runtime_diagnostics(config: OpenTelemetryLogConfig) -> Result { + fn new_with_runtime_diagnostics(mut config: OpenTelemetryLogConfig) -> Result { config.validate()?; + config.headers = resolve_header_env(&config.headers, &config.header_env)?; + validate_signal_headers(&config.headers)?; let minimum_severity = config.minimum_severity; let completed_span_context_ttl = config.completed_span_context_ttl; let instrumentation_scope = config.instrumentation_scope.clone(); diff --git a/crates/core/src/observability/otel_metrics.rs b/crates/core/src/observability/otel_metrics.rs index 36632ecb5..28f4a8746 100644 --- a/crates/core/src/observability/otel_metrics.rs +++ b/crates/core/src/observability/otel_metrics.rs @@ -37,8 +37,8 @@ use super::otel::{OpenTelemetryError, OtlpTransport, Result, normalize_shutdown_ use super::otel_signal::{ MetricMarkClassification, SignalExporterRuntime, SignalRuntimeDiagnostics, build_grpc_metadata, build_in_owned_runtime, classify_metric_mark, reject_signal_header_environment, - resolve_http_signal_endpoint, should_relog_runtime_diagnostic, signal_resource, - validate_signal_headers, + resolve_header_env, resolve_http_signal_endpoint, should_relog_runtime_diagnostic, + signal_resource, validate_signal_headers, }; const DEFAULT_EXPORT_INTERVAL: Duration = Duration::from_secs(60); @@ -97,6 +97,7 @@ impl std::str::FromStr for MetricTemporality { pub struct OpenTelemetryMetricConfig { endpoint: String, headers: HashMap, + header_env: HashMap, resource_attributes: HashMap, service_name: String, service_namespace: Option, @@ -117,6 +118,7 @@ impl OpenTelemetryMetricConfig { Self { endpoint: endpoint.into(), headers: HashMap::new(), + header_env: HashMap::new(), resource_attributes: HashMap::new(), service_name: "unknown_service".to_string(), service_namespace: None, @@ -144,6 +146,12 @@ impl OpenTelemetryMetricConfig { self } + /// Map an exporter header name to the environment variable supplying its value. + pub fn with_header_env(mut self, key: impl Into, variable: impl Into) -> Self { + self.header_env.insert(key.into(), variable.into()); + self + } + /// Add an OpenTelemetry resource attribute. pub fn with_resource_attribute( mut self, @@ -282,8 +290,10 @@ impl OpenTelemetryMetricSubscriber { Self::new_with_runtime_diagnostics(config) } - fn new_with_runtime_diagnostics(config: OpenTelemetryMetricConfig) -> Result { + fn new_with_runtime_diagnostics(mut config: OpenTelemetryMetricConfig) -> Result { config.validate()?; + config.headers = resolve_header_env(&config.headers, &config.header_env)?; + validate_signal_headers(&config.headers)?; let instrumentation_scope = config.instrumentation_scope.clone(); let max_instruments = config.max_instruments; let cardinality_limit = config.cardinality_limit; diff --git a/crates/core/src/observability/otel_signal.rs b/crates/core/src/observability/otel_signal.rs index 011acf3fa..bec44c9e2 100644 --- a/crates/core/src/observability/otel_signal.rs +++ b/crates/core/src/observability/otel_signal.rs @@ -315,6 +315,80 @@ pub(super) fn validate_signal_headers(headers: &HashMap) -> Resu Ok(()) } +pub(super) fn resolve_header_env( + headers: &HashMap, + header_env: &HashMap, +) -> Result> { + let mut normalized = HashSet::new(); + for key in headers.keys() { + normalized.insert(key.to_ascii_lowercase()); + } + + for (key, variable) in header_env { + if !normalized.insert(key.to_ascii_lowercase()) { + return Err(OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: + "header names must be unique across headers and header_env ignoring ASCII case" + .to_string(), + }); + } + reqwest::header::HeaderName::from_bytes(key.as_bytes()).map_err(|error| { + OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: error.to_string(), + } + })?; + if variable.trim().is_empty() + || variable.trim() != variable + || variable.contains(['\0', '=']) + { + return Err(OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: "header_env must name a nonblank environment variable without surrounding whitespace, '=' or NUL" + .to_string(), + }); + } + } + + let mut resolved = headers.clone(); + for (key, variable) in header_env { + let value = match std::env::var(variable) { + Ok(value) => value, + Err(std::env::VarError::NotPresent) => { + return Err(OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: format!("environment variable {variable:?} is not set"), + }); + } + Err(std::env::VarError::NotUnicode(_)) => { + return Err(OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: format!("environment variable {variable:?} is not valid Unicode"), + }); + } + }; + if value.trim().is_empty() || value.trim() != value { + return Err(OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: format!( + "environment variable {variable:?} must contain a nonblank value without surrounding whitespace" + ), + }); + } + reqwest::header::HeaderValue::from_str(&value).map_err(|_| { + OpenTelemetryError::InvalidHeader { + key: key.clone(), + message: format!( + "environment variable {variable:?} does not contain a valid header value" + ), + } + })?; + resolved.insert(key.clone(), value); + } + Ok(resolved) +} + pub(super) fn reject_signal_header_environment(signal_variable: &'static str) -> Result<()> { for variable in ["OTEL_EXPORTER_OTLP_HEADERS", signal_variable] { if std::env::var_os(variable).is_some_and(|value| !value.is_empty()) { diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 03d0410f1..de57844e3 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -2199,9 +2199,9 @@ fn resolve_signal_headers( "OpenTelemetry {signal}.endpoints[{index}] header {key:?} cannot appear in both headers and header_env" ))); } - let value = std::env::var(variable).map_err(|error| { + let value = std::env::var(variable).map_err(|_| { PluginError::InvalidConfig(format!( - "OpenTelemetry {signal}.endpoints[{index}].header_env.{key} could not read environment variable {variable:?}: {error}" + "OpenTelemetry {signal}.endpoints[{index}].header_env.{key} could not read environment variable {variable:?}" )) })?; if value.trim().is_empty() || value.trim() != value { @@ -3269,9 +3269,9 @@ fn apply_otel_environment_headers( header_env: HashMap, ) -> PluginResult { for (key, variable) in header_env { - let value = std::env::var(&variable).map_err(|error| { + let value = std::env::var(&variable).map_err(|_| { PluginError::InvalidConfig(format!( - "OpenTelemetry endpoints[{index}].header_env.{key} could not read environment variable {variable:?}: {error}" + "OpenTelemetry endpoints[{index}].header_env.{key} could not read environment variable {variable:?}" )) })?; if value.trim().is_empty() || value.trim() != value { diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index b108143d6..11f646d45 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -1202,6 +1202,7 @@ fn make_mark_event_with_metadata(parent_uuid: Option, metadata: Json) -> E struct CapturedHttpRequest { path: String, content_type: String, + authorization: Option, body: Vec, } @@ -1227,6 +1228,7 @@ fn read_http_request(stream: &mut impl Read) -> CapturedHttpRequest { CapturedHttpRequest { path: request_line.split_whitespace().nth(1).unwrap().to_string(), content_type: header_value(&headers_text, "content-type").unwrap_or_default(), + authorization: header_value(&headers_text, "authorization"), body: bytes[header_end..header_end + content_length].to_vec(), } } @@ -1283,6 +1285,7 @@ fn config_defaults_and_builder_overrides_are_applied() { OpenTelemetryConfig::new(OpenTelemetryType::Full, "http://localhost:4318/v1/traces") .with_service_name("demo-agent") .with_header("authorization", "Bearer token") + .with_header_env("x-api-key", "NEMO_RELAY_TEST_API_KEY") .with_resource_attribute("deployment.environment", "test") .with_service_namespace("agents") .with_service_version("1.2.3") @@ -1302,6 +1305,10 @@ fn assert_config_builder_overrides(config: &OpenTelemetryConfig) { config.headers.get("authorization"), Some(&"Bearer token".into()) ); + assert_eq!( + config.header_env.get("x-api-key"), + Some(&"NEMO_RELAY_TEST_API_KEY".into()) + ); assert_eq!( config.resource_attributes.get("deployment.environment"), Some(&"test".into()) @@ -1525,6 +1532,109 @@ fn direct_config_rejects_invalid_and_case_duplicate_headers() { } } +#[test] +fn direct_config_rejects_invalid_header_env_without_exposing_values() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + let missing = format!("NEMO_RELAY_TEST_MISSING_HEADER_{}", Uuid::now_v7().simple()); + + for (variable, value, expected) in [ + (missing.as_str(), None, "is not set"), + ("NEMO_RELAY_TEST_BLANK_HEADER", Some(" "), "nonblank value"), + ( + "NEMO_RELAY_TEST_SECRET_HEADER", + Some("relay-secret\ninvalid"), + "valid header value", + ), + ] { + if let Some(value) = value { + unsafe { std::env::set_var(variable, value) }; + } + let error = match OpenTelemetrySubscriber::new( + OpenTelemetryConfig::new(OpenTelemetryType::Full, "http://localhost:4318/v1/traces") + .with_header_env("authorization", variable), + ) { + Ok(_) => panic!("invalid header_env should fail activation"), + Err(error) => error, + }; + let message = error.to_string(); + assert!(message.contains(expected), "unexpected error: {message}"); + assert!(!message.contains("relay-secret")); + unsafe { std::env::remove_var(variable) }; + } + + for variable in ["", " padded ", "INVALID=NAME", "INVALID\0NAME"] { + let error = match OpenTelemetrySubscriber::new( + OpenTelemetryConfig::new(OpenTelemetryType::Full, "http://localhost:4318/v1/traces") + .with_header_env("authorization", variable), + ) { + Ok(_) => panic!("invalid environment variable reference should fail activation"), + Err(error) => error, + }; + assert!(error.to_string().contains("header_env must name")); + } + + let error = match OpenTelemetrySubscriber::new( + OpenTelemetryConfig::new(OpenTelemetryType::Full, "http://localhost:4318/v1/traces") + .with_header("Authorization", "static") + .with_header_env("authorization", &missing), + ) { + Ok(_) => panic!("case-insensitive duplicate should fail before environment lookup"), + Err(error) => error, + }; + assert!( + error + .to_string() + .contains("unique across headers and header_env") + ); + + unsafe { std::env::set_var(&missing, "valid") }; + let exact_error = match OpenTelemetrySubscriber::new( + OpenTelemetryConfig::new(OpenTelemetryType::Full, "http://localhost:4318/v1/traces") + .with_header("authorization", "static") + .with_header_env("authorization", &missing), + ) { + Ok(_) => panic!("exact duplicate should fail activation"), + Err(error) => error, + }; + assert!( + exact_error + .to_string() + .contains("unique across headers and header_env") + ); + unsafe { std::env::remove_var(&missing) }; +} + +#[cfg(unix)] +#[test] +fn direct_config_non_unicode_header_env_errors_do_not_expose_values() { + use std::os::unix::ffi::OsStringExt; + + let _guard = crate::observability::test_mutex() + .lock() + .unwrap_or_else(|error| error.into_inner()); + let variable = format!( + "NEMO_RELAY_TEST_NON_UNICODE_DIRECT_HEADER_{}", + Uuid::now_v7().simple() + ); + let secret = "relay-direct-secret"; + let mut value = vec![0xff]; + value.extend_from_slice(secret.as_bytes()); + unsafe { std::env::set_var(&variable, std::ffi::OsString::from_vec(value)) }; + + let error = match OpenTelemetrySubscriber::new( + OpenTelemetryConfig::new(OpenTelemetryType::Full, "http://localhost:4318/v1/traces") + .with_header_env("authorization", &variable), + ) { + Ok(_) => panic!("non-Unicode header environment value should fail activation"), + Err(error) => error, + }; + let message = error.to_string(); + assert!(message.contains(&variable)); + assert!(!message.contains(secret)); + + unsafe { std::env::remove_var(variable) }; +} + #[test] fn direct_config_rejects_process_global_otel_headers() { const CHILD_MARKER: &str = "NEMO_RELAY_TEST_GLOBAL_OTEL_HEADER_CHILD"; @@ -3165,8 +3275,17 @@ fn http_config_exports_scope_push_pop_and_marks_without_tokio_runtime() { let (request_tx, request_rx) = mpsc::channel(); spawn_http_collector(listener, request_tx); - let config = OpenTelemetryConfig::http_binary("demo-agent").with_endpoint(endpoint); + let variable = format!( + "NEMO_RELAY_TEST_HEADER_SNAPSHOT_{}", + Uuid::now_v7().simple() + ); + let secret = "Bearer activation-secret"; + unsafe { std::env::set_var(&variable, secret) }; + let config = OpenTelemetryConfig::http_binary("demo-agent") + .with_endpoint(endpoint) + .with_header_env("authorization", &variable); let subscriber = OpenTelemetrySubscriber::new(config).unwrap(); + unsafe { std::env::set_var(&variable, "Bearer changed-secret") }; let name = format!("otel_http_{}", Uuid::now_v7().simple()); subscriber.register(&name).unwrap(); @@ -3204,7 +3323,22 @@ fn http_config_exports_scope_push_pop_and_marks_without_tokio_runtime() { .expect("expected an OTLP request"); assert_eq!(request.path, "/v1/traces"); assert_eq!(request.content_type, "application/x-protobuf"); + assert_eq!(request.authorization.as_deref(), Some(secret)); assert!(!request.body.is_empty()); + assert!( + !request + .body + .windows(secret.len()) + .any(|window| window == secret.as_bytes()) + ); + assert!( + subscriber + .runtime_diagnostics() + .entries() + .iter() + .all(|diagnostic| !diagnostic.message.contains(secret)) + ); + unsafe { std::env::remove_var(variable) }; } #[test] diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index f5baea3cb..7989d27b9 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -1516,6 +1516,44 @@ fn opentelemetry_endpoint_header_env_rejects_missing_and_duplicate_headers() { clear_plugin_configuration().unwrap(); } +#[cfg(unix)] +#[test] +fn opentelemetry_header_env_non_unicode_errors_do_not_expose_values() { + use std::os::unix::ffi::OsStringExt; + + let _guard = crate::observability::test_mutex() + .lock() + .unwrap_or_else(|error| error.into_inner()); + let variable = "NEMO_RELAY_TEST_NON_UNICODE_OTEL_HEADER_ENV"; + let secret = "relay-plugin-secret"; + let mut value = vec![0xff]; + value.extend_from_slice(secret.as_bytes()); + unsafe { std::env::set_var(variable, std::ffi::OsString::from_vec(value)) }; + + let trace: OpenTelemetryEndpointConfig = serde_json::from_value(json!({ + "type": "full", + "endpoint": "http://localhost:4318/v1/traces", + "header_env": {"authorization": variable} + })) + .unwrap(); + let trace_error = build_otel_config(0, trace).unwrap_err().to_string(); + assert!(trace_error.contains(variable)); + assert!(!trace_error.contains(secret)); + + let logs: OpenTelemetrySignalEndpointConfig = serde_json::from_value(json!({ + "endpoint": "http://localhost:4318/v1/logs", + "header_env": {"authorization": variable} + })) + .unwrap(); + let log_error = resolve_signal_headers("logs", 0, &logs) + .unwrap_err() + .to_string(); + assert!(log_error.contains(variable)); + assert!(!log_error.contains(secret)); + + unsafe { std::env::remove_var(variable) }; +} + #[test] fn invalid_log_endpoint_keeps_valid_signal_subscriber() { let _guard = crate::observability::test_mutex() diff --git a/crates/ffi/nemo_relay.h b/crates/ffi/nemo_relay.h index c5b83dc1f..ae6c44102 100644 --- a/crates/ffi/nemo_relay.h +++ b/crates/ffi/nemo_relay.h @@ -1781,6 +1781,33 @@ NemoRelayStatus nemo_relay_otel_subscriber_create_with_projection_options(const * # Safety * Any non-null C strings must be valid and `out` must be non-null. */ +NemoRelayStatus nemo_relay_otel_subscriber_create_with_projection_options_v4(const char *otel_type, + const char *transport, + const char *endpoint, + const char *headers_json, + const char *header_env_json, + const char *resource_attributes_json, + const char *service_name, + const char *service_namespace, + const char *service_version, + const char *instrumentation_scope, + uint64_t timeout_millis, + const char *mark_projection, + const char *mark_exclude_names_json, + const char *attribute_mappings_json, + const char *promote_metadata_prefixes_json, + uint64_t completed_span_context_ttl_millis, + struct FfiOpenTelemetrySubscriber **out); + +/** + * Creates one typed OpenTelemetry exporter subscriber with projection, metadata, and lineage controls. + * + * This compatibility entrypoint configures only static headers. Use + * `nemo_relay_otel_subscriber_create_with_projection_options_v4` for `header_env`. + * + * # Safety + * Any non-null C strings must be valid and `out` must be non-null. + */ NemoRelayStatus nemo_relay_otel_subscriber_create_with_projection_options_v3(const char *otel_type, const char *transport, const char *endpoint, @@ -1893,6 +1920,29 @@ NemoRelayStatus nemo_relay_otel_log_subscriber_create(const char *transport, uint64_t completed_span_context_ttl_millis, struct FfiOpenTelemetryLogSubscriber **out); +/** + * Creates an independently managed OpenTelemetry log subscriber with `header_env` support. + * + * # Safety + * Any non-null C strings must be valid and `out` must be non-null. + */ +NemoRelayStatus nemo_relay_otel_log_subscriber_create_v2(const char *transport, + const char *endpoint, + const char *headers_json, + const char *header_env_json, + const char *resource_attributes_json, + const char *service_name, + const char *service_namespace, + const char *service_version, + const char *instrumentation_scope, + uint64_t timeout_millis, + const char *minimum_severity, + uint64_t max_queue_size, + uint64_t max_export_batch_size, + uint64_t scheduled_delay_millis, + uint64_t completed_span_context_ttl_millis, + struct FfiOpenTelemetryLogSubscriber **out); + /** * Registers an OpenTelemetry log subscriber globally. * @@ -1965,6 +2015,28 @@ NemoRelayStatus nemo_relay_otel_metric_subscriber_create(const char *transport, uint64_t cardinality_limit, struct FfiOpenTelemetryMetricSubscriber **out); +/** + * Creates an independently managed OpenTelemetry metric subscriber with `header_env` support. + * + * # Safety + * Any non-null C strings must be valid and `out` must be non-null. + */ +NemoRelayStatus nemo_relay_otel_metric_subscriber_create_v2(const char *transport, + const char *endpoint, + const char *headers_json, + const char *header_env_json, + const char *resource_attributes_json, + const char *service_name, + const char *service_namespace, + const char *service_version, + const char *instrumentation_scope, + uint64_t timeout_millis, + uint64_t export_interval_millis, + const char *temporality, + uint64_t max_instruments, + uint64_t cardinality_limit, + struct FfiOpenTelemetryMetricSubscriber **out); + /** * Registers an OpenTelemetry metric subscriber globally. * @@ -3313,10 +3385,11 @@ void nemo_relay_atof_exporter_free(struct FfiAtofExporter *ptr); /** * Free an OpenTelemetry subscriber handle previously returned by - * `nemo_relay_otel_subscriber_create`. + * an OpenTelemetry subscriber constructor. * * # Safety - * `ptr` must be a valid pointer returned by `nemo_relay_otel_subscriber_create`, or null. + * `ptr` must be a valid pointer returned by `nemo_relay_otel_subscriber_create` or + * `nemo_relay_otel_subscriber_create_with_projection_options_v4`, or null. */ void nemo_relay_otel_subscriber_free(struct FfiOpenTelemetrySubscriber *ptr); @@ -3325,7 +3398,8 @@ void nemo_relay_otel_subscriber_free(struct FfiOpenTelemetrySubscriber *ptr); * * # Safety * `ptr` must be a valid pointer returned by - * `nemo_relay_otel_log_subscriber_create`, or null. + * `nemo_relay_otel_log_subscriber_create` or + * `nemo_relay_otel_log_subscriber_create_v2`, or null. */ void nemo_relay_otel_log_subscriber_free(struct FfiOpenTelemetryLogSubscriber *ptr); @@ -3334,7 +3408,8 @@ void nemo_relay_otel_log_subscriber_free(struct FfiOpenTelemetryLogSubscriber *p * * # Safety * `ptr` must be a valid pointer returned by - * `nemo_relay_otel_metric_subscriber_create`, or null. + * `nemo_relay_otel_metric_subscriber_create` or + * `nemo_relay_otel_metric_subscriber_create_v2`, or null. */ void nemo_relay_otel_metric_subscriber_free(struct FfiOpenTelemetryMetricSubscriber *ptr); diff --git a/crates/ffi/src/api/observability.rs b/crates/ffi/src/api/observability.rs index 2a33ae051..f79819465 100644 --- a/crates/ffi/src/api/observability.rs +++ b/crates/ffi/src/api/observability.rs @@ -773,6 +773,7 @@ fn build_ffi_otel_config( transport: *const c_char, endpoint: *const c_char, headers_json: *const c_char, + header_env_json: *const c_char, resource_attributes_json: *const c_char, service_name: *const c_char, service_namespace: *const c_char, @@ -812,6 +813,12 @@ fn build_ffi_otel_config( "headers", OpenTelemetryConfig::with_header, )?; + config = apply_string_map( + config, + header_env_json, + "header_env", + OpenTelemetryConfig::with_header_env, + )?; apply_string_map( config, resource_attributes_json, @@ -849,6 +856,7 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create( transport, endpoint, headers_json, + std::ptr::null(), resource_attributes_json, service_name, service_namespace, @@ -924,11 +932,12 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_optio /// Any non-null C strings must be valid and `out` must be non-null. #[allow(clippy::too_many_arguments)] #[unsafe(no_mangle)] -pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_options_v3( +pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_options_v4( otel_type: *const c_char, transport: *const c_char, endpoint: *const c_char, headers_json: *const c_char, + header_env_json: *const c_char, resource_attributes_json: *const c_char, service_name: *const c_char, service_namespace: *const c_char, @@ -951,6 +960,7 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_optio transport, endpoint, headers_json, + header_env_json, resource_attributes_json, service_name, service_namespace, @@ -993,6 +1003,56 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_optio NemoRelayStatus::Ok } +/// Creates one typed OpenTelemetry exporter subscriber with projection, metadata, and lineage controls. +/// +/// This compatibility entrypoint configures only static headers. Use +/// `nemo_relay_otel_subscriber_create_with_projection_options_v4` for `header_env`. +/// +/// # Safety +/// Any non-null C strings must be valid and `out` must be non-null. +#[allow(clippy::too_many_arguments)] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_options_v3( + otel_type: *const c_char, + transport: *const c_char, + endpoint: *const c_char, + headers_json: *const c_char, + resource_attributes_json: *const c_char, + service_name: *const c_char, + service_namespace: *const c_char, + service_version: *const c_char, + instrumentation_scope: *const c_char, + timeout_millis: u64, + mark_projection: *const c_char, + mark_exclude_names_json: *const c_char, + attribute_mappings_json: *const c_char, + promote_metadata_prefixes_json: *const c_char, + completed_span_context_ttl_millis: u64, + out: *mut *mut FfiOpenTelemetrySubscriber, +) -> NemoRelayStatus { + unsafe { + nemo_relay_otel_subscriber_create_with_projection_options_v4( + otel_type, + transport, + endpoint, + headers_json, + std::ptr::null(), + resource_attributes_json, + service_name, + service_namespace, + service_version, + instrumentation_scope, + timeout_millis, + mark_projection, + mark_exclude_names_json, + attribute_mappings_json, + promote_metadata_prefixes_json, + completed_span_context_ttl_millis, + out, + ) + } +} + /// Creates one typed OpenTelemetry exporter subscriber with projection, metadata, and lineage controls. /// /// This compatibility entrypoint preserves the 60-second completed-context retention default. @@ -1170,6 +1230,7 @@ fn build_ffi_otel_log_config( transport: *const c_char, endpoint: *const c_char, headers_json: *const c_char, + header_env_json: *const c_char, resource_attributes_json: *const c_char, service_name: *const c_char, service_namespace: *const c_char, @@ -1244,6 +1305,12 @@ fn build_ffi_otel_log_config( "headers", OpenTelemetryLogConfig::with_header, )?; + config = apply_string_map( + config, + header_env_json, + "header_env", + OpenTelemetryLogConfig::with_header_env, + )?; apply_string_map( config, resource_attributes_json, @@ -1277,6 +1344,52 @@ pub unsafe extern "C" fn nemo_relay_otel_log_subscriber_create( scheduled_delay_millis: u64, completed_span_context_ttl_millis: u64, out: *mut *mut FfiOpenTelemetryLogSubscriber, +) -> NemoRelayStatus { + unsafe { + nemo_relay_otel_log_subscriber_create_v2( + transport, + endpoint, + headers_json, + std::ptr::null(), + resource_attributes_json, + service_name, + service_namespace, + service_version, + instrumentation_scope, + timeout_millis, + minimum_severity, + max_queue_size, + max_export_batch_size, + scheduled_delay_millis, + completed_span_context_ttl_millis, + out, + ) + } +} + +/// Creates an independently managed OpenTelemetry log subscriber with `header_env` support. +/// +/// # Safety +/// Any non-null C strings must be valid and `out` must be non-null. +#[unsafe(no_mangle)] +#[allow(clippy::too_many_arguments)] +pub unsafe extern "C" fn nemo_relay_otel_log_subscriber_create_v2( + transport: *const c_char, + endpoint: *const c_char, + headers_json: *const c_char, + header_env_json: *const c_char, + resource_attributes_json: *const c_char, + service_name: *const c_char, + service_namespace: *const c_char, + service_version: *const c_char, + instrumentation_scope: *const c_char, + timeout_millis: u64, + minimum_severity: *const c_char, + max_queue_size: u64, + max_export_batch_size: u64, + scheduled_delay_millis: u64, + completed_span_context_ttl_millis: u64, + out: *mut *mut FfiOpenTelemetryLogSubscriber, ) -> NemoRelayStatus { clear_last_error(); if let Err(status) = required_out_ptr(out) { @@ -1286,6 +1399,7 @@ pub unsafe extern "C" fn nemo_relay_otel_log_subscriber_create( transport, endpoint, headers_json, + header_env_json, resource_attributes_json, service_name, service_namespace, @@ -1428,6 +1542,7 @@ fn build_ffi_otel_metric_config( transport: *const c_char, endpoint: *const c_char, headers_json: *const c_char, + header_env_json: *const c_char, resource_attributes_json: *const c_char, service_name: *const c_char, service_namespace: *const c_char, @@ -1487,6 +1602,12 @@ fn build_ffi_otel_metric_config( "headers", OpenTelemetryMetricConfig::with_header, )?; + config = apply_string_map( + config, + header_env_json, + "header_env", + OpenTelemetryMetricConfig::with_header_env, + )?; apply_string_map( config, resource_attributes_json, @@ -1518,6 +1639,50 @@ pub unsafe extern "C" fn nemo_relay_otel_metric_subscriber_create( max_instruments: u64, cardinality_limit: u64, out: *mut *mut FfiOpenTelemetryMetricSubscriber, +) -> NemoRelayStatus { + unsafe { + nemo_relay_otel_metric_subscriber_create_v2( + transport, + endpoint, + headers_json, + std::ptr::null(), + resource_attributes_json, + service_name, + service_namespace, + service_version, + instrumentation_scope, + timeout_millis, + export_interval_millis, + temporality, + max_instruments, + cardinality_limit, + out, + ) + } +} + +/// Creates an independently managed OpenTelemetry metric subscriber with `header_env` support. +/// +/// # Safety +/// Any non-null C strings must be valid and `out` must be non-null. +#[unsafe(no_mangle)] +#[allow(clippy::too_many_arguments)] +pub unsafe extern "C" fn nemo_relay_otel_metric_subscriber_create_v2( + transport: *const c_char, + endpoint: *const c_char, + headers_json: *const c_char, + header_env_json: *const c_char, + resource_attributes_json: *const c_char, + service_name: *const c_char, + service_namespace: *const c_char, + service_version: *const c_char, + instrumentation_scope: *const c_char, + timeout_millis: u64, + export_interval_millis: u64, + temporality: *const c_char, + max_instruments: u64, + cardinality_limit: u64, + out: *mut *mut FfiOpenTelemetryMetricSubscriber, ) -> NemoRelayStatus { clear_last_error(); if let Err(status) = required_out_ptr(out) { @@ -1527,6 +1692,7 @@ pub unsafe extern "C" fn nemo_relay_otel_metric_subscriber_create( transport, endpoint, headers_json, + header_env_json, resource_attributes_json, service_name, service_namespace, diff --git a/crates/ffi/src/types/mod.rs b/crates/ffi/src/types/mod.rs index 235d27ae2..c652085ac 100644 --- a/crates/ffi/src/types/mod.rs +++ b/crates/ffi/src/types/mod.rs @@ -385,10 +385,11 @@ pub unsafe extern "C" fn nemo_relay_atof_exporter_free(ptr: *mut FfiAtofExporter } /// Free an OpenTelemetry subscriber handle previously returned by -/// `nemo_relay_otel_subscriber_create`. +/// an OpenTelemetry subscriber constructor. /// /// # Safety -/// `ptr` must be a valid pointer returned by `nemo_relay_otel_subscriber_create`, or null. +/// `ptr` must be a valid pointer returned by `nemo_relay_otel_subscriber_create` or +/// `nemo_relay_otel_subscriber_create_with_projection_options_v4`, or null. #[unsafe(no_mangle)] pub unsafe extern "C" fn nemo_relay_otel_subscriber_free(ptr: *mut FfiOpenTelemetrySubscriber) { if !ptr.is_null() { @@ -400,7 +401,8 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_free(ptr: *mut FfiOpenTeleme /// /// # Safety /// `ptr` must be a valid pointer returned by -/// `nemo_relay_otel_log_subscriber_create`, or null. +/// `nemo_relay_otel_log_subscriber_create` or +/// `nemo_relay_otel_log_subscriber_create_v2`, or null. #[unsafe(no_mangle)] pub unsafe extern "C" fn nemo_relay_otel_log_subscriber_free( ptr: *mut FfiOpenTelemetryLogSubscriber, @@ -414,7 +416,8 @@ pub unsafe extern "C" fn nemo_relay_otel_log_subscriber_free( /// /// # Safety /// `ptr` must be a valid pointer returned by -/// `nemo_relay_otel_metric_subscriber_create`, or null. +/// `nemo_relay_otel_metric_subscriber_create` or +/// `nemo_relay_otel_metric_subscriber_create_v2`, or null. #[unsafe(no_mangle)] pub unsafe extern "C" fn nemo_relay_otel_metric_subscriber_free( ptr: *mut FfiOpenTelemetryMetricSubscriber, diff --git a/crates/ffi/tests/unit/api/plugin_tests.rs b/crates/ffi/tests/unit/api/plugin_tests.rs index efdb2fc83..eeceee646 100644 --- a/crates/ffi/tests/unit/api/plugin_tests.rs +++ b/crates/ffi/tests/unit/api/plugin_tests.rs @@ -1763,6 +1763,73 @@ fn test_ffi_otel_projection_options_v3_configures_completed_context_ttl() { } } +#[test] +fn test_ffi_otel_projection_options_v4_accepts_header_env_and_keeps_values_secret() { + let _lock = TEST_MUTEX.lock().unwrap_or_else(|error| error.into_inner()); + reset_globals(); + + let variable = "NEMO_RELAY_FFI_OTEL_HEADER_V4"; + let secret = "relay-ffi-secret"; + let header_env_json = format!(r#"{{"authorization":"{variable}"}}"#); + let header_env = cstring(&header_env_json); + let endpoint = c"http://localhost:4318/v1/traces"; + unsafe { + std::env::set_var(variable, "Bearer activation-value"); + let mut subscriber: *mut FfiOpenTelemetrySubscriber = ptr::null_mut(); + assert_status!( + nemo_relay_otel_subscriber_create_with_projection_options_v4( + c"full".as_ptr(), + ptr::null(), + endpoint.as_ptr(), + ptr::null(), + header_env.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + c"inherit".as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + 60_000, + &mut subscriber, + ), + NemoRelayStatus::Ok + ); + nemo_relay_otel_subscriber_free(subscriber); + + std::env::set_var(variable, format!("{secret}\ninvalid")); + let mut invalid_subscriber: *mut FfiOpenTelemetrySubscriber = ptr::null_mut(); + assert_status!( + nemo_relay_otel_subscriber_create_with_projection_options_v4( + c"full".as_ptr(), + ptr::null(), + endpoint.as_ptr(), + ptr::null(), + header_env.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + c"inherit".as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + 60_000, + &mut invalid_subscriber, + ), + NemoRelayStatus::Internal + ); + assert!(invalid_subscriber.is_null()); + assert!(!read_last_error().unwrap_or_default().contains(secret)); + std::env::remove_var(variable); + } +} + #[test] fn test_ffi_specialized_constructor_invalid_utf8_and_malformed_json_sweep() { let _lock = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 92628c81a..377c5dd92 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -317,6 +317,9 @@ fn build_otel_config( for (key, value) in parse_string_map(options.headers, "headers")? { config = config.with_header(key, value); } + for (key, variable) in parse_string_map(options.header_env, "headerEnv")? { + config = config.with_header_env(key, variable); + } for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? { config = config.with_resource_attribute(key, value); } @@ -399,6 +402,9 @@ fn build_otel_log_config( for (key, value) in parse_string_map(options.headers, "headers")? { config = config.with_header(key, value); } + for (key, variable) in parse_string_map(options.header_env, "headerEnv")? { + config = config.with_header_env(key, variable); + } for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? { config = config.with_resource_attribute(key, value); } @@ -447,6 +453,9 @@ fn build_otel_metric_config( for (key, value) in parse_string_map(options.headers, "headers")? { config = config.with_header(key, value); } + for (key, variable) in parse_string_map(options.header_env, "headerEnv")? { + config = config.with_header_env(key, variable); + } for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? { config = config.with_resource_attribute(key, value); } @@ -5147,6 +5156,9 @@ pub struct OpenTelemetryConfig { pub endpoint: String, /// Extra exporter headers/metadata as string key/value pairs. pub headers: Option, + /// Header names mapped to environment variables resolved during subscriber activation. + #[napi(ts_type = "Record")] + pub header_env: Option, /// Extra OpenTelemetry resource attributes as string key/value pairs. pub resource_attributes: Option, /// `service.name` resource attribute. Defaults to `"unknown_service"`. @@ -5243,6 +5255,9 @@ pub struct OpenTelemetryLogConfig { /// Extra exporter headers/metadata as string key/value pairs. #[napi(ts_type = "Record")] pub headers: Option, + /// Header names mapped to environment variables resolved during subscriber activation. + #[napi(ts_type = "Record")] + pub header_env: Option, /// Extra OpenTelemetry resource attributes as string key/value pairs. #[napi(ts_type = "Record")] pub resource_attributes: Option, @@ -5339,6 +5354,9 @@ pub struct OpenTelemetryMetricConfig { /// Extra exporter headers/metadata as string key/value pairs. #[napi(ts_type = "Record")] pub headers: Option, + /// Header names mapped to environment variables resolved during subscriber activation. + #[napi(ts_type = "Record")] + pub header_env: Option, /// Extra OpenTelemetry resource attributes as string key/value pairs. #[napi(ts_type = "Record")] pub resource_attributes: Option, diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs index 57c5427ef..266dfc289 100644 --- a/crates/node/tests/otel_tests.mjs +++ b/crates/node/tests/otel_tests.mjs @@ -83,6 +83,17 @@ describe('OpenTelemetrySubscriber', () => { }), /headers must be an object of string values/i, ); + assert.throws( + () => + new OpenTelemetrySubscriber({ + type: 'full', + endpoint: 'http://localhost:4318/v1/traces', + headerEnv: { + authorization: 1, + }, + }), + /headerEnv must be an object of string values/i, + ); assert.throws( () => new OpenTelemetrySubscriber({ @@ -148,12 +159,17 @@ describe('OpenTelemetrySubscriber', () => { it('exports scope push/pop and mark events end to end', async () => { const collector = await startCollector(); + const variable = `NEMO_RELAY_NODE_HEADER_${Date.now()}`; + const secret = 'Bearer node-activation-secret'; + process.env[variable] = secret; const subscriber = new OpenTelemetrySubscriber({ type: 'full', endpoint: collector.endpoint, serviceName: 'node-agent', promoteMetadataPrefixes: ['nv.'], + headerEnv: { authorization: variable }, }); + process.env[variable] = 'Bearer node-changed-secret'; const name = uniqueId('node_otel_e2e'); subscriber.register(name); @@ -179,16 +195,67 @@ describe('OpenTelemetrySubscriber', () => { const request = await collector.nextRequest(); assert.equal(request.url, '/v1/traces'); assert.equal(request.headers['content-type'], 'application/x-protobuf'); + assert.equal(request.headers.authorization, secret); assert.ok(request.body.length > 0); + assert.equal(request.body.includes(Buffer.from(secret, 'utf8')), false); assertBodyContains(request.body, 'nemo_relay.mark.metadata.source'); assertOtlpStringAttribute(request.body, 'nv.binding', 'node'); + assert.equal( + subscriber.runtimeDiagnostics().some((entry) => entry.message.includes(secret)), + false, + ); } finally { subscriber.deregister(name); subscriber.shutdown(); + delete process.env[variable]; await collector.close(); } }); + it('rejects header_env names that collide with static headers ignoring case', () => { + const variable = `NEMO_RELAY_NODE_DUPLICATE_${Date.now()}`; + process.env[variable] = 'Bearer secret'; + try { + assert.throws( + () => + new OpenTelemetrySubscriber({ + type: 'full', + endpoint: 'http://localhost:4318/v1/traces', + headers: { Authorization: 'static' }, + headerEnv: { authorization: variable }, + }), + /unique across headers and header_env/i, + ); + } finally { + delete process.env[variable]; + } + }); + + it('rejects unset, blank, and invalid header_env values without exposing secrets', () => { + const variable = `NEMO_RELAY_NODE_INVALID_HEADER_${Date.now()}`; + const config = { + type: 'full', + endpoint: 'http://localhost:4318/v1/traces', + headerEnv: { authorization: variable }, + }; + delete process.env[variable]; + assert.throws(() => new OpenTelemetrySubscriber(config), /is not set/i); + + process.env[variable] = ' '; + assert.throws(() => new OpenTelemetrySubscriber(config), /nonblank value/i); + + const secret = 'relay-node-secret'; + process.env[variable] = `${secret}\ninvalid`; + try { + assert.throws( + () => new OpenTelemetrySubscriber(config), + (error) => /valid header value/i.test(error.message) && !error.message.includes(secret), + ); + } finally { + delete process.env[variable]; + } + }); + it('exports the GenAI agent projection end to end', async () => { const collector = await startCollector(); const subscriber = new OpenTelemetrySubscriber({ @@ -217,7 +284,10 @@ describe('OpenTelemetrySubscriber', () => { }); describe('OpenTelemetry log and metric subscribers', () => { - it('constructs signal-specific subscribers and supports lifecycle methods', () => { + it('constructs signal-specific subscribers and supports lifecycle methods', (t) => { + const variable = uniqueId('NEMO_RELAY_NODE_SIGNAL_HEADER'); + process.env[variable] = 'signal-route'; + t.after(() => delete process.env[variable]); const logSubscriber = new OpenTelemetryLogSubscriber({ endpoint: 'http://localhost:4318/v1/logs', minimumSeverity: LogSeverity.Warn, @@ -225,6 +295,7 @@ describe('OpenTelemetry log and metric subscribers', () => { maxExportBatchSize: 16, scheduledDelayMillis: 100, headers: { authorization: 'Bearer token' }, + headerEnv: { 'x-relay-route': variable }, resourceAttributes: { 'deployment.environment': 'test' }, }); const logName = uniqueId('node_otel_log'); @@ -243,6 +314,7 @@ describe('OpenTelemetry log and metric subscribers', () => { maxInstruments: 32, cardinalityLimit: 100, headers: { authorization: 'Bearer token' }, + headerEnv: { 'x-relay-route': variable }, resourceAttributes: { 'deployment.environment': 'test' }, }); const metricName = uniqueId('node_otel_metric'); diff --git a/crates/node/tests/public_observability_api_fixture.ts b/crates/node/tests/public_observability_api_fixture.ts index 2be31e2c9..48779001c 100644 --- a/crates/node/tests/public_observability_api_fixture.ts +++ b/crates/node/tests/public_observability_api_fixture.ts @@ -18,15 +18,18 @@ const dataSchema: DataSchema = { name: 'example.fixture', version: '1' }; const logSubscriber = new OpenTelemetryLogSubscriber({ endpoint: 'http://localhost:4318/v1/logs', + headerEnv: { authorization: 'OTEL_LOG_AUTHORIZATION' }, minimumSeverity: LogSeverity.Warn, }); const metricSubscriber = new OpenTelemetryMetricSubscriber({ endpoint: 'http://localhost:4318/v1/metrics', + headerEnv: { authorization: 'OTEL_METRIC_AUTHORIZATION' }, temporality: MetricTemporality.Delta, }); const traceSubscriber = new OpenTelemetrySubscriber({ type: 'full', endpoint: 'http://localhost:4318/v1/traces', + headerEnv: { authorization: 'OTEL_TRACE_AUTHORIZATION' }, completedSpanContextTtlMillis: 4_294_967_296n, }); diff --git a/crates/python/src/py_types/observability.rs b/crates/python/src/py_types/observability.rs index 387d5253c..d1745d1cb 100644 --- a/crates/python/src/py_types/observability.rs +++ b/crates/python/src/py_types/observability.rs @@ -490,6 +490,7 @@ pub struct PyOpenTelemetryConfig { #[pyo3(get, set)] pub(crate) mark_exclude_names: Vec, pub(crate) headers: HashMap, + pub(crate) header_env: HashMap, pub(crate) resource_attributes: HashMap, pub(crate) attribute_mappings: Vec, #[pyo3(get, set)] @@ -533,6 +534,9 @@ impl PyOpenTelemetryConfig { for (key, value) in &self.headers { config = config.with_header(key.clone(), value.clone()); } + for (key, variable) in &self.header_env { + config = config.with_header_env(key.clone(), variable.clone()); + } for (key, value) in &self.resource_attributes { config = config.with_resource_attribute(key.clone(), value.clone()); } @@ -570,6 +574,7 @@ impl PyOpenTelemetryConfig { mark_projection: "inherit".to_string(), mark_exclude_names: nemo_relay::observability::default_mark_exclude_names(), headers: HashMap::new(), + header_env: HashMap::new(), resource_attributes: HashMap::new(), attribute_mappings: Vec::new(), promote_metadata_prefixes: Vec::new(), @@ -587,6 +592,20 @@ impl PyOpenTelemetryConfig { Ok(()) } + #[getter] + pub(crate) fn header_env(&self, py: Python<'_>) -> PyResult> { + json_to_py( + py, + &serde_json::to_value(&self.header_env).unwrap_or_default(), + ) + } + + #[setter] + pub(crate) fn set_header_env(&mut self, header_env: &Bound<'_, PyAny>) -> PyResult<()> { + self.header_env = py_string_map(header_env, "header_env")?; + Ok(()) + } + #[getter] pub(crate) fn resource_attributes(&self, py: Python<'_>) -> PyResult> { json_to_py( @@ -632,6 +651,10 @@ impl PyOpenTelemetryConfig { self.headers.insert(key, value); } + pub(crate) fn set_header_from_env(&mut self, key: String, variable: String) { + self.header_env.insert(key, variable); + } + pub(crate) fn set_resource_attribute(&mut self, key: String, value: String) { self.resource_attributes.insert(key, value); } @@ -787,6 +810,7 @@ pub struct PyOpenTelemetryLogConfig { #[pyo3(get, set)] pub(crate) completed_span_context_ttl_millis: u64, pub(crate) headers: HashMap, + pub(crate) header_env: HashMap, pub(crate) resource_attributes: HashMap, } @@ -818,6 +842,9 @@ impl PyOpenTelemetryLogConfig { for (key, value) in &self.headers { config = config.with_header(key.clone(), value.clone()); } + for (key, variable) in &self.header_env { + config = config.with_header_env(key.clone(), variable.clone()); + } for (key, value) in &self.resource_attributes { config = config.with_resource_attribute(key.clone(), value.clone()); } @@ -843,6 +870,7 @@ impl PyOpenTelemetryLogConfig { scheduled_delay_millis: 1_000, completed_span_context_ttl_millis: 60_000, headers: HashMap::new(), + header_env: HashMap::new(), resource_attributes: HashMap::new(), } } @@ -858,6 +886,20 @@ impl PyOpenTelemetryLogConfig { Ok(()) } + #[getter] + fn header_env(&self, py: Python<'_>) -> PyResult> { + json_to_py( + py, + &serde_json::to_value(&self.header_env).unwrap_or_default(), + ) + } + + #[setter] + fn set_header_env(&mut self, header_env: &Bound<'_, PyAny>) -> PyResult<()> { + self.header_env = py_string_map(header_env, "header_env")?; + Ok(()) + } + #[getter] fn resource_attributes(&self, py: Python<'_>) -> PyResult> { json_to_py( @@ -876,6 +918,10 @@ impl PyOpenTelemetryLogConfig { self.headers.insert(key, value); } + fn set_header_from_env(&mut self, key: String, variable: String) { + self.header_env.insert(key, variable); + } + fn set_resource_attribute(&mut self, key: String, value: String) { self.resource_attributes.insert(key, value); } @@ -987,6 +1033,7 @@ pub struct PyOpenTelemetryMetricConfig { #[pyo3(get, set)] pub(crate) cardinality_limit: usize, pub(crate) headers: HashMap, + pub(crate) header_env: HashMap, pub(crate) resource_attributes: HashMap, } @@ -1015,6 +1062,9 @@ impl PyOpenTelemetryMetricConfig { for (key, value) in &self.headers { config = config.with_header(key.clone(), value.clone()); } + for (key, variable) in &self.header_env { + config = config.with_header_env(key.clone(), variable.clone()); + } for (key, value) in &self.resource_attributes { config = config.with_resource_attribute(key.clone(), value.clone()); } @@ -1039,6 +1089,7 @@ impl PyOpenTelemetryMetricConfig { max_instruments: 256, cardinality_limit: 2_000, headers: HashMap::new(), + header_env: HashMap::new(), resource_attributes: HashMap::new(), } } @@ -1054,6 +1105,20 @@ impl PyOpenTelemetryMetricConfig { Ok(()) } + #[getter] + fn header_env(&self, py: Python<'_>) -> PyResult> { + json_to_py( + py, + &serde_json::to_value(&self.header_env).unwrap_or_default(), + ) + } + + #[setter] + fn set_header_env(&mut self, header_env: &Bound<'_, PyAny>) -> PyResult<()> { + self.header_env = py_string_map(header_env, "header_env")?; + Ok(()) + } + #[getter] fn resource_attributes(&self, py: Python<'_>) -> PyResult> { json_to_py( @@ -1072,6 +1137,10 @@ impl PyOpenTelemetryMetricConfig { self.headers.insert(key, value); } + fn set_header_from_env(&mut self, key: String, variable: String) { + self.header_env.insert(key, variable); + } + fn set_resource_attribute(&mut self, key: String, value: String) { self.resource_attributes.insert(key, value); } diff --git a/docs/configure-plugins/observability/openinference.mdx b/docs/configure-plugins/observability/openinference.mdx index a016db096..4dc7f93de 100644 --- a/docs/configure-plugins/observability/openinference.mdx +++ b/docs/configure-plugins/observability/openinference.mdx @@ -71,6 +71,7 @@ config = OpenTelemetryConfig( "http://localhost:6006/v1/traces", ) config.service_name = "agent-service" +config.header_env = {"authorization": "OTEL_AUTHORIZATION"} subscriber = OpenTelemetrySubscriber(config) ``` @@ -82,6 +83,7 @@ const subscriber = new OpenTelemetrySubscriber({ type: "openinference", endpoint: "http://localhost:6006/v1/traces", serviceName: "agent-service", + headerEnv: { authorization: "OTEL_AUTHORIZATION" }, }); ``` @@ -94,7 +96,8 @@ let config = OpenTelemetryConfig::new( OpenTelemetryType::OpenInference, "http://localhost:6006/v1/traces", ) -.with_service_name("agent-service"); +.with_service_name("agent-service") +.with_header_env("authorization", "OTEL_AUTHORIZATION"); let subscriber = OpenTelemetrySubscriber::new(config)?; ``` diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index fc54f5702..4d3ae5e8a 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -683,6 +683,7 @@ config = OpenTelemetryConfig( "http://localhost:4318/v1/traces", ) config.service_name = "agent-service" +config.header_env = {"authorization": "OTEL_AUTHORIZATION"} subscriber = OpenTelemetrySubscriber(config) ``` @@ -694,6 +695,7 @@ const subscriber = new OpenTelemetrySubscriber({ type: "gen_ai", endpoint: "http://localhost:4318/v1/traces", serviceName: "agent-service", + headerEnv: { authorization: "OTEL_AUTHORIZATION" }, }); ``` @@ -706,12 +708,29 @@ let config = OpenTelemetryConfig::new( OpenTelemetryType::GenAi, "http://localhost:4318/v1/traces", ) -.with_service_name("agent-service"); +.with_service_name("agent-service") +.with_header_env("authorization", "OTEL_AUTHORIZATION"); let subscriber = OpenTelemetrySubscriber::new(config)?; ``` +Set each referenced environment variable before constructing the subscriber. +Direct trace, log, and metric configs resolve `header_env` when the subscriber +is constructed and retain that value for the subscriber's activation. Changing +the process environment affects only a subsequently constructed subscriber. +Static `headers` remain unchanged. A header name cannot appear in both maps, +including names that differ only by ASCII case, and names within `header_env` +must also be unique ignoring ASCII case. + +Each `header_env` reference must be nonblank, have no surrounding whitespace, +and contain neither `=` nor NUL. Its environment value must be set, nonblank, +unpadded, valid Unicode, and a valid HTTP header value. Validation errors name +the header and environment variable but do not include the resolved value. +Relay supplies resolved values only as outbound OTLP request headers; it does +not copy them into Event data, OpenTelemetry payloads, resource attributes, or +runtime diagnostics. + The log and metric equivalents are `OpenTelemetryLogConfig` with `OpenTelemetryLogSubscriber`, and `OpenTelemetryMetricConfig` with `OpenTelemetryMetricSubscriber`. Each config takes one required endpoint and diff --git a/go/nemo_relay/nemo_relay.go b/go/nemo_relay/nemo_relay.go index 2631dc6a1..301e52525 100644 --- a/go/nemo_relay/nemo_relay.go +++ b/go/nemo_relay/nemo_relay.go @@ -284,6 +284,7 @@ extern int32_t nemo_relay_otel_subscriber_create(const char*, const char*, const extern int32_t nemo_relay_otel_subscriber_create_with_projection_options(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, const char*, const char*, void**); extern int32_t nemo_relay_otel_subscriber_create_with_projection_options_v2(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, const char*, const char*, const char*, void**); extern int32_t nemo_relay_otel_subscriber_create_with_projection_options_v3(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, const char*, const char*, const char*, uint64_t, void**); +extern int32_t nemo_relay_otel_subscriber_create_with_projection_options_v4(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, const char*, const char*, const char*, uint64_t, void**); extern int32_t nemo_relay_otel_subscriber_register(const void*, const char*); extern int32_t nemo_relay_otel_subscriber_deregister(const char*); extern int32_t nemo_relay_otel_subscriber_force_flush(const void*); @@ -291,6 +292,7 @@ extern int32_t nemo_relay_otel_subscriber_runtime_diagnostics_json(const void*, extern int32_t nemo_relay_otel_subscriber_shutdown(const void*); extern void nemo_relay_otel_subscriber_free(void*); extern int32_t nemo_relay_otel_log_subscriber_create(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, uint64_t, uint64_t, uint64_t, uint64_t, void**); +extern int32_t nemo_relay_otel_log_subscriber_create_v2(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, uint64_t, uint64_t, uint64_t, uint64_t, void**); extern int32_t nemo_relay_otel_log_subscriber_register(const void*, const char*); extern int32_t nemo_relay_otel_log_subscriber_deregister(const char*); extern int32_t nemo_relay_otel_log_subscriber_force_flush(const void*); @@ -298,6 +300,7 @@ extern int32_t nemo_relay_otel_log_subscriber_runtime_diagnostics_json(const voi extern int32_t nemo_relay_otel_log_subscriber_shutdown(const void*); extern void nemo_relay_otel_log_subscriber_free(void*); extern int32_t nemo_relay_otel_metric_subscriber_create(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, uint64_t, const char*, uint64_t, uint64_t, void**); +extern int32_t nemo_relay_otel_metric_subscriber_create_v2(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, uint64_t, const char*, uint64_t, uint64_t, void**); extern int32_t nemo_relay_otel_metric_subscriber_register(const void*, const char*); extern int32_t nemo_relay_otel_metric_subscriber_deregister(const char*); extern int32_t nemo_relay_otel_metric_subscriber_force_flush(const void*); @@ -2388,10 +2391,12 @@ const ( // Create it with [NewOpenTelemetryConfig], then mutate fields as needed before // passing it to [NewOpenTelemetrySubscriber]. type OpenTelemetryConfig struct { - Type OpenTelemetryType - Transport OpenTelemetryTransport - Endpoint string - Headers map[string]string + Type OpenTelemetryType + Transport OpenTelemetryTransport + Endpoint string + Headers map[string]string + // HeaderEnv maps outbound header names to environment variables resolved at activation. + HeaderEnv map[string]string ResourceAttributes map[string]string ServiceName string ServiceNamespace string @@ -2413,6 +2418,7 @@ func NewOpenTelemetryConfig(otelType OpenTelemetryType, endpoint string) OpenTel Transport: OpenTelemetryTransportHTTPBinary, Endpoint: endpoint, Headers: map[string]string{}, + HeaderEnv: map[string]string{}, ResourceAttributes: map[string]string{}, ServiceName: "unknown_service", InstrumentationScope: "opentelemetry", @@ -2481,6 +2487,9 @@ func normalizeOpenTelemetryConfig(config OpenTelemetryConfig) (OpenTelemetryConf if config.Headers == nil { config.Headers = map[string]string{} } + if config.HeaderEnv == nil { + config.HeaderEnv = map[string]string{} + } if config.ResourceAttributes == nil { config.ResourceAttributes = map[string]string{} } @@ -2527,6 +2536,12 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc } cHeadersJSON := C.CString(string(headersJSON)) defer C.free(unsafe.Pointer(cHeadersJSON)) + headerEnvJSON, err := jsonMarshal(config.HeaderEnv) + if err != nil { + return nil, err + } + cHeaderEnvJSON := C.CString(string(headerEnvJSON)) + defer C.free(unsafe.Pointer(cHeaderEnvJSON)) resourceAttrsJSON, err := jsonMarshal(config.ResourceAttributes) if err != nil { @@ -2568,11 +2583,12 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc defer C.free(unsafe.Pointer(cPromoteMetadataPrefixesJSON)) var ptr unsafe.Pointer - status := C.nemo_relay_otel_subscriber_create_with_projection_options_v3( + status := C.nemo_relay_otel_subscriber_create_with_projection_options_v4( cType, cTransport, cEndpoint, cHeadersJSON, + cHeaderEnvJSON, cResourceAttrsJSON, cServiceName, cServiceNamespace, @@ -2640,9 +2656,11 @@ func (s *OpenTelemetrySubscriber) Close() { // OpenTelemetryLogConfig configures an independent OTLP log subscriber. type OpenTelemetryLogConfig struct { - Transport OpenTelemetryTransport - Endpoint string - Headers map[string]string + Transport OpenTelemetryTransport + Endpoint string + Headers map[string]string + // HeaderEnv maps outbound header names to environment variables resolved at activation. + HeaderEnv map[string]string ResourceAttributes map[string]string ServiceName string ServiceNamespace string @@ -2662,6 +2680,7 @@ func NewOpenTelemetryLogConfig(endpoint string) OpenTelemetryLogConfig { Transport: OpenTelemetryTransportHTTPBinary, Endpoint: endpoint, Headers: map[string]string{}, + HeaderEnv: map[string]string{}, ResourceAttributes: map[string]string{}, ServiceName: "unknown_service", InstrumentationScope: "opentelemetry", @@ -2690,9 +2709,11 @@ const ( // OpenTelemetryMetricConfig configures an independent OTLP metric subscriber. type OpenTelemetryMetricConfig struct { - Transport OpenTelemetryTransport - Endpoint string - Headers map[string]string + Transport OpenTelemetryTransport + Endpoint string + Headers map[string]string + // HeaderEnv maps outbound header names to environment variables resolved at activation. + HeaderEnv map[string]string ResourceAttributes map[string]string ServiceName string ServiceNamespace string @@ -2711,6 +2732,7 @@ func NewOpenTelemetryMetricConfig(endpoint string) OpenTelemetryMetricConfig { Transport: OpenTelemetryTransportHTTPBinary, Endpoint: endpoint, Headers: map[string]string{}, + HeaderEnv: map[string]string{}, ResourceAttributes: map[string]string{}, ServiceName: "unknown_service", InstrumentationScope: "opentelemetry", @@ -2731,6 +2753,7 @@ type openTelemetrySignalCStrings struct { transport *C.char endpoint *C.char headers *C.char + headerEnv *C.char resourceAttributes *C.char serviceName *C.char serviceNamespace *C.char @@ -2742,6 +2765,7 @@ type openTelemetrySignalConfig struct { transport OpenTelemetryTransport endpoint string headers map[string]string + headerEnv map[string]string resourceAttributes map[string]string serviceName string serviceNamespace string @@ -2754,6 +2778,10 @@ func newOpenTelemetrySignalCStrings(config openTelemetrySignalConfig) (openTelem if err != nil { return openTelemetrySignalCStrings{}, err } + encodedHeaderEnv, err := jsonMarshal(config.headerEnv) + if err != nil { + return openTelemetrySignalCStrings{}, err + } encodedResources, err := jsonMarshal(config.resourceAttributes) if err != nil { return openTelemetrySignalCStrings{}, err @@ -2762,6 +2790,7 @@ func newOpenTelemetrySignalCStrings(config openTelemetrySignalConfig) (openTelem transport: C.CString(string(config.transport)), endpoint: C.CString(config.endpoint), headers: C.CString(string(encodedHeaders)), + headerEnv: C.CString(string(encodedHeaderEnv)), resourceAttributes: C.CString(string(encodedResources)), serviceName: C.CString(config.serviceName), serviceNamespace: optionalCString(config.serviceNamespace), @@ -2774,6 +2803,7 @@ func (values *openTelemetrySignalCStrings) free() { C.free(unsafe.Pointer(values.transport)) C.free(unsafe.Pointer(values.endpoint)) C.free(unsafe.Pointer(values.headers)) + C.free(unsafe.Pointer(values.headerEnv)) C.free(unsafe.Pointer(values.resourceAttributes)) C.free(unsafe.Pointer(values.serviceName)) C.free(unsafe.Pointer(values.serviceNamespace)) @@ -2834,6 +2864,9 @@ func normalizeOpenTelemetryLogConfig(config OpenTelemetryLogConfig) (OpenTelemet if config.Headers == nil { config.Headers = map[string]string{} } + if config.HeaderEnv == nil { + config.HeaderEnv = map[string]string{} + } if config.ResourceAttributes == nil { config.ResourceAttributes = map[string]string{} } @@ -2847,7 +2880,7 @@ func NewOpenTelemetryLogSubscriber(config OpenTelemetryLogConfig) (*OpenTelemetr return nil, err } common, err := newOpenTelemetrySignalCStrings(openTelemetrySignalConfig{ - config.Transport, config.Endpoint, config.Headers, config.ResourceAttributes, + config.Transport, config.Endpoint, config.Headers, config.HeaderEnv, config.ResourceAttributes, config.ServiceName, config.ServiceNamespace, config.ServiceVersion, config.InstrumentationScope, }) if err != nil { @@ -2857,10 +2890,11 @@ func NewOpenTelemetryLogSubscriber(config OpenTelemetryLogConfig) (*OpenTelemetr cSeverity := C.CString(string(config.MinimumSeverity)) defer C.free(unsafe.Pointer(cSeverity)) var ptr unsafe.Pointer - status := C.nemo_relay_otel_log_subscriber_create( + status := C.nemo_relay_otel_log_subscriber_create_v2( common.transport, common.endpoint, common.headers, + common.headerEnv, common.resourceAttributes, common.serviceName, common.serviceNamespace, @@ -2962,6 +2996,9 @@ func normalizeOpenTelemetryMetricConfig(config OpenTelemetryMetricConfig) (OpenT if config.Headers == nil { config.Headers = map[string]string{} } + if config.HeaderEnv == nil { + config.HeaderEnv = map[string]string{} + } if config.ResourceAttributes == nil { config.ResourceAttributes = map[string]string{} } @@ -2975,7 +3012,7 @@ func NewOpenTelemetryMetricSubscriber(config OpenTelemetryMetricConfig) (*OpenTe return nil, err } common, err := newOpenTelemetrySignalCStrings(openTelemetrySignalConfig{ - config.Transport, config.Endpoint, config.Headers, config.ResourceAttributes, + config.Transport, config.Endpoint, config.Headers, config.HeaderEnv, config.ResourceAttributes, config.ServiceName, config.ServiceNamespace, config.ServiceVersion, config.InstrumentationScope, }) if err != nil { @@ -2985,10 +3022,11 @@ func NewOpenTelemetryMetricSubscriber(config OpenTelemetryMetricConfig) (*OpenTe cTemporality := C.CString(string(config.Temporality)) defer C.free(unsafe.Pointer(cTemporality)) var ptr unsafe.Pointer - status := C.nemo_relay_otel_metric_subscriber_create( + status := C.nemo_relay_otel_metric_subscriber_create_v2( common.transport, common.endpoint, common.headers, + common.headerEnv, common.resourceAttributes, common.serviceName, common.serviceNamespace, diff --git a/go/nemo_relay/otel_signals_test.go b/go/nemo_relay/otel_signals_test.go index 5c528eacd..547e8dff0 100644 --- a/go/nemo_relay/otel_signals_test.go +++ b/go/nemo_relay/otel_signals_test.go @@ -213,13 +213,21 @@ func TestOpenTelemetrySignalConfigRejectsFractionalMillisecondDurations(t *testi func TestOpenTelemetrySubscribersExposeRuntimeDiagnostics(t *testing.T) { endpoint := "http://127.0.0.1:4318/v1/traces" - traceSubscriber, err := NewOpenTelemetrySubscriber(NewOpenTelemetryConfig(OpenTelemetryTypeFull, endpoint)) + variable := "NEMO_RELAY_GO_SIGNAL_HEADER_" + time.Now().Format(otelTimeFormat) + t.Setenv(variable, "signal-route") + traceConfig := NewOpenTelemetryConfig(OpenTelemetryTypeFull, endpoint) + traceConfig.HeaderEnv["x-relay-route"] = variable + traceSubscriber, err := NewOpenTelemetrySubscriber(traceConfig) requireNoError(t, err, "NewOpenTelemetrySubscriber failed") defer traceSubscriber.Close() - logSubscriber, err := NewOpenTelemetryLogSubscriber(NewOpenTelemetryLogConfig(endpoint)) + logConfig := NewOpenTelemetryLogConfig(endpoint) + logConfig.HeaderEnv["x-relay-route"] = variable + logSubscriber, err := NewOpenTelemetryLogSubscriber(logConfig) requireNoError(t, err, "NewOpenTelemetryLogSubscriber failed") defer logSubscriber.Close() - metricSubscriber, err := NewOpenTelemetryMetricSubscriber(NewOpenTelemetryMetricConfig(endpoint)) + metricConfig := NewOpenTelemetryMetricConfig(endpoint) + metricConfig.HeaderEnv["x-relay-route"] = variable + metricSubscriber, err := NewOpenTelemetryMetricSubscriber(metricConfig) requireNoError(t, err, "NewOpenTelemetryMetricSubscriber failed") defer metricSubscriber.Close() diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go index 4f1e13392..82a842fd9 100644 --- a/go/nemo_relay/otel_test.go +++ b/go/nemo_relay/otel_test.go @@ -10,6 +10,8 @@ import ( "io" "net/http" "net/http/httptest" + "os" + "strings" "testing" "time" ) @@ -54,6 +56,9 @@ func TestNewOpenTelemetryConfigDefaults(t *testing.T) { if config.Headers == nil || len(config.Headers) != 0 { t.Fatalf("expected empty headers map, got %#v", config.Headers) } + if config.HeaderEnv == nil || len(config.HeaderEnv) != 0 { + t.Fatalf("expected empty header environment map, got %#v", config.HeaderEnv) + } if config.ResourceAttributes == nil || len(config.ResourceAttributes) != 0 { t.Fatalf("expected empty resource attributes map, got %#v", config.ResourceAttributes) } @@ -185,9 +190,10 @@ func TestOpenTelemetrySubscriberRejectsInvalidRequiredFields(t *testing.T) { func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { type otelRequest struct { - Path string - ContentType string - Body []byte + Path string + ContentType string + Authorization string + Body []byte } requests := make(chan otelRequest, 4) @@ -197,9 +203,10 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { t.Errorf("read request body: %v", err) } requests <- otelRequest{ - Path: r.URL.Path, - ContentType: r.Header.Get("Content-Type"), - Body: body, + Path: r.URL.Path, + ContentType: r.Header.Get("Content-Type"), + Authorization: r.Header.Get("Authorization"), + Body: body, } w.WriteHeader(http.StatusOK) })) @@ -208,11 +215,18 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { config := NewOpenTelemetryConfig(OpenTelemetryTypeFull, server.URL+otelTestPath) config.ServiceName = "go-agent" config.PromoteMetadataPrefixes = []string{"nv."} + variable := "NEMO_RELAY_GO_HEADER_" + time.Now().Format(otelTimeFormat) + secret := "Bearer go-activation-secret" + t.Setenv(variable, secret) + config.HeaderEnv["authorization"] = variable subscriber, err := NewOpenTelemetrySubscriber(config) if err != nil { t.Fatalf(newOpenTelemetrySubscriberFailed, err) } defer subscriber.Close() + if err := os.Setenv(variable, "Bearer go-changed-secret"); err != nil { + t.Fatalf("change header environment: %v", err) + } name := "go_otel_e2e_" + time.Now().Format(otelTimeFormat) if err := subscriber.Register(name); err != nil { @@ -256,16 +270,64 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { if request.ContentType != "application/x-protobuf" { t.Fatalf("expected protobuf content type, got %q", request.ContentType) } + if request.Authorization != secret { + t.Fatalf("expected activation-time authorization header, got %q", request.Authorization) + } if len(request.Body) == 0 { t.Fatal("expected non-empty OTLP request body") } + if bytes.Contains(request.Body, []byte(secret)) { + t.Fatal("authorization value must not appear in the OTLP payload") + } assertOtlpStringAttribute(t, request.Body, "nemo_relay.scope_type", "agent") assertOtlpStringAttribute(t, request.Body, "nv.binding", "go") + diagnostics, err := subscriber.RuntimeDiagnostics() + if err != nil { + t.Fatalf("RuntimeDiagnostics failed: %v", err) + } + for _, diagnostic := range diagnostics { + if strings.Contains(diagnostic.Message, secret) { + t.Fatal("authorization value must not appear in runtime diagnostics") + } + } case <-time.After(5 * time.Second): t.Fatal("timed out waiting for OTLP request") } } +func TestOpenTelemetrySubscriberRejectsInvalidHeaderEnvWithoutSecretValues(t *testing.T) { + variable := "NEMO_RELAY_GO_INVALID_HEADER_" + time.Now().Format(otelTimeFormat) + secret := "relay-go-secret" + + config := NewOpenTelemetryConfig(OpenTelemetryTypeFull, otelTestEndpoint) + config.HeaderEnv["authorization"] = variable + if _, err := NewOpenTelemetrySubscriber(config); err == nil { + t.Fatal("expected unset header environment variable to fail") + } + + t.Setenv(variable, " ") + if _, err := NewOpenTelemetrySubscriber(config); err == nil { + t.Fatal("expected blank header environment variable to fail") + } + + if err := os.Setenv(variable, secret+"\ninvalid"); err != nil { + t.Fatalf("set invalid header value: %v", err) + } + if _, err := NewOpenTelemetrySubscriber(config); err == nil { + t.Fatal("expected invalid header value to fail") + } else if strings.Contains(err.Error(), secret) { + t.Fatal("invalid header error exposed the environment-derived value") + } + + if err := os.Setenv(variable, "valid"); err != nil { + t.Fatalf("set valid header value: %v", err) + } + config.Headers["Authorization"] = "static" + if _, err := NewOpenTelemetrySubscriber(config); err == nil { + t.Fatal("expected case-insensitive header collision to fail") + } +} + func TestOpenTelemetrySubscriberExportsGenAIAgentProjection(t *testing.T) { requests := make(chan otelRequest, 1) server := NewOtelTestServer(t, requests) diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index b5169e8c0..82bb04809 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -1171,6 +1171,14 @@ class OpenTelemetryConfig: """Replace additional exporter headers.""" ... @property + def header_env(self) -> dict[str, str]: + """Return header names mapped to environment variable names.""" + ... + @header_env.setter + def header_env(self, value: dict[str, str]) -> None: + """Replace environment-backed exporter header references.""" + ... + @property def resource_attributes(self) -> dict[str, str]: """Return additional OpenTelemetry resource attributes.""" ... @@ -1189,6 +1197,9 @@ class OpenTelemetryConfig: def set_header(self, key: str, value: str) -> None: """Set one exporter header key/value pair.""" ... + def set_header_from_env(self, key: str, variable: str) -> None: + """Map one exporter header to an environment variable.""" + ... def set_resource_attribute(self, key: str, value: str) -> None: """Set one OpenTelemetry resource attribute key/value pair.""" ... @@ -1262,10 +1273,15 @@ class OpenTelemetryLogConfig: @headers.setter def headers(self, value: dict[str, str]) -> None: ... @property + def header_env(self) -> dict[str, str]: ... + @header_env.setter + def header_env(self, value: dict[str, str]) -> None: ... + @property def resource_attributes(self) -> dict[str, str]: ... @resource_attributes.setter def resource_attributes(self, value: dict[str, str]) -> None: ... def set_header(self, key: str, value: str) -> None: ... + def set_header_from_env(self, key: str, variable: str) -> None: ... def set_resource_attribute(self, key: str, value: str) -> None: ... class OpenTelemetryLogSubscriber: @@ -1306,10 +1322,15 @@ class OpenTelemetryMetricConfig: @headers.setter def headers(self, value: dict[str, str]) -> None: ... @property + def header_env(self) -> dict[str, str]: ... + @header_env.setter + def header_env(self, value: dict[str, str]) -> None: ... + @property def resource_attributes(self) -> dict[str, str]: ... @resource_attributes.setter def resource_attributes(self, value: dict[str, str]) -> None: ... def set_header(self, key: str, value: str) -> None: ... + def set_header_from_env(self, key: str, variable: str) -> None: ... def set_resource_attribute(self, key: str, value: str) -> None: ... class OpenTelemetryMetricSubscriber: diff --git a/python/tests/test_types.py b/python/tests/test_types.py index 88994c9fb..50228625e 100644 --- a/python/tests/test_types.py +++ b/python/tests/test_types.py @@ -616,16 +616,21 @@ def test_signal_subscribers_expose_runtime_diagnostics(self, signal): subscriber.deregister(subscriber_name) subscriber.shutdown() - def test_signal_config_defaults_and_lifecycle(self): + def test_signal_config_defaults_and_lifecycle(self, monkeypatch: pytest.MonkeyPatch): + variable = f"NEMO_RELAY_PY_SIGNAL_HEADER_{uuid4().hex}" + monkeypatch.setenv(variable, "signal-route") log_config = OpenTelemetryLogConfig("http://localhost:4318/v1/logs") assert log_config.minimum_severity == LogSeverity.Info assert log_config.max_queue_size == 2048 assert log_config.max_export_batch_size == 512 assert log_config.scheduled_delay_millis == 1000 assert log_config.completed_span_context_ttl_millis == 60000 + assert log_config.header_env == {} log_config.minimum_severity = LogSeverity.Warn log_config.headers = {"authorization": "Bearer token"} + log_config.set_header_from_env("x-relay-route", variable) log_config.resource_attributes = {"deployment.environment": "test"} + assert log_config.header_env == {"x-relay-route": variable} log_subscriber = OpenTelemetryLogSubscriber(log_config) log_name = f"py_otel_log_{uuid4().hex}" @@ -642,9 +647,12 @@ def test_signal_config_defaults_and_lifecycle(self): assert metric_config.temporality == MetricTemporality.Cumulative assert metric_config.max_instruments == 256 assert metric_config.cardinality_limit == 2000 + assert metric_config.header_env == {} metric_config.temporality = MetricTemporality.Delta metric_config.headers = {"authorization": "Bearer token"} + metric_config.set_header_from_env("x-relay-route", variable) metric_config.resource_attributes = {"deployment.environment": "test"} + assert metric_config.header_env == {"x-relay-route": variable} metric_subscriber = OpenTelemetryMetricSubscriber(metric_config) metric_name = f"py_otel_metric_{uuid4().hex}" @@ -746,6 +754,7 @@ def test_config_defaults_mutation_and_repr(self): assert config.instrumentation_scope == "opentelemetry" assert config.timeout_millis == 3000 assert config.headers == {} + assert config.header_env == {} assert config.resource_attributes == {} assert config.mark_projection == "inherit" assert config.mark_exclude_names == ["llm.chunk"] @@ -758,6 +767,7 @@ def test_config_defaults_mutation_and_repr(self): config.instrumentation_scope = "py-tests" config.timeout_millis = 1250 config.set_header("authorization", "Bearer token") + config.set_header_from_env("x-api-key", "NEMO_RELAY_API_KEY") config.set_resource_attribute("deployment.environment", "test") config.mark_projection = "tool" config.mark_exclude_names = ["custom.mark"] @@ -765,6 +775,7 @@ def test_config_defaults_mutation_and_repr(self): config.promote_metadata_prefixes = ["nv."] assert config.headers == {"authorization": "Bearer token"} + assert config.header_env == {"x-api-key": "NEMO_RELAY_API_KEY"} assert config.resource_attributes == {"deployment.environment": "test"} assert config.mark_projection == "tool" assert config.mark_exclude_names == ["custom.mark"] @@ -778,6 +789,9 @@ def test_config_rejects_invalid_map_values(self): with pytest.raises(ValueError, match="dict\\[str, str\\]"): config.headers = cast(dict[str, str], []) + with pytest.raises(ValueError, match="dict\\[str, str\\]"): + config.header_env = cast(dict[str, str], {"authorization": 1}) + with pytest.raises(ValueError, match="dict\\[str, str\\]"): config.resource_attributes = cast(dict[str, str], {"env": 1}) @@ -831,14 +845,19 @@ def test_subscriber_rejects_missing_or_invalid_required_fields(self): with pytest.raises(RuntimeError, match="completed_span_context_ttl must be greater than 0"): OpenTelemetrySubscriber(zero_ttl) - def test_subscriber_exports_scope_and_mark_events_end_to_end(self): + def test_subscriber_exports_scope_and_mark_events_end_to_end(self, monkeypatch: pytest.MonkeyPatch): with _OtelCollector() as collector: source = "python-é" * 20 + variable = f"NEMO_RELAY_PY_HEADER_{uuid4().hex}" + secret = "Bearer python-activation-secret" + monkeypatch.setenv(variable, secret) config = OpenTelemetryConfig("full", collector.endpoint) config.service_name = "py-agent" config.promote_metadata_prefixes = ["nv."] + config.header_env = {"authorization": variable} subscriber = OpenTelemetrySubscriber(config) + monkeypatch.setenv(variable, "Bearer python-changed-secret") subscriber_name = f"py_otel_e2e_{uuid4().hex}" subscriber.register(subscriber_name) @@ -863,13 +882,45 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): request = collector.wait_for_request() assert request["path"] == "/v1/traces" assert request["headers"]["content-type"] == "application/x-protobuf" + assert request["headers"]["authorization"] == secret assert request["body"] + assert secret.encode() not in request["body"] assert b"nemo_relay.mark.metadata.source" in request["body"] assert _otlp_string_attribute("nv.binding", "python") in request["body"] + assert all(secret not in entry.message for entry in subscriber.runtime_diagnostics().entries) finally: subscriber.deregister(subscriber_name) subscriber.shutdown() + def test_subscriber_rejects_header_env_case_collision(self, monkeypatch: pytest.MonkeyPatch): + variable = f"NEMO_RELAY_PY_DUPLICATE_HEADER_{uuid4().hex}" + monkeypatch.setenv(variable, "Bearer secret") + config = OpenTelemetryConfig("full", "http://localhost:4318/v1/traces") + config.headers = {"Authorization": "static"} + config.header_env = {"authorization": variable} + + with pytest.raises(RuntimeError, match="unique across headers and header_env"): + OpenTelemetrySubscriber(config) + + def test_subscriber_rejects_unset_blank_and_invalid_header_env_values(self, monkeypatch: pytest.MonkeyPatch): + variable = f"NEMO_RELAY_PY_INVALID_HEADER_{uuid4().hex}" + config = OpenTelemetryConfig("full", "http://localhost:4318/v1/traces") + config.header_env = {"authorization": variable} + + monkeypatch.delenv(variable, raising=False) + with pytest.raises(RuntimeError, match="is not set"): + OpenTelemetrySubscriber(config) + + monkeypatch.setenv(variable, " ") + with pytest.raises(RuntimeError, match="nonblank value"): + OpenTelemetrySubscriber(config) + + secret = "relay-python-secret" + monkeypatch.setenv(variable, f"{secret}\ninvalid") + with pytest.raises(RuntimeError, match="valid header value") as failure: + OpenTelemetrySubscriber(config) + assert secret not in str(failure.value) + def test_trace_export_failure_stays_unhealthy_until_a_later_export_succeeds(self): with _OtelCollector(response_status=503) as collector: subscriber = OpenTelemetrySubscriber(OpenTelemetryConfig("full", collector.endpoint))