diff --git a/docs/how-to/run_openai_chat_front.md b/docs/how-to/run_openai_chat_front.md index dc3c89fe..789f2056 100644 --- a/docs/how-to/run_openai_chat_front.md +++ b/docs/how-to/run_openai_chat_front.md @@ -242,7 +242,8 @@ Every error is an OpenAI envelope, `{"error": {"message", "type", "param", "code `code` written as null when they do not apply. `metadata` holds what is known: `request_id`, the horde's `rc`, `retry_after`, and `submit_outcome: "unknown"` with the request's `operation_key` when a submit was lost and recovery did not settle it. A horde error keeps the horde's status, and its -`rc` becomes `code`; a horde 429 keeps `Retry-After` and the `X-RateLimit-*` headers, and a horde 503 without +`rc` becomes `code`; a horde 429 keeps `Retry-After` and the `X-RateLimit-*` headers, and has code +`rate_limit_exceeded` when the horde sends no `rc`, such as for its per-route request limit. A horde 503 without `Retry-After` (such as `ChatAdmissionDisabled`) gets `Retry-After: 30`. A status read that ends the wait with a horde error other than 404, 429 or a 5xx, such as 401 or 403, cancels the request before the error is returned; a 404 means the horde no longer holds it. @@ -329,7 +330,8 @@ anything. | `DEPLOYMENT_ENVIRONMENT` | `development` | The `deployment.environment` resource attribute. | | `OTEL_TRACES_SAMPLER_ARG` | `1.0` | Head sampling ratio, as for the horde. | -The front does not start Pyroscope, and `PYROSCOPE_ENABLED` alone does not turn its export on. +`PYROSCOPE_ENABLED=true` starts Pyroscope continuous profiling with the horde's `PYROSCOPE_*` variables (see +`horde.telemetry`). The application name is the front's service name, so its profiles are separate from the horde's. Verification: with an endpoint set, a collector receives `horde.oai.*` metrics within the SDK's export interval after the first request, one `Chat request finished: ...` log record per finished request, and a trace per diff --git a/docs/how-to/send_chat_completions.md b/docs/how-to/send_chat_completions.md index 2bcc0414..9c191d02 100644 --- a/docs/how-to/send_chat_completions.md +++ b/docs/how-to/send_chat_completions.md @@ -388,8 +388,9 @@ Every error is `{"error": {"message", "type", "param", "code", "metadata"}}`, wi do not apply and `metadata` carrying the Horde's `request_id` and return code where known ([ADR 0034](../decisions/0034-map-chat-errors-to-the-openai-envelope-with-openrouter-status-vocabulary.md)). A refusal from the Horde keeps its status and has its return code (see `README_return_codes.md` at the repository root) as -`code`; a Horde 429 keeps `Retry-After` and the `X-RateLimit-*` headers. A native payload validation failure, which -has no return code, is 400 `BadRequest` with the failing key as `param`, such as `temperature` or `max_tokens`. +`code`; a Horde 429 keeps `Retry-After` and the `X-RateLimit-*` headers, and has code `rate_limit_exceeded` when the +Horde sends no return code. A native payload validation failure, which does not carry a return code, is 400 +`BadRequest` with the failing key as `param`, such as `temperature` or `max_tokens`. The front's own codes: diff --git a/docs/reference/chat_front.md b/docs/reference/chat_front.md index c4e4feae..43e02a2e 100644 --- a/docs/reference/chat_front.md +++ b/docs/reference/chat_front.md @@ -294,11 +294,23 @@ another, and the views from `horde.oai.metrics.histogram_views`. It bridges Logu instruments httpx, so each call to the horde is a client span whose `traceparent` the horde's Flask instrumentation continues. The Flask application's `init_telemetry_early` is not reused because it instruments a Flask app. +With `PYROSCOPE_ENABLED=true`, `configure_telemetry` starts Pyroscope through `horde.telemetry.init_pyroscope`, the +function the Flask application uses, with the same `PYROSCOPE_*` variables and the front's service name as the +application name. It starts also when `OTEL_SDK_DISABLED=true`; span profiles need the SDK. + `instrument_app` adds the server spans (`POST /v1/chat/completions`, `GET /v1/models`) once `create_app` has built the application; `main` calls it only when telemetry was configured. `/healthz` is excluded, since the listener and the container poll it every few seconds. The instrumentation does not capture headers or bodies. A streamed request's span lasts until the stream ends. +## Spans + +Besides the server and client spans, the front records these spans. Attribute keys carry the `horde.` prefix. + +| Name | Attributes | Meaning | +| --- | --- | --- | +| `horde.oai.delta_read` | `entry_count`, `byte_count`, `first_entry_lag_seconds` | One delta stream read that returned entries, across every registered request's stream; an empty read does not record a span. `entry_count` is the entries returned, `byte_count` the UTF-8 size of their field names and values, and `first_entry_lag_seconds` the time from the oldest returned entry's append, the millisecond part of its stream id on the Redis server's clock, to the read's return. The reader is one background task, so the span is a root span and not part of a request's trace. | + `create_app` takes the meter from the global meter provider unless one is passed as `meter`; tests pass a meter of a separate `MeterProvider` with an `InMemoryMetricReader`. Unconfigured, the global provider does not record anything. @@ -306,14 +318,14 @@ anything. ## Metrics Attribute keys carry the `horde.` prefix. Every attribute is bounded: call types, outcomes, wake sources, HTTP -statuses, error codes, booleans and the model id. The model id comes from a delivered completion, so it is a model a +statuses, error codes, refusal reasons, booleans and the model id. The model id comes from a delivered completion, so it is a model a worker served. Client addresses, API keys, prompts, client titles and request ids are never attributes. | Name | Type | Attributes | Meaning | | --- | --- | --- | --- | | `horde.oai.requests.admitted` | counter | `streamed` | Chat requests that took an in-flight slot. | | `horde.oai.requests.refused` | counter | `status_code`, `code` | Chat requests refused before an in-flight slot: body size (413), body read timeout (408 `RequestBodyTimeout`), validation (400, 404), authorization (401), `FrontAtCapacity` (503). | -| `horde.oai.requests.errors` | counter | `status_code`, `code`, `streamed` | Admitted requests answered with an error status before any header: horde refusals at submit, 408 `ChatRequestTimeout`, 404 `no_worker_supports_features`, 400 `no_worker_supports_limit`, 503 `no_eligible_worker`, 503 `queue_unavailable`, 502 `ChatRequestFaulted`, `ChatRequestNoGeneration` and `HordeUnavailable`. | +| `horde.oai.requests.errors` | counter | `status_code`, `code`, `streamed`, `reason` | Admitted requests answered with an error status before any header: horde refusals at submit (429 `rate_limit_exceeded` when the horde's 429 does not carry an `rc`), 408 `ChatRequestTimeout`, 404 `no_worker_supports_features`, 400 `no_worker_supports_limit`, 503 `no_eligible_worker`, 503 `queue_unavailable`, 502 `ChatRequestFaulted`, `ChatRequestNoGeneration` and `HordeUnavailable`. `reason` is `no_eligible_threads` or `wait_exceeds_budget` for `queue_unavailable` and empty for every other code. | | `horde.oai.stream.failures` | counter | `code` | Streamed requests whose 200 headers were sent and which ended in an SSE error event. | | `horde.oai.inflight` | observable gauge | none | Requests holding an in-flight slot. | | `horde.oai.inflight.sources` | observable gauge | none | Distinct client addresses holding at least one slot. | @@ -345,11 +357,15 @@ duration use `BUCKETS_REQUEST_LIFECYCLE_SECONDS` (both in `horde/metrics.py`); b Every admitted request logs one Loguru `INFO` record when it ends. The message is ```text -Chat request finished: streamed= submitted_at= woke_at= woke_by= waited=s status= code= wakes= estimated_usage= done_without_generation_reads= +Chat request finished: streamed= submitted_at= woke_at= woke_by= waited=s status= code= wakes= estimated_usage= done_without_generation_reads= requested_model= served_model= retry_after= ``` -and the same values are bound to the record's `extra`, which the OpenTelemetry log bridge exports as attributes. A -request refused at submit has no id; its message starts `Unsubmitted chat request finished:`. +A `queue_unavailable` refusal appends +`queue_reason= wait_time= eligible_worker_threads=`. Every line then ends with +`first_read_wait_time= first_read_eligible_worker_threads=`. + +The same values are bound to the record's `extra`, which the OpenTelemetry log bridge exports as attributes. A +request refused at submit does not have an id; its message starts `Unsubmitted chat request finished:`. `tests/stress/chat_scale_demo.py` parses the message up to `waited`. | Field | Meaning | @@ -365,5 +381,13 @@ request refused at submit has no id; its message starts `Unsubmitted chat reques | `wake_count` | How many status reads the wait made. | | `estimated_usage` | Whether the delivered completion's usage was estimated; null when none was delivered. | | `done_without_generation_reads` | How many status reads showed the request done, not faulted and without a generation. | +| `requested_model` | The models the request accepts, comma-separated. | +| `served_model` | The model of the delivered completion; null when none was delivered. | +| `retry_after` | The `Retry-After` the error response or event carried, in seconds; null when it carried none. | +| `queue_reason` | `queue_unavailable` only: `no_eligible_threads` or `wait_exceeds_budget`. | +| `wait_time` | `queue_unavailable` only: the horde-wide wait estimate the deciding status read reported; null when it had none. | +| `eligible_worker_threads` | `queue_unavailable` only: the eligible worker threads that read reported; null when it had none. | +| `first_read_wait_time` | `fail_fast` only: the wait estimate of the status read that took the first-read decision, whether it admitted or refused the request; null in the `wait` mode, before a decision, or when the read had none. | +| `first_read_eligible_worker_threads` | `fail_fast` only: the eligible worker threads that read reported; null as for `first_read_wait_time`. | The line does not carry the API key, the messages, the client's title or address, or the operation key. diff --git a/horde/oai/__main__.py b/horde/oai/__main__.py index 17165f6c..341dee8a 100644 --- a/horde/oai/__main__.py +++ b/horde/oai/__main__.py @@ -15,6 +15,8 @@ off, and ``OTEL_SERVICE_NAME``, ``DEPLOYMENT_ENVIRONMENT`` and ``OTEL_TRACES_SAM sampling. The service name defaults to [`FRONT_SERVICE_NAME`][horde.oai.__main__.FRONT_SERVICE_NAME]. With none of them set the front exports nothing. Configured, the front records one server span per request (health checks excluded) and a client span per call to the horde, carrying the trace context into the horde's own spans. +``PYROSCOPE_ENABLED=true`` starts continuous profiling through the horde's ``PYROSCOPE_*`` variables, with the +same service name as the front's telemetry, so the front's profiles are kept apart from the horde's. """ from __future__ import annotations @@ -30,7 +32,7 @@ from loguru import logger from horde.oai.app import create_app from horde.oai.metrics import histogram_views from horde.oai.settings import load_settings -from horde.telemetry import build_sampling_options, otel_sdk_enabled, telemetry_enabled +from horde.telemetry import build_sampling_options, init_pyroscope, otel_sdk_enabled, telemetry_enabled __all__ = [ "FRONT_SERVICE_NAME", @@ -54,19 +56,26 @@ def configure_telemetry() -> bool: The Flask application's ``init_telemetry_early`` also instruments Flask, so the front configures Logfire itself with the same variables, its own service name and the bucket boundaries of its histograms. Loguru records, - the finished-request line included, are bridged to OpenTelemetry logs. Pyroscope profiling is not started. + the finished-request line included, are bridged to OpenTelemetry logs. Pyroscope profiling starts through the + horde's ``init_pyroscope`` under the front's service name, also when ``OTEL_SDK_DISABLED=true`` leaves the SDK + off, and its span processors join the tracer provider when span profiles are on. httpx is instrumented here, before the application creates its clients, so every call to the horde is a client span that carries the trace context (``traceparent``) the horde's Flask instrumentation continues; the server spans are added by [`instrument_app`][horde.oai.__main__.instrument_app] once the application exists. Returns: - Whether telemetry was configured. + Whether the OpenTelemetry SDK was configured. Side Effects: - Installs the global OpenTelemetry tracer and meter providers, instruments httpx and adds a Loguru sink. + Starts Pyroscope when ``PYROSCOPE_ENABLED=true``, installs the global OpenTelemetry tracer and meter + providers, instruments httpx and adds a Loguru sink. """ - if not telemetry_enabled() or not otel_sdk_enabled(): + if not telemetry_enabled(): + return False + sdk_enabled = otel_sdk_enabled() + span_processors = init_pyroscope(span_profiles_enabled=sdk_enabled, default_service_name=FRONT_SERVICE_NAME) + if not sdk_enabled: return False logfire.configure( send_to_logfire=False, @@ -75,6 +84,7 @@ def configure_telemetry() -> bool: environment=os.environ.get("DEPLOYMENT_ENVIRONMENT", "development"), sampling=build_sampling_options(), metrics=logfire.MetricsOptions(views=histogram_views()), + additional_span_processors=span_processors or None, ) loguru_handler = logfire.loguru_handler() logger.add(**loguru_handler) diff --git a/horde/oai/app.py b/horde/oai/app.py index 4dea0bee..e46d7544 100644 --- a/horde/oai/app.py +++ b/horde/oai/app.py @@ -48,7 +48,7 @@ from fastapi.responses import JSONResponse from horde_sdk.generation_parameters.text.chat_policy import ChatGuarantee from loguru import logger from opentelemetry.metrics import Meter -from pydantic import ValidationError +from pydantic import BaseModel, ConfigDict, ValidationError from sse_starlette import EventSourceResponse, ServerSentEvent from starlette.background import BackgroundTask from starlette.requests import ClientDisconnect @@ -146,6 +146,7 @@ __all__ = [ "InflightLimiter", "InflightSlot", "OpenRequest", + "QueueRefusalFacts", "RequestTrace", "create_app", ] @@ -273,6 +274,10 @@ MAX_KUDOS_EXCEEDED_CODE: str = "max_kudos_exceeded" """The error code of a request whose kudos estimate at current pricing exceeds its ``horde.max_kudos``; the Horde refuses it with ``ChatKudosCapExceeded`` before creating it.""" +RATE_LIMIT_EXCEEDED_CODE: str = "rate_limit_exceeded" +"""The error code of a Horde 429 that carries no ``rc``, such as a per-route request limit; OpenAI's code for a rate +limit refusal. A 429 with an ``rc``, such as ``TooManyPrompts``, keeps the ``rc``.""" + MAX_KUDOS_PARAMETER: str = "horde.max_kudos" """The request path of the client's kudos cap.""" @@ -337,6 +342,22 @@ CLIENT_RETRY_COUNT_HEADER: str = "x-stainless-retry-count" """The request header OpenAI SDKs send with how many times they retried the call; a hint, never an identity.""" +class QueueRefusalFacts(BaseModel): + """Represents the pool a ``fail_fast`` request's first status read reported when it refused the request. + + The facts go to the error metric and the finished-request line; the client reads them in the error message only. + """ + + model_config = ConfigDict(frozen=True) + + reason: QueueUnavailableReason + """Which rule refused the request.""" + wait_time: int | None + """The Horde's wait estimate in seconds, when the status carried one.""" + eligible_worker_threads: int | None + """The eligible worker threads online, when the status carried the count.""" + + class ChatFrontError(Exception): """An error the front answers with an OpenAI error envelope, or an SSE error event after headers.""" @@ -350,6 +371,7 @@ class ChatFrontError(Exception): metadata: ErrorMetadata | None = None, headers: Mapping[str, str] | None = None, job_may_exist: bool = False, + queue_refusal: QueueRefusalFacts | None = None, ) -> None: """Create the error. @@ -361,6 +383,7 @@ class ChatFrontError(Exception): metadata: The request id, return code and retry details that are known. headers: Extra response headers, such as ``Retry-After``. job_may_exist: Whether the Horde holds, or may hold, a request for this call; adds ``x-should-retry: false``. + queue_refusal: The pool facts of a ``queue_unavailable`` refusal, for the metric and log; not sent. """ super().__init__(message) self.status_code: int = status_code @@ -377,6 +400,8 @@ class ChatFrontError(Exception): """Extra response headers.""" self.job_may_exist: bool = job_may_exist """Whether a retry could create a second charged job.""" + self.queue_refusal: QueueRefusalFacts | None = queue_refusal + """The pool facts of a ``queue_unavailable`` refusal.""" def to_response(self) -> JSONResponse: """Create the JSON error response, with ``x-should-retry: false`` when a job may exist.""" @@ -767,7 +792,8 @@ def _horde_error(error: HordeAPIError, *, request_id: str | None, job_may_exist: """Create the front error for a Horde error, keeping its status, ``rc`` and forwarded headers. A native payload validation failure, which has no ``rc``, is reported as ``BadRequest`` with its first failing - field as ``param``. A refusal over the client's kudos cap is reported as ``max_kudos_exceeded``. + field as ``param``. A 429 without an ``rc`` is reported as ``rate_limit_exceeded``, keeping the Horde's message and + ``Retry-After``. A refusal over the client's kudos cap is reported as ``max_kudos_exceeded``. """ if error.return_code == CHAT_KUDOS_CAP_EXCEEDED_RC: return _kudos_cap_error(error, request_id=request_id, job_may_exist=job_may_exist) @@ -790,6 +816,8 @@ def _horde_error(error: HordeAPIError, *, request_id: str | None, job_may_exist: param = _request_path_of_native_field(native_path) or None message = f"{param}: {field_message}" if param is not None else field_message code = code if code is not None else BAD_REQUEST_RC + if code is None and error.status_code == HTTPStatus.TOO_MANY_REQUESTS: + code = RATE_LIMIT_EXCEEDED_CODE return ChatFrontError( status_code=error.status_code, message=message, @@ -1435,21 +1463,28 @@ class RequestTrace: """Time one admitted chat request and record how it ended, once. The line logged at the end carries the request id, the submit and wake times, what woke the request, the wait, - the status and code, the wake count, whether usage was estimated and how many reads showed the request done - without a generation. It does not carry the API key, the messages - or the client's attribution. + the status and code, the wake count, whether usage was estimated, how many reads showed the request done without + a generation, the requested and served models and the ``Retry-After`` sent, for a ``queue_unavailable`` + refusal the pool facts that decided it, and for a ``fail_fast`` request the wait estimate and eligible worker + threads of the status that took its first-read decision. It does not carry the API key, the messages or the + client's attribution. """ - def __init__(self, *, front_metrics: FrontMetrics, streamed: bool, received_at: float) -> None: + def __init__(self, *, front_metrics: FrontMetrics, streamed: bool, received_at: float, requested_models: Sequence[str]) -> None: """Create the trace of an admitted request. Args: front_metrics: The front's instruments. streamed: Whether the request is streamed. received_at: The monotonic time the request arrived. + requested_models: The models the request accepts. """ self.streamed: bool = streamed """Whether the request is streamed.""" + self.requested_models: tuple[str, ...] = tuple(requested_models) + """The models the request accepts.""" + self.served_model: str | None = None + """The model of the delivered completion, once one is recorded.""" self.request_id: str | None = None """The Horde request id, once the submit is acknowledged.""" self.submitted_at: float | None = None @@ -1486,6 +1521,7 @@ class RequestTrace: def record_completion(self, completion: ChatCompletionResponse, status: HordeChatStatus) -> None: """Record the content, usage, empty-return and queue-wait facts of a completion about to be sent.""" self.record_first_content() + self.served_model = completion.model self._front_metrics.record_completion( model=completion.model, estimated_usage=completion.x_horde.estimated_usage, @@ -1500,16 +1536,21 @@ class RequestTrace: def error_response(self, error: ChatFrontError, *, watcher: ChatRequestWatcher | None = None) -> JSONResponse: """Return the error response sent before headers, recording the error and the request's end.""" - self._front_metrics.record_error(status_code=int(error.status_code), code=error.code, streamed=self.streamed) + self._front_metrics.record_error( + status_code=int(error.status_code), + code=error.code, + streamed=self.streamed, + reason=error.queue_refusal.reason if error.queue_refusal is not None else None, + ) if error.metadata.submit_outcome == SubmitOutcome.UNKNOWN: self._front_metrics.record_submit_outcome_unknown() - self.finish(status_code=int(error.status_code), code=error.code or "", watcher=watcher) + self.finish(status_code=int(error.status_code), code=error.code or "", watcher=watcher, front_error=error) return error.to_response() def error_events(self, error: ChatFrontError, *, watcher: ChatRequestWatcher | None) -> list[ServerSentEvent]: """Return the error events sent after a stream's 200 headers, recording the failure and the request's end.""" self._front_metrics.record_stream_failure(code=error.code) - self.finish(status_code=HTTPStatus.OK, code=error.code or "", watcher=watcher) + self.finish(status_code=HTTPStatus.OK, code=error.code or "", watcher=watcher, front_error=error) return error.to_events() def finish_disconnected(self, *, status_code: int, watcher: ChatRequestWatcher | None) -> None: @@ -1530,6 +1571,7 @@ class RequestTrace: watcher: ChatRequestWatcher | None, estimated_usage: bool | None = None, woke_by: WakeSource | None = None, + front_error: ChatFrontError | None = None, ) -> None: """Record the request's duration and log its finished line; only the first call counts. @@ -1540,6 +1582,7 @@ class RequestTrace: watcher: The request's watcher, when the wait started. estimated_usage: Whether the delivered completion's usage was estimated; None when none was delivered. woke_by: What ended the wait when the watcher did not find the request finished. + front_error: The error the request was answered with, for its ``Retry-After`` and queue refusal facts. """ if self._finished: return @@ -1568,16 +1611,43 @@ class RequestTrace: "wake_count": watcher.wake_count if watcher is not None else 0, "estimated_usage": estimated_usage, "done_without_generation_reads": watcher.done_without_generation_reads if watcher is not None else 0, + "requested_model": ",".join(self.requested_models), + "served_model": self.served_model, + "retry_after": front_error.metadata.retry_after if front_error is not None else None, } + queue_refusal = front_error.queue_refusal if front_error is not None else None + if queue_refusal is not None: + finished_fields["queue_reason"] = str(queue_refusal.reason) + finished_fields["wait_time"] = queue_refusal.wait_time + finished_fields["eligible_worker_threads"] = queue_refusal.eligible_worker_threads + first_read_status = watcher.first_read_status if watcher is not None else None + finished_fields["first_read_wait_time"] = first_read_status.wait_time if first_read_status is not None else None + finished_fields["first_read_eligible_worker_threads"] = ( + first_read_status.eligible_worker_threads if first_read_status is not None else None + ) request_label = f"Chat request {self.request_id}" if self.request_id is not None else "Unsubmitted chat request" - # tests/stress/chat_scale_demo.py parses the leading fields in this order and format. - logger.bind(**finished_fields).info( + # tests/stress/chat_scale_demo.py parses the leading fields in this order and format, and the horde-chat + # dashboard's LogQL reads streamed, woke_by and code with greedy matches, so later fields go at the end and no + # later key ends in "code", "woke_by" or "streamed". + message = ( f"{request_label} finished: streamed={self.streamed} submitted_at={_format_seconds(self.submitted_at)} " f"woke_at={_format_seconds(woke_at)} woke_by={finished_fields['woke_by']} " f"waited={_format_seconds(waited_seconds)}s status={int(status_code)} code={code} " f"wakes={finished_fields['wake_count']} estimated_usage={estimated_usage} " - f"done_without_generation_reads={finished_fields['done_without_generation_reads']}", + f"done_without_generation_reads={finished_fields['done_without_generation_reads']} " + f"requested_model={finished_fields['requested_model']} served_model={_format_optional(self.served_model)} " + f"retry_after={_format_optional(finished_fields['retry_after'])}" ) + if queue_refusal is not None: + message += ( + f" queue_reason={queue_refusal.reason} wait_time={_format_optional(queue_refusal.wait_time)} " + f"eligible_worker_threads={_format_optional(queue_refusal.eligible_worker_threads)}" + ) + message += ( + f" first_read_wait_time={_format_optional(finished_fields['first_read_wait_time'])} " + f"first_read_eligible_worker_threads={_format_optional(finished_fields['first_read_eligible_worker_threads'])}" + ) + logger.bind(**finished_fields).info(message) def _format_seconds(seconds: float | None) -> str: @@ -1585,6 +1655,11 @@ def _format_seconds(seconds: float | None) -> str: return f"{seconds:.3f}" if seconds is not None else "none" +def _format_optional(value: object) -> str: + """Return ``value`` as text, or ``none``.""" + return str(value) if value is not None else "none" + + async def _wait_for_disconnect(request: Request) -> None: """Return once the client of ``request`` disconnects; the request body must already be read.""" while True: @@ -1763,6 +1838,11 @@ def _queue_unavailable_error(error: QueueUnavailableError, *, request_id: str) - metadata=ErrorMetadata(request_id=request_id, retry_after=retry_after), headers={RETRY_AFTER_HEADER: str(retry_after)}, job_may_exist=True, + queue_refusal=QueueRefusalFacts( + reason=error.reason, + wait_time=status.wait_time, + eligible_worker_threads=status.eligible_worker_threads, + ), ) @@ -1966,7 +2046,12 @@ async def _chat_completions(request: Request) -> Response: raise streamed = bool(chat_request.stream) front_metrics.record_admitted(streamed=streamed) - request_trace = RequestTrace(front_metrics=front_metrics, streamed=streamed, received_at=received_at) + request_trace = RequestTrace( + front_metrics=front_metrics, + streamed=streamed, + received_at=received_at, + requested_models=chat_request.requested_models(), + ) budget_seconds = settings.stream_timeout_seconds if chat_request.stream else settings.nonstream_timeout_seconds deadline = time.monotonic() + budget_seconds # A fresh key per client request: a client's own retry is a new request and is not deduplicated (ADR 0026). diff --git a/horde/oai/metrics.py b/horde/oai/metrics.py index 1e380b3d..da90ab43 100644 --- a/horde/oai/metrics.py +++ b/horde/oai/metrics.py @@ -151,6 +151,7 @@ class MetricAttribute(StrEnum): ESTIMATED_USAGE = "horde.estimated_usage" EXISTING_CLAIM = "horde.existing_claim" RELAYED = "horde.relayed" + REASON = "horde.reason" class NativeCallType(StrEnum): @@ -373,14 +374,22 @@ class FrontMetrics: """Record a chat request refused before it took an in-flight slot.""" self._refused.add(1, {MetricAttribute.STATUS_CODE: int(status_code), MetricAttribute.CODE: code or ""}) - def record_error(self, *, status_code: int, code: str | None, streamed: bool) -> None: - """Record an admitted chat request answered with an error status before any header was sent.""" + def record_error(self, *, status_code: int, code: str | None, streamed: bool, reason: str | None) -> None: + """Record an admitted chat request answered with an error status before any header was sent. + + Args: + status_code: The HTTP status sent. + code: The error code. + streamed: Whether the request is streamed. + reason: Which rule refused a ``queue_unavailable`` request; None for every other error. + """ self._errors.add( 1, { MetricAttribute.STATUS_CODE: int(status_code), MetricAttribute.CODE: code or "", MetricAttribute.STREAMED: streamed, + MetricAttribute.REASON: reason or "", }, ) diff --git a/horde/oai/waiter.py b/horde/oai/waiter.py index d8575f01..9125f8cc 100644 --- a/horde/oai/waiter.py +++ b/horde/oai/waiter.py @@ -64,6 +64,7 @@ from enum import StrEnum from http import HTTPStatus import httpx +import logfire import redis.asyncio import redis.exceptions from loguru import logger @@ -127,6 +128,29 @@ to join, and a delta reaches its client within it. DELTA_READ_COUNT: int = 100 """The most entries one ``XREAD`` returns per stream; a reader that fell behind catches up over several reads.""" +DELTA_READ_SPAN_NAME: str = "horde.oai.delta_read" +"""The span recorded for each delta stream read that returns entries. + +An empty read does not get a span: with any streamed request registered the reader reads every block, so empty +reads would add several spans a second per process without a relayed chunk. +""" + +DELTA_READ_ENTRY_COUNT_ATTRIBUTE: str = "horde.entry_count" +"""The read span attribute holding how many stream entries the read returned, across every stream.""" + +DELTA_READ_BYTE_COUNT_ATTRIBUTE: str = "horde.byte_count" +"""The read span attribute holding the UTF-8 size of the returned entries' field names and values.""" + +DELTA_READ_FIRST_ENTRY_LAG_ATTRIBUTE: str = "horde.first_entry_lag_seconds" +"""The read span attribute holding the seconds from the oldest returned entry's append to the read's return. + +The append time is the millisecond part of the entry's auto-generated stream id, taken from the Redis server's +clock, so a skew between that clock and the front's shifts the value by the same amount. +""" + +STREAM_ID_MILLISECONDS_PER_SECOND: int = 1000 +"""The unit of the time part of a Redis stream entry id, which counts milliseconds.""" + DELTA_READ_TIMEOUT_SECONDS: float = 5.0 """How long one ``XREAD`` may take before the connection is treated as dead and the read is made again. @@ -648,20 +672,37 @@ class DeltaStreamReader: ), DELTA_READ_TIMEOUT_SECONDS, ) - for stream_key, stream_entries in replies or (): - request_id = _decode_text(stream_key).removeprefix(DELTA_STREAM_KEY_PREFIX) - subscription = self._subscriptions.get(request_id) - if subscription is None: - continue - for entry_id, entry_fields in stream_entries: - subscription.last_entry_id = _decode_text(entry_id) - fields = {_decode_text(field): _decode_text(value) for field, value in entry_fields.items()} - try: - entry = DeltaEntry.model_validate(fields) - except ValidationError as error: - logger.warning(f"Ignoring a malformed delta of chat request {request_id}: {error}") - continue - subscription.deliver(entry) + if not replies: + return + read_at = time.time() + with logfire.span(DELTA_READ_SPAN_NAME) as span: + entry_count = 0 + byte_count = 0 + oldest_appended_at: float | None = None + for stream_key, stream_entries in replies: + request_id = _decode_text(stream_key).removeprefix(DELTA_STREAM_KEY_PREFIX) + if stream_entries: + first_appended_at = _stream_entry_appended_at(_decode_text(stream_entries[0][0])) + if first_appended_at is not None and (oldest_appended_at is None or first_appended_at < oldest_appended_at): + oldest_appended_at = first_appended_at + subscription = self._subscriptions.get(request_id) + for entry_id, entry_fields in stream_entries: + fields = {_decode_text(field): _decode_text(value) for field, value in entry_fields.items()} + entry_count += 1 + byte_count += sum(len(field.encode()) + len(value.encode()) for field, value in fields.items()) + if subscription is None: + continue + subscription.last_entry_id = _decode_text(entry_id) + try: + entry = DeltaEntry.model_validate(fields) + except ValidationError as error: + logger.warning(f"Ignoring a malformed delta of chat request {request_id}: {error}") + continue + subscription.deliver(entry) + span.set_attribute(DELTA_READ_ENTRY_COUNT_ATTRIBUTE, entry_count) + span.set_attribute(DELTA_READ_BYTE_COUNT_ATTRIBUTE, byte_count) + if oldest_appended_at is not None: + span.set_attribute(DELTA_READ_FIRST_ENTRY_LAG_ATTRIBUTE, max(0.0, read_at - oldest_appended_at)) async def _read_forever(self, redis_client: redis.asyncio.Redis) -> None: """Read the registered streams until cancelled, waiting while none is registered and reading again after a @@ -721,6 +762,9 @@ class ChatRequestWatcher: """How many status reads the wait made, the read that found the request finished included.""" self.done_without_generation_reads: int = 0 """How many status reads showed the request done, not faulted and without a finished generation.""" + self.first_read_status: HordeChatStatus | None = None + """The status that took a ``fail_fast`` request's first-read decision, whether it admitted or refused the + request; None in the ``wait`` mode and before the decision.""" self._front_metrics = front_metrics self._horde_client = horde_client self._credentials = credentials @@ -895,7 +939,7 @@ class ChatRequestWatcher: In the ``fail_fast`` mode a status without queue information does not decide the wait rule until [`POSITIONED_READ_WAIT_SECONDS`][horde.oai.waiter.POSITIONED_READ_WAIT_SECONDS] after the first read: it schedules a re-read instead, and the first status that carries the information, or the latest once the bound - has passed, decides. Zero eligible worker threads needs no position and refuses on the first read. + has passed, decides. With zero eligible worker threads the request does not need a position and is refused on the first read. Raises: QueueUnavailableError: The pool has zero eligible worker threads, or the wait estimate exceeds the time @@ -911,6 +955,7 @@ class ChatRequestWatcher: self._positioned_reread_at = math.inf if not is_fail_fast: return + self.first_read_status = status if status.processing > 0 or status.finished > 0: return remaining_seconds = self._budget.deadline - self._clock() @@ -1128,6 +1173,18 @@ def _decode_text(text: object) -> str: return text.decode("utf-8", "replace") if isinstance(text, bytes) else str(text) +def _stream_entry_appended_at(entry_id: str) -> float | None: + """Return the Unix time in seconds a stream entry was appended, from its ``-`` id. + + Returns None for an id Redis did not generate, which does not carry a time. + """ + milliseconds, _, _ = entry_id.partition("-") + try: + return int(milliseconds) / STREAM_ID_MILLISECONDS_PER_SECOND + except ValueError: + return None + + def _is_finished(status: HordeChatStatus) -> bool: """Return whether ``status`` is terminal: faulted, or done with a finished generation to deliver.""" has_generation = len(status.generations) > 0 diff --git a/horde/telemetry.py b/horde/telemetry.py index 1fa03ed2..412c9740 100644 --- a/horde/telemetry.py +++ b/horde/telemetry.py @@ -40,7 +40,7 @@ import threading import time from collections.abc import Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Protocol, cast, override +from typing import TYPE_CHECKING, Any, cast, override import logfire from loguru import logger as _loguru_logger @@ -52,22 +52,53 @@ if TYPE_CHECKING: from contextlib import AbstractContextManager from flask import Flask -class _HordeLogger(Protocol): - """Subset of the loguru logger augmented with Horde's custom INIT levels. +_INIT_LOG_CALLER_DEPTH: int = 2 +"""The frames between the loguru call and the code reporting an init status: ``_log_init`` and the ``init_*`` +method, so the record carries the reporting function and line.""" - ``horde.logger`` binds ``init_ok`` / ``init_warn`` / ``init_err`` onto the - loguru ``Logger`` class at import time via ``partialmethod``; those dynamic - attributes are invisible to static analysis, so this Protocol re-declares - the subset used in this module. +_INIT_FALLBACK_LEVEL: str = "INFO" +"""The level an init status is logged at in a process that has not registered Horde's INIT levels.""" + + +class _HordeLogger: + """The loguru logger with Horde's INIT levels, usable before or without ``horde.logger``. + + ``horde.logger`` registers the ``INIT_OK``, ``INIT_WARN`` and ``INIT_ERR`` + levels at import time; the Flask application imports it, so its init + statuses keep those levels. The chat front shares Pyroscope startup with + this module but configures its own logging and does not import + ``horde.logger``, so there a status is logged at INFO with the status in + the message. """ - def init_ok(self, message: str, *, status: str) -> None: ... - def init_warn(self, message: str, *, status: str) -> None: ... - def init_err(self, message: str, *, status: str) -> None: ... - def add(self, sink: Any, **kwargs: Any) -> int: ... + def init_ok(self, message: str, *, status: str) -> None: + """Log ``message`` with ``status`` at the ``INIT_OK`` level.""" + self._log_init("INIT_OK", message, status) + + def init_warn(self, message: str, *, status: str) -> None: + """Log ``message`` with ``status`` at the ``INIT_WARN`` level.""" + self._log_init("INIT_WARN", message, status) + + def init_err(self, message: str, *, status: str) -> None: + """Log ``message`` with ``status`` at the ``INIT_ERR`` level.""" + self._log_init("INIT_ERR", message, status) + + def add(self, sink: Any, **kwargs: Any) -> int: + """Add a loguru sink and return its id.""" + return _loguru_logger.add(sink, **kwargs) + + @staticmethod + def _log_init(level: str, message: str, status: str) -> None: + caller = _loguru_logger.opt(depth=_INIT_LOG_CALLER_DEPTH) + try: + _loguru_logger.level(level) + except ValueError: + caller.log(_INIT_FALLBACK_LEVEL, "{message}: {status}", message=message, status=status) + return + caller.log(level, message, status=status) -logger: _HordeLogger = cast("_HordeLogger", _loguru_logger) +logger: _HordeLogger = _HordeLogger() _initialized_early = False _initialized_late = False @@ -314,7 +345,7 @@ def init_telemetry_early(app: Flask) -> None: _initialized_early = True sdk_enabled = otel_sdk_enabled() - span_processors = _init_pyroscope(span_profiles_enabled=sdk_enabled) + span_processors = init_pyroscope(span_profiles_enabled=sdk_enabled) if not sdk_enabled: logger.init_warn("Telemetry", status="OpenTelemetry disabled") @@ -418,12 +449,28 @@ def build_sampling_options() -> logfire.SamplingOptions: return logfire.SamplingOptions(head=ratio) -def _init_pyroscope(*, span_profiles_enabled: bool = True) -> list[SpanProcessor]: +def init_pyroscope( + *, + span_profiles_enabled: bool = True, + default_service_name: str = "ai-horde", +) -> list[SpanProcessor]: """Start continuous profiling and optionally enable span correlation. ``PyroscopeSpanProcessor`` adds per-root-span identifiers to profiling samples. Keep it behind its own explicit gate so ordinary continuous profiling does not create an unbounded series for every request. + + Args: + span_profiles_enabled: Whether the OpenTelemetry SDK runs, which span + correlation requires. + default_service_name: The Pyroscope application name when + ``OTEL_SERVICE_NAME`` is unset. Each process passes its own + OpenTelemetry default so its profiles are not merged with + another service's. + + Returns: + The span processors to add to the tracer provider; empty when span + profiles are off or unavailable. """ if os.environ.get("PYROSCOPE_ENABLED", "").lower() != "true": return [] @@ -432,7 +479,7 @@ def _init_pyroscope(*, span_profiles_enabled: bool = True) -> list[SpanProcessor import pyroscope pyroscope.configure( - application_name=os.environ.get("OTEL_SERVICE_NAME", "ai-horde"), + application_name=os.environ.get("OTEL_SERVICE_NAME", default_service_name), server_address=os.environ.get("PYROSCOPE_SERVER_ADDRESS", "http://localhost:4040"), tags={ "environment": os.environ.get("DEPLOYMENT_ENVIRONMENT", "development"), diff --git a/tests/unit/oai/test_delta_relay.py b/tests/unit/oai/test_delta_relay.py index aff8a7a6..15aa8dca 100644 --- a/tests/unit/oai/test_delta_relay.py +++ b/tests/unit/oai/test_delta_relay.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import json from collections.abc import Awaitable, Callable, Coroutine +from types import SimpleNamespace from typing import Any import fakeredis @@ -174,6 +175,52 @@ async def test_reader_queues_each_registered_requests_entries_in_stream_order() assert entries_after_second_read == [] +class RecordedSpan: + """A stand-in for a Logfire span that keeps its name and the attributes set on it.""" + + def __init__(self, name: str) -> None: + self.name = name + self.attributes: dict[str, Any] = {} + + def __enter__(self) -> RecordedSpan: + return self + + def __exit__(self, *_exc_info: object) -> None: + return None + + def set_attribute(self, key: str, value: Any) -> None: + self.attributes[key] = value + + +async def test_reader_records_one_span_per_read_that_returns_entries(monkeypatch: pytest.MonkeyPatch) -> None: + """The span counts every returned entry and its bytes, and the lag runs from the oldest entry's stream id.""" + spans: list[RecordedSpan] = [] + monkeypatch.setattr(waiter.logfire, "span", lambda name, **_attributes: spans.append(RecordedSpan(name)) or spans[-1]) + monkeypatch.setattr(waiter, "time", SimpleNamespace(time=lambda: 1_000_002.5)) + redis_client = fakeredis.FakeAsyncRedis() + reader = DeltaStreamReader(redis_client) + try: + with reader.register("request-a"), reader.register("request-b"): + await reader.read_once(redis_client) + spans_after_empty_read = len(spans) + await redis_client.xadd(f"{DELTA_STREAM_KEY_PREFIX}request-a", {"gen": "g1", "seq": "1", "content": "Hi"}, id="1000001000-0") + await redis_client.xadd(f"{DELTA_STREAM_KEY_PREFIX}request-b", {"gen": "g2", "ended": "1"}, id="1000000000-0") + await reader.read_once(redis_client) + finally: + await redis_client.close() + + assert spans_after_empty_read == 0 + (span,) = spans + assert span.name == waiter.DELTA_READ_SPAN_NAME + assert span.attributes == { + waiter.DELTA_READ_ENTRY_COUNT_ATTRIBUTE: 2, + waiter.DELTA_READ_BYTE_COUNT_ATTRIBUTE: sum( + len(text) for text in ("gen", "g1", "seq", "1", "content", "Hi", "gen", "g2", "ended", "1") + ), + waiter.DELTA_READ_FIRST_ENTRY_LAG_ATTRIBUTE: pytest.approx(2.5), + } + + @pytest.mark.parametrize( ("fields", "worker_id", "model"), [ diff --git a/tests/unit/oai/test_errors.py b/tests/unit/oai/test_errors.py index be06059f..c03f0e40 100644 --- a/tests/unit/oai/test_errors.py +++ b/tests/unit/oai/test_errors.py @@ -224,6 +224,31 @@ async def test_horde_rate_limit_passes_through_with_its_headers(start_front: Sta } +async def test_horde_rate_limit_without_an_rc_is_rate_limit_exceeded_with_its_retry_after(start_front: StartFront) -> None: + async with start_front() as front: + front.horde.submit_route.respond( + 429, + json={"message": "90 per 1 minute"}, + headers={"Retry-After": "7", "X-RateLimit-Limit": "90", "X-RateLimit-Remaining": "0"}, + ) + response = await front.client.post("/v1/chat/completions", json=build_chat_body()) + + assert response.status_code == 429 + assert response.headers["retry-after"] == "7" + assert response.headers["x-ratelimit-limit"] == "90" + assert response.headers["x-ratelimit-remaining"] == "0" + assert SHOULD_RETRY not in response.headers + assert response.json() == { + "error": { + "message": "90 per 1 minute", + "type": "rate_limit_error", + "param": None, + "code": "rate_limit_exceeded", + "metadata": {"retry_after": 7}, + }, + } + + async def test_chat_admission_disabled_is_503_with_retry_after(start_front: StartFront) -> None: async with start_front() as front: front.horde.submit_route.respond(503, json={"message": "Chat is switched off.", "rc": "ChatAdmissionDisabled"}) diff --git a/tests/unit/oai/test_observability.py b/tests/unit/oai/test_observability.py index 9ece7573..aa1b4046 100644 --- a/tests/unit/oai/test_observability.py +++ b/tests/unit/oai/test_observability.py @@ -44,6 +44,7 @@ from tests.unit.oai.conftest import ( done_status, metric_points, wait_for_call, + waiting_status, ) if TYPE_CHECKING: @@ -73,10 +74,28 @@ FINISHED_FIELDS: frozenset[str] = frozenset( "wake_count", "estimated_usage", "done_without_generation_reads", + "requested_model", + "served_model", + "retry_after", + "first_read_wait_time", + "first_read_eligible_worker_threads", }, ) """The structured fields of the finished-request line.""" +FIRST_READ_WAIT_TIME: int = 2 +"""The wait estimate of the first status an admitted request reads; under the 5 s test budget, so ``fail_fast`` +admits.""" + +FIRST_READ_ELIGIBLE_WORKER_THREADS: int = 3 +"""The eligible worker threads of the first status an admitted request reads.""" + +FINISH_DELAY_SECONDS: float = 0.2 +"""How long after the request starts a test marks it done and signals it, so the first read sees it waiting.""" + +QUEUE_REFUSAL_FIELDS: frozenset[str] = frozenset({"queue_reason", "wait_time", "eligible_worker_threads"}) +"""The structured fields a ``queue_unavailable`` refusal adds to the finished-request line.""" + class CapturedLogs: """Hold the Loguru records written while a capture is open.""" @@ -237,6 +256,14 @@ async def test_finished_line_carries_the_wait_fields_and_no_key_prompt_or_title( assert finished_fields["estimated_usage"] is False assert finished_fields["woke_at"] >= finished_fields["submitted_at"] assert finished_fields["waited_seconds"] >= 0 + assert finished_fields["requested_model"] == MODEL_NAME + assert finished_fields["served_model"] == MODEL_NAME + assert finished_fields["retry_after"] is None + assert not QUEUE_REFUSAL_FIELDS & finished_fields.keys() + assert finished_record["message"].endswith( + f" requested_model={MODEL_NAME} served_model={MODEL_NAME} retry_after=none " + "first_read_wait_time=none first_read_eligible_worker_threads=none", + ) logged_text = finished_record["message"] + repr(finished_fields) for private_text in (API_KEY, CLIENT_TITLE, PROMPT_TEXT): assert private_text not in logged_text @@ -255,3 +282,107 @@ async def test_unsubmitted_request_logs_a_finished_line_without_wake_fields( assert finished_record["extra"]["request_id"] is None assert finished_record["extra"]["code"] == "ChatAdmissionDisabled" assert finished_record["extra"]["wake_count"] == 0 + + +@pytest.mark.parametrize( + ("status_payload", "reason", "retry_after", "wait_time", "eligible_worker_threads"), + [ + (waiting_status(is_possible=False, wait_time=2), "no_eligible_threads", 30, 2, 0), + (waiting_status(wait_time=600, eligible_worker_threads=4), "wait_exceeds_budget", 600, 600, 4), + ], + ids=["no_eligible_threads", "wait_exceeds_budget"], +) +async def test_queue_unavailable_records_the_reason_and_logs_the_estimate_and_retry_after( + start_front: StartFront, + captured_logs: CapturedLogs, + status_payload: dict[str, object], + reason: str, + retry_after: int, + wait_time: int, + eligible_worker_threads: int, +) -> None: + async with start_front() as front: + front.horde.status_payload = status_payload + response = await front.client.post("/v1/chat/completions", json=build_chat_body(horde={"availability_mode": "fail_fast"})) + + assert response.status_code == 503 + assert front.metric_total(ERRORS_NAME, horde_status_code=503, horde_code="queue_unavailable", horde_reason=reason) == 1 + (finished_record,) = captured_logs.finished_records() + finished_fields = finished_record["extra"] + assert finished_fields["code"] == "queue_unavailable" + assert finished_fields["requested_model"] == MODEL_NAME + assert finished_fields["served_model"] is None + assert finished_fields["retry_after"] == retry_after + assert finished_fields["queue_reason"] == reason + assert finished_fields["wait_time"] == wait_time + assert finished_fields["eligible_worker_threads"] == eligible_worker_threads + assert finished_fields["first_read_wait_time"] == wait_time + assert finished_fields["first_read_eligible_worker_threads"] == eligible_worker_threads + assert finished_record["message"].endswith( + f" requested_model={MODEL_NAME} served_model=none retry_after={retry_after} queue_reason={reason} " + f"wait_time={wait_time} eligible_worker_threads={eligible_worker_threads} " + f"first_read_wait_time={wait_time} first_read_eligible_worker_threads={eligible_worker_threads}", + ) + + +@pytest.mark.parametrize( + ("availability_mode", "logged_wait_time", "logged_eligible_worker_threads"), + [ + ("fail_fast", FIRST_READ_WAIT_TIME, FIRST_READ_ELIGIBLE_WORKER_THREADS), + ("wait", None, None), + ], + ids=["fail_fast", "wait"], +) +async def test_admitted_request_logs_the_first_reads_estimate_only_in_the_fail_fast_mode( + start_front: StartFront, + captured_logs: CapturedLogs, + availability_mode: str, + logged_wait_time: int | None, + logged_eligible_worker_threads: int | None, +) -> None: + """A later read with another estimate does not change the logged first-read facts.""" + async with start_front() as front: + front.horde.status_payload = waiting_status( + wait_time=FIRST_READ_WAIT_TIME, + eligible_worker_threads=FIRST_READ_ELIGIBLE_WORKER_THREADS, + ) + finisher = asyncio.create_task(front.horde.finish_later(FINISH_DELAY_SECONDS)) + response = await front.client.post( + "/v1/chat/completions", + json=build_chat_body(horde={"availability_mode": availability_mode}), + ) + await finisher + + assert response.status_code == 200 + (finished_record,) = captured_logs.finished_records() + finished_fields = finished_record["extra"] + assert finished_fields["code"] == "ok" + assert finished_fields["first_read_wait_time"] == logged_wait_time + assert finished_fields["first_read_eligible_worker_threads"] == logged_eligible_worker_threads + assert finished_record["message"].endswith( + f" first_read_wait_time={logged_wait_time if logged_wait_time is not None else 'none'} " + f"first_read_eligible_worker_threads={logged_eligible_worker_threads if logged_eligible_worker_threads is not None else 'none'}", + ) + + +async def test_an_error_other_than_queue_unavailable_records_an_empty_reason(start_front: StartFront) -> None: + async with start_front(FAST_TEST_SETTINGS.model_copy(update={"nonstream_timeout_seconds": 0.3})) as front: + response = await front.client.post("/v1/chat/completions", json=build_chat_body()) + + assert response.status_code == 408 + assert front.metric_total(ERRORS_NAME, horde_code="ChatRequestTimeout", horde_reason="") == 1 + + +async def test_horde_rate_limit_without_an_rc_records_rate_limit_exceeded( + start_front: StartFront, + captured_logs: CapturedLogs, +) -> None: + async with start_front() as front: + front.horde.submit_route.respond(429, json={"message": "90 per 1 minute"}, headers={"Retry-After": "7"}) + response = await front.client.post("/v1/chat/completions", json=build_chat_body()) + + assert response.status_code == 429 + assert front.metric_total(ERRORS_NAME, horde_status_code=429, horde_code="rate_limit_exceeded") == 1 + (finished_record,) = captured_logs.finished_records() + assert finished_record["extra"]["code"] == "rate_limit_exceeded" + assert finished_record["extra"]["retry_after"] == 7 diff --git a/tests/unit/oai/test_telemetry_setup.py b/tests/unit/oai/test_telemetry_setup.py index 5dfb4ec0..dd553407 100644 --- a/tests/unit/oai/test_telemetry_setup.py +++ b/tests/unit/oai/test_telemetry_setup.py @@ -81,3 +81,39 @@ def test_sdk_disabled_keeps_telemetry_off(telemetry_environment: pytest.MonkeyPa assert front_main.configure_telemetry() is False assert configure_calls == [] + + +def test_profiling_only_starts_pyroscope_as_horde_oai_without_the_sdk( + telemetry_environment: pytest.MonkeyPatch, +) -> None: + pyroscope_calls: list[dict[str, Any]] = [] + configure_calls: list[dict[str, Any]] = [] + telemetry_environment.setenv("PYROSCOPE_ENABLED", "true") + telemetry_environment.setenv("OTEL_SDK_DISABLED", "true") + telemetry_environment.setattr(front_main, "init_pyroscope", lambda **kwargs: pyroscope_calls.append(kwargs) or []) + telemetry_environment.setattr(logfire, "configure", lambda **kwargs: configure_calls.append(kwargs)) + + assert front_main.configure_telemetry() is False + assert pyroscope_calls == [{"span_profiles_enabled": False, "default_service_name": "horde-oai"}] + assert configure_calls == [] + + +def test_pyroscope_span_processors_join_the_front_tracer_provider(telemetry_environment: pytest.MonkeyPatch) -> None: + span_processor = object() + pyroscope_calls: list[dict[str, Any]] = [] + configure_calls: list[dict[str, Any]] = [] + telemetry_environment.setenv("PYROSCOPE_ENABLED", "true") + telemetry_environment.setattr( + front_main, + "init_pyroscope", + lambda **kwargs: pyroscope_calls.append(kwargs) or [span_processor], + ) + telemetry_environment.setattr(logfire, "configure", lambda **kwargs: configure_calls.append(kwargs)) + telemetry_environment.setattr(logfire, "loguru_handler", lambda: {"sink": print}) + telemetry_environment.setattr(logger, "add", lambda **_kwargs: None) + telemetry_environment.setattr(logfire, "instrument_httpx", lambda **_kwargs: None) + + assert front_main.configure_telemetry() is True + assert pyroscope_calls == [{"span_profiles_enabled": True, "default_service_name": "horde-oai"}] + (configure_kwargs,) = configure_calls + assert configure_kwargs["additional_span_processors"] == [span_processor] diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 5f60ffdc..b3f07567 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -339,7 +339,7 @@ def test_profiling_only_initializes_pyroscope_without_logfire(monkeypatch): monkeypatch.setattr(telemetry, "_initialized_early", False) monkeypatch.setattr( telemetry, - "_init_pyroscope", + "init_pyroscope", lambda *, span_profiles_enabled: pyroscope_calls.append(span_profiles_enabled) or [], ) monkeypatch.setattr( @@ -379,7 +379,7 @@ def test_pyroscope_span_processor_is_disabled_by_default(monkeypatch, span_profi else: monkeypatch.setenv("PYROSCOPE_SPAN_PROFILES", span_profiles) - assert telemetry._init_pyroscope() == [] + assert telemetry.init_pyroscope() == [] assert len(configure_calls) == 1 @@ -402,7 +402,7 @@ def test_pyroscope_span_processor_requires_explicit_opt_in(monkeypatch): monkeypatch.setenv("PYROSCOPE_SPAN_PROFILES", "true") monkeypatch.setenv("PYROSCOPE_SPAN_PROFILES_SAMPLE_RATE", "1") - processors = telemetry._init_pyroscope() + processors = telemetry.init_pyroscope() assert len(processors) == 1 assert isinstance(processors[0], telemetry._TraceRatioSpanProcessor) @@ -430,7 +430,7 @@ def test_profiling_only_skips_span_processor_when_otel_sdk_is_disabled(monkeypat monkeypatch.setenv("PYROSCOPE_ENABLED", "true") monkeypatch.setenv("PYROSCOPE_SPAN_PROFILES", "true") - assert telemetry._init_pyroscope(span_profiles_enabled=False) == [] + assert telemetry.init_pyroscope(span_profiles_enabled=False) == [] assert len(configure_calls) == 1