feat: configure OpenTelemetry batching per endpoint (#719)

#### Overview

Add endpoint-local OpenTelemetry batch processor configuration for observability config version 3. Omitted values continue to inherit the standard `OTEL_BSP_*` environment settings and SDK defaults.

- [x] I confirm this contribution is my own work, or I have the right to submit it under this project's license.
- [x] I searched existing issues and open pull requests, and this does not duplicate existing work.

#### Details

- Add optional `max_queue_size`, `max_export_batch_size`, and `scheduled_delay_millis` fields to each configured OpenTelemetry endpoint.
- Apply endpoint overrides after the environment-aware SDK batch defaults are constructed, preserving independent endpoint behavior and existing diagnostics, flush, and shutdown semantics.
- Reject zero values and explicitly configured export batches larger than their queue with indexed configuration diagnostics.
- Add configuration parity and serialization coverage for Rust, Python, Node.js, and Go while leaving manual subscriber APIs and the C FFI unchanged.
- Preserve the existing Python `OpenTelemetryEndpointConfig` positional constructor order by appending the new fields after all pre-existing fields, with regression coverage for legacy positional calls.
- Document precedence, sizing semantics, queue overflow behavior, scheduled-delay behavior, and the endpoint-local OTLP request timeout.

Validation:

- `uv run pre-commit run --all-files`
- `cargo clippy --workspace --all-targets -- -D warnings`
- `just test-go`
- `just docs`
- Full isolated Python suite: 640 passed, 45 skipped
- Focused Python observability suite: 16 passed
- Focused Rust and Node.js observability configuration tests

The canonical Rust, Python, and Node.js commands were also attempted. Configuration-sensitive cases discovered a local `/Users/wkillian/.nemo-relay/plugins.toml`, producing unrelated inherited-configuration warnings and conflicts. The canonical Python run completed with 629 passing tests and 11 affected tests; the full suite passes when run with isolated configuration discovery.

#### Where should the reviewer start?

Start with `crates/core/src/observability/plugin_component.rs` for the public endpoint schema and validation, then `crates/core/src/observability/otel.rs` for environment-aware batch processor construction and endpoint override precedence. The Python compatibility follow-up is covered by `python/tests/test_observability_plugin.py`.

#### Related Issues: (use one of the action keywords Closes / Fixes / Resolves / Relates to)

- Relates to: RELAY-626



## Summary by CodeRabbit

* **New Features**
  * Added per-endpoint OpenTelemetry settings for queue size, export batch size, and scheduled export delay.
  * Added support across Rust, Node.js, Go, and Python configuration interfaces.
  * Endpoint-specific settings take precedence over process-wide defaults.
  * Added validation for zero values and batch sizes larger than the queue.

* **Documentation**
  * Updated configuration guidance, examples, fallback behavior, and known issues.

* **Tests**
  * Added coverage for serialization, validation, defaults, and runtime propagation.

Authors:
  - Will Killian (https://github.com/willkill07)

Approvers:
  - Zhongxuan (Daniel) Wang (https://github.com/zhongxuanwang-nv)

URL: #719

Author: willkill07

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<OtlpAttributeMapping>,
     timeout: Duration,
     transport: OtlpTransport,
+    max_queue_size: Option<usize>,
+    max_export_batch_size: Option<usize>,
+    scheduled_delay: Option<Duration>,
 }
 
 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<usize>, Option<usize>, Option<Duration>) {
+        (
+            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<String>) -> 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<E: SpanExporter + 'static>(
-        exporter: E,
-        endpoint: String,
-        diagnostic_field: Option<String>,
-    ) -> Self {
-        Self::new_with_batch_config(
-            exporter,
-            endpoint,
-            diagnostic_field,
-            opentelemetry_sdk::trace::BatchConfig::default(),
-        )
-    }
-
     fn new_with_batch_config<E: SpanExporter + 'static>(
         exporter: E,
         endpoint: String,

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<usize>,
+    /// Maximum spans exported in one batch.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_export_batch_size: Option<usize>,
+    /// Maximum delay before exporting a non-full batch, in milliseconds.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub scheduled_delay_millis: Option<u64>,
 }
 
 /// 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, &section)?;
+    validate_otel_batch_config(index, &section)?;
     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(&section.endpoints) {
@@ -2430,6 +2491,50 @@ fn validate_opentelemetry_section(
     validate_opentelemetry_feature_support(diagnostics, policy, section);
 }
 
+fn validate_opentelemetry_batch_config(
+    diagnostics: &mut Vec<ConfigDiagnostic>,
+    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,

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();