OpenSearch Shard Allocation Recovery with Kestra

id: opensearch-shard-allocation-disabled-recovery-guard
namespace: company.team
description: |
  Detect an OpenSearch cluster whose shard allocation is stuck at none or
  primaries after a rolling restart, re-enable allocation, retry failed
  shards, then re-read health and the setting and fail if the cluster is
  still red or allocation did not return to all.

inputs:
  - id: opensearch_url
    type: STRING
    defaults: "http://localhost:9200"
    description: Base URL of the OpenSearch cluster, scheme host and port.
  - id: opensearch_username
    type: STRING
    defaults: "{{ secret('OPENSEARCH_USERNAME') }}"
    description: Username for basic auth. Clusters with security disabled ignore it.
  - id: opensearch_password
    type: STRING
    defaults: "{{ secret('OPENSEARCH_PASSWORD') }}"
    description: Password for basic auth. Clusters with security disabled ignore it.
  - id: dry_run
    type: BOOL
    defaults: true
    description: When true the flow only reports, it never changes the cluster.
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false the flow re-enables allocation and
      reroutes failed shards.

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scan for disabled shard allocation every morning in safe report-only mode.
    cron: "0 6 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false
tasks:
  - id: detect_health
    type: io.kestra.plugin.opensearch.Request
    description: Read cluster health so the flow has the status color and the
      unassigned shard count.
    connection:
      hosts:
        - "{{ inputs.opensearch_url }}"
      basicAuth:
        username: "{{ inputs.opensearch_username }}"
        password: "{{ inputs.opensearch_password }}"
    method: GET
    endpoint: "_cluster/health"

  - id: detect_settings
    type: io.kestra.plugin.opensearch.Request
    description: Read cluster settings so the flow can see the current
      cluster.routing.allocation.enable value.
    connection:
      hosts:
        - "{{ inputs.opensearch_url }}"
      basicAuth:
        username: "{{ inputs.opensearch_username }}"
        password: "{{ inputs.opensearch_password }}"
    method: GET
    endpoint: "_cluster/settings"

  - id: check_reads
    type: io.kestra.plugin.core.flow.If
    description: The Request task already fails on non-2xx statuses, so this is a
      second guard that both reads returned 200.
    condition: "{{ outputs.detect_health.status != 200 or
      outputs.detect_settings.status != 200 }}"
    then:
      - id: fail_reads
        type: io.kestra.plugin.core.execution.Fail
        description: Fail when the cluster could not be read so a bad response never
          passes for a healthy cluster.
        errorMessage: "Could not read cluster state: health status {{
          outputs.detect_health.status }}, settings status {{
          outputs.detect_settings.status }}."

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: Record the allocation setting, the unassigned shard count, the
      cluster status and the mode.
    message: |
      Shard allocation scan.
      allocation.enable: {{ outputs.detect_settings.response | jq('(.transient.cluster.routing.allocation.enable) // (.persistent.cluster.routing.allocation.enable) // "all"') | first }}.
      Cluster status: {{ outputs.detect_health.response | jq('.status') | first }}.
      Unassigned shards: {{ outputs.detect_health.response | jq('.unassigned_shards') | first }}.
      Mode dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}.
  - id: alert_detected_gate
    type: io.kestra.plugin.core.flow.If
    description: Alert Slack the moment disabled allocation or unassigned shards are
      found, before and apart from any remediation decision.
    condition: "{{ (outputs.detect_settings.response |
      jq('((.transient.cluster.routing.allocation.enable) //
      (.persistent.cluster.routing.allocation.enable) // \"all\") != \"all\"') |
      first) or (outputs.detect_health.response | jq('.unassigned_shards > 0') |
      first) }}"
    then:
      - id: alert_detected
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the allocation setting, the unassigned shard count and whether
          a change will run on this pass.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "OpenSearch shard allocation problem detected. allocation.enable {{ outputs.detect_settings.response | jq('(.transient.cluster.routing.allocation.enable) // (.persistent.cluster.routing.allocation.enable) // "all"') | first }}, cluster status {{ outputs.detect_health.response | jq('.status') | first }}, unassigned shards {{ outputs.detect_health.response | jq('.unassigned_shards') | first }}. Remediation will run {{ inputs.auto_remediate and not inputs.dry_run }} (auto_remediate {{ inputs.auto_remediate }} dry_run {{ inputs.dry_run }})."
          }

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: Act only when a problem is present and the operator asked for a
      real remediation.
    condition: "{{ ((outputs.detect_settings.response |
      jq('((.transient.cluster.routing.allocation.enable) //
      (.persistent.cluster.routing.allocation.enable) // \"all\") != \"all\"') |
      first) or (outputs.detect_health.response | jq('.unassigned_shards > 0') |
      first)) and inputs.auto_remediate and not inputs.dry_run }}"
    then:
      - id: enable_allocation
        type: io.kestra.plugin.opensearch.Request
        description: Re-enable shard allocation for every shard kind by setting the
          transient cluster setting back to all.
        connection:
          hosts:
            - "{{ inputs.opensearch_url }}"
          basicAuth:
            username: "{{ inputs.opensearch_username }}"
            password: "{{ inputs.opensearch_password }}"
        method: PUT
        endpoint: "_cluster/settings"
        body: |
          {"transient": {"cluster.routing.allocation.enable": "all"}}
      - id: reroute
        type: io.kestra.plugin.opensearch.Request
        description: Retry shards that stopped allocating after too many failed attempts
          while allocation was disabled.
        connection:
          hosts:
            - "{{ inputs.opensearch_url }}"
          basicAuth:
            username: "{{ inputs.opensearch_username }}"
            password: "{{ inputs.opensearch_password }}"
        method: POST
        endpoint: "_cluster/reroute"
        parameters:
          retry_failed: "true"
        body: |
          {}
      - id: reverify_health
        type: io.kestra.plugin.opensearch.Request
        description: Re-read cluster health after the change so the result is confirmed
          against the system itself.
        connection:
          hosts:
            - "{{ inputs.opensearch_url }}"
          basicAuth:
            username: "{{ inputs.opensearch_username }}"
            password: "{{ inputs.opensearch_password }}"
        method: GET
        endpoint: "_cluster/health"
      - id: reverify_settings
        type: io.kestra.plugin.opensearch.Request
        description: Re-read cluster settings so the new allocation.enable value is on
          the record.
        connection:
          hosts:
            - "{{ inputs.opensearch_url }}"
          basicAuth:
            username: "{{ inputs.opensearch_username }}"
            password: "{{ inputs.opensearch_password }}"
        method: GET
        endpoint: "_cluster/settings"
      - id: verify_report
        type: io.kestra.plugin.core.log.Log
        description: Record the state after the change so the audit trail shows the result.
        message: |
          After remediation.
          allocation.enable: {{ outputs.reverify_settings.response | jq('(.transient.cluster.routing.allocation.enable) // (.persistent.cluster.routing.allocation.enable) // "all"') | first }}.
          Cluster status: {{ outputs.reverify_health.response | jq('.status') | first }}.
          Unassigned shards: {{ outputs.reverify_health.response | jq('.unassigned_shards') | first }}.
          enable_allocation status {{ outputs.enable_allocation.status }}, reroute status {{ outputs.reroute.status }}.
      - id: landed
        type: io.kestra.plugin.core.flow.If
        description: Pass only when allocation is back to all and the cluster is not
          red, otherwise fail loudly.
        condition: "{{ (outputs.reverify_settings.response |
          jq('((.transient.cluster.routing.allocation.enable) //
          (.persistent.cluster.routing.allocation.enable) // \"all\") ==
          \"all\"') | first) and (outputs.reverify_health.response | jq('.status
          != \"red\"') | first) }}"
        then:
          - id: success
            type: io.kestra.plugin.core.log.Log
            description: Record that allocation is re-enabled and the cluster is no longer
              red.
            message: "Allocation re-enabled on the cluster. Status {{
              outputs.reverify_health.response | jq('.status') | first }},
              unassigned shards {{ outputs.reverify_health.response |
              jq('.unassigned_shards') | first }}."
        else:
          - id: fail_not_landed
            type: io.kestra.plugin.core.execution.Fail
            description: Fail when the cluster is still red or allocation did not return to
              all so a half-recovered cluster is never reported healthy.
            errorMessage: "Allocation is {{ outputs.reverify_settings.response |
              jq('(.transient.cluster.routing.allocation.enable) //
              (.persistent.cluster.routing.allocation.enable) // \"all\"') |
              first }} and status is {{ outputs.reverify_health.response |
              jq('.status') | first }} with {{ outputs.reverify_health.response
              | jq('.unassigned_shards') | first }} unassigned shards. See
              verify_report."
    else:
      - id: report_only
        type: io.kestra.plugin.core.log.Log
        description: Record a no-change outcome, either nothing is wrong or the run is
          in report mode.
        message: "No action taken. allocation.enable {{ outputs.detect_settings.response
          | jq('(.transient.cluster.routing.allocation.enable) //
          (.persistent.cluster.routing.allocation.enable) // \"all\"') | first
          }}, status {{ outputs.detect_health.response | jq('.status') | first
          }}, unassigned {{ outputs.detect_health.response |
          jq('.unassigned_shards') | first }}, dry_run {{ inputs.dry_run }},
          auto_remediate {{ inputs.auto_remediate }}."
