fix(promsd): look lane tags up by name, not by a computed slug

README.md

@@ -10,6 +10,8 @@ Part of [dmfdeploy](https://github.com/dmfdeploy/dmfdeploy), an independent open
 
 The dmfdeploy platform keeps NetBox as its source of truth for inventory. `dmf-promsd` reads NetBox services, devices and virtual machines, and serves the ones that carry a monitoring tag as Prometheus [HTTP service discovery](https://prometheus.io/docs/prometheus/latest/http_sd/) target groups. Prometheus then scrapes or probes new workloads without anyone editing its configuration.
 
+The service looks each lane tag up in NetBox by name and filters by the slug NetBox returns, so a tag created in the NetBox UI works with whatever slug NetBox chose. Every lane tag must exist in NetBox and be visible to the token. If any lane tag is missing or hidden, the refresh fails, the service keeps serving the last good lanes, and `last_error` in `/healthz` names the tags it didn't see. Any error on a record query fails the refresh the same way. The service reports not ready once the kept lanes go stale.
+
 A record joins a lane when it has the matching tag in NetBox and the lane reads that record type:
 
 | Tag | Endpoint | Reads | Lane |
@@ -26,7 +28,7 @@ The service skips a record in a lane when it lacks a required field: `metrics_po
 
 ## Run it
 
-You need Python 3.12 or later, and a NetBox instance with an API token that can read services, devices and virtual machines.
+You need Python 3.12 or later, and a NetBox instance with an API token with view access to services, devices, virtual machines and all three lane tags (`extras | tag | view`). NetBox hides objects a token can't view, so promsd can't tell a hidden lane tag from a missing one. In either case the refresh fails, the service keeps the last good lanes, and `last_error` names the tags it didn't see and mentions `extras.view_tag`. The service reports not ready until a refresh succeeds, or once the last good lanes go stale.
 
 Supply the NetBox address and token through environment variables. The service refuses to start without `NETBOX_URL` and `NETBOX_TOKEN`. Set the token from your secret store at run time and keep it out of files in this repository.
 

src/dmf_promsd/cache.py

@@ -70,9 +70,19 @@ def refresh(self) -> bool:
         try:
             records: list[Record] = []
             source_counts: dict[str, int] = {}
+            slugs = self._client.tag_slugs(lane.tag for lane in LANES)
+            unseen = sorted({lane.tag for lane in LANES} - slugs.keys())
+            if unseen:
+                # NetBox hides objects the token may not view, so a hidden lane tag looks
+                # the same as a missing one. Publishing an empty lane would pass for "no
+                # records", so fail the refresh and keep the last snapshot.
+                raise RuntimeError(
+                    f"Lane tags not seen in NetBox: {', '.join(unseen)}. Each is either "
+                    "missing from NetBox or not visible to the token (extras.view_tag)."
+                )
             for kind in RECORD_KINDS:
-                tags = [lane.tag for lane in LANES if kind in lane.kinds]
-                kind_records, unreadable = self._client.list_records(kind, tags)
+                kind_slugs = [slugs[lane.tag] for lane in LANES if kind in lane.kinds]
+                kind_records, unreadable = self._client.list_records(kind, kind_slugs)
                 records.extend(kind_records)
                 source_counts[kind] = len(kind_records) + unreadable
             payloads, skipped = build_payloads(records)

src/dmf_promsd/netbox.py

@@ -7,11 +7,7 @@
 
 from .records import DEVICES, SERVICES, VIRTUAL_MACHINES, Record
 
-
-def _tag_slug(name: str) -> str:
-    # NetBox ?tag= filters by tag SLUG (colon -> hyphen), not the name.
-    return name.replace(":", "-")
-
+_TAGS_PATH = "/api/extras/tags/"
 
 _LIST_PATHS = {
     SERVICES: "/api/ipam/services/",
@@ -98,20 +94,32 @@ def _list_paginated(
             query = None
         return results
 
-    def list_tagged(self, kind: str, tags: Iterable[str]) -> list[dict[str, Any]]:
+    def tag_slugs(self, names: Iterable[str]) -> dict[str, str]:
+        """Map each tag name that exists in NetBox to the slug NetBox gave it.
+
+        NetBox derives the slug when a client sends none, and the rule differs between
+        clients, so the slug is read back instead of computed. Names NetBox doesn't know
+        are left out.
+        """
+        wanted = list(names)
+        params = [("limit", "1000"), *(("name", name) for name in wanted)]
+        found = {tag["name"]: tag["slug"] for tag in self._list_paginated(_TAGS_PATH, params)}
+        return {name: found[name] for name in wanted if name in found}
+
+    def list_tagged(self, kind: str, slugs: Iterable[str]) -> list[dict[str, Any]]:
         path = _LIST_PATHS[kind]
         # NetBox ?tag= filters are ANDed; query each tag separately and union by
         # id so an object tagged with ANY of the tags is included (OR).
         merged: dict[Any, dict[str, Any]] = {}
-        for tag in tags:
-            for obj in self._list_paginated(path, [("limit", "1000"), ("tag", _tag_slug(tag))]):
+        for slug in slugs:
+            for obj in self._list_paginated(path, [("limit", "1000"), ("tag", slug)]):
                 merged[obj.get("id", id(obj))] = obj
         return list(merged.values())
 
-    def list_records(self, kind: str, tags: Iterable[str]) -> tuple[list[Record], int]:
+    def list_records(self, kind: str, slugs: Iterable[str]) -> tuple[list[Record], int]:
         records: list[Record] = []
         unreadable = 0
-        for raw in self.list_tagged(kind, tags):
+        for raw in self.list_tagged(kind, slugs):
             try:
                 records.append(parse_record(kind, raw))
             except (KeyError, TypeError, AttributeError):

tests/netbox45.py

@@ -11,7 +11,7 @@
 import itertools
 import re
 from collections import defaultdict
-from collections.abc import Iterable, Iterator
+from collections.abc import Callable, Iterable, Iterator
 from typing import Any
 
 import httpx
@@ -204,9 +204,9 @@ class FakeNetbox:
     """The three list endpoints promsd calls, answering as NetBox v4.5.0 does.
 
     While `down` is set, every request gets the 503 of a proxy whose NetBox is
-    gone. A request without the token is a 403. `?tag=` matches tag slugs, repeated
-    `tag` parameters must all match, and a slug that is neither seeded nor on a
-    record is a 400. Every match comes back in one page.
+    gone. A request without the token is a 403. `?name=` on the tags endpoint matches
+    exactly. `?tag=` matches tag slugs, repeated `tag` parameters must all match, and a
+    slug that is neither seeded nor on a record is a 400. Every match comes back in one page.
     """
 
     def __init__(
@@ -215,20 +215,40 @@ def __init__(
         services: Iterable[dict[str, Any]] = (),
         devices: Iterable[dict[str, Any]] = (),
         virtual_machines: Iterable[dict[str, Any]] = (),
+        tag_slugs: dict[str, str] | None = None,
+        absent_tags: Iterable[str] = (),
     ) -> None:
         self._records = {
             _LIST_PATHS["services"]: list(services),
             _LIST_PATHS["devices"]: list(devices),
             _LIST_PATHS["virtual_machines"]: list(virtual_machines),
         }
-        self._slugs = {_slug(name) for name in SEEDED_TAGS} | {
-            tag["slug"]
-            for records in self._records.values()
-            for record in records
-            for tag in record["tags"]
-        }
+        # A tag created in the NetBox UI gets NetBox's own slug, `monitoringscrape` for
+        # `monitoring:scrape`. `tag_slugs` overrides the slug of a named tag everywhere.
+        overrides = tag_slugs or {}
+        self._tags = {name: overrides.get(name, _slug(name)) for name in SEEDED_TAGS}
+        for records in self._records.values():
+            for record in records:
+                for record_tag in record["tags"]:
+                    record_tag["slug"] = overrides.get(record_tag["name"], record_tag["slug"])
+                    self._tags[record_tag["name"]] = record_tag["slug"]
+        self._slugs = set(self._tags.values())
+        # Answers 400 to a list request this returns true for, such as one for a later page.
+        self.refuse: Callable[[httpx.Request], bool] = lambda request: False
+        # Page size of list endpoints; None puts every match in one page.
+        self.page_size: int | None = None
         self.requests: list[httpx.Request] = []
         self.down = False
+        for name in absent_tags:
+            self.remove_tag(name)
+
+    def remove_tag(self, name: str) -> None:
+        """Delete a tag. NetBox strips it from every object that carried it."""
+        self._tags.pop(name, None)
+        self._slugs = set(self._tags.values())
+        for records in self._records.values():
+            for record in records:
+                record["tags"] = [t for t in record["tags"] if t["name"] != name]
 
     def client(self, token: str = TOKEN) -> NetboxClient:
         return NetboxClient(base_url=BASE_URL, token=token, transport=httpx.MockTransport(self))
@@ -239,9 +259,21 @@ def __call__(self, request: httpx.Request) -> httpx.Response:
             return httpx.Response(503, text="Service Unavailable")
         if request.headers.get("Authorization") not in {f"Token {TOKEN}", f"Bearer {TOKEN}"}:
             return httpx.Response(403, json={"detail": "Invalid v1 token"})
+        if request.url.path == "/api/extras/tags/":
+            names = request.url.params.get_list("name")
+            found = [
+                {"id": i, "name": name, "slug": slug}
+                for i, (name, slug) in enumerate(self._tags.items(), 1)
+                if name in names
+            ]
+            return httpx.Response(
+                200, json={"count": len(found), "next": None, "previous": None, "results": found}
+            )
         records = self._records.get(request.url.path)
         if records is None:
             return httpx.Response(404, json={"detail": "Not found."})
+        if self.refuse(request):
+            return httpx.Response(400, json={"detail": "Bad request."})
         slugs = request.url.params.get_list("tag")
         unknown = [slug for slug in slugs if slug not in self._slugs]
         if unknown:
@@ -250,7 +282,17 @@ def __call__(self, request: httpx.Request) -> httpx.Response:
         matches = [
             record for record in records if set(slugs) <= {tag["slug"] for tag in record["tags"]}
         ]
+        offset = int(request.url.params.get("offset", 0))
+        size = self.page_size or max(len(matches), 1)
+        following = None
+        if offset + size < len(matches):
+            following = str(request.url.copy_set_param("offset", offset + size))
         return httpx.Response(
             200,
-            json={"count": len(matches), "next": None, "previous": None, "results": matches},
+            json={
+                "count": len(matches),
+                "next": following,
+                "previous": None,
+                "results": matches[offset : offset + size],
+            },
         )

tests/test_main.py

@@ -7,6 +7,7 @@
 from fastapi.testclient import TestClient
 
 from dmf_promsd import __version__
+from dmf_promsd.cache import PromSDCache
 from dmf_promsd.main import create_app
 from dmf_promsd.settings import Settings
 from tests.netbox45 import BASE_URL, TOKEN, FakeNetbox, device, service, virtual_machine
@@ -232,3 +233,105 @@ def test_a_record_promsd_cannot_read_costs_the_others_nothing():
     assert [group["targets"] for group in scrape] == [["192.0.2.11:9100"]]
     assert health["source_counts"]["devices"] == 2
     assert health["last_error"] is None
+
+
+def _lanes(http: TestClient) -> dict[str, list[str]]:
+    return {
+        lane: [t for group in http.get(f"/sd/{lane}").json() for t in group["targets"]]
+        for lane in ("scrape", "probe", "snmp")
+    }
+
+
+def _serve(netbox: FakeNetbox) -> tuple[dict[str, list[str]], dict]:
+    client = netbox.client()
+    app = create_app(Settings(netbox_url=BASE_URL, netbox_token=TOKEN), client=client)
+    try:
+        with TestClient(app) as http:
+            served = _lanes(http)
+            health = http.get("/healthz").json()
+            ready = http.get("/readyz")
+    finally:
+        client.close()
+    assert ready.status_code == 200
+    return served, health
+
+
+def _one_device_per_lane() -> list[dict]:
+    return [
+        device("node-1", ["monitoring:scrape"], address="192.0.2.11/24", metrics_port=9100),
+        device("encoder-1", ["monitoring:probe"], address="192.0.2.21/24", probe_module="icmp"),
+        device("switch-1", ["monitoring:snmp"], address="192.0.2.31/24"),
+    ]
+
+
+ALL_LANES = {"scrape": ["192.0.2.11:9100"], "probe": ["192.0.2.21"], "snmp": ["192.0.2.31"]}
+
+
+# promsd must use the slug NetBox chose, not one it computes from the tag name.
+def test_lanes_follow_the_slugs_netbox_chose():
+    ui_slugs = {f"monitoring:{lane}": f"monitoring{lane}" for lane in ("scrape", "probe", "snmp")}
+    served, health = _serve(FakeNetbox(devices=_one_device_per_lane(), tag_slugs=ui_slugs))
+
+    assert served == ALL_LANES
+    assert health["last_error"] is None
+
+
+def test_legacy_hyphen_slugs_still_work():
+    served, _ = _serve(FakeNetbox(devices=_one_device_per_lane()))
+
+    assert served == ALL_LANES
+
+
+def _refused_from_page(offset: int):
+    def refuse(request) -> bool:
+        return (
+            request.url.path == "/api/dcim/devices/"
+            and int(request.url.params.get("offset", 0)) >= offset
+        )
+
+    return refuse
+
+
+# A 400 on a record query fails the refresh and the last snapshot stays.
+def test_a_refused_record_query_keeps_the_last_snapshot():
+    for first_bad_offset in (0, 1):
+        netbox = FakeNetbox(devices=[*_one_device_per_lane(), _extra_scrape_device()])
+        netbox.page_size = 1
+        client = netbox.client()
+        cache = PromSDCache(client=client, ttl_seconds=60)
+        try:
+            assert cache.refresh()
+            good = cache.get_snapshot()
+            assert good is not None
+
+            netbox.refuse = _refused_from_page(first_bad_offset)
+            assert not cache.refresh()
+
+            state = cache.state()
+            assert state.snapshot is good
+            assert state.last_error is not None and "NetBox API 400" in state.last_error
+        finally:
+            client.close()
+
+
+def _extra_scrape_device() -> dict:
+    return device("node-2", ["monitoring:scrape"], address="192.0.2.12/24", metrics_port=9100)
+
+
+# With a lane tag unseen on the first refresh there is no snapshot to serve, so the
+# service is not ready and says which tags it didn't see.
+def test_a_missing_lane_tag_on_the_first_refresh_leaves_the_service_not_ready():
+    netbox = FakeNetbox(devices=_one_device_per_lane(), absent_tags=["monitoring:probe"])
+    client = netbox.client()
+    app = create_app(Settings(netbox_url=BASE_URL, netbox_token=TOKEN), client=client)
+    try:
+        with TestClient(app) as http:
+            ready = http.get("/readyz")
+            health = http.get("/healthz").json()
+    finally:
+        client.close()
+
+    assert ready.status_code == 503
+    assert ready.json()["reason"] == "no snapshot"
+    assert "monitoring:probe" in health["last_error"]
+    assert "missing_tags" not in health

tests/test_netbox.py

@@ -36,11 +36,11 @@ def handler(request: httpx.Request) -> httpx.Response:
     )
 
     try:
-        results = client.list_tagged("services", ("monitoring:scrape",))
+        results = client.list_tagged("services", ("monitoring-scrape",))
     finally:
         client.close()
 
     assert [item["id"] for item in results] == [1, 2]
     assert len(requests) == 2
-    # NetBox ?tag= filters by slug (colon -> hyphen), not the tag name.
+    # NetBox ?tag= filters by slug, not the tag name.
     assert requests[0].url.params.get("tag") == "monitoring-scrape"