Verified OpenSearch reindex and atomic alias cutover

id: opensearch-reindex-alias-cutover
namespace: company.search
description: Reindex a write-blocked OpenSearch index, check the copy and a
  search canary, then move a read alias in one atomic request.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: source_index
    type: STRING
    defaults: catalog-v1
  - id: read_alias
    type: STRING
    defaults: catalog
  - id: mappings
    type: JSON
    defaults:
      dynamic: strict
      properties:
        sku:
          type: keyword
        title:
          type: text
          analyzer: english
        price:
          type: double
  - id: canary_query
    type: JSON
    defaults:
      match:
        title: wireless
  - id: canary_min_hits
    type: INT
    defaults: 1
    min: 1
  - id: replicas
    type: INT
    defaults: 1
    min: 0
    max: 3

variables:
  candidate: "{{ inputs.read_alias }}-kestra-{{ execution.id | lower }}"

tasks:
  - id: validate_names
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ inputs.source_index | regexMatch('^[a-z][a-z0-9_-]{0,119}$') }}"
      - "{{ inputs.read_alias | regexMatch('^[a-z][a-z0-9_-]{0,119}$') }}"
      - "{{ inputs.source_index != inputs.read_alias }}"
    errorMessage: Use distinct concrete index and alias names, without wildcards,
      paths or system-index prefixes.

  - id: source_settings
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "{{ inputs.source_index }}/_settings"
    parameters:
      flat_settings: "true"

  - id: require_frozen_source
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{
        outputs.source_settings.response[inputs.source_index].settings['index.b\
        locks.write'] == 'true' }}"
    errorMessage: Stop ingestion and set index.blocks.write=true on the concrete
      source before reindexing.

  - id: read_alias
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "_alias/{{ inputs.read_alias }}"

  - id: require_single_plain_alias
    type: io.kestra.plugin.core.execution.Assert
    description: Reject multi-index, filtered, routed or hidden aliases instead of
      dropping their semantics.
    conditions:
      - "{{ outputs.read_alias.response | length == 1 }}"
      - "{{ outputs.read_alias.response[inputs.source_index] is defined }}"
      - "{{
        (outputs.read_alias.response[inputs.source_index].aliases[inputs.read_a\
        lias] | length == 0) or
        ((outputs.read_alias.response[inputs.source_index].aliases[inputs.read_\
        alias] | length == 1) and
        (outputs.read_alias.response[inputs.source_index].aliases[inputs.read_a\
        lias].is_write_index ?? true) == false) }}"

  - id: refresh_source
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: POST
    endpoint: "{{ inputs.source_index }}/_refresh"

  - id: count_source
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "{{ inputs.source_index }}/_count"

  - id: require_nonempty_source
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ outputs.refresh_source.response._shards.failed == 0 }}"
      - "{{ outputs.count_source.response._shards.failed == 0 }}"
      - "{{ outputs.count_source.response.count > 0 }}"

  - id: create_candidate
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: PUT
    endpoint: "{{ render(vars.candidate) }}"
    # Webhook defaults are JSON text; manual inputs are maps.
    body: |
      {
        "settings": {"number_of_shards": 1, "number_of_replicas": {{ inputs.replicas }}},
        "mappings": {{ (inputs.mappings is map ? inputs.mappings : fromJson(inputs.mappings)) | toJson }}
      }

  - id: start_reindex
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: POST
    endpoint: _reindex
    parameters:
      wait_for_completion: "false"
      refresh: "true"
    body: |
      {
        "conflicts": "abort",
        "source": {"index": {{ inputs.source_index | toJson }}},
        "dest": {"index": {{ render(vars.candidate) | toJson }}, "op_type": "create"}
      }

  - id: wait_for_copy
    type: io.kestra.plugin.core.flow.LoopUntil
    condition: "{{ outputs.poll_reindex.response.completed == true }}"
    failOnMaxReached: true
    checkFrequency:
      interval: PT2S
      maxDuration: PT10M
    tasks:
      - id: poll_reindex
        type: io.kestra.plugin.opensearch.Request
        connection:
          hosts:
            - "{{ secret('OPENSEARCH_URL') }}"
          basicAuth:
            username: "{{ secret('OPENSEARCH_USER') }}"
            password: "{{ secret('OPENSEARCH_PASSWORD') }}"
        endpoint: "_tasks/{{ outputs.start_reindex.response.task }}"

  - id: require_complete_copy
    type: io.kestra.plugin.core.execution.Assert
    description: HTTP success alone does not mean every reindex item succeeded.
    conditions:
      - "{{ outputs.poll_reindex.response.error is not defined }}"
      - "{{ outputs.poll_reindex.response.response.failures | length == 0 }}"
      - "{{ outputs.poll_reindex.response.response.timed_out == false }}"
      - "{{ outputs.poll_reindex.response.response.version_conflicts == 0 }}"
      - "{{ outputs.poll_reindex.response.response.created ==
        outputs.count_source.response.count }}"
      - "{{ outputs.poll_reindex.response.response.total ==
        outputs.count_source.response.count }}"
    errorMessage: Reindex did not copy every source document. The serving alias has
      not been moved.

  - id: freeze_candidate
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: PUT
    endpoint: "{{ render(vars.candidate) }}/_settings"
    body:
      index.blocks.write: true

  - id: count_candidate
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "{{ render(vars.candidate) }}/_count"

  - id: search_canary
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    method: POST
    endpoint: "{{ render(vars.candidate) }}/_search"
    body: |
      {"size": 0, "track_total_hits": true, "query": {{ (inputs.canary_query is map ? inputs.canary_query : fromJson(inputs.canary_query)) | toJson }}}

  - id: require_searchable_copy
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ outputs.count_candidate.response._shards.failed == 0 }}"
      - "{{ outputs.count_candidate.response.count ==
        outputs.count_source.response.count }}"
      - "{{ outputs.search_canary.response.timed_out == false }}"
      - "{{ outputs.search_canary.response._shards.failed == 0 }}"
      - "{{ outputs.search_canary.response.hits.total.relation == 'eq' }}"
      - "{{ outputs.search_canary.response.hits.total.value >=
        inputs.canary_min_hits }}"
    errorMessage: Count or search canary failed. Inspect the candidate; the serving
      alias has not been moved.

  - id: source_settings_again
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "{{ inputs.source_index }}/_settings"
    parameters:
      flat_settings: "true"

  - id: read_alias_again
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "_alias/{{ inputs.read_alias }}"

  - id: require_same_source
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{
        outputs.source_settings_again.response[inputs.source_index].settings['i\
        ndex.uuid'] ==
        outputs.source_settings.response[inputs.source_index].settings['index.u\
        uid'] }}"
      - "{{
        outputs.source_settings_again.response[inputs.source_index].settings['i\
        ndex.blocks.write'] == 'true' }}"
      - "{{ outputs.read_alias_again.response == outputs.read_alias.response }}"
    errorMessage: The source or alias changed during reindexing. No cutover was requested.

  - id: promote_candidate
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    description: Both alias actions run in one atomic request; keep the old index
      for rollback.
    method: POST
    endpoint: _aliases
    body: |
      {"actions": [
        {"remove": {"index": {{ inputs.source_index | toJson }}, "alias": {{ inputs.read_alias | toJson }}, "must_exist": true}},
        {"add": {"index": {{ render(vars.candidate) | toJson }}, "alias": {{ inputs.read_alias | toJson }}, "is_write_index": false}}
      ]}

  - id: verify_alias
    type: io.kestra.plugin.opensearch.Request
    connection:
      hosts:
        - "{{ secret('OPENSEARCH_URL') }}"
      basicAuth:
        username: "{{ secret('OPENSEARCH_USER') }}"
        password: "{{ secret('OPENSEARCH_PASSWORD') }}"
    endpoint: "_alias/{{ inputs.read_alias }}"

  - id: require_promoted_alias
    type: io.kestra.plugin.core.execution.Assert
    conditions:
      - "{{ outputs.promote_candidate.response.acknowledged == true }}"
      - "{{ outputs.verify_alias.response | length == 1 }}"
      - "{{ outputs.verify_alias.response[render(vars.candidate)] is defined }}"
      - "{{
        outputs.verify_alias.response[render(vars.candidate)].aliases[inputs.re\
        ad_alias].is_write_index == false }}"

