k3s: Pod-Monitoring in Grafana (Task 6.5)

kube-state-metrics (NodePort 30808) und ServiceAccount prometheus-infra (nur
nodes/metrics) im Cluster; Prometheus auf infra liest kubelet/cAdvisor per TLS
gegen die k3s-CA. Ziele per file_sd aus kubectl get nodes, cAdvisor auf die
nötigen Reihen begrenzt. Dashboard "k3s – Pods" (Vorgabe nex), Alerts K3s*
nur mit notify. Playbook monitoring-k3s.yml.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

core/tests/test_artifacts.py

@@ -39,13 +39,18 @@ def test_healer_labels_available_on_alerts():
     assert set(OPS_LABELS) == {"docker_host", "container", "pve_node", "vmid"}
 
 
-def _monitoring_targets() -> dict:
+def _targets_module():
     import importlib.util
 
     spec = importlib.util.spec_from_file_location(
         "monitoring_targets", REPO_ROOT / "platform/proxmox/scripts/monitoring-targets.py")
     mod = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(mod)
+    return mod
+
+
+def _monitoring_targets() -> dict:
+    mod = _targets_module()
     registry = yaml.safe_load((REPO_ROOT / "sites/knaster/registry.yaml").read_text(encoding="utf-8"))
     gpu_vms = {  # Form wie `tofu output -json gpu_vms`
         "gpu-pve44": {"node": "pve44", "vmid": 244, "ipv4": "192.168.100.186"},
@@ -67,6 +72,55 @@ def test_monitoring_targets_from_tofu_output():
     assert t["node"][2]["labels"]["slot"] == "none"   # gpu-pve52 trägt noch keinen Slot
 
 
+def _k3s_node(name: str, ip: str | None, ready: str = "True") -> dict:
+    addresses = [{"type": "Hostname", "address": name}] + ([{"type": "InternalIP", "address": ip}] if ip else [])
+    return {"metadata": {"name": name},
+            "status": {"addresses": addresses, "conditions": [{"type": "Ready", "status": ready}]}}
+
+
+def test_k3s_targets_from_kubectl_nodes():
+    nodes = {"items": [  # Form wie `kubectl get nodes -o json`
+        _k3s_node("gpu-pve52", "192.168.100.174"),
+        _k3s_node("k3s", "192.168.100.124"),
+        _k3s_node("gpu-pve44", "192.168.100.186", ready="False"),
+        _k3s_node("gpu-pve46", None),
+    ]}
+    t = _targets_module().build_k3s_targets(nodes)
+    assert [(e["targets"][0], e["labels"]["instance"]) for e in t["kubelet"]] == [
+        ("192.168.100.186:10250", "gpu-pve44"), ("192.168.100.174:10250", "gpu-pve52"),
+        ("192.168.100.124:10250", "k3s")]   # gpu-pve46 ohne IP fehlt
+    assert t["kube-state-metrics"] == [
+        {"targets": ["192.168.100.124:30808"], "labels": {"instance": "kube-state-metrics"}}]   # Dienst-VM bevorzugt
+    t = _targets_module().build_k3s_targets({"items": [nodes["items"][0], nodes["items"][2]]})
+    assert t["kube-state-metrics"][0]["targets"] == ["192.168.100.174:30808"]   # erster bereiter Knoten
+    assert _targets_module().build_k3s_targets({}) == {"kubelet": [], "kube-state-metrics": []}
+
+
+def test_k3s_jobs_use_generated_targets_and_nodeport():
+    prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text())
+    jobs = {j["job_name"]: j for j in prom["scrape_configs"]}
+    mod = _targets_module()
+    written = set(mod.build_k3s_targets({}))
+    for job, sd in {"kubelet": "kubelet", "cadvisor": "kubelet", "kube-state-metrics": "kube-state-metrics"}.items():
+        assert jobs[job]["file_sd_configs"] == [{"files": [f"/etc/knaster-sd/{sd}.json"]}]
+        assert sd in written
+    for job in ("kubelet", "cadvisor"):
+        assert jobs[job]["scheme"] == "https" and jobs[job]["tls_config"]["ca_file"].startswith("/etc/knaster-k3s/")
+    docs = list(yaml.safe_load_all((DEPLOY / "k8s/monitoring/kube-state-metrics.yaml").read_text()))
+    svc = next(d for d in docs if d["kind"] == "Service")
+    assert svc["spec"]["ports"][0]["nodePort"] == mod.KSM_NODEPORT
+
+
+DASHBOARDS = sorted((DEPLOY / "grafana/dashboards").glob("*.json"))
+
+
+@pytest.mark.parametrize("path", DASHBOARDS, ids=[p.name for p in DASHBOARDS])
+def test_grafana_dashboards_parse_with_unique_uid(path):
+    d = json.loads(path.read_text(encoding="utf-8"))
+    assert d["uid"] and d["title"] and d["panels"]
+    assert [x for x in DASHBOARDS if json.loads(x.read_text(encoding="utf-8"))["uid"] == d["uid"]] == [path]
+
+
 def test_alertmanager_targets_router_endpoint():
     am = yaml.safe_load((DEPLOY / "alertmanager/alertmanager.yml").read_text())
     url = am["receivers"][0]["webhook_configs"][0]["url"]