From 8f3814406010e7e539b5e3b2aa7beb5ce0642fcc Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 09:46:28 -0400 Subject: [PATCH 1/2] fix: report dropped OpenTelemetry spans Signed-off-by: Will Killian --- crates/core/src/observability/otel.rs | 182 +++++++++++++++++- .../src/observability/plugin_component.rs | 12 +- crates/core/src/plugin.rs | 14 +- .../tests/unit/observability/otel_tests.rs | 110 ++++++++++- crates/core/tests/unit/plugin_tests.rs | 2 +- .../observability/opentelemetry.mdx | 14 +- 6 files changed, 309 insertions(+), 25 deletions(-) diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 5999442d1..362ea5ca3 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -17,6 +17,7 @@ use std::borrow::Cow; use std::cell::RefCell; use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::mpsc; use std::sync::{Arc, Mutex}; use std::thread; @@ -42,13 +43,22 @@ use opentelemetry::trace::{ Tracer, TracerProvider as _, }; use opentelemetry::{Context, KeyValue}; -use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig}; +use opentelemetry_otlp::{ + Protocol, SpanExporter as OtlpSpanExporter, WithExportConfig, WithHttpConfig, +}; use opentelemetry_sdk::Resource; +use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult}; use opentelemetry_sdk::trace::{ - IdGenerator, RandomIdGenerator, SdkTracer, SdkTracerProvider, Span, + BatchSpanProcessor, IdGenerator, RandomIdGenerator, SdkTracer, SdkTracerProvider, Span, + SpanData, SpanExporter, SpanProcessor, }; use uuid::Uuid; +use crate::plugin::{ + OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, RuntimeDiagnostic, + record_active_plugin_runtime_diagnostic, +}; + pub(super) const COMPLETED_SPAN_CONTEXT_LIMIT: usize = 4096; use opentelemetry_otlp::WithTonicConfig; @@ -397,6 +407,25 @@ impl Drop for ExporterRuntime { impl OpenTelemetrySubscriber { /// Builds a subscriber backed by a new OTLP tracer provider. pub fn new(config: OpenTelemetryConfig) -> Result { + Self::new_with_runtime_diagnostics(config, None) + } + + pub(crate) fn new_for_plugin( + config: OpenTelemetryConfig, + endpoint_index: usize, + ) -> Result { + Self::new_with_runtime_diagnostics( + config, + Some(format!( + "opentelemetry.endpoints[{endpoint_index}].endpoint" + )), + ) + } + + fn new_with_runtime_diagnostics( + config: OpenTelemetryConfig, + diagnostic_field: Option, + ) -> Result { if config.endpoint.trim().is_empty() { return Err(OpenTelemetryError::ExporterBuild( "endpoint must be a nonblank string".to_string(), @@ -406,7 +435,7 @@ impl OpenTelemetrySubscriber { .map_err(OpenTelemetryError::InvalidAttributeMappings)?; reject_global_header_environment()?; validate_headers(&config.headers)?; - let (provider, runtime) = build_owned_tracer_provider(config.clone())?; + let (provider, runtime) = build_owned_tracer_provider(config.clone(), diagnostic_field)?; Ok(Self::from_tracer_provider_with_scope_and_type( provider, config.instrumentation_scope, @@ -617,6 +646,7 @@ impl OpenTelemetrySubscriber { fn build_owned_tracer_provider( config: OpenTelemetryConfig, + diagnostic_field: Option, ) -> Result<(SdkTracerProvider, ExporterRuntime)> { let (result_sender, result_receiver) = mpsc::sync_channel(1); let (stop_sender, stop_receiver) = mpsc::channel(); @@ -637,7 +667,7 @@ fn build_owned_tracer_provider( }; let provider = { let _guard = runtime.enter(); - build_tracer_provider(&config) + build_tracer_provider(&config, diagnostic_field) }; let keep_runtime_alive = provider.is_ok(); let _ = result_sender.send(provider); @@ -696,10 +726,13 @@ pub(crate) fn validate_headers(headers: &HashMap) -> Result<()> Ok(()) } -fn build_tracer_provider(config: &OpenTelemetryConfig) -> Result { +fn build_tracer_provider( + config: &OpenTelemetryConfig, + diagnostic_field: Option, +) -> Result { let exporter = match config.transport { OtlpTransport::HttpBinary => { - let mut builder = SpanExporter::builder() + let mut builder = OtlpSpanExporter::builder() .with_http() .with_protocol(Protocol::HttpBinary) .with_timeout(config.timeout); @@ -713,7 +746,7 @@ fn build_tracer_provider(config: &OpenTelemetryConfig) -> Result { - let mut builder = SpanExporter::builder() + let mut builder = OtlpSpanExporter::builder() .with_tonic() .with_protocol(Protocol::Grpc) .with_timeout(config.timeout); @@ -754,7 +787,140 @@ fn build_tracer_provider(config: &OpenTelemetryConfig) -> Result { + inner: E, + accepted_spans: Arc, +} + +impl SpanExporter for CountingSpanExporter { + async fn export(&self, batch: Vec) -> OTelSdkResult { + self.accepted_spans + .fetch_add(batch.len() as u64, Ordering::Relaxed); + self.inner.export(batch).await + } + + fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult { + self.inner.shutdown_with_timeout(timeout) + } + + fn force_flush(&self) -> OTelSdkResult { + self.inner.force_flush() + } + + fn set_resource(&mut self, resource: &Resource) { + self.inner.set_resource(resource); + } +} + +#[derive(Debug)] +struct DiagnosticBatchSpanProcessor { + inner: BatchSpanProcessor, + completed_spans: AtomicU64, + accepted_spans: Arc, + endpoint: String, + diagnostic_field: Option, + diagnostic_reported: AtomicBool, +} + +impl DiagnosticBatchSpanProcessor { + fn new( + exporter: E, + endpoint: String, + diagnostic_field: Option, + ) -> Self { + Self::new_with_batch_config( + exporter, + endpoint, + diagnostic_field, + opentelemetry_sdk::trace::BatchConfig::default(), + ) + } + + fn new_with_batch_config( + exporter: E, + endpoint: String, + diagnostic_field: Option, + batch_config: opentelemetry_sdk::trace::BatchConfig, + ) -> Self { + let accepted_spans = Arc::new(AtomicU64::new(0)); + let exporter = CountingSpanExporter { + inner: exporter, + accepted_spans: Arc::clone(&accepted_spans), + }; + Self { + inner: BatchSpanProcessor::builder(exporter) + .with_batch_config(batch_config) + .build(), + completed_spans: AtomicU64::new(0), + accepted_spans, + endpoint, + diagnostic_field, + diagnostic_reported: AtomicBool::new(false), + } + } + + fn record_dropped_spans(&self) -> u64 { + let dropped = self + .completed_spans + .load(Ordering::Relaxed) + .saturating_sub(self.accepted_spans.load(Ordering::Relaxed)); + if dropped == 0 + || self.diagnostic_field.is_none() + || self.diagnostic_reported.swap(true, Ordering::Relaxed) + { + return dropped; + } + record_active_plugin_runtime_diagnostic(RuntimeDiagnostic { + code: "otel.spans_dropped".to_string(), + component: "observability".to_string(), + field: self.diagnostic_field.clone(), + message: format!( + "OpenTelemetry dropped {dropped} spans before export to endpoint {} because the batch queue was full", + self.endpoint + ), + session_id: None, + count: dropped, + }); + dropped + } +} + +impl SpanProcessor for DiagnosticBatchSpanProcessor { + fn on_start(&self, span: &mut Span, cx: &Context) { + self.inner.on_start(span, cx); + } + + fn on_end(&self, span: SpanData) { + self.completed_spans.fetch_add(1, Ordering::Relaxed); + self.inner.on_end(span); + } + + fn force_flush(&self) -> OTelSdkResult { + self.inner.force_flush() + } + + fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult { + let result = self.inner.shutdown_with_timeout(timeout); + if result.is_ok() { + let dropped = self.record_dropped_spans(); + if dropped > 0 && self.diagnostic_field.is_some() { + return Err(OTelSdkError::InternalFailure(format!( + "{OTEL_RUNTIME_DELIVERY_FAILURE_MARKER}: otel.spans_dropped ({dropped})" + ))); + } + } + result + } + + fn set_resource(&mut self, resource: &Resource) { + self.inner.set_resource(resource); + } } fn build_grpc_metadata(headers: &HashMap) -> Result { diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 4374e7da7..2b703ca7a 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -60,17 +60,15 @@ use crate::observability::{ validate_attribute_mappings, }; use crate::plugin::{ - ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, Plugin, PluginComponentSpec, PluginError, - PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior, - apply_global_config_policy, deregister_plugin, register_builtin_plugin, + ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, Plugin, + PluginComponentSpec, PluginError, PluginRegistration, PluginRegistrationContext, + Result as PluginResult, UnsupportedBehavior, apply_global_config_policy, deregister_plugin, + register_builtin_plugin, }; use crate::plugin::{RuntimeDiagnostic, record_active_plugin_runtime_diagnostic}; /// The plugin kind registered by the core crate. pub const OBSERVABILITY_PLUGIN_KIND: &str = "observability"; -/// Identifies teardown errors caused by recoverable ATIF delivery failures. -pub(crate) const ATIF_RUNTIME_DELIVERY_FAILURE_MARKER: &str = "ATIF runtime delivery failures"; - /// Top-level observability component wrapper. /// /// Use this wrapper when constructing a [`PluginComponentSpec`] from Rust @@ -1036,7 +1034,7 @@ fn build_opentelemetry_subscribers( let mut subscribers = Vec::with_capacity(endpoints.len()); for (index, endpoint) in endpoints.into_iter().enumerate() { let subscriber = build_otel_config(index, endpoint).and_then(|config| { - OpenTelemetrySubscriber::new(config) + OpenTelemetrySubscriber::new_for_plugin(config, index) .map(Arc::new) .map_err(observability_registration_error) }); diff --git a/crates/core/src/plugin.rs b/crates/core/src/plugin.rs index 789d7176e..842155706 100644 --- a/crates/core/src/plugin.rs +++ b/crates/core/src/plugin.rs @@ -44,7 +44,6 @@ use crate::api::runtime::{ ToolExecutionFn, ToolInterceptFn, ToolSanitizeFn, }; use crate::api::subscriber::{deregister_subscriber, register_subscriber}; -use crate::observability::plugin_component::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER; pub use nemo_relay_types::plugin::{ConfigDiagnostic, DiagnosticLevel}; pub mod dynamic; @@ -127,6 +126,12 @@ pub enum PluginError { /// Specialized [`Result`](std::result::Result) type for plugin operations. pub type Result = std::result::Result; +/// Identifies teardown errors caused by recoverable ATIF delivery failures. +pub(crate) const ATIF_RUNTIME_DELIVERY_FAILURE_MARKER: &str = "ATIF runtime delivery failures"; +/// Identifies teardown errors caused by recoverable OpenTelemetry delivery failures. +pub(crate) const OTEL_RUNTIME_DELIVERY_FAILURE_MARKER: &str = + "OpenTelemetry runtime delivery failures"; + /// Canonical plugin configuration document. #[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] @@ -2136,9 +2141,10 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { // Runtime delivery failures are reported by an otherwise successful // deregistration callback. They must propagate without treating callback // removal itself as unsafe. - let callbacks_cleared = deregistration_errors - .iter() - .all(|error| error.contains(ATIF_RUNTIME_DELIVERY_FAILURE_MARKER)); + let callbacks_cleared = deregistration_errors.iter().all(|error| { + error.contains(ATIF_RUNTIME_DELIVERY_FAILURE_MARKER) + || error.contains(OTEL_RUNTIME_DELIVERY_FAILURE_MARKER) + }); let deregistration_error = (!deregistration_errors.is_empty()).then(|| { PluginError::RegistrationFailed(format!( "plugin teardown failed: {}", diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 10f21e6ae..279944973 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -25,13 +25,14 @@ use crate::json::Json; use crate::observability::atif::{AtifAgentInfo, AtifExporter, AtifStepExtra}; use crate::observability::{relay_span_id, relay_trace_id}; use opentelemetry::trace::TraceContextExt; -use opentelemetry_sdk::trace::InMemorySpanExporterBuilder; +use opentelemetry_sdk::trace::{BatchConfigBuilder, InMemorySpanExporterBuilder}; use serde_json::json; use std::collections::BTreeSet; use std::collections::HashMap; use std::io::{Read, Write}; use std::net::TcpListener; use std::sync::mpsc; +use std::sync::{Arc, Condvar, Mutex}; use std::thread; use uuid::Uuid; @@ -51,6 +52,53 @@ impl Drop for RestoreThreadScopeStackGuard { } } +#[derive(Clone, Debug, Default)] +struct BlockingSpanExporter { + state: Arc<(Mutex, Condvar)>, +} + +#[derive(Debug, Default)] +struct BlockingExporterState { + export_started: bool, + release_export: bool, +} + +impl BlockingSpanExporter { + fn wait_until_export_starts(&self) { + let (state, changed) = &*self.state; + let guard = state.lock().unwrap(); + let (guard, timeout) = changed + .wait_timeout_while(guard, Duration::from_secs(5), |state| !state.export_started) + .unwrap(); + assert!(guard.export_started && !timeout.timed_out()); + } + + fn release(&self) { + let (state, changed) = &*self.state; + state.lock().unwrap().release_export = true; + changed.notify_all(); + } +} + +impl Drop for BlockingSpanExporter { + fn drop(&mut self) { + self.release(); + } +} + +impl SpanExporter for BlockingSpanExporter { + async fn export(&self, _batch: Vec) -> OTelSdkResult { + let (state, changed) = &*self.state; + let mut state = state.lock().unwrap(); + state.export_started = true; + changed.notify_all(); + while !state.release_export { + state = changed.wait(state).unwrap(); + } + Ok(()) + } +} + fn empty_annotated_response() -> AnnotatedLlmResponse { AnnotatedLlmResponse { id: None, @@ -3062,6 +3110,7 @@ fn provider_builders_cover_success_paths() { .with_resource_attribute("deployment.environment", "test") .with_service_namespace("agents") .with_service_version("1.2.3"), + None, ) .unwrap(); http_provider.force_flush().unwrap(); @@ -3076,6 +3125,64 @@ fn provider_builders_cover_success_paths() { subscriber.shutdown().unwrap(); } +#[test] +fn dropped_spans_are_recorded_in_the_active_plugin_report() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + let _ = crate::plugin::clear_plugin_configuration(); + futures::executor::block_on(crate::plugin::initialize_plugins_exact( + crate::plugin::PluginConfig::default(), + )) + .unwrap(); + + let exporter = BlockingSpanExporter::default(); + let processor = DiagnosticBatchSpanProcessor::new_with_batch_config( + exporter.clone(), + "https://collector.example/v1/traces".to_string(), + Some("opentelemetry.endpoints[2].endpoint".to_string()), + BatchConfigBuilder::default() + .with_max_queue_size(1) + .with_max_export_batch_size(1) + .with_scheduled_delay(Duration::from_secs(60)) + .build(), + ); + let provider = SdkTracerProvider::builder() + .with_span_processor(processor) + .build(); + let tracer = provider.tracer("dropped-span-diagnostic-test"); + + tracer.start("export-in-progress").end(); + exporter.wait_until_export_starts(); + tracer.start("queued").end(); + tracer.start("dropped-1").end(); + tracer.start("dropped-2").end(); + exporter.release(); + let shutdown = provider.shutdown().unwrap_err(); + assert!( + shutdown + .to_string() + .contains(OTEL_RUNTIME_DELIVERY_FAILURE_MARKER) + ); + + let report = crate::plugin::active_plugin_report().unwrap(); + let diagnostic = report + .runtime_diagnostics + .iter() + .find(|diagnostic| diagnostic.code == "otel.spans_dropped") + .unwrap(); + assert_eq!(diagnostic.count, 2); + assert_eq!( + diagnostic.field.as_deref(), + Some("opentelemetry.endpoints[2].endpoint") + ); + assert!( + diagnostic + .message + .contains("https://collector.example/v1/traces") + ); + + crate::plugin::clear_plugin_configuration().unwrap(); +} + #[test] fn grpc_metadata_and_runtime_builder_paths_succeed() { let metadata = build_grpc_metadata(&HashMap::from([( @@ -3097,6 +3204,7 @@ fn grpc_metadata_and_runtime_builder_paths_succeed() { &OpenTelemetryConfig::grpc("grpc-demo") .with_endpoint("http://127.0.0.1:4317") .with_header("authorization", "Bearer token"), + None, ) .unwrap(); provider.force_flush().ok(); diff --git a/crates/core/tests/unit/plugin_tests.rs b/crates/core/tests/unit/plugin_tests.rs index 8f483143f..2dfdea069 100644 --- a/crates/core/tests/unit/plugin_tests.rs +++ b/crates/core/tests/unit/plugin_tests.rs @@ -1526,7 +1526,7 @@ fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { }); Err(PluginError::RegistrationFailed(format!( "{}: atif.remote_delivery_failed (1)", - crate::observability::plugin_component::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER + crate::plugin::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER ))) }), )], diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index 25187fa80..b660c3f85 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -120,14 +120,20 @@ The OpenTelemetry SDK logs first drops a span. It suppresses additional first-drop warnings for that endpoint to avoid a log storm. During graceful shutdown, it logs `BatchSpanProcessor.SpansDropped` with the endpoint processor's exact -`dropped_span_count` and `max_queue_size`. The SDK diagnostics do not include -the Relay endpoint index or URL. +`dropped_span_count` and `max_queue_size`. + +For plugin-managed exporters, Relay also records `otel.spans_dropped` in the +active plugin report's `runtime_diagnostics`. Its `count` is the exact number +of dropped spans, `field` identifies the affected +`opentelemetry.endpoints[N].endpoint`, and `message` includes the configured +endpoint URL. Clearing the plugin reports the delivery failure and retains the +diagnostic for inspection, without disabling later plugin configuration. Increasing `OTEL_BSP_MAX_QUEUE_SIZE` can reduce the risk for a known burst size, but a finite queue does not guarantee lossless telemetry. Always clear -the plugin during graceful shutdown so the SDK can report its final drop count -and attempt to export queued spans. +the plugin during graceful shutdown so Relay can record the final drop count +and the SDK can attempt to export queued spans. ## Endpoint Capacity and Sizing From 302f54b4baf20ecceffa35733b0b44372fc0d0ab Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 12:32:19 -0400 Subject: [PATCH 2/2] fix: preserve OpenTelemetry shutdown failures Signed-off-by: Will Killian --- .../src/observability/plugin_component.rs | 51 +++++++----- crates/core/src/plugin.rs | 16 +++- .../tests/unit/observability/otel_tests.rs | 11 ++- .../observability/plugin_component_tests.rs | 81 ++++++++++++++++++- crates/core/tests/unit/plugin_tests.rs | 53 ++++++++++++ .../observability/opentelemetry.mdx | 9 ++- 6 files changed, 191 insertions(+), 30 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 2b703ca7a..b622f3016 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -60,10 +60,10 @@ use crate::observability::{ validate_attribute_mappings, }; use crate::plugin::{ - ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, Plugin, - PluginComponentSpec, PluginError, PluginRegistration, PluginRegistrationContext, - Result as PluginResult, UnsupportedBehavior, apply_global_config_policy, deregister_plugin, - register_builtin_plugin, + ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, + OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, Plugin, PluginComponentSpec, PluginError, + PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior, + apply_global_config_policy, deregister_plugin, register_builtin_plugin, }; use crate::plugin::{RuntimeDiagnostic, record_active_plugin_runtime_diagnostic}; @@ -1041,7 +1041,7 @@ fn build_opentelemetry_subscribers( match subscriber { Ok(subscriber) => subscribers.push(subscriber), Err(error) => { - if let Some(_rollback_error) = shutdown_opentelemetry_providers(&subscribers) { + if !shutdown_opentelemetry_providers(&subscribers).is_empty() { log::warn!( target: "nemo_relay.plugin", event = "plugin_resource_rollback_failed", @@ -1061,28 +1061,43 @@ fn build_opentelemetry_subscribers( fn shutdown_opentelemetry_subscribers( subscribers: &[Arc], ) -> Option { - let mut first_error = flush_subscribers().err().map(|error| { - observability_registration_error(crate::observability::otel::OpenTelemetryError::Core( - error, - )) - }); - let provider_error = shutdown_opentelemetry_providers(subscribers); - if first_error.is_none() { - first_error = provider_error; + let mut errors = Vec::new(); + if let Err(error) = flush_subscribers() { + errors.push(crate::observability::otel::OpenTelemetryError::Core(error)); } - first_error + errors.extend(shutdown_opentelemetry_providers(subscribers)); + if errors.is_empty() { + return None; + } + + let all_delivery_failures = errors.iter().all(|error| { + error + .to_string() + .contains(OTEL_RUNTIME_DELIVERY_FAILURE_MARKER) + }); + let summary = errors + .into_iter() + .map(|error| error.to_string()) + .collect::>() + .join("; "); + let message = if all_delivery_failures { + format!("{OTEL_RUNTIME_DELIVERY_FAILURE_MARKER}: {summary}") + } else { + format!("OpenTelemetry shutdown failures: {summary}") + }; + Some(PluginError::RegistrationFailed(message)) } fn shutdown_opentelemetry_providers( subscribers: &[Arc], -) -> Option { - let mut first_error = None; +) -> Vec { + let mut errors = Vec::new(); for subscriber in subscribers { if let Err(error) = subscriber.shutdown_provider() { - first_error.get_or_insert_with(|| observability_registration_error(error)); + errors.push(error); } } - first_error + errors } struct AtifDispatcher { diff --git a/crates/core/src/plugin.rs b/crates/core/src/plugin.rs index 842155706..9fbce2c81 100644 --- a/crates/core/src/plugin.rs +++ b/crates/core/src/plugin.rs @@ -2141,10 +2141,9 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { // Runtime delivery failures are reported by an otherwise successful // deregistration callback. They must propagate without treating callback // removal itself as unsafe. - let callbacks_cleared = deregistration_errors.iter().all(|error| { - error.contains(ATIF_RUNTIME_DELIVERY_FAILURE_MARKER) - || error.contains(OTEL_RUNTIME_DELIVERY_FAILURE_MARKER) - }); + let callbacks_cleared = deregistration_errors + .iter() + .all(|error| is_runtime_delivery_failure(error)); let deregistration_error = (!deregistration_errors.is_empty()).then(|| { PluginError::RegistrationFailed(format!( "plugin teardown failed: {}", @@ -2175,6 +2174,15 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { } } +fn is_runtime_delivery_failure(error: &str) -> bool { + [ + ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, + OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, + ] + .iter() + .any(|marker| error.contains(&format!("registration failed: {marker}:"))) +} + pub(crate) fn plugin_configuration_is_active() -> Result { ACTIVE_PLUGIN_CONFIGURATION .lock() diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 279944973..88bf5ecea 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -44,6 +44,14 @@ impl Drop for ResetPricingResolverGuard { } } +struct ClearPluginConfigurationGuard; + +impl Drop for ClearPluginConfigurationGuard { + fn drop(&mut self) { + let _ = crate::plugin::clear_plugin_configuration(); + } +} + struct RestoreThreadScopeStackGuard(ThreadScopeStackBinding); impl Drop for RestoreThreadScopeStackGuard { @@ -3129,6 +3137,7 @@ fn provider_builders_cover_success_paths() { fn dropped_spans_are_recorded_in_the_active_plugin_report() { let _guard = crate::observability::test_mutex().lock().unwrap(); let _ = crate::plugin::clear_plugin_configuration(); + let _clear_guard = ClearPluginConfigurationGuard; futures::executor::block_on(crate::plugin::initialize_plugins_exact( crate::plugin::PluginConfig::default(), )) @@ -3179,8 +3188,6 @@ fn dropped_spans_are_recorded_in_the_active_plugin_report() { .message .contains("https://collector.example/v1/traces") ); - - crate::plugin::clear_plugin_configuration().unwrap(); } #[test] diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 494b7b731..f6e4eb362 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -20,11 +20,36 @@ use serde_json::json; use std::fs; use std::io::{Read, Write}; use std::net::TcpListener; -use std::sync::mpsc; +use std::sync::Arc; #[cfg(feature = "atof-streaming")] -use std::sync::{Arc, Mutex}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::mpsc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +#[derive(Debug)] +struct ShutdownFailureSpanProcessor { + message: String, + shutdown_calls: Arc, +} + +impl opentelemetry_sdk::trace::SpanProcessor for ShutdownFailureSpanProcessor { + fn on_start(&self, _span: &mut opentelemetry_sdk::trace::Span, _cx: &opentelemetry::Context) {} + + fn on_end(&self, _span: opentelemetry_sdk::trace::SpanData) {} + + fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult { + Ok(()) + } + + fn shutdown_with_timeout(&self, _timeout: Duration) -> opentelemetry_sdk::error::OTelSdkResult { + self.shutdown_calls.fetch_add(1, Ordering::SeqCst); + Err(opentelemetry_sdk::error::OTelSdkError::InternalFailure( + self.message.clone(), + )) + } +} + fn temp_dir(prefix: &str) -> PathBuf { let id = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -2846,6 +2871,58 @@ fn opentelemetry_shutdown_helper_attempts_every_constructed_endpoint() { } } +#[test] +fn opentelemetry_shutdown_helper_retains_every_endpoint_failure() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dropped_calls = Arc::new(AtomicUsize::new(0)); + let timeout_calls = Arc::new(AtomicUsize::new(0)); + let subscribers = [ + ( + format!("{OTEL_RUNTIME_DELIVERY_FAILURE_MARKER}: otel.spans_dropped (2)"), + Arc::clone(&dropped_calls), + ), + ( + "endpoint shutdown timed out".to_string(), + Arc::clone(&timeout_calls), + ), + ] + .into_iter() + .enumerate() + .map(|(index, (message, shutdown_calls))| { + let processor = ShutdownFailureSpanProcessor { + message, + shutdown_calls, + }; + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_span_processor(processor) + .build(); + Arc::new(OpenTelemetrySubscriber::from_tracer_provider( + provider, + format!("shutdown-failure-{index}"), + )) + }) + .collect::>(); + + let error = shutdown_opentelemetry_subscribers(&subscribers) + .expect("mixed endpoint shutdown failures should be reported") + .to_string(); + + assert_eq!(dropped_calls.load(Ordering::SeqCst), 1); + assert_eq!(timeout_calls.load(Ordering::SeqCst), 1); + assert!( + error.contains(OTEL_RUNTIME_DELIVERY_FAILURE_MARKER), + "{error}" + ); + assert!(error.contains("endpoint shutdown timed out"), "{error}"); + assert!( + !error.contains(&format!( + "registration failed: {OTEL_RUNTIME_DELIVERY_FAILURE_MARKER}:" + )), + "mixed failures must not use the recoverable marker prefix: {error}" + ); +} + #[test] fn opentelemetry_delivery_continues_after_an_endpoint_panics() { let delivered = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); diff --git a/crates/core/tests/unit/plugin_tests.rs b/crates/core/tests/unit/plugin_tests.rs index 2dfdea069..816f2ab8d 100644 --- a/crates/core/tests/unit/plugin_tests.rs +++ b/crates/core/tests/unit/plugin_tests.rs @@ -1549,6 +1549,59 @@ fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { reset_global(); } +#[test] +fn test_opentelemetry_delivery_failure_allows_later_plugin_configuration() { + let _guard = lock_runtime_owner(); + reset_global(); + store_active_plugin_configuration( + PluginConfig::default(), + ConfigReport::default(), + vec![PluginRegistration::new( + "fixture", + "opentelemetry-shutdown", + Box::new(|| { + Err(PluginError::RegistrationFailed(format!( + "{}: otel.spans_dropped (2)", + crate::plugin::OTEL_RUNTIME_DELIVERY_FAILURE_MARKER + ))) + }), + )], + ) + .unwrap(); + + let outcome = clear_plugin_configuration_inner(); + assert!(outcome.callbacks_cleared); + assert!(outcome.result.is_err()); + reset_global(); +} + +#[test] +fn test_mixed_opentelemetry_shutdown_failure_blocks_later_configuration() { + let _guard = lock_runtime_owner(); + reset_global(); + store_active_plugin_configuration( + PluginConfig::default(), + ConfigReport::default(), + vec![PluginRegistration::new( + "fixture", + "opentelemetry-shutdown", + Box::new(|| { + Err(PluginError::RegistrationFailed(format!( + "OpenTelemetry shutdown failures: provider error: {}: otel.spans_dropped (2); endpoint shutdown timed out", + crate::plugin::OTEL_RUNTIME_DELIVERY_FAILURE_MARKER + ))) + }), + )], + ) + .unwrap(); + + let outcome = clear_plugin_configuration_inner(); + assert!(!outcome.callbacks_cleared); + let error = outcome.result.unwrap_err().to_string(); + assert!(error.contains("endpoint shutdown timed out"), "{error}"); + reset_global(); +} + #[test] fn test_legacy_clear_retains_mutation_owner_after_incomplete_teardown() { let _guard = lock_runtime_owner(); diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index b660c3f85..7b21893f2 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -122,17 +122,18 @@ endpoint to avoid a log storm. During graceful shutdown, it logs `BatchSpanProcessor.SpansDropped` with the endpoint processor's exact `dropped_span_count` and `max_queue_size`. -For plugin-managed exporters, Relay also records `otel.spans_dropped` in the +For plugin-managed exporters, NeMo Relay also records `otel.spans_dropped` in the active plugin report's `runtime_diagnostics`. Its `count` is the exact number of dropped spans, `field` identifies the affected `opentelemetry.endpoints[N].endpoint`, and `message` includes the configured -endpoint URL. Clearing the plugin reports the delivery failure and retains the -diagnostic for inspection, without disabling later plugin configuration. +endpoint URL. If spans were dropped, clearing the plugin returns a delivery +failure error and retains the diagnostic for inspection. This error does not +disable later plugin configuration. Increasing `OTEL_BSP_MAX_QUEUE_SIZE` can reduce the risk for a known burst size, but a finite queue does not guarantee lossless telemetry. Always clear -the plugin during graceful shutdown so Relay can record the final drop count +the plugin during graceful shutdown so NeMo Relay can record the final drop count and the SDK can attempt to export queued spans. ## Endpoint Capacity and Sizing