diff --git a/docs/content/docs/proxy.mdx b/docs/content/docs/proxy.mdx index 35bb7a77f..3390a086d 100644 --- a/docs/content/docs/proxy.mdx +++ b/docs/content/docs/proxy.mdx @@ -284,6 +284,7 @@ Rewrite the upstream model per request — for example, send small, tool-free ca | `--telemetry` / `HEADROOM_TELEMETRY` | off | **Local-only** usage stats for your own `/stats`, `/metrics`, and dashboard. Nothing leaves the machine. | | `--log-file` / `HEADROOM_LOG_FILE` | none | JSONL request/response log. | | `--log-messages` | `false` | Include full message bodies in the log (may contain sensitive data). | +| `HEADROOM_LOG_LEVEL` | `warning` | uvicorn's log level (`critical`, `error`, `warning`, `info`, `debug`, `trace`). Raise to `info` for the per-request access log when diagnosing a deployed proxy. An unrecognized value warns and falls back to `warning`. | | `HEADROOM_OTEL_METRICS_ENABLED` | `false` | Export OpenTelemetry metrics (`HEADROOM_OTEL_METRICS_ENDPOINT`, …). See [OTLP export](/docs/metrics#opentelemetry-otlp-export). | | `HEADROOM_LANGFUSE_ENABLED` | `false` | Emit Langfuse traces (`LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY`). | diff --git a/headroom/proxy/handlers/streaming.py b/headroom/proxy/handlers/streaming.py index 55c03e1fa..5f86d48a4 100644 --- a/headroom/proxy/handlers/streaming.py +++ b/headroom/proxy/handlers/streaming.py @@ -1340,11 +1340,22 @@ class StreamingMixin: if upstream_response is None: raise last_connect_error or RuntimeError("upstream connection did not start") # Retries exhausted (or a transport failure escaped the loop): emit a - # clean SSE error instead of letting an h2 StreamReset bubble up as a - # 502. Covers ConnectError/timeouts and Local/RemoteProtocolError. (#1639) + # clean SSE error instead of letting an h2 StreamReset bubble up as an + # unhandled 502. Covers ConnectError/timeouts and Local/RemoteProtocol- + # Error. (#1639) + # + # The status MUST stay non-2xx. No body byte has been forwarded yet, so + # the status line is still ours to set, and a 200 here is + # indistinguishable — to every Anthropic/OpenAI SDK — from a successful + # stream that produced no events: the client reports "empty or malformed + # response (HTTP 200)" and cannot retry, because 200 is not retryable. + # 502 keeps the structured SSE body for clients that read it while + # letting SDK retry logic treat a transient upstream transport failure + # as what it is. except httpx.TransportError as e: error_msg = str(e) or repr(e) logger.error(f"[{request_id}] Connection error to upstream API: {error_msg}") + self.metrics.record_upstream_connection_error(provider) async def _error_gen(): error_event = { @@ -1357,7 +1368,11 @@ class StreamingMixin: yield f"event: error\ndata: {json.dumps(error_event)}\n\n".encode() self._cleanup_mid_turn_stream(session_key) - return StreamingResponse(_error_gen(), media_type="text/event-stream") + return StreamingResponse( + _error_gen(), + status_code=502, + media_type="text/event-stream", + ) # Capture Codex rate-limit window data from the upstream response # headers, for *every* status. Codex (gpt-5.x) almost always streams, so diff --git a/headroom/proxy/prometheus_metrics.py b/headroom/proxy/prometheus_metrics.py index 3586c7a7b..503a7ffe2 100644 --- a/headroom/proxy/prometheus_metrics.py +++ b/headroom/proxy/prometheus_metrics.py @@ -152,6 +152,14 @@ class PrometheusMetrics: # whether the compression budget is too tight vs. a real bug. self.compression_failed_by_reason: dict[str, int] = defaultdict(int) + # Upstream transport failures on the streaming path, keyed by provider. + # Raised when every connect retry is exhausted and the proxy synthesizes + # its own error response instead of forwarding an upstream status. That + # path emits no upstream status code to attribute the failure to, so + # without this counter it is invisible in metrics and survives only as a + # log line. + self.upstream_connection_errors_by_provider: dict[str, int] = defaultdict(int) + # Kompress size-gate outcomes, keyed by outcome ("within", # "exceeded"). The gate routes oversized blocks away from ML # compression (ContentRouter._kompress_max_tokens, #1171). This @@ -355,6 +363,7 @@ class PrometheusMetrics: self.savings_by_source.clear() with self._obs_counter_lock: self.compression_failed_by_reason.clear() + self.upstream_connection_errors_by_provider.clear() self.kompress_size_gate_by_outcome.clear() self.compression_quarantine_by_event.clear() @@ -568,6 +577,18 @@ class PrometheusMetrics: with self._obs_counter_lock: self.compression_failed_by_reason[reason or "error"] += 1 + def record_upstream_connection_error(self, provider: str) -> None: + """Record one exhausted-retries upstream transport failure. + + Called from the streaming handler's ``httpx.TransportError`` fallback + (handlers/streaming.py), where the proxy synthesizes its own 502 + because no upstream response ever arrived. Guarded by + ``_obs_counter_lock`` for the same reason as + ``record_compression_failed``. + """ + with self._obs_counter_lock: + self.upstream_connection_errors_by_provider[provider or "unknown"] += 1 + def record_kompress_size_gate(self, outcome: str) -> None: """Record one kompress size-gate decision, bucketed by ``outcome``. @@ -1404,9 +1425,23 @@ class PrometheusMetrics: # then format outside it (see _obs_counter_lock). with self._obs_counter_lock: compression_failed = dict(self.compression_failed_by_reason) + upstream_conn_errors = dict(self.upstream_connection_errors_by_provider) kompress_size_gate = dict(self.kompress_size_gate_by_outcome) compression_quarantine = dict(self.compression_quarantine_by_event) + if upstream_conn_errors: + lines.extend( + [ + "# HELP headroom_upstream_connection_errors_total Exhausted-retries upstream transport failures by provider; the proxy answered 502 itself because no upstream response arrived", + "# TYPE headroom_upstream_connection_errors_total counter", + ] + ) + for prov, count in upstream_conn_errors.items(): + lines.append( + f'headroom_upstream_connection_errors_total{{provider="{_escape_label_value(prov)}"}} {count}' + ) + lines.append("") + if compression_failed: lines.extend( [ diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index d38b44209..f90b082ee 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -5485,11 +5485,18 @@ def run_server( else: app_target = create_app(config) + # "warning" keeps the default terminal quiet (uvicorn's access log is one line + # per request). It was previously hardcoded, which left deployed proxies with + # no way to turn request logging on: operators diagnosing a production + # incident could not see uvicorn's view of the traffic at all, with no env var + # and no CLI flag to change it. Overridable now; the default is unchanged. + uvicorn_log_level = _resolve_uvicorn_log_level() + uvicorn.run( app_target, host=config.host, port=config.port, - log_level="warning", + log_level=uvicorn_log_level, workers=workers if workers > 1 else None, # None = single process (default) limit_concurrency=limit_concurrency, # Defense-in-depth: the loopback guard for /debug/* endpoints trusts @@ -5556,6 +5563,30 @@ def _get_env_str(name: str, default: str) -> str: return os.environ.get(name, default) +# uvicorn rejects anything outside this set with a KeyError during startup, so an +# operator typo in HEADROOM_LOG_LEVEL must not be able to stop the proxy booting. +_UVICORN_LOG_LEVELS = frozenset({"critical", "error", "warning", "info", "debug", "trace"}) +_UVICORN_LOG_LEVEL_DEFAULT = "warning" + + +def _resolve_uvicorn_log_level() -> str: + """Resolve uvicorn's log level from ``HEADROOM_LOG_LEVEL``. + + Falls back to the previous hardcoded default on an unset or unrecognized + value, warning loudly rather than failing the boot. + """ + raw = _get_env_str("HEADROOM_LOG_LEVEL", _UVICORN_LOG_LEVEL_DEFAULT).strip().lower() + if raw in _UVICORN_LOG_LEVELS: + return raw + logger.warning( + "Ignoring unrecognized HEADROOM_LOG_LEVEL=%r; using %r. Valid values: %s", + raw, + _UVICORN_LOG_LEVEL_DEFAULT, + ", ".join(sorted(_UVICORN_LOG_LEVELS)), + ) + return _UVICORN_LOG_LEVEL_DEFAULT + + def _parse_exclude_tools(cli_excludes: str | None) -> set[str]: """Parse extra never-compress tool names from CLI args and env var. diff --git a/tests/test_h2_stream_reset_retry.py b/tests/test_h2_stream_reset_retry.py index f1774999c..14d916b7b 100644 --- a/tests/test_h2_stream_reset_retry.py +++ b/tests/test_h2_stream_reset_retry.py @@ -30,6 +30,7 @@ def _mock_proxy(): proxy._config.retry_max_delay_ms = 0 proxy.config = proxy._config proxy.memory_handler = None + proxy.metrics = MagicMock() proxy._parse_sse_usage_from_buffer = MagicMock(return_value=None) proxy._finalize_stream_response = AsyncMock(return_value=None) return proxy @@ -142,3 +143,37 @@ async def test_stream_reset_exhaustion_yields_sse_error_not_crash(): assert proxy.http_client.send.await_count == 2 assert b"event: error" in body assert b"connection_error" in body + + +@pytest.mark.asyncio +async def test_stream_reset_exhaustion_status_is_not_200(): + """The synthesized error response must not claim success. + + A 200 here is indistinguishable, to every Anthropic/OpenAI SDK, from a + successful stream that produced no events — the client reports "empty or + malformed response (HTTP 200)" and cannot retry, because 200 is not a + retryable status. This asserts the status line specifically: the body + assertions above passed for the entire time the status was 200. + """ + proxy = _mock_proxy() + proxy.http_client.build_request = MagicMock(return_value=MagicMock()) + proxy.http_client.send = AsyncMock(side_effect=httpx.RemoteProtocolError("reset")) + + result = await _run_stream(proxy) + + assert result.status_code == 502 + proxy.metrics.record_upstream_connection_error.assert_called_once() + + +@pytest.mark.asyncio +async def test_successful_stream_still_returns_200(): + """Guard the happy path against the 502 change above.""" + proxy = _mock_proxy() + good = _good_stream_response([b'event: message_stop\ndata: {"type":"message_stop"}\n\n']) + proxy.http_client.build_request = MagicMock(return_value=MagicMock()) + proxy.http_client.send = AsyncMock(return_value=good) + + result = await _run_stream(proxy) + + assert result.status_code == 200 + proxy.metrics.record_upstream_connection_error.assert_not_called() diff --git a/tests/test_prometheus_obs_counters.py b/tests/test_prometheus_obs_counters.py index 258cb23e1..41c4a7683 100644 --- a/tests/test_prometheus_obs_counters.py +++ b/tests/test_prometheus_obs_counters.py @@ -8,6 +8,8 @@ Covers the related counters added to ``PrometheusMetrics``: via the observer hook, split into "exceeded" vs "within". * ``headroom_compression_quarantine_total{event}`` — records quarantine activation and immediate executor skips while a timed-out worker remains. +* ``headroom_upstream_connection_errors_total{provider}`` — recorded when the + streaming path exhausts its connect retries and answers 502 itself. Imports only the metrics module so the test stays free of heavy ML deps. """ @@ -41,6 +43,35 @@ def test_record_compression_failed_empty_reason_defaults_to_error() -> None: assert metrics.compression_failed_by_reason["error"] == 1 +def test_record_upstream_connection_error_buckets_by_provider() -> None: + metrics = PrometheusMetrics() + + metrics.record_upstream_connection_error("anthropic") + metrics.record_upstream_connection_error("openai") + metrics.record_upstream_connection_error("openai") + + assert metrics.upstream_connection_errors_by_provider["anthropic"] == 1 + assert metrics.upstream_connection_errors_by_provider["openai"] == 2 + + +def test_record_upstream_connection_error_empty_provider_defaults_to_unknown() -> None: + metrics = PrometheusMetrics() + + metrics.record_upstream_connection_error("") + + assert metrics.upstream_connection_errors_by_provider["unknown"] == 1 + + +async def test_upstream_connection_errors_exported_in_prometheus_text() -> None: + metrics = PrometheusMetrics() + metrics.record_upstream_connection_error("anthropic") + + text = await metrics.export() + + assert "# TYPE headroom_upstream_connection_errors_total counter" in text + assert 'headroom_upstream_connection_errors_total{provider="anthropic"} 1' in text + + def test_record_kompress_size_gate_buckets_by_outcome() -> None: metrics = PrometheusMetrics() diff --git a/tests/test_uvicorn_log_level_env.py b/tests/test_uvicorn_log_level_env.py new file mode 100644 index 000000000..64b8f8ced --- /dev/null +++ b/tests/test_uvicorn_log_level_env.py @@ -0,0 +1,45 @@ +"""HEADROOM_LOG_LEVEL resolution for uvicorn's log level. + +uvicorn's level was previously hardcoded to "warning", so a deployed proxy had +no way to turn request logging on — no env var, no CLI flag. These tests pin the +override and, more importantly, pin that an operator typo cannot stop the proxy +booting (uvicorn raises on an unknown level). +""" + +from __future__ import annotations + +import pytest + +from headroom.proxy.server import _resolve_uvicorn_log_level + + +@pytest.fixture(autouse=True) +def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("HEADROOM_LOG_LEVEL", raising=False) + + +def test_defaults_to_warning_when_unset() -> None: + assert _resolve_uvicorn_log_level() == "warning" + + +@pytest.mark.parametrize("level", ["critical", "error", "warning", "info", "debug", "trace"]) +def test_accepts_every_uvicorn_level(level: str, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("HEADROOM_LOG_LEVEL", level) + + assert _resolve_uvicorn_log_level() == level + + +def test_normalizes_case_and_whitespace(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("HEADROOM_LOG_LEVEL", " INFO \n") + + assert _resolve_uvicorn_log_level() == "info" + + +@pytest.mark.parametrize("bad", ["verbose", "WARN", "", "10"]) +def test_unrecognized_value_falls_back_instead_of_raising( + bad: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """A typo must degrade to the old default, never fail the boot.""" + monkeypatch.setenv("HEADROOM_LOG_LEVEL", bad) + + assert _resolve_uvicorn_log_level() == "warning"