homelab/stack
1
0
Fork 0
Code Issues 53 Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat(llm,infra): OTel metrics for dispatch/embeddings/chat/indexing + LLM Grafana dashboard (refs #579)

Browse Source 
New `llm.metrics`: one lazily-built instrument set off `perf.meter("stack.llm")`,
so every call site is a no-op mock when telemetry is disabled (as it is in CI and,
today, in the container — see below). A build lock guards first use because
`embed_texts` fans batches across a thread pool.

Instruments, all `stack_llm_…` per the P26 convention:

  stack_llm_dispatch_total{host,kind}         pool slots handed out
  stack_llm_inflight{host}                    slots held right now
  stack_llm_dispatch_seconds{host,kind}       how long a slot was held
  stack_llm_embedded_texts_total{host}        texts embedded
  stack_llm_embed_batch_seconds{host}         one /api/embed round trip
  stack_llm_chat_seconds{stage}               retrieve | generate
  stack_llm_chat_total{outcome}               ok | error
  stack_llm_chat_tokens_total{direction}      prompt | completion (Ollama's
                                              final frame; skipped when absent)
  stack_llm_indexed_chunks_total{collection}
  stack_llm_indexed_items_total{collection,outcome}   indexed | skipped

Attribute values are closed sets — host, kind, stage, outcome, collection — never
item keys, dockets or question text, so the series count stays bounded.

Call sites: `HostPool.acquire`/`acquire_generation` wrap the held slot (kind embed
vs generate); `embed_texts` times each batch around the POST. `stream_answer` is
now a thin wrapper that counts the outcome once (ok, or error and re-raise) around
`_stream_answer`, which records the retrieve/generate stage split and the token
counts. The indexer counts one item at each of its five existing decision points;
no behaviour changed — every added line is a metric call.

The #575 tagging chain does not exist yet, so tag throughput and abstain rate from
the issue's scope are deliberately left out; they belong with that module.

Dashboard `infra/grafana/dashboards/llm.json` (uid `llm`, "LLM service",
schemaVersion 39): request rate by status filtered to `service_name="llm"`, chat
outcomes, dispatch rate and in-flight per host, p50/p95 dispatch latency by
host+kind, chat latency by stage, embedding throughput, chat tokens, indexed
chunks and items per collection, plus a Loki tail.

Export path — the metrics do NOT reach Prometheus yet, and this commit does not
change that. `perf._meter.setup_meter_provider` installs only a
`PrometheusMetricReader` (an in-process prometheus_client registry); the
`OTEL_EXPORTER_OTLP_ENDPOINT` the llm container sets is read by `_tracer.py` for
spans only, so nothing is pushed to otel-collector:4317 and nothing is exposed for
a scrape (`llm` has no /metrics route and no target in
infra/prometheus/targets/services.yml). On top of that `[telemetry] enabled` is
false in stack.toml and the llm service sets no STACK_TELEMETRY, so `perf.meter`
hands back the no-op mock today. Making #579's "metrics scraped" true needs a
follow-up that (a) enables telemetry for llm, (b) gives metrics a real exporter,
and (c) accounts for the collector's `namespace: stack`, which would re-prefix
OTLP-delivered names to `stack_stack_llm_…`. Filed as a note on #579.

Tests: tests/llm/test_metrics.py patches `perf.meter` with a recording fake and
asserts every instrument name, kind and attribute set, drives the real pool, chat
and indexer paths through it, and checks the helpers stay inert against the real
`MockMeter`; tests/test_observability.py gains the llm dashboard to
EXPECTED_DASHBOARDS plus panel/datasource/quantile assertions.
This commit is contained in:
kert
2026-09-11 18:54:27 -04:00
parent b9349c2958
commit 60018f47e9
7 changed files with 932 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

256
infra/grafana/dashboards/llm.json Normal file
View File

@@ -0,0 +1,256 @@
{
"title": "LLM service",
"uid": "llm",
"tags": [
"homelab",
"stack"
],
"timezone": "browser",
"schemaVersion": 39,
"time": {
"from": "now-6h",
"to": "now"
},
"refresh": "30s",
"panels": [
{
"id": 1,
"title": "HTTP request rate (status)",
"type": "timeseries",
"gridPos": {
"x": 0,
"y": 0,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (status) (rate(stack_http_server_requests_total{service_name=\"llm\"}[1m]))",
"legendFormat": "{{status}}"
}
]
},
{
"id": 2,
"title": "Chat answers by outcome",
"type": "timeseries",
"gridPos": {
"x": 12,
"y": 0,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (outcome) (rate(stack_llm_chat_total[5m]))",
"legendFormat": "{{outcome}}"
}
]
},
{
"id": 3,
"title": "Dispatch rate per host",
"type": "timeseries",
"gridPos": {
"x": 0,
"y": 8,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (host, kind) (rate(stack_llm_dispatch_total[5m]))",
"legendFormat": "{{host}} {{kind}}"
}
]
},
{
"id": 4,
"title": "In-flight requests per host",
"type": "timeseries",
"gridPos": {
"x": 12,
"y": 8,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (host) (stack_llm_inflight)",
"legendFormat": "{{host}}"
}
]
},
{
"id": 5,
"title": "Dispatch latency p50/p95 (host, kind)",
"type": "timeseries",
"gridPos": {
"x": 0,
"y": 16,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"fieldConfig": {
"defaults": {
"unit": "s"
},
"overrides": []
},
"targets": [
{
"expr": "histogram_quantile(0.50, sum by (le, host, kind) (rate(stack_llm_dispatch_seconds_bucket[5m])))",
"legendFormat": "p50 {{host}} {{kind}}"
},
{
"expr": "histogram_quantile(0.95, sum by (le, host, kind) (rate(stack_llm_dispatch_seconds_bucket[5m])))",
"legendFormat": "p95 {{host}} {{kind}}"
}
]
},
{
"id": 6,
"title": "Chat latency by stage (p50/p95)",
"type": "timeseries",
"gridPos": {
"x": 12,
"y": 16,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"fieldConfig": {
"defaults": {
"unit": "s"
},
"overrides": []
},
"targets": [
{
"expr": "histogram_quantile(0.50, sum by (le, stage) (rate(stack_llm_chat_seconds_bucket[5m])))",
"legendFormat": "p50 {{stage}}"
},
{
"expr": "histogram_quantile(0.95, sum by (le, stage) (rate(stack_llm_chat_seconds_bucket[5m])))",
"legendFormat": "p95 {{stage}}"
}
]
},
{
"id": 7,
"title": "Embedding throughput per host (texts/s)",
"type": "timeseries",
"gridPos": {
"x": 0,
"y": 24,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (host) (rate(stack_llm_embedded_texts_total[5m]))",
"legendFormat": "{{host}}"
}
]
},
{
"id": 8,
"title": "Chat tokens by direction",
"type": "timeseries",
"gridPos": {
"x": 12,
"y": 24,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (direction) (rate(stack_llm_chat_tokens_total[5m]))",
"legendFormat": "{{direction}}"
}
]
},
{
"id": 9,
"title": "Indexed chunks per collection",
"type": "timeseries",
"gridPos": {
"x": 0,
"y": 32,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (collection) (rate(stack_llm_indexed_chunks_total[5m]))",
"legendFormat": "{{collection}}"
}
]
},
{
"id": 10,
"title": "Indexed items (indexed vs skipped)",
"type": "timeseries",
"gridPos": {
"x": 12,
"y": 32,
"w": 12,
"h": 8
},
"datasource": {
"uid": "prometheus"
},
"targets": [
{
"expr": "sum by (collection, outcome) (rate(stack_llm_indexed_items_total[5m]))",
"legendFormat": "{{collection}} {{outcome}}"
}
]
},
{
"id": 11,
"title": "LLM service log tail",
"type": "logs",
"gridPos": {
"x": 0,
"y": 40,
"w": 24,
"h": 8
},
"datasource": {
"uid": "loki"
},
"targets": [
{
"expr": "{container=\"llm\"}"
}
]
}
]
}

