CrateDB Shard Allocation & Reroute Sentinel

id: cratedb-shard-allocation-and-reroute-sentinel
namespace: company.data

inputs:
  - id: cratedb_host
    type: STRING
    displayName: CrateDB Host
    defaults: "cratedb.internal"
    description: Hostname or IP address of the CrateDB node exposing the PostgreSQL
      wire protocol port.

  - id: cratedb_port
    type: INT
    displayName: CrateDB PostgreSQL Port
    defaults: 5432
    description: Port number for the PostgreSQL wire compatibility interface on
      CrateDB (default 5432).

  - id: cratedb_database
    type: STRING
    displayName: Database Name
    defaults: "doc"
    description: Target database or schema name to establish the JDBC session
      against (default doc).

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

  - id: auto_reroute
    type: BOOL
    displayName: Auto Reroute Failed Allocations
    defaults: true
    description: Automatically dispatch ALTER CLUSTER REROUTE RETRY FAILED when
      unassigned shards are detected.

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

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

tasks:
  - id: probe_unassigned_shards
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "jdbc:postgresql://{{ inputs.cratedb_host }}:{{ inputs.cratedb_port }}/{{
      inputs.cratedb_database }}"
    username: "{{ secret('CRATEDB_USERNAME') }}"
    password: "{{ secret('CRATEDB_PASSWORD') }}"
    fetchType: FETCH
    allowFailure: true
    sql: |
      SELECT
        COALESCE(schema_name, 'doc') AS schema_name,
        COALESCE(table_name, 'unknown') AS table_name,
        COALESCE(id, 0) AS shard_id,
        COALESCE("primary", false) AS is_primary,
        COALESCE(routing_state, 'UNKNOWN') AS routing_state,
        COALESCE(state, 'UNKNOWN') AS shard_state
      FROM sys.shards
      WHERE routing_state != 'STARTED'
      ORDER BY "primary" DESC, table_name ASC;

  - id: route_by_health
    type: io.kestra.plugin.core.flow.Switch
    value: >-
      {% if outputs.probe_unassigned_shards.rows is null %} PROBE_FAILED {% else
      %}
        {% set unassigned_primaries = 0 %}
        {% set unassigned_replicas = 0 %}
        {% for row in outputs.probe_unassigned_shards.rows %}
          {% if row.is_primary == true %}
            {% set unassigned_primaries = unassigned_primaries + 1 %}
          {% else %}
            {% set unassigned_replicas = unassigned_replicas + 1 %}
          {% endif %}
        {% endfor %}
        {% if unassigned_primaries > 0 %}
        CRITICAL
        {% elseif unassigned_replicas > 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] CrateDB Unassigned Primary Shards Detected - Immediate Cluster Intervention Required",
              "attachments": [
                {
                  "color": "#D32F2F",
                  "title": "CrateDB Unassigned Primary Shards Incident",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.cratedb_host }}:{{ inputs.cratedb_port }}",
                      "short": true
                    },
                    {
                      "title": "Impacted Shards Count",
                      "value": "{{ outputs.probe_unassigned_shards.size }}",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "CRITICAL - Primary Shard Offline",
                      "short": true
                    },
                    {
                      "title": "Remediation Status",
                      "value": "{{ (inputs.auto_reroute and not inputs.dry_run) ? 'Triggering ALTER CLUSTER REROUTE RETRY FAILED' : 'Remediation Skipped (Dry Run or Disabled)' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra CrateDB Shard Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

        - id: check_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_reroute == true and inputs.dry_run == false }}"
          then:
            - id: execute_reroute_retry
              type: io.kestra.plugin.jdbc.postgresql.Query
              url: "jdbc:postgresql://{{ inputs.cratedb_host }}:{{ inputs.cratedb_port }}/{{
                inputs.cratedb_database }}"
              username: "{{ secret('CRATEDB_USERNAME') }}"
              password: "{{ secret('CRATEDB_PASSWORD') }}"
              sql: "ALTER CLUSTER REROUTE RETRY FAILED;"

            - id: notify_remediation_slack
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "channel": "{{ inputs.slack_channel }}",
                  "text": "[REMEDIATION] CrateDB cluster reroute command ALTER CLUSTER REROUTE RETRY FAILED executed successfully."
                }

      WARNING:
        - id: alert_warning_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[WARNING] CrateDB Unassigned Replica Shards Detected - Cluster Under-Replicated",
              "attachments": [
                {
                  "color": "#F57C00",
                  "title": "CrateDB Under-Replication Warning",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.cratedb_host }}:{{ inputs.cratedb_port }}",
                      "short": true
                    },
                    {
                      "title": "Unassigned Replicas Count",
                      "value": "{{ outputs.probe_unassigned_shards.size }}",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "WARNING - Reduced Fault Tolerance",
                      "short": true
                    },
                    {
                      "title": "Auto Reroute",
                      "value": "{{ inputs.auto_reroute ? 'Enabled' : 'Disabled' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra CrateDB Shard Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

        - id: check_warning_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_reroute == true and inputs.dry_run == false }}"
          then:
            - id: execute_warning_reroute
              type: io.kestra.plugin.jdbc.postgresql.Query
              url: "jdbc:postgresql://{{ inputs.cratedb_host }}:{{ inputs.cratedb_port }}/{{
                inputs.cratedb_database }}"
              username: "{{ secret('CRATEDB_USERNAME') }}"
              password: "{{ secret('CRATEDB_PASSWORD') }}"
              sql: "ALTER CLUSTER REROUTE RETRY FAILED;"

      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] CrateDB Shard Sentinel Probe Failed - Unable to query sys.shards. Verify network routing and credentials."
            }

      HEALTHY:
        - id: log_cluster_healthy
          type: io.kestra.plugin.core.log.Log
          message: "CrateDB cluster shard routing is healthy. All evaluated shards are in
            STARTED state."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: "CrateDB shard health audit finished. Status: {{
      outputs.route_by_health.value ?? 'PROCESSED' }}. Evaluated shards count:
      {{ outputs.probe_unassigned_shards.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] CrateDB Shard Allocation Sentinel flow execution failed unexpectedly. Execution ID: {{ execution.id }}."
      }

