MQTT Broker Health Sentinel for Kestra

id: mqtt-broker-health-sentinel
namespace: company.team
description: |
  On a schedule, subscribe to the MQTT broker's own reserved $SYS diagnostic
  topic tree for a short bounded window (no application or device topic
  involved) and classify the broker's self-reporting channel as HEALTHY,
  SILENT, or UNREACHABLE. Log clean runs and page Slack the moment the
  broker itself stops publishing its own internal statistics.

inputs:
  - id: mqtt_server
    type: STRING
    defaults: "tcp://localhost:1883"
    displayName: "MQTT Broker URI"
    description: "Broker connection URI, e.g. tcp://broker.example.com:1883."
  - id: sys_topic_filter
    type: STRING
    defaults: "$SYS/#"
    displayName: "Broker $SYS Diagnostic Topic Filter"
    description: "The broker's reserved, broker-published diagnostic topic tree (not
      an application or device topic). Standards-compliant brokers such as
      Mosquitto, EMQX, and HiveMQ publish here automatically with no producer
      code required."
  - id: probe_duration
    type: STRING
    defaults: "PT10S"
    displayName: "Probe Duration"
    description: "ISO-8601 duration; maximum wall-clock time the subscription is
      allowed to run before it stops and returns its count."
  - id: min_expected_messages
    type: INT
    defaults: 1
    displayName: "Minimum Expected $SYS Messages"
    description: "Below this count within the probe window, the broker's own
      diagnostic channel is classified SILENT."

triggers:
  - id: broker_health_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Probe the broker's $SYS channel every 15 minutes. Shipped disabled;
      enable it once mqtt_server points at a real broker and a manual run has
      been verified.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      probe_duration: "PT10S"
      min_expected_messages: 1

tasks:
  - id: probe_broker_diagnostics
    type: io.kestra.plugin.mqtt.Subscribe
    description: Subscribe to the broker's reserved $SYS topic tree for a short
      bounded window. This checks the broker's own self-reporting loop, not any
      device or application topic, so it is unaffected by whether any device has
      ever connected.
    server: "{{ inputs.mqtt_server }}"
    clientId: kestra-broker-health-sentinel
    topic: "{{ inputs.sys_topic_filter }}"
    serdeType: STRING
    qos: 1
    maxRecords: 50
    maxDuration: "{{ inputs.probe_duration }}"
    allowFailure: true

  - id: classify_broker_health
    type: io.kestra.plugin.core.debug.Return
    description: Classify the sampled $SYS message count, treating a missing output
      as a broker connectivity failure rather than silence.
    format: "{{ outputs.probe_broker_diagnostics.messagesCount is not defined ?
      'UNREACHABLE' : (outputs.probe_broker_diagnostics.messagesCount <
      inputs.min_expected_messages ? 'SILENT' : 'HEALTHY') }}"

  - id: route_by_broker_health
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log when the broker's diagnostic channel is
      active, or a Slack page when it has gone silent or is unreachable.
    value: "{{ outputs.classify_broker_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "MQTT broker {{ inputs.mqtt_server }} is healthy: {{
            outputs.probe_broker_diagnostics.messagesCount }} message(s)
            observed on {{ inputs.sys_topic_filter }} in this window."
      "SILENT":
        - id: alert_silent_broker
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the broker's own diagnostic channel produced fewer
            messages than expected, meaning the broker's self-stats loop itself
            may be degraded, not just one application's traffic.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "MQTT broker sentinel SILENT: broker {{ inputs.mqtt_server }} returned only {{ outputs.probe_broker_diagnostics.messagesCount }} message(s) on {{ inputs.sys_topic_filter }} in the last {{ inputs.probe_duration }} (minimum expected {{ inputs.min_expected_messages }}). The broker's own self-reporting loop may be disabled or degraded. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_broker
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the subscription could not return a message count
            at all, indicating the broker is unreachable or rejected the
            connection.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "MQTT broker sentinel UNREACHABLE: subscribing to {{ inputs.sys_topic_filter }} on {{ inputs.mqtt_server }} did not return a message count (connection refused, authentication failure, or network unreachable). Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "MQTT broker sentinel completed with an unexpected classification: {{
          outputs.classify_broker_health.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record a single audit line for every run regardless of branch
      taken, so the check's own history stays queryable independent of Slack.
    message: "MQTT broker sentinel audit: broker={{ inputs.mqtt_server }} topic={{
      inputs.sys_topic_filter }} status={{ outputs.classify_broker_health.value
      }} sampled_count={{ outputs.probe_broker_diagnostics.messagesCount is
      defined ? outputs.probe_broker_diagnostics.messagesCount : 'n/a' }}
      execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails outright, so a
      broken check is never mistaken for a healthy broker.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "MQTT broker sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Diagnostic-channel health of broker {{ inputs.mqtt_server }} is now unknown; check the task logs and the broker manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_broker_health.value }}"

An MQTT broker can degrade in ways no single device's heartbeat would ever reveal: its own internal stats loop can stop publishing while individual application topics still look fine, or look fine right up until the broker itself becomes unreachable. This blueprint turns a scheduled, bounded subscription to the broker's reserved $SYS topic tree into a Day 2 infrastructure health gate, distinct from monitoring any one device or application's traffic.

How it works

