Skip to main content

Sensor Telemetry over MQTT

Subscribe to every sensor topic on an MQTT broker, take the line and sensor from the topic name, convert Fahrenheit to Celsius, flag readings above 80 °C as alerts, and keep malformed readings out of the telemetry stream with the reason they were rejected.

Components: mqtt input (wildcard topic, QoS 1) → mapping → switch output to three file outputs: rejects, alerts, and telemetry.

Needs: an MQTT broker the node can reach, with devices publishing JSON readings such as {"seq": 1, "ts": "2026-09-17T12:01:00Z", "temp_f": 69.0, "pressure_kpa": 102} to topics shaped plant/<line>/<sensor>/telemetry.

Proved over MQTT with synthetic readings only

This job ran end to end on an expanso-edge v2.1.21 node in local mode, not through Expanso Cloud, with a local Mosquitto broker and a script publishing synthetic readings. No sensor hardware, OPC UA, Modbus or protocol gateway was involved.

Complete job​

sensor-telemetry.yaml
name: sensor-telemetry
type: pipeline
selector:
  match_labels:
    pipeline_role: telemetry
config:
  input:
    mqtt:
      urls: ["tcp://127.0.0.1:1883"]
      topics: ["plant/+/+/telemetry"]
      client_id: "sensor-telemetry"
      qos: 1
  pipeline:
    processors:
      - mapping: |
          let parts = @mqtt_topic.split("/")
          root.site = "plant"
          root.line = $parts.index(1)
          root.sensor = $parts.index(2)
          root.seq = this.seq
          root.ts = this.ts
          root.temp_c = if this.temp_f == null {
            throw("missing temp_f")
          } else {
            ((this.temp_f - 32) * 5 / 9).round()
          }
          root.pressure_kpa = this.pressure_kpa
          root.alert = root.temp_c > 80
  output:
    switch:
      cases:
        - check: errored()
          output:
            file:
              path: "/var/tmp/expanso-telemetry/rejects.jsonl"
              codec: lines
            processors:
              - mapping: 'root = {"error": error(),
                  "topic": @mqtt_topic, "raw": content().string()}'
        - check: this.alert
          continue: true
          output:
            file:
              path: "/var/tmp/expanso-telemetry/alerts.jsonl"
              codec: lines
        - output:
            file:
              path: "/var/tmp/expanso-telemetry/telemetry.jsonl"
              codec: lines

The selector sends the job to a node labelled pipeline_role: telemetry; set that label on your node, or change the selector to a label it already carries. With no matching node, the job is stored and never runs.

Replace tcp://127.0.0.1:1883 with your broker. The three output paths are on the node that runs the job; create their directory there first:

mkdir -p /var/tmp/expanso-telemetry

A reading is an alert when temp_c is above 80. Alerts go to alerts.jsonl and, because that case sets continue: true, also to telemetry.jsonl.

Try it with a local broker​

On the node, start a Mosquitto broker in Docker, deploy the job, and publish three readings: one normal, one hot, one with no temperature:

printf 'listener 1883\nallow_anonymous true\n' > mosquitto.conf
docker run -d --name mqtt -p 127.0.0.1:1883:1883 \
  -v "$PWD/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro" \
  eclipse-mosquitto:2
expanso-cli job deploy sensor-telemetry.yaml
pub() { docker exec mqtt mosquitto_pub -q 1 -t "$1" -m "$2"; }
pub plant/line1/sensor-a/telemetry \
  '{"seq":1,"ts":"2026-09-17T12:01:00Z",'\
'"temp_f":69.0,"pressure_kpa":102}'
pub plant/line1/sensor-b/telemetry \
  '{"seq":3,"ts":"2026-09-17T12:03:00Z",'\
'"temp_f":220.0,"pressure_kpa":104}'
pub plant/line2/sensor-c/telemetry \
  '{"seq":5,"ts":"2026-09-17T12:05:00Z",'\
'"pressure_kpa":106}'

Then check the three files on the node:

cd /var/tmp/expanso-telemetry
wc -l telemetry.jsonl alerts.jsonl rejects.jsonl

Expect 2 lines in telemetry, 1 alert, and 1 reject. The reject records the reason, the topic and the raw payload. From the run, with raw shown separately:

{
  "error": "failed assignment (line 7): missing temp_f",
  "topic": "plant/line2/sensor-c/telemetry",
  "raw": "..."
}

Its raw value was the payload exactly as received: {"seq": 5, "ts": "2026-09-17T12:05:00Z", "pressure_kpa": 106}.

The job keeps its subscription open until you stop it.

What the run proved​

  • 30 synthetic readings from 3 sensors were published at QoS 1.
  • 28 valid readings reached telemetry, with line and sensor taken from the topic and each Fahrenheit value converted correctly.
  • Exactly 3 readings, at 104 °C, were flagged as alerts, and each also appeared in telemetry.
  • The 2 readings without temp_f were rejected with that reason.

Limits​

  • Proved on a local-mode node only, not through Expanso Cloud.
  • MQTT only. The catalog has opcua input and output components, but no OPC UA server, industrial gateway or real sensor hardware was part of this run.
  • site is fixed to "plant" in the mapping; the topic supplies only line and sensor.
  • The outputs are files on the node. Replace them to send telemetry and alerts elsewhere.