From 4fec371e09472df37fc37c2f1f104a923e513571 Mon Sep 17 00:00:00 2001 From: alsruf36 Date: Wed, 13 May 2026 19:24:30 +0900 Subject: [PATCH 1/2] fix(observability): set engine-originated span resource identity --- engine/src/logging.rs | 10 + engine/src/workers/observability/otel.rs | 157 +++++++++++--- .../otel_tracer_provider_resource_test.rs | 197 ++++++++++++++++++ 3 files changed, 332 insertions(+), 32 deletions(-) create mode 100644 engine/tests/otel_tracer_provider_resource_test.rs diff --git a/engine/src/logging.rs b/engine/src/logging.rs index 59178efa4f..d61b066533 100644 --- a/engine/src/logging.rs +++ b/engine/src/logging.rs @@ -408,6 +408,12 @@ fn extract_otel_config(cfg: &EngineConfig) -> OtelConfig { if let Some(service_name) = module_config.service_name { otel_cfg.service_name = service_name; } + if let Some(service_version) = module_config.service_version { + otel_cfg.service_version = service_version; + } + if let Some(service_namespace) = module_config.service_namespace { + otel_cfg.service_namespace = Some(service_namespace); + } if let Some(exporter) = module_config.exporter { otel_cfg.exporter = match exporter { crate::workers::observability::config::OtelExporterType::Memory => ExporterType::Memory, @@ -1042,6 +1048,8 @@ mod tests { config: Some(serde_json::json!({ "enabled": true, "service_name": "test-service", + "service_version": "1.2.3", + "service_namespace": "production", "exporter": "memory", "endpoint": "http://collector:4317", "sampling_ratio": 0.25, @@ -1054,6 +1062,8 @@ mod tests { let otel = extract_otel_config(&cfg); assert!(otel.enabled); assert_eq!(otel.service_name, "test-service"); + assert_eq!(otel.service_version, "1.2.3"); + assert_eq!(otel.service_namespace.as_deref(), Some("production")); assert!(matches!(otel.exporter, ExporterType::Memory)); assert_eq!(otel.endpoint, "http://collector:4317"); assert_eq!(otel.sampling_ratio, 0.25); diff --git a/engine/src/workers/observability/otel.rs b/engine/src/workers/observability/otel.rs index 73b096d34b..6a7c800044 100644 --- a/engine/src/workers/observability/otel.rs +++ b/engine/src/workers/observability/otel.rs @@ -110,6 +110,22 @@ pub struct OtelConfig { pub memory_max_spans: usize, } +fn non_blank(value: String) -> Option { + (!value.trim().is_empty()).then_some(value) +} + +fn service_name_or_default(configured: &str) -> String { + non_blank(configured.to_string()) + .or_else(|| env::var("OTEL_SERVICE_NAME").ok().and_then(non_blank)) + .unwrap_or_else(|| "iii".to_string()) +} + +fn service_version_or_default(configured: &str) -> String { + non_blank(configured.to_string()) + .or_else(|| env::var("SERVICE_VERSION").ok().and_then(non_blank)) + .unwrap_or_else(|| "unknown".to_string()) +} + impl Default for OtelConfig { fn default() -> Self { // First check global config from YAML, then fall back to environment variables @@ -124,19 +140,22 @@ impl Default for OtelConfig { }) .unwrap_or(false); - let service_name = global_cfg - .and_then(|c| c.service_name.clone()) - .or_else(|| env::var("OTEL_SERVICE_NAME").ok()) - .unwrap_or_else(|| "iii".to_string()); + let service_name = service_name_or_default( + global_cfg + .and_then(|c| c.service_name.as_deref()) + .unwrap_or(""), + ); - let service_version = global_cfg - .and_then(|c| c.service_version.clone()) - .or_else(|| env::var("SERVICE_VERSION").ok()) - .unwrap_or_else(|| "unknown".to_string()); + let service_version = service_version_or_default( + global_cfg + .and_then(|c| c.service_version.as_deref()) + .unwrap_or(""), + ); let service_namespace = global_cfg .and_then(|c| c.service_namespace.clone()) - .or_else(|| env::var("SERVICE_NAMESPACE").ok()); + .and_then(non_blank) + .or_else(|| env::var("SERVICE_NAMESPACE").ok().and_then(non_blank)); let exporter = global_cfg .and_then(|c| c.exporter.clone()) @@ -852,19 +871,24 @@ where // Build the sampler using advanced configuration if available let sampler = build_sampler(config); - // Build resource attributes with OTEL semantic conventions - // Using string keys for attributes not available in the crate version + let service_name = service_name_or_default(&config.service_name); + let service_version = service_version_or_default(&config.service_version); + let service_namespace = config + .service_namespace + .clone() + .and_then(non_blank) + .or_else(|| env::var("SERVICE_NAMESPACE").ok().and_then(non_blank)); + let mut resource_builder = Resource::builder() - .with_service_name(config.service_name.clone()) + .with_service_name(service_name.clone()) .with_attributes([ - KeyValue::new("service.version", config.service_version.clone()), + KeyValue::new("service.version", service_version), KeyValue::new("service.instance.id", uuid::Uuid::new_v4().to_string()), ]); - // Only add namespace if provided (optional attribute) - if let Some(namespace) = &config.service_namespace { + if let Some(namespace) = service_namespace { resource_builder = - resource_builder.with_attribute(KeyValue::new("service.namespace", namespace.clone())); + resource_builder.with_attribute(KeyValue::new("service.namespace", namespace)); } let resource = resource_builder.build(); @@ -899,10 +923,8 @@ where "Failed to create OTLP exporter, falling back to memory-only mode" ); // Fall back to memory-only mode - let exporter = InMemorySpanExporter::new( - config.memory_max_spans, - config.service_name.clone(), - ); + let exporter = + InMemorySpanExporter::new(config.memory_max_spans, service_name.clone()); SdkTracerProvider::builder() .with_simple_exporter(exporter) .with_sampler(sampler) @@ -913,8 +935,7 @@ where } } ExporterType::Memory => { - let exporter = - InMemorySpanExporter::new(config.memory_max_spans, config.service_name.clone()); + let exporter = InMemorySpanExporter::new(config.memory_max_spans, service_name.clone()); SdkTracerProvider::builder() .with_simple_exporter(exporter) @@ -940,11 +961,8 @@ where init_sdk_span_forwarder(&config.endpoint); // Create tee exporter that sends to both - let tee_exporter = TeeSpanExporter::new( - otlp_exporter, - memory_storage, - config.service_name.clone(), - ); + let tee_exporter = + TeeSpanExporter::new(otlp_exporter, memory_storage, service_name.clone()); SdkTracerProvider::builder() .with_batch_exporter(tee_exporter) @@ -960,10 +978,8 @@ where "Failed to create OTLP exporter for 'both' mode, using memory-only" ); // Fall back to memory-only with our already-created storage - let exporter = InMemorySpanExporter::with_storage( - memory_storage, - config.service_name.clone(), - ); + let exporter = + InMemorySpanExporter::with_storage(memory_storage, service_name.clone()); SdkTracerProvider::builder() .with_simple_exporter(exporter) .with_sampler(sampler) @@ -997,7 +1013,7 @@ where println!( "OpenTelemetry initialized: exporter={}, service_name={}, sampling_ratio={}", - exporter_info, config.service_name, config.sampling_ratio + exporter_info, service_name, config.sampling_ratio ); Some(OpenTelemetryLayer::new(tracer)) @@ -3183,6 +3199,83 @@ mod tests { use super::*; use serial_test::serial; + fn restore_env_var(name: &str, value: Option) { + unsafe { + match value { + Some(value) => env::set_var(name, value), + None => env::remove_var(name), + } + } + } + + struct EnvVarGuard { + name: &'static str, + value: Option, + } + + impl EnvVarGuard { + fn set(name: &'static str, value: &str) -> Self { + let guard = Self { + name, + value: env::var_os(name), + }; + unsafe { + env::set_var(name, value); + } + guard + } + + fn remove(name: &'static str) -> Self { + let guard = Self { + name, + value: env::var_os(name), + }; + unsafe { + env::remove_var(name); + } + guard + } + } + + impl Drop for EnvVarGuard { + fn drop(&mut self) { + restore_env_var(self.name, self.value.clone()); + } + } + + #[test] + #[serial] + fn test_service_name_or_default_uses_env_when_config_is_blank() { + let _service_name = EnvVarGuard::set("OTEL_SERVICE_NAME", "iii-engine"); + + assert_eq!(service_name_or_default(""), "iii-engine"); + assert_eq!(service_name_or_default(" "), "iii-engine"); + } + + #[test] + #[serial] + fn test_service_name_or_default_uses_builtin_default_when_config_and_env_are_blank() { + let _service_name = EnvVarGuard::set("OTEL_SERVICE_NAME", ""); + + assert_eq!(service_name_or_default(""), "iii"); + } + + #[test] + #[serial] + fn test_service_version_or_default_uses_env_when_config_is_blank() { + let _service_version = EnvVarGuard::set("SERVICE_VERSION", "1.2.3"); + + assert_eq!(service_version_or_default(""), "1.2.3"); + } + + #[test] + #[serial] + fn test_service_version_or_default_uses_builtin_default_when_config_and_env_are_blank() { + let _service_version = EnvVarGuard::remove("SERVICE_VERSION"); + + assert_eq!(service_version_or_default(""), "unknown"); + } + #[tokio::test] #[serial] async fn test_ingest_otlp_metrics_gauge() { diff --git a/engine/tests/otel_tracer_provider_resource_test.rs b/engine/tests/otel_tracer_provider_resource_test.rs new file mode 100644 index 0000000000..58e5562be8 --- /dev/null +++ b/engine/tests/otel_tracer_provider_resource_test.rs @@ -0,0 +1,197 @@ +// Regression test for iii-hq/iii#1619 — engine-originated spans must +// export a non-empty service.name resource through the engine SdkTracerProvider. +// +// This test intentionally lives in its own integration-test binary: init_otel +// stores TRACER_PROVIDER in a process-global OnceLock and installs a global +// OpenTelemetry provider, so sharing a process with other OTEL tests would make +// the assertions order-dependent. + +use std::sync::Arc; +use std::time::Duration; + +use opentelemetry_proto::tonic::collector::trace::v1::{ + ExportTraceServiceRequest, ExportTraceServiceResponse, + trace_service_server::{TraceService, TraceServiceServer}, +}; +use tokio::sync::Mutex; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; +use tracing_subscriber::prelude::*; + +#[derive(Default, Clone)] +struct CapturingTraceService { + received: Arc>>, +} + +#[tonic::async_trait] +impl TraceService for CapturingTraceService { + async fn export( + &self, + request: Request, + ) -> Result, Status> { + self.received.lock().await.push(request.into_inner()); + Ok(Response::new(ExportTraceServiceResponse::default())) + } +} + +struct EnvVarGuard { + name: &'static str, + value: Option, +} + +impl EnvVarGuard { + fn set(name: &'static str, value: &str) -> Self { + let guard = Self { + name, + value: std::env::var_os(name), + }; + unsafe { + std::env::set_var(name, value); + } + guard + } + + fn remove(name: &'static str) -> Self { + let guard = Self { + name, + value: std::env::var_os(name), + }; + unsafe { + std::env::remove_var(name); + } + guard + } +} + +impl Drop for EnvVarGuard { + fn drop(&mut self) { + unsafe { + match &self.value { + Some(value) => std::env::set_var(self.name, value), + None => std::env::remove_var(self.name), + } + } + } +} + +async fn spawn_mock_collector() -> (String, Arc>>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let endpoint = format!("http://{addr}"); + + let service = CapturingTraceService::default(); + let received = service.received.clone(); + + tokio::spawn(async move { + let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener); + Server::builder() + .add_service(TraceServiceServer::new(service)) + .serve_with_incoming(incoming) + .await + .expect("mock OTLP collector failed to start; see tonic transport error above") + }); + + let deadline = std::time::Instant::now() + Duration::from_secs(2); + loop { + match tokio::net::TcpStream::connect(addr).await { + Ok(_) => break, + Err(_) if std::time::Instant::now() < deadline => { + tokio::task::yield_now().await; + } + Err(e) => panic!("mock collector not reachable within 2s: {e}"), + } + } + + (endpoint, received) +} + +fn find_string_attr<'a>( + attrs: &'a [opentelemetry_proto::tonic::common::v1::KeyValue], + key: &str, +) -> Option<&'a str> { + attrs + .iter() + .find(|kv| kv.key == key) + .and_then(|kv| kv.value.as_ref()) + .and_then(|v| v.value.as_ref()) + .and_then(|v| match v { + opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue(s) => { + Some(s.as_str()) + } + _ => None, + }) +} + +fn resource_for_span<'a>( + req: &'a ExportTraceServiceRequest, + span_name: &str, +) -> Option<&'a opentelemetry_proto::tonic::resource::v1::Resource> { + req.resource_spans.iter().find_map(|resource_spans| { + let has_span = resource_spans + .scope_spans + .iter() + .any(|scope_spans| scope_spans.spans.iter().any(|span| span.name == span_name)); + + has_span.then(|| resource_spans.resource.as_ref()).flatten() + }) +} + +#[tokio::test] +async fn engine_originated_spans_fall_back_to_otel_service_name_when_config_service_name_is_blank() +{ + let _service_name = EnvVarGuard::set("OTEL_SERVICE_NAME", "iii-engine"); + let _resource_attrs = EnvVarGuard::remove("OTEL_RESOURCE_ATTRIBUTES"); + + let (endpoint, received) = spawn_mock_collector().await; + let config = iii::workers::observability::otel::OtelConfig { + enabled: true, + service_name: String::new(), + service_version: "test-version".to_string(), + service_namespace: None, + exporter: iii::workers::observability::otel::ExporterType::Otlp, + endpoint, + sampling_ratio: 1.0, + memory_max_spans: 10, + }; + + let otel_layer = iii::workers::observability::otel::init_otel(&config) + .expect("OpenTelemetry should initialize for the mock collector"); + let subscriber = tracing_subscriber::registry().with(otel_layer); + + tracing::subscriber::with_default(subscriber, || { + let span = tracing::info_span!("engine-originated-test-span"); + let _enter = span.enter(); + tracing::info!("emitting engine-originated span for iii#1619 regression test"); + }); + + iii::workers::observability::otel::shutdown_otel(); + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if !received.lock().await.is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + + let captured = received.lock().await; + assert_eq!( + captured.len(), + 1, + "Mock collector should have received exactly one engine-originated ExportTraceServiceRequest" + ); + + let resource = resource_for_span(&captured[0], "engine-originated-test-span") + .expect("engine-originated test span should be exported with a Resource block"); + + assert_eq!( + find_string_attr(&resource.attributes, "service.name"), + Some("iii-engine"), + "engine-originated spans must use OTEL_SERVICE_NAME when config service_name is blank" + ); + assert_eq!( + find_string_attr(&resource.attributes, "service.version"), + Some("test-version"), + "engine-originated spans should preserve the configured service.version resource attribute" + ); +} From 71f7cb054b51ee96dd9403a22aa8f40c7b3f601f Mon Sep 17 00:00:00 2001 From: alsruf36 Date: Wed, 13 May 2026 23:48:21 +0900 Subject: [PATCH 2/2] fix(observability): forward tee exporter resource identity --- engine/src/workers/observability/otel.rs | 4 + ...otel_tracer_provider_both_resource_test.rs | 196 ++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 engine/tests/otel_tracer_provider_both_resource_test.rs diff --git a/engine/src/workers/observability/otel.rs b/engine/src/workers/observability/otel.rs index 6a7c800044..341ea80545 100644 --- a/engine/src/workers/observability/otel.rs +++ b/engine/src/workers/observability/otel.rs @@ -676,6 +676,10 @@ impl SpanExporter for TeeSpanExporter { fn shutdown_with_timeout(&mut self, timeout: std::time::Duration) -> OTelSdkResult { self.otlp_exporter.shutdown_with_timeout(timeout) } + + fn set_resource(&mut self, resource: &Resource) { + self.otlp_exporter.set_resource(resource); + } } static TRACER_PROVIDER: OnceLock = OnceLock::new(); diff --git a/engine/tests/otel_tracer_provider_both_resource_test.rs b/engine/tests/otel_tracer_provider_both_resource_test.rs new file mode 100644 index 0000000000..2b9ecd5797 --- /dev/null +++ b/engine/tests/otel_tracer_provider_both_resource_test.rs @@ -0,0 +1,196 @@ +// Regression test for iii-hq/iii#1619 — engine-originated spans must +// export a non-empty service.name resource through the `both` exporter path. +// +// This test intentionally lives in its own integration-test binary: init_otel +// stores TRACER_PROVIDER in a process-global OnceLock and installs a global +// OpenTelemetry provider, so sharing a process with other OTEL tests would make +// the assertions order-dependent. + +use std::sync::Arc; +use std::time::Duration; + +use opentelemetry_proto::tonic::collector::trace::v1::{ + ExportTraceServiceRequest, ExportTraceServiceResponse, + trace_service_server::{TraceService, TraceServiceServer}, +}; +use tokio::sync::Mutex; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; +use tracing_subscriber::prelude::*; + +#[derive(Default, Clone)] +struct CapturingTraceService { + received: Arc>>, +} + +#[tonic::async_trait] +impl TraceService for CapturingTraceService { + async fn export( + &self, + request: Request, + ) -> Result, Status> { + self.received.lock().await.push(request.into_inner()); + Ok(Response::new(ExportTraceServiceResponse::default())) + } +} + +struct EnvVarGuard { + name: &'static str, + value: Option, +} + +impl EnvVarGuard { + fn set(name: &'static str, value: &str) -> Self { + let guard = Self { + name, + value: std::env::var_os(name), + }; + unsafe { + std::env::set_var(name, value); + } + guard + } + + fn remove(name: &'static str) -> Self { + let guard = Self { + name, + value: std::env::var_os(name), + }; + unsafe { + std::env::remove_var(name); + } + guard + } +} + +impl Drop for EnvVarGuard { + fn drop(&mut self) { + unsafe { + match &self.value { + Some(value) => std::env::set_var(self.name, value), + None => std::env::remove_var(self.name), + } + } + } +} + +async fn spawn_mock_collector() -> (String, Arc>>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let endpoint = format!("http://{addr}"); + + let service = CapturingTraceService::default(); + let received = service.received.clone(); + + tokio::spawn(async move { + let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener); + Server::builder() + .add_service(TraceServiceServer::new(service)) + .serve_with_incoming(incoming) + .await + .expect("mock OTLP collector failed to start; see tonic transport error above") + }); + + let deadline = std::time::Instant::now() + Duration::from_secs(2); + loop { + match tokio::net::TcpStream::connect(addr).await { + Ok(_) => break, + Err(_) if std::time::Instant::now() < deadline => { + tokio::task::yield_now().await; + } + Err(e) => panic!("mock collector not reachable within 2s: {e}"), + } + } + + (endpoint, received) +} + +fn find_string_attr<'a>( + attrs: &'a [opentelemetry_proto::tonic::common::v1::KeyValue], + key: &str, +) -> Option<&'a str> { + attrs + .iter() + .find(|kv| kv.key == key) + .and_then(|kv| kv.value.as_ref()) + .and_then(|v| v.value.as_ref()) + .and_then(|v| match v { + opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue(s) => { + Some(s.as_str()) + } + _ => None, + }) +} + +fn resource_for_span<'a>( + req: &'a ExportTraceServiceRequest, + span_name: &str, +) -> Option<&'a opentelemetry_proto::tonic::resource::v1::Resource> { + req.resource_spans.iter().find_map(|resource_spans| { + let has_span = resource_spans + .scope_spans + .iter() + .any(|scope_spans| scope_spans.spans.iter().any(|span| span.name == span_name)); + + has_span.then(|| resource_spans.resource.as_ref()).flatten() + }) +} + +#[tokio::test] +async fn both_exporter_forwards_engine_resource_to_otlp_collector() { + let _service_name = EnvVarGuard::set("OTEL_SERVICE_NAME", "iii-engine"); + let _resource_attrs = EnvVarGuard::remove("OTEL_RESOURCE_ATTRIBUTES"); + + let (endpoint, received) = spawn_mock_collector().await; + let config = iii::workers::observability::otel::OtelConfig { + enabled: true, + service_name: String::new(), + service_version: "both-test-version".to_string(), + service_namespace: None, + exporter: iii::workers::observability::otel::ExporterType::Both, + endpoint, + sampling_ratio: 1.0, + memory_max_spans: 10, + }; + + let otel_layer = iii::workers::observability::otel::init_otel(&config) + .expect("OpenTelemetry should initialize for the mock collector"); + let subscriber = tracing_subscriber::registry().with(otel_layer); + + tracing::subscriber::with_default(subscriber, || { + let span = tracing::info_span!("engine-originated-both-test-span"); + let _enter = span.enter(); + tracing::info!("emitting engine-originated span through the both exporter"); + }); + + iii::workers::observability::otel::shutdown_otel(); + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if !received.lock().await.is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + + let captured = received.lock().await; + assert_eq!( + captured.len(), + 1, + "Mock collector should have received exactly one engine-originated ExportTraceServiceRequest" + ); + + let resource = resource_for_span(&captured[0], "engine-originated-both-test-span") + .expect("engine-originated test span should be exported with a Resource block"); + + assert_eq!( + find_string_attr(&resource.attributes, "service.name"), + Some("iii-engine"), + "the both exporter must forward the provider Resource to its inner OTLP exporter" + ); + assert_eq!( + find_string_attr(&resource.attributes, "service.version"), + Some("both-test-version"), + "the both exporter should preserve the configured service.version resource attribute" + ); +}