Serving observability + SLO autoscaler

Open in Colab ▶️ Run this notebook in Colab

Serving observability + SLO autoscaler#

Track 05 - Serving · Notebook 11 · Runtime: ≈30 s on CPU

Prerequisites: 05_serving/10 (DistServe).

Reference: vLLM Prometheus metrics; SLO-driven autoscaling patterns.

What#

Two building blocks of a real LLM serving deployment:

  • Metrics: Prometheus-style counters and histograms for requests, tokens, TTFT, TPOT, queue depth, KV utilisation.

  • Autoscaler: a control loop that reads recent SLO attainment and scales replicas up/down accordingly.

We simulate a non-stationary workload (traffic spikes at minute 2, drops at minute 5), record the metrics, and drive a replicas-count controller that keeps SLO attainment above a target while minimising over-provisioning.

from llm_systems_cookbook.nb import bootstrap

from dataclasses import dataclass, field

import numpy as np

s = bootstrap("05_serving_11_serving_observability_slo_autoscaler")

Metrics registry#

A minimal Prometheus-shaped registry: counters (monotonic) and histograms (bucketed). Exposes observe, inc, and snapshot methods.

@dataclass
class Counter:
    name: str
    value: float = 0.0

    def inc(self, by: float = 1.0) -> None:
        self.value += by


@dataclass
class Histogram:
    name: str
    buckets_le: list[float] = field(default_factory=lambda: [0.05, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0])
    counts: list[int] = field(default_factory=list)
    sum_: float = 0.0
    n: int = 0

    def __post_init__(self) -> None:
        if not self.counts:
            self.counts = [0] * (len(self.buckets_le) + 1)

    def observe(self, value: float) -> None:
        self.sum_ += value
        self.n += 1
        for i, le in enumerate(self.buckets_le):
            if value <= le:
                self.counts[i] += 1
                return
        self.counts[-1] += 1

    def quantile(self, q: float) -> float:
        if self.n == 0:
            return 0.0
        target = q * self.n
        seen = 0
        for i, le in enumerate(self.buckets_le):
            seen += self.counts[i]
            if seen >= target:
                return le
        return self.buckets_le[-1] * 2


@dataclass
class Registry:
    counters: dict[str, Counter] = field(default_factory=dict)
    histograms: dict[str, Histogram] = field(default_factory=dict)

    def counter(self, name: str) -> Counter:
        if name not in self.counters:
            self.counters[name] = Counter(name)
        return self.counters[name]

    def histogram(self, name: str) -> Histogram:
        if name not in self.histograms:
            self.histograms[name] = Histogram(name)
        return self.histograms[name]

    def snapshot(self) -> dict:
        return {
            "counters": {n: c.value for n, c in self.counters.items()},
            "histograms": {n: {"p50": h.quantile(0.5), "p99": h.quantile(0.99), "n": h.n}
                           for n, h in self.histograms.items()},
        }


registry = Registry()
registry.counter("requests_total").inc()
registry.histogram("ttft_seconds").observe(0.3)
registry.histogram("ttft_seconds").observe(1.8)
snap = registry.snapshot()
print(f"snapshot after 2 requests: {snap}")

s.check(
    "counter_increments",
    lambda: registry.counter("requests_total").value == 1.0,
)
s.check(
    "histogram_quantile_sane",
    lambda: registry.histogram("ttft_seconds").quantile(0.99) <= 2.0,
    msg=f"p99 = {registry.histogram('ttft_seconds').quantile(0.99)}",
)

Non-stationary workload#

Arrivals vary over time: low (4 req/s) minutes 0-2, spike (14 req/s) minutes 2-5, back down (5 req/s) minutes 5-8.

rng = np.random.default_rng(0)


