feat: add configurable log processing parameters to telemetry system

packages/node/iii/src/telemetry-system/index.ts

@@ -44,6 +44,7 @@ import { SharedEngineConnection } from './connection'
 import { EngineSpanExporter, EngineMetricsExporter, EngineLogExporter } from './exporters'
 import { extractTraceparent } from './context'
 import { patchGlobalFetch, unpatchGlobalFetch } from './fetch-instrumentation'
+import { parseIntegerEnv, parseNumberEnv } from './utils'
 
 // Re-export everything from submodules
 export * from './types'
@@ -166,11 +167,27 @@ export function initOtel(config: OtelConfig = {}): void {
 
   // Initialize logs (always enabled when OTEL is enabled)
   const logExporter = new EngineLogExporter(sharedConnection)
+  const logsScheduledDelayMillis =
+    config.logsFlushIntervalMs ??
+    parseNumberEnv(process.env.OTEL_LOGS_FLUSH_INTERVAL_MS, 0) ??
+    DEFAULT_OTEL_CONFIG.logsFlushIntervalMs
+  const logsMaxExportBatchSize =
+    config.logsBatchSize ??
+    parseIntegerEnv(process.env.OTEL_LOGS_BATCH_SIZE, 1) ??
+    DEFAULT_OTEL_CONFIG.logsBatchSize
+
   loggerProvider = new LoggerProvider({ resource })
-  loggerProvider.addLogRecordProcessor(new BatchLogRecordProcessor(logExporter))
+  loggerProvider.addLogRecordProcessor(
+    new BatchLogRecordProcessor(logExporter, {
+      scheduledDelayMillis: logsScheduledDelayMillis,
+      maxExportBatchSize: logsMaxExportBatchSize,
+    }),
+  )
   logger = loggerProvider.getLogger(serviceName)
 
-  console.debug('[OTel] Logs initialized')
+  console.debug(
+    `[OTel] Logs initialized: delay=${logsScheduledDelayMillis}ms, batch=${logsMaxExportBatchSize}`,
+  )
 }
 
 /**

packages/node/iii/src/telemetry-system/types.ts

@@ -65,6 +65,10 @@ export interface OtelConfig {
   metricsEnabled?: boolean
   /** Metrics export interval in milliseconds. Defaults to 60000 (60 seconds). */
   metricsExportIntervalMs?: number
+  /** Log processor flush delay in milliseconds. Defaults to 100ms. */
+  logsFlushIntervalMs?: number
+  /** Maximum number of log records exported per batch. Defaults to 1. */
+  logsBatchSize?: number
   /** Whether to auto-instrument globalThis.fetch calls. Defaults to true. Works on Node.js, Bun, and Deno. Set to false to disable. */
   fetchInstrumentationEnabled?: boolean
   /** Optional reconnection configuration for the WebSocket connection. */
@@ -79,6 +83,8 @@ export const DEFAULT_OTEL_CONFIG = {
   engineWsUrl: 'ws://localhost:49134',
   metricsEnabled: true,
   metricsExportIntervalMs: 60000,
+  logsFlushIntervalMs: 100,
+  logsBatchSize: 1,
   fetchInstrumentationEnabled: true,
 } as const satisfies Partial<OtelConfig>
 

packages/node/iii/src/telemetry-system/utils.ts

@@ -0,0 +1,21 @@
+/**
+ * Parse a numeric environment variable with optional minimum bound.
+ */
+export function parseNumberEnv(value: string | undefined, minimum: number = 0): number | undefined {
+  if (value === undefined) return undefined
+  const parsed = Number(value)
+  if (!Number.isFinite(parsed) || parsed < minimum) return undefined
+  return parsed
+}
+
+/**
+ * Parse an integer environment variable with optional minimum bound.
+ */
+export function parseIntegerEnv(
+  value: string | undefined,
+  minimum: number = 0,
+): number | undefined {
+  const parsed = parseNumberEnv(value, minimum)
+  if (parsed === undefined || !Number.isInteger(parsed)) return undefined
+  return parsed
+}

packages/python/iii/src/iii/telemetry.py

@@ -10,7 +10,7 @@
 import logging
 import os
 import uuid
-from typing import Any
+from typing import Any, cast
 
 from .telemetry_types import OtelConfig
 
