fix(observability): persist Kubernetes events through OTLP

scripts/reboot-recovery/tests/test_event_exporter_metadata_rbac.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+
+
+ROOT = Path(__file__).resolve().parents[3]
+MANIFEST = ROOT / "k8s" / "observability" / "event-exporter.yaml"
+
+
+def test_event_exporter_has_minimal_metadata_enrichment_reads() -> None:
+    documents = [
+        doc for doc in yaml.safe_load_all(MANIFEST.read_text(encoding="utf-8")) if doc
+    ]
+    role = next(doc for doc in documents if doc["kind"] == "ClusterRole")
+    rules = role["rules"]
+
+    expected = {
+        ("", "pods"),
+        ("apps", "deployments"),
+        ("apps", "replicasets"),
+        ("apps", "daemonsets"),
+        ("batch", "jobs"),
+        ("batch", "cronjobs"),
+        ("argoproj.io", "applications"),
+    }
+    granted = {
+        (api_group, resource)
+        for rule in rules
+        for api_group in rule["apiGroups"]
+        for resource in rule["resources"]
+        if rule["verbs"] == ["get"]
+    }
+    assert expected <= granted
+    assert all(resource not in {"secrets", "configmaps"} for _, resource in granted)
+
+
+def test_event_exporter_metadata_rules_never_write_or_delete() -> None:
+    documents = [
+        doc for doc in yaml.safe_load_all(MANIFEST.read_text(encoding="utf-8")) if doc
+    ]
+    role = next(doc for doc in documents if doc["kind"] == "ClusterRole")
+    forbidden = {"create", "update", "patch", "delete", "deletecollection"}
+    assert not any(forbidden.intersection(rule["verbs"]) for rule in role["rules"])
+
+
+def _event_exporter_config() -> dict:
+    documents = [
+        doc for doc in yaml.safe_load_all(MANIFEST.read_text(encoding="utf-8")) if doc
+    ]
+    configmap = next(doc for doc in documents if doc["kind"] == "ConfigMap")
+    return yaml.safe_load(configmap["data"]["config.yaml"])
+
+
+def test_event_exporter_webhook_emits_a_real_otlp_json_envelope() -> None:
+    config = _event_exporter_config()
+    receivers = config["receivers"]
+    assert [receiver["name"] for receiver in receivers] == ["signoz-otlp"]
+
+    webhook = receivers[0]["webhook"]
+    assert webhook["endpoint"].endswith("/v1/logs")
+    assert webhook["headers"]["Content-Type"] == "application/json"
+
+    resource_logs = webhook["layout"]["resourceLogs"]
+    log_record = resource_logs[0]["scopeLogs"][0]["logRecords"][0]
+    assert log_record["body"]["stringValue"] == "{{ .Message }}"
+    assert log_record["severityText"] == "{{ .Type }}"
+    attribute_keys = {item["key"] for item in log_record["attributes"]}
+    assert {
+        "k8s.event.reason",
+        "k8s.event.type",
+        "k8s.namespace.name",
+        "k8s.object.kind",
+        "k8s.object.name",
+    } <= attribute_keys
+
+
+def test_event_exporter_has_one_route_and_drops_only_known_normal_noise() -> None:
+    config = _event_exporter_config()
+    routes = config["route"]["routes"]
+    assert len(routes) == 1
+    assert routes[0]["match"] == [{"receiver": "signoz-otlp"}]
+    assert {item["type"] for item in routes[0]["drop"]} == {"Normal"}
+    assert {item["reason"] for item in routes[0]["drop"]} == {
+        "Scheduled",
+        "Pulling",
+        "Pulled",
+        "Created",
+        "Started",
+    }