CrateDB is a distributed, open-source SQL database built on Lucene, optimized for machine data, IoT telemetry, and real-time analytical workloads. In a distributed CrateDB cluster, data tables are partitioned across primary and replica shards distributed across cluster nodes. When data nodes restart, suffer transient network partitions, or encounter temporary disk watermarks, shards can fail allocation and remain stuck in UNASSIGNED states. Unassigned primary shards cause immediate query failures and partition unreachability, while unassigned replica shards leave tables vulnerable to complete data loss if an active node fails. This blueprint provides automated surveillance over sys.shards, detects unassigned primary and replica states, alerts engineering teams via Slack, and automates safe cluster rerouting using ALTER CLUSTER REROUTE RETRY FAILED.

How it works

Scheduled trigger (every_fifteen_minutes) initiates cluster health checks every 15 minutes, with manual runs supported via customizable inputs.
Diagnostic probe (probe_unassigned_shards) connects to CrateDB over the PostgreSQL wire protocol (port 5432) via io.kestra.plugin.jdbc.postgresql.Query and queries sys.shards for any partitions whose routing state is not STARTED.
Classification logic evaluates the query output using Pebble templating, discerning whether any unassigned partitions are primaries (CRITICAL), replicas only (WARNING), none (HEALTHY), or if the database connection failed (PROBE_FAILED).
Routing task (route_by_health) branches execution along dedicated handling paths according to the classified severity.
On CRITICAL or WARNING states, structured Slack incident cards notify operations channels with impacted cluster endpoints, shard counts, and partition severities.
Guarded remediation (check_remediation) checks whether auto_reroute is enabled and dry_run is disabled. If confirmed, execute_reroute_retry dispatches ALTER CLUSTER REROUTE RETRY FAILED; to re-trigger allocation for stuck shards without administrative intervention.
On PROBE_FAILED, dedicated alerts notify administrators of unreachable ports or authentication failures.
Audit log task (log_audit_summary) writes execution metadata and evaluated shard metrics to task logs.

What you get

Automated continuous surveillance of CrateDB shard routing states across all cluster schemas and tables.
Differentiated incident alerting separating critical primary partition drops from degraded replica allocations.
Automated self-healing cluster reroute capabilities that safely resolve transient shard allocation failures.
Safe operational gating with dry-run toggles and permission separation via namespace secrets.
Full execution traceability and audit logging without leaking database credentials in logs or alert payloads.

Who it's for

Platform Engineers and SREs maintaining production CrateDB clusters for IoT, logging, and metrics.
Database Administrators managing distributed storage partitions and replica durability.
Data Engineers deploying real-time telemetry ingestion pipelines that depend on constant table write availability.

Why orchestrate this with Kestra

Declarative YAML definition consolidates cluster health monitoring, conditional severity branching, and automated self-healing into version-controlled GitOps workflows.
Native PostgreSQL JDBC plugin connectivity connects directly to CrateDB's native wire protocol without external agent binaries or Python container overhead.
Built-in conditional routing and dry-run safety gates isolate discovery from remediation commands.
Immutable execution history provides cryptographic compliance and operational logs documenting cluster recovery actions.
Resilient error trapping ensures cluster monitoring failures trigger immediate operational escalations.

Prerequisites

CrateDB cluster running with PostgreSQL wire compatibility enabled (default port 5432).
Database user with read access to sys.shards and administrative permissions to run ALTER CLUSTER REROUTE.
Slack incoming webhook URL for operational incident dispatch.

Secrets

CRATEDB_USERNAME: Username for authenticating to the CrateDB PostgreSQL wire interface (e.g. crate).
CRATEDB_PASSWORD: Password for the CrateDB database user.
SLACK_WEBHOOK_URL: Incoming webhook URL for posting incident cards to Slack.

Quick start

Configure CRATEDB_USERNAME, CRATEDB_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 CrateDB cluster.
Review the execution log in the Kestra UI to verify shard query execution from sys.shards.
Enable the every_fifteen_minutes schedule trigger to activate automated continuous monitoring and remediation.

How to extend

Add allocation reason diagnostics by querying sys.allocations when unassigned shards persist across consecutive runs.
Integrate PagerDuty alerting via io.kestra.plugin.pagerduty.notifications.PagerDutyAlert for immediate on-call escalation during critical primary shard outages.
Attach table-level optimization using OPTIMIZE TABLE <table> WITH (max_num_segments = 1) to compact fragmented Lucene segments during off-peak hours.

CrateDB Shard Allocation and Cluster Reroute 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?

CrateDB Shard Allocation and Cluster Reroute 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

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

AI Synthetic Relational Data Generator and Referential Integrity Gate

Autonomous Mathematical Research and Scientific Computation Pipeline

Airbyte Connection Sync Failure Sentinel

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

More Infrastructure Blueprints

Create HubSpot Tickets Automatically for Incident Management

CouchDB Active Task and Continuous Replication Sentinel

Couchbase TTL expiry triage sentinel

Create Linear Issues Automatically for Incident Management

Cancel runaway Couchbase N1QL queries that starve the query service

Audit Couchbase GSI Secondary Indexes and Detect Offline States

New to Kestra?