diff --git a/infra/grafana/dashboards/llm.json b/infra/grafana/dashboards/llm.json new file mode 100644 index 0000000..b96c971 --- /dev/null +++ b/infra/grafana/dashboards/llm.json @@ -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\"}" + } + ] + } + ] +} diff --git a/src/llm/index.py b/src/llm/index.py index c704615..45bf64e 100644 --- a/src/llm/index.py +++ b/src/llm/index.py @@ -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 diff --git a/src/llm/metrics.py b/src/llm/metrics.py new file mode 100644 index 0000000..bf3036d --- /dev/null +++ b/src/llm/metrics.py @@ -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}) diff --git a/src/llm/pool.py b/src/llm/pool.py index 69dd719..ff34ed6 100644 --- a/src/llm/pool.py +++ b/src/llm/pool.py @@ -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: diff --git a/src/llm/rag.py b/src/llm/rag.py index ea64108..283e397 100644 --- a/src/llm/rag.py +++ b/src/llm/rag.py @@ -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, diff --git a/tests/llm/test_metrics.py b/tests/llm/test_metrics.py new file mode 100644 index 0000000..3a4c7c2 --- /dev/null +++ b/tests/llm/test_metrics.py @@ -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) diff --git a/tests/test_observability.py b/tests/test_observability.py index 7afa323..f91a976 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -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