test(perf): gate the gateway's own overhead header, redis latency and postgres load in the perf cases (#47472)

* test(perf): gate the gateway's own overhead header, redis latency and postgres load in the perf cases

* test(perf): keep the gateway_added_seconds limits next to the header ones

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/e2e/gateway/perf_ci_config.yml

@@ -27,6 +27,8 @@ litellm_settings:
   callbacks:
     - prometheus
     - otel
+  service_callback:
+    - prometheus_system
   require_auth_for_metrics_endpoint: false
   drop_params: true
   request_timeout: 120

tests/e2e/perf/cases/mixed_size_requests_2rps.py

@@ -64,6 +64,14 @@ CASE: Final = PerformanceTestCase(
         Limit("unexpected_status_count", max=3),
         Limit("gateway_added_seconds", percentile=50, max=0.58),
         Limit("gateway_added_seconds", percentile=99, max=6.1),
+        Limit("litellm_overhead_seconds", percentile=50, max=0.05),
+        Limit("litellm_overhead_seconds", percentile=99, max=2.2),
+        Limit("litellm_response_seconds", percentile=50, max=1.2),
+        Limit("litellm_response_seconds", percentile=99, max=3.8),
+        Limit("prometheus_redis_latency_mean_seconds", max=0.003),
+        Limit("postgres_active_sessions", max=0.004),
+        Limit("postgres_slowest_statement_seconds", max=0.045),
+        Limit("postgres_peak_connections", max=30),
         Limit("time_to_first_token_seconds", percentile=50, max=2.7),
         Limit("time_to_first_token_seconds", percentile=99, max=8.8),
         Limit("memory_growth_megabytes", max=600),

tests/e2e/perf/cases/small_bodies_5rps.py

@@ -65,6 +65,14 @@ CASE: Final = PerformanceTestCase(
     limits=(
         Limit("unexpected_status_count", max=3),
         Limit("gateway_added_seconds", percentile=50, max=0.29),
+        Limit("litellm_overhead_seconds", percentile=50, max=0.02),
+        Limit("litellm_overhead_seconds", percentile=99, max=0.8),
+        Limit("litellm_response_seconds", percentile=50, max=0.8),
+        Limit("litellm_response_seconds", percentile=99, max=2.8),
+        Limit("prometheus_redis_latency_mean_seconds", max=0.006),
+        Limit("postgres_active_sessions", max=0.006),
+        Limit("postgres_slowest_statement_seconds", max=0.035),
+        Limit("postgres_peak_connections", max=30),
         Limit("time_to_first_token_seconds", percentile=99, max=3.7),
         Limit("memory_growth_megabytes", max=100),
         Limit("postgres_calls_per_request", max=1.8),

tests/e2e/perf/setup/datastore.py

@@ -1,7 +1,10 @@
 """Postgres and Redis work the gateway did during the traffic phase, split into what a request waits on and
 what runs after the response. Both come from the servers' own accounting, so nothing here depends on how the
 gateway talks to them: Postgres statement counts from `pg_stat_statements` (read before and after the run) and
-Redis commands from a `MONITOR` connection held open for the run. `DATABASE_URL` and `REDIS_HOST` are the
+Redis commands from a `MONITOR` connection held open for the run, Postgres execution time per statement from the
+same `pg_stat_statements` rows (`postgres_active_sessions`, the average-active-sessions number RDS Performance
+Insights calls DB load, and the slowest statement's mean) and the connection count from `pg_stat_activity` sampled
+once a second (`postgres_peak_connections`). `DATABASE_URL` and `REDIS_HOST` are the
 runner's own env, as in the other suites; without them the counters are None and any Limit on them reports as
 missing. `E2E_DATABASE_ADMIN_URL`, when set, is used once to create the `pg_stat_statements` extension. Only the
 runtime user's own statements are read: other users' rows come back with a NULL queryid and hidden text."""
@@ -26,13 +29,18 @@ Bucket = Literal["awaited", "background", "plumbing", "unclassified"]
 
 _COMMITS_QUERY: Final = "SELECT xact_commit FROM pg_stat_database WHERE datname = current_database()"
 _STATEMENTS_QUERY: Final = (
-    "SELECT queryid, calls, query FROM pg_stat_statements "
+    "SELECT queryid, calls, query, total_exec_time FROM pg_stat_statements "
     "WHERE dbid = (SELECT oid FROM pg_database WHERE datname = current_database()) "
     "AND userid = (SELECT usesysid FROM pg_user WHERE usename = current_user) AND queryid IS NOT NULL"
 )
 _CREATE_EXTENSION: Final = "CREATE EXTENSION IF NOT EXISTS pg_stat_statements"
+_CONNECTIONS_QUERY: Final = (
+    "SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() "
+    "AND backend_type = 'client backend' AND pid <> pg_backend_pid()"
+)
+_CONNECTIONS_SAMPLE_SECONDS: Final = 1.0
 _COMMITS_ROW: Final = TypeAdapter(tuple[int])
-_STATEMENT_ROWS: Final = TypeAdapter(list[tuple[int, int, str]])
+_STATEMENT_ROWS: Final = TypeAdapter(list[tuple[int, int, str, float]])
 _INFO: Final = TypeAdapter(dict[str, object])
 _INT: Final = TypeAdapter(int)
 _MONITOR_LINE: Final = TypeAdapter(dict[str, object])
@@ -64,6 +72,11 @@ class StatementCalls:
     queryid: int
     query: str
     calls: int
+    total_exec_ms: float = 0.0
+
+    @property
+    def mean_exec_seconds(self) -> float | None:
+        return self.total_exec_ms / 1000 / self.calls if self.calls > 0 else None
 
     @property
     def bucket(self) -> Bucket:
@@ -136,16 +149,85 @@ def statement_deltas(before: DbCounters | None, after: DbCounters | None) -> tup
     """Calls per statement during the run: `after` minus `before` by queryid, new statements whole."""
     if before is None or after is None or before.statements is None or after.statements is None:
         return None
-    earlier: Final = {statement.queryid: statement.calls for statement in before.statements}
+    earlier: Final = {statement.queryid: statement for statement in before.statements}
     return tuple(
-        StatementCalls(
-            queryid=statement.queryid, query=statement.query, calls=statement.calls - earlier.get(statement.queryid, 0)
-        )
+        _statement_delta(statement, earlier.get(statement.queryid))
         for statement in after.statements
-        if statement.calls > earlier.get(statement.queryid, 0)
+        if statement.calls > (0 if statement.queryid not in earlier else earlier[statement.queryid].calls)
     )
 
 
+def _statement_delta(statement: StatementCalls, earlier: StatementCalls | None) -> StatementCalls:
+    return StatementCalls(
+        queryid=statement.queryid,
+        query=statement.query,
+        calls=statement.calls - (earlier.calls if earlier is not None else 0),
+        total_exec_ms=statement.total_exec_ms - (earlier.total_exec_ms if earlier is not None else 0.0),
+    )
+
+
+def active_sessions(statements: tuple[StatementCalls, ...] | None, wall_seconds: float) -> float | None:
+    """Seconds Postgres spent executing the run's statements per second of wall time, the average active
+    sessions figure RDS Performance Insights reports as DB load."""
+    if statements is None or wall_seconds <= 0:
+        return None
+    return sum(statement.total_exec_ms for statement in statements) / 1000 / wall_seconds
+
+
+def slowest_statement(statements: tuple[StatementCalls, ...] | None) -> StatementCalls | None:
+    """The run's statement with the highest mean execution time."""
+    timed: Final = tuple(statement for statement in statements or () if statement.mean_exec_seconds is not None)
+    return max(timed, key=lambda statement: statement.mean_exec_seconds or 0.0, default=None)
+
+
+class ConnectionsSampler:
+    """`pg_stat_activity` client backends on the gateway's database once a second on a background thread, minus
+    this sampler's own connection, so the peak is what the gateway (and the test's own reads) held open."""
+
+    def __init__(self, url: str) -> None:
+        self._url: Final = url
+        self._stop: Final = threading.Event()
+        self._thread: Final = threading.Thread(target=self._run, name="postgres-connections", daemon=True)
+        self.peak: int | None = None
+
+    def start(self) -> None:
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        self._thread.join(timeout=10)
+
+    def _run(self) -> None:
+        import psycopg
+
+        with psycopg.connect(self._url, autocommit=True) as conn:
+            while not self._stop.is_set():
+                (count,) = _COMMITS_ROW.validate_python(conn.execute(_CONNECTIONS_QUERY).fetchone())
+                self.peak = count if self.peak is None else max(self.peak, count)
+                _ = self._stop.wait(_CONNECTIONS_SAMPLE_SECONDS)
+
+
+class NoConnectionsSampler:
+    peak: None = None
+
+    def start(self) -> None:
+        return None
+
+    def stop(self) -> None:
+        return None
+
+
+@contextmanager
+def connections_sampler() -> Generator[ConnectionsSampler | NoConnectionsSampler]:
+    url: Final = os.environ.get("DATABASE_URL")
+    sampler: Final = ConnectionsSampler(url) if url is not None else NoConnectionsSampler()
+    sampler.start()
+    try:
+        yield sampler
+    finally:
+        sampler.stop()
+
+
 def calls_in(bucket: Bucket, items: tuple[StatementCalls, ...] | tuple[RedisCommand, ...] | None) -> int | None:
     if items is None:
         return None
@@ -171,7 +253,10 @@ def read_db_counters() -> DbCounters | None:
             return DbCounters(commits=commits, statements=None)
     return DbCounters(
         commits=commits,
-        statements=tuple(StatementCalls(queryid=queryid, query=query, calls=calls) for queryid, calls, query in rows),
+        statements=tuple(
+            StatementCalls(queryid=queryid, query=query, calls=calls, total_exec_ms=total_ms)
+            for queryid, calls, query, total_ms in rows
+        ),
     )
 
 

tests/e2e/perf/setup/locustfile.py

@@ -137,6 +137,15 @@ def _header(response: FastResponse, name: str) -> str | None:
     return value if isinstance(value, str) else None
 
 
+def _ms_header(response: FastResponse | None, name: str) -> float | None:
+    """A gateway timing header as a number; the gateway writes the literal `None` when it has no value."""
+    value: Final = _header(response, name) if response is not None else None
+    try:
+        return float(value) if value is not None else None
+    except ValueError:
+        return None
+
+
 def _quit(user: FastHttpUser) -> None:
     """End the run once every record is written. Quitting from a separate greenlet: `runner.quit`
     kills the user greenlets, and the user calling it would be killed mid-shutdown."""
@@ -309,6 +318,8 @@ class PlanUser(FastHttpUser):
                 call_id=_header(response, "x-litellm-call-id") if response is not None else None,
                 error=error,
                 response_id=identifier,
+                litellm_overhead_ms=_ms_header(response, "x-litellm-overhead-duration-ms"),
+                litellm_response_ms=_ms_header(response, "x-litellm-response-duration-ms"),
             )
         )
         if _PROGRESS.drained:

tests/e2e/perf/setup/observe.py

@@ -16,6 +16,8 @@ from memory_readings import RssCapture, RssReading, WorkerKey
 from pydantic import BaseModel
 
 from setup.datastore import (
+    active_sessions,
+    slowest_statement,
     DbCounters,
     RedisCommand,
     RedisCounters,
@@ -25,6 +27,7 @@ from setup.datastore import (
 )
 from setup.mocks.reply_common import Provider
 from setup.performance_test_case import Limit, Metric, MockLatency, PerformanceTestCase
+from setup.telemetry import GatewayTelemetry
 from setup.wire import WireRecord
 
 
@@ -115,6 +118,16 @@ class Observation:
     redis_after: RedisCounters | None = None
     redis_commands: tuple[RedisCommand, ...] | None = None
     gateway_cpu_seconds: float | None = None
+    telemetry_before: GatewayTelemetry | None = None
+    telemetry_after: GatewayTelemetry | None = None
+    peak_connections: int | None = None
+
+    @property
+    def telemetry(self) -> GatewayTelemetry | None:
+        """What the gateway's `/metrics` counted during the run."""
+        if self.telemetry_before is None or self.telemetry_after is None:
+            return None
+        return self.telemetry_after.minus(self.telemetry_before)
 
     def per_request(self, before: int | None, after: int | None) -> float | None:
         if before is None or after is None or not self.records:
@@ -182,6 +195,18 @@ class Observation:
         ]
         return non_stream + stream
 
+    def _first_attempts(self) -> list[WireRecord]:
+        return [record for record, _ in self._clean(stream=False)] + [record for record, _ in self._clean(stream=True)]
+
+    def litellm_overhead_seconds(self) -> list[float]:
+        """`x-litellm-overhead-duration-ms` per first-attempt request, the gateway's own account of what it added
+        (receive time to response end, minus the provider call), in seconds to sit next to `gateway_added_seconds`."""
+        return [r.litellm_overhead_ms / 1000 for r in self._first_attempts() if r.litellm_overhead_ms is not None]
+
+    def litellm_response_seconds(self) -> list[float]:
+        """`x-litellm-response-duration-ms` per first-attempt request, the gateway's own end-to-end time."""
+        return [r.litellm_response_ms / 1000 for r in self._first_attempts() if r.litellm_response_ms is not None]
+
     def time_to_first_token_seconds(self) -> list[float]:
         """Seconds from the client sending each first-attempt streamed request to its first token, mock delay
         included, comparable across versions because the mocks generate the same replies after the same wait."""
@@ -217,11 +242,23 @@ def observe(
     per_request: dict[str, list[float]] = {
         "gateway_added_seconds": observation.gateway_added_seconds(),
         "time_to_first_token_seconds": observation.time_to_first_token_seconds(),
+        "litellm_overhead_seconds": observation.litellm_overhead_seconds(),
+        "litellm_response_seconds": observation.litellm_response_seconds(),
     }
     if metric in per_request:
         return percentile(per_request[metric], 50 if percentile_rank is None else percentile_rank)
     if metric == "unexpected_status_count":
         return float(observation.unexpected_status_count())
+    if metric == "prometheus_redis_latency_mean_seconds":
+        counted: Final = observation.telemetry
+        return None if counted is None else counted.redis_latency.mean
+    if metric == "postgres_active_sessions":
+        return active_sessions(observation.statements, observation.wall_seconds)
+    if metric == "postgres_slowest_statement_seconds":
+        slowest: Final = slowest_statement(observation.statements)
+        return None if slowest is None else slowest.mean_exec_seconds
+    if metric == "postgres_peak_connections":
+        return None if observation.peak_connections is None else float(observation.peak_connections)
     if metric == "postgres_calls_per_request":
         return observation.count_per_request(calls_in("awaited", observation.statements))
     if metric == "postgres_background_calls_per_request":
@@ -275,6 +312,14 @@ REPORTED: Final[tuple[tuple[Metric, int | None], ...]] = (
     ("unexpected_status_count", None),
     ("gateway_added_seconds", 50),
     ("gateway_added_seconds", 99),
+    ("litellm_overhead_seconds", 50),
+    ("litellm_overhead_seconds", 99),
+    ("litellm_response_seconds", 50),
+    ("litellm_response_seconds", 99),
+    ("prometheus_redis_latency_mean_seconds", None),
+    ("postgres_active_sessions", None),
+    ("postgres_slowest_statement_seconds", None),
+    ("postgres_peak_connections", None),
     ("time_to_first_token_seconds", 50),
     ("time_to_first_token_seconds", 99),
     ("memory_growth_megabytes", None),
@@ -325,6 +370,9 @@ def _detail(metric: Metric, observation: Observation, exempt_tables: frozenset[s
     if metric == "postgres_max_background_statement_per_request":
         busiest_background: Final = observation.busiest_background_statement(exempt_tables)
         return "" if busiest_background is None else busiest_background.summary
+    if metric == "postgres_slowest_statement_seconds":
+        slowest: Final = slowest_statement(observation.statements)
+        return "" if slowest is None else slowest.summary
     return ""
 
 

tests/e2e/perf/setup/performance_test_case.py

@@ -30,6 +30,12 @@ Metric = Literal[
     "cpu_seconds_per_request",
     "redis_calls_per_request",
     "redis_background_calls_per_request",
+    "litellm_overhead_seconds",
+    "litellm_response_seconds",
+    "prometheus_redis_latency_mean_seconds",
+    "postgres_active_sessions",
+    "postgres_slowest_statement_seconds",
+    "postgres_peak_connections",
 ]
 
 

tests/e2e/perf/setup/telemetry.py

@@ -0,0 +1,65 @@
+"""The gateway's own `/metrics`, read before and after a run for the numbers nothing outside the gateway
+can see. Today that is `litellm_redis_latency`, the histogram customers' Grafana panels show when Redis
+slows the gateway down. The gateway's latency and request counts are read per request from its response
+headers instead (`setup/wire.py`), which gives percentiles where a histogram delta only gives a mean.
+
+RSS and CPU are not here: a multi-worker gateway serves `/metrics` through the prometheus multiprocess
+collector, which drops `process_resident_memory_bytes` and `process_cpu_seconds_total`, so those stay
+on psutil (`proxy_usage.py`)."""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Iterable
+from dataclasses import dataclass
+from itertools import chain
+from typing import Final
+
+from e2e_http import NoBody
+from prometheus_client.parser import text_string_to_metric_families
+from prometheus_client.samples import Sample
+from proxy_client import ProxyClient
+
+
+@dataclass(frozen=True, slots=True)
+class HistogramTotals:
+    """`_sum` and `_count` of one histogram across every label set, enough for a mean over a window."""
+
+    total: float
+    count: float
+
+    def minus(self, earlier: HistogramTotals) -> HistogramTotals:
+        return HistogramTotals(total=self.total - earlier.total, count=self.count - earlier.count)
+
+    @property
+    def mean(self) -> float | None:
+        return self.total / self.count if self.count > 0 else None
+
+
+@dataclass(frozen=True, slots=True)
+class GatewayTelemetry:
+    redis_latency: HistogramTotals
+    """`litellm_redis_latency`, seconds per Redis call the gateway made."""
+
+    def minus(self, earlier: GatewayTelemetry) -> GatewayTelemetry:
+        return GatewayTelemetry(redis_latency=self.redis_latency.minus(earlier.redis_latency))
+
+
+def _value(samples: Iterable[Sample], name: str) -> float:
+    return sum(sample.value for sample in samples if sample.name == name and math.isfinite(sample.value))
+
+
+def _histogram(samples: tuple[Sample, ...], family: str) -> HistogramTotals:
+    return HistogramTotals(total=_value(samples, f"{family}_sum"), count=_value(samples, f"{family}_count"))
+
+
+def parse_gateway_telemetry(text: str) -> GatewayTelemetry:
+    samples: Final = tuple(chain.from_iterable(family.samples for family in text_string_to_metric_families(text)))
+    return GatewayTelemetry(redis_latency=_histogram(samples, "litellm_redis_latency"))
+
+
+def read_gateway_telemetry(proxy: ProxyClient) -> GatewayTelemetry | None:
+    """One `/metrics` scrape, or None when the gateway does not serve it (no `prometheus` callback), in which
+    case every limit on a gateway-reported metric fails as missing rather than reading a silent zero."""
+    scrape: Final = proxy.probe("/metrics", params=NoBody())
+    return parse_gateway_telemetry(scrape.body) if scrape.status_code == 200 else None

tests/e2e/perf/setup/wire.py

@@ -72,6 +72,8 @@ class WireRecord(BaseModel):
     call_id: str | None
     error: str | None
     response_id: str | None = None
+    litellm_overhead_ms: float | None = None
+    litellm_response_ms: float | None = None
 
 
 def response_id(payload: Mapping[str, object]) -> str | None:

tests/e2e/perf/test_perf_cases.py

@@ -18,7 +18,7 @@ from lifecycle import ResourceManager
 from memory_readings import read_rss_everywhere
 from models import SpendLogsParams
 from proxy_client import ProxyClient
-from setup.datastore import read_db_counters, read_redis_counters, redis_monitor
+from setup.datastore import connections_sampler, read_db_counters, read_redis_counters, redis_monitor
 from setup.event_loop import (
     LOG_FILE_ENV,
     OPT_IN_ENV,
@@ -44,6 +44,7 @@ from setup.plan import PlannedRequest, build_plan
 from setup.podcpu import gateway_cpu_sampler
 from setup.provision import MOCK_PROVIDERS, build_setup, guardrail_url, mock_url
 from setup.run import run_plan
+from setup.telemetry import read_gateway_telemetry
 
 pytestmark = [pytest.mark.perf, pytest.mark.quiet_stack]
 _SPEND_SAMPLE: Final = 40
@@ -108,16 +109,18 @@ def test_perf_case(
     _ = run_plan(case, _warmup_plan(case, plan), setup)
     _arm_mocks(mock_urls, case.mock_latency)
     rss_before: Final = read_rss_everywhere(proxy)
+    telemetry_before: Final = read_gateway_telemetry(proxy)
     db_before: Final = read_db_counters()
     redis_before: Final = read_redis_counters()
     run_started: Final = time.monotonic()
     traffic_started_at: Final = datetime.now(tz=UTC)
-    with gateway_cpu_sampler() as cpu, redis_monitor() as monitor:
+    with gateway_cpu_sampler() as cpu, redis_monitor() as monitor, connections_sampler() as connections:
         outcome: Final = run_plan(case, plan, setup)
     run_finished: Final = time.monotonic()
     db_after: Final = read_db_counters()
     redis_after: Final = read_redis_counters()
     rss_after: Final = read_rss_everywhere(proxy)
+    telemetry_after: Final = read_gateway_telemetry(proxy)
     rss_read: Final = time.monotonic()
     assert outcome.records, f"locust wrote no records (exit {outcome.locust_exit}):\n{outcome.locust_output}"
     served_call_ids: Final = tuple(
@@ -144,6 +147,9 @@ def test_perf_case(
         redis_after=redis_after,
         redis_commands=monitor.commands,
         gateway_cpu_seconds=cpu.seconds,
+        telemetry_before=telemetry_before,
+        telemetry_after=telemetry_after,
+        peak_connections=connections.peak,
     )
     rows, exceeded_limits = evaluate(case, observation)
     stall_rows: Final = _event_loop_rows(case, traffic_started_at) if case.event_loop is not None else ()

tests/e2e_harness/perf/setup/test_observe.py

@@ -36,7 +36,12 @@ def served(seq: int, total_s: float) -> WireRecord:
 
 
 def _key_lookup(calls: int) -> StatementCalls:
-    return StatementCalls(queryid=1, query='SELECT * FROM "LiteLLM_VerificationToken" WHERE token = $1', calls=calls)
+    return StatementCalls(
+        queryid=1,
+        query='SELECT * FROM "LiteLLM_VerificationToken" WHERE token = $1',
+        calls=calls,
+        total_exec_ms=calls * 2.0,
+    )
 
 
 def test_junit_properties_carry_each_observed_metric_and_each_limit_including_unobserved_ones() -> None:
@@ -71,6 +76,8 @@ def test_junit_properties_carry_each_observed_metric_and_each_limit_including_un
         "perf_postgres_background_calls_per_request": 0.0,
         "perf_postgres_max_statement_per_request": 10.0,
         "perf_postgres_max_background_statement_per_request": 0.0,
+        "perf_postgres_active_sessions": 0.02,
+        "perf_postgres_slowest_statement_seconds": 0.002,
         "perf_redis_calls_per_request_limit": 35.0,
     }