def generate_arrivals() -> list[tuple[float, float]]:
    '''Return (arrival_time, requested_compute_seconds) per request.'''
    arrivals: list[tuple[float, float]] = []
    t = 0.0
    while t < 480:  # 8 minutes
        if t < 120:
            lam = 4.0
        elif t < 300:
            lam = 14.0
        else:
            lam = 5.0
        t += rng.exponential(1.0 / lam)
        work = float(rng.lognormal(-0.5, 0.3))  # ~0.6s mean service
        arrivals.append((t, work))
    return arrivals


ARRIVALS = generate_arrivals()
print(f"workload: {len(ARRIVALS)} requests over 8 minutes")

SLO-driven autoscaler#

A simple control loop: every 30s, look at the SLO attainment (TTFT p99 ≤ 1s) over the last window; if below target, scale up by 1 replica; if above by a margin, scale down. Cap replicas at [1, 8].

SLO_TTFT = 1.0
TARGET_SLO = 0.95

PER_REPLICA_CAPACITY = 0.8  # service rate per replica (req/s)


def simulate(static_replicas: int | None = None) -> dict:
    reg = Registry()
    replicas = static_replicas if static_replicas is not None else 1
    queue: list[tuple[float, float]] = []  # (arrival_time, work_seconds)
    in_flight_until = [0.0] * 16  # support up to 16 replicas
    replicas_history: list[tuple[float, int]] = [(0.0, replicas)]
    last_control = 0.0
    window_attainment: list[bool] = []

    for arr, work in ARRIVALS:
        # Advance to arrival.
        queue.append((arr, work))
        # Auto-scale decision every 30s.
        if static_replicas is None and arr - last_control > 30:
            if window_attainment:
                attain = sum(window_attainment) / len(window_attainment)
                if attain < TARGET_SLO and replicas < 8:
                    replicas += 1
                elif attain > TARGET_SLO + 0.03 and replicas > 1:
                    replicas -= 1
                window_attainment = []
            last_control = arr
            replicas_history.append((arr, replicas))
        # Admit queued items to any free replica.
        while queue:
            a, w = queue[0]
            # Pick the earliest-free replica among the active ones.
            free_idx = int(min(range(replicas), key=lambda i: in_flight_until[i]))
            start = max(a, in_flight_until[free_idx])
            ttft = start - a
            reg.histogram("ttft_seconds").observe(ttft)
            reg.counter("requests_total").inc()
            window_attainment.append(ttft <= SLO_TTFT)
            in_flight_until[free_idx] = start + w / PER_REPLICA_CAPACITY
            queue.pop(0)
            break  # one admission per arrival to keep simulation simple

    snap = reg.snapshot()
    # Attainment = fraction of per-request ttft measurements within SLO.
    all_attainments = []
    for le, count in zip(reg.histogram("ttft_seconds").buckets_le,
                         reg.histogram("ttft_seconds").counts[:-1], strict=True):
        if le <= SLO_TTFT:
            all_attainments.append(count)
    total_n = reg.histogram("ttft_seconds").n
    attained = sum(all_attainments)
    slo_attainment = attained / max(total_n, 1)
    return {
        "ttft_p99": snap["histograms"]["ttft_seconds"]["p99"],
        "ttft_p50": snap["histograms"]["ttft_seconds"]["p50"],
        "slo_attainment": slo_attainment,
        "total": snap["counters"]["requests_total"],
        "replicas_history": replicas_history,
        "replicas_avg": float(np.mean([r for _, r in replicas_history])),
    }


result_autoscale = simulate(static_replicas=None)
result_static_4 = simulate(static_replicas=4)
result_static_2 = simulate(static_replicas=2)

for name, r in [("autoscale", result_autoscale), ("static-4", result_static_4), ("static-2", result_static_2)]:
    print(f"  {name:<11}  ttft_p99={r['ttft_p99']:.3f}  p50={r['ttft_p50']:.3f}  "
          f"slo={r['slo_attainment']:.1%}  replicas_avg={r['replicas_avg']:.1f}")