triggers:
  - id: request_migration
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('OPENSEARCH_REINDEX_WEBHOOK_KEY') }}"

outputs:
  - id: new_index
    type: STRING
    value: "{{ render(vars.candidate) }}"
  - id: old_index
    type: STRING
    value: "{{ inputs.source_index }}"
  - id: copied_documents
    type: INT
    value: "{{ outputs.count_candidate.response.count ?? 0 }}"

Change OpenSearch mappings without serving a partially built index. Use a frozen concrete source and a single-index read alias owned by this flow.

How it works

Verify the source write block and a plain, single-index alias. Refresh and count the source; empty sources fail before a candidate is created.
Create an execution-specific candidate and start an asynchronous reindex. LoopUntil waits up to ten minutes; all item failures and counts are checked.
Freeze the candidate, check its count and run a configurable search canary.
Recheck the source UUID, write block and alias. Remove the old alias target and add the candidate in one atomic request. Verify the resulting alias.

What you get

A read alias moved only after a complete, searchable copy is verified.
Old and candidate index names, copied count and per-task API responses.
The old index and failed candidates retained for diagnosis or rollback.

Who it's for

Search teams migrating mappings or analyzers during a planned ingestion pause.

Why orchestrate this with Kestra

Native OpenSearch tasks retain copy results and canary outcomes. The bounded wait and explicit gates make a partial reindex fail before alias promotion.

