From db3658c4b1178fef483d85383cb48969c2a00385 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Wed, 5 Aug 2026 15:53:35 -0400 Subject: [PATCH 1/2] feat: configure OpenTelemetry batching per endpoint Signed-off-by: Will Killian --- crates/core/src/observability/otel.rs | 68 ++++++++--- .../src/observability/plugin_component.rs | 107 +++++++++++++++++- .../tests/unit/observability/otel_tests.rs | 5 +- .../observability/plugin_component_tests.rs | 107 ++++++++++++++++++ crates/node/observability.d.ts | 7 +- .../node/tests/observability_plugin_tests.mjs | 11 +- docs/about-nemo-relay/release-notes/index.mdx | 7 +- .../observability/configuration.mdx | 2 +- .../observability/opentelemetry.mdx | 41 +++++-- go/nemo_relay/observability_plugin.go | 3 + go/nemo_relay/observability_plugin_test.go | 28 +++++ python/nemo_relay/observability.py | 6 + python/nemo_relay/observability.pyi | 3 + python/tests/test_observability_plugin.py | 6 + 14 files changed, 360 insertions(+), 41 deletions(-) diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 5a80992de..521d5f1bc 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -49,8 +49,8 @@ use opentelemetry_otlp::{ use opentelemetry_sdk::Resource; use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult}; use opentelemetry_sdk::trace::{ - BatchSpanProcessor, IdGenerator, RandomIdGenerator, SdkTracer, SdkTracerProvider, Span, - SpanData, SpanExporter, SpanProcessor, + BatchConfigBuilder, BatchSpanProcessor, IdGenerator, RandomIdGenerator, SdkTracer, + SdkTracerProvider, Span, SpanData, SpanExporter, SpanProcessor, }; use uuid::Uuid; @@ -197,6 +197,9 @@ pub struct OpenTelemetryConfig { attribute_mappings: Vec, timeout: Duration, transport: OtlpTransport, + max_queue_size: Option, + max_export_batch_size: Option, + scheduled_delay: Option, } impl OpenTelemetryConfig { @@ -215,6 +218,9 @@ impl OpenTelemetryConfig { attribute_mappings: Vec::new(), timeout: Duration::from_secs(3), transport: OtlpTransport::HttpBinary, + max_queue_size: None, + max_export_batch_size: None, + scheduled_delay: None, } } @@ -292,6 +298,33 @@ impl OpenTelemetryConfig { self } + /// Overrides the batch processor queue size for this endpoint. + pub(crate) fn with_max_queue_size(mut self, max_queue_size: usize) -> Self { + self.max_queue_size = Some(max_queue_size); + self + } + + /// Overrides the maximum export batch size for this endpoint. + pub(crate) fn with_max_export_batch_size(mut self, max_export_batch_size: usize) -> Self { + self.max_export_batch_size = Some(max_export_batch_size); + self + } + + /// Overrides the maximum delay before exporting a non-full batch. + pub(crate) fn with_scheduled_delay(mut self, scheduled_delay: Duration) -> Self { + self.scheduled_delay = Some(scheduled_delay); + self + } + + #[cfg(test)] + pub(crate) fn batch_overrides(&self) -> (Option, Option, Option) { + ( + self.max_queue_size, + self.max_export_batch_size, + self.scheduled_delay, + ) + } + /// Sets the service namespace resource attribute. pub fn with_service_namespace(mut self, namespace: impl Into) -> Self { self.service_namespace = Some(namespace.into()); @@ -787,8 +820,22 @@ fn build_tracer_provider( .with_max_attributes_per_span(u32::MAX) .with_max_attributes_per_event(u32::MAX); - let processor = - DiagnosticBatchSpanProcessor::new(exporter, config.endpoint.clone(), diagnostic_field); + let mut batch_config = BatchConfigBuilder::default(); + if let Some(max_queue_size) = config.max_queue_size { + batch_config = batch_config.with_max_queue_size(max_queue_size); + } + if let Some(max_export_batch_size) = config.max_export_batch_size { + batch_config = batch_config.with_max_export_batch_size(max_export_batch_size); + } + if let Some(scheduled_delay) = config.scheduled_delay { + batch_config = batch_config.with_scheduled_delay(scheduled_delay); + } + let processor = DiagnosticBatchSpanProcessor::new_with_batch_config( + exporter, + config.endpoint.clone(), + diagnostic_field, + batch_config.build(), + ); Ok(builder.with_span_processor(processor).build()) } @@ -829,19 +876,6 @@ struct DiagnosticBatchSpanProcessor { } 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, diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 32101a29c..ddaf46854 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -206,9 +206,18 @@ pub struct OpenTelemetryEndpointConfig { /// Instrumentation scope name. #[serde(default = "default_otel_instrumentation_scope")] pub instrumentation_scope: String, - /// Export timeout in milliseconds. + /// OTLP request timeout in milliseconds. #[serde(default = "default_timeout_millis")] pub timeout_millis: u64, + /// Maximum completed spans buffered before the endpoint drops new spans. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_queue_size: Option, + /// Maximum spans exported in one batch. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_export_batch_size: Option, + /// Maximum delay before exporting a non-full batch, in milliseconds. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scheduled_delay_millis: Option, } /// Multi-sink ATOF JSONL exporter config. @@ -531,6 +540,14 @@ impl EditorConfig for OpenTelemetryEndpointConfig { otel_editor_field("service_version", EditorFieldKind::String, &[], true), otel_editor_field("instrumentation_scope", EditorFieldKind::String, &[], false), otel_editor_field("timeout_millis", EditorFieldKind::Integer, &[], false), + otel_editor_field("max_queue_size", EditorFieldKind::Integer, &[], true), + otel_editor_field("max_export_batch_size", EditorFieldKind::Integer, &[], true), + otel_editor_field( + "scheduled_delay_millis", + EditorFieldKind::Integer, + &[], + true, + ), otel_editor_field("headers", EditorFieldKind::StringMap, &[], false), otel_editor_field("header_env", EditorFieldKind::StringMap, &[], false), otel_editor_field( @@ -1976,6 +1993,7 @@ fn build_otel_config( } }; validate_otel_header_env(index, §ion)?; + validate_otel_batch_config(index, §ion)?; let mut config = CoreOpenTelemetryConfig::new(section.otel_type, section.endpoint) .with_transport(transport) .with_service_name(section.service_name) @@ -1984,6 +2002,15 @@ fn build_otel_config( .with_mark_projection(section.mark_projection) .with_mark_exclude_names(section.mark_exclude_names) .with_attribute_mappings(section.attribute_mappings); + if let Some(max_queue_size) = section.max_queue_size { + config = config.with_max_queue_size(max_queue_size); + } + if let Some(max_export_batch_size) = section.max_export_batch_size { + config = config.with_max_export_batch_size(max_export_batch_size); + } + if let Some(scheduled_delay_millis) = section.scheduled_delay_millis { + config = config.with_scheduled_delay(Duration::from_millis(scheduled_delay_millis)); + } if let Some(namespace) = section.service_namespace { config = config.with_service_namespace(namespace); } @@ -2000,6 +2027,36 @@ fn build_otel_config( Ok(config) } +fn validate_otel_batch_config( + index: usize, + section: &OpenTelemetryEndpointConfig, +) -> PluginResult<()> { + for (field, value) in [ + ("max_queue_size", section.max_queue_size), + ("max_export_batch_size", section.max_export_batch_size), + ] { + if value == Some(0) { + return Err(PluginError::InvalidConfig(format!( + "OpenTelemetry endpoints[{index}].{field} must be greater than 0" + ))); + } + } + if section.scheduled_delay_millis == Some(0) { + return Err(PluginError::InvalidConfig(format!( + "OpenTelemetry endpoints[{index}].scheduled_delay_millis must be greater than 0" + ))); + } + if matches!( + (section.max_export_batch_size, section.max_queue_size), + (Some(batch), Some(queue)) if batch > queue + ) { + return Err(PluginError::InvalidConfig(format!( + "OpenTelemetry endpoints[{index}].max_export_batch_size must be less than or equal to max_queue_size" + ))); + } + Ok(()) +} + fn validate_otel_header_env( index: usize, section: &OpenTelemetryEndpointConfig, @@ -2238,6 +2295,9 @@ fn validate_opentelemetry_endpoint_fields( "service_version", "instrumentation_scope", "timeout_millis", + "max_queue_size", + "max_export_batch_size", + "scheduled_delay_millis", ]; const REMOVED: &[&str] = &["semantic_selector", "capture_content"]; let Some(endpoints) = opentelemetry.get("endpoints").and_then(Json::as_array) else { @@ -2416,6 +2476,7 @@ fn validate_opentelemetry_section( error, ); } + validate_opentelemetry_batch_config(diagnostics, policy, index, endpoint); validate_opentelemetry_headers(diagnostics, policy, index, endpoint); } for error in opentelemetry_destination_collision_errors(§ion.endpoints) { @@ -2430,6 +2491,50 @@ fn validate_opentelemetry_section( validate_opentelemetry_feature_support(diagnostics, policy, section); } +fn validate_opentelemetry_batch_config( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + index: usize, + endpoint: &OpenTelemetryEndpointConfig, +) { + for (field, is_zero) in [ + ("max_queue_size", endpoint.max_queue_size == Some(0)), + ( + "max_export_batch_size", + endpoint.max_export_batch_size == Some(0), + ), + ( + "scheduled_delay_millis", + endpoint.scheduled_delay_millis == Some(0), + ), + ] { + if is_zero { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("opentelemetry".to_string()), + Some(format!("endpoints[{index}].{field}")), + format!("OpenTelemetry endpoint {field} must be greater than 0"), + ); + } + } + if matches!( + (endpoint.max_export_batch_size, endpoint.max_queue_size), + (Some(batch), Some(queue)) if batch > queue + ) { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("opentelemetry".to_string()), + Some(format!("endpoints[{index}].max_export_batch_size")), + "OpenTelemetry endpoint max_export_batch_size must be less than or equal to max_queue_size" + .to_string(), + ); + } +} + struct OpenTelemetryDestinationCollision { index: usize, message: String, diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 1d8172947..ebbd57e2b 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -3427,7 +3427,10 @@ fn provider_builders_cover_success_paths() { .with_header("authorization", "Bearer token") .with_resource_attribute("deployment.environment", "test") .with_service_namespace("agents") - .with_service_version("1.2.3"), + .with_service_version("1.2.3") + .with_max_queue_size(16) + .with_max_export_batch_size(4) + .with_scheduled_delay(Duration::from_millis(10)), None, ) .unwrap(); diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index abddf29be..e12ff0d09 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -299,6 +299,15 @@ fn editor_schema_tracks_observability_config_types() { .kind, EditorFieldKind::StringMap ); + for field in [ + "max_queue_size", + "max_export_batch_size", + "scheduled_delay_millis", + ] { + let field = otlp_endpoint_schema.field(field).expect("batch field"); + assert_eq!(field.kind, EditorFieldKind::Integer); + assert!(field.optional); + } assert_eq!( default_opentelemetry_endpoint_editor_value(), json!({ @@ -453,6 +462,9 @@ fn default_config_and_component_conversion_cover_public_shape() { service_version: None, instrumentation_scope: default_otel_instrumentation_scope(), timeout_millis: default_timeout_millis(), + max_queue_size: None, + max_export_batch_size: None, + scheduled_delay_millis: None, headers: HashMap::new(), header_env: HashMap::new(), resource_attributes: HashMap::new(), @@ -473,6 +485,30 @@ fn default_config_and_component_conversion_cover_public_shape() { assert!(generic.enabled); assert_eq!(generic.config["version"], json!(3)); assert_eq!(generic.config["atif"]["agent_name"], json!("NeMo Relay")); + let serialized_endpoint = &generic.config["opentelemetry"]["endpoints"][0]; + for field in [ + "max_queue_size", + "max_export_batch_size", + "scheduled_delay_millis", + ] { + assert!(serialized_endpoint.get(field).is_none()); + } + + assert_endpoint_batch_fields_deserialize(); +} + +fn assert_endpoint_batch_fields_deserialize() { + let endpoint: OpenTelemetryEndpointConfig = serde_json::from_value(json!({ + "type": "full", + "endpoint": "http://localhost:4318/v1/traces", + "max_queue_size": 4096, + "max_export_batch_size": 256, + "scheduled_delay_millis": 750, + })) + .unwrap(); + assert_eq!(endpoint.max_queue_size, Some(4096)); + assert_eq!(endpoint.max_export_batch_size, Some(256)); + assert_eq!(endpoint.scheduled_delay_millis, Some(750)); } fn assert_default_stream_sink_shape() { @@ -537,6 +573,9 @@ fn opentelemetry_endpoint_header_env_is_resolved_and_snapshotted() { service_version: None, instrumentation_scope: default_otel_instrumentation_scope(), timeout_millis: default_timeout_millis(), + max_queue_size: None, + max_export_batch_size: None, + scheduled_delay_millis: None, headers: HashMap::new(), header_env: HashMap::from([("authorization".to_string(), variable.to_string())]), resource_attributes: HashMap::new(), @@ -561,6 +600,9 @@ fn test_opentelemetry_endpoint() -> OpenTelemetryEndpointConfig { service_version: None, instrumentation_scope: default_otel_instrumentation_scope(), timeout_millis: default_timeout_millis(), + max_queue_size: None, + max_export_batch_size: None, + scheduled_delay_millis: None, headers: HashMap::new(), header_env: HashMap::new(), resource_attributes: HashMap::new(), @@ -607,6 +649,39 @@ fn build_otel_config_rejects_each_activation_only_invalid_value() { .insert("authorization".to_string(), variable.to_string()); assert!(build_otel_config(3, endpoint).is_err()); unsafe { std::env::remove_var(variable) }; + + let mut endpoint = test_opentelemetry_endpoint(); + endpoint.max_queue_size = Some(0); + assert!(build_otel_config(4, endpoint).is_err()); + + let mut endpoint = test_opentelemetry_endpoint(); + endpoint.max_export_batch_size = Some(0); + assert!(build_otel_config(4, endpoint).is_err()); + + let mut endpoint = test_opentelemetry_endpoint(); + endpoint.scheduled_delay_millis = Some(0); + assert!(build_otel_config(4, endpoint).is_err()); + + let mut endpoint = test_opentelemetry_endpoint(); + endpoint.max_queue_size = Some(8); + endpoint.max_export_batch_size = Some(9); + assert!(build_otel_config(4, endpoint).is_err()); +} + +#[test] +fn build_otel_config_carries_endpoint_batch_overrides() { + let mut endpoint = test_opentelemetry_endpoint(); + endpoint.max_queue_size = Some(4096); + endpoint.max_export_batch_size = Some(256); + endpoint.scheduled_delay_millis = Some(750); + let config = build_otel_config(0, endpoint).unwrap(); + assert_eq!( + config.batch_overrides(), + (Some(4096), Some(256), Some(Duration::from_millis(750))) + ); + + let config = build_otel_config(1, test_opentelemetry_endpoint()).unwrap(); + assert_eq!(config.batch_overrides(), (None, None, None)); } #[test] @@ -657,6 +732,32 @@ fn validate_opentelemetry_section_reports_empty_and_malformed_endpoints() { "missing diagnostic for {field}: {diagnostics:?}" ); } + + let mut endpoint = test_opentelemetry_endpoint(); + endpoint.max_queue_size = Some(0); + endpoint.max_export_batch_size = Some(2); + endpoint.scheduled_delay_millis = Some(0); + diagnostics.clear(); + validate_opentelemetry_section( + &mut diagnostics, + &policy, + &OpenTelemetrySectionConfig { + enabled: true, + endpoints: vec![endpoint], + }, + ); + for field in [ + "endpoints[0].max_queue_size", + "endpoints[0].max_export_batch_size", + "endpoints[0].scheduled_delay_millis", + ] { + assert!( + diagnostics + .iter() + .any(|diagnostic| diagnostic.field.as_deref() == Some(field)), + "missing diagnostic for {field}: {diagnostics:?}" + ); + } } #[test] @@ -826,6 +927,9 @@ fn opentelemetry_endpoint_accepts_legacy_projection_controls_and_rejects_unknown "mark_projection": "tool", "mark_exclude_names": ["notification"], "attribute_mappings": [{"key": "nemo_relay.model_name", "alias": "model.alias"}], + "max_queue_size": 4096, + "max_export_batch_size": 256, + "scheduled_delay_millis": 750, "capture_content": true }] } @@ -851,6 +955,9 @@ fn opentelemetry_endpoint_accepts_legacy_projection_controls_and_rejects_unknown Some("endpoints[0].mark_projection") | Some("endpoints[0].mark_exclude_names") | Some("endpoints[0].attribute_mappings") + | Some("endpoints[0].max_queue_size") + | Some("endpoints[0].max_export_batch_size") + | Some("endpoints[0].scheduled_delay_millis") ) })); } diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts index 214954b5e..f5f1e9831 100644 --- a/crates/node/observability.d.ts +++ b/crates/node/observability.d.ts @@ -81,6 +81,9 @@ export interface OpenTelemetryEndpointConfig { service_version?: string; instrumentation_scope?: string; timeout_millis?: number; + max_queue_size?: number; + max_export_batch_size?: number; + scheduled_delay_millis?: number; } export interface OpenTelemetrySectionConfig { @@ -113,9 +116,7 @@ export declare function atifConfig(config?: AtifConfig): AtifConfig; /** Create one typed OpenTelemetry endpoint. */ export declare function openTelemetryEndpoint(config: OpenTelemetryEndpointConfig): OpenTelemetryEndpointConfig; /** Create multi-endpoint OpenTelemetry settings. */ -export declare function openTelemetryConfig( - config?: OpenTelemetrySectionConfig, -): OpenTelemetrySectionConfig; +export declare function openTelemetryConfig(config?: OpenTelemetrySectionConfig): OpenTelemetrySectionConfig; /** Wrap observability config as a top-level plugin component. */ export declare function ComponentSpec( config: Config, diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs index 59e5d477c..04e29ec5d 100644 --- a/crates/node/tests/observability_plugin_tests.mjs +++ b/crates/node/tests/observability_plugin_tests.mjs @@ -36,6 +36,9 @@ describe('observability plugin helpers', () => { type: 'gen_ai', endpoint: 'http://localhost:4318/v1/traces', header_env: { authorization: 'OTEL_AUTHORIZATION' }, + max_queue_size: 4096, + max_export_batch_size: 256, + scheduled_delay_millis: 750, }), { type: 'gen_ai', @@ -47,6 +50,9 @@ describe('observability plugin helpers', () => { service_name: 'unknown_service', instrumentation_scope: 'opentelemetry', timeout_millis: 3000, + max_queue_size: 4096, + max_export_batch_size: 256, + scheduled_delay_millis: 750, }, ); @@ -61,10 +67,7 @@ describe('observability plugin helpers', () => { () => observability.openTelemetryEndpoint({ type: 'invalid', endpoint: 'http://localhost' }), /type must be/, ); - assert.throws( - () => observability.openTelemetryEndpoint({ type: 'full', endpoint: ' ' }), - /nonblank/, - ); + assert.throws(() => observability.openTelemetryEndpoint({ type: 'full', endpoint: ' ' }), /nonblank/); assert.equal(plugin.listKinds().includes(observability.OBSERVABILITY_PLUGIN_KIND), true); const report = plugin.validate({ version: 1, diff --git a/docs/about-nemo-relay/release-notes/index.mdx b/docs/about-nemo-relay/release-notes/index.mdx index 3f1b3137c..40228bee6 100644 --- a/docs/about-nemo-relay/release-notes/index.mdx +++ b/docs/about-nemo-relay/release-notes/index.mdx @@ -93,9 +93,10 @@ Migration guidance for upgrading from 0.7 to 0.8 will be added to the endpoint queue drops completed spans without applying backpressure and can leave an incomplete trace, including a missing root span. The SDK warns on the first drop and reports the exact dropped-span count during graceful - shutdown. Increase `OTEL_BSP_MAX_QUEUE_SIZE` before plugin activation to - reduce the risk for a known burst size; endpoint-specific batch sizing is not - available, and a larger finite queue does not guarantee lossless telemetry. + shutdown. Configure `max_queue_size`, `max_export_batch_size`, and + `scheduled_delay_millis` independently on each endpoint, or use the standard + `OTEL_BSP_*` environment variables as process-wide fallbacks. A larger finite + queue does not guarantee lossless telemetry. - Operational logging configuration and sink lifecycle are available, but broad operational log coverage across commands is not yet available. diff --git a/docs/configure-plugins/observability/configuration.mdx b/docs/configure-plugins/observability/configuration.mdx index 88b8b61ca..bf8ca15aa 100644 --- a/docs/configure-plugins/observability/configuration.mdx +++ b/docs/configure-plugins/observability/configuration.mdx @@ -98,7 +98,7 @@ unreachable collector can delay `plugin.clear()` or process shutdown by approximately the OpenTelemetry SDK's five-second shutdown bound plus one endpoint export timeout. Use each endpoint's `timeout_millis` value to bound that export attempt. This timeout behavior does not make a full batch queue -lossless; refer to [OpenTelemetry](/configure-plugins/observability/opentelemetry#batch-processor-environment-variables) +lossless; refer to [OpenTelemetry](/configure-plugins/observability/opentelemetry#batch-processor-configuration) for queue sizing and drop-warning behavior. Top-level component `config` lists concatenate across configuration layers, diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index edfc432d7..8225769a4 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -64,6 +64,9 @@ type = "gen_ai" endpoint = "http://localhost:4318/v1/traces" transport = "http_binary" service_name = "agent-service" +max_queue_size = 4096 +max_export_batch_size = 512 +scheduled_delay_millis = 1000 [components.config.opentelemetry.endpoints.header_env] authorization = "OTEL_AUTHORIZATION" @@ -90,7 +93,10 @@ work or delivery to the other exporters. | `service_namespace` | Omitted | Optional `service.namespace`. | | `service_version` | Omitted | Optional `service.version`. | | `instrumentation_scope` | `opentelemetry` | Instrumentation scope name. | -| `timeout_millis` | `3000` | Export timeout. | +| `timeout_millis` | `3000` | OTLP request timeout. | +| `max_queue_size` | Environment or `2048` | Maximum completed spans buffered before this endpoint drops new spans. | +| `max_export_batch_size` | Environment or `512` | Maximum spans exported in one batch; capped at the effective queue size. | +| `scheduled_delay_millis` | Environment or `5000` ms | Maximum delay before this endpoint exports a non-full batch. A full batch exports sooner. | | `headers` | `{}` | String-to-string exporter headers. | | `header_env` | `{}` | Header names mapped to environment variable names containing secret values. | | `resource_attributes` | `{}` | String-to-string resource attributes. | @@ -98,11 +104,16 @@ work or delivery to the other exporters. | `mark_exclude_names` | `["llm.chunk"]` | Mark names excluded from `full` and `openinference` projection. | | `attribute_mappings` | `[]` | `{ key, alias }` copies applied by `full` and `openinference` projection. | -## Batch Processor Environment Variables +## Batch Processor Configuration -OpenTelemetry's standard batch processor settings apply process-wide to every -configured endpoint. Set them before the plugin activates; endpoint-specific -batch sizing is not supported. +Configure batch processing independently on each endpoint with +`max_queue_size`, `max_export_batch_size`, and `scheduled_delay_millis`. +When an endpoint omits a field, the corresponding standard OpenTelemetry +environment variable applies process-wide. If neither is set, the SDK default +applies. + +The precedence for each setting is endpoint value, then environment variable, +then SDK default. Set environment variables before the plugin activates. | Variable | Default | Notes | |---|---:|---| @@ -110,9 +121,16 @@ batch sizing is not supported. | `OTEL_BSP_MAX_EXPORT_BATCH_SIZE` | `512` | Maximum spans exported in one batch; capped at the queue size. | | `OTEL_BSP_SCHEDULE_DELAY` | `5000` ms | Maximum delay before exporting a non-full batch. | -Use positive integer values. The SDK falls back to its defaults for malformed -values but accepts zero; do not use zero values. Queue capacity counts spans, -not bytes. +Endpoint values must be positive integers. If both endpoint size fields are +set, `max_export_batch_size` must not exceed `max_queue_size`. When one size is +inherited, the SDK caps the effective batch size at the effective queue size. +The SDK falls back to its default for malformed environment values. Queue and +batch sizes count spans, not bytes. + +Relay's thread-based batch processor exports serially, so it does not expose +the SDK's concurrent-export setting. It also does not expose a separate batch +processor export timeout; use the endpoint's `timeout_millis` to bound each +OTLP request. A full queue drops completed spans instead of applying backpressure to @@ -135,9 +153,10 @@ 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 NeMo Relay can record the final drop count +Increasing an endpoint's `max_queue_size`, or the process-wide +`OTEL_BSP_MAX_QUEUE_SIZE` fallback, 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 NeMo Relay can record the final drop count and the SDK can attempt to export queued spans. ## Endpoint Capacity and Sizing diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index a52c3237c..b9c40f0f2 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -39,6 +39,9 @@ type ObservabilityOpenTelemetryEndpointConfig struct { ServiceVersion string `json:"service_version,omitempty"` InstrumentationScope string `json:"instrumentation_scope,omitempty"` TimeoutMillis uint64 `json:"timeout_millis,omitempty"` + MaxQueueSize *uint64 `json:"max_queue_size,omitempty"` + MaxExportBatchSize *uint64 `json:"max_export_batch_size,omitempty"` + ScheduledDelayMillis *uint64 `json:"scheduled_delay_millis,omitempty"` } // ObservabilityAtofConfig configures filesystem-backed raw ATOF JSONL export. diff --git a/go/nemo_relay/observability_plugin_test.go b/go/nemo_relay/observability_plugin_test.go index 2b65de7c0..9061ddb70 100644 --- a/go/nemo_relay/observability_plugin_test.go +++ b/go/nemo_relay/observability_plugin_test.go @@ -71,6 +71,12 @@ func TestObservabilityConfigHelpers(t *testing.T) { NewObservabilityOpenTelemetryEndpointConfig(OpenTelemetryTypeFull, "http://localhost:4318/v1/traces"), } otel.Endpoints[0].HeaderEnv["authorization"] = "OTEL_AUTHORIZATION" + maxQueueSize := uint64(4096) + maxExportBatchSize := uint64(256) + scheduledDelayMillis := uint64(750) + otel.Endpoints[0].MaxQueueSize = &maxQueueSize + otel.Endpoints[0].MaxExportBatchSize = &maxExportBatchSize + otel.Endpoints[0].ScheduledDelayMillis = &scheduledDelayMillis config.Atof = &atof config.Atif = &atif @@ -138,6 +144,28 @@ func assertWrappedObservabilityConfig(t *testing.T, wrapped PluginComponentSpec) if otelEndpoints[0].(map[string]any)["header_env"].(map[string]any)["authorization"] != "OTEL_AUTHORIZATION" { t.Fatalf("expected OpenTelemetry header_env in serialized config: %#v", wrapped.Config) } + if otelEndpoints[0].(map[string]any)["max_queue_size"] != float64(4096) || + otelEndpoints[0].(map[string]any)["max_export_batch_size"] != float64(256) || + otelEndpoints[0].(map[string]any)["scheduled_delay_millis"] != float64(750) { + t.Fatalf("expected OpenTelemetry batch settings in serialized config: %#v", wrapped.Config) + } +} + +func TestObservabilityOpenTelemetryEndpointPreservesExplicitZeroBatchSettings(t *testing.T) { + zero := uint64(0) + config := NewObservabilityOpenTelemetryEndpointConfig(OpenTelemetryTypeFull, "http://localhost:4318/v1/traces") + config.MaxQueueSize = &zero + config.MaxExportBatchSize = &zero + config.ScheduledDelayMillis = &zero + payload, err := json.Marshal(config) + if err != nil { + t.Fatalf("marshal OpenTelemetry endpoint config: %v", err) + } + for _, field := range []string{"max_queue_size", "max_export_batch_size", "scheduled_delay_millis"} { + if !strings.Contains(string(payload), `"`+field+`":0`) { + t.Fatalf("expected explicit zero %s in serialized config: %s", field, payload) + } + } } func assertS3StorageConfig(t *testing.T, storage ObservabilityS3StorageConfig) { diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index c705bca58..da00e4d08 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -226,6 +226,9 @@ class OpenTelemetryEndpointConfig: service_version: str | None = None instrumentation_scope: str = "opentelemetry" timeout_millis: int = 3000 + max_queue_size: int | None = None + max_export_batch_size: int | None = None + scheduled_delay_millis: int | None = None headers: dict[str, str] = field(default_factory=dict) header_env: dict[str, str] = field(default_factory=dict) resource_attributes: dict[str, str] = field(default_factory=dict) @@ -245,6 +248,9 @@ def to_dict(self) -> JsonObject: "service_version": self.service_version, "instrumentation_scope": self.instrumentation_scope, "timeout_millis": self.timeout_millis, + "max_queue_size": self.max_queue_size, + "max_export_batch_size": self.max_export_batch_size, + "scheduled_delay_millis": self.scheduled_delay_millis, "headers": self.headers, "header_env": self.header_env, "resource_attributes": self.resource_attributes, diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index 5c2d5bf41..578a821f7 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -89,6 +89,9 @@ class OpenTelemetryEndpointConfig: service_version: str | None = ... instrumentation_scope: str = ... timeout_millis: int = ... + max_queue_size: int | None = ... + max_export_batch_size: int | None = ... + scheduled_delay_millis: int | None = ... headers: dict[str, str] = field(default_factory=dict) header_env: dict[str, str] = field(default_factory=dict) resource_attributes: dict[str, str] = field(default_factory=dict) diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index b457f703a..5a4b39ecf 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -100,6 +100,9 @@ def test_defaults_and_component_wrapper(self): "gen_ai", "http://localhost:4318/v1/traces", header_env={"authorization": "OTEL_AUTHORIZATION"}, + max_queue_size=4096, + max_export_batch_size=256, + scheduled_delay_millis=750, ).to_dict() == { "type": "gen_ai", "endpoint": "http://localhost:4318/v1/traces", @@ -110,6 +113,9 @@ def test_defaults_and_component_wrapper(self): "service_name": "unknown_service", "instrumentation_scope": "opentelemetry", "timeout_millis": 3000, + "max_queue_size": 4096, + "max_export_batch_size": 256, + "scheduled_delay_millis": 750, "headers": {}, "header_env": {"authorization": "OTEL_AUTHORIZATION"}, "resource_attributes": {}, From 3146222b4a957160320843d4004a355473498540 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Wed, 5 Aug 2026 18:36:26 -0400 Subject: [PATCH 2/2] fix(python): preserve OpenTelemetry positional arguments Signed-off-by: Will Killian --- python/nemo_relay/observability.py | 6 +++--- python/nemo_relay/observability.pyi | 6 +++--- python/tests/test_observability_plugin.py | 25 +++++++++++++++++++++++ 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index da00e4d08..05a40b1fe 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -226,12 +226,12 @@ class OpenTelemetryEndpointConfig: service_version: str | None = None instrumentation_scope: str = "opentelemetry" timeout_millis: int = 3000 - max_queue_size: int | None = None - max_export_batch_size: int | None = None - scheduled_delay_millis: int | None = None headers: dict[str, str] = field(default_factory=dict) header_env: dict[str, str] = field(default_factory=dict) resource_attributes: dict[str, str] = field(default_factory=dict) + max_queue_size: int | None = None + max_export_batch_size: int | None = None + scheduled_delay_millis: int | None = None def to_dict(self) -> JsonObject: """Serialize this endpoint to the canonical plugin shape.""" diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index 578a821f7..16f14bba3 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -89,12 +89,12 @@ class OpenTelemetryEndpointConfig: service_version: str | None = ... instrumentation_scope: str = ... timeout_millis: int = ... - max_queue_size: int | None = ... - max_export_batch_size: int | None = ... - scheduled_delay_millis: int | None = ... headers: dict[str, str] = field(default_factory=dict) header_env: dict[str, str] = field(default_factory=dict) resource_attributes: dict[str, str] = field(default_factory=dict) + max_queue_size: int | None = ... + max_export_batch_size: int | None = ... + scheduled_delay_millis: int | None = ... def to_dict(self) -> JsonObject: ... @dataclass(slots=True) diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index 5a4b39ecf..4a317cfe2 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -84,6 +84,31 @@ def wait_for_requests(self, expected: int, timeout: float = 5.0) -> list[tuple[d class TestObservabilityConfigHelpers: + def test_opentelemetry_endpoint_preserves_existing_positional_arguments(self): + endpoint = OpenTelemetryEndpointConfig( + "full", + "http://localhost:4318/v1/traces", + "event", + ["custom.mark"], + [{"key": "source", "alias": "destination"}], + "grpc", + "relay-service", + "relay-namespace", + "1.2.3", + "relay-instrumentation", + 1234, + {"x-test": "header"}, + {"authorization": "OTEL_AUTHORIZATION"}, + {"deployment.environment": "test"}, + ) + + assert endpoint.headers == {"x-test": "header"} + assert endpoint.header_env == {"authorization": "OTEL_AUTHORIZATION"} + assert endpoint.resource_attributes == {"deployment.environment": "test"} + assert endpoint.max_queue_size is None + assert endpoint.max_export_batch_size is None + assert endpoint.scheduled_delay_millis is None + def test_defaults_and_component_wrapper(self): assert AtofConfig().to_dict() == {"enabled": False} assert AtifConfig().to_dict() == {