SingleStore Pipeline Ingestion Stall Sentinel

id: singlestore-pipeline-ingestion-stall-sentinel
namespace: company.data

inputs:
  - id: singlestore_host
    type: STRING
    displayName: SingleStore Host
    defaults: "singlestore.internal"
    description: Hostname or IP address of the SingleStore master aggregator or leaf
      node exposing port 3306.

  - id: singlestore_port
    type: INT
    displayName: SingleStore Port
    defaults: 3306
    description: Port number for the MySQL wire protocol interface on SingleStore
      (default 3306).

  - id: singlestore_database
    type: STRING
    displayName: Database Name
    defaults: "information_schema"
    description: Database name to establish the session against (default
      information_schema).

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#data-platform-alerts"
    description: Slack channel destination for operational pipeline incident cards.

  - id: auto_restart
    type: BOOL
    displayName: Auto Restart Failed Pipelines
    defaults: true
    description: Automatically dispatch START PIPELINE commands for pipelines stuck
      in an Error state.

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    defaults: false
    description: When true, audits pipeline states and dispatches notifications but
      skips remediation restarts.

triggers:
  - id: every_fifteen_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"

tasks:
  - id: probe_stalled_pipelines
    type: io.kestra.plugin.jdbc.mysql.Query
    url: "jdbc:mysql://{{ inputs.singlestore_host }}:{{ inputs.singlestore_port
      }}/{{ inputs.singlestore_database }}"
    username: "{{ secret('SINGLESTORE_USERNAME') }}"
    password: "{{ secret('SINGLESTORE_PASSWORD') }}"
    fetchType: FETCH
    allowFailure: true
    sql: |
      SELECT
        p.DATABASE_NAME AS database_name,
        p.PIPELINE_NAME AS pipeline_name,
        p.STATE AS pipeline_state,
        COALESCE(e.ERROR_KIND, 'None') AS error_kind,
        COALESCE(e.ERROR_CODE, 0) AS error_code,
        COALESCE(SUBSTRING(e.ERROR_MESSAGE, 1, 200), 'No error message recorded') AS error_message
      FROM information_schema.PIPELINES p
      LEFT JOIN (
        SELECT
          DATABASE_NAME,
          PIPELINE_NAME,
          ERROR_KIND,
          ERROR_CODE,
          ERROR_MESSAGE,
          ROW_NUMBER() OVER (PARTITION BY DATABASE_NAME, PIPELINE_NAME ORDER BY ERROR_UNIX_TIMESTAMP DESC) AS rn
        FROM information_schema.PIPELINES_ERRORS
      ) e ON p.DATABASE_NAME = e.DATABASE_NAME AND p.PIPELINE_NAME = e.PIPELINE_NAME AND e.rn = 1
      WHERE p.STATE != 'Running'
      ORDER BY p.STATE ASC, p.PIPELINE_NAME ASC
      LIMIT 50;

  - id: route_by_pipeline_health
    type: io.kestra.plugin.core.flow.Switch
    value: >-
      {% if outputs.probe_stalled_pipelines.rows is null %} PROBE_FAILED {% else
      %}
        {% set error_count = 0 %}
        {% set stopped_count = 0 %}
        {% for row in outputs.probe_stalled_pipelines.rows %}
          {% if row.pipeline_state == 'Error' %}
            {% set error_count = error_count + 1 %}
          {% else %}
            {% set stopped_count = stopped_count + 1 %}
          {% endif %}
        {% endfor %}
        {% if error_count > 0 %}
        CRITICAL
        {% elseif stopped_count > 0 %}
        WARNING
        {% else %}
        HEALTHY
        {% endif %}
      {% endif %}
    cases:
      CRITICAL:
        - id: alert_critical_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] SingleStore Ingestion Pipelines in Error State - Streaming Ingestion Halted",
              "attachments": [
                {
                  "color": "#D32F2F",
                  "title": "SingleStore Ingestion Pipeline Failure Incident",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.singlestore_host }}:{{ inputs.singlestore_port }}",
                      "short": true
                    },
                    {
                      "title": "Failed Pipelines Count",
                      "value": "{{ outputs.probe_stalled_pipelines.size }}",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "CRITICAL - Pipeline Ingestion Stalled",
                      "short": true
                    },
                    {
                      "title": "Remediation Status",
                      "value": "{{ (inputs.auto_restart and not inputs.dry_run) ? 'Dispatching START PIPELINE' : 'Remediation Skipped (Dry Run or Disabled)' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra SingleStore Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

        - id: check_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_restart == true and inputs.dry_run == false }}"
          then:
            - id: restart_pipelines
              type: io.kestra.plugin.core.flow.Loop
              values: "{{ outputs.probe_stalled_pipelines.rows }}"
              tasks:
                - id: execute_start_pipeline
                  type: io.kestra.plugin.jdbc.mysql.Query
                  url: "jdbc:mysql://{{ inputs.singlestore_host }}:{{ inputs.singlestore_port
                    }}/{{ inputs.singlestore_database }}"
                  username: "{{ secret('SINGLESTORE_USERNAME') }}"
                  password: "{{ secret('SINGLESTORE_PASSWORD') }}"
                  sql: "START PIPELINE `{{ item.value.database_name }}`.`{{
                    item.value.pipeline_name }}`;"

            - id: notify_remediation_slack
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "channel": "{{ inputs.slack_channel }}",
                  "text": "[REMEDIATION] SingleStore START PIPELINE command dispatched for all failed pipelines."
                }

      WARNING:
        - id: alert_warning_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[WARNING] SingleStore Stopped Ingestion Pipelines Detected - Ingestion Inactive",
              "attachments": [
                {
                  "color": "#F57C00",
                  "title": "SingleStore Inactive Pipeline Warning",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.singlestore_host }}:{{ inputs.singlestore_port }}",
                      "short": true
                    },
                    {
                      "title": "Inactive Pipelines Count",
                      "value": "{{ outputs.probe_stalled_pipelines.size }}",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "WARNING - Pipeline Stopped",
                      "short": true
                    },
                    {
                      "title": "Auto Restart",
                      "value": "{{ inputs.auto_restart ? 'Enabled' : 'Disabled' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra SingleStore Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

      PROBE_FAILED:
        - id: alert_probe_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] SingleStore Sentinel Probe Failed - Unable to query information_schema.PIPELINES. Verify network routing and database credentials."
            }

      HEALTHY:
        - id: log_pipelines_healthy
          type: io.kestra.plugin.core.log.Log
          message: "SingleStore cluster pipelines health is normal. All evaluated
            pipelines are in Running state."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: "SingleStore pipeline health audit completed. Status: {{
      outputs.route_by_pipeline_health.value ?? 'PROCESSED' }}. Evaluated
      stalled pipelines count: {{ outputs.probe_stalled_pipelines.size ?? 0 }}."

errors:
  - id: on_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "[ERROR] SingleStore Pipeline Ingestion Stall Sentinel execution aborted unexpectedly. Execution ID: {{ execution.id }}."
      }

SingleStore is a high-performance, distributed SQL database engineered for operational and real-time analytical workloads. Its native Pipelines feature provides scalable, exactly-once streaming ingestion from Kafka, AWS S3, Google Cloud Storage, Azure Blob Storage, and file systems directly into relational tables without third-party ETL plumbing. In production, unannounced schema drift, corrupted JSON payloads, expired cloud storage access keys, or transient upstream broker timeouts cause pipelines to enter an Error state and halt ingestion. When a pipeline stops, real-time dashboards and analytics applications serve increasingly stale data without raising query-level SQL errors. This blueprint provides continuous automated surveillance over information_schema.PIPELINES and information_schema.PIPELINES_ERRORS, classifies ingestion health, escalates incident alerts to Slack, and safely executes START PIPELINE remediation to recover streaming data flows.

How it works

Scheduled trigger (every_fifteen_minutes) initiates diagnostic pipeline health checks every 15 minutes, with manual runs supported via customizable inputs.
Diagnostic probe (probe_stalled_pipelines) connects to SingleStore over the MySQL wire protocol (port 3306) via io.kestra.plugin.jdbc.mysql.Query and queries information_schema.PIPELINES joined with recent error details from information_schema.PIPELINES_ERRORS for any pipeline whose state is not Running.
Classification logic evaluates the query output using Pebble templating, discerning whether any pipelines are in an Error state (CRITICAL), manually stopped (WARNING), all operating normally (HEALTHY), or if the database connection failed (PROBE_FAILED).
Routing task (route_by_pipeline_health) switches execution along dedicated paths according to the classified severity.
On CRITICAL or WARNING states, structured Slack incident cards notify operations channels with impacted database names, pipeline identifiers, failure counts, and remediation states.
Guarded remediation (check_remediation) checks whether auto_restart is enabled and dry_run is disabled. If confirmed, restart_pipelines iterates across stalled pipelines and executes START PIPELINE commands to safely resume streaming ingestion.
On PROBE_FAILED, dedicated alerts notify administrators of unreachable ports or authentication failures.
Audit log task (log_audit_summary) writes execution metadata and evaluated pipeline counts to task logs.

What you get

Automated continuous surveillance of SingleStore streaming pipelines across all databases and schemas.
Rapid detection of stalled pipelines and extracted error diagnostics before data freshness SLAs breach.
Automated self-healing recovery that safely dispatches START PIPELINE commands for failed ingestion streams.
Differentiated incident routing separating active failure errors from intentionally stopped pipelines.
Full execution traceability and audit logging without leaking database credentials in logs or alert payloads.

Who it's for

Data Engineers and Analytics Platform Teams operating streaming ingestion into SingleStore.
Database Administrators managing real-time analytical tables and high-throughput data pipelines.
Site Reliability Engineers ensuring continuous operational metric ingestion and dashboard freshness.

Why orchestrate this with Kestra

Declarative YAML definition unifies streaming pipeline health monitoring, conditional error triage, and automated self-healing into version-controlled GitOps workflows.
Native MySQL JDBC plugin connectivity communicates directly with SingleStore's native wire protocol without external agent daemons or Python container overhead.
Built-in conditional routing and looping isolate discovery, dry-run safety gates, and recovery executions without custom scripts.
Immutable execution history provides operational evidence documenting pipeline recovery timelines and error recurrence patterns.
Cloud-native scheduling and flow-level error-trapping ensure continuous sentinel operation independent of external cron infrastructure.

Prerequisites

SingleStore cluster running with MySQL wire compatibility enabled (default port 3306).
Database user with read permissions on information_schema.PIPELINES and information_schema.PIPELINES_ERRORS, plus administrative privileges to run START PIPELINE.
Slack incoming webhook URL for operational incident dispatch.

Secrets

SINGLESTORE_USERNAME: Username for authenticating to the SingleStore MySQL wire protocol port.
SINGLESTORE_PASSWORD: Password for the SingleStore database user.
SLACK_WEBHOOK_URL: Incoming webhook URL for posting incident cards to Slack.

Quick start

Configure SINGLESTORE_USERNAME, SINGLESTORE_PASSWORD, and SLACK_WEBHOOK_URL in your Kestra namespace secrets.
Deploy this blueprint to your Kestra instance and trigger an initial run with dry_run: true to verify connectivity against your SingleStore cluster.
Review the execution log in the Kestra UI to verify pipeline query execution from information_schema.PIPELINES.
Enable the every_fifteen_minutes schedule trigger to activate automated continuous monitoring and remediation.

How to extend

Add CLEAR PIPELINE ERRORS FOR <pipeline>; preceding START PIPELINE for pipelines blocked by error buffer limits.
Integrate PagerDuty alerting via io.kestra.plugin.pagerduty.notifications.PagerDutyAlert for high-priority production data landing stalls.
Attach Kafka lag checks to compare upstream topic offset consumer positions with SingleStore committed batch timestamps.

SingleStore Pipeline Ingestion Stall & Auto-Restart Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

SingleStore Pipeline Ingestion Stall & Auto-Restart Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

More Infrastructure Blueprints

New to Kestra?