diff --git a/litellm/__init__.py b/litellm/__init__.py index 1784deeb797..984e3e7cde7 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -156,6 +156,7 @@ _custom_logger_compatible_callbacks_literal = Literal[ "compression_interception", "newrelic", "signoz", + "nops", ] cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None diff --git a/litellm/integrations/callback_configs.json b/litellm/integrations/callback_configs.json index 38928f67f42..4ccb506dfca 100644 --- a/litellm/integrations/callback_configs.json +++ b/litellm/integrations/callback_configs.json @@ -330,6 +330,27 @@ }, "description": "New Relic AI Monitoring Integration" }, + { + "id": "nops", + "displayName": "nOps", + "logo": "nops.png", + "supports_key_team_logging": false, + "dynamic_params": { + "NOPS_API_KEY": { + "type": "password", + "ui_name": "API Key", + "description": "nOps API key with the telemetry:write permission (Inform + Operate plan)", + "required": true + }, + "NOPS_OTEL_ENDPOINT": { + "type": "text", + "ui_name": "OTel Endpoint", + "description": "nOps OTLP endpoint. Leave blank to use https://api.nops.io/otel", + "required": false + } + }, + "description": "nOps Logging Integration - sends LiteLLM traces (cost, tokens, model, team) to nOps" + }, { "id": "openmeter", "displayName": "OpenMeter", diff --git a/litellm/integrations/otel/model/config.py b/litellm/integrations/otel/model/config.py index cca12b885c1..a13e9bb6f22 100644 --- a/litellm/integrations/otel/model/config.py +++ b/litellm/integrations/otel/model/config.py @@ -44,6 +44,7 @@ class ExporterOwner(str, Enum): AGENTOPS = "agentops" NEWRELIC = "newrelic" SIGNOZ = "signoz" + NOPS = "nops" class _OTelV2Flag(BaseSettings): diff --git a/litellm/integrations/otel/presets/__init__.py b/litellm/integrations/otel/presets/__init__.py index 7c891c29409..ed6b07203bd 100644 --- a/litellm/integrations/otel/presets/__init__.py +++ b/litellm/integrations/otel/presets/__init__.py @@ -26,6 +26,7 @@ from litellm.integrations.otel.presets.newrelic import ( newrelic_dynamic_headers, newrelic_preset, ) +from litellm.integrations.otel.presets.nops import nops_preset from litellm.integrations.otel.presets.phoenix import ( phoenix_preset, phoenix_project_headers, @@ -49,6 +50,7 @@ PRESET_BY_CALLBACK: Final[Mapping[str, Preset]] = MappingProxyType( "langtrace": langtrace_preset, "levo": levo_preset, "newrelic": newrelic_preset, + "nops": nops_preset, "signoz": signoz_preset, "weave_otel": weave_preset, } @@ -159,6 +161,7 @@ __all__ = [ "langtrace_preset", "levo_preset", "newrelic_preset", + "nops_preset", "phoenix_preset", "project_routing_headers", "signoz_preset", diff --git a/litellm/integrations/otel/presets/nops.py b/litellm/integrations/otel/presets/nops.py new file mode 100644 index 00000000000..2f351cabdda --- /dev/null +++ b/litellm/integrations/otel/presets/nops.py @@ -0,0 +1,48 @@ +from types import MappingProxyType +from typing import Final + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + +from litellm.integrations.otel.model.config import ( + ExporterOwner, + ExporterSpec, + OpenTelemetryV2Config, +) +from litellm.integrations.otel.presets.utils import ensure_mappers + +NOPS_API_KEY_ENV: Final = "NOPS_API_KEY" +NOPS_OTEL_ENDPOINT_ENV: Final = "NOPS_OTEL_ENDPOINT" +NOPS_DEFAULT_OTEL_ENDPOINT: Final = "https://api.nops.io/otel" + + +class _NopsSettings(BaseSettings): + model_config = SettingsConfigDict(case_sensitive=False, extra="ignore", frozen=True) + + endpoint: str | None = Field(default=None, validation_alias=NOPS_OTEL_ENDPOINT_ENV) + api_key: str | None = Field(default=None, validation_alias=NOPS_API_KEY_ENV) + + +def nops_preset( + *, + config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, +) -> OpenTelemetryV2Config: + settings: Final = _NopsSettings() + base: Final = config_overrides or OpenTelemetryV2Config() + key: Final = settings.api_key + spec: Final = ExporterSpec( + kind="otlp_http", + endpoint=settings.endpoint or NOPS_DEFAULT_OTEL_ENDPOINT, + headers=(f"Authorization=Bearer {key}" if key else None), + owner=ExporterOwner.NOPS, + requires_headers=True, + ) + return base.model_copy( + update=MappingProxyType( + { + "exporters": (*base.exporters, spec), + "mapper_names": ensure_mappers(base.mapper_names, "genai"), + } + ) + ) diff --git a/litellm/litellm_core_utils/custom_logger_registry.py b/litellm/litellm_core_utils/custom_logger_registry.py index 74badc2c8c1..c20d77e3e0d 100644 --- a/litellm/litellm_core_utils/custom_logger_registry.py +++ b/litellm/litellm_core_utils/custom_logger_registry.py @@ -97,6 +97,7 @@ class CustomLoggerRegistry: "weave_otel": OpenTelemetry, "levo": OpenTelemetry, "signoz": OpenTelemetry, + "nops": OpenTelemetry, "mlflow": MlflowLogger, "langfuse": LangfusePromptManagement, "otel": OpenTelemetry, diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 45349bb0e5a..94c5cc057bd 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -5541,6 +5541,44 @@ def build_custom_logger_compatible_class( _in_memory_loggers.append(_signoz_logger) return _signoz_logger + elif logging_integration == "nops": + from litellm.integrations.otel.presets.nops import ( + NOPS_API_KEY_ENV, + NOPS_DEFAULT_OTEL_ENDPOINT, + NOPS_OTEL_ENDPOINT_ENV, + ) + + _nops_key: Final = os.getenv(NOPS_API_KEY_ENV) + if not _nops_key: + raise ValueError(f"{NOPS_API_KEY_ENV} not found in environment variables") + + _nops_v2: Final = _maybe_construct_otel_v2("nops", _in_memory_loggers) + if _nops_v2 is not None: + return _nops_v2 + + from litellm.integrations.opentelemetry import ( + CAPTURE_MODE_NO_CONTENT, + OpenTelemetry, + OpenTelemetryConfig, + ) + + _nops_base: Final = (os.getenv(NOPS_OTEL_ENDPOINT_ENV) or NOPS_DEFAULT_OTEL_ENDPOINT).rstrip("/") + _nops_config: Final = OpenTelemetryConfig( + exporter="otlp_http", + skip_set_global=True, + endpoint=(_nops_base if _nops_base.endswith("/v1/traces") else f"{_nops_base}/v1/traces"), + headers=f"Authorization=Bearer {_nops_key}", + capture_message_content=( + os.getenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT") or CAPTURE_MODE_NO_CONTENT + ), + ) + for callback in _in_memory_loggers: + if isinstance(callback, OpenTelemetry) and callback.callback_name == "nops": + return callback + _nops_logger: Final = OpenTelemetry(config=_nops_config, callback_name="nops") + _in_memory_loggers.append(_nops_logger) + return _nops_logger + elif logging_integration == "mlflow": for callback in _in_memory_loggers: if isinstance(callback, MlflowLogger): diff --git a/litellm/proxy/_experimental/out/assets/logos/nops.png b/litellm/proxy/_experimental/out/assets/logos/nops.png new file mode 100644 index 00000000000..3aa8e0a6b31 Binary files /dev/null and b/litellm/proxy/_experimental/out/assets/logos/nops.png differ diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 54acf59ef80..a2c2a864a90 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -3653,6 +3653,12 @@ class AllCallbacks(LiteLLMPydanticObjectBase): ], ) + nops: CallbackOnUI = CallbackOnUI( + litellm_callback_name="nops", + ui_callback_name="nOps", + litellm_callback_params=("NOPS_API_KEY", "NOPS_OTEL_ENDPOINT"), + ) + pointfive: CallbackOnUI = CallbackOnUI( litellm_callback_name="pointfive", ui_callback_name="PointFive", diff --git a/litellm/proxy/health_endpoints/_health_endpoints.py b/litellm/proxy/health_endpoints/_health_endpoints.py index 684f1619512..f9e23eb8a27 100644 --- a/litellm/proxy/health_endpoints/_health_endpoints.py +++ b/litellm/proxy/health_endpoints/_health_endpoints.py @@ -269,6 +269,7 @@ services = ( "arize", "galileo", "newrelic", + "nops", "pointfive", "signoz", "sqs", diff --git a/tests/integration/observability/test_nops_delivery.py b/tests/integration/observability/test_nops_delivery.py new file mode 100644 index 00000000000..ecd8adc263c --- /dev/null +++ b/tests/integration/observability/test_nops_delivery.py @@ -0,0 +1,516 @@ +import asyncio +import base64 +import json +import re +import threading +import uuid +from collections import deque +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import anthropic +import httpx +import openai +import pytest +import yaml +from integration._support.client import Gateway, eventually, gateway_from_environment, object_value, string_value +from integration._support.database import read_rows +from integration._support.process import OwnedProxy, owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest +from opentelemetry.proto.common.v1.common_pb2 import AnyValue +from pydantic import JsonValue, TypeAdapter + +MARKER: Final = re.compile(rb"nops-[0-9a-f]{32}") +JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) +RESPONSE_ID: Final = "gen_ai.response.id" +REPLY_TEXT: Final = "nops upstream reply text" +OPERATOR_KEY: Final = "nops-operator-" + uuid.uuid4().hex + + +def _marker() -> str: + return "nops-" + uuid.uuid4().hex + + +def _chat_reply(identity: str, stream: bool) -> Reply: + if not stream: + return Reply( + body=json.dumps( + { + "id": identity, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": REPLY_TEXT}, "finish_reason": "stop"} + ], + "usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}, + } + ).encode() + ) + chunk: Final = {"id": identity, "object": "chat.completion.chunk", "created": 1, "model": "gpt-4o-mini"} + return Reply( + content_type="text/event-stream", + chunks=( + b"data: " + + json.dumps( + {**chunk, "choices": [{"index": 0, "delta": {"role": "assistant", "content": REPLY_TEXT}}]} + ).encode() + + b"\n\n", + b"data: " + + json.dumps({**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}).encode() + + b"\n\n", + b"data: " + + json.dumps( + {**chunk, "choices": [], "usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}} + ).encode() + + b"\n\n", + b"data: [DONE]\n\n", + ), + ) + + +def _responses_reply(identity: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": identity, + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-4o-mini", + "output": [ + { + "id": "msg_" + identity, + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": REPLY_TEXT, "annotations": []}], + } + ], + "usage": {"input_tokens": 7, "output_tokens": 2, "total_tokens": 9}, + } + ).encode() + ) + + +def _upstream(request: Request) -> Reply: + found: Final = MARKER.search(request.body) + if found is None: + return Reply(status=404, body=b'{"error":"no marker"}') + marker: Final = found.group(0).decode() + if request.target.endswith("/responses"): + return _responses_reply(f"resp_{marker}") + stream: Final = object_value(JSON.validate_json(request.body)).get("stream") is True + return _chat_reply(f"chatcmpl-{marker}", stream) + + +def _decoded_responses_id(identity: str) -> str: + try: + return base64.b64decode(identity.removeprefix("resp_").encode()).decode() + except (ValueError, UnicodeDecodeError): + return identity + + +def _canonical_id(identity: str) -> str: + return _decoded_responses_id(identity).rpartition("response_id:")[2] + + +def _text_at(payload: JsonValue, *path: str) -> str: + if not path: + return string_value(payload) + return _text_at(object_value(payload)[path[0]], *path[1:]) + + +def _body_id(response: httpx.Response) -> str: + return _text_at(JSON.validate_json(response.content), "id") + + +@dataclass(frozen=True, slots=True) +class Span: + target: str + authorization: str | None + attributes: Mapping[str, str] + + +def _attribute_text(value: AnyValue) -> str: + match value.WhichOneof("value"): + case "string_value": + return value.string_value + case "int_value": + return str(value.int_value) + case "double_value": + return str(value.double_value) + case "bool_value": + return str(value.bool_value) + case _: + return "" + + +@dataclass(frozen=True, slots=True) +class Collector: + wire: Wire + rejection: threading.Event + missing: threading.Event + accepted: Sequence[Request] + refused: Sequence[Request] + guard: threading.Lock + + def refused_batch_carrying(self, response_id: str) -> Request: + def carrying() -> tuple[Request, ...]: + with self.guard: + return tuple(batch for batch in self.refused if response_id.encode() in batch.body) + + return eventually(carrying, lambda found: len(found) >= 1, seconds=30)[0] + + def bodies(self) -> tuple[bytes, ...]: + with self.guard: + return tuple(batch.body for batch in self.accepted) + + def spans(self) -> tuple[Span, ...]: + with self.guard: + batches: Final = tuple(batch for batch in self.accepted if batch.target.endswith("/v1/traces")) + return tuple( + Span( + batch.target, + batch.headers.get("authorization"), + {attribute.key: _attribute_text(attribute.value) for attribute in span.attributes}, + ) + for batch in batches + for resource in ExportTraceServiceRequest.FromString(batch.body).resource_spans + for scope in resource.scope_spans + for span in scope.spans + ) + + def spans_for(self, response_id: str) -> tuple[Span, ...]: + return tuple( + span + for span in self.spans() + if RESPONSE_ID in span.attributes + and _canonical_id(span.attributes[RESPONSE_ID]) == _canonical_id(response_id) + ) + + def single_span(self, response_id: str) -> Span: + found: Final = eventually( + lambda: self.spans_for(response_id), lambda spans: len(spans) == 1, seconds=30, return_last_on_timeout=True + ) + assert len(found) == 1, f"{len(found)} spans for {response_id} at the sink" + return found[0] + + def landed(self, response_ids: Sequence[str]) -> dict[str, int]: + spans: Final = self.spans() + return { + _canonical_id(identity): sum( + 1 + for span in spans + if RESPONSE_ID in span.attributes + and _canonical_id(span.attributes[RESPONSE_ID]) == _canonical_id(identity) + ) + for identity in response_ids + } + + +@pytest.fixture(scope="session") +def sink() -> Iterator[Collector]: + rejection: Final = threading.Event() + missing: Final = threading.Event() + accepted: Final[deque[Request]] = deque() # mutable-ok: the sink thread records each accepted batch as it arrives + refused: Final[deque[Request]] = deque() # mutable-ok: the sink thread records each refused batch as it arrives + guard: Final = threading.Lock() + + def refuse(request: Request, status: int, body: bytes) -> Reply: + with guard: + refused.append(request) + return Reply(status=status, body=body) + + def receive(request: Request) -> Reply: + if rejection.is_set(): + return refuse(request, 403, b'{"error":"forbidden"}') + if missing.is_set(): + return refuse(request, 404, b'{"error":"not found"}') + with guard: + accepted.append(request) + return Reply() + + with wire_server(receive) as wire: + yield Collector(wire, rejection, missing, accepted, refused, guard) + + +@pytest.fixture(scope="session") +def provider() -> Iterator[Wire]: + with wire_server(_upstream) as wire: + yield wire + + +@dataclass(frozen=True, slots=True) +class Rig: + proxy: Gateway + process: OwnedProxy + model: str + upstream: Wire + sink: Collector + + def openai_client(self) -> openai.OpenAI: + return openai.OpenAI(base_url=str(self.proxy.client.base_url) + "/v1", api_key=self.proxy.key, max_retries=0) + + def async_openai_client(self) -> openai.AsyncOpenAI: + return openai.AsyncOpenAI( + base_url=str(self.proxy.client.base_url) + "/v1", api_key=self.proxy.key, max_retries=0 + ) + + def anthropic_client(self) -> anthropic.Anthropic: + return anthropic.Anthropic(base_url=str(self.proxy.client.base_url), api_key=self.proxy.key, max_retries=0) + + def chat(self, content: str) -> httpx.Response: + return self.proxy.request( + "POST", + "/v1/chat/completions", + {"model": self.model, "messages": [{"role": "user", "content": content}], "cache": {"no-cache": True}}, + ) + + def upstream_bodies(self, marker: str) -> tuple[dict[str, JsonValue], ...]: + return tuple( + object_value(JSON.validate_json(request.body)) + for request in self.upstream.drain() + if marker.encode() in request.body + ) + + def spend_rows(self, response_id: str) -> tuple[dict[str, JsonValue], ...]: + return tuple( + eventually( + lambda: read_rows( + 'SELECT request_id, spend FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (response_id,) + ), + lambda values: len(values) == 1, + seconds=70, + ) + ) + + +@dataclass(frozen=True, slots=True) +class RigFactory: + provider: Wire + sink: Collector + directory: Path + otel_v2: bool + endpoint: str + api_key: str | None = OPERATOR_KEY + + def config_path(self) -> Path: + loaded: Final = object_value( + JSON.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())) + ) + config: Final = { + **loaded, + "litellm_settings": {**object_value(loaded["litellm_settings"]), "callbacks": ["nops"]}, + } + path: Final = self.directory / f"nops-{uuid.uuid4().hex}.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + def overrides(self) -> dict[str, str]: + return { + "LITELLM_OTEL_V2": "1" if self.otel_v2 else "0", + "OTEL_BSP_SCHEDULE_DELAY": "300", + "NOPS_OTEL_ENDPOINT": self.endpoint, + **({"NOPS_API_KEY": self.api_key} if self.api_key is not None else {}), + } + + def start(self) -> Iterator[Rig]: + with ( + gateway_from_environment() as gateway, + owned_proxy_process( + gateway, + self.directory, + self.overrides(), + config=self.config_path(), + remove_environment=( + "NOPS_API_KEY", + "NOPS_OTEL_ENDPOINT", + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", + ), + workers=2, + ) as owned, + owned.gateway.scenario() as scenario, + ): + model: Final = scenario.model(api_base=self.provider.url + "/v1") + yield Rig(owned.gateway, owned, model, self.provider, self.sink) + + +@pytest.fixture(scope="session") +def rig(provider: Wire, sink: Collector, tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + yield from RigFactory(provider, sink, tmp_path_factory.mktemp("nops"), False, sink.wire.url).start() + + +@pytest.fixture(scope="session") +def v2_rig(provider: Wire, sink: Collector, tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + yield from RigFactory(provider, sink, tmp_path_factory.mktemp("nops-v2"), True, sink.wire.url).start() + + +def _assert_bearer_span(rig: Rig, response_id: str, marker: str) -> Span: + span: Final = rig.sink.single_span(response_id) + assert span.target == "/v1/traces", span + assert span.authorization == f"Bearer {OPERATOR_KEY}", span + bodies: Final = rig.upstream_bodies(marker) + assert len(bodies) == 1, bodies + assert OPERATOR_KEY not in json.dumps(bodies[0]), bodies[0] + return span + + +def test_nops_is_registered_as_an_opentelemetry_callback(rig: Rig) -> None: + listed: Final = rig.proxy.request("GET", "/active/callbacks") + assert listed.status_code == 200, listed.text + assert "OpenTelemetry" in json.dumps(listed.json()), listed.text + assert "NOPS_API_KEY not found" not in rig.process.log.read_text() + + +def test_chat_completion_sdk_span_lands_at_the_sink_with_the_bearer_key(rig: Rig) -> None: + marker: Final = _marker() + completion: Final = rig.openai_client().chat.completions.create( + model=rig.model, messages=[{"role": "user", "content": marker}] + ) + assert completion.id == f"chatcmpl-{marker}" + span: Final = _assert_bearer_span(rig, completion.id, marker) + assert span.attributes.get("gen_ai.request.model") or span.attributes.get("llm.request.model"), span + rows: Final = rig.spend_rows(completion.id) + assert rows[0]["request_id"] == completion.id, rows + + +def test_chat_stream_async_sdk_span_lands_once_after_the_stream_is_consumed(rig: Rig) -> None: + marker: Final = _marker() + + async def consume() -> frozenset[str]: + stream: Final = await rig.async_openai_client().chat.completions.create( + model=rig.model, messages=[{"role": "user", "content": marker}], stream=True + ) + return frozenset([chunk.id async for chunk in stream]) + + identities: Final = asyncio.run(consume()) + assert identities == {f"chatcmpl-{marker}"}, identities + _assert_bearer_span(rig, f"chatcmpl-{marker}", marker) + + +def test_messages_sdk_span_lands_at_the_sink_with_the_bearer_key(rig: Rig) -> None: + marker: Final = _marker() + message: Final = rig.anthropic_client().messages.create( + model=rig.model, max_tokens=16, messages=[{"role": "user", "content": marker}] + ) + _assert_bearer_span(rig, message.id, marker) + + +def test_responses_sdk_span_lands_at_the_sink_with_the_bearer_key(rig: Rig) -> None: + marker: Final = _marker() + response: Final = rig.openai_client().responses.create(model=rig.model, input=marker) + _assert_bearer_span(rig, response.id, marker) + + +def test_v2_flag_on_still_delivers_the_span_with_the_bearer_key(v2_rig: Rig) -> None: + marker: Final = _marker() + response: Final = v2_rig.chat(marker) + assert response.status_code == 200, response.text + _assert_bearer_span(v2_rig, _body_id(response), marker) + + +def _assert_span_without_content(rig: Rig) -> None: + marker: Final = _marker() + prompt: Final = "prompt-" + uuid.uuid4().hex + response: Final = rig.chat(f"{marker} {prompt}") + assert response.status_code == 200, response.text + span: Final = rig.sink.single_span(_body_id(response)) + assert "gen_ai.input.messages" not in span.attributes, span + assert "gen_ai.output.messages" not in span.attributes, span + bodies: Final = rig.sink.bodies() + assert not any(prompt.encode() in body for body in bodies) + assert not any(REPLY_TEXT.encode() in body for body in bodies) + + +def test_legacy_span_carries_no_prompt_or_response_content_by_default(rig: Rig) -> None: + _assert_span_without_content(rig) + + +def test_v2_span_carries_no_prompt_or_response_content_by_default(v2_rig: Rig) -> None: + _assert_span_without_content(v2_rig) + + +def test_endpoint_already_ending_in_v1_traces_is_not_doubled( + provider: Wire, sink: Collector, tmp_path_factory: pytest.TempPathFactory +) -> None: + factory: Final = RigFactory( + provider, sink, tmp_path_factory.mktemp("nops-suffixed"), False, sink.wire.url + "/v1/traces" + ) + suffixed: Final = next(started := factory.start()) + marker: Final = _marker() + response: Final = suffixed.chat(marker) + assert response.status_code == 200, response.text + span: Final = suffixed.sink.single_span(_body_id(response)) + assert span.target == "/v1/traces", span + assert tuple(started) == () + + +def test_three_identical_requests_produce_one_span_each(rig: Rig) -> None: + markers: Final = tuple(_marker() for _ in range(3)) + responses: Final = tuple(rig.chat(marker) for marker in markers) + assert all(response.status_code == 200 for response in responses), [response.text for response in responses] + identities: Final = tuple(_body_id(response) for response in responses) + landed: Final = eventually( + lambda: rig.sink.landed(identities), lambda seen: all(count >= 1 for count in seen.values()), seconds=30 + ) + assert landed == {identity: 1 for identity in identities}, landed + assert rig.sink.landed(identities) == landed + + +def test_health_services_accepts_nops(rig: Rig) -> None: + response: Final = rig.proxy.request("GET", "/health/services", params={"service": "nops"}) + assert response.status_code == 200, response.text + assert "nops" in _text_at(JSON.validate_json(response.content), "message"), response.text + + +def test_sink_answering_403_drops_those_spans_and_later_spans_still_land(rig: Rig) -> None: + rig.sink.rejection.set() + try: + rejected: Final = rig.chat(_marker()) + assert rejected.status_code == 200, rejected.text + rig.sink.refused_batch_carrying(_body_id(rejected)) + finally: + rig.sink.rejection.clear() + later: Final = rig.chat(_marker()) + assert later.status_code == 200, later.text + rig.sink.single_span(_body_id(later)) + assert rig.sink.landed((_body_id(rejected),)) == {_body_id(rejected): 0} + + +def test_sink_answering_404_drops_those_spans_and_later_spans_still_land(rig: Rig) -> None: + rig.sink.missing.set() + try: + dropped: Final = rig.chat(_marker()) + assert dropped.status_code == 200, dropped.text + rig.sink.refused_batch_carrying(_body_id(dropped)) + finally: + rig.sink.missing.clear() + later: Final = rig.chat(_marker()) + assert later.status_code == 200, later.text + rig.sink.single_span(_body_id(later)) + assert rig.sink.landed((_body_id(dropped),)) == {_body_id(dropped): 0} + + +@pytest.mark.parametrize("api_key", [None, ""], ids=["unset", "empty"]) +def test_missing_api_key_fails_loudly_at_boot( + provider: Wire, sink: Collector, tmp_path_factory: pytest.TempPathFactory, api_key: str | None +) -> None: + factory: Final = RigFactory(provider, sink, tmp_path_factory.mktemp("nops-nokey"), False, sink.wire.url, api_key) + started: Final = factory.start() + broken: Final = next(started) + try: + response: Final = broken.chat(_marker()) + assert response.status_code == 200, response.text + eventually( + lambda: broken.process.log.read_text(), + lambda text: "NOPS_API_KEY not found in environment variables" in text, + seconds=30, + ) + assert sink.landed((_body_id(response),)) == {_body_id(response): 0} + finally: + with pytest.raises(StopIteration): + next(started) diff --git a/tests/unit/integrations/otel/test_otel_v2_presets.py b/tests/unit/integrations/otel/test_otel_v2_presets.py index 64366626784..328f62b255b 100644 --- a/tests/unit/integrations/otel/test_otel_v2_presets.py +++ b/tests/unit/integrations/otel/test_otel_v2_presets.py @@ -2,6 +2,8 @@ event loop: the preset does no network I/O, and a custom exporter mints the JWT lazily on its first export (in the BatchSpanProcessor worker thread).""" +from typing import Final + import httpx import pytest @@ -278,3 +280,50 @@ def test_signoz_preset_accepts_the_factory_call_shape(monkeypatch): cfg = PRESET_BY_CALLBACK["signoz"](allow_missing_credentials=True) assert any(e.owner == ExporterOwner.SIGNOZ for e in cfg.exporters) + + +@pytest.mark.parametrize("endpoint", [None, ""], ids=["unset", "empty"]) +def test_nops_preset_defaults_to_the_nops_endpoint_with_a_bearer_key(monkeypatch, endpoint): + monkeypatch.setenv("NOPS_API_KEY", "nops-env-key") + if endpoint is None: + monkeypatch.delenv("NOPS_OTEL_ENDPOINT", raising=False) + else: + monkeypatch.setenv("NOPS_OTEL_ENDPOINT", endpoint) + from litellm.integrations.otel.model.config import ExporterOwner + from litellm.integrations.otel.plumbing.providers import _exports_to_the_wire, _otlp_traces_endpoint + from litellm.integrations.otel.presets import PRESET_BY_CALLBACK + + cfg: Final = PRESET_BY_CALLBACK["nops"](allow_missing_credentials=True) + spec: Final = next(e for e in cfg.exporters if e.owner == ExporterOwner.NOPS) + assert spec.kind == "otlp_http" + assert _otlp_traces_endpoint(spec.endpoint) == "https://api.nops.io/otel/v1/traces" + assert spec.headers == "Authorization=Bearer nops-env-key" + assert _exports_to_the_wire(spec) is True + + +@pytest.mark.parametrize( + "endpoint", + ["http://clara.local:8080/otel", "http://clara.local:8080/otel/", "http://clara.local:8080/otel/v1/traces"], + ids=["base", "trailing-slash", "signal-path"], +) +def test_nops_preset_endpoint_override(monkeypatch, endpoint): + monkeypatch.setenv("NOPS_API_KEY", "nops-env-key") + monkeypatch.setenv("NOPS_OTEL_ENDPOINT", endpoint) + from litellm.integrations.otel.model.config import ExporterOwner + from litellm.integrations.otel.plumbing.providers import _otlp_traces_endpoint + from litellm.integrations.otel.presets.nops import nops_preset + + spec: Final = next(e for e in nops_preset().exporters if e.owner == ExporterOwner.NOPS) + assert _otlp_traces_endpoint(spec.endpoint) == "http://clara.local:8080/otel/v1/traces" + + +def test_nops_preset_without_key_skips_the_exporter(monkeypatch): + monkeypatch.delenv("NOPS_API_KEY", raising=False) + monkeypatch.delenv("NOPS_OTEL_ENDPOINT", raising=False) + from litellm.integrations.otel.model.config import ExporterOwner + from litellm.integrations.otel.plumbing.providers import _exports_to_the_wire + from litellm.integrations.otel.presets.nops import nops_preset + + spec: Final = next(e for e in nops_preset().exporters if e.owner == ExporterOwner.NOPS) + assert spec.headers is None + assert _exports_to_the_wire(spec) is False diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index 79645ffe8f6..d7ab7a5a08f 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -11371,6 +11371,137 @@ def test_uncached_price_keeps_an_earlier_price_failure(pr44187_priced_model: str assert logging_obj.model_call_details["response_cost_failure_debug_information"] == earlier_failure +def test_nops_dispatch_prefers_otel_v2_when_flag_on(monkeypatch): + from litellm.integrations.otel.logger import OpenTelemetryV2 + from litellm.integrations.otel.model.config import ExporterOwner, is_otel_v2_enabled + from litellm.integrations.otel.plumbing.providers import _otlp_traces_endpoint + from litellm.litellm_core_utils import litellm_logging as logging_module + + logging_module._in_memory_loggers.clear() + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + monkeypatch.setenv("NOPS_API_KEY", "test-key") + monkeypatch.delenv("NOPS_OTEL_ENDPOINT", raising=False) + is_otel_v2_enabled.cache_clear() + try: + v2_logger: Final = logging_module._init_custom_logger_compatible_class( + logging_integration="nops", + internal_usage_cache=None, + llm_router=None, + custom_logger_init_args={}, + ) + assert isinstance(v2_logger, OpenTelemetryV2) + assert v2_logger.callback_name == "nops" + spec: Final = next(e for e in v2_logger.config.exporters if e.owner == ExporterOwner.NOPS) + assert spec.kind == "otlp_http" + assert _otlp_traces_endpoint(spec.endpoint) == "https://api.nops.io/otel/v1/traces" + assert spec.headers == "Authorization=Bearer test-key" + again: Final = logging_module._init_custom_logger_compatible_class( + logging_integration="nops", + internal_usage_cache=None, + llm_router=None, + custom_logger_init_args={}, + ) + assert again is v2_logger + finally: + logging_module._in_memory_loggers.clear() + monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) + is_otel_v2_enabled.cache_clear() + + +@pytest.mark.parametrize( + ("endpoint", "expected"), + [ + (None, "https://api.nops.io/otel/v1/traces"), + ("", "https://api.nops.io/otel/v1/traces"), + ("http://clara.local:8080/otel/", "http://clara.local:8080/otel/v1/traces"), + ("http://clara.local:8080/otel/v1/traces", "http://clara.local:8080/otel/v1/traces"), + ], + ids=["default", "empty", "override", "signal-path"], +) +def test_nops_dispatch_keeps_legacy_otel_when_flag_off(monkeypatch, endpoint, expected): + from litellm.integrations.opentelemetry import OpenTelemetry + from litellm.integrations.otel.model.config import is_otel_v2_enabled + from litellm.litellm_core_utils import litellm_logging as logging_module + + logging_module._in_memory_loggers.clear() + monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) + monkeypatch.setenv("NOPS_API_KEY", "legacy-key") + if endpoint is None: + monkeypatch.delenv("NOPS_OTEL_ENDPOINT", raising=False) + else: + monkeypatch.setenv("NOPS_OTEL_ENDPOINT", endpoint) + monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False) + monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_HEADERS", raising=False) + monkeypatch.setattr(litellm, "turn_off_message_logging", False) + is_otel_v2_enabled.cache_clear() + try: + legacy: Final = logging_module._init_custom_logger_compatible_class( + logging_integration="nops", + internal_usage_cache=None, + llm_router=None, + custom_logger_init_args={}, + ) + assert isinstance(legacy, OpenTelemetry) + assert legacy.callback_name == "nops" + assert legacy.config.exporter == "otlp_http" + assert legacy.config.endpoint == expected + assert legacy.config.headers == "Authorization=Bearer legacy-key" + assert legacy._capture_in_span() is False + assert legacy._capture_in_event() is False + assert "OTEL_EXPORTER_OTLP_TRACES_HEADERS" not in os.environ + again: Final = logging_module._init_custom_logger_compatible_class( + logging_integration="nops", + internal_usage_cache=None, + llm_router=None, + custom_logger_init_args={}, + ) + assert again is legacy + logging_module._in_memory_loggers.clear() + monkeypatch.setenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "SPAN_AND_EVENT") + opted_in: Final = logging_module._init_custom_logger_compatible_class( + logging_integration="nops", + internal_usage_cache=None, + llm_router=None, + custom_logger_init_args={}, + ) + assert isinstance(opted_in, OpenTelemetry) + assert opted_in._capture_in_span() is True + assert opted_in._capture_in_event() is True + finally: + logging_module._in_memory_loggers.clear() + is_otel_v2_enabled.cache_clear() + + +@pytest.mark.parametrize("flag", ["true", "false"], ids=["v2", "legacy"]) +@pytest.mark.parametrize("key", [None, ""], ids=["unset", "empty"]) +def test_nops_dispatch_requires_an_api_key(monkeypatch, flag, key): + from litellm.integrations.otel.model.config import is_otel_v2_enabled + from litellm.litellm_core_utils import litellm_logging as logging_module + + logging_module._in_memory_loggers.clear() + monkeypatch.setenv("LITELLM_OTEL_V2", flag) + if key is None: + monkeypatch.delenv("NOPS_API_KEY", raising=False) + else: + monkeypatch.setenv("NOPS_API_KEY", key) + is_otel_v2_enabled.cache_clear() + try: + with pytest.raises(ValueError, match="NOPS_API_KEY not found in environment variables"): + logging_module.build_custom_logger_compatible_class("nops", internal_usage_cache=None, llm_router=None) + created: Final = logging_module._init_custom_logger_compatible_class( + logging_integration="nops", + internal_usage_cache=None, + llm_router=None, + custom_logger_init_args={}, + ) + assert created is None + assert not [cb for cb in logging_module._in_memory_loggers if getattr(cb, "callback_name", None) == "nops"] + finally: + logging_module._in_memory_loggers.clear() + monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) + is_otel_v2_enabled.cache_clear() + + def test_enterprise_alerting_loggers_resolves_real_classes(): from litellm_enterprise.enterprise_callbacks.pagerduty.pagerduty import PagerDutyAlerting from litellm_enterprise.enterprise_callbacks.send_emails.resend_email import ResendEmailLogger diff --git a/ui/litellm-dashboard/public/assets/logos/nops.png b/ui/litellm-dashboard/public/assets/logos/nops.png new file mode 100644 index 00000000000..3aa8e0a6b31 Binary files /dev/null and b/ui/litellm-dashboard/public/assets/logos/nops.png differ diff --git a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx index 6ee092f030d..ef78068c9a3 100644 --- a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx +++ b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx @@ -7,6 +7,7 @@ import lagoLogo from "../../public/assets/logos/lago.svg"; import langfuseLogo from "../../public/assets/logos/langfuse.png"; import langsmithLogo from "../../public/assets/logos/langsmith.png"; import newrelicLogo from "../../public/assets/logos/newrelic.png"; +import nopsLogo from "../../public/assets/logos/nops.png"; import openmeterLogo from "../../public/assets/logos/openmeter.png"; import otelLogo from "../../public/assets/logos/otel.png"; import pointfiveLogo from "../../public/assets/logos/pointfive.png"; @@ -155,6 +156,17 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [ }, description: "Langsmith Logging Integration", }, + { + id: "nops", + displayName: "nOps", + logo: nopsLogo.src, + supports_key_team_logging: false, + dynamic_params: { + NOPS_API_KEY: "password", + NOPS_OTEL_ENDPOINT: "text", + }, + description: "nOps Logging Integration", + }, { id: "openmeter", displayName: "OpenMeter", diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 2b174a1141a..cad86c1d30e 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -62149,7 +62149,7 @@ export interface operations { parameters: { query: { /** @description Specify the service being hit. */ - service: ("slack_budget_alerts" | "langfuse" | "langfuse_otel" | "slack" | "ms_teams" | "openmeter" | "webhook" | "email" | "braintrust" | "datadog" | "datadog_llm_observability" | "generic_api" | "arize" | "galileo" | "newrelic" | "pointfive" | "signoz" | "sqs") | string; + service: ("slack_budget_alerts" | "langfuse" | "langfuse_otel" | "slack" | "ms_teams" | "openmeter" | "webhook" | "email" | "braintrust" | "datadog" | "datadog_llm_observability" | "generic_api" | "arize" | "galileo" | "newrelic" | "nops" | "pointfive" | "signoz" | "sqs") | string; }; header?: never; path?: never;