Skip to main content

End-to-end: fleet watch

Advanced ⏱ 20 min iot · end-to-end · windows + timers

You will build: the fleet-health pipeline from the IoT use case — per-device battery/signal health in sliding windows, plus the watchdog that catches the device that stopped talking.

telemetry → streaming:health (window) → rule-based:flag → webhook (NOC) · + watchdog (keyed timers)

Prerequisites​

  • The quickstart completed. Fully deterministic — no LLM required.

1 · Scaffold​

pulse new fleet --source webhook \
  --stage streaming:health --stage rule-based:flag --sink webhook

2 · The health stage — per-device windows​

stages:
  - name: health
    engine: streaming
    operators:
      - { type: keyBy, field: device_id }
      - type: window
        spec: "sliding(5m,1m)"
        aggregations:
          batt_min: "min(battery)"
          rssi_avg: "avg(rssi)"

3 · The flag stage​

  - name: flag
    engine: rule-based
    rules:
      - { condition: "batt_min < 15", action: flag }

4 · Sink to the NOC and deploy​

pulse secret set NOC_WEBHOOK https://webhook.site/<your-id>
pulse secret set PULSE_WEBHOOK_SECRET dev-secret   # the scaffolded webhook source requires it
pulse deploy .

5 · Heartbeat a healthy device and a dying one​

pulse events publish --topic fleet.in --value '{"device_id":"d-1","battery":88,"rssi":-61}'
pulse events publish --topic fleet.in --value '{"device_id":"d-2","battery":12,"rssi":-74}'
pulse events tail --topic fleet.flag.out

d-2's window computes batt_min: 12 → flagged to the NOC. d-1 stays quiet and free.

6 · Add the silence watchdog​

Windows only see what arrives. Add the keyed-timer stage from Keyed timers — detect the silence as a second streaming stage:

  - name: watchdog
    engine: streaming
    operators:
      - type: process
        key: device_id
        timeout: "10m"
        timer: processing
        onTimeout: emit
        passThrough: true

Now the fleet is covered on both failure axes: degrading devices (windows) and disappearing devices (timers).

What just happened​

Two small declarative stages replaced a monitoring service: per-device state maintained by the engine, thresholds in reviewable config, and the whole thing running on one self-hosted node — the same shape that scales to the enterprise mesh without a rewrite when the fleet outgrows the box.

Troubleshooting​

Nothing on fleet.flag.out

Check fleet.health.out first — is batt_min actually below your threshold inside the current window? Rule field names must match aggregation names exactly.

Next​