AI-Powered Downtime Triage for Manufacturing - MQTT and Postgres

id: downtime-root-cause-triage
namespace: company.team

description: |
  **Use case:** Real-time machine downtime monitoring with AI-powered root-cause analysis. 
  Triggered by MQTT events from PLCs/IoT sensors, queries historical downtime patterns from Postgres, 
  analyzes root causes with AI, and alerts the team via Slack.
  **Trigger:** Event-based MQTT RealtimeTrigger for machine stop events.
  **Highlights:**
  - Realtime event-driven architecture with MQTT broker.
  - Conditional logic (If task) to distinguish micro-stops from critical downtime.
  - AI analysis using Gemini to generate root-cause summaries.
  - Postgres integration for historical context and audit logging.
  - Slack notifications for critical events.

variables:
  mqtt_server: tcp://host.docker.internal:1883
  mqtt_topic: plant/+/stops
  db_url: jdbc:postgresql://host.docker.internal:5433/plant
  long_stop_minutes: 2
  history_limit: 10

tasks:
  - id: ensure_table
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ vars.db_url }}"
    username: "{{ secret('PLANT_DB_USER') }}"
    password: "{{ secret('PLANT_DB_PASSWORD') }}"
    sql: |
      CREATE TABLE IF NOT EXISTS downtime_log (
        id SERIAL PRIMARY KEY,
        machine_id TEXT NOT NULL,
        line TEXT,
        reason_code TEXT,
        duration_min NUMERIC NOT NULL,
        ai_summary TEXT,
        severity TEXT,
        created_at TIMESTAMPTZ DEFAULT now()
      );

  - id: fetch_history
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ vars.db_url }}"
    username: "{{ secret('PLANT_DB_USER') }}"
    password: "{{ secret('PLANT_DB_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT reason_code, duration_min, created_at
      FROM downtime_log
      WHERE machine_id = :machine
      ORDER BY created_at DESC
      LIMIT {{ vars.history_limit }};
    parameters:
      machine: "{{ trigger.payload.machine_id }}"

  - id: log_event
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ vars.db_url }}"
    username: "{{ secret('PLANT_DB_USER') }}"
    password: "{{ secret('PLANT_DB_PASSWORD') }}"
    sql: |
      INSERT INTO downtime_log (machine_id, line, reason_code, duration_min)
      VALUES (:machine, :line, :reason, :duration);
    parameters:
      machine: "{{ trigger.payload.machine_id }}"
      line: "{{ trigger.payload.line }}"
      reason: "{{ trigger.payload.reason_code }}"
      duration: "{{ trigger.payload.duration_min }}"

  - id: is_long_stop
    type: io.kestra.plugin.core.flow.If
    condition: "{{ trigger.payload.duration_min >= vars.long_stop_minutes }}"
    then:
      - id: analyze
        type: io.kestra.plugin.ai.completion.ChatCompletion
        provider:
          type: io.kestra.plugin.ai.provider.GoogleGemini
          apiKey: "{{ secret('GEMINI_API_KEY') }}"
          modelName: gemini-2.5-flash
        messages:
          - type: SYSTEM
            content: >
              You are a manufacturing reliability engineer. Reply in exactly 3
              lines: Line 1: Likely cause Line 2: Suggested check Line 3:
              Severity (LOW/MEDIUM/HIGH)
          - type: USER
            content: |
              Stop event: {{ trigger.payload }}
              Recent stops on this machine (last {{ vars.history_limit }}): {{ outputs.fetch_history.rows }}

      - id: update_summary
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "{{ vars.db_url }}"
        username: "{{ secret('PLANT_DB_USER') }}"
        password: "{{ secret('PLANT_DB_PASSWORD') }}"
        sql: |
          UPDATE downtime_log SET ai_summary = :summary
          WHERE id = (SELECT max(id) FROM downtime_log WHERE machine_id = :machine);
        parameters:
          summary: "{{ outputs.analyze.textOutput }}"
          machine: "{{ trigger.payload.machine_id }}"

      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "🚨 Downtime Alert: {{ trigger.payload.machine_id }}",
            "blocks": [
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "*Machine:* {{ trigger.payload.machine_id }}\n*Line:* {{ trigger.payload.line }}\n*Duration:* {{ trigger.payload.duration_min }} min\n*Reason Code:* {{ trigger.payload.reason_code }}"
                }
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "*AI Analysis:*\n{{ outputs.analyze.textOutput }}"
                }
              }
            ]
          }
    else:
      - id: log_micro_stop
        type: io.kestra.plugin.core.log.Log
        message: "Micro-stop on {{ trigger.payload.machine_id }} ({{
          trigger.payload.duration_min }} min), logged to database only."

triggers:
  - id: machine_stop_event
    type: io.kestra.plugin.mqtt.RealtimeTrigger
    server: "{{ vars.mqtt_server }}"
    topic:
      - "{{ vars.mqtt_topic }}"
    serdeType: JSON

This blueprint automates the root-cause analysis of machine downtime in manufacturing environments. When a production line stops (triggered by an MQTT event from a PLC or sensor), the workflow:

Captures the event via MQTT RealtimeTrigger (e.g., plant/filler1/stops).
Queries historical context from Postgres: recent stops on the same machine.
Applies conditional logic: Micro-stops (< 2 min) are logged; longer stops trigger AI analysis.
Analyzes root causes using Google Gemini to synthesize event data + history into a likely cause, suggested check, and severity.
Stores the summary back to the downtime_log table for audit trails.
Alerts the team via Slack with the AI-generated summary, enabling faster response.

How it works

Architecture

Trigger: MQTT RealtimeTrigger listens on plant/+/stops for machine stop events (JSON payloads).
Task 1 (ensure_table): Idempotent table creation in Postgres for durability.
Task 2 (fetch_history): Queries recent downtime records for the affected machine to provide context.
Task 3 (log_event): Inserts the current stop into the audit log immediately.
Task 4 (is_long_stop): Flowable If condition, branching on stop duration.
- Then branch: For stops of at least 2 minutes, runs AI analysis and sends a Slack alert.
- Else branch: For micro-stops, logs a message without running AI.

Sample MQTT Payload

{
  "machine_id": "filler1",
  "line": "CSD3",
  "reason_code": "LOW_PRESSURE",
  "duration_min": 7
}

AI Prompt

The Gemini ChatCompletion receives the current stop, the machine's recent stop history, and generates:

Likely cause (e.g., "Recurring pressure regulation issue")
Suggested check (e.g., "Inspect pressure relief valve on filler1")
Severity (LOW, MEDIUM, HIGH)

Slack Alert Format

Includes machine ID, line, duration, reason code, and the AI-generated summary, formatted as a rich Block Kit message.

What you get

Event-driven monitoring: No polling, no cron. Reacts instantly to MQTT events.
Audit trail: Every downtime event persists to Postgres with AI analysis attached.
Contextual AI: Gemini analyzes patterns, not isolated events.
Conditional logic: Reduces noise by ignoring micro-stops.
Slack integration: Team gets instant, actionable alerts.
Extensible: Easy to add email, PagerDuty, or Kafka notifications.

Who it's for

Manufacturing plants running PLCs and IoT sensors.
Reliability engineers who need faster root-cause analysis cycles.
Operations teams who want visibility into downtime patterns.
Anyone integrating MQTT-based machinery with a data warehouse.

Prerequisites

MQTT broker (e.g., Mosquitto) publishing machine stop events.
Postgres database for historical downtime logs.
Google Gemini API key for AI analysis.
Slack workspace with an incoming webhook URL for alerts.
Kestra instance with MQTT, JDBC PostgreSQL, and Slack notification plugins.

Environment setup

Use the included docker-compose.yml to spin up a local MQTT broker and Postgres:

cd kestra-dev/
docker compose up -d

This starts:

Mosquitto MQTT broker on localhost:1883
PostgreSQL on localhost:5433 (user: plant, password: plant, db: plant)

Secrets

In the Kestra UI, add the following namespace secrets (or set as environment variables prefixed SECRET_):

Secret	Example	Source
`PLANT_DB_USER`	`plant`	Postgres username
`PLANT_DB_PASSWORD`	`plant`	Postgres password
`GEMINI_API_KEY`	`AIza...`	Google AI Studio
`SLACK_WEBHOOK_URL`	`https://hooks.slack.com/...`	Slack App incoming webhook

Quick start

Deploy MQTT + Postgres:
```
cd kestra-dev/
docker compose up -d
```
Add secrets to Kestra (Namespace → Secrets):
- PLANT_DB_USER = plant
- PLANT_DB_PASSWORD = plant
- GEMINI_API_KEY = your API key
- SLACK_WEBHOOK_URL = your Slack webhook
Copy both flows (downtime-root-cause-triage.yaml and simulate-machine-stops.yaml) into the Kestra editor under the company.team namespace.
Trigger a test:
- Manually execute simulate-machine-stops, which publishes two MQTT events (a 7-min stop and a 1-min stop).
- Watch the Kestra execution topology; you should see:
  - is_long_stop branches on the 7-minute stop (goes to then, runs AI).
  - is_long_stop evaluates to false on the 1-minute stop (goes to else, just logs).
- Check Postgres for 2 new rows in downtime_log.
- Check Slack for 1 alert (only the long stop triggers an alert).
Inspect outputs:
- Click the analyze task to see the AI-generated summary.
- Click alert to confirm the Slack message was sent.
- Query Postgres directly to see the full audit log.

Scaling and customization

Adjust long_stop_minutes (default 2) to change the cutoff for AI analysis.
Increase history_limit (default 10) to give Gemini more historical context.
Extend the payload with additional fields (e.g., OEE, shift ID, production target) and reference them in the AI prompt.
Integrate with your PLC: Update the MQTT topic and payload schema to match your sensors.

Manufacturing Downtime Root-Cause Triage with AI and MQTT

How it works

Architecture

Sample MQTT Payload

AI Prompt

Slack Alert Format

What you get

Who it's for

Prerequisites

Environment setup

Secrets

Quick start

Scaling and customization

Tags

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Manufacturing Downtime Root-Cause Triage with AI and MQTT

How it works

Architecture

Sample MQTT Payload

AI Prompt

Slack Alert Format

What you get

Who it's for

Prerequisites

Environment setup

Secrets

Quick start

Scaling and customization

Tags

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

More AI Blueprints

New to Kestra?