errors:
  - id: alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack on any failure so stuck allocation or a cluster that
      stayed red is never silent.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "opensearch-shard-allocation-disabled-recovery-guard failed in flow {{ flow.id }} (execution {{ execution.id }}). Allocation is still disabled, the cluster stayed red or a request errored. Check the execution logs."
      }

After a rolling restart an operator often sets cluster.routing.allocation.enable to none or primaries so shards do not thrash while nodes bounce. If that setting is left behind, replicas and new shards never allocate: the cluster sits yellow or red, ingest for new indices stalls and nobody notices until an alert fires hours later. This blueprint reads cluster health and the live setting, re-enables allocation with a PUT _cluster/settings, retries shards stuck behind the failed-allocation limit with POST _cluster/reroute?retry_failed=true, then re-reads health and the setting and fails if the cluster is still red or allocation did not return to all. It alerts Slack the moment a problem is detected, before any remediation decision.

Findings

io.kestra.plugin.opensearch.Request fails the task on any non-2xx HTTP status. The gates here also read outputs.<id>.status or re-read the cluster.
cluster.routing.allocation.enable lives under transient or persistent in GET _cluster/settings. A value that was never set is absent, which means the built-in default all, so the flow treats an absent value as all.
Re-enabling allocation is set with the value all. The flow re-reads the setting and cluster health after the change, because the write returning a 200 does not prove the shards actually allocated.

How it works

detect_health (io.kestra.plugin.opensearch.Request) reads GET _cluster/health for the status color and the unassigned shard count. detect_settings reads GET _cluster/settings for the live cluster.routing.allocation.enable value.
check_reads fails the run if either read did not return HTTP 200. The Request task already fails on non-2xx.
report logs the allocation value, the status, the unassigned count and the mode. alert_detected then posts a Slack findings alert whenever allocation is not all or shards are unassigned, before and apart from the remediation gate, so a plain detect run or a dry_run run both alert.
route (io.kestra.plugin.core.flow.If) acts only when a problem is present, auto_remediate is true and dry_run is false. Otherwise report_only records a no-change outcome.
enable_allocation sends PUT _cluster/settings with transient cluster.routing.allocation.enable set to all. reroute sends POST _cluster/reroute?retry_failed=true to retry shards stuck behind the failed-allocation limit.
reverify_health and reverify_settings re-read the system. landed fails the run through fail_not_landed if the cluster is still red or allocation did not return to all.
The errors block posts a Slack alert on any failure.
A disabled Schedule trigger runs the scan every morning in report-only mode.

What you get

A one-command recovery for allocation left disabled after a rolling restart, driven through the plugin and logged as one execution.
A findings alert the moment allocation is disabled or shards are unassigned, independent of whether a change will run.
Proof the fix landed, from a re-read of health and the setting, not from a write that returned 200.
A read-only default: nothing changes unless auto_remediate is true and dry_run is false.
A Slack failure alert whenever the cluster stays red or allocation does not return to all.

Who it's for

Platform and SRE teams who run rolling restarts and need the allocation toggle flipped back without a manual curl.
Teams whose new indices silently stop allocating because enable was left at none or primaries.
Anyone who has re-enabled allocation by hand during an incident and wants it gated, logged and alertable instead.

Why orchestrate this with Kestra

Re-enabling allocation is a cluster-wide change that should be observed. Running it as a flow records what the allocation setting and health were, whether a change was allowed and whether the cluster actually recovered, all against one execution id. The read-only default stops a careless change. The re-read plus the red-cluster gate mean a green run really means allocation is back on and the cluster is not red.

Tested end to end

Verified on Kestra 2.0.5 against OpenSearch 2.19 (single node, security disabled). Allocation was disabled with PUT _cluster/settings transient cluster.routing.allocation.enable: none and a fresh index was created so its primary could not allocate, which took the cluster to red with unassigned shards. The clean, detect, dry-run, remediate and failure paths were all run. The remediate path moved the cluster from red with unassigned shards and enable: none to green with zero unassigned shards and enable: all, confirmed with a direct GET _cluster/health and GET _cluster/settings, not only the Kestra log. A separate index pinned to a non-existent node kept the cluster red after allocation was re-enabled. The flow failed that run through the re-read gate instead of reporting success.

Prerequisites

A reachable OpenSearch cluster. For a secured cluster, credentials allowed to read cluster health and settings and to update cluster settings.
A Slack incoming webhook for alerts.

Secrets

OPENSEARCH_USERNAME: username for basic auth, when the cluster has security enabled.
OPENSEARCH_PASSWORD: password for basic auth, when the cluster has security enabled.
SLACK_WEBHOOK_URL: Slack incoming webhook URL for alerts.

Inputs

opensearch_url: base URL of the cluster.
dry_run and auto_remediate: the two safety gates. A change runs only when auto_remediate is true and dry_run is false.

Quick start

Add the Slack webhook secret plus the OpenSearch credentials if your cluster is secured.
Set opensearch_url to your cluster. Run once with the defaults (dry_run true) to see the current allocation value, the status and the unassigned shard count.
When you want the toggle flipped back, run with auto_remediate true and dry_run false. The flow re-enables allocation, reroutes failed shards and fails if the cluster is still red.
Set disabled: false on the daily trigger to scan every morning in report-only mode.

How to extend

Add a Pause before enable_allocation if your change policy needs a human approval on every write.
Add a branch that pages a different channel when the cluster stays red after allocation is re-enabled, which points at a shard problem a reroute cannot fix.
Chain a downstream flow off this flow's FAILED state to open a ticket when the cluster cannot be recovered.
Narrow the reroute to specific shards by posting allocation commands in the reroute body.

Detect and Recover Disabled OpenSearch Shard Allocation

Findings

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Detect and Recover Disabled OpenSearch Shard Allocation

Findings

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Gate OIDC Signing-Key Rollovers with a Pinned JWKS

Run a task on an on-demand Databricks cluster

1Password SCIM Bridge Sync Status Sentinel

Odoo Scheduled Action Overdue Sentinel

Ollama Model Storage Headroom and Cache Eviction Sentinel

Daily Cross-Stack Compliance Audit with OPA

New to Kestra?