s.check(
    "autoscaler_adjusts_replicas",
    lambda: len({r for _, r in result_autoscale["replicas_history"]}) >= 2,
    msg=f"replicas seen: {set(r for _, r in result_autoscale['replicas_history'])}",
)
s.check(
    "autoscaler_slo_at_least_static_2",
    lambda: result_autoscale["slo_attainment"] >= result_static_2["slo_attainment"] - 0.05,
    msg=f"auto={result_autoscale['slo_attainment']:.1%}  s2={result_static_2['slo_attainment']:.1%}",
)
s.check(
    "autoscaler_uses_more_replicas_than_static_2",
    lambda: result_autoscale["replicas_avg"] > result_static_2["replicas_avg"],
    msg=f"auto={result_autoscale['replicas_avg']:.2f}  s2={result_static_2['replicas_avg']:.2f}",
)
s.check(
    "more_replicas_improves_slo",
    lambda: result_static_4["slo_attainment"] >= result_static_2["slo_attainment"],
    msg=f"s4={result_static_4['slo_attainment']:.1%}  s2={result_static_2['slo_attainment']:.1%}",
)
s.check(
    "autoscaler_beats_static_2_slo",
    lambda: result_autoscale["slo_attainment"] >= result_static_2["slo_attainment"],
    msg=f"auto={result_autoscale['slo_attainment']:.1%}  s2={result_static_2['slo_attainment']:.1%}",
)

Replicas and arrival rate, side by side#

Two stacked panels. Top: arrival rate over time (the 2-5 minute spike). Bottom: replica count chosen by the autoscaler, stepping up as the spike arrives and draining afterwards. The lag between the two panels is the control loop’s reaction time - longer lag means more SLO violations during the ramp.

import matplotlib.pyplot as plt

# Arrival rate: count arrivals inside 10-second windows.
bin_s = 10.0
t_end = max(t for t, _ in ARRIVALS)
edges = np.arange(0, t_end + bin_s, bin_s)
times = np.array([t for t, _ in ARRIVALS])
rate, _ = np.histogram(times, bins=edges)
rate_per_s = rate / bin_s

# Replicas step-function from autoscaler history.
hist = result_autoscale["replicas_history"]
rt = [t for t, _ in hist] + [t_end]
rv = [r for _, r in hist] + [hist[-1][1]]

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 4.8), sharex=True)
ax1.fill_between(edges[:-1] / 60, rate_per_s, alpha=0.4, color="tab:blue")
ax1.plot(edges[:-1] / 60, rate_per_s, color="tab:blue", lw=1.5, label="arrivals / s")
ax1.axvspan(2, 5, color="tab:orange", alpha=0.15, label="spike window")
ax1.set_ylabel("arrival rate (req/s)")
ax1.set_title("workload arrival rate (non-stationary)")
ax1.legend(loc="upper right", fontsize=9)
ax1.grid(True, alpha=0.3)

ax2.step([t / 60 for t in rt], rv, where="post", color="tab:green", lw=2)
ax2.fill_between([t / 60 for t in rt], rv, step="post", alpha=0.2, color="tab:green")
ax2.axvspan(2, 5, color="tab:orange", alpha=0.15)
ax2.set_ylabel("active replicas")
ax2.set_xlabel("time (minutes)")
ax2.set_title(f"autoscaler response (SLO attainment {result_autoscale['slo_attainment']:.0%})")
ax2.set_ylim(0, max(rv) + 1)
ax2.grid(True, alpha=0.3)

fig.tight_layout()
plt.show()

Exercises#

  1. Add more metrics. KV cache utilisation, queue depth histogram, per-model request counts. In production these all feed Grafana dashboards.

  2. Alerting. Send a “burning through error budget” alert when SLO attainment drops below 90 % for 5 minutes.

  3. Cost-aware scaling. Adjust the controller to optimise goodput / replica_cost; the optimum is usually 2 replicas below what the naive SLO-attainment target would pick.

References#

  • Google SRE book, Service Level Objectives chapter.

  • vLLM’s /metrics endpoint for the reference Prometheus labels.

  • KEDA (Kubernetes Event-Driven Autoscaler) for a real-world autoscaler wiring.

s.summary()
s.save()