From 9c91a6b05256a70cd9e27d5e8e6c4e0928cd2514 Mon Sep 17 00:00:00 2001 From: Jaya Surya P <210020040@iitdh.ac.in> Date: Tue, 11 Aug 2026 18:20:06 +0530 Subject: [PATCH] [fix] expose BatchSpanProcessor queue, exporter and sampler tuning in register() Previously documented tuning params were silently dropped, making span drops under high throughput unavoidable via the Python API. --- python/CHANGELOG.md | 14 + python/fi_instrumentation/__init__.py | 7 + python/fi_instrumentation/otel.py | 163 ++++++-- python/pyproject.toml | 2 +- .../tests/test_batch_span_processor_tuning.py | 382 ++++++++++++++++++ 5 files changed, 529 insertions(+), 39 deletions(-) create mode 100644 python/tests/test_batch_span_processor_tuning.py diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 2cab70c5..557c712a 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -1,3 +1,17 @@ +## [1.1.0] - 2026-08-11 +### Feature +- `register()` now accepts span-processor / exporter / provider tuning: + `max_queue_size`, `schedule_delay_millis`, `max_export_batch_size`, + `export_timeout_millis`, `timeout`, `sampler`, `span_limits`, `span_exporter`. + Unset values fall back to the OTEL env vars / upstream defaults (no behavior change). +- `BatchSpanProcessor` forwards its documented tuning params to the OpenTelemetry + base class (previously silently dropped, causing unavoidable span drops under load). +- Re-exported `Sampler`, `ParentBased`, `TraceIdRatioBased`, `SpanLimits`. +### Fixed +- `register(batch=False)` warns when batch-only tuning params are passed. +- `register(span_exporter=...)` no longer crashes with non-OTLP exporters. +- Corrected docstrings that advertised an unsupported `endpoint` argument. + ## [0.1.7] - 2025-06-10 ### Feature - Added support for ai-evaluation diff --git a/python/fi_instrumentation/__init__.py b/python/fi_instrumentation/__init__.py index 9acbf35e..308f1013 100644 --- a/python/fi_instrumentation/__init__.py +++ b/python/fi_instrumentation/__init__.py @@ -16,6 +16,7 @@ ) from fi_instrumentation.instrumentation.helpers import safe_json_dumps from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased from .otel import ( PROJECT_NAME, @@ -24,8 +25,10 @@ SEMANTIC_CONVENTION, BatchSpanProcessor, HTTPSpanExporter, + Sampler, SemanticConvention, SimpleSpanProcessor, + SpanLimits, TracerProvider, Transport, register, @@ -52,6 +55,10 @@ "SimpleSpanProcessor", "BatchSpanProcessor", "HTTPSpanExporter", + "Sampler", + "ParentBased", + "TraceIdRatioBased", + "SpanLimits", "Resource", "PROJECT_NAME", "PROJECT_TYPE", diff --git a/python/fi_instrumentation/otel.py b/python/fi_instrumentation/otel.py index 60abc744..67cd76e9 100644 --- a/python/fi_instrumentation/otel.py +++ b/python/fi_instrumentation/otel.py @@ -6,6 +6,7 @@ import signal import sys import uuid +import warnings from enum import Enum from typing import Any, Dict, List, Optional, Tuple, Type, Union from urllib.parse import ParseResult, urlparse @@ -35,11 +36,12 @@ OTLPSpanExporter as _HTTPSpanExporter, ) from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import SpanProcessor +from opentelemetry.sdk.trace import SpanLimits, SpanProcessor from opentelemetry.sdk.trace import TracerProvider as _TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor as _BatchSpanProcessor from opentelemetry.sdk.trace.export import SimpleSpanProcessor as _SimpleSpanProcessor from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult +from opentelemetry.sdk.trace.sampling import Sampler from opentelemetry.trace import Status, StatusCode try: @@ -102,7 +104,47 @@ def register( verbose: bool = True, transport: Transport = Transport.HTTP, semantic_convention: SemanticConvention = SemanticConvention.FI, + max_queue_size: Optional[int] = None, + schedule_delay_millis: Optional[float] = None, + max_export_batch_size: Optional[int] = None, + export_timeout_millis: Optional[float] = None, + span_exporter: Optional[SpanExporter] = None, + timeout: Optional[float] = None, + sampler: Optional[Sampler] = None, + span_limits: Optional[SpanLimits] = None, ) -> _TracerProvider: + """Configure and return a Future AGI `TracerProvider`. + + All tuning args below default to `None`, meaning the corresponding OpenTelemetry + environment variable / upstream default applies (no behavior change when unset). + + Args: + project_name, project_type, project_version_name, eval_tags, metadata: + Future AGI project/eval configuration. + batch (bool): Use `BatchSpanProcessor` (default) vs `SimpleSpanProcessor`. + set_global_tracer_provider (bool): Register as the global OTel provider. + headers (dict): Extra headers sent to the collector. + verbose (bool): Print configuration details to stdout. + transport (Transport): HTTP (default) or gRPC. + semantic_convention (SemanticConvention): Attribute naming convention. + + Batch processor tuning (ignored unless `batch=True`): + max_queue_size (int): Max spans buffered before overflow drops + (`OTEL_BSP_MAX_QUEUE_SIZE`, upstream default 2048). + schedule_delay_millis (float): Delay between exports, in **milliseconds**. + max_export_batch_size (int): Max spans per export batch (upstream default 512). + export_timeout_millis (float): Batch export timeout, in **milliseconds**. + + Exporter tuning (used only when `span_exporter` is not provided): + span_exporter (SpanExporter): Pre-built exporter; bypasses the default one + (and `timeout` below). + timeout (float): Per-export timeout, in **seconds** (note: seconds here vs + milliseconds for the batch params above). + + Provider tuning: + sampler (Sampler): Head sampler (e.g. `TraceIdRatioBased`) to shed volume. + span_limits (SpanLimits): Caps on attributes/events/links per span. + """ eval_tags = eval_tags or [] metadata = metadata or {} @@ -154,20 +196,48 @@ def register( resource = Resource(attributes=resource_attributes) + # Only forward provider tuning when set, so unset -> upstream env/default. + provider_kwargs: Dict[str, Any] = {} + if sampler is not None: + provider_kwargs["sampler"] = sampler + if span_limits is not None: + provider_kwargs["span_limits"] = span_limits + tracer_provider = TracerProvider( resource=resource, verbose=False, id_generator=UuidIdGenerator(), - transport=transport + transport=transport, **provider_kwargs, ) span_processor: SpanProcessor if batch: span_processor = BatchSpanProcessor( + span_exporter=span_exporter, headers=headers, transport=transport, + max_queue_size=max_queue_size, + schedule_delay_millis=schedule_delay_millis, + max_export_batch_size=max_export_batch_size, + export_timeout_millis=export_timeout_millis, + timeout=timeout, ) else: + batch_only = { + "max_queue_size": max_queue_size, + "schedule_delay_millis": schedule_delay_millis, + "max_export_batch_size": max_export_batch_size, + "export_timeout_millis": export_timeout_millis, + } + ignored = [name for name, value in batch_only.items() if value is not None] + if ignored: + warnings.warn( + f"batch=False: ignoring batch-only tuning {ignored}; " + "these apply only to BatchSpanProcessor.", + stacklevel=2, + ) span_processor = SimpleSpanProcessor( + span_exporter=span_exporter, headers=headers, transport=transport, + timeout=timeout, ) tracer_provider.add_span_processor(span_processor) tracer_provider._default_processor = True @@ -198,13 +268,15 @@ class TracerProvider(_TracerProvider): Extended keyword arguments are documented in the `Args` section. For further documentation, see the OpenTelemetry documentation at https://opentelemetry.io/docs/specs/otel/trace/sdk/. + On construction a default `SimpleSpanProcessor` is added, exporting over `transport`. + The collector endpoint is inferred from `transport` and the `FI_BASE_URL`/`FI_GRPC_URL` + environment variables. Call `add_span_processor` to replace the default. In addition to + the arguments below, all upstream `opentelemetry.sdk.trace.TracerProvider` keyword + arguments (`sampler`, `resource`, `span_limits`, `id_generator`, ...) are accepted and + forwarded. + Args: - endpoint (str, optional): The collector endpoint to which spans will be exported. If - specified, a default SpanProcessor will be created and added to this TracerProvider. - If not provided, the `BASE_URL` environment variable will be - used to infer which collector endpoint to use, defaults to the gRPC endpoint. When - specifying the endpoint, the transport method (HTTP or gRPC) will be inferred from the - URL. + transport (Transport): Transport (HTTP or gRPC) used for the default exporter. verbose (bool): If True, configuration details will be printed to stdout. """ @@ -282,11 +354,11 @@ def _tracing_details(self) -> str: if processors := self._active_span_processor._span_processors: if len(processors) == 1: span_processor = self._active_span_processor._span_processors[0] - if exporter := getattr(span_processor, "span_exporter"): + if exporter := getattr(span_processor, "span_exporter", None): processor_name = span_processor.__class__.__name__ - endpoint = exporter._endpoint + endpoint = getattr(exporter, "_endpoint", None) transport = _exporter_transport(exporter) - headers = _printable_headers(exporter._headers) + headers = _printable_headers(getattr(exporter, "_headers", {}) or {}) else: processor_name = "Multiple Span Processors" endpoint = "Multiple Span Exporters" @@ -375,6 +447,21 @@ def _auto_set_ok_status(span: Any) -> None: span._status = Status(StatusCode.OK) +def _build_default_exporter( + transport: Transport, + headers: Optional[Dict[str, str]], + timeout: Optional[float], +) -> SpanExporter: + """Build the default OTLP exporter for `transport` (endpoint from env).""" + if transport == Transport.HTTP: + _, endpoint = _normalized_endpoint(get_env_collector_endpoint()) + return HTTPSpanExporter(endpoint=endpoint, headers=headers, timeout=timeout) + if transport == Transport.GRPC: + endpoint = get_env_grpc_collector_endpoint() + return GRPCSpanExporter(endpoint=endpoint, headers=headers, timeout=timeout) + raise ValueError(f"Invalid transport: {transport}") + + class SimpleSpanProcessor(_SimpleSpanProcessor): """ Simple SpanProcessor implementation. @@ -384,14 +471,14 @@ class SimpleSpanProcessor(_SimpleSpanProcessor): Args: span_exporter (SpanExporter, optional): The `SpanExporter` to which ended spans will be - passed. - endpoint (str, optional): The collector endpoint to which spans will be exported. If not - provided, the `BASE_URL` environment variable will be used to - infer which collector endpoint to use, defaults to the gRPC endpoint. When specifying - the endpoint, the transport method (HTTP or gRPC) will be inferred from the URL. + passed. If not provided, a default OTLP exporter is created for `transport`; its + collector endpoint is taken from the `FI_BASE_URL`/`FI_GRPC_URL` environment + variables. headers (dict, optional): Optional headers to include in the request to the collector. If not provided, the `FI_API_KEY` and `FI_SECRET_KEY` environment variable will be used. + transport (Transport, optional): Transport (HTTP or gRPC) used for the default exporter. + timeout (float, optional): Per-export timeout in seconds for the default exporter. """ def __init__( @@ -399,17 +486,12 @@ def __init__( span_exporter: Optional[SpanExporter] = None, headers: Optional[Dict[str, str]] = None, transport: Transport = Transport.HTTP, + timeout: Optional[float] = None, ): self._active_spans = {} if span_exporter is None: - if transport == Transport.HTTP: - endpoint = get_env_collector_endpoint() - parsed_url, endpoint = _normalized_endpoint(endpoint) - span_exporter = HTTPSpanExporter(endpoint=endpoint, headers=headers) - elif transport == Transport.GRPC: - endpoint = get_env_grpc_collector_endpoint() - span_exporter = GRPCSpanExporter(endpoint=endpoint, headers=headers) + span_exporter = _build_default_exporter(transport, headers, timeout) super().__init__(span_exporter) @@ -478,19 +560,19 @@ class BatchSpanProcessor(_BatchSpanProcessor): Args: span_exporter (SpanExporter, optional): The `SpanExporter` to which ended spans will be - passed. - endpoint (str, optional): The collector endpoint to which spans will be exported. If not - provided, the `BASE_URL` environment variable will be used to - infer which collector endpoint to use, defaults to the gRPC endpoint. When specifying - the endpoint, the transport method (HTTP or gRPC) will be inferred from the URL. + passed. If not provided, a default OTLP exporter is created for `transport`; its + collector endpoint is taken from the `FI_BASE_URL`/`FI_GRPC_URL` environment + variables. headers (dict, optional): Optional headers to include in the request to the collector. If not provided, the `FI_API_KEY` and `FI_SECRET_KEY` environment variable will be used. + transport (Transport, optional): Transport (HTTP or gRPC) used for the default exporter. max_queue_size (int, optional): The maximum queue size. schedule_delay_millis (float, optional): The delay between two consecutive exports in milliseconds. max_export_batch_size (int, optional): The maximum batch size. - export_timeout_millis (float, optional): The batch timeout in milliseconds. + export_timeout_millis (float, optional): The batch export timeout in milliseconds. + timeout (float, optional): Per-export timeout in seconds for the default exporter. """ def __init__( @@ -498,17 +580,22 @@ def __init__( span_exporter: Optional[SpanExporter] = None, headers: Optional[Dict[str, str]] = None, transport: Transport = Transport.HTTP, + max_queue_size: Optional[int] = None, + schedule_delay_millis: Optional[float] = None, + max_export_batch_size: Optional[int] = None, + export_timeout_millis: Optional[float] = None, + timeout: Optional[float] = None, ): if span_exporter is None: - if transport == Transport.HTTP: - endpoint = get_env_collector_endpoint() - parsed_url, endpoint = _normalized_endpoint(endpoint) - span_exporter = HTTPSpanExporter(endpoint=endpoint, headers=headers) - elif transport == Transport.GRPC: - endpoint = get_env_grpc_collector_endpoint() - span_exporter = GRPCSpanExporter(endpoint=endpoint, headers=headers) - - super().__init__(span_exporter) + span_exporter = _build_default_exporter(transport, headers, timeout) + + super().__init__( + span_exporter, + max_queue_size=max_queue_size, + schedule_delay_millis=schedule_delay_millis, + max_export_batch_size=max_export_batch_size, + export_timeout_millis=export_timeout_millis, + ) def on_end(self, span: Any) -> None: """Auto-set OK status for UNSET spans before batching.""" diff --git a/python/pyproject.toml b/python/pyproject.toml index 45ff78c8..d90b8566 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "fi-instrumentation-otel" -version = "1.0.0" +version = "1.1.0" description = "OpenTelemetry instrumentation" authors = ["Future AGI "] readme = "README.md" diff --git a/python/tests/test_batch_span_processor_tuning.py b/python/tests/test_batch_span_processor_tuning.py new file mode 100644 index 00000000..2e30eb5a --- /dev/null +++ b/python/tests/test_batch_span_processor_tuning.py @@ -0,0 +1,382 @@ +"""Tests for the span-processor / exporter / provider tuning exposed by the SDK. + +These verify that the tuning params (previously silently dropped) are now accepted +and forwarded, and that env vars / upstream defaults still apply when unset. + +Helpers below read upstream OTel internals in a version-tolerant way: OTel SDK +>=1.34 delegates to a `_batch_processor`, older versions store the values directly +on the processor. +""" + +import os +import warnings +from unittest.mock import patch + +import pytest +from opentelemetry.sdk.trace import SpanLimits +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.sdk.trace.sampling import TraceIdRatioBased + +from fi_instrumentation.fi_types import ProjectType +from fi_instrumentation.otel import ( + BatchSpanProcessor, + SimpleSpanProcessor, + TracerProvider, + Transport, + register, +) + + +def _bp_get(processor, name): + """Read a batch-processor setting across OTel layouts.""" + holder = getattr(processor, "_batch_processor", processor) + for obj, attr in ((holder, "_" + name), (processor, name), (processor, "_" + name)): + if hasattr(obj, attr): + return getattr(obj, attr) + raise AttributeError(name) + + +def _queue_batch(processor): + return _bp_get(processor, "max_queue_size"), _bp_get(processor, "max_export_batch_size") + + +def _exporter_of(processor): + """Effective exporter across OTel layouts (batch and simple processors).""" + bp = getattr(processor, "_batch_processor", None) + if bp is not None and hasattr(bp, "_exporter"): + return bp._exporter + return getattr(processor, "span_exporter", None) + + +def _clear_env(keys): + saved = {k: os.environ.pop(k, None) for k in keys} + try: + yield + finally: + for k, v in saved.items(): + if v is not None: + os.environ[k] = v + else: + os.environ.pop(k, None) + + +@pytest.fixture +def clean_bsp_env(): + yield from _clear_env( + [ + "OTEL_BSP_MAX_QUEUE_SIZE", + "OTEL_BSP_MAX_EXPORT_BATCH_SIZE", + "OTEL_BSP_SCHEDULE_DELAY", + "OTEL_BSP_EXPORT_TIMEOUT", + ] + ) + + +@pytest.fixture +def clean_exporter_env(): + yield from _clear_env( + ["OTEL_EXPORTER_OTLP_TIMEOUT", "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT"] + ) + + +class TestBatchSpanProcessorTuning: + def test_defaults_match_upstream(self, clean_bsp_env): + """No kwargs, no env -> upstream OTel defaults (unchanged).""" + p = BatchSpanProcessor(span_exporter=InMemorySpanExporter()) + assert _queue_batch(p) == (2048, 512) + + def test_explicit_kwargs_are_forwarded(self, clean_bsp_env): + """The core fix: kwargs are now honored instead of dropped.""" + p = BatchSpanProcessor( + span_exporter=InMemorySpanExporter(), + max_queue_size=10000, + max_export_batch_size=1024, + ) + assert _queue_batch(p) == (10000, 1024) + + def test_schedule_delay_and_export_timeout_forwarded(self, clean_bsp_env): + p = BatchSpanProcessor( + span_exporter=InMemorySpanExporter(), + schedule_delay_millis=1234, + export_timeout_millis=5678, + ) + assert _bp_get(p, "schedule_delay_millis") == 1234 + assert _bp_get(p, "export_timeout_millis") == 5678 + + def test_env_vars_respected_when_unset(self, clean_bsp_env): + os.environ["OTEL_BSP_MAX_QUEUE_SIZE"] = "3333" + p = BatchSpanProcessor(span_exporter=InMemorySpanExporter()) + queue, batch = _queue_batch(p) + assert queue == 3333 + assert batch == 512 # untouched -> upstream default + + def test_env_var_batch_size_respected_when_unset(self, clean_bsp_env): + os.environ["OTEL_BSP_MAX_EXPORT_BATCH_SIZE"] = "128" + p = BatchSpanProcessor(span_exporter=InMemorySpanExporter()) + assert _queue_batch(p)[1] == 128 + + def test_explicit_kwarg_overrides_env(self, clean_bsp_env): + os.environ["OTEL_BSP_MAX_QUEUE_SIZE"] = "3333" + p = BatchSpanProcessor( + span_exporter=InMemorySpanExporter(), max_queue_size=9000 + ) + assert _queue_batch(p)[0] == 9000 + + +class TestExporterTuning: + """timeout must reach the default OTLP exporter (both processors).""" + + def test_batch_forwards_timeout(self): + p = BatchSpanProcessor(timeout=42) + assert _exporter_of(p)._timeout == 42 + + def test_simple_forwards_timeout(self): + p = SimpleSpanProcessor(timeout=7) + assert p.span_exporter._timeout == 7 + + def test_unset_timeout_uses_upstream_default(self, clean_exporter_env): + """Backward-compat: no timeout -> upstream default (10s).""" + p = BatchSpanProcessor(span_exporter=None) + assert _exporter_of(p)._timeout == 10 + + def test_prebuilt_span_exporter_is_used_as_is(self): + """A caller-supplied exporter is used verbatim; timeout ignored.""" + exp = InMemorySpanExporter() + p = BatchSpanProcessor(span_exporter=exp, timeout=99) + assert _exporter_of(p) is exp + + +class TestRegisterPlumbsTuning: + """register() must forward the tuning kwargs to the processor / provider.""" + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.BatchSpanProcessor") + def test_register_forwards_tuning(self, mock_bsp, _mock_tp, _mock_check): + register( + batch=True, + verbose=False, + max_queue_size=7777, + schedule_delay_millis=1000, + max_export_batch_size=512, + export_timeout_millis=15000, + ) + _, kwargs = mock_bsp.call_args + assert kwargs["max_queue_size"] == 7777 + assert kwargs["schedule_delay_millis"] == 1000 + assert kwargs["max_export_batch_size"] == 512 + assert kwargs["export_timeout_millis"] == 15000 + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.BatchSpanProcessor") + def test_register_defaults_to_none(self, mock_bsp, _mock_tp, _mock_check): + """Unset -> None, so upstream env/default behavior is preserved.""" + register(batch=True, verbose=False) + _, kwargs = mock_bsp.call_args + assert kwargs["max_queue_size"] is None + assert kwargs["schedule_delay_millis"] is None + assert kwargs["max_export_batch_size"] is None + assert kwargs["export_timeout_millis"] is None + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.BatchSpanProcessor") + def test_register_forwards_exporter_and_provider_config( + self, mock_bsp, mock_tp, _mock_check + ): + sampler = TraceIdRatioBased(0.5) + limits = SpanLimits(max_attributes=10) + exp = InMemorySpanExporter() + register( + verbose=False, + timeout=30, + sampler=sampler, + span_limits=limits, + span_exporter=exp, + ) + _, bkw = mock_bsp.call_args + assert bkw["timeout"] == 30 + assert bkw["span_exporter"] is exp + + _, pkw = mock_tp.call_args + assert pkw["sampler"] is sampler + assert pkw["span_limits"] is limits + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.BatchSpanProcessor") + def test_register_omits_provider_kwargs_when_unset( + self, _mock_bsp, mock_tp, _mock_check + ): + """Unset sampler/span_limits are not forwarded -> upstream defaults apply.""" + register(verbose=False) + _, pkw = mock_tp.call_args + assert "sampler" not in pkw + assert "span_limits" not in pkw + + +class TestRegisterBatchFalse: + """batch=False -> SimpleSpanProcessor path: exporter tuning wired, batch tuning warned.""" + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.SimpleSpanProcessor") + def test_wires_simple_processor(self, mock_ssp, _mock_tp, _mock_check): + register(batch=False, verbose=False, timeout=5) + _, kw = mock_ssp.call_args + assert kw["timeout"] == 5 + # batch-only tunables are not passed to SimpleSpanProcessor + assert "max_queue_size" not in kw + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.SimpleSpanProcessor") + def test_warns_on_batch_only_params(self, _mock_ssp, _mock_tp, _mock_check): + with pytest.warns(UserWarning, match="batch=False"): + register(batch=False, verbose=False, max_queue_size=10000) + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.SimpleSpanProcessor") + def test_no_warning_without_batch_params(self, _mock_ssp, _mock_tp, _mock_check): + with warnings.catch_warnings(): + warnings.simplefilter("error") + register(batch=False, verbose=False, timeout=5) + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + @patch("fi_instrumentation.otel.TracerProvider") + @patch("fi_instrumentation.otel.SimpleSpanProcessor") + def test_forwards_span_exporter(self, mock_ssp, _mock_tp, _mock_check): + exp = InMemorySpanExporter() + register(batch=False, verbose=False, span_exporter=exp) + _, kw = mock_ssp.call_args + assert kw["span_exporter"] is exp + + def test_warning_fires_across_python_versions(self): + """Regression: the warning must not rely on locals() inside a comprehension + (broken on <3.12), so it fires regardless of interpreter.""" + with patch( + "fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False + ), patch("fi_instrumentation.otel.TracerProvider"), patch( + "fi_instrumentation.otel.SimpleSpanProcessor" + ), pytest.warns( + UserWarning, match=r"schedule_delay_millis" + ): + register(batch=False, verbose=False, schedule_delay_millis=1000) + + +class TestTracerProviderForwarding: + def test_accepts_sampler_and_span_limits(self): + """Direct construction forwards provider kwargs to upstream.""" + sampler = TraceIdRatioBased(0.25) + tp = TracerProvider(sampler=sampler, span_limits=SpanLimits(max_attributes=3)) + assert tp.sampler is sampler + assert tp._span_limits.max_attributes == 3 + + +class TestPublicReExports: + """Concrete samplers are re-exported so `sampler=` is usable without reaching + into opentelemetry.*""" + + def test_samplers_and_limits_reexported(self): + import fi_instrumentation as fi + + for name in ("Sampler", "ParentBased", "TraceIdRatioBased", "SpanLimits"): + assert name in fi.__all__ + assert hasattr(fi, name) + # constructable + accepted by the provider + tp = TracerProvider(sampler=fi.ParentBased(fi.TraceIdRatioBased(0.5)), verbose=False) + assert tp.sampler is not None + + +class TestEnvPrecedence: + """schedule_delay / export_timeout also honor their OTEL_BSP_* env vars.""" + + def test_schedule_delay_env_respected(self, clean_bsp_env): + os.environ["OTEL_BSP_SCHEDULE_DELAY"] = "7000" + p = BatchSpanProcessor(span_exporter=InMemorySpanExporter()) + assert _bp_get(p, "schedule_delay_millis") == 7000 + + def test_export_timeout_env_respected(self, clean_bsp_env): + os.environ["OTEL_BSP_EXPORT_TIMEOUT"] = "4000" + p = BatchSpanProcessor(span_exporter=InMemorySpanExporter()) + assert _bp_get(p, "export_timeout_millis") == 4000 + + +class TestGrpcTransport: + def test_grpc_forwards_timeout(self): + pytest.importorskip("grpc") + p = BatchSpanProcessor(transport=Transport.GRPC, timeout=11) + assert _exporter_of(p)._timeout == 11 + + +class TestEndpointRejected: + """The corrected docstrings: these classes do NOT accept `endpoint`.""" + + def test_batch_rejects_endpoint(self): + with pytest.raises(TypeError): + BatchSpanProcessor(endpoint="http://x") + + def test_simple_rejects_endpoint(self): + with pytest.raises(TypeError): + SimpleSpanProcessor(endpoint="http://x") + + def test_provider_rejects_endpoint(self): + with pytest.raises(TypeError): + TracerProvider(endpoint="http://x") + + +class TestSamplerEffectiveness: + """sampler is not just stored -> it actually drops/keeps spans.""" + + def _emit_and_count(self, sampler) -> int: + exp = InMemorySpanExporter() + tp = TracerProvider(sampler=sampler, verbose=False) + tp.add_span_processor(SimpleSpanProcessor(span_exporter=exp)) + tracer = tp.get_tracer(__name__) + for i in range(20): + with tracer.start_as_current_span(f"s{i}"): + pass + tp.force_flush() + return len(exp.get_finished_spans()) + + def test_ratio_zero_drops_all(self): + assert self._emit_and_count(TraceIdRatioBased(0.0)) == 0 + + def test_ratio_one_keeps_all(self): + assert self._emit_and_count(TraceIdRatioBased(1.0)) == 20 + + +class TestEndToEndEmit: + """Full chain (no mocks): tuned processor actually exports spans.""" + + def test_batch_processor_emits(self): + exp = InMemorySpanExporter() + tp = TracerProvider(verbose=False) + tp.add_span_processor( + BatchSpanProcessor(span_exporter=exp, max_queue_size=100, max_export_batch_size=32) + ) + tracer = tp.get_tracer(__name__) + with tracer.start_as_current_span("root"): + with tracer.start_as_current_span("child"): + pass + tp.force_flush() + assert len(exp.get_finished_spans()) == 2 + + @patch("fi_instrumentation.otel.check_custom_eval_config_exists", return_value=False) + def test_register_with_custom_exporter_does_not_crash(self, _mock_check): + """Regression: register() called _tracing_details() unconditionally, which + crashed on exporters lacking _endpoint/_headers (e.g. InMemorySpanExporter).""" + exp = InMemorySpanExporter() + tp = register( + project_type=ProjectType.OBSERVE, + verbose=False, + span_exporter=exp, + ) + tracer = tp.get_tracer(__name__) + with tracer.start_as_current_span("s"): + pass + tp.force_flush() + assert len(exp.get_finished_spans()) == 1 + tp.shutdown()