From 3f708ba0e6fb911cbac3971394eba6bc6ffe1475 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:48:08 +0200 Subject: [PATCH 01/11] Add health message parser --- .../pyxbot2_diagnostics/aggregator/health.py | 329 ++++++++++++++++++ 1 file changed, 329 insertions(+) create mode 100644 python/src/pyxbot2_diagnostics/aggregator/health.py diff --git a/python/src/pyxbot2_diagnostics/aggregator/health.py b/python/src/pyxbot2_diagnostics/aggregator/health.py new file mode 100644 index 0000000..3518b36 --- /dev/null +++ b/python/src/pyxbot2_diagnostics/aggregator/health.py @@ -0,0 +1,329 @@ +"""Validation and normalization for device health diagnostics.""" + +from __future__ import annotations + +import json +import math +from dataclasses import dataclass +from datetime import datetime, timezone +from decimal import Decimal +from typing import Any, Iterable + +from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticKeyValue, DiagnosticsMessage + +HEALTH_SCHEMA_NAME = "xbot.device_health" +HEALTH_SCHEMA_VERSION = 1 +HEALTH_STATUS_SUFFIXES = frozenset({"health", "health_status"}) + +_SCHEMA_NAME_KEY = "schema.name" +_SCHEMA_VERSION_KEY = "schema.version" +_BOOT_ID_KEY = "device.boot_id" +_ACTIVE_KEY = "faults.active" +_RAISE_COUNT_KEY = "faults.raise_count_total" +_LAST_RAISED_KEY = "faults.last_raised" +_LAST_CLEARED_KEY = "faults.last_cleared" + +REQUIRED_HEALTH_KEYS = frozenset( + { + _SCHEMA_NAME_KEY, + _SCHEMA_VERSION_KEY, + _BOOT_ID_KEY, + _ACTIVE_KEY, + _RAISE_COUNT_KEY, + _LAST_RAISED_KEY, + _LAST_CLEARED_KEY, + } +) + + +class HealthMessageValidationError(ValueError): + """Raised when a health diagnostic does not satisfy the health contract.""" + + +@dataclass(frozen=True) +class FaultHealthRecord: + """Normalized state for one fault code.""" + + code: str + active: bool + raise_count_total: int + last_raised_ns: int | None + last_cleared_ns: int | None + + +@dataclass(frozen=True) +class HealthStatus: + """Normalized device health snapshot.""" + + device_path: str + schema_name: str + schema_version: int + boot_id: str + faults: tuple[FaultHealthRecord, ...] + extra_values: tuple[DiagnosticKeyValue, ...] + + @property + def active_fault_count(self) -> int: + return sum(fault.active for fault in self.faults) + + +def is_health_message(message: DiagnosticsMessage) -> bool: + """Return whether *message* is identified as a device health status.""" + + parts = _path_parts(message.node) + return bool(parts) and parts[-1] in HEALTH_STATUS_SUFFIXES + + +def parse_health_message(message: DiagnosticsMessage) -> HealthStatus: + """Validate and normalize a ``/health`` or ``/health_status`` message.""" + + parts = _path_parts(message.node) + if not parts or parts[-1] not in HEALTH_STATUS_SUFFIXES: + raise HealthMessageValidationError( + "health status name must end with '/health' or '/health_status'" + ) + if len(parts) < 2: + raise HealthMessageValidationError("health status name must include a device path") + if not message.hw_id.strip(): + raise HealthMessageValidationError("health status requires a non-empty hardware_id") + if not math.isfinite(message.stamp) or message.stamp < 0: + raise HealthMessageValidationError("health status stamp must be finite and non-negative") + + values = _unique_value_map(message.values) + missing = sorted(REQUIRED_HEALTH_KEYS - values.keys()) + if missing: + raise HealthMessageValidationError( + "health status is missing required keys: " + ", ".join(missing) + ) + + schema_name = _require_non_empty_string(values[_SCHEMA_NAME_KEY], _SCHEMA_NAME_KEY) + if schema_name != HEALTH_SCHEMA_NAME: + raise HealthMessageValidationError( + f"{_SCHEMA_NAME_KEY} must be '{HEALTH_SCHEMA_NAME}', got '{schema_name}'" + ) + + schema_version = _parse_schema_version(values[_SCHEMA_VERSION_KEY]) + if schema_version != HEALTH_SCHEMA_VERSION: + raise HealthMessageValidationError( + f"unsupported health schema version {schema_version}; " + f"expected {HEALTH_SCHEMA_VERSION}" + ) + + boot_id = _require_non_empty_string(values[_BOOT_ID_KEY], _BOOT_ID_KEY) + active_codes = _parse_active_faults(values[_ACTIVE_KEY]) + counts = _parse_counter_map(values[_RAISE_COUNT_KEY]) + last_raised = _parse_timestamp_map(values[_LAST_RAISED_KEY], _LAST_RAISED_KEY) + last_cleared = _parse_timestamp_map(values[_LAST_CLEARED_KEY], _LAST_CLEARED_KEY) + + referenced_codes = set(active_codes) | set(last_raised) | set(last_cleared) + unknown_codes = sorted(referenced_codes - counts.keys()) + if unknown_codes: + raise HealthMessageValidationError( + f"{_RAISE_COUNT_KEY} is missing referenced fault codes: " + + ", ".join(unknown_codes) + ) + + faults: list[FaultHealthRecord] = [] + active_set = set(active_codes) + for code in sorted(counts): + count = counts[code] + raised_ns = last_raised.get(code) + cleared_ns = last_cleared.get(code) + active = code in active_set + + if count == 0 and raised_ns is not None: + raise HealthMessageValidationError( + f"fault '{code}' has zero raises but a non-null last-raised timestamp" + ) + if count > 0 and raised_ns is None: + raise HealthMessageValidationError( + f"fault '{code}' has a positive raise counter but no last-raised timestamp" + ) + if active and count == 0: + raise HealthMessageValidationError( + f"active fault '{code}' must have a positive raise counter" + ) + if raised_ns is not None and cleared_ns is not None: + if active and cleared_ns >= raised_ns: + raise HealthMessageValidationError( + f"active fault '{code}' has last-cleared >= last-raised" + ) + if not active and raised_ns > cleared_ns: + raise HealthMessageValidationError( + f"inactive fault '{code}' has last-raised > last-cleared" + ) + + faults.append( + FaultHealthRecord( + code=code, + active=active, + raise_count_total=count, + last_raised_ns=raised_ns, + last_cleared_ns=cleared_ns, + ) + ) + + extra_values = tuple( + entry for entry in message.values if entry.key not in REQUIRED_HEALTH_KEYS + ) + device_path = "/" + "/".join(parts[:-1]) + return HealthStatus( + device_path=device_path, + schema_name=schema_name, + schema_version=schema_version, + boot_id=boot_id, + faults=tuple(faults), + extra_values=extra_values, + ) + + +def timestamp_seconds_to_ns(value: int | float) -> int: + """Convert finite non-negative epoch seconds to integer nanoseconds.""" + + seconds = Decimal(str(value)) + if not seconds.is_finite() or seconds < 0: + raise HealthMessageValidationError( + "timestamp seconds must be finite and non-negative" + ) + return int(seconds * Decimal(1_000_000_000)) + + +def _path_parts(path: str) -> list[str]: + return [part for part in path.split("/") if part] + + +def _unique_value_map(values: Iterable[DiagnosticKeyValue]) -> dict[str, Any]: + result: dict[str, Any] = {} + duplicates: list[str] = [] + for entry in values: + if entry.key in result: + duplicates.append(entry.key) + else: + result[entry.key] = entry.value + if duplicates: + raise HealthMessageValidationError( + "health status contains duplicate keys: " + ", ".join(sorted(set(duplicates))) + ) + return result + + +def _require_non_empty_string(value: Any, key: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise HealthMessageValidationError(f"{key} must be a non-empty string") + return value.strip() + + +def _parse_schema_version(value: Any) -> int: + if isinstance(value, bool): + raise HealthMessageValidationError(f"{_SCHEMA_VERSION_KEY} must be an integer") + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value.strip()) + except ValueError as exc: + raise HealthMessageValidationError( + f"{_SCHEMA_VERSION_KEY} must be an integer" + ) from exc + raise HealthMessageValidationError(f"{_SCHEMA_VERSION_KEY} must be an integer") + + +def _decode_json(value: Any, key: str) -> Any: + if isinstance(value, str): + try: + return json.loads(value) + except json.JSONDecodeError as exc: + raise HealthMessageValidationError(f"{key} contains invalid JSON: {exc.msg}") from exc + return value + + +def _parse_active_faults(value: Any) -> tuple[str, ...]: + decoded = _decode_json(value, _ACTIVE_KEY) + if not isinstance(decoded, list): + raise HealthMessageValidationError(f"{_ACTIVE_KEY} must be a JSON array") + + result: list[str] = [] + for index, code in enumerate(decoded): + if not isinstance(code, str) or not code.strip(): + raise HealthMessageValidationError( + f"{_ACTIVE_KEY}[{index}] must be a non-empty string" + ) + result.append(code.strip()) + + if len(result) != len(set(result)): + raise HealthMessageValidationError(f"{_ACTIVE_KEY} must not contain duplicates") + return tuple(result) + + +def _parse_counter_map(value: Any) -> dict[str, int]: + decoded = _decode_json(value, _RAISE_COUNT_KEY) + if not isinstance(decoded, dict): + raise HealthMessageValidationError(f"{_RAISE_COUNT_KEY} must be a JSON object") + + result: dict[str, int] = {} + for raw_code, count in decoded.items(): + code = _validate_fault_code(raw_code, _RAISE_COUNT_KEY) + if isinstance(count, bool) or not isinstance(count, int) or count < 0: + raise HealthMessageValidationError( + f"{_RAISE_COUNT_KEY}['{code}'] must be a non-negative integer" + ) + result[code] = count + return result + + +def _parse_timestamp_map(value: Any, key: str) -> dict[str, int | None]: + decoded = _decode_json(value, key) + if not isinstance(decoded, dict): + raise HealthMessageValidationError(f"{key} must be a JSON object") + + result: dict[str, int | None] = {} + for raw_code, timestamp in decoded.items(): + code = _validate_fault_code(raw_code, key) + result[code] = _parse_timestamp(timestamp, f"{key}['{code}']") + return result + + +def _validate_fault_code(value: Any, key: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise HealthMessageValidationError(f"{key} fault codes must be non-empty strings") + return value.strip() + + +def _parse_timestamp(value: Any, location: str) -> int | None: + if value is None: + return None + if isinstance(value, bool): + raise HealthMessageValidationError( + f"{location} must be null, epoch seconds, or an ISO-8601 timestamp" + ) + if isinstance(value, (int, float)): + seconds = float(value) + if not math.isfinite(seconds) or seconds < 0: + raise HealthMessageValidationError( + f"{location} epoch seconds must be finite and non-negative" + ) + return timestamp_seconds_to_ns(seconds) + if not isinstance(value, str) or not value.strip(): + raise HealthMessageValidationError( + f"{location} must be null, epoch seconds, or an ISO-8601 timestamp" + ) + + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise HealthMessageValidationError( + f"{location} must be a valid ISO-8601 timestamp" + ) from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise HealthMessageValidationError(f"{location} must include a timezone") + utc = parsed.astimezone(timezone.utc) + epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) + delta = utc - epoch + return ( + delta.days * 86_400 * 1_000_000_000 + + delta.seconds * 1_000_000_000 + + delta.microseconds * 1_000 + ) From ed90410fbb30b79692c53ab0ed3f3878adab1799 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:48:30 +0200 Subject: [PATCH 02/11] Normalize health diagnostics for InfluxDB --- .../aggregator/sinks/influxdb_sink.py | 153 +++++++++++++++--- 1 file changed, 132 insertions(+), 21 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index 2cc2326..e73623b 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -7,11 +7,20 @@ from typing import Any from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticsMessage +from pyxbot2_diagnostics.aggregator.health import ( + HealthMessageValidationError, + HealthStatus, + is_health_message, + parse_health_message, + timestamp_seconds_to_ns, +) LOGGER = logging.getLogger(__name__) -# Single measurement name for all robot diagnostics. +# Generic measurement name for ordinary robot diagnostics. _MEASUREMENT = "robot_diagnostics" +_DEVICE_HEALTH_MEASUREMENT = "device_health" +_FAULT_COUNTER_MEASUREMENT = "fault_counter" # Minimum seconds between batch writes to InfluxDB. _FLUSH_INTERVAL_SEC = 1.0 @@ -20,15 +29,21 @@ class InfluxDBSink: """Write diagnostics to InfluxDB v2. - Schema - ------ - measurement : robot_diagnostics - tags : hw_id, path (full status name), name (last path segment) - fields : level (int), one float field per kv-pair in status.values, - message (str, only when non-empty) + Ordinary diagnostics retain the existing schema. Messages whose path ends in + ``/health`` or ``/health_status`` are validated against the device-health + contract and normalized into two measurements: - Points are buffered in handle_message and flushed as a single batch write - at most once per _FLUSH_INTERVAL_SEC to avoid per-message HTTP overhead. + ``device_health`` + tags: hw_id, path, device_path, schema, schema_version, boot_id + fields: level, active_fault_count, message, plus non-contract scalar values + + ``fault_counter`` + one point per fault code + tags: hw_id, path, device_path, fault_code, schema_version, boot_id + fields: active, raise_count_total, last_raised_ns, last_cleared_ns + + Invalid health messages are logged and omitted from InfluxDB rather than being + written as generic diagnostics with opaque JSON fields. """ def __init__( @@ -79,6 +94,27 @@ def handle_message(self, message: DiagnosticsMessage) -> None: if not self._enabled or self._write_api is None: return + if is_health_message(message): + self._handle_health_message(message) + return + + self._pending.append(self._generic_point(message)) + + def _handle_health_message(self, message: DiagnosticsMessage) -> None: + try: + health = parse_health_message(message) + except HealthMessageValidationError as exc: + LOGGER.warning("Rejecting invalid health message %s: %s", message.node, exc) + return + + sample_time_ns = timestamp_seconds_to_ns(message.stamp) + self._pending.append(self._device_health_point(message, health, sample_time_ns)) + self._pending.extend( + self._fault_counter_points(message, health, sample_time_ns) + ) + + @staticmethod + def _generic_point(message: DiagnosticsMessage) -> dict[str, Any]: path = message.node parts = [p for p in path.split("/") if p] @@ -108,19 +144,83 @@ def handle_message(self, message: DiagnosticsMessage) -> None: name = parts[-2] if len(parts) >= 2 else measurement component = "/".join(parts[:-2]) - self._pending.append( - { - "measurement": measurement, - "tags": { - "hw_id": message.hw_id if message.hw_id else "unknown", - "path": path, - "name": name, - "component": component, - }, - "fields": fields, - "time": int(1e9 * time.time()), + return { + "measurement": measurement, + "tags": { + "hw_id": message.hw_id if message.hw_id else "unknown", + "path": path, + "name": name, + "component": component, + }, + "fields": fields, + "time": int(1e9 * time.time()), + } + + @staticmethod + def _device_health_point( + message: DiagnosticsMessage, + health: HealthStatus, + sample_time_ns: int, + ) -> dict[str, Any]: + fields: dict[str, Any] = { + "level": message.level, + "active_fault_count": health.active_fault_count, + } + if message.msg: + fields["message"] = message.msg + + # Preserve optional health metadata when it is scalar. Structured optional + # values remain JSON strings to avoid dynamic nested InfluxDB schemas. + for entry in health.extra_values: + fields[entry.key] = _coerce_health_field(entry.value) + + return { + "measurement": _DEVICE_HEALTH_MEASUREMENT, + "tags": { + "hw_id": message.hw_id, + "path": message.node, + "device_path": health.device_path, + "schema": health.schema_name, + "schema_version": str(health.schema_version), + "boot_id": health.boot_id, + }, + "fields": fields, + "time": sample_time_ns, + } + + @staticmethod + def _fault_counter_points( + message: DiagnosticsMessage, + health: HealthStatus, + sample_time_ns: int, + ) -> list[dict[str, Any]]: + points: list[dict[str, Any]] = [] + for fault in health.faults: + fields: dict[str, Any] = { + "active": fault.active, + "raise_count_total": fault.raise_count_total, } - ) + if fault.last_raised_ns is not None: + fields["last_raised_ns"] = fault.last_raised_ns + if fault.last_cleared_ns is not None: + fields["last_cleared_ns"] = fault.last_cleared_ns + + points.append( + { + "measurement": _FAULT_COUNTER_MEASUREMENT, + "tags": { + "hw_id": message.hw_id, + "path": message.node, + "device_path": health.device_path, + "fault_code": fault.code, + "schema_version": str(health.schema_version), + "boot_id": health.boot_id, + }, + "fields": fields, + "time": sample_time_ns, + } + ) + return points def publish_state(self, states: dict[str, DiagnosticsMessage]) -> None: del states @@ -150,3 +250,14 @@ def close(self) -> None: LOGGER.warning("InfluxDB final flush failed: %s", exc) if self._client is not None: self._client.close() + + +def _coerce_health_field(value: Any) -> Any: + if isinstance(value, (bool, int, float, str)): + return value + try: + import json + + return json.dumps(value, separators=(",", ":"), sort_keys=True) + except (TypeError, ValueError): + return str(value) From ee458489cde789a336ea1ce3b837c7412a1a3b3c Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:48:50 +0200 Subject: [PATCH 03/11] Test health message validation and Influx mapping --- tests/test_health.py | 231 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 tests/test_health.py diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..b69cd11 --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,231 @@ +import logging + +import pytest + +from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticKeyValue, DiagnosticsMessage +from pyxbot2_diagnostics.aggregator.health import ( + HealthMessageValidationError, + is_health_message, + parse_health_message, +) +from pyxbot2_diagnostics.aggregator.sinks.influxdb_sink import InfluxDBSink + + +class FakeWriteApi: + def __init__(self) -> None: + self.calls = [] + + def write(self, *, bucket, org, record): + self.calls.append({"bucket": bucket, "org": org, "record": record}) + + +def _health_msg( + *, + node: str = "/xbot/joint/knee/motor/health_status", + hw_id: str = "SN-1", + stamp: float = 1785614401.25, + values: tuple[DiagnosticKeyValue, ...] | None = None, +) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=1, + node=node, + hw_id=hw_id, + stamp=stamp, + level=2, + msg="OVERCURRENT active", + values=values + or ( + DiagnosticKeyValue("schema.name", "xbot.device_health"), + DiagnosticKeyValue("schema.version", "1"), + DiagnosticKeyValue("device.boot_id", "boot-a"), + DiagnosticKeyValue("faults.active", '["OVERCURRENT"]'), + DiagnosticKeyValue( + "faults.raise_count_total", + '{"OVERCURRENT":4,"ENCODER_CRC":2}', + ), + DiagnosticKeyValue( + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:00Z",' + '"ENCODER_CRC":"2026-08-01T19:00:00+00:00"}', + ), + DiagnosticKeyValue( + "faults.last_cleared", + '{"OVERCURRENT":"2026-08-01T18:00:00Z",' + '"ENCODER_CRC":"2026-08-01T19:00:01Z"}', + ), + DiagnosticKeyValue("thermal.temperature", 72.5), + DiagnosticKeyValue("communication.degraded", False), + ), + ) + + +def _replace_value( + message: DiagnosticsMessage, + key: str, + value, +) -> DiagnosticsMessage: + values = tuple( + DiagnosticKeyValue(entry.key, value if entry.key == key else entry.value) + for entry in message.values + ) + return DiagnosticsMessage( + v=message.v, + node=message.node, + hw_id=message.hw_id, + stamp=message.stamp, + level=message.level, + msg=message.msg, + values=values, + ) + + +def test_parse_valid_health_message() -> None: + health = parse_health_message(_health_msg()) + + assert health.device_path == "/xbot/joint/knee/motor" + assert health.boot_id == "boot-a" + assert health.active_fault_count == 1 + assert [fault.code for fault in health.faults] == ["ENCODER_CRC", "OVERCURRENT"] + assert health.faults[0].active is False + assert health.faults[1].active is True + assert [entry.key for entry in health.extra_values] == [ + "thermal.temperature", + "communication.degraded", + ] + + +def test_accepts_native_json_values_and_health_alias() -> None: + message = _health_msg(node="xbot/motor/health") + message = _replace_value(message, "faults.active", ["OVERCURRENT"]) + message = _replace_value( + message, + "faults.raise_count_total", + {"OVERCURRENT": 4, "ENCODER_CRC": 2}, + ) + message = _replace_value( + message, + "faults.last_raised", + {"OVERCURRENT": 1785614400.0, "ENCODER_CRC": 1785610800.0}, + ) + message = _replace_value( + message, + "faults.last_cleared", + {"OVERCURRENT": 1785607200.0, "ENCODER_CRC": 1785610801.0}, + ) + + assert is_health_message(message) + assert parse_health_message(message).device_path == "/xbot/motor" + + +@pytest.mark.parametrize( + "key,value,match", + [ + ("faults.active", '["OVERCURRENT","OVERCURRENT"]', "duplicates"), + ("faults.raise_count_total", '{"OVERCURRENT":-1}', "non-negative"), + ( + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:00"}', + "timezone", + ), + ], +) +def test_rejects_invalid_fault_payloads(key, value, match) -> None: + with pytest.raises(HealthMessageValidationError, match=match): + parse_health_message(_replace_value(_health_msg(), key, value)) + + +def test_rejects_duplicate_diagnostic_keys() -> None: + original = _health_msg() + message = _health_msg( + values=original.values + (DiagnosticKeyValue("faults.active", "[]"),) + ) + with pytest.raises(HealthMessageValidationError, match="duplicate keys"): + parse_health_message(message) + + +def test_rejects_active_fault_missing_from_counter_map() -> None: + message = _replace_value( + _health_msg(), "faults.raise_count_total", '{"ENCODER_CRC":2}' + ) + with pytest.raises( + HealthMessageValidationError, match="missing referenced fault codes" + ): + parse_health_message(message) + + +def test_rejects_positive_counter_without_last_raise() -> None: + message = _replace_value( + _health_msg(), + "faults.last_raised", + '{"OVERCURRENT":null,"ENCODER_CRC":"2026-08-01T19:00:00Z"}', + ) + with pytest.raises(HealthMessageValidationError, match="positive raise counter"): + parse_health_message(message) + + +def test_requires_health_suffix_and_device_path() -> None: + assert not is_health_message(_health_msg(node="/xbot/motor/temperature")) + with pytest.raises(HealthMessageValidationError, match="must end"): + parse_health_message(_health_msg(node="/xbot/motor/temperature")) + with pytest.raises(HealthMessageValidationError, match="device path"): + parse_health_message(_health_msg(node="/health")) + + +def test_influx_sink_normalizes_health_message() -> None: + fake = FakeWriteApi() + sink = InfluxDBSink( + enabled=True, + url="", + token="", + org="xbot2", + bucket="diagnostics", + write_api=fake, + ) + + sink.handle_message(_health_msg()) + sink._last_flush = 0.0 + sink.publish_state({}) + + points = fake.calls[0]["record"] + assert [point["measurement"] for point in points] == [ + "device_health", + "fault_counter", + "fault_counter", + ] + + health_point = points[0] + assert health_point["tags"]["hw_id"] == "SN-1" + assert health_point["tags"]["device_path"] == "/xbot/joint/knee/motor" + assert health_point["tags"]["boot_id"] == "boot-a" + assert health_point["fields"]["level"] == 2 + assert health_point["fields"]["active_fault_count"] == 1 + assert health_point["fields"]["thermal.temperature"] == 72.5 + assert health_point["fields"]["communication.degraded"] is False + assert health_point["time"] == 1785614401250000000 + + fault_points = {point["tags"]["fault_code"]: point for point in points[1:]} + assert fault_points["OVERCURRENT"]["fields"]["active"] is True + assert fault_points["OVERCURRENT"]["fields"]["raise_count_total"] == 4 + assert fault_points["ENCODER_CRC"]["fields"]["active"] is False + assert fault_points["ENCODER_CRC"]["fields"]["last_cleared_ns"] > 0 + + +def test_influx_sink_omits_invalid_health_message(caplog) -> None: + fake = FakeWriteApi() + sink = InfluxDBSink( + enabled=True, + url="", + token="", + org="xbot2", + bucket="diagnostics", + write_api=fake, + ) + invalid = _replace_value(_health_msg(), "schema.version", "99") + + with caplog.at_level(logging.WARNING): + sink.handle_message(invalid) + sink._last_flush = 0.0 + sink.publish_state({}) + + assert fake.calls == [] + assert "unsupported health schema version" in caplog.text From 4c5928ba161920d7cc9908a2edf48d916af1dfde Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:49:06 +0200 Subject: [PATCH 04/11] Document draft device health schema --- docs/device_health_schema.md | 121 +++++++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 docs/device_health_schema.md diff --git a/docs/device_health_schema.md b/docs/device_health_schema.md new file mode 100644 index 0000000..3243ca5 --- /dev/null +++ b/docs/device_health_schema.md @@ -0,0 +1,121 @@ +# Device health diagnostic schema (draft v1) + +This document describes the provisional schema recognized by the Python diagnostics +aggregator. It is intentionally narrow while the device-health contract is being +reviewed. + +## Identification + +A diagnostic is treated as a device-health message when the final segment of its +`DiagnosticStatus.name` / aggregator `node` is either: + +- `health` +- `health_status` + +The preceding path identifies the logical device. `hardware_id` must contain the +non-empty physical device identifier. + +Example: + +```text +name: /xbot/joint/left_knee/motor/health_status +device path: /xbot/joint/left_knee/motor +hardware_id: SN-0028417 +``` + +## Required key-value fields + +All required values may arrive as JSON strings, as they do in ROS +`diagnostic_msgs/KeyValue`. Native JSON values are also accepted by the ZMQ input. + +| Key | Type | Meaning | +|---|---|---| +| `schema.name` | string | Must equal `xbot.device_health` | +| `schema.version` | integer or integer string | Must equal `1` | +| `device.boot_id` | non-empty string | Counter epoch identifier | +| `faults.active` | array of unique strings | Complete currently active fault set | +| `faults.raise_count_total` | object: fault code to non-negative integer | Monotonic raise count within `device.boot_id` | +| `faults.last_raised` | object: fault code to timestamp or null | Latest raise time | +| `faults.last_cleared` | object: fault code to timestamp or null | Latest clear time | + +Timestamps may be non-negative Unix epoch seconds or timezone-aware ISO-8601 +strings. They are normalized to integer nanoseconds. + +Optional non-contract key-value fields are retained on the `device_health` InfluxDB +point. Scalar values remain scalar fields; structured values are serialized as +compact JSON strings. + +## Consistency rules + +- Diagnostic keys must be unique. +- Every active or timestamped fault code must exist in + `faults.raise_count_total`. +- A positive raise counter requires a non-null last-raise time. +- A zero raise counter requires a null or absent last-raise time. +- An active fault must have a positive raise counter. +- For an active fault, `last_cleared` must precede `last_raised` when both exist. +- For an inactive fault, `last_cleared` must not precede `last_raised` when both + exist. + +Malformed health messages are logged and omitted from InfluxDB health measurements. +Other diagnostic messages continue through the generic InfluxDB path unchanged. + +## InfluxDB measurements + +### `device_health` + +One point is written per valid health snapshot. + +Tags: + +- `hw_id` +- `path` +- `device_path` +- `schema` +- `schema_version` +- `boot_id` + +Fields: + +- `level` +- `active_fault_count` +- `message`, when non-empty +- optional non-contract health values + +The point timestamp is the diagnostic source timestamp. + +### `fault_counter` + +One point is written per known fault code in each valid health snapshot. + +Tags: + +- `hw_id` +- `path` +- `device_path` +- `fault_code` +- `schema_version` +- `boot_id` + +Fields: + +- `active` +- `raise_count_total` +- `last_raised_ns`, when known +- `last_cleared_ns`, when known + +The point timestamp is the diagnostic source timestamp. + +## Decisions still open + +1. Whether the canonical suffix should be only `/health_status`, only `/health`, + or whether both aliases should remain supported. +2. Whether `device.boot_id` is mandatory and whether counters are boot-scoped or + persisted for the lifetime of the device. +3. Whether `boot_id` should be an InfluxDB tag. Keeping it as a tag simplifies + counter-epoch filtering but creates a new series for every device reboot. +4. Whether per-fault timestamps should stay as integer nanosecond fields or be + represented differently for easier Grafana formatting. +5. Whether optional device-specific health values should share the + `device_health` measurement or be written through the existing generic + diagnostic measurement. From 0c262d0e01b2023969160ebd40de140578858eb6 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:58:54 +0200 Subject: [PATCH 05/11] Add fault occurrence Influx measurement --- .../aggregator/sinks/influxdb_sink.py | 123 ++++++++++++++++-- 1 file changed, 109 insertions(+), 14 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index e73623b..aef5070 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -21,26 +21,47 @@ _MEASUREMENT = "robot_diagnostics" _DEVICE_HEALTH_MEASUREMENT = "device_health" _FAULT_COUNTER_MEASUREMENT = "fault_counter" +_FAULT_OCCURRENCE_MEASUREMENT = "fault_occurrence" # Minimum seconds between batch writes to InfluxDB. _FLUSH_INTERVAL_SEC = 1.0 +# (hardware id, status path, fault code) -> (boot id, last total counter) +FaultCounterKey = tuple[str, str, str] +FaultCounterState = tuple[str, int] + class InfluxDBSink: """Write diagnostics to InfluxDB v2. Ordinary diagnostics retain the existing schema. Messages whose path ends in ``/health`` or ``/health_status`` are validated against the device-health - contract and normalized into two measurements: + contract and normalized into three measurements: ``device_health`` - tags: hw_id, path, device_path, schema, schema_version, boot_id - fields: level, active_fault_count, message, plus non-contract scalar values + tags: hw_id, path, device_path, schema, schema_version + fields: level, active_fault_count, boot_id, message, and optional values ``fault_counter`` - one point per fault code - tags: hw_id, path, device_path, fault_code, schema_version, boot_id - fields: active, raise_count_total, last_raised_ns, last_cleared_ns + one point per fault code and health snapshot + tags: hw_id, path, device_path, fault_code, schema_version + fields: active, raise_count_total, boot_id, last_raised_ms, + last_cleared_ms + + ``fault_occurrence`` + sparse point emitted when a cumulative raise counter increases + tags: hw_id, path, device_path, fault_code, schema_version + fields: occurrences, counter_before, counter_after, boot_id, + last_raised_ms + + ``boot_id`` is deliberately a field rather than a tag to avoid creating a new + series on every device reboot. Transition timestamps are stored as Unix epoch + milliseconds so Grafana can format them directly as date/time fields. InfluxDB + point timestamps remain nanoseconds. + + The first sample for a source/fault or a new boot establishes a baseline and + does not emit an occurrence. A counter decrease within the same boot is logged + and also establishes a new baseline. Invalid health messages are logged and omitted from InfluxDB rather than being written as generic diagnostics with opaque JSON fields. @@ -63,6 +84,7 @@ def __init__( self._write_api = write_api self._pending: list[dict[str, Any]] = [] self._last_flush = 0.0 + self._fault_counter_state: dict[FaultCounterKey, FaultCounterState] = {} if not enabled: return @@ -109,9 +131,8 @@ def _handle_health_message(self, message: DiagnosticsMessage) -> None: sample_time_ns = timestamp_seconds_to_ns(message.stamp) self._pending.append(self._device_health_point(message, health, sample_time_ns)) - self._pending.extend( - self._fault_counter_points(message, health, sample_time_ns) - ) + self._pending.extend(self._fault_counter_points(message, health, sample_time_ns)) + self._pending.extend(self._fault_occurrence_points(message, health, sample_time_ns)) @staticmethod def _generic_point(message: DiagnosticsMessage) -> dict[str, Any]: @@ -165,6 +186,7 @@ def _device_health_point( fields: dict[str, Any] = { "level": message.level, "active_fault_count": health.active_fault_count, + "boot_id": health.boot_id, } if message.msg: fields["message"] = message.msg @@ -182,7 +204,6 @@ def _device_health_point( "device_path": health.device_path, "schema": health.schema_name, "schema_version": str(health.schema_version), - "boot_id": health.boot_id, }, "fields": fields, "time": sample_time_ns, @@ -199,11 +220,12 @@ def _fault_counter_points( fields: dict[str, Any] = { "active": fault.active, "raise_count_total": fault.raise_count_total, + "boot_id": health.boot_id, } if fault.last_raised_ns is not None: - fields["last_raised_ns"] = fault.last_raised_ns + fields["last_raised_ms"] = _timestamp_ns_to_ms(fault.last_raised_ns) if fault.last_cleared_ns is not None: - fields["last_cleared_ns"] = fault.last_cleared_ns + fields["last_cleared_ms"] = _timestamp_ns_to_ms(fault.last_cleared_ns) points.append( { @@ -214,7 +236,76 @@ def _fault_counter_points( "device_path": health.device_path, "fault_code": fault.code, "schema_version": str(health.schema_version), - "boot_id": health.boot_id, + }, + "fields": fields, + "time": sample_time_ns, + } + ) + return points + + def _fault_occurrence_points( + self, + message: DiagnosticsMessage, + health: HealthStatus, + sample_time_ns: int, + ) -> list[dict[str, Any]]: + points: list[dict[str, Any]] = [] + for fault in health.faults: + key: FaultCounterKey = (message.hw_id, message.node, fault.code) + previous = self._fault_counter_state.get(key) + self._fault_counter_state[key] = ( + health.boot_id, + fault.raise_count_total, + ) + + if previous is None: + continue + + previous_boot_id, previous_total = previous + if previous_boot_id != health.boot_id: + LOGGER.info( + "Fault counter epoch changed for %s %s (%s -> %s); " + "establishing new baseline", + message.node, + fault.code, + previous_boot_id, + health.boot_id, + ) + continue + + if fault.raise_count_total < previous_total: + LOGGER.warning( + "Fault counter decreased within boot for %s %s: %d -> %d; " + "establishing new baseline", + message.node, + fault.code, + previous_total, + fault.raise_count_total, + ) + continue + + occurrences = fault.raise_count_total - previous_total + if occurrences == 0: + continue + + fields: dict[str, Any] = { + "occurrences": occurrences, + "counter_before": previous_total, + "counter_after": fault.raise_count_total, + "boot_id": health.boot_id, + } + if fault.last_raised_ns is not None: + fields["last_raised_ms"] = _timestamp_ns_to_ms(fault.last_raised_ns) + + points.append( + { + "measurement": _FAULT_OCCURRENCE_MEASUREMENT, + "tags": { + "hw_id": message.hw_id, + "path": message.node, + "device_path": health.device_path, + "fault_code": fault.code, + "schema_version": str(health.schema_version), }, "fields": fields, "time": sample_time_ns, @@ -242,7 +333,7 @@ def _flush(self) -> None: LOGGER.warning("InfluxDB write failed (%d points dropped): %s", len(points), exc) def close(self) -> None: - # Final flush on shutdown — ignore the rate limit. + # Final flush on shutdown - ignore the rate limit. if self._pending and self._enabled and self._write_api is not None: try: self._write_api.write(bucket=self._bucket, org=self._org, record=self._pending) @@ -252,6 +343,10 @@ def close(self) -> None: self._client.close() +def _timestamp_ns_to_ms(value: int) -> int: + return value // 1_000_000 + + def _coerce_health_field(value: Any) -> Any: if isinstance(value, (bool, int, float, str)): return value From c514badcf2129e1ddd404f6e3e51150e1f23deae Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:59:42 +0200 Subject: [PATCH 06/11] Test fault occurrence Influx mapping --- tests/test_health.py | 121 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 105 insertions(+), 16 deletions(-) diff --git a/tests/test_health.py b/tests/test_health.py index b69cd11..a0b16fb 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -171,9 +171,8 @@ def test_requires_health_suffix_and_device_path() -> None: parse_health_message(_health_msg(node="/health")) -def test_influx_sink_normalizes_health_message() -> None: - fake = FakeWriteApi() - sink = InfluxDBSink( +def _sink(fake: FakeWriteApi) -> InfluxDBSink: + return InfluxDBSink( enabled=True, url="", token="", @@ -182,10 +181,19 @@ def test_influx_sink_normalizes_health_message() -> None: write_api=fake, ) - sink.handle_message(_health_msg()) + +def _flush(sink: InfluxDBSink) -> None: sink._last_flush = 0.0 sink.publish_state({}) + +def test_influx_sink_normalizes_health_message() -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg()) + _flush(sink) + points = fake.calls[0]["record"] assert [point["measurement"] for point in points] == [ "device_health", @@ -196,7 +204,8 @@ def test_influx_sink_normalizes_health_message() -> None: health_point = points[0] assert health_point["tags"]["hw_id"] == "SN-1" assert health_point["tags"]["device_path"] == "/xbot/joint/knee/motor" - assert health_point["tags"]["boot_id"] == "boot-a" + assert "boot_id" not in health_point["tags"] + assert health_point["fields"]["boot_id"] == "boot-a" assert health_point["fields"]["level"] == 2 assert health_point["fields"]["active_fault_count"] == 1 assert health_point["fields"]["thermal.temperature"] == 72.5 @@ -206,26 +215,106 @@ def test_influx_sink_normalizes_health_message() -> None: fault_points = {point["tags"]["fault_code"]: point for point in points[1:]} assert fault_points["OVERCURRENT"]["fields"]["active"] is True assert fault_points["OVERCURRENT"]["fields"]["raise_count_total"] == 4 + assert fault_points["OVERCURRENT"]["fields"]["boot_id"] == "boot-a" + assert "boot_id" not in fault_points["OVERCURRENT"]["tags"] assert fault_points["ENCODER_CRC"]["fields"]["active"] is False - assert fault_points["ENCODER_CRC"]["fields"]["last_cleared_ns"] > 0 + assert fault_points["ENCODER_CRC"]["fields"]["last_cleared_ms"] > 0 -def test_influx_sink_omits_invalid_health_message(caplog) -> None: +def test_influx_sink_emits_fault_occurrence_from_counter_delta() -> None: fake = FakeWriteApi() - sink = InfluxDBSink( - enabled=True, - url="", - token="", - org="xbot2", - bucket="diagnostics", - write_api=fake, + sink = _sink(fake) + + sink.handle_message(_health_msg(stamp=1785614401.0)) + updated = _replace_value( + _health_msg(stamp=1785614402.0), + "faults.raise_count_total", + '{"OVERCURRENT":7,"ENCODER_CRC":2}', ) + updated = _replace_value( + updated, + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:01.500Z",' + '"ENCODER_CRC":"2026-08-01T19:00:00Z"}', + ) + sink.handle_message(updated) + _flush(sink) + + points = fake.calls[0]["record"] + occurrences = [ + point for point in points if point["measurement"] == "fault_occurrence" + ] + assert len(occurrences) == 1 + point = occurrences[0] + assert point["tags"]["fault_code"] == "OVERCURRENT" + assert "boot_id" not in point["tags"] + assert point["fields"]["boot_id"] == "boot-a" + assert point["fields"]["occurrences"] == 3 + assert point["fields"]["counter_before"] == 4 + assert point["fields"]["counter_after"] == 7 + assert point["fields"]["last_raised_ms"] == 1785614401500 + assert point["time"] == 1785614402000000000 + + +def test_influx_sink_reboot_establishes_new_counter_baseline() -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg(stamp=1785614401.0)) + restarted = _replace_value(_health_msg(stamp=1785614402.0), "device.boot_id", "boot-b") + restarted = _replace_value( + restarted, + "faults.raise_count_total", + '{"OVERCURRENT":1,"ENCODER_CRC":0}', + ) + restarted = _replace_value( + restarted, + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:01Z","ENCODER_CRC":null}', + ) + restarted = _replace_value( + restarted, + "faults.last_cleared", + '{"OVERCURRENT":null,"ENCODER_CRC":null}', + ) + sink.handle_message(restarted) + _flush(sink) + + assert not any( + point["measurement"] == "fault_occurrence" + for point in fake.calls[0]["record"] + ) + + +def test_influx_sink_counter_decrease_establishes_new_baseline(caplog) -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg(stamp=1785614401.0)) + decreased = _replace_value( + _health_msg(stamp=1785614402.0), + "faults.raise_count_total", + '{"OVERCURRENT":3,"ENCODER_CRC":2}', + ) + with caplog.at_level(logging.WARNING): + sink.handle_message(decreased) + _flush(sink) + + assert not any( + point["measurement"] == "fault_occurrence" + for point in fake.calls[0]["record"] + ) + assert "Fault counter decreased within boot" in caplog.text + + +def test_influx_sink_omits_invalid_health_message(caplog) -> None: + fake = FakeWriteApi() + sink = _sink(fake) invalid = _replace_value(_health_msg(), "schema.version", "99") with caplog.at_level(logging.WARNING): sink.handle_message(invalid) - sink._last_flush = 0.0 - sink.publish_state({}) + _flush(sink) assert fake.calls == [] assert "unsupported health schema version" in caplog.text From 87052bff206815218923cf0d84746f542c746378 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 22:59:56 +0200 Subject: [PATCH 07/11] Document fault occurrence measurement --- docs/device_health_schema.md | 65 ++++++++++++++++++++++++------------ 1 file changed, 44 insertions(+), 21 deletions(-) diff --git a/docs/device_health_schema.md b/docs/device_health_schema.md index 3243ca5..fb87264 100644 --- a/docs/device_health_schema.md +++ b/docs/device_health_schema.md @@ -1,17 +1,12 @@ # Device health diagnostic schema (draft v1) This document describes the provisional schema recognized by the Python diagnostics -aggregator. It is intentionally narrow while the device-health contract is being -reviewed. +aggregator. ## Identification A diagnostic is treated as a device-health message when the final segment of its -`DiagnosticStatus.name` / aggregator `node` is either: - -- `health` -- `health_status` - +`DiagnosticStatus.name` / aggregator `node` is either `health` or `health_status`. The preceding path identifies the logical device. `hardware_id` must contain the non-empty physical device identifier. @@ -39,7 +34,8 @@ All required values may arrive as JSON strings, as they do in ROS | `faults.last_cleared` | object: fault code to timestamp or null | Latest clear time | Timestamps may be non-negative Unix epoch seconds or timezone-aware ISO-8601 -strings. They are normalized to integer nanoseconds. +strings. Internally they are normalized to nanoseconds. InfluxDB fields exposed to +Grafana use Unix epoch milliseconds (`last_raised_ms`, `last_cleared_ms`). Optional non-contract key-value fields are retained on the `device_health` InfluxDB point. Scalar values remain scalar fields; structured values are serialized as @@ -73,12 +69,12 @@ Tags: - `device_path` - `schema` - `schema_version` -- `boot_id` Fields: - `level` - `active_fault_count` +- `boot_id` - `message`, when non-empty - optional non-contract health values @@ -95,27 +91,54 @@ Tags: - `device_path` - `fault_code` - `schema_version` -- `boot_id` Fields: - `active` - `raise_count_total` -- `last_raised_ns`, when known -- `last_cleared_ns`, when known +- `boot_id` +- `last_raised_ms`, when known +- `last_cleared_ms`, when known The point timestamp is the diagnostic source timestamp. +### `fault_occurrence` + +A sparse point is written when a cumulative raise counter increases. + +Tags: + +- `hw_id` +- `path` +- `device_path` +- `fault_code` +- `schema_version` + +Fields: + +- `occurrences`: counter delta since the previous received sample +- `counter_before` +- `counter_after` +- `boot_id` +- `last_raised_ms`, when known + +The point timestamp is the current health diagnostic source timestamp. When the +counter delta is greater than one, the exact timestamps of all raises are not known; +`last_raised_ms` records the latest raise supplied by the producer. + +The first sample for a fault establishes a baseline. A changed `boot_id` also +establishes a new baseline. A counter decrease within the same boot is logged as a +warning and establishes a new baseline. None of these baseline cases emits a +`fault_occurrence` point. + +`boot_id` is stored as an Influx field, not a tag, to avoid creating a new series on +every reboot. + ## Decisions still open 1. Whether the canonical suffix should be only `/health_status`, only `/health`, or whether both aliases should remain supported. -2. Whether `device.boot_id` is mandatory and whether counters are boot-scoped or - persisted for the lifetime of the device. -3. Whether `boot_id` should be an InfluxDB tag. Keeping it as a tag simplifies - counter-epoch filtering but creates a new series for every device reboot. -4. Whether per-fault timestamps should stay as integer nanosecond fields or be - represented differently for easier Grafana formatting. -5. Whether optional device-specific health values should share the - `device_health` measurement or be written through the existing generic - diagnostic measurement. +2. Whether `device.boot_id` remains mandatory and whether counters are boot-scoped + or persisted for the lifetime of the device. +3. Whether optional device-specific health values should share the + `device_health` measurement or use the generic diagnostic measurement. From 6e73b6d92915e064cab09d59d0820820f41e5dbf Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 23:12:44 +0200 Subject: [PATCH 08/11] Simplify health contract identifiers --- .../pyxbot2_diagnostics/aggregator/health.py | 38 ------------------- 1 file changed, 38 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/health.py b/python/src/pyxbot2_diagnostics/aggregator/health.py index 3518b36..d9b3918 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/health.py +++ b/python/src/pyxbot2_diagnostics/aggregator/health.py @@ -11,12 +11,8 @@ from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticKeyValue, DiagnosticsMessage -HEALTH_SCHEMA_NAME = "xbot.device_health" -HEALTH_SCHEMA_VERSION = 1 HEALTH_STATUS_SUFFIXES = frozenset({"health", "health_status"}) -_SCHEMA_NAME_KEY = "schema.name" -_SCHEMA_VERSION_KEY = "schema.version" _BOOT_ID_KEY = "device.boot_id" _ACTIVE_KEY = "faults.active" _RAISE_COUNT_KEY = "faults.raise_count_total" @@ -25,8 +21,6 @@ REQUIRED_HEALTH_KEYS = frozenset( { - _SCHEMA_NAME_KEY, - _SCHEMA_VERSION_KEY, _BOOT_ID_KEY, _ACTIVE_KEY, _RAISE_COUNT_KEY, @@ -56,8 +50,6 @@ class HealthStatus: """Normalized device health snapshot.""" device_path: str - schema_name: str - schema_version: int boot_id: str faults: tuple[FaultHealthRecord, ...] extra_values: tuple[DiagnosticKeyValue, ...] @@ -96,19 +88,6 @@ def parse_health_message(message: DiagnosticsMessage) -> HealthStatus: "health status is missing required keys: " + ", ".join(missing) ) - schema_name = _require_non_empty_string(values[_SCHEMA_NAME_KEY], _SCHEMA_NAME_KEY) - if schema_name != HEALTH_SCHEMA_NAME: - raise HealthMessageValidationError( - f"{_SCHEMA_NAME_KEY} must be '{HEALTH_SCHEMA_NAME}', got '{schema_name}'" - ) - - schema_version = _parse_schema_version(values[_SCHEMA_VERSION_KEY]) - if schema_version != HEALTH_SCHEMA_VERSION: - raise HealthMessageValidationError( - f"unsupported health schema version {schema_version}; " - f"expected {HEALTH_SCHEMA_VERSION}" - ) - boot_id = _require_non_empty_string(values[_BOOT_ID_KEY], _BOOT_ID_KEY) active_codes = _parse_active_faults(values[_ACTIVE_KEY]) counts = _parse_counter_map(values[_RAISE_COUNT_KEY]) @@ -169,8 +148,6 @@ def parse_health_message(message: DiagnosticsMessage) -> HealthStatus: device_path = "/" + "/".join(parts[:-1]) return HealthStatus( device_path=device_path, - schema_name=schema_name, - schema_version=schema_version, boot_id=boot_id, faults=tuple(faults), extra_values=extra_values, @@ -213,21 +190,6 @@ def _require_non_empty_string(value: Any, key: str) -> str: return value.strip() -def _parse_schema_version(value: Any) -> int: - if isinstance(value, bool): - raise HealthMessageValidationError(f"{_SCHEMA_VERSION_KEY} must be an integer") - if isinstance(value, int): - return value - if isinstance(value, str): - try: - return int(value.strip()) - except ValueError as exc: - raise HealthMessageValidationError( - f"{_SCHEMA_VERSION_KEY} must be an integer" - ) from exc - raise HealthMessageValidationError(f"{_SCHEMA_VERSION_KEY} must be an integer") - - def _decode_json(value: Any, key: str) -> Any: if isinstance(value, str): try: From 49fd9d779375a57790dbf6de9cb56ab926939fd4 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 23:13:10 +0200 Subject: [PATCH 09/11] Use lean health Influx tags --- .../aggregator/sinks/influxdb_sink.py | 69 +++++-------------- 1 file changed, 17 insertions(+), 52 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index aef5070..d58dee0 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -26,7 +26,7 @@ # Minimum seconds between batch writes to InfluxDB. _FLUSH_INTERVAL_SEC = 1.0 -# (hardware id, status path, fault code) -> (boot id, last total counter) +# (hardware id, device path, fault code) -> (boot id, last total counter) FaultCounterKey = tuple[str, str, str] FaultCounterState = tuple[str, int] @@ -34,37 +34,31 @@ class InfluxDBSink: """Write diagnostics to InfluxDB v2. - Ordinary diagnostics retain the existing schema. Messages whose path ends in - ``/health`` or ``/health_status`` are validated against the device-health - contract and normalized into three measurements: + Ordinary diagnostics retain the existing generic representation. Messages whose + path ends in ``/health`` or ``/health_status`` are validated and normalized into + three measurements with a deliberately small tag set: ``device_health`` - tags: hw_id, path, device_path, schema, schema_version - fields: level, active_fault_count, boot_id, message, and optional values + tags: hw_id, device_path + fields: level, active_fault_count, boot_id, message, optional values ``fault_counter`` - one point per fault code and health snapshot - tags: hw_id, path, device_path, fault_code, schema_version + tags: hw_id, device_path, fault_code fields: active, raise_count_total, boot_id, last_raised_ms, last_cleared_ms ``fault_occurrence`` - sparse point emitted when a cumulative raise counter increases - tags: hw_id, path, device_path, fault_code, schema_version + tags: hw_id, device_path, fault_code fields: occurrences, counter_before, counter_after, boot_id, last_raised_ms - ``boot_id`` is deliberately a field rather than a tag to avoid creating a new - series on every device reboot. Transition timestamps are stored as Unix epoch - milliseconds so Grafana can format them directly as date/time fields. InfluxDB - point timestamps remain nanoseconds. + ``boot_id`` is a field rather than a tag to avoid creating a new series on every + reboot. Transition timestamps are Unix epoch milliseconds so Grafana can format + them directly as date/time fields. InfluxDB point timestamps remain nanoseconds. The first sample for a source/fault or a new boot establishes a baseline and does not emit an occurrence. A counter decrease within the same boot is logged and also establishes a new baseline. - - Invalid health messages are logged and omitted from InfluxDB rather than being - written as generic diagnostics with opaque JSON fields. """ def __init__( @@ -107,8 +101,6 @@ def __init__( return self._client = InfluxDBClient(url=url, token=token, org=org) - # SYNCHRONOUS so errors surface immediately rather than being silently - # dropped by the async batch queue. self._write_api = self._client.write_api(write_options=SYNCHRONOUS) LOGGER.info("InfluxDB sink enabled: url=%s bucket=%s org=%s", url, bucket, org) @@ -138,10 +130,8 @@ def _handle_health_message(self, message: DiagnosticsMessage) -> None: def _generic_point(message: DiagnosticsMessage) -> dict[str, Any]: path = message.node parts = [p for p in path.split("/") if p] - fields: dict[str, Any] = {"level": message.level} - # Coerce each kv-value to float; fall back to string for non-numeric ones. for kv in message.values: try: fields[kv.key] = float(kv.value) @@ -151,16 +141,6 @@ def _generic_point(message: DiagnosticsMessage) -> dict[str, Any]: if message.msg: fields["message"] = message.msg - """ - node schema is defined as follows: - // - - example: /xbot/joint/knee_pitch_1/pos_ref --> - component = /xbot/joint - name = knee_pitch_1 - measurement = pos_ref - """ - measurement = parts[-1] if parts else "unknown" name = parts[-2] if len(parts) >= 2 else measurement component = "/".join(parts[:-2]) @@ -191,8 +171,6 @@ def _device_health_point( if message.msg: fields["message"] = message.msg - # Preserve optional health metadata when it is scalar. Structured optional - # values remain JSON strings to avoid dynamic nested InfluxDB schemas. for entry in health.extra_values: fields[entry.key] = _coerce_health_field(entry.value) @@ -200,10 +178,7 @@ def _device_health_point( "measurement": _DEVICE_HEALTH_MEASUREMENT, "tags": { "hw_id": message.hw_id, - "path": message.node, "device_path": health.device_path, - "schema": health.schema_name, - "schema_version": str(health.schema_version), }, "fields": fields, "time": sample_time_ns, @@ -232,10 +207,8 @@ def _fault_counter_points( "measurement": _FAULT_COUNTER_MEASUREMENT, "tags": { "hw_id": message.hw_id, - "path": message.node, "device_path": health.device_path, "fault_code": fault.code, - "schema_version": str(health.schema_version), }, "fields": fields, "time": sample_time_ns, @@ -251,12 +224,9 @@ def _fault_occurrence_points( ) -> list[dict[str, Any]]: points: list[dict[str, Any]] = [] for fault in health.faults: - key: FaultCounterKey = (message.hw_id, message.node, fault.code) + key: FaultCounterKey = (message.hw_id, health.device_path, fault.code) previous = self._fault_counter_state.get(key) - self._fault_counter_state[key] = ( - health.boot_id, - fault.raise_count_total, - ) + self._fault_counter_state[key] = (health.boot_id, fault.raise_count_total) if previous is None: continue @@ -264,9 +234,8 @@ def _fault_occurrence_points( previous_boot_id, previous_total = previous if previous_boot_id != health.boot_id: LOGGER.info( - "Fault counter epoch changed for %s %s (%s -> %s); " - "establishing new baseline", - message.node, + "Fault counter epoch changed for %s %s (%s -> %s); establishing new baseline", + health.device_path, fault.code, previous_boot_id, health.boot_id, @@ -275,9 +244,8 @@ def _fault_occurrence_points( if fault.raise_count_total < previous_total: LOGGER.warning( - "Fault counter decreased within boot for %s %s: %d -> %d; " - "establishing new baseline", - message.node, + "Fault counter decreased within boot for %s %s: %d -> %d; establishing new baseline", + health.device_path, fault.code, previous_total, fault.raise_count_total, @@ -302,10 +270,8 @@ def _fault_occurrence_points( "measurement": _FAULT_OCCURRENCE_MEASUREMENT, "tags": { "hw_id": message.hw_id, - "path": message.node, "device_path": health.device_path, "fault_code": fault.code, - "schema_version": str(health.schema_version), }, "fields": fields, "time": sample_time_ns, @@ -333,7 +299,6 @@ def _flush(self) -> None: LOGGER.warning("InfluxDB write failed (%d points dropped): %s", len(points), exc) def close(self) -> None: - # Final flush on shutdown - ignore the rate limit. if self._pending and self._enabled and self._write_api is not None: try: self._write_api.write(bucket=self._bucket, org=self._org, record=self._pending) From a7708f24984094ef59160d2c1e4fe8655323dd71 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 23:13:35 +0200 Subject: [PATCH 10/11] Update health tests for lean contract --- tests/test_health.py | 54 ++++++++++++++++++++++++++++++-------------- 1 file changed, 37 insertions(+), 17 deletions(-) diff --git a/tests/test_health.py b/tests/test_health.py index a0b16fb..06a1f0e 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -35,8 +35,6 @@ def _health_msg( msg="OVERCURRENT active", values=values or ( - DiagnosticKeyValue("schema.name", "xbot.device_health"), - DiagnosticKeyValue("schema.version", "1"), DiagnosticKeyValue("device.boot_id", "boot-a"), DiagnosticKeyValue("faults.active", '["OVERCURRENT"]'), DiagnosticKeyValue( @@ -59,11 +57,7 @@ def _health_msg( ) -def _replace_value( - message: DiagnosticsMessage, - key: str, - value, -) -> DiagnosticsMessage: +def _replace_value(message: DiagnosticsMessage, key: str, value) -> DiagnosticsMessage: values = tuple( DiagnosticKeyValue(entry.key, value if entry.key == key else entry.value) for entry in message.values @@ -79,6 +73,18 @@ def _replace_value( ) +def _remove_value(message: DiagnosticsMessage, key: str) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=message.v, + node=message.node, + hw_id=message.hw_id, + stamp=message.stamp, + level=message.level, + msg=message.msg, + values=tuple(entry for entry in message.values if entry.key != key), + ) + + def test_parse_valid_health_message() -> None: health = parse_health_message(_health_msg()) @@ -143,6 +149,11 @@ def test_rejects_duplicate_diagnostic_keys() -> None: parse_health_message(message) +def test_rejects_missing_required_key() -> None: + with pytest.raises(HealthMessageValidationError, match="device.boot_id"): + parse_health_message(_remove_value(_health_msg(), "device.boot_id")) + + def test_rejects_active_fault_missing_from_counter_map() -> None: message = _replace_value( _health_msg(), "faults.raise_count_total", '{"ENCODER_CRC":2}' @@ -202,9 +213,10 @@ def test_influx_sink_normalizes_health_message() -> None: ] health_point = points[0] - assert health_point["tags"]["hw_id"] == "SN-1" - assert health_point["tags"]["device_path"] == "/xbot/joint/knee/motor" - assert "boot_id" not in health_point["tags"] + assert health_point["tags"] == { + "hw_id": "SN-1", + "device_path": "/xbot/joint/knee/motor", + } assert health_point["fields"]["boot_id"] == "boot-a" assert health_point["fields"]["level"] == 2 assert health_point["fields"]["active_fault_count"] == 1 @@ -213,10 +225,14 @@ def test_influx_sink_normalizes_health_message() -> None: assert health_point["time"] == 1785614401250000000 fault_points = {point["tags"]["fault_code"]: point for point in points[1:]} + assert fault_points["OVERCURRENT"]["tags"] == { + "hw_id": "SN-1", + "device_path": "/xbot/joint/knee/motor", + "fault_code": "OVERCURRENT", + } assert fault_points["OVERCURRENT"]["fields"]["active"] is True assert fault_points["OVERCURRENT"]["fields"]["raise_count_total"] == 4 assert fault_points["OVERCURRENT"]["fields"]["boot_id"] == "boot-a" - assert "boot_id" not in fault_points["OVERCURRENT"]["tags"] assert fault_points["ENCODER_CRC"]["fields"]["active"] is False assert fault_points["ENCODER_CRC"]["fields"]["last_cleared_ms"] > 0 @@ -240,14 +256,18 @@ def test_influx_sink_emits_fault_occurrence_from_counter_delta() -> None: sink.handle_message(updated) _flush(sink) - points = fake.calls[0]["record"] occurrences = [ - point for point in points if point["measurement"] == "fault_occurrence" + point + for point in fake.calls[0]["record"] + if point["measurement"] == "fault_occurrence" ] assert len(occurrences) == 1 point = occurrences[0] - assert point["tags"]["fault_code"] == "OVERCURRENT" - assert "boot_id" not in point["tags"] + assert point["tags"] == { + "hw_id": "SN-1", + "device_path": "/xbot/joint/knee/motor", + "fault_code": "OVERCURRENT", + } assert point["fields"]["boot_id"] == "boot-a" assert point["fields"]["occurrences"] == 3 assert point["fields"]["counter_before"] == 4 @@ -310,11 +330,11 @@ def test_influx_sink_counter_decrease_establishes_new_baseline(caplog) -> None: def test_influx_sink_omits_invalid_health_message(caplog) -> None: fake = FakeWriteApi() sink = _sink(fake) - invalid = _replace_value(_health_msg(), "schema.version", "99") + invalid = _remove_value(_health_msg(), "faults.active") with caplog.at_level(logging.WARNING): sink.handle_message(invalid) _flush(sink) assert fake.calls == [] - assert "unsupported health schema version" in caplog.text + assert "missing required keys" in caplog.text From 9b10c8603bda561fed1d4b77a601e6429e2a8674 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Sat, 1 Aug 2026 23:13:48 +0200 Subject: [PATCH 11/11] Remove schema metadata from health docs --- docs/device_health_schema.md | 30 ++++++++---------------------- 1 file changed, 8 insertions(+), 22 deletions(-) diff --git a/docs/device_health_schema.md b/docs/device_health_schema.md index fb87264..4c5881b 100644 --- a/docs/device_health_schema.md +++ b/docs/device_health_schema.md @@ -1,9 +1,4 @@ -# Device health diagnostic schema (draft v1) - -This document describes the provisional schema recognized by the Python diagnostics -aggregator. - -## Identification +# Device health diagnostic contract A diagnostic is treated as a device-health message when the final segment of its `DiagnosticStatus.name` / aggregator `node` is either `health` or `health_status`. @@ -20,13 +15,11 @@ hardware_id: SN-0028417 ## Required key-value fields -All required values may arrive as JSON strings, as they do in ROS +Values may arrive as JSON strings, as they do in ROS `diagnostic_msgs/KeyValue`. Native JSON values are also accepted by the ZMQ input. | Key | Type | Meaning | |---|---|---| -| `schema.name` | string | Must equal `xbot.device_health` | -| `schema.version` | integer or integer string | Must equal `1` | | `device.boot_id` | non-empty string | Counter epoch identifier | | `faults.active` | array of unique strings | Complete currently active fault set | | `faults.raise_count_total` | object: fault code to non-negative integer | Monotonic raise count within `device.boot_id` | @@ -37,11 +30,11 @@ Timestamps may be non-negative Unix epoch seconds or timezone-aware ISO-8601 strings. Internally they are normalized to nanoseconds. InfluxDB fields exposed to Grafana use Unix epoch milliseconds (`last_raised_ms`, `last_cleared_ms`). -Optional non-contract key-value fields are retained on the `device_health` InfluxDB -point. Scalar values remain scalar fields; structured values are serialized as -compact JSON strings. +Optional device-specific key-value fields are retained on the `device_health` +InfluxDB point. Scalar values remain scalar fields; structured values are serialized +as compact JSON strings. -## Consistency rules +## Validation rules - Diagnostic keys must be unique. - Every active or timestamped fault code must exist in @@ -65,10 +58,7 @@ One point is written per valid health snapshot. Tags: - `hw_id` -- `path` - `device_path` -- `schema` -- `schema_version` Fields: @@ -76,7 +66,7 @@ Fields: - `active_fault_count` - `boot_id` - `message`, when non-empty -- optional non-contract health values +- optional device-specific health values The point timestamp is the diagnostic source timestamp. @@ -87,10 +77,8 @@ One point is written per known fault code in each valid health snapshot. Tags: - `hw_id` -- `path` - `device_path` - `fault_code` -- `schema_version` Fields: @@ -109,10 +97,8 @@ A sparse point is written when a cumulative raise counter increases. Tags: - `hw_id` -- `path` - `device_path` - `fault_code` -- `schema_version` Fields: @@ -134,7 +120,7 @@ warning and establishes a new baseline. None of these baseline cases emits a `boot_id` is stored as an Influx field, not a tag, to avoid creating a new series on every reboot. -## Decisions still open +## Open decisions 1. Whether the canonical suffix should be only `/health_status`, only `/health`, or whether both aliases should remain supported.