SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(perception): stabilize pre-capture methodology

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-28 17:47:06 +03:00
parent 1f20e0d7d9
commit d729abab31
65 changed files with 9698 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/telemetry-plane/normalizer/normalizer.py
+191 -19
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import contextlib
import json
import os
import re
@@ -25,6 +26,40 @@ SOURCE_SCHEMAS: Final = {
TELEGRAF_SCHEMA: Final = "telegraf.metric-json/v1"
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
MAX_TAGS: Final = 256
MAX_SERIES_KEY_BYTES: Final = 4096
RETENTION_INTERVAL_SECONDS: Final = 86_400
ALLOWED_TELEMETRY_TAGS: Final = frozenset(
{
"agent_id",
"container",
"container_id",
"container_image",
"container_name",
"container_status",
"container_version",
"contour_id",
"cpu",
"device",
"engine_host",
"gpu_name",
"host",
"interface",
"lab_id",
"method_id",
"name",
"node_id",
"path",
"request_id",
"run_id",
"server_version",
"source_id",
"source_package_id",
"stage_id",
"stage_state",
}
)
def _required(name: str) -> str:
@@ -68,26 +103,48 @@ def _series_key(document: dict[str, Any]) -> str:
if not isinstance(tags, dict):
value = document.get("series_key")
return value if isinstance(value, str) else ""
if len(tags) > MAX_TAGS:
raise ValueError("telemetry payload contains too many tags")
stable_tags = {
str(name): str(value)
for name, value in tags.items()
if name not in {"agent_id", "contour_id", "host", "node_id"}
}
return json.dumps(
encoded = json.dumps(
stable_tags,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
if len(encoded.encode("utf-8")) > MAX_SERIES_KEY_BYTES:
raise ValueError("telemetry series identity is too large")
return encoded
def _sanitize_tags(document: dict[str, Any]) -> dict[str, Any]:
tags = document.get("tags")
if not isinstance(tags, dict):
return document
if len(tags) > MAX_TAGS:
raise ValueError("telemetry payload contains too many tags")
sanitized = {
str(name): str(value)
for name, value in tags.items()
if name in ALLOWED_TELEMETRY_TAGS
}
return {**document, "tags": sanitized}
def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
if len(payload) > MAX_PAYLOAD_BYTES:
raise ValueError("telemetry payload exceeds the 1 MiB contract")
match = TOPIC_RE.fullmatch(topic)
if match is None:
raise ValueError("topic is outside the telemetry contract")
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")
source_schema = document.get("schema_version")
if source_schema == SOURCE_SCHEMAS[kind]:
@@ -101,14 +158,18 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
source_schema = TELEGRAF_SCHEMA
observed_value = document.get("timestamp")
measurement = document["name"].strip()
if not measurement:
if not measurement or len(measurement) > 128:
raise ValueError("Telegraf measurement name is required")
else:
raise ValueError("payload schema does not match topic kind")
node_id = document.get("node_id") or _tag_text(document, "node_id")
if not node_id:
node_id = _tag_text(document, "host")
if not isinstance(node_id, str) or not node_id.strip():
if (
not isinstance(node_id, str)
or not node_id.strip()
or len(node_id.strip()) > 128
):
raise ValueError("node_id is required")
observed_at = _timestamp(observed_value)
return (
@@ -132,9 +193,53 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
class TelemetryQueryServer(ThreadingHTTPServer):
daemon_threads = True
def __init__(self, server_address: tuple[str, int], dsn: str) -> None:
def __init__(
self,
server_address: tuple[str, int],
dsn: str,
runtime_health: TelemetryRuntimeHealth,
) -> None:
super().__init__(server_address, TelemetryQueryHandler)
self.dsn = dsn
self.runtime_health = runtime_health
class TelemetryRuntimeHealth:
"""Thread-safe health of the actual MQTT-to-database ingestion path."""
def __init__(self) -> None:
self._lock = threading.Lock()
self._mqtt_connected = False
self._database_connected = False
self._last_ingested_at_utc: str | None = None
self._last_error: str | None = None
def set_mqtt(self, connected: bool, error: str | None = None) -> None:
with self._lock:
self._mqtt_connected = connected
self._last_error = error
def set_database(self, connected: bool, error: str | None = None) -> None:
with self._lock:
self._database_connected = connected
self._last_error = error
def mark_ingested(self) -> None:
with self._lock:
self._database_connected = True
self._last_ingested_at_utc = (
datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
)
self._last_error = None
def snapshot(self) -> dict[str, object]:
with self._lock:
return {
"mqtt_connected": self._mqtt_connected,
"database_connected": self._database_connected,
"last_ingested_at_utc": self._last_ingested_at_utc,
"last_error": self._last_error,
}
class TelemetryQueryHandler(BaseHTTPRequestHandler):
@@ -174,6 +279,7 @@ class TelemetryQueryHandler(BaseHTTPRequestHandler):
def _health(self) -> None:
import psycopg # type: ignore[import-not-found]
database_reachable = False
try:
with (
psycopg.connect(self.server.dsn) as connection,
@@ -181,13 +287,24 @@ class TelemetryQueryHandler(BaseHTTPRequestHandler):
):
cursor.execute("SELECT 1")
cursor.fetchone()
database_reachable = True
except psycopg.Error:
self._json(
HTTPStatus.SERVICE_UNAVAILABLE,
{"ok": False, "schema_version": QUERY_SCHEMA},
)
return
self._json(HTTPStatus.OK, {"ok": True, "schema_version": QUERY_SCHEMA})
database_reachable = False
runtime = self.server.runtime_health.snapshot()
ok = (
database_reachable
and runtime["mqtt_connected"] is True
and runtime["database_connected"] is True
)
self._json(
HTTPStatus.OK if ok else HTTPStatus.SERVICE_UNAVAILABLE,
{
"ok": ok,
"schema_version": QUERY_SCHEMA,
"database_reachable": database_reachable,
**runtime,
},
)
def _latest(
self,
@@ -270,10 +387,13 @@ class TelemetryQueryHandler(BaseHTTPRequestHandler):
self.wfile.write(body)
def _start_query_server(dsn: str) -> ThreadingHTTPServer:
def _start_query_server(
dsn: str,
runtime_health: TelemetryRuntimeHealth,
) -> ThreadingHTTPServer:
host = os.environ.get("MISSIONCORE_QUERY_HOST", "0.0.0.0")
port = int(os.environ.get("MISSIONCORE_QUERY_PORT", "18030"))
server = TelemetryQueryServer((host, port), dsn)
server = TelemetryQueryServer((host, port), dsn, runtime_health)
thread = threading.Thread(
target=server.serve_forever,
name="telemetry-query",
@@ -294,8 +414,11 @@ def main() -> None:
port = int(os.environ.get("MISSIONCORE_MQTT_PORT", "1883"))
username = _required("MISSIONCORE_MQTT_USERNAME")
password = _required("MISSIONCORE_MQTT_PASSWORD")
_start_query_server(dsn)
runtime_health = TelemetryRuntimeHealth()
connection = psycopg.connect(dsn, autocommit=True)
last_retention_monotonic = 0.0
runtime_health.set_database(True)
_start_query_server(dsn, runtime_health)
client = mqtt.Client(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id="ndc-mission-core-telemetry-normalizer",
@@ -311,14 +434,44 @@ def main() -> None:
_properties: Any,
) -> None:
if reason_code.is_failure:
runtime_health.set_mqtt(False, f"MQTT connection rejected: {reason_code}")
raise RuntimeError(f"MQTT connection rejected: {reason_code}")
connected_client.subscribe(TOPIC, qos=1)
runtime_health.set_mqtt(True)
def on_disconnect(
_connected_client: Any,
_userdata: object,
_disconnect_flags: Any,
reason_code: Any,
_properties: Any,
) -> None:
runtime_health.set_mqtt(False, f"MQTT disconnected: {reason_code}")
def on_message(_client: Any, _userdata: object, message: Any) -> None:
nonlocal connection, last_retention_monotonic
try:
row = _normalize(message.topic, message.payload)
with connection.cursor() as cursor:
cursor.execute(
except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
print(f"telemetry sample rejected: {exc}", flush=True)
return
for attempt in range(2):
try:
with connection.cursor() as cursor:
now_monotonic = time.monotonic()
if (
now_monotonic - last_retention_monotonic
>= RETENTION_INTERVAL_SECONDS
):
cursor.execute(
"""
DELETE FROM contour_telemetry_samples
WHERE observed_at
< CURRENT_TIMESTAMP - INTERVAL '30 days'
"""
)
last_retention_monotonic = now_monotonic
cursor.execute(
"""
INSERT INTO contour_telemetry_samples (
observed_at, contour_id, agent_id, node_id, kind, measurement,
@@ -358,18 +511,37 @@ def main() -> None:
ELSE EXCLUDED.payload
END
""",
(*row, TELEGRAF_SCHEMA, TELEGRAF_SCHEMA),
)
except (ValueError, UnicodeDecodeError, json.JSONDecodeError, psycopg.Error) as exc:
print(f"telemetry sample rejected: {exc}", flush=True)
(*row, TELEGRAF_SCHEMA, TELEGRAF_SCHEMA),
)
runtime_health.mark_ingested()
return
except psycopg.Error as exc:
runtime_health.set_database(False, f"database ingestion failed: {exc}")
with contextlib.suppress(psycopg.Error):
connection.close()
if attempt == 0:
try:
connection = psycopg.connect(dsn, autocommit=True)
runtime_health.set_database(True)
except psycopg.Error as reconnect_error:
runtime_health.set_database(
False,
f"database reconnect failed: {reconnect_error}",
)
break
else:
break
print("telemetry sample rejected: database unavailable", flush=True)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message
while True:
try:
client.connect(host, port, keepalive=30)
client.loop_forever(retry_first_connection=True)
except (OSError, MQTTException, psycopg.Error) as exc:
runtime_health.set_mqtt(False, f"MQTT reconnecting: {exc}")
print(f"telemetry normalizer reconnecting: {exc}", flush=True)
time.sleep(3)