feat(telemetry): add bounded native pipeline lifecycle

deploy/telemetry-plane/normalizer/normalizer.py

@@ -24,6 +24,8 @@ SOURCE_SCHEMAS: Final = {
     "heartbeat": "missioncore.agent-heartbeat/v1",
 }
 TELEGRAF_SCHEMA: Final = "telegraf.metric-json/v1"
+PIPELINE_RECORD_SCHEMA: Final = "missioncore.pipeline-telemetry-record/v1"
+PIPELINE_EVENT_MEASUREMENT: Final = "missioncore_pipeline_event"
 QUERY_SCHEMA: Final = "missioncore.telemetry-query/v1"
 SAFE_IDENTIFIER: Final = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")
 MAX_PAYLOAD_BYTES: Final = 1024 * 1024
@@ -43,6 +45,7 @@ ALLOWED_TELEMETRY_TAGS: Final = frozenset(
         "cpu",
         "device",
         "engine_host",
+        "event_type",
         "gpu_name",
         "host",
         "interface",
@@ -53,6 +56,7 @@ ALLOWED_TELEMETRY_TAGS: Final = frozenset(
         "path",
         "request_id",
         "run_id",
+        "run_state",
         "server_version",
         "source_id",
         "source_package_id",
@@ -135,6 +139,30 @@ def _sanitize_tags(document: dict[str, Any]) -> dict[str, Any]:
     return {**document, "tags": sanitized}
 
 
+def _unwrap_pipeline_event(
+    topic: str,
+    kind: str,
+    document: dict[str, Any],
+) -> dict[str, Any]:
+    if kind != "pipeline" or document.get("name") != PIPELINE_EVENT_MEASUREMENT:
+        return document
+    fields = document.get("fields")
+    if not isinstance(fields, dict) or not isinstance(fields.get("value"), str):
+        raise ValueError("Telegraf pipeline event value is missing")
+    record = json.loads(fields["value"])
+    if (
+        not isinstance(record, dict)
+        or record.get("schema_version") != PIPELINE_RECORD_SCHEMA
+        or record.get("topic") != topic
+        or not isinstance(record.get("payload"), dict)
+    ):
+        raise ValueError("pipeline event record is invalid")
+    payload = record["payload"]
+    if payload.get("schema_version") != SOURCE_SCHEMAS["pipeline"]:
+        raise ValueError("pipeline event payload schema is invalid")
+    return payload
+
+
 def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
     if len(payload) > MAX_PAYLOAD_BYTES:
         raise ValueError("telemetry payload exceeds the 1 MiB contract")
@@ -144,8 +172,9 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
     document = json.loads(payload.decode("utf-8"))
     if not isinstance(document, dict):
         raise ValueError("payload must be an object")
-    document = _sanitize_tags(document)
     kind = match.group("kind")
+    document = _unwrap_pipeline_event(topic, kind, document)
+    document = _sanitize_tags(document)
     source_schema = document.get("schema_version")
     if source_schema == SOURCE_SCHEMAS[kind]:
         observed_value = document.get("observed_at_utc")