Telemetry Service

Implementation reference for backend/telemetry. For its architectural role, see Telemetry Service Architecture; for the domain model, see Telemetry Ingestion.

A Rust service on Axum, backed by Postgres + TimescaleDB. It ingests readings from constrained hardware, persists them, publishes them for live distribution, manages per-building and per-room thresholds, and forwards commands to physical devices. Its design answers two pressures: adding a metric must not touch the core, and no I/O technology may reach the logic that decides what a reading means.

Source Layout

Configuration

Adding a Metric

Three steps, none of which touch the kernel.

1. Declare it. A new file in src/plugins/:

static DESCRIPTOR: MetricDescriptor = MetricDescriptor {
    key: "humidity",
    value_field: "humidity",
    label: "Humidity",
    interface_name: "IHumidity",
    unit: Some("%"),
    fields: &[
        FieldSpec { name: "buildingId", kind: FieldKind::NonEmptyString, required: true },
        FieldSpec { name: "roomId",     kind: FieldKind::NonEmptyString, required: true },
        FieldSpec { name: "timestamp",  kind: FieldKind::Finite,         required: true },
        FieldSpec { name: "humidity",   kind: FieldKind::Finite,         required: true },
    ],
};

static BOUNDS: &[BoundSpec] = &[
    BoundSpec { key: "maxHumidity", field: "humidity", label: "Humidity", unit: Some("%"), direction: BoundDirection::Above },
];

A bound names the payload field it compares — not the reading’s value, since one metric can bound several fields — plus the label and unit its alert carries. One alert per breached field.

2. Implement SensorPlugin. validate is check_fields plus reading the value; actions() defaults to &[] and is only overridden for metrics with an actuator.

3. Register it in plugins::all(), which main.rs hands to PluginRegistry::new. A plugin with bounds also joins telemetry_schema::ALERTABLE_METRICS — a test fails until it does — which gives its breaches a delivery path in notification.

GET /contracts, POST /ingest, threshold validation and breach evaluation all pick it up from the declaration. There is no switch statement to extend.

GET /contracts serialises that declaration into ServiceMetricsContract from the shared schemas/telemetry-schema crate — the same types dashboard parses, so the two cannot drift.

Request Lifecycle

Ingestion — POST /ingest, device-facing so no user JWT and still exempt from the edge gate, but not open: the caller signs the request itself. The route takes a batch, always — a lone device sends one reading in the array, so there is no second route and no second code path.

0. X-Signature must equal HMAC-SHA256(TELEMETRY_INGEST_SECRET, raw_body) in lowercase hex — missing, malformed or mismatched is 401 before any parsing, persistence or fanout. Bodies over 1 MiB are 413.
1. Envelope {buildingId, readings[]}, capped at 500 readings. Each reading carries its own type, which selects the plugin and is stripped; the envelope supplies buildingId. A reading naming a different building is rejected — a batch maps to exactly one fan-out channel.
2. Every reading is validated before anything is written. All-or-nothing: one failure persists nothing, publishes nothing, and one 422 names every offending field, each reading’s prefixed with its index (readings[3]: ...). An unknown metric is one of those errors, not a 404 — within a batch it is a bad element of a payload, not an unroutable request.
3. One bulk insert, with one threshold lookup for the whole tick running alongside it.
4. Thresholds resolve room-first, then building; each breached field publishes one AlertEvent, carrying the building and room names from registration, to the alerts topic, keyed buildingId:roomId — but only once the write has committed.
5. One telemetry:raw message for the whole tick.
6. 202 {accepted, readings}.

The signature covers the exact bytes on the wire, so a re-serialised body fails: a signing client must send the same string it hashed.

A threshold-evaluation failure is logged and does not fail the ingest; a persistence failure does, and is counted by telemetry_ingest_persist_failures_total.

Reads — GET /{sensorType}/latest, /entireBuilding, /dashboard. Each resolves the building’s domains through digital-twin and evaluates Cedar before touching data. latest on an empty room is 404.

A read is served in the same flat shape as the telemetry socket event: envelope plus every metric field at the top level under its own name. The trimmed payload column is a storage detail — it is re-inflated on the way out and never appears in a response. A client that reads a REST snapshot and then live events must not have to handle two shapes; when it did, a dashboard column stayed blank until the first socket tick overwrote the row.

Actions — POST /executeAction. The kernel validates the command against the plugin’s ActionSpec; the dispatch adapter translates it into the device’s own vocabulary. See Sensor Actions & the Device ACL.

Data Model

Five tables. Full retention and compression design in Telemetry Storage & Retention.

Testing

just test telemetry runs the unit and fitness suites — the whole kernel proven against in-memory fakes, with no database. just test telemetry-integration stands up TimescaleDB, Redis and Kafka in a throwaway compose project and runs tests/*.rs against them, each test on its own freshly created database.

tests/architecture.rs is a gate, not a formality: it fails if the kernel gains an I/O import, if a plugin imports a sibling, or if contracts grows a dependency.