6
src/llm/index.py
View File

@@ -27,6 +27,7 @@ from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine
from sqlalchemy.exc import ProgrammingError
from llm import metrics
from llm.chunk import Doc, chunk_doc, content_hash
from llm.config import LlmConfig, pg_url
from llm.migrate import ensure_hnsw, migrate
@@ -198,6 +199,7 @@ def index_refs(
def _unindexable(ref: DocRef) -> None:
stats["skipped"] += 1
metrics.indexed_item(collection, "skipped")
if ref.docket:
unindexable[ref.docket] = unindexable.get(ref.docket, 0) + 1
@@ -207,11 +209,13 @@ def index_refs(
prev_hash, prev_fp = seen.get(ref.key, ("", ""))
if not force and ref.fingerprint and ref.fingerprint == prev_fp:
stats["fingerprint_skipped"] += 1
metrics.indexed_item(collection, "skipped")
continue
doc = ref.load()
if doc is None or not doc.text.strip():
_record_state(engine, ref.key, collection, "", 0, ref.fingerprint)
stats["skipped"] += 1
metrics.indexed_item(collection, "skipped")
continue
h = content_hash(doc.text)
if not force and prev_hash == h:
@@ -224,6 +228,7 @@ def index_refs(
{"f": ref.fingerprint, "k": ref.key, "c": collection},
)
stats["hash_skipped"] += 1
metrics.indexed_item(collection, "skipped")
continue
chunks = enrich_pdf_pages(doc, chunk_doc(doc, code_index=code_index))
if not chunks:
@@ -240,6 +245,7 @@ def index_refs(
_record_state(engine, ref.key, collection, h, len(chunks), ref.fingerprint)
stats["indexed"] += 1
stats["chunks"] += len(chunks)
metrics.indexed_item(collection, "indexed", len(chunks))
# needs 100+ docs in one run to hit this line
if stats["indexed"] % 100 == 0:
log.info("indexed %(indexed)s (+%(chunks)s)", stats) # pragma: no cover

174
src/llm/metrics.py Normal file
View File

@@ -0,0 +1,174 @@
"""OpenTelemetry metric instruments for the llm service (#579).
Follows the P26 pipeline-telemetry convention: ``perf.meter`` hands back
a real OTel meter when telemetry is enabled and a zero-overhead
``MockMeter`` when it is not, so every call site here is free in tests
and in a container started without ``STACK_TELEMETRY``.
Instruments (all exported as ``stack_llm_…``)::
stack_llm_dispatch_total{host,kind} host slots handed out
stack_llm_inflight{host} slots held right now
stack_llm_dispatch_seconds{host,kind} how long a slot was held
stack_llm_embedded_texts_total{host} texts embedded
stack_llm_embed_batch_seconds{host} one /api/embed round trip
stack_llm_chat_seconds{stage} retrieve / generate
stack_llm_chat_total{outcome} ok / error
stack_llm_chat_tokens_total{direction} prompt / completion
stack_llm_indexed_chunks_total{collection} chunks written to pgvector
stack_llm_indexed_items_total{collection,outcome}
Attribute values are deliberately closed sets — host, kind, stage,
outcome, collection — never item keys, dockets or question text, so the
time series count stays bounded.
The tagging chain (#575) does not exist yet, so tag throughput and
abstain rate are not instrumented here; they belong with that module.
"""
from __future__ import annotations
import threading
import time
from contextlib import contextmanager
from typing import Any, Iterator
#: Instrumentation scope, mirroring ``perf.collector``'s "stack.pipeline".
METER_NAME = "stack.llm"
_instruments: "_Instruments | None" = None
#: ``embed_texts`` fans batches across a thread pool, so first use can be
#: concurrent — build the instrument set exactly once all the same.
_build_lock = threading.Lock()
class _Instruments:
"""The service's instrument set, built once on first use."""
def __init__(self) -> None:
from perf import meter
m: Any = meter(METER_NAME)
self.dispatch_total = m.create_counter(
"stack_llm_dispatch_total",
unit="1",
description="Ollama host slots handed out by the pool",
)
self.inflight = m.create_up_down_counter(
"stack_llm_inflight",
unit="1",
description="Requests currently holding an Ollama host slot",
)
self.dispatch_seconds = m.create_histogram(
"stack_llm_dispatch_seconds",
unit="s",
description="Time a request held an Ollama host slot",
)
self.embedded_texts_total = m.create_counter(
"stack_llm_embedded_texts_total",
unit="1",
description="Texts embedded through the host pool",
)
self.embed_batch_seconds = m.create_histogram(
"stack_llm_embed_batch_seconds",
unit="s",
description="Latency of one /api/embed batch",
)
self.chat_seconds = m.create_histogram(
"stack_llm_chat_seconds",
unit="s",
description="Chat answer latency by stage",
)
self.chat_total = m.create_counter(
"stack_llm_chat_total",
unit="1",
description="Chat answers by outcome",
)
self.chat_tokens_total = m.create_counter(
"stack_llm_chat_tokens_total",
unit="1",
description="Chat tokens reported by Ollama, by direction",
)
self.indexed_chunks_total = m.create_counter(
"stack_llm_indexed_chunks_total",
unit="1",
description="Chunks written to pgvector by the indexer",
)
self.indexed_items_total = m.create_counter(
"stack_llm_indexed_items_total",
unit="1",
description="Items the indexer embedded or skipped",
)
def instruments() -> _Instruments:
"""The cached instrument set (created on first call)."""
global _instruments
if _instruments is None:
with _build_lock:
if _instruments is None:
_instruments = _Instruments()
return _instruments
def reset() -> None:
"""Drop the cached instruments so the next call rebuilds them.
Only tests need this — they patch ``perf.meter`` and then want the
instruments built against the patched meter.
"""
global _instruments
_instruments = None
@contextmanager
def dispatch(host: str, kind: str) -> Iterator[None]:
"""Count and time one held pool slot on *host* (``embed``/``generate``)."""
i = instruments()
slot = {"host": host, "kind": kind}
i.dispatch_total.add(1, slot)
i.inflight.add(1, {"host": host})
started = time.monotonic()
try:
yield
finally:
i.inflight.add(-1, {"host": host})
i.dispatch_seconds.record(time.monotonic() - started, slot)
def embedded(host: str, texts: int, seconds: float) -> None:
"""Record one finished embed batch of *texts* items on *host*."""
i = instruments()
i.embedded_texts_total.add(texts, {"host": host})
i.embed_batch_seconds.record(seconds, {"host": host})
def chat_stage(stage: str, seconds: float) -> None:
"""Record the duration of a chat *stage* (``retrieve``/``generate``)."""
instruments().chat_seconds.record(seconds, {"stage": stage})
def chat_outcome(outcome: str) -> None:
"""Count one finished chat answer (``ok``/``error``)."""
instruments().chat_total.add(1, {"outcome": outcome})
def chat_tokens(prompt: int, completion: int) -> None:
"""Count the tokens Ollama reported for one answer.
Either count is absent on a host/model that does not report it; a
zero is not recorded rather than being counted as a real zero.
"""
i = instruments()
if prompt:
i.chat_tokens_total.add(prompt, {"direction": "prompt"})
if completion:
i.chat_tokens_total.add(completion, {"direction": "completion"})
def indexed_item(collection: str, outcome: str, chunks: int = 0) -> None:
"""Count one item the indexer finished (``indexed``/``skipped``)."""
i = instruments()
i.indexed_items_total.add(1, {"collection": collection, "outcome": outcome})
if chunks:
i.indexed_chunks_total.add(chunks, {"collection": collection})

11
src/llm/pool.py
View File

@@ -14,6 +14,7 @@ stay least-loaded (``acquire``).
from __future__ import annotations
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from typing import Iterator, Mapping, Sequence
@@ -21,6 +22,8 @@ from typing import Iterator, Mapping, Sequence
import httpx
from langchain_core.embeddings import Embeddings
from llm import metrics
_TIMEOUT = httpx.Timeout(120.0, connect=5.0)
@@ -111,7 +114,8 @@ class HostPool:
host = min(self._in_flight, key=self._in_flight.__getitem__)
self._in_flight[host] += 1
try:
yield host
with metrics.dispatch(host, "embed"):
yield host
finally:
self._release(host)
@@ -129,7 +133,8 @@ class HostPool:
)
self._in_flight[host] += 1
try:
yield host
with metrics.dispatch(host, "generate"):
yield host
finally:
self._release(host)
@@ -162,11 +167,13 @@ def embed_texts(
def run(start: int, batch: list[str]) -> None:
with pool.acquire() as host, httpx.Client(timeout=_TIMEOUT) as client:
began = time.monotonic()
resp = client.post(
f"{host}/api/embed", json={"model": model, "input": batch}
)
resp.raise_for_status()
results[start] = resp.json()["embeddings"]
metrics.embedded(host, len(batch), time.monotonic() - began)
workers = max(1, min(len(pool.hosts) * 2, len(batches)))
with ThreadPoolExecutor(max_workers=workers) as ex:

33
src/llm/rag.py
View File

@@ -27,12 +27,14 @@ import logging
import os
import re
import threading
import time
from dataclasses import replace
from datetime import date
from typing import Iterator, Sequence
import httpx
from llm import metrics
from llm.config import LlmConfig
from llm.evidence import (
ValuationEvidence,
@@ -633,7 +635,29 @@ def stream_answer(
then one ``{"type":"sources", "sources": […], "model": …, "host":
…, "mode": "timeline"|"recent"}`` (the resolved mode) and a final
``{"type":"done"}``.
Instrumented (#579): ``stack_llm_chat_total{outcome}`` counts each
answer once, ``ok`` or ``error``; the retrieve/generate stage split
and the token counts are recorded by ``_stream_answer``.
"""
try:
yield from _stream_answer(question, cfg=cfg, pool=pool, since=since, mode=mode)
except Exception:
metrics.chat_outcome("error")
raise
metrics.chat_outcome("ok")
def _stream_answer(
question: str,
*,
cfg: LlmConfig,
pool: HostPool,
since: str = "",
mode: str = "auto",
) -> Iterator[dict]:
"""The answer stream itself; ``stream_answer`` wraps it."""
began = time.monotonic()
resolved_mode = _resolve_mode(question, mode)
# Detection first: a detected family's acronym conflicts (CCM = chronic
# care management here, cardiac contractility modulation in the CY2027
@@ -715,6 +739,7 @@ def stream_answer(
yield lineage.payload()
if evidence is not None:
yield evidence.payload()
metrics.chat_stage("retrieve", time.monotonic() - began)
pool.check(cfg.instruct_model)
messages = build_messages(
question,
@@ -727,6 +752,7 @@ def stream_answer(
n_manual=n_manual,
budget_chars=cfg.chat_num_ctx * 3,
)
began = time.monotonic()
with pool.acquire_generation() as host, httpx.Client(timeout=_TIMEOUT) as client:
model = pick_model(cfg, pool, host)
with client.stream(
@@ -752,7 +778,14 @@ def stream_answer(
if chunk:
yield {"type": "token", "text": chunk}
if data.get("done"):
# Ollama reports its token counts on the final frame;
# older hosts/models omit them (chat_tokens skips 0).
metrics.chat_tokens(
int(data.get("prompt_eval_count") or 0),
int(data.get("eval_count") or 0),
)
break
metrics.chat_stage("generate", time.monotonic() - began)
yield {
"type": "sources",
"sources": sources,

408
tests/llm/test_metrics.py Normal file
View File

@@ -0,0 +1,408 @@
"""llm.metrics — OTel instruments for dispatch, embeddings, chat, indexing.
Two halves: with ``perf.meter`` patched to a recording fake (so the
instrument names and attributes are asserted exactly), and against the
real ``perf`` no-op ``MockMeter`` that every test run gets with
telemetry disabled — the call sites must be inert, not merely quiet.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from llm import metrics
from llm.config import LlmConfig
from llm.pool import HostPool, embed_texts
CFG = LlmConfig(
ollama_hosts=("http://h1:11434",),
host_vram={"http://h1:11434": 24},
embed_model="embed",
instruct_model="chat",
instruct_model_large="big",
embed_dim=768,
pg_host="x",
pg_port=5432,
pg_db="llm",
pg_user="llm",
build_ann_index=False,
top_n=3,
)
H1 = "http://h1:11434"
H2 = "http://h2:11434"
class FakeInstrument:
"""Records every ``add``/``record`` call as (amount, attributes)."""
def __init__(self, name: str, kind: str) -> None:
self.name = name
self.kind = kind
self.calls: list[tuple[float, dict]] = []
def add(self, amount, attributes=None):
self.calls.append((amount, dict(attributes or {})))
def record(self, amount, attributes=None):
self.calls.append((amount, dict(attributes or {})))
class FakeMeter:
def __init__(self) -> None:
self.name = ""
self.instruments: dict[str, FakeInstrument] = {}
def _make(self, name, kind):
self.instruments[name] = FakeInstrument(name, kind)
return self.instruments[name]
def create_counter(self, name, **kw):
return self._make(name, "counter")
def create_histogram(self, name, **kw):
return self._make(name, "histogram")
def create_up_down_counter(self, name, **kw):
return self._make(name, "up_down_counter")
@pytest.fixture()
def fake_meter():
"""Build the instrument set against a recording meter."""
meter = FakeMeter()
def _meter(name):
meter.name = name
return meter
metrics.reset()
with patch("perf.meter", _meter):
metrics.instruments()
yield meter
metrics.reset()
@pytest.fixture(autouse=True)
def _reset_instruments():
"""No test may inherit another's cached instrument set."""
metrics.reset()
yield
metrics.reset()
def _calls(meter: FakeMeter, name: str) -> list[tuple[float, dict]]:
return meter.instruments[name].calls
class TestInstrumentSet:
def test_names_and_kinds(self, fake_meter):
assert {n: i.kind for n, i in fake_meter.instruments.items()} == {
"stack_llm_dispatch_total": "counter",
"stack_llm_inflight": "up_down_counter",
"stack_llm_dispatch_seconds": "histogram",
"stack_llm_embedded_texts_total": "counter",
"stack_llm_embed_batch_seconds": "histogram",
"stack_llm_chat_seconds": "histogram",
"stack_llm_chat_total": "counter",
"stack_llm_chat_tokens_total": "counter",
"stack_llm_indexed_chunks_total": "counter",
"stack_llm_indexed_items_total": "counter",
}
def test_meter_scope(self, fake_meter):
assert fake_meter.name == "stack.llm"
def test_instruments_built_once(self, fake_meter):
first = metrics.instruments()
assert metrics.instruments() is first
assert len(fake_meter.instruments) == 10
def test_reset_rebuilds(self, fake_meter):
first = metrics.instruments()
metrics.reset()
with patch("perf.meter", lambda name: fake_meter):
assert metrics.instruments() is not first
class TestHelpers:
def test_dispatch_counts_times_and_balances_inflight(self, fake_meter):
with metrics.dispatch(H1, "embed"):
held = list(_calls(fake_meter, "stack_llm_inflight"))
assert held == [(1, {"host": H1})]
assert _calls(fake_meter, "stack_llm_dispatch_total") == [
(1, {"host": H1, "kind": "embed"})
]
assert _calls(fake_meter, "stack_llm_inflight") == [
(1, {"host": H1}),
(-1, {"host": H1}),
]
(seconds, attrs) = _calls(fake_meter, "stack_llm_dispatch_seconds")[0]
assert seconds >= 0
assert attrs == {"host": H1, "kind": "embed"}
def test_dispatch_releases_inflight_on_error(self, fake_meter):
with pytest.raises(RuntimeError):
with metrics.dispatch(H2, "generate"):
raise RuntimeError("boom")
assert _calls(fake_meter, "stack_llm_inflight")[-1] == (-1, {"host": H2})
assert len(_calls(fake_meter, "stack_llm_dispatch_seconds")) == 1
def test_embedded(self, fake_meter):
metrics.embedded(H1, 64, 1.5)
assert _calls(fake_meter, "stack_llm_embedded_texts_total") == [
(64, {"host": H1})
]
assert _calls(fake_meter, "stack_llm_embed_batch_seconds") == [
(1.5, {"host": H1})
]
def test_chat_stage_and_outcome(self, fake_meter):
metrics.chat_stage("retrieve", 0.25)
metrics.chat_outcome("ok")
assert _calls(fake_meter, "stack_llm_chat_seconds") == [
(0.25, {"stage": "retrieve"})
]
assert _calls(fake_meter, "stack_llm_chat_total") == [(1, {"outcome": "ok"})]
def test_chat_tokens_both_directions(self, fake_meter):
metrics.chat_tokens(120, 40)
assert _calls(fake_meter, "stack_llm_chat_tokens_total") == [
(120, {"direction": "prompt"}),
(40, {"direction": "completion"}),
]
def test_chat_tokens_skips_zero(self, fake_meter):
metrics.chat_tokens(0, 0)
assert _calls(fake_meter, "stack_llm_chat_tokens_total") == []
def test_indexed_item_with_chunks(self, fake_meter):
metrics.indexed_item("comments", "indexed", 7)
assert _calls(fake_meter, "stack_llm_indexed_items_total") == [
(1, {"collection": "comments", "outcome": "indexed"})
]
assert _calls(fake_meter, "stack_llm_indexed_chunks_total") == [
(7, {"collection": "comments"})
]
def test_indexed_item_skipped_records_no_chunks(self, fake_meter):
metrics.indexed_item("rules", "skipped")
assert _calls(fake_meter, "stack_llm_indexed_items_total") == [
(1, {"collection": "rules", "outcome": "skipped"})
]
assert _calls(fake_meter, "stack_llm_indexed_chunks_total") == []
class TestPool:
def test_acquire_dispatches_as_embed(self, fake_meter):
with HostPool([H1]).acquire() as host:
assert host == H1
assert _calls(fake_meter, "stack_llm_dispatch_total") == [
(1, {"host": H1, "kind": "embed"})
]
def test_acquire_generation_dispatches_as_generate(self, fake_meter):
with HostPool([H1, H2], vram_gb={H2: 24.0}).acquire_generation() as host:
assert host == H2
assert _calls(fake_meter, "stack_llm_dispatch_total") == [
(1, {"host": H2, "kind": "generate"})
]
@patch("llm.pool.httpx.Client")
def test_embed_texts_counts_texts_per_host(self, MockClient, fake_meter):
resp = MagicMock()
resp.json.return_value = {"embeddings": [[0.1], [0.2]]}
MockClient.return_value.__enter__.return_value.post.return_value = resp
out = embed_texts(HostPool([H1]), "embed", ["a", "b"], batch_size=2)
assert out == [[0.1], [0.2]]
assert _calls(fake_meter, "stack_llm_embedded_texts_total") == [
(2, {"host": H1})
]
(seconds, attrs) = _calls(fake_meter, "stack_llm_embed_batch_seconds")[0]
assert seconds >= 0
assert attrs == {"host": H1}
def _pool_mock():
pool = MagicMock()
pool.acquire_generation.return_value.__enter__.return_value = H1
pool.vram.return_value = 24.0
pool.serves.return_value = True
return pool
def _stream(MockClient, lines):
client = MockClient.return_value.__enter__.return_value
resp = client.stream.return_value.__enter__.return_value
resp.iter_lines.return_value = iter(lines)
return resp
class TestChat:
@patch("llm.rag.valuation_evidence", return_value=None)
@patch("llm.rag.lineage_evidence", return_value=None)
@patch("llm.rag.httpx.Client")
@patch("llm.rag.retrieve", return_value=[])
def test_ok_records_stages_outcome_and_tokens(
self, _retrieve, MockClient, _lin, _ev, fake_meter
):
from llm.rag import stream_answer
_stream(
MockClient,
[
'{"message":{"content":"hi"},"done":false}',
'{"message":{"content":""},"done":true,'
'"prompt_eval_count":120,"eval_count":40}',
],
)
events = list(stream_answer("q", cfg=CFG, pool=_pool_mock()))
assert events[-1] == {"type": "done"}
assert [
a[1]["stage"] for a in _calls(fake_meter, "stack_llm_chat_seconds")
] == [
"retrieve",
"generate",
]
assert _calls(fake_meter, "stack_llm_chat_total") == [(1, {"outcome": "ok"})]
assert _calls(fake_meter, "stack_llm_chat_tokens_total") == [
(120, {"direction": "prompt"}),
(40, {"direction": "completion"}),
]
@patch("llm.rag.valuation_evidence", return_value=None)
@patch("llm.rag.lineage_evidence", return_value=None)
@patch("llm.rag.httpx.Client")
@patch("llm.rag.retrieve", return_value=[])
def test_missing_token_counts_are_not_recorded(
self, _retrieve, MockClient, _lin, _ev, fake_meter
):
from llm.rag import stream_answer
_stream(MockClient, ['{"message":{"content":""},"done":true}'])
list(stream_answer("q", cfg=CFG, pool=_pool_mock()))
assert _calls(fake_meter, "stack_llm_chat_tokens_total") == []
@patch("llm.rag.valuation_evidence", return_value=None)
@patch("llm.rag.lineage_evidence", return_value=None)
@patch("llm.rag.httpx.Client")
@patch("llm.rag.retrieve", return_value=[])
def test_error_outcome_and_no_generate_stage(
self, _retrieve, MockClient, _lin, _ev, fake_meter
):
from llm.rag import stream_answer
resp = _stream(MockClient, [])
resp.raise_for_status.side_effect = RuntimeError("ollama down")
with pytest.raises(RuntimeError, match="ollama down"):
list(stream_answer("q", cfg=CFG, pool=_pool_mock()))
assert _calls(fake_meter, "stack_llm_chat_total") == [(1, {"outcome": "error"})]
stages = [a[1]["stage"] for a in _calls(fake_meter, "stack_llm_chat_seconds")]
assert stages == ["retrieve"]
class TestIndexer:
def _run(self, refs, state_rows, fake_meter):
from llm.index import index_refs
engine = MagicMock()
conn = engine.begin.return_value.__enter__.return_value
def fake_execute(clause, *a, **k):
r = MagicMock()
sql = str(clause)
r.fetchall.return_value = state_rows if "FROM index_state" in sql else []
return r
conn.execute.side_effect = fake_execute
with (
patch("llm.index.migrate"),
patch("llm.index.vectorstore", return_value=MagicMock()),
patch("llm.index._code_index", return_value={}),
patch("llm.index.embed_texts", return_value=[[0.1]]),
patch("llm.index.enrich_pdf_pages", side_effect=lambda doc, chunks: chunks),
):
return index_refs(
refs,
collection="comments",
cfg=CFG,
pool=MagicMock(),
engine=engine,
)
def _ref(self, key, fingerprint, doc):
from llm.source import DocRef
return DocRef(
key=key,
collection="comments",
docket="D",
fingerprint=fingerprint,
load=lambda: doc,
)
def test_indexed_and_skipped_items(self, fake_meter):
from llm.chunk import Doc, content_hash
body = Doc(key="K1", text="Some body text.", metadata={"docket": "D"})
same = Doc(key="K2", text="Unchanged text.", metadata={"docket": "D"})
blank = Doc(key="K3", text=" ", metadata={"docket": "D"})
stats = self._run(
[
self._ref("K1", "fp-new", body),
self._ref("K2", "fp-changed", same),
self._ref("K3", "fp-blank", blank),
self._ref("K4", "fp-old", body),
],
[("K2", content_hash(same.text), "fp-stale"), ("K4", "h", "fp-old")],
fake_meter,
)
assert (stats["indexed"], stats["chunks"]) == (1, 1)
assert _calls(fake_meter, "stack_llm_indexed_items_total") == [
(1, {"collection": "comments", "outcome": "indexed"}),
(1, {"collection": "comments", "outcome": "skipped"}),
(1, {"collection": "comments", "outcome": "skipped"}),
(1, {"collection": "comments", "outcome": "skipped"}),
]
assert _calls(fake_meter, "stack_llm_indexed_chunks_total") == [
(1, {"collection": "comments"})
]
def test_unindexable_ref_counts_skipped(self, fake_meter):
from llm.chunk import Doc
doc = Doc(key="K9", text="text that chunks to nothing", metadata={})
with patch("llm.index.chunk_doc", return_value=[]):
stats = self._run([self._ref("K9", "fp", doc)], [], fake_meter)
assert stats["skipped"] == 1
assert _calls(fake_meter, "stack_llm_indexed_items_total") == [
(1, {"collection": "comments", "outcome": "skipped"})
]
class TestNoopMeter:
"""Telemetry is disabled in CI — the real ``perf`` mock must absorb
every call without the caller noticing."""
def test_every_helper_is_inert(self, monkeypatch):
monkeypatch.setenv("STACK_TELEMETRY", "false")
from perf import meter
from perf._meter import MockMeter, _MockInstrument
assert isinstance(meter(metrics.METER_NAME), MockMeter)
assert isinstance(metrics.instruments().dispatch_total, _MockInstrument)
with metrics.dispatch(H1, "embed"):
pass
metrics.embedded(H1, 3, 0.1)
metrics.chat_stage("generate", 0.2)
metrics.chat_outcome("ok")
metrics.chat_tokens(1, 2)
metrics.indexed_item("rules", "indexed", 4)

46
tests/test_observability.py
View File

@@ -271,6 +271,7 @@ EXPECTED_DASHBOARDS = {
"data-pipelines",
"ci",
"gpu-notebooks",
"llm",
}
@@ -285,3 +286,48 @@ class TestDashboards:
data = json.load(f)
assert "panels" in data, f"{name}: no panels[]"
assert data["panels"], f"{name}: empty panels[]"
class TestLlmDashboard:
"""#579: the LLM service dashboard — panels for the metrics
``llm.metrics`` emits, on the provisioned prometheus datasource."""
def _dash(self) -> dict:
return json.loads((ROOT / "infra/grafana/dashboards/llm.json").read_text())
def test_identity_matches_the_other_dashboards(self):
d = self._dash()
assert d["uid"] == "llm"
assert d["title"] == "LLM service"
assert d["schemaVersion"] == 39
assert set(d["tags"]) == {"homelab", "stack"}
def test_panels_cover_every_instrument(self):
exprs = " ".join(
t["expr"] for p in self._dash()["panels"] for t in p.get("targets", [])
)
for metric in (
"stack_http_server_requests_total",
"stack_llm_dispatch_total",
"stack_llm_inflight",
"stack_llm_dispatch_seconds_bucket",
"stack_llm_embedded_texts_total",
"stack_llm_chat_seconds_bucket",
"stack_llm_chat_total",
"stack_llm_chat_tokens_total",
"stack_llm_indexed_chunks_total",
"stack_llm_indexed_items_total",
):
assert metric in exprs, f"no panel queries {metric}"
def test_panel_ids_unique_and_datasources_provisioned(self):
panels = self._dash()["panels"]
assert len({p["id"] for p in panels}) == len(panels)
assert {p["datasource"]["uid"] for p in panels} <= {"prometheus", "loki"}
def test_latency_panels_use_quantiles(self):
exprs = [
t["expr"] for p in self._dash()["panels"] for t in p.get("targets", [])
]
quantiles = [e for e in exprs if e.startswith("histogram_quantile")]
assert len(quantiles) == 4 # p50 + p95 for dispatch and for chat