@@ -44,10 +44,11 @@ def init_otel(
 
     cfg = config or OtelConfig()
 
-    enabled = cfg.enabled
-    if enabled is None:
-        env = os.environ.get("OTEL_ENABLED", "").lower()
+    if cfg.enabled is not None:
+        enabled = cfg.enabled
+    else:
         # Enabled by default; set OTEL_ENABLED=false/0/no/off to disable
+        env = os.environ.get("OTEL_ENABLED", "").lower()
         enabled = env not in ("false", "0", "no", "off")
 
     if not enabled:
@@ -113,7 +114,7 @@ def init_otel(
     # --- Log exporter ---
     logs_enabled = cfg.logs_enabled if cfg.logs_enabled is not None else True
     if logs_enabled:
-        _configure_log_provider(resource, _connection)
+        _configure_log_provider(resource, _connection, cfg)
 
     _initialized = True
 
@@ -152,7 +153,32 @@ def _configure_meter_provider(
     _meter = meter_provider.get_meter(service_name)
 
 
-def _configure_log_provider(resource: Any, connection: Any) -> None:
+def _resolve_int(
+    config_value: int | None,
+    env_var: str,
+    default: int,
+    minimum: int = 0,
+) -> int:
+    """Resolve an integer setting: explicit config > env var > default.
+
+    Matches the Node SDK's resolution order for cross-SDK consistency.
+    """
+    if config_value is not None:
+        return config_value
+
+    raw = os.environ.get(env_var)
+    if raw is not None:
+        try:
+            val = int(raw)
+            if val >= minimum:
+                return val
+        except (ValueError, TypeError):
+            pass
+
+    return default
+
+
+def _configure_log_provider(resource: Any, connection: Any, cfg: OtelConfig) -> None:
     """Set up a global SdkLoggerProvider with EngineLogExporter."""
     global _log_provider
     try:
@@ -168,11 +194,31 @@ def _configure_log_provider(resource: Any, connection: Any) -> None:
         return
 
     log_exporter = EngineLogExporter(connection)
+
+    logs_flush_interval_ms = _resolve_int(
+        cfg.logs_flush_interval_ms, "OTEL_LOGS_FLUSH_INTERVAL_MS", default=100,
+    )
+    logs_batch_size = _resolve_int(
+        cfg.logs_batch_size, "OTEL_LOGS_BATCH_SIZE", default=1, minimum=1,
+    )
+
     log_provider = SdkLoggerProvider(resource=resource)
-    log_provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter))  # type: ignore[arg-type]
+    log_provider.add_log_record_processor(
+        BatchLogRecordProcessor(
+            cast(Any, log_exporter),
+            schedule_delay_millis=logs_flush_interval_ms,
+            max_export_batch_size=logs_batch_size,
+        )
+    )
     _logs.set_logger_provider(log_provider)
     _log_provider = log_provider
 
+    logging.getLogger("iii.telemetry").debug(
+        "Log provider configured: flush_interval=%dms, batch_size=%d",
+        logs_flush_interval_ms,
+        logs_batch_size,
+    )
+
 
 _original_opener_open: Any = None
 
@@ -297,6 +343,15 @@ async def shutdown_otel_async() -> None:
     _reset_state()
 
 
+def _shutdown_provider(provider: Any) -> None:
+    """Call shutdown() on a provider, silently ignoring errors."""
+    try:
+        if provider is not None and hasattr(provider, "shutdown"):
+            provider.shutdown()
+    except Exception:
+        pass
+
+
 def _reset_state() -> None:
     global _tracer, _meter, _meter_provider, _log_provider, _connection, _initialized, _fetch_patched
 
@@ -312,21 +367,11 @@ def _reset_state() -> None:
     if _initialized:
         try:
             from opentelemetry import trace
-            provider = trace.get_tracer_provider()
-            if hasattr(provider, "shutdown"):
-                provider.shutdown()
-        except Exception:
-            pass
-        try:
-            if _meter_provider and hasattr(_meter_provider, "shutdown"):
-                _meter_provider.shutdown()
-        except Exception:
-            pass
-        try:
-            if _log_provider and hasattr(_log_provider, "shutdown"):
-                _log_provider.shutdown()
+            _shutdown_provider(trace.get_tracer_provider())
         except Exception:
             pass
+        _shutdown_provider(_meter_provider)
+        _shutdown_provider(_log_provider)
 
     _tracer = None
     _meter = None

packages/python/iii/src/iii/telemetry_types.py

@@ -32,6 +32,12 @@ class OtelConfig:
     logs_enabled: bool | None = None
     """Enable OTel log export via EngineLogExporter. Defaults to True when OTel is enabled."""
 
+    logs_flush_interval_ms: int | None = None
+    """Log processor flush delay in milliseconds. Defaults to 100ms when not set."""
+
+    logs_batch_size: int | None = None
+    """Maximum number of log records exported per batch. Defaults to 1 when not set."""
+
     metrics_enabled: bool = True
     """Enable OTel metrics export via EngineMetricsExporter. Defaults to True."""
 

packages/python/iii/tests/test_telemetry.py

@@ -9,29 +9,25 @@
 ORIGINAL_OPENER_OPEN = urllib.request.OpenerDirector.open
 
 
+def _reset_otel_singleton(module_path: str, provider_attr: str, set_once_attr: str) -> None:
+    """Reset one OTel global singleton (provider + its SetOnce guard)."""
+    try:
+        import importlib
+        mod = importlib.import_module(module_path)
+        setattr(mod, provider_attr, None)
+        getattr(mod, set_once_attr)._done = False
+    except Exception:
+        pass
+
+
 @pytest.fixture(autouse=True)
 def cleanup():
     yield
     shutdown_otel()
     # Reset all OTel global singletons so tests don't bleed state
-    try:
-        import opentelemetry._logs._internal as _li
-        _li._LOGGER_PROVIDER = None
-        _li._LOGGER_PROVIDER_SET_ONCE._done = False
-    except Exception:
-        pass
-    try:
-        import opentelemetry.trace._internal as _ti
-        _ti._TRACER_PROVIDER = None
-        _ti._TRACER_PROVIDER_SET_ONCE._done = False
-    except Exception:
-        pass
-    try:
-        import opentelemetry.metrics._internal as _mi
-        _mi._METER_PROVIDER = None
-        _mi._METER_PROVIDER_SET_ONCE._done = False
-    except Exception:
-        pass
+    _reset_otel_singleton("opentelemetry._logs._internal", "_LOGGER_PROVIDER", "_LOGGER_PROVIDER_SET_ONCE")
+    _reset_otel_singleton("opentelemetry.trace._internal", "_TRACER_PROVIDER", "_TRACER_PROVIDER_SET_ONCE")
+    _reset_otel_singleton("opentelemetry.metrics._internal", "_METER_PROVIDER", "_METER_PROVIDER_SET_ONCE")
     urllib.request.OpenerDirector.open = ORIGINAL_OPENER_OPEN
 
 
@@ -143,3 +139,64 @@ def test_shutdown_closes_connection():
         init_otel(OtelConfig(enabled=True))
         asyncio.run(shutdown_otel_async())
         mock_shutdown.assert_called_once()
+
+
+def test_init_configures_log_batch_params_defaults():
+    """Default batch config: 100ms flush, batch size 1."""
+    from opentelemetry._logs import get_logger_provider
+    from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider
+    from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+
+    init_otel(OtelConfig(enabled=True))
+    lp = get_logger_provider()
+    assert isinstance(lp, SdkLoggerProvider)
+    # Verify processor was created with no errors (config was applied)
+    processors = lp._multi_log_record_processor._log_record_processors
+    blrp = next((p for p in processors if isinstance(p, BatchLogRecordProcessor)), None)
+    assert blrp is not None
+
+
+def test_init_log_batch_config_explicit_overrides():
+    """Explicit config values override env vars and defaults."""
+    from opentelemetry._logs import get_logger_provider
+    from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider
+
+    init_otel(OtelConfig(enabled=True, logs_flush_interval_ms=200, logs_batch_size=5))
+    lp = get_logger_provider()
+    assert isinstance(lp, SdkLoggerProvider)
+
+
+def test_init_log_batch_config_env_var_override(monkeypatch):
+    """Env vars override defaults when config fields are None."""
+    from opentelemetry._logs import get_logger_provider
+    from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider
+
+    monkeypatch.setenv("OTEL_LOGS_FLUSH_INTERVAL_MS", "500")
+    monkeypatch.setenv("OTEL_LOGS_BATCH_SIZE", "10")
+    init_otel(OtelConfig(enabled=True))
+    lp = get_logger_provider()
+    assert isinstance(lp, SdkLoggerProvider)
+
+
+def test_init_log_batch_config_explicit_beats_env(monkeypatch):
+    """Explicit config value takes precedence over env var."""
+    from opentelemetry._logs import get_logger_provider
+    from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider
+
+    monkeypatch.setenv("OTEL_LOGS_FLUSH_INTERVAL_MS", "999")
+    monkeypatch.setenv("OTEL_LOGS_BATCH_SIZE", "99")
+    init_otel(OtelConfig(enabled=True, logs_flush_interval_ms=200, logs_batch_size=5))
+    lp = get_logger_provider()
+    assert isinstance(lp, SdkLoggerProvider)
+
+
+def test_init_log_batch_config_invalid_env_ignored(monkeypatch):
+    """Invalid env var values are silently ignored, falling through to defaults."""
+    from opentelemetry._logs import get_logger_provider
+    from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider
+
+    monkeypatch.setenv("OTEL_LOGS_FLUSH_INTERVAL_MS", "not-a-number")
+    monkeypatch.setenv("OTEL_LOGS_BATCH_SIZE", "-5")
+    init_otel(OtelConfig(enabled=True))
+    lp = get_logger_provider()
+    assert isinstance(lp, SdkLoggerProvider)

packages/python/iii/tests/test_telemetry_types.py

@@ -36,3 +36,15 @@ def test_logs_enabled_defaults_to_none():
 def test_engine_ws_url_exists():
     cfg = OtelConfig(engine_ws_url="ws://custom:1234")
     assert cfg.engine_ws_url == "ws://custom:1234"
+
+
+def test_logs_batch_config_defaults_to_none():
+    cfg = OtelConfig()
+    assert cfg.logs_flush_interval_ms is None
+    assert cfg.logs_batch_size is None
+
+
+def test_logs_batch_config_explicit():
+    cfg = OtelConfig(logs_flush_interval_ms=500, logs_batch_size=10)
+    assert cfg.logs_flush_interval_ms == 500
+    assert cfg.logs_batch_size == 10