Merge pull request #492 from ggozad/feat/telemetry-improvements

Improve logfire telemetry

CHANGELOG.md

@@ -1,6 +1,11 @@
 # Changelog
 ## [Unreleased]
 
+### Changed
+
+- Logfire spans carry `service.version` and a per-process `service.name` (`haiku-ingester`, `haiku-rag`, `haiku-rag-app`); `OTEL_SERVICE_NAME` / `LOGFIRE_SERVICE_NAME` override the default.
+- Each docling-serve request emits a `docling_serve.request` span carrying the instance `url` and `attempt`.
+
 ## [0.65.0] - 2026-07-09
 
 ### Added

app/backend/main.py

@@ -20,6 +20,7 @@ from haiku.rag.client import HaikuRAG
 from haiku.rag.config import load_yaml_config
 from haiku.rag.config.models import AppConfig
 from haiku.rag.skills.rag import create_skill, get_agent_preamble
+from haiku.rag.telemetry import configure as configure_telemetry
 from haiku.rag.utils import get_model
 from haiku.skills import (
     SkillDeps,
@@ -30,14 +31,7 @@ from haiku.skills.prompts import build_system_prompt
 
 load_dotenv(find_dotenv(usecwd=True))
 
-# Configure logfire (only sends data if LOGFIRE_TOKEN is present)
-try:
-    import logfire
-
-    logfire.configure(send_to_logfire="if-token-present", console=False)
-    logfire.instrument_pydantic_ai()
-except Exception:
-    pass
+configure_telemetry(service_name="haiku-rag-app")
 
 logging.basicConfig(
     level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"

docs/ingester.md

@@ -640,7 +640,21 @@ INFO     Processing upsert file:///.../a.md (job 5d9a...)
 INFO     Job 5d9a... succeeded in 0.34s: file:///.../a.md
 ```
 
-When `LOGFIRE_TOKEN` is set, spans are also shipped to Logfire.
+When `LOGFIRE_TOKEN` is set, spans are also shipped to Logfire. Spans carry
+`service.name` (`haiku-ingester`) and `service.version`. To tell concurrent
+ingestions apart in Logfire, give each process a distinct name via the standard
+`OTEL_SERVICE_NAME` (or `LOGFIRE_SERVICE_NAME`) environment variable, which
+overrides the default:
+
+```bash
+OTEL_SERVICE_NAME=ingester-tenant-a haiku-ingester serve
+```
+
+The span tree is `ingester.poller.sweep` -> `ingester.job` (tagged with
+`source_id` and `uri`) -> `document.convert` / `document.chunk`. When a source
+uses docling-serve, each request emits a `docling_serve.request` span carrying
+the instance `url` and `attempt`, so a failed conversion can be traced to the
+exact instance that served it.
 
 ### Operating against the API
 

evaluations/evaluations/benchmark.py

@@ -4,7 +4,6 @@ from collections.abc import Awaitable, Callable, Mapping
 from pathlib import Path
 from typing import Any, Literal, cast
 
-import logfire
 import typer
 from dotenv import find_dotenv, load_dotenv
 from huggingface_hub import HfApi, snapshot_download
@@ -26,6 +25,7 @@ from haiku.rag.client import HaikuRAG
 from haiku.rag.config import AppConfig, find_config_file, load_yaml_config
 from haiku.rag.config.models import ModelConfig
 from haiku.rag.logging import configure_cli_logging
+from haiku.rag.telemetry import configure as configure_telemetry
 from haiku.rag.utils import get_model, parse_model_option
 
 Target = Literal["rag-skill", "analysis-skill"]
@@ -42,10 +42,7 @@ HF_REPO_ID = "ggozad/haiku-rag-eval-dbs"
 
 # Scrubbing off: eval outputs are financial answers with words like "authorized"
 # that trip Logfire's secret scrubber and redact the model's answer text.
-logfire.configure(
-    send_to_logfire="if-token-present", service_name="evals", scrubbing=False
-)
-logfire.instrument_pydantic_ai()
+configure_telemetry(service_name="evals", scrubbing=False)
 configure_cli_logging()
 console = Console()
 

haiku_rag_slim/README.md

@@ -52,7 +52,7 @@ Adds support for 40+ file formats including PDF, DOCX, HTML, and more.
 ```bash
 # Common combinations
 uv pip install haiku.rag-slim[docling,anthropic,mxbai]
-uv pip install haiku.rag-slim[docling,groq,logfire]
+uv pip install haiku.rag-slim[docling,groq]
 ```
 
 ## Usage

haiku_rag_slim/haiku/rag/chat/app.py

@@ -40,7 +40,7 @@ from haiku.skills.agent import (
 from haiku.skills.models import Skill
 from haiku.skills.prompts import build_system_prompt
 
-configure_telemetry()
+configure_telemetry(service_name="haiku-rag")
 
 if TYPE_CHECKING:
     from textual.app import ComposeResult

haiku_rag_slim/haiku/rag/cli.py

@@ -137,7 +137,9 @@ def main(
     from haiku.rag.telemetry import configure as configure_telemetry
 
     is_production = get_config().environment != "development"
-    configure_telemetry(console=False if is_production else None)
+    configure_telemetry(
+        service_name="haiku-rag", console=False if is_production else None
+    )
 
     if get_config().environment != "development":
         # Suppress warnings in production

haiku_rag_slim/haiku/rag/providers/docling_serve.py

@@ -11,6 +11,7 @@ import httpx
 
 from haiku.rag.circuit_breaker import CircuitBreaker
 from haiku.rag.config import CircuitBreakerConfig, DoclingServeConfig
+from haiku.rag.telemetry import logfire
 
 logger = logging.getLogger(__name__)
 
@@ -156,8 +157,14 @@ class DoclingServeClient:
             base_url = self._pick_url(exclude=frozenset(tried))
             breaker = self._breaker_for(base_url)
             try:
-                async with self._httpx_client() as client:
-                    result = await attempt(client, base_url)
+                with logfire.span(
+                    "docling_serve.request",
+                    name=name,
+                    url=base_url,
+                    attempt=attempt_no,
+                ):
+                    async with self._httpx_client() as client:
+                        result = await attempt(client, base_url)
             except Exception as exc:
                 if not _is_retryable(exc):
                     raise

haiku_rag_slim/haiku/rag/telemetry.py

@@ -1,3 +1,5 @@
+import os
+from importlib import metadata
 from typing import Literal
 
 from logfire import Logfire, attach_context, get_context
@@ -19,25 +21,45 @@ def configure(
     *,
     service_name: str | None = None,
     console: Literal[False] | None = False,
+    scrubbing: Literal[False] | None = None,
 ) -> None:
     """Configure Logfire and enable pydantic-ai instrumentation for the
     running process. Each CLI entry point calls this once at startup.
     Silently no-ops on failure so a missing/misconfigured LOGFIRE_TOKEN
     never crashes the app.
 
-    - service_name: identifies the process in the Logfire UI (e.g.
-      "haiku-ingester"). Falls back to logfire's default when None.
+    - service_name: the default name for this process in the Logfire UI
+      (e.g. "haiku-ingester"). The OTEL_SERVICE_NAME / LOGFIRE_SERVICE_NAME
+      env vars, when set, take precedence so operators can distinguish
+      concurrent processes.
     - console: False (default) suppresses span lines on stderr so they
       don't interleave with RichHandler logs. Pass None to let logfire
       decide (its own default applies).
+    - scrubbing: None (default) keeps logfire's secret scrubbing on. Pass
+      False to disable it when span content legitimately contains tokens
+      that trip the scrubber (e.g. eval answer text).
     """
     try:
         import logfire as _lf
 
+        # An explicit service_name arg would beat the env vars in logfire's
+        # precedence; deferring to None when an env var is set lets the
+        # operator's OTEL_SERVICE_NAME / LOGFIRE_SERVICE_NAME win over our
+        # per-process default.
+        env_service = os.environ.get("OTEL_SERVICE_NAME") or os.environ.get(
+            "LOGFIRE_SERVICE_NAME"
+        )
+        try:
+            service_version = metadata.version("haiku.rag-slim")
+        except metadata.PackageNotFoundError:  # pragma: no cover
+            service_version = None
+
         _lf.configure(
-            service_name=service_name,
+            service_name=None if env_service else service_name,
+            service_version=service_version,
             send_to_logfire="if-token-present",
             console=console,
+            scrubbing=scrubbing,
         )
         _lf.instrument_pydantic_ai()
     except Exception:  # pragma: no cover

tests/test_docling_serve_client.py

@@ -301,6 +301,36 @@ async def test_task_failure_is_not_retried():
     assert set(seen) == {"h"}
 
 
+@pytest.mark.asyncio
+async def test_request_span_records_instance_per_attempt(monkeypatch):
+    """Each attempt opens a docling_serve.request span tagged with the
+    instance URL, so failover is traceable in Logfire."""
+    from contextlib import nullcontext
+
+    from haiku.rag.providers import docling_serve as ds_module
+
+    spans: list[dict] = []
+
+    def _fake_span(span_name, /, **attrs):
+        spans.append({"span_name": span_name, **attrs})
+        return nullcontext()
+
+    monkeypatch.setattr(ds_module.logfire, "span", _fake_span)
+
+    transport, _ = _failover_transport({"down-s"}, "t", {"ok": True})
+    client = DoclingServeClient(
+        base_urls=["http://down-s:5001", "http://up-s:5001"],
+        transport=transport,
+        retry_base_delay=0.0,
+    )
+
+    await _poll(client)
+
+    requests = [s for s in spans if s["span_name"] == "docling_serve.request"]
+    assert [s["url"] for s in requests] == ["http://down-s:5001", "http://up-s:5001"]
+    assert [s["attempt"] for s in requests] == [0, 1]
+
+
 def test_pick_url_skips_excluded_instances():
     """On retry, _pick_url advances past every excluded instance."""
     urls = ["http://p1:5001", "http://p2:5001", "http://p3:5001"]

tests/test_telemetry.py

@@ -0,0 +1,70 @@
+from importlib import metadata
+
+import logfire
+import pytest
+
+from haiku.rag import telemetry
+
+
+@pytest.fixture
+def captured_configure(monkeypatch):
+    """Capture the kwargs telemetry.configure() passes to logfire.configure,
+    and no-op the instrumentation so tests don't touch a real exporter."""
+    captured: dict = {}
+
+    def _fake_configure(**kwargs):
+        captured.update(kwargs)
+
+    monkeypatch.setattr(logfire, "configure", _fake_configure)
+    monkeypatch.setattr(logfire, "instrument_pydantic_ai", lambda: None)
+    return captured
+
+
+def test_default_service_name_used_when_env_unset(captured_configure, monkeypatch):
+    monkeypatch.delenv("OTEL_SERVICE_NAME", raising=False)
+    monkeypatch.delenv("LOGFIRE_SERVICE_NAME", raising=False)
+
+    telemetry.configure(service_name="haiku-ingester")
+
+    assert captured_configure["service_name"] == "haiku-ingester"
+
+
+def test_otel_service_name_overrides_default(captured_configure, monkeypatch):
+    monkeypatch.setenv("OTEL_SERVICE_NAME", "customer-ingester")
+
+    telemetry.configure(service_name="haiku-ingester")
+
+    # Deferring to logfire (service_name=None) lets it read the env var,
+    # so the customer's OTEL_SERVICE_NAME wins over our default.
+    assert captured_configure["service_name"] is None
+
+
+def test_logfire_service_name_overrides_default(captured_configure, monkeypatch):
+    monkeypatch.delenv("OTEL_SERVICE_NAME", raising=False)
+    monkeypatch.setenv("LOGFIRE_SERVICE_NAME", "customer-ingester")
+
+    telemetry.configure(service_name="haiku-ingester")
+
+    assert captured_configure["service_name"] is None
+
+
+def test_service_version_is_package_version(captured_configure, monkeypatch):
+    monkeypatch.delenv("OTEL_SERVICE_NAME", raising=False)
+    monkeypatch.delenv("LOGFIRE_SERVICE_NAME", raising=False)
+
+    telemetry.configure(service_name="haiku-rag")
+
+    assert captured_configure["service_version"] == metadata.version("haiku.rag-slim")
+
+
+def test_scrubbing_defaults_to_enabled(captured_configure):
+    telemetry.configure(service_name="haiku-rag")
+
+    # None is logfire's "scrubbing enabled" default.
+    assert captured_configure["scrubbing"] is None
+
+
+def test_scrubbing_can_be_disabled(captured_configure):
+    telemetry.configure(service_name="evals", scrubbing=False)
+
+    assert captured_configure["scrubbing"] is False