diff --git a/README.md b/README.md index c19e2bc..aa9aa01 100644 --- a/README.md +++ b/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. diff --git a/src/dmf_promsd/cache.py b/src/dmf_promsd/cache.py index 74d2703..a9b3bcf 100644 --- a/src/dmf_promsd/cache.py +++ b/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) diff --git a/src/dmf_promsd/netbox.py b/src/dmf_promsd/netbox.py index 433a2ff..2e95f32 100644 --- a/src/dmf_promsd/netbox.py +++ b/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): diff --git a/tests/netbox45.py b/tests/netbox45.py index b82e15b..9e03e09 100644 --- a/tests/netbox45.py +++ b/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], + }, ) diff --git a/tests/test_main.py b/tests/test_main.py index 1f828b3..7c97f11 100644 --- a/tests/test_main.py +++ b/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 diff --git a/tests/test_netbox.py b/tests/test_netbox.py index a6368d3..b8f2b54 100644 --- a/tests/test_netbox.py +++ b/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" diff --git a/tests/test_tag_visibility.py b/tests/test_tag_visibility.py new file mode 100644 index 0000000..b1e991f --- /dev/null +++ b/tests/test_tag_visibility.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +from dmf_promsd.cache import PromSDCache +from tests.netbox45 import FakeNetbox, device + +LANE_TAGS = ["monitoring:scrape", "monitoring:probe", "monitoring:snmp"] + + +def _devices() -> list[dict]: + return [ + device("node-1", ["monitoring:scrape"], address="192.0.2.11/24", metrics_port=9100), + device("switch-1", ["monitoring:snmp"], address="192.0.2.31/24"), + ] + + +def _devices_for_every_lane() -> list[dict]: + return [ + *_devices(), + device("encoder-1", ["monitoring:probe"], address="192.0.2.21/24", probe_module="icmp"), + ] + + +# A token without tag-view access sees an empty tag list, the same as a NetBox with no +# lane tags. The refresh fails so the lanes don't go empty without a signal. +def test_no_visible_lane_tag_fails_the_refresh_and_keeps_the_last_snapshot(): + netbox = FakeNetbox(devices=_devices()) + client = netbox.client() + cache = PromSDCache(client=client, ttl_seconds=60) + try: + assert cache.refresh() + good = cache.get_snapshot() + assert good is not None + + for name in LANE_TAGS: + netbox.remove_tag(name) + assert not cache.refresh() + + state = cache.state() + assert state.snapshot is good + assert state.last_error is not None + for name in LANE_TAGS: + assert name in state.last_error + assert "extras.view_tag" in state.last_error + finally: + client.close() + + +# A token limited to some lane tags can't tell a hidden tag from a missing one, so any +# unseen lane tag fails the refresh and the earlier lanes stay. +def test_a_partly_visible_lane_tag_set_fails_the_refresh_and_keeps_every_lane(): + netbox = FakeNetbox(devices=_devices_for_every_lane()) + client = netbox.client() + cache = PromSDCache(client=client, ttl_seconds=60) + try: + assert cache.refresh() + good = cache.get_snapshot() + assert good is not None + assert good.lanes["probe"] and good.lanes["snmp"] + + netbox.remove_tag("monitoring:probe") + netbox.remove_tag("monitoring:snmp") + assert not cache.refresh() + + state = cache.state() + assert state.snapshot is good + assert cache.lane_payload("probe") == good.lanes["probe"] + assert cache.lane_payload("snmp") == good.lanes["snmp"] + assert state.last_error is not None + assert "monitoring:probe, monitoring:snmp" in state.last_error + assert "monitoring:scrape" not in state.last_error + assert "extras.view_tag" in state.last_error + finally: + client.close() + + +def test_an_unseen_lane_tag_on_the_first_refresh_leaves_no_snapshot(): + client = FakeNetbox(devices=_devices(), absent_tags=["monitoring:probe"]).client() + cache = PromSDCache(client=client, ttl_seconds=60) + try: + assert not cache.refresh() + assert cache.get_snapshot() is None + assert cache.state().last_error is not None + finally: + client.close()