probe_broker_diagnostics (io.kestra.plugin.mqtt.Subscribe) connects to mqtt_server and subscribes to sys_topic_filter ($SYS/# by default), the broker's own reserved, broker-published diagnostic topic tree. Standards-compliant brokers (Mosquitto, EMQX, HiveMQ) publish here automatically, independent of whether any device or application has ever connected. The subscription is bounded by maxDuration and returns messagesCount. allowFailure: true keeps a connection failure from aborting the flow outright.
classify_broker_health (io.kestra.plugin.core.debug.Return) evaluates a single Pebble ternary against messagesCount: missing entirely becomes UNREACHABLE, below min_expected_messages becomes SILENT, otherwise HEALTHY.
route_by_broker_health (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY logs a quiet confirmation; SILENT and UNREACHABLE each page Slack with the sampled count and the execution id; a defaults fallback logs the raw value for any classification outside the three known cases.
log_audit_result (io.kestra.plugin.core.log.Log) writes one audit line per run, independent of which Switch branch fired, so the check's own history stays queryable even if Slack delivery fails.
The flow-level errors block pages Slack separately if the flow fails outright, so a broken sentinel is never mistaken for a healthy broker.
The broker_health_schedule trigger probes every 15 minutes and ships disabled: true, with explicit inputs: overrides for probe_duration and min_expected_messages so the scheduled run's probe window is pinned regardless of this flow's own defaults.

What you get

A scheduled broker-infrastructure health check that is independent of any single device's or application's traffic pattern.
A single status output (HEALTHY / SILENT / UNREACHABLE) other flows can gate on.
Two distinct Slack pages, one per failure mode, each carrying the sampled count and the execution id.
An audit log line on every run, separate from Slack, so the check's own history survives a webhook outage.

Who it's for

Platform teams running a shared MQTT broker who need a signal on the broker itself, separate from per-device or per-application monitors already watching their own topics.
On-call engineers who want to tell "one device went quiet" apart from "the broker's own diagnostics stopped," two very different incidents with the same symptom from a single device monitor's point of view.
Teams that already run a device-level heartbeat monitor and want a complementary, broker-level signal rather than a duplicate of it.

Why orchestrate this with Kestra

A device heartbeat monitor answers "is my fleet talking," which says nothing about the broker's own health when the fleet happens to be quiet for legitimate reasons, or says nothing is wrong right up until the broker process itself stops responding. Kestra's scheduled, bounded Subscribe task against the broker's own $SYS tree gives an independent, broker-scoped signal, with the schedule, the classification, and the Slack call all captured as one auditable execution history instead of a side-channel script nobody reviews.

Prerequisites

An MQTT broker reachable at mqtt_server with its $SYS diagnostic publishing enabled (the default on Mosquitto, EMQX, and HiveMQ unless explicitly turned off).
A Slack incoming webhook for the SILENT, UNREACHABLE, and flow-failure alerts.

Secrets

SLACK_WEBHOOK: Slack incoming webhook used for every alert branch and the flow-failure alert.
For an authenticated broker, add username and password properties to the probe_broker_diagnostics task, sourced from secrets such as {{ secret('MQTT_USERNAME') }} and {{ secret('MQTT_PASSWORD') }}, rather than inlining credentials.

In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }}, for example SECRET_SLACK_WEBHOOK=$(echo -n 'https://hooks.slack.com/services/<placeholder>' | base64). This keeps the webhook out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Quick start

Start a local Mosquitto broker, which publishes $SYS statistics by default, on a port well away from Kestra's own UI: docker run -d --name mosquitto-broker -p 1883:1883 eclipse-mosquitto:2
If Kestra is also running in Docker, check docker network ls for Kestra's network and attach the broker to it so the worker can reach it by container name: docker network connect YOUR_KESTRA_NETWORK mosquitto-broker Then set mqtt_server to tcp://mosquitto-broker:1883. From a host-based Kestra install, use tcp://localhost:1883 instead.
Add the SLACK_WEBHOOK secret and run the flow once manually with no application traffic at all. status should still read HEALTHY, since $SYS publishes on its own.
Stop the broker container and run the flow again to confirm status reads UNREACHABLE.
Enable broker_health_schedule once pointed at a real broker.

How to extend

Pair this flow with a device-level heartbeat monitor (subscribing to an application topic such as heartbeat/+) so silence on one side and silence on the other are distinguishable incidents instead of one undifferentiated alert.
Narrow sys_topic_filter to a single metric such as $SYS/broker/clients/connected and read its payload from {{ outputs.probe_broker_diagnostics.uri }} for a connected-client-count threshold instead of a presence check.
Push status into a KV store and alert only on state transitions instead of every run.
Add username/password properties sourced from secrets for a broker that requires authentication.

Inputs

mqtt_server: broker connection URI.
sys_topic_filter: the broker's reserved $SYS diagnostic topic tree to subscribe to.
probe_duration: ISO-8601 duration cap on how long the subscription runs.
min_expected_messages: floor below which the broker's diagnostic channel is classified SILENT.

Outputs

status: one of HEALTHY, SILENT, UNREACHABLE.

Pitfalls

Some managed or hardened broker deployments disable $SYS publishing entirely for security reasons; confirm sys_interval-style diagnostics are enabled before trusting this check, or this flow will permanently read SILENT against a perfectly healthy broker.
$SYS/# is a broad wildcard; on a very busy broker it can return a large batch within maxRecords. Narrow sys_topic_filter to a single metric topic if you only need a presence check rather than full diagnostics.
This flow is deliberately independent of device/application traffic; it will report HEALTHY even if every device is disconnected, as long as the broker's own diagnostics are publishing. Pair it with a device-level heartbeat monitor rather than using it as a substitute.
For an authenticated broker, a missing username/password on the task produces the same UNREACHABLE classification as a genuine network failure; check credentials first when triaging an alert.