diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 0466e00a959..13d3b49e074 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -902,12 +902,31 @@ def publish_global_otel_v2_provider( """ global _published_v2_provider logger: Final = select_global_otel_v2_logger(in_memory_loggers, registered=registered) - attach_tenant_fan_out(logger.tracer_provider, *_v2_configs(in_memory_loggers, logger)) + attach_tenant_fan_out( + logger.tracer_provider, + *_v2_configs(in_memory_loggers, logger), + excluded_db_systems=_excluded_db_systems(in_memory_loggers, logger), + ) set_global_provider(logger.tracer_provider) _published_v2_provider = logger.tracer_provider # rebind-ok: startup records the one provider carrying the fan-out return logger +def _excluded_db_systems(in_memory_loggers: Sequence[object], logger: "OpenTelemetryV2") -> frozenset[str]: + """The datastore services withheld from tenant destinations. + + Proxy-wide the set lives on exactly one config: the ``otel`` callback's, the + only one ``callback_settings.otel`` writes. A preset (langfuse_otel, …) builds + its config env-only, so reading ``excluded_services`` off every logger and + unioning them reintroduces the ``LITELLM_OTEL_EXCLUDED_SERVICES`` value that + ``callback_settings.otel`` already overrode on the ``otel`` config. Falls back + to ``logger`` when no ``otel`` callback exists. + """ + loggers: Final = tuple(cb for cb in in_memory_loggers if isinstance(cb, OpenTelemetryV2)) + owner: Final = next((cb for cb in (logger, *loggers) if cb.callback_name == "otel"), logger) + return owner.config.excluded_services + + def _v2_configs(in_memory_loggers: Sequence[object], logger: "OpenTelemetryV2") -> tuple[OpenTelemetryV2Config, ...]: """Every v2 logger's config, the published logger's first. @@ -967,7 +986,13 @@ def fan_out_provider() -> ApiTracerProvider: return published logger: Final = _registered_v2_logger() if logger is not None: - attach_tenant_fan_out(logger.tracer_provider, logger.config) + from litellm.litellm_core_utils.litellm_logging import _in_memory_loggers + + attach_tenant_fan_out( + logger.tracer_provider, + logger.config, + excluded_db_systems=_excluded_db_systems(_in_memory_loggers, logger), + ) return logger.tracer_provider return get_tracer_provider() diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 050ebd51f37..66b69b7c729 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -1167,7 +1167,9 @@ def build_tracer_provider( _FAN_OUT_ATTACH_LOCK: Final = threading.Lock() -def attach_tenant_fan_out(provider: TracerProvider, *configs: OpenTelemetryV2Config) -> None: +def attach_tenant_fan_out( + provider: TracerProvider, *configs: OpenTelemetryV2Config, excluded_db_systems: frozenset[str] = frozenset() +) -> None: """Give ``provider`` the fan-out that delivers spans to key/team destinations. Called on the one provider published as the OTel global, and idempotent so a @@ -1176,7 +1178,9 @@ def attach_tenant_fan_out(provider: TracerProvider, *configs: OpenTelemetryV2Con so exactly one fan-out lands. ``configs`` name the operator's own exporters, one config per v2 logger since each keeps its own provider and still writes its account, so an additive destination pointing at any of them is delivered once - rather than twice. + rather than twice. ``excluded_db_systems`` comes from the ``otel`` callback's + config alone (see ``_excluded_db_systems``); unioning it across every logger's + config would reintroduce the env value that ``callback_settings.otel`` overrode. """ with _FAN_OUT_ATTACH_LOCK: if any(isinstance(processor, TenantFanOutSpanProcessor) for processor in _attached_processors(provider)): @@ -1184,7 +1188,7 @@ def attach_tenant_fan_out(provider: TracerProvider, *configs: OpenTelemetryV2Con provider.add_span_processor( TenantFanOutSpanProcessor( operator_sinks=operator_sink_scopes(*configs), - excluded_db_systems=frozenset().union(*(config.excluded_services for config in configs)), + excluded_db_systems=excluded_db_systems, ) ) diff --git a/tests/integration/observability/test_otel_excluded_services.py b/tests/integration/observability/test_otel_excluded_services.py index 8beaecf95f1..7c45ddcdbc0 100644 --- a/tests/integration/observability/test_otel_excluded_services.py +++ b/tests/integration/observability/test_otel_excluded_services.py @@ -212,7 +212,12 @@ def test_config_excluded_services_wins_over_env( langfuse_vars: dict[str, JsonValue], tmp_path: Path, ) -> None: - config: Final = _config_with(tmp_path, otel_audit_config, otel={"excluded_services": ["postgres"]}) + def with_langfuse_otel(config: dict) -> None: + config["litellm_settings"]["callbacks"] = ["otel", "langfuse_otel"] + + config: Final = _config_with( + tmp_path, otel_audit_config, otel={"excluded_services": ["postgres"]}, extra=with_langfuse_otel + ) with owned_proxy( gateway, tmp_path, {"LITELLM_OTEL_V2": "1", "LITELLM_OTEL_EXCLUDED_SERVICES": "redis"}, config=config, workers=2 ) as candidate: diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index ebc4747a502..eee7d24a475 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -1069,6 +1069,43 @@ class TestProviderWiring: assert kinds(published).count("TenantFanOutSpanProcessor") == 1 assert "TenantFanOutSpanProcessor" not in kinds(other) + def test_excluded_services_come_from_the_otel_callback_config_only(self): + """A preset builds its config env-only, so unioning ``excluded_services`` + across loggers reintroduces the env value ``callback_settings.otel`` + overrode. The fan-out must take the set from the ``otel`` config alone.""" + otel = OpenTelemetryV2( + config=OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory")], excluded_services=["postgres"]), + callback_name="otel", + ) + preset = OpenTelemetryV2( + config=OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory")], excluded_services=["redis"]), + callback_name="langfuse_otel", + ) + + publish_global_otel_v2_provider([preset], lambda _p: None, registered=otel) + + fan_out = next( + processor + for processor in otel._tracer_provider._active_span_processor._span_processors + if isinstance(processor, TenantFanOutSpanProcessor) + ) + assert fan_out._excluded_db_systems == frozenset({"postgresql"}) + + def test_excluded_services_fall_back_to_the_published_logger_without_an_otel_callback(self): + preset = OpenTelemetryV2( + config=OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory")], excluded_services=["redis"]), + callback_name="langfuse_otel", + ) + + publish_global_otel_v2_provider([], lambda _p: None, registered=preset) + + fan_out = next( + processor + for processor in preset._tracer_provider._active_span_processor._span_processors + if isinstance(processor, TenantFanOutSpanProcessor) + ) + assert fan_out._excluded_db_systems == frozenset({"redis"}) + @pytest.mark.parametrize("canonical", ["langfuse_otel", "arize"]) def test_publishing_tells_the_fan_out_about_every_v2_loggers_account(self, monkeypatch, canonical): monkeypatch.setenv("LITELLM_OTEL_TENANT_DESTINATION_MODE", "additive")