Prerequisites

Kestra 2.0, the OpenSearch plugin, and cluster access. Stop ingestion and set index.blocks.write=true on the concrete source first. No other operator may modify the source or alias during the run. This is a read-alias migration; at cutover both indexes are write-blocked, and the alias has is_write_index=false. Filtered, routed, hidden and multi-index aliases are rejected.

Secrets

OPENSEARCH_URL: cluster URL; use trusted TLS for your deployment.
OPENSEARCH_USER and OPENSEARCH_PASSWORD: credentials with index read, create, write, settings, reindex/task and alias-management permissions.
OPENSEARCH_REINDEX_WEBHOOK_KEY: key for the migration webhook.

Quick start

Configure secrets, source index, read alias and desired mappings.
Pause writers, block source writes, and choose a representative canary. Defaults expect sku, title, price and a title matching wireless.
Execute or call the keyed webhook; verify the new index and alias.

How to extend

Add relevance checks or custom analyzer settings in candidate creation. After failure, inspect or cancel the recorded reindex task before a new execution. Keep enough disk for both copies. Roll back with an atomic alias request to the old index; clean up indexes only after the rollback window. Restrict the webhook to your maintenance caller; update defaults for each new migration.

Reindex OpenSearch with a search canary and atomic read-alias cutover

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

New to Kestra?

Reindex OpenSearch with a search canary and atomic read-alias cutover

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

More Data Blueprints

Object Storage Integrity and Streaming Replica Parity Audit

OpenMetadata Catalog Health and Change-Event Sentinel

OceanBase Major Compaction Stall & Auto-Resume Sentinel

NocoDB Database Platform Health, Storage Engine, and Project Sentinel

Detect and Resume Stuck Paused NSQ Channels with Accumulating Backlog

Search OpenAlex for Related Scholarly Works

New to Kestra?