From 683a049ca88c1dcc5686c30eda8a97acdfac2197 Mon Sep 17 00:00:00 2001 From: yucheng Date: Thu, 24 Sep 2026 07:58:51 +0000 Subject: [PATCH] test(otel v2): retarget the tenant span scope integration cells to otel_span_scope Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../integrationCritical/otelSpanScope.spec.ts | 79 +++++ ...pans.py => test_otel_tenant_span_scope.py} | 309 ++++++++++-------- ...y => test_otel_tenant_span_scope_chaos.py} | 8 +- 3 files changed, 248 insertions(+), 148 deletions(-) create mode 100644 tests/e2e/ui/tests/integrationCritical/otelSpanScope.spec.ts rename tests/integration/observability/{test_otel_tenant_internal_spans.py => test_otel_tenant_span_scope.py} (82%) rename tests/integration/observability/{test_otel_tenant_internal_spans_chaos.py => test_otel_tenant_span_scope_chaos.py} (97%) diff --git a/tests/e2e/ui/tests/integrationCritical/otelSpanScope.spec.ts b/tests/e2e/ui/tests/integrationCritical/otelSpanScope.spec.ts new file mode 100644 index 00000000000..fee862db9fe --- /dev/null +++ b/tests/e2e/ui/tests/integrationCritical/otelSpanScope.spec.ts @@ -0,0 +1,79 @@ +import { test, expect } from "@playwright/test"; +import { randomUUID } from "node:crypto"; +import { Page } from "../../fixtures/pages"; +import { navigateToPage } from "../../helpers/navigation"; +import { captureRequestBody, readBack } from "../../helpers/roundTrip"; + +test("team logging settings persist otel_span_scope no_internal through the callback editor", async ({ + page, + request, +}) => { + const master = process.env.LITELLM_MASTER_KEY ?? "sk-integration-master"; + const headers = { Authorization: `Bearer ${master}` }; + const prefix = `integration-browser-${randomUUID()}`; + const post = async (url: string, data: object) => { + const response = await request.post(url, { headers, data }); + expect(response.ok(), `${url}: ${await response.text()}`).toBe(true); + return response.json(); + }; + const team = await post("/team/new", { team_alias: prefix }); + try { + await page.goto("/ui/login"); + await page.getByPlaceholder("Enter your username").fill("admin"); + await page.getByPlaceholder("Enter your password").fill(master); + await page.getByRole("button", { name: "Login", exact: true }).click(); + await expect(page).toHaveURL( + (url) => url.pathname.startsWith("/ui") && !url.pathname.includes("login"), + ); + await navigateToPage(page, Page.Teams); + await page.getByRole("button", { name: new RegExp(`^${prefix}`) }).click(); + await page.getByRole("tab", { name: "Settings" }).click(); + await page.getByRole("button", { name: "Edit Settings" }).click(); + await page + .getByRole("button", { name: "Add Integration", exact: true }) + .click(); + await page + .getByRole("combobox") + .filter({ hasText: "Select integration" }) + .click(); + await page + .getByRole("option", { name: /Langfuse OTEL/ }) + .click(); + const spanScope = page.getByRole("combobox", { + name: "otel span scope", + }); + await spanScope.click(); + await expect( + page.getByRole("option", { name: "full", exact: true }), + ).toBeVisible(); + await page + .getByRole("option", { name: "no_internal", exact: true }) + .click(); + const update = await captureRequestBody( + page, + { method: "POST", urlIncludes: "/team/update" }, + async () => { + await page.getByRole("button", { name: "Save Changes" }).click(); + }, + ); + const logging = (update.metadata as { logging: any[] }).logging; + expect(logging).toEqual([ + { + callback_name: "langfuse_otel", + callback_type: "success", + callback_vars: { otel_span_scope: "no_internal" }, + }, + ]); + await expect( + page.getByText("Team settings updated successfully"), + ).toBeVisible(); + const info = await readBack<{ + team_info: { metadata: { logging: any[] } }; + }>(page, `/team/info?team_id=${team.team_id}`); + expect( + info.team_info.metadata.logging[0].callback_vars.otel_span_scope, + ).toBe("no_internal"); + } finally { + await post("/team/delete", { team_ids: [team.team_id] }); + } +}); diff --git a/tests/integration/observability/test_otel_tenant_internal_spans.py b/tests/integration/observability/test_otel_tenant_span_scope.py similarity index 82% rename from tests/integration/observability/test_otel_tenant_internal_spans.py rename to tests/integration/observability/test_otel_tenant_span_scope.py index d1a14d89222..bd98f192895 100644 --- a/tests/integration/observability/test_otel_tenant_internal_spans.py +++ b/tests/integration/observability/test_otel_tenant_span_scope.py @@ -37,7 +37,7 @@ from pydantic import JsonValue AuditConfigWriter = Callable[[Path, Mapping[str, JsonValue]], Path] ARIZE_VARS: Final[dict[str, str]] = {"arize_space_id": "audit-space", "arize_api_key": "audit-arize-key"} -INTERNAL_SPANS_VAR: Final = "otel_internal_spans" +SPAN_SCOPE_VAR: Final = "otel_span_scope" @pytest.fixture(scope="module") @@ -140,7 +140,7 @@ def _classes(spans: tuple[Span, ...]) -> dict[str, int]: return {name: sum(1 for span in spans if span_class(span) == name) for name in ("root", "tenant", "internal")} -def _assert_excluded(trace_spans: tuple[Span, ...]) -> None: +def _assert_no_internal(trace_spans: tuple[Span, ...]) -> None: counts: Final = _classes(trace_spans) names: Final = sorted(str(span["name"]) for span in trace_spans) assert counts["root"] == 1 and counts["internal"] == 0 and counts["tenant"] >= 1, ( @@ -156,13 +156,13 @@ def _assert_full(trace_spans: tuple[Span, ...]) -> None: ) -def _excluded_flow( +def _no_internal_flow( gateway: Gateway, upstream_url: str, langfuse_vars: Mapping[str, JsonValue], send: Callable[[Gateway, str, str, str], httpx.Response], *, - internal_spans: str | None = "exclude", + span_scope: str | None = "no_internal", callback_vars: Mapping[str, JsonValue] | None = None, ) -> tuple[httpx.Response, str]: with gateway.scenario() as scenario: @@ -171,7 +171,7 @@ def _excluded_flow( vars: Final = { **dict(langfuse_vars), **(dict(callback_vars) if callback_vars else {}), - **({INTERNAL_SPANS_VAR: internal_spans} if internal_spans is not None else {}), + **({SPAN_SCOPE_VAR: span_scope} if span_scope is not None else {}), } response: Final = _add_callback(gateway, team_id, vars) assert response.status_code == 200, f"callback setup failed: {response.status_code} {response.text}" @@ -188,7 +188,7 @@ def _assert_trace_split(audit_sinks: SpanSinks, traffic: httpx.Response) -> None operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id, response_id=response_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id, response_id=response_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) assert tenant_trace == operator_trace @@ -281,7 +281,7 @@ def _responses_stream_id(response: httpx.Response) -> str | None: return None -def _send_and_assert_excluded( +def _send_and_assert_no_internal( gateway: Gateway, upstream_url: str, audit_sinks: SpanSinks, @@ -290,7 +290,7 @@ def _send_and_assert_excluded( ) -> None: traffic: Final[httpx.Response] nonce: Final[str] - traffic, nonce = _excluded_flow(gateway, upstream_url, langfuse_vars, send) + traffic, nonce = _no_internal_flow(gateway, upstream_url, langfuse_vars, send) _assert_trace_split(audit_sinks, traffic) _assert_upstream_saw(upstream_url, nonce) @@ -311,62 +311,62 @@ def _candidate( yield candidate -def test_team_exclude_chat_completions_openai_sync(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_no_internal_chat_completions_openai_sync(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, f"callback setup failed: {callback.status_code} {callback.text}" key: Final = _key_on_team(scenario, team_id) nonce = _nonce() response_id, call_id = _openai_sdk(gateway, key, model, nonce) tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id, response_id=response_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id, response_id=response_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) assert tenant_trace == operator_trace _assert_upstream_saw(gateway.upstream_url, nonce) -def test_team_exclude_chat_completions_stream_openai_async(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_no_internal_chat_completions_stream_openai_async(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) nonce: Final = _nonce() response_id, call_id = _openai_sdk_async_stream(gateway, key, model, nonce) assert response_id is not None, "stream produced no response id" tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id, response_id=response_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id, response_id=response_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) _assert_upstream_saw(gateway.upstream_url, nonce) -def test_team_exclude_messages_anthropic_sync(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_no_internal_messages_anthropic_sync(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) nonce: Final = _nonce() response_id, call_id = _anthropic_sdk(gateway, key, model, nonce) assert call_id is not None, "no x-litellm-call-id header on /v1/messages" tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) _assert_upstream_saw(gateway.upstream_url, nonce) -def test_team_exclude_messages_stream_anthropic_async(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_no_internal_messages_stream_anthropic_async(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) nonce: Final = _nonce() @@ -386,17 +386,17 @@ def test_team_exclude_messages_stream_anthropic_async(gateway: Gateway, audit_si call_id: Final = _call_id(response) assert call_id is not None tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) _assert_upstream_saw(gateway.upstream_url, nonce) -def test_team_exclude_responses_api(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_no_internal_responses_api(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) nonce: Final = _nonce() @@ -405,17 +405,17 @@ def test_team_exclude_responses_api(gateway: Gateway, audit_sinks: SpanSinks, la call_id: Final = _call_id(response) response_id: Final = _response_id(response.json()) tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id, response_id=response_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id, response_id=response_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) _assert_upstream_saw(gateway.upstream_url, nonce) -def test_team_exclude_responses_stream(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_no_internal_responses_stream(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) nonce: Final = _nonce() @@ -425,17 +425,17 @@ def test_team_exclude_responses_stream(gateway: Gateway, audit_sinks: SpanSinks, call_id: Final = _call_id(response) assert response_id is not None or call_id is not None, response.text[:400] tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id, response_id=response_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id, response_id=response_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) _assert_upstream_saw(gateway.upstream_url, nonce) -def test_team_include_explicit_delivers_full_trace(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_full_explicit_delivers_full_trace(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "include"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "full"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(gateway, key, model, _nonce()) @@ -446,7 +446,7 @@ def test_team_include_explicit_delivers_full_trace(gateway: Gateway, audit_sinks _assert_full(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_var_absent_defaults_to_include(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_var_absent_defaults_to_full(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() @@ -463,30 +463,30 @@ def _key_logging_entry(callback_vars: Mapping[str, JsonValue], callback_name: st return [{"callback_name": callback_name, "callback_vars": dict(callback_vars)}] -def test_key_level_exclude_without_team_callbacks(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_key_level_no_internal_without_team_callbacks(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) key: Final = scenario.key( - metadata={"logging": _key_logging_entry({**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"})} + metadata={"logging": _key_logging_entry({**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"})} ) response: Final = _chat(gateway, key, model, _nonce()) assert response.status_code == 200, response.text call_id: Final = _call_id(response) tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) -def test_key_include_wins_over_team_exclude(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_key_full_wins_over_team_no_internal(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = scenario.key( team_id=team_id, - metadata={"logging": _key_logging_entry({**langfuse_vars, INTERNAL_SPANS_VAR: "include"})}, + metadata={"logging": _key_logging_entry({**langfuse_vars, SPAN_SCOPE_VAR: "full"})}, ) response: Final = _chat(gateway, key, model, _nonce()) assert response.status_code == 200, response.text @@ -494,30 +494,30 @@ def test_key_include_wins_over_team_exclude(gateway: Gateway, audit_sinks: SpanS _assert_full(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_key_exclude_wins_over_team_include(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_key_no_internal_wins_over_team_full(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "include"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "full"}) assert callback.status_code == 200, callback.text key: Final = scenario.key( team_id=team_id, - metadata={"logging": _key_logging_entry({**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"})}, + metadata={"logging": _key_logging_entry({**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"})}, ) response: Final = _chat(gateway, key, model, _nonce()) assert response.status_code == 200, response.text tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_langfuse_exclude_arize_include_split(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_langfuse_no_internal_arize_full_split(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - langfuse: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + langfuse: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert langfuse.status_code == 200, langfuse.text arize: Final = _add_callback( - gateway, team_id, {**ARIZE_VARS, INTERNAL_SPANS_VAR: "include"}, callback_name="arize" + gateway, team_id, {**ARIZE_VARS, SPAN_SCOPE_VAR: "full"}, callback_name="arize" ) assert arize.status_code == 200, arize.text key: Final = _key_on_team(scenario, team_id) @@ -525,45 +525,66 @@ def test_langfuse_exclude_arize_include_split(gateway: Gateway, audit_sinks: Spa assert response.status_code == 200, response.text call_id: Final = _call_id(response) langfuse_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, langfuse_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, langfuse_trace)) arize_trace: Final = _trace_id(audit_sinks.arize, call_id=call_id) arize_spans: Final = _trace_spans(audit_sinks.arize, arize_trace) _assert_full(arize_spans) assert arize_trace == langfuse_trace -def test_llm_only_scope_under_exclude_keeps_model_span(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_llm_only_scope_on_arize_grpc_keeps_model_span( + gateway: Gateway, audit_sinks: SpanSinks, arize_grpc_sink: GrpcSink, otel_audit_config: AuditConfigWriter, tmp_path: Path +) -> None: + overrides: Final = { + "LITELLM_OTEL_V2": "1", + "ARIZE_ENDPOINT": arize_grpc_sink.url, + "ARIZE_HTTP_ENDPOINT": audit_sinks.arize, + } + with owned_proxy(gateway, tmp_path, overrides, config=otel_audit_config(tmp_path, {}), workers=2) as candidate: + with candidate.scenario() as scenario: + model: Final = _audit_model(scenario, candidate.upstream_url) + team_id: Final = scenario.team() + callback: Final = _add_callback( + candidate, + team_id, + {**ARIZE_VARS, SPAN_SCOPE_VAR: "llm_only"}, + callback_name="arize", + ) + assert callback.status_code == 200, callback.text + key: Final = _key_on_team(scenario, team_id) + response: Final = _chat(candidate, key, model, _nonce()) + assert response.status_code == 200, response.text + arize_trace: Final = _trace_id(arize_grpc_sink.control_url, call_id=_call_id(response)) + + def settled() -> tuple[Span, ...] | None: + _, spans = recorded_spans(arize_grpc_sink.control_url) + group: Final = spans_for_trace(spans, arize_trace) + return group if group else None + + group: Final = eventually(settled, lambda value: value is not None, seconds=30) + assert group is not None + names: Final = sorted(str(span["name"]) for span in group) + assert all("gen_ai.operation.name" in span["attributes"] for span in group), names + + +def test_langfuse_alias_and_otel_scope_disagree_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: - model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback( + response: Final = _add_callback( gateway, team_id, - {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude", "langfuse_span_scope": "llm_only"}, + {**langfuse_vars, "langfuse_span_scope": "llm_only", SPAN_SCOPE_VAR: "no_internal"}, ) - assert callback.status_code == 200, callback.text - key: Final = _key_on_team(scenario, team_id) - response: Final = _chat(gateway, key, model, _nonce()) - assert response.status_code == 200, response.text - tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - - def settled() -> tuple[Span, ...] | None: - _, spans = recorded_spans(audit_sinks.tenant) - group: Final = spans_for_trace(spans, tenant_trace) - return group if group else None - - group: Final = eventually(settled, lambda value: value is not None, seconds=30) - assert group is not None - names: Final = sorted(str(span["name"]) for span in group) - assert all("gen_ai.operation.name" in span["attributes"] for span in group), names + assert response.status_code == 400, f"expected 400, got {response.status_code}: {response.text}" + assert "span_scope" in response.text, response.text -def test_additive_mode_operator_full_tenant_excluded(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: +def test_additive_mode_operator_full_tenant_no_internal(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, settings={"otel_tenant_destination_mode": "additive"}) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(candidate, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(candidate, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(candidate, key, model, _nonce()) @@ -572,14 +593,14 @@ def test_additive_mode_operator_full_tenant_excluded(gateway: Gateway, audit_sin operator_trace: Final = _trace_id(audit_sinks.operator, call_id=call_id) _assert_full(_trace_spans(audit_sinks.operator, operator_trace)) tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_operator_sink_keeps_internal_spans_under_exclude(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_operator_sink_keeps_span_scope_under_no_internal(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(gateway, key, model, _nonce()) @@ -593,7 +614,7 @@ def test_operator_sink_keeps_internal_spans_under_exclude(gateway: Gateway, audi assert any("auth" in name or "redis" in name or "postgres" in name for name in internal_names), internal_names -def test_guardrail_span_survives_exclude(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: +def test_guardrail_span_survives_no_internal(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: guardrail_name: Final = f"audit-filter-{uuid.uuid4().hex[:8]}" config: Final = yaml.safe_load(otel_audit_config(tmp_path, {}).read_text()) config["guardrails"] = [ @@ -622,7 +643,7 @@ def test_guardrail_span_survives_exclude(gateway: Gateway, audit_sinks: SpanSink with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(candidate, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(candidate, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(candidate, key, model, _nonce()) @@ -645,8 +666,8 @@ def test_guardrail_span_survives_exclude(gateway: Gateway, audit_sinks: SpanSink assert kept is not None, f"guardrail span missing at tenant sink: {tenant_spans}" -def test_env_exclude_applies_when_var_absent(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_INTERNAL_SPANS": "exclude"}) as candidate: +def test_env_no_internal_applies_when_var_absent(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_SPAN_SCOPE": "no_internal"}) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() @@ -656,15 +677,15 @@ def test_env_exclude_applies_when_var_absent(gateway: Gateway, audit_sinks: Span response: Final = _chat(candidate, key, model, _nonce()) assert response.status_code == 200, response.text tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_var_include_beats_env_exclude(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_INTERNAL_SPANS": "exclude"}) as candidate: +def test_var_full_beats_env_no_internal(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_SPAN_SCOPE": "no_internal"}) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(candidate, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "include"}) + callback: Final = _add_callback(candidate, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "full"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(candidate, key, model, _nonce()) @@ -673,8 +694,8 @@ def test_var_include_beats_env_exclude(gateway: Gateway, audit_sinks: SpanSinks, _assert_full(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_litellm_setting_exclude_applies_when_var_absent(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, settings={"otel_tenant_internal_spans": "exclude"}) as candidate: +def test_litellm_setting_no_internal_applies_when_var_absent(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, settings={"otel_tenant_span_scope": "no_internal"}) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() @@ -684,11 +705,11 @@ def test_litellm_setting_exclude_applies_when_var_absent(gateway: Gateway, audit response: Final = _chat(candidate, key, model, _nonce()) assert response.status_code == 200, response.text tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_setting_include_beats_env_exclude(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_INTERNAL_SPANS": "exclude"}, settings={"otel_tenant_internal_spans": "include"} +def test_setting_full_beats_env_no_internal(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_SPAN_SCOPE": "no_internal"}, settings={"otel_tenant_span_scope": "full"} ) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) @@ -702,8 +723,8 @@ def test_setting_include_beats_env_exclude(gateway: Gateway, audit_sinks: SpanSi _assert_full(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_env_exclude_with_whitespace_and_case(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_INTERNAL_SPANS": " EXCLUDE "}) as candidate: +def test_env_no_internal_with_whitespace_and_case(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_SPAN_SCOPE": " NO_INTERNAL "}) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() @@ -713,11 +734,11 @@ def test_env_exclude_with_whitespace_and_case(gateway: Gateway, audit_sinks: Spa response: Final = _chat(candidate, key, model, _nonce()) assert response.status_code == 200, response.text tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_env_bogus_value_falls_back_to_include(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_INTERNAL_SPANS": "bogus"}) as candidate: +def test_env_bogus_value_falls_back_to_full(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_SPAN_SCOPE": "bogus"}) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) team_id: Final = scenario.team() @@ -731,7 +752,7 @@ def test_env_bogus_value_falls_back_to_include(gateway: Gateway, audit_sinks: Sp def test_empty_setting_falls_through_to_env(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue], otel_audit_config: AuditConfigWriter, tmp_path: Path) -> None: - with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_INTERNAL_SPANS": "exclude"}, settings={"otel_tenant_internal_spans": ""} + with _candidate(gateway, tmp_path, audit_sinks, otel_audit_config, env={"LITELLM_OTEL_TENANT_SPAN_SCOPE": "no_internal"}, settings={"otel_tenant_span_scope": ""} ) as candidate: with candidate.scenario() as scenario: model: Final = _audit_model(scenario, candidate.upstream_url) @@ -742,18 +763,18 @@ def test_empty_setting_falls_through_to_env(gateway: Gateway, audit_sinks: SpanS response: Final = _chat(candidate, key, model, _nonce()) assert response.status_code == 200, response.text tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) def _bad_var_rejected(response: httpx.Response) -> None: assert response.status_code in (400, 422), f"expected rejection, got {response.status_code}: {response.text}" - assert INTERNAL_SPANS_VAR in response.text or "callback" in response.text, response.text + assert SPAN_SCOPE_VAR in response.text or "callback" in response.text, response.text def test_callback_var_int_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() - response: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: 1}) + response: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: 1}) _bad_var_rejected(response) @@ -761,7 +782,7 @@ def test_callback_var_list_rejected(gateway: Gateway, langfuse_vars: dict[str, J with gateway.scenario() as scenario: team_id: Final = scenario.team() response: Final = _add_callback( - gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: ["exclude"]} + gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: ["no_internal"]} ) _bad_var_rejected(response) @@ -769,100 +790,100 @@ def test_callback_var_list_rejected(gateway: Gateway, langfuse_vars: dict[str, J def test_callback_var_empty_string_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() - response: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: ""}) + response: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: ""}) _bad_var_rejected(response) def test_callback_var_oversized_string_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() - response: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "x" * 5120}) + response: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "x" * 5120}) _bad_var_rejected(response) def test_callback_var_case_sensitive_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() - response: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "Exclude"}) + response: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "No_Internal"}) _bad_var_rejected(response) -def test_same_internal_spans_value_on_second_entry_accepted(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_same_span_scope_value_on_second_entry_accepted(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() first: Final = _add_callback( - gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}, callback_type="success_and_failure" + gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}, callback_type="success_and_failure" ) assert first.status_code == 200, first.text second: Final = _add_callback( - gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}, callback_type="success" + gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}, callback_type="success" ) assert second.status_code == 200, f"identical value rejected: {second.status_code} {second.text}" listed: Final = gateway.get(f"/team/{team_id}/callback") data: Final = object_value(listed["data"]) assert "langfuse_otel" in data.get("success_callbacks", []), data - assert object_value(data["callback_vars"]).get(INTERNAL_SPANS_VAR) == "exclude", data + assert object_value(data["callback_vars"]).get(SPAN_SCOPE_VAR) == "no_internal", data key: Final = _key_on_team(scenario, team_id) response: Final = _chat(gateway, key, model, _nonce()) assert response.status_code == 200, response.text tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=_call_id(response)) - _assert_excluded(_trace_spans(audit_sinks.tenant, tenant_trace)) + _assert_no_internal(_trace_spans(audit_sinks.tenant, tenant_trace)) -def test_conflicting_internal_spans_value_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: +def test_conflicting_span_scope_value_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() first: Final = _add_callback( - gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}, callback_type="success_and_failure" + gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}, callback_type="success_and_failure" ) assert first.status_code == 200, first.text second: Final = _add_callback( - gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "include"}, callback_type="success" + gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "full"}, callback_type="success" ) assert second.status_code == 400, f"expected 400 conflict, got {second.status_code}: {second.text}" - assert INTERNAL_SPANS_VAR in second.text, second.text + assert SPAN_SCOPE_VAR in second.text, second.text -def test_internal_spans_on_non_otel_callback_rejected(gateway: Gateway) -> None: +def test_span_scope_on_non_otel_callback_rejected(gateway: Gateway) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() response: Final = _add_callback( gateway, team_id, - {"langsmith_api_key": "sk-ls-audit", INTERNAL_SPANS_VAR: "exclude"}, + {"langsmith_api_key": "sk-ls-audit", SPAN_SCOPE_VAR: "no_internal"}, callback_name="langsmith", ) assert response.status_code == 400, f"expected 400, got {response.status_code}: {response.text}" assert "callback" in response.text, response.text -def test_internal_spans_on_newrelic_accepted(gateway: Gateway) -> None: +def test_span_scope_on_newrelic_accepted(gateway: Gateway) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() response: Final = _add_callback( gateway, team_id, - {"newrelic_api_key": "nr-audit-key", INTERNAL_SPANS_VAR: "exclude"}, + {"newrelic_api_key": "nr-audit-key", SPAN_SCOPE_VAR: "no_internal"}, callback_name="newrelic", ) assert response.status_code == 200, f"expected 200, got {response.status_code}: {response.text}" listed: Final = gateway.get(f"/team/{team_id}/callback") data: Final = object_value(listed["data"]) - assert object_value(data["callback_vars"]).get(INTERNAL_SPANS_VAR) == "exclude", data + assert object_value(data["callback_vars"]).get(SPAN_SCOPE_VAR) == "no_internal", data assert "newrelic" in data.get("success_callbacks", []) or "newrelic" in data.get("failure_callbacks", []), data -def test_newrelic_exclude_over_connect( +def test_newrelic_no_internal_over_connect( gateway: Gateway, audit_sinks: SpanSinks, newrelic_sink: ConnectSink, otel_audit_config: AuditConfigWriter, tmp_path: Path ) -> None: - _newrelic_flow(gateway, audit_sinks, newrelic_sink, otel_audit_config, tmp_path, internal_spans="exclude") + _newrelic_flow(gateway, audit_sinks, newrelic_sink, otel_audit_config, tmp_path, span_scope="no_internal") -def test_newrelic_include_over_connect( +def test_newrelic_full_over_connect( gateway: Gateway, audit_sinks: SpanSinks, newrelic_sink: ConnectSink, otel_audit_config: AuditConfigWriter, tmp_path: Path ) -> None: - _newrelic_flow(gateway, audit_sinks, newrelic_sink, otel_audit_config, tmp_path, internal_spans="include") + _newrelic_flow(gateway, audit_sinks, newrelic_sink, otel_audit_config, tmp_path, span_scope="full") def _newrelic_flow( @@ -872,7 +893,7 @@ def _newrelic_flow( otel_audit_config: AuditConfigWriter, tmp_path: Path, *, - internal_spans: str, + span_scope: str, ) -> None: overrides: Final = { "LITELLM_OTEL_V2": "1", @@ -888,7 +909,7 @@ def _newrelic_flow( callback: Final = _add_callback( candidate, team_id, - {"newrelic_api_key": "nr-synthetic-audit", INTERNAL_SPANS_VAR: internal_spans}, + {"newrelic_api_key": "nr-synthetic-audit", SPAN_SCOPE_VAR: span_scope}, callback_name="newrelic", ) assert callback.status_code == 200, callback.text @@ -898,8 +919,8 @@ def _newrelic_flow( call_id: Final = _call_id(response) trace: Final = _trace_id(newrelic_sink.control_url, call_id=call_id) group: Final = _trace_spans(newrelic_sink.control_url, trace) - if internal_spans == "exclude": - _assert_excluded(group) + if span_scope == "no_internal": + _assert_no_internal(group) else: _assert_full(group) requests: Final = recorded_requests(newrelic_sink.control_url) @@ -912,16 +933,16 @@ def _newrelic_flow( assert isinstance(headers, Mapping) and headers.get("api-key") == "nr-synthetic-audit", entry -def test_arize_grpc_exclude( +def test_arize_grpc_no_internal( gateway: Gateway, audit_sinks: SpanSinks, arize_grpc_sink: GrpcSink, otel_audit_config: AuditConfigWriter, tmp_path: Path ) -> None: - _arize_grpc_flow(gateway, audit_sinks, arize_grpc_sink, otel_audit_config, tmp_path, internal_spans="exclude") + _arize_grpc_flow(gateway, audit_sinks, arize_grpc_sink, otel_audit_config, tmp_path, span_scope="no_internal") -def test_arize_grpc_include( +def test_arize_grpc_full( gateway: Gateway, audit_sinks: SpanSinks, arize_grpc_sink: GrpcSink, otel_audit_config: AuditConfigWriter, tmp_path: Path ) -> None: - _arize_grpc_flow(gateway, audit_sinks, arize_grpc_sink, otel_audit_config, tmp_path, internal_spans="include") + _arize_grpc_flow(gateway, audit_sinks, arize_grpc_sink, otel_audit_config, tmp_path, span_scope="full") def _arize_grpc_flow( @@ -931,7 +952,7 @@ def _arize_grpc_flow( otel_audit_config: AuditConfigWriter, tmp_path: Path, *, - internal_spans: str, + span_scope: str, ) -> None: overrides: Final = { "LITELLM_OTEL_V2": "1", @@ -945,7 +966,7 @@ def _arize_grpc_flow( callback: Final = _add_callback( candidate, team_id, - {**ARIZE_VARS, INTERNAL_SPANS_VAR: internal_spans}, + {**ARIZE_VARS, SPAN_SCOPE_VAR: span_scope}, callback_name="arize", ) assert callback.status_code == 200, callback.text @@ -955,8 +976,8 @@ def _arize_grpc_flow( call_id: Final = _call_id(response) trace: Final = _trace_id(arize_grpc_sink.control_url, call_id=call_id) group: Final = _trace_spans(arize_grpc_sink.control_url, trace) - if internal_spans == "exclude": - _assert_excluded(group) + if span_scope == "no_internal": + _assert_no_internal(group) else: _assert_full(group) requests: Final = recorded_requests(arize_grpc_sink.control_url) @@ -977,41 +998,41 @@ def test_unauthenticated_callback_post_rejected(gateway: Gateway, langfuse_vars: assert response.status_code == 401, f"expected 401, got {response.status_code}: {response.text}" -def test_key_generate_bogus_internal_spans_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: +def test_key_generate_bogus_span_scope_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: response: Final = gateway.request( - "POST", "/key/generate", {"metadata": {"logging": _key_logging_entry({**langfuse_vars, INTERNAL_SPANS_VAR: "bogus"})}} + "POST", "/key/generate", {"metadata": {"logging": _key_logging_entry({**langfuse_vars, SPAN_SCOPE_VAR: "bogus"})}} ) assert response.status_code == 400, f"expected 400, got {response.status_code}: {response.text}" -def test_key_update_bogus_internal_spans_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: +def test_key_update_bogus_span_scope_rejected(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: key: Final = scenario.key() response: Final = gateway.request( "POST", "/key/update", - {"key": key, "metadata": {"logging": _key_logging_entry({**langfuse_vars, INTERNAL_SPANS_VAR: "bogus"})}}, + {"key": key, "metadata": {"logging": _key_logging_entry({**langfuse_vars, SPAN_SCOPE_VAR: "bogus"})}}, ) assert response.status_code == 400, f"expected 400, got {response.status_code}: {response.text}" -def test_team_logging_metadata_rejects_bogus_internal_spans(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_team_logging_metadata_rejects_bogus_span_scope(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() registered: Final = _add_callback(gateway, team_id, langfuse_vars) assert registered.status_code == 200, registered.text bogus_metadata: Final[Mapping[str, JsonValue]] = { - "logging": _key_logging_entry({**langfuse_vars, INTERNAL_SPANS_VAR: "bogus"}) + "logging": _key_logging_entry({**langfuse_vars, SPAN_SCOPE_VAR: "bogus"}) } update: Final = gateway.request( "POST", "/team/update", {"team_id": team_id, "metadata": bogus_metadata} ) assert update.status_code == 400, update.text - assert "otel_internal_spans" in update.text, update.text + assert "otel_span_scope" in update.text, update.text created: Final = gateway.request("POST", "/team/new", {"metadata": bogus_metadata}) assert created.status_code == 400, created.text - assert "otel_internal_spans" in created.text, created.text + assert "otel_span_scope" in created.text, created.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(gateway, key, model, _nonce()) assert response.status_code == 200, response.text @@ -1065,7 +1086,7 @@ def test_tenant_sink_404_does_not_break_caller( _sink_status_flow(gateway, audit_sinks, langfuse_vars, 404) -def test_upstream_500_under_exclude_keeps_internal_spans_back(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_upstream_500_under_no_internal_keeps_span_scope_back(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: upstream_model: Final = "audit-chat" httpx.post( f"{gateway.upstream_url}/__scripts/{upstream_model}", json={"statuses": [500]}, trust_env=False, timeout=15 @@ -1074,7 +1095,7 @@ def test_upstream_500_under_exclude_keeps_internal_spans_back(gateway: Gateway, with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = _chat(gateway, key, model, _nonce()) @@ -1111,21 +1132,21 @@ def test_unrelated_key_unaffected_by_tenant_sink_failure(gateway: Gateway, audit configure_sink(audit_sinks.tenant, status=200) -def test_key_health_with_excluded_team_callback(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: +def test_key_health_with_no_internal_team_callback(gateway: Gateway, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) response: Final = gateway.request("POST", "/key/health", key=key) assert response.status_code == 200, f"/key/health failed: {response.status_code} {response.text}" -def test_callback_var_update_include_to_exclude_takes_effect(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_callback_var_update_full_to_no_internal_takes_effect(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - first: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "include"}) + first: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "full"}) assert first.status_code == 200, first.text key: Final = _key_on_team(scenario, team_id) warm: Final = _chat(gateway, key, model, _nonce()) @@ -1134,7 +1155,7 @@ def test_callback_var_update_include_to_exclude_takes_effect(gateway: Gateway, a _assert_full(_trace_spans(audit_sinks.tenant, warm_trace)) deleted: Final = gateway.request("DELETE", f"/team/{team_id}/callback/langfuse_otel") assert deleted.status_code == 200, deleted.text - updated: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + updated: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert updated.status_code == 200, updated.text issued: Final[list[str | None]] = [] @@ -1161,7 +1182,7 @@ def test_callback_var_update_include_to_exclude_takes_effect(gateway: Gateway, a return None trace: Final = eventually(flipped, lambda value: value is not None, seconds=70) - assert trace is not None, "exclude never took effect within the cache TTL" + assert trace is not None, "no_internal never took effect within the cache TTL" def test_callback_delete_stops_tenant_export(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: @@ -1198,7 +1219,7 @@ def test_identical_requests_export_exactly_once(gateway: Gateway, audit_sinks: S with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) call_ids: Final = [] @@ -1227,12 +1248,12 @@ def test_identical_requests_export_exactly_once(gateway: Gateway, audit_sinks: S assert len(operator_matching) == 1, f"operator exported {response_id} {len(operator_matching)} times" -def test_concurrent_requests_all_excluded_once(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: +def test_concurrent_requests_all_no_internal_once(gateway: Gateway, audit_sinks: SpanSinks, langfuse_vars: dict[str, JsonValue]) -> None: with gateway.scenario() as scenario: model: Final = _audit_model(scenario, gateway.upstream_url) team_id: Final = scenario.team() - callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}) + callback: Final = _add_callback(gateway, team_id, {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}) assert callback.status_code == 200, callback.text key: Final = _key_on_team(scenario, team_id) responses: Final[list[httpx.Response]] = [] @@ -1251,7 +1272,7 @@ def test_concurrent_requests_all_excluded_once(gateway: Gateway, audit_sinks: Sp call_id: Final = _call_id(response) tenant_trace: Final = _trace_id(audit_sinks.tenant, call_id=call_id) group: Final = _trace_spans(audit_sinks.tenant, tenant_trace) - _assert_excluded(group) + _assert_no_internal(group) workers: Final = { str(span["resource"].get("process.pid")) for span in group } @@ -1265,7 +1286,7 @@ def test_failure_only_callback_entry_anchors_no_destination(gateway: Gateway, au callback: Final = _add_callback( gateway, team_id, - {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}, + {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}, callback_type="failure", ) assert callback.status_code == 200, callback.text diff --git a/tests/integration/observability/test_otel_tenant_internal_spans_chaos.py b/tests/integration/observability/test_otel_tenant_span_scope_chaos.py similarity index 97% rename from tests/integration/observability/test_otel_tenant_internal_spans_chaos.py rename to tests/integration/observability/test_otel_tenant_span_scope_chaos.py index 3a406e0a72a..3f29f5d9883 100644 --- a/tests/integration/observability/test_otel_tenant_internal_spans_chaos.py +++ b/tests/integration/observability/test_otel_tenant_span_scope_chaos.py @@ -16,7 +16,7 @@ from integration._support.process import owned_proxy, owned_proxy_process from pydantic import JsonValue AuditConfigWriter = Callable[[Path, Mapping[str, JsonValue]], Path] -INTERNAL_SPANS_VAR: Final = "otel_internal_spans" +SPAN_SCOPE_VAR: Final = "otel_span_scope" @pytest.fixture(scope="module") @@ -102,7 +102,7 @@ def test_frozen_tenant_sink_receives_every_span_after_resume(gateway: Gateway, a model: Final = scenario.model(model="openai/audit-chat", api_base=f"{gateway.upstream_url}/v1") team_id: Final = scenario.team() callback: Final = gateway.request( - "POST", f"/team/{team_id}/callback", {"callback_name": "langfuse_otel", "callback_vars": {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}} + "POST", f"/team/{team_id}/callback", {"callback_name": "langfuse_otel", "callback_vars": {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}} ) assert callback.status_code == 200, callback.text key: Final = scenario.key(team_id=team_id) @@ -130,7 +130,7 @@ def test_sink_outage_keeps_diagnostics_green_and_recovers(gateway: Gateway, audi model: Final = scenario.model(model="openai/audit-chat", api_base=f"{gateway.upstream_url}/v1") team_id: Final = scenario.team() callback: Final = gateway.request( - "POST", f"/team/{team_id}/callback", {"callback_name": "langfuse_otel", "callback_vars": {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}} + "POST", f"/team/{team_id}/callback", {"callback_name": "langfuse_otel", "callback_vars": {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}} ) assert callback.status_code == 200, callback.text key: Final = scenario.key(team_id=team_id) @@ -166,7 +166,7 @@ def test_slow_tenant_sink_exports_each_span_once(gateway: Gateway, audit_sinks: model: Final = scenario.model(model="openai/audit-chat", api_base=f"{gateway.upstream_url}/v1") team_id: Final = scenario.team() callback: Final = gateway.request( - "POST", f"/team/{team_id}/callback", {"callback_name": "langfuse_otel", "callback_vars": {**langfuse_vars, INTERNAL_SPANS_VAR: "exclude"}} + "POST", f"/team/{team_id}/callback", {"callback_name": "langfuse_otel", "callback_vars": {**langfuse_vars, SPAN_SCOPE_VAR: "no_internal"}} ) assert callback.status_code == 200, callback.text key: Final = scenario.key(team_id=team_id)