Migrating from Elasticsearch to OpenSearch — Zero-Downtime Playbook

Why Teams Are Migrating from Elasticsearch to OpenSearch

In January 2021, Elastic announced it would change the license of Elasticsearch and Kibana from Apache 2.0 to the Server Side Public License (SSPL) starting with version 7.11. SSPL is not an OSI-approved open-source license — it requires organisations that run Elasticsearch as a service to open-source their entire service stack. For cloud providers and SaaS companies, this was a dealbreaker.

Amazon Web Services responded by forking Elasticsearch 7.10.2 (the last Apache 2.0 release) and creating OpenSearch under Apache 2.0. OpenSearch has since grown into a mature independent project with a governance model, plugin ecosystem, and release cadence decoupled from Elastic. Teams running self-managed clusters are migrating for three reasons: license compliance, access to OpenSearch-native features (like vector search improvements and ML Commons), and alignment with managed services like Amazon OpenSearch Service.

This playbook covers the full migration path: pre-migration assessment, strategy selection, zero-downtime alias-based cutover, client SDK changes, index template migration, security plugin differences, post-migration verification, and rollback procedures.

Pre-Migration Assessment

A migration that skips the assessment phase encounters surprises mid-cutover. Run the following audit against your existing Elasticsearch cluster before writing a single line of migration code.

Cluster Inventory Script

The script below queries the Elasticsearch REST API to produce a migration readiness report: version, nodes, indices, plugins, and deprecated feature usage.

#!/usr/bin/env python3
# es_audit.py — Pre-migration assessment for Elasticsearch clusters
# pip install elasticsearch requests

import json
import sys
from elasticsearch import Elasticsearch

ES_HOST = "https://localhost:9200"
ES_USER = "elastic"
ES_PASS = "changeme"   # replace with env var in production

client = Elasticsearch(
    ES_HOST,
    basic_auth=(ES_USER, ES_PASS),
    verify_certs=False,
)


def audit_cluster() -> dict:
    report = {}

    # Cluster version and health
    info = client.info()
    health = client.cluster.health()
    report["version"] = info["version"]["number"]
    report["cluster_name"] = info["cluster_name"]
    report["health"] = health["status"]
    report["node_count"] = health["number_of_nodes"]
    report["data_nodes"] = health["number_of_data_nodes"]

    # Node details
    nodes = client.nodes.info(metric=["os", "plugins", "settings"])
    report["nodes"] = [
        {
            "name": n["name"],
            "roles": n.get("roles", []),
            "os": n["os"]["name"],
            "plugins": [p["name"] for p in n.get("plugins", [])],
        }
        for n in nodes["nodes"].values()
    ]

    # Index summary
    indices = client.cat.indices(format="json", h="index,docs.count,store.size,status")
    report["index_count"] = len(indices)
    report["total_docs"] = sum(int(i.get("docs.count") or 0) for i in indices)
    report["large_indices"] = [
        {"index": i["index"], "docs": i["docs.count"], "size": i["store.size"]}
        for i in indices
        if int(i.get("docs.count") or 0) > 1_000_000
    ]

    # Index templates
    templates = client.indices.get_index_template()
    report["index_templates"] = [t["name"] for t in templates.get("index_templates", [])]

    # Ingest pipelines
    pipelines = client.ingest.get_pipeline()
    report["ingest_pipelines"] = list(pipelines.keys())

    # Snapshot repositories
    repos = client.snapshot.get_repository()
    report["snapshot_repositories"] = list(repos.keys())

    # Lifecycle policies (ILM)
    ilm = client.ilm.get_lifecycle()
    report["ilm_policies"] = list(ilm.keys())

    # Deprecated API usage (check _xpack endpoint — absent on OpenSearch)
    try:
        client.perform_request("GET", "/_xpack")
        report["xpack_in_use"] = True
    except Exception:
        report["xpack_in_use"] = False

    return report


if __name__ == "__main__":
    report = audit_cluster()
    print(json.dumps(report, indent=2))

    # Flag blockers
    blockers = []
    major = int(report["version"].split(".")[0])
    if major < 7:
        blockers.append(f"ES version {report['version']} is below 7.x — upgrade to 7.10.x first")
    if report["health"] != "green":
        blockers.append(f"Cluster health is {report['health']} — resolve before migrating")

    if blockers:
        print("\n[BLOCKERS]")
        for b in blockers:
            print(f"  ✗ {b}")
        sys.exit(1)
    else:
        print("\n[OK] No blockers found — cluster is migration-ready")

Plugin Compatibility Matrix

Not every Elasticsearch plugin has a direct OpenSearch equivalent. The table below covers the most common ones — verify the full list against your node plugin output before committing to a migration timeline.

Migration Strategy Selection

There is no single right migration strategy — the correct choice depends on cluster size, tolerable downtime window, network topology, and whether you are migrating to a managed service or self-managed OpenSearch. The three viable strategies are:

Zero-Downtime Playbook — Alias-Based Cutover

The alias-based approach is the recommended path for production clusters that serve live search traffic. The key insight is that applications should never reference index names directly — they should always read and write through aliases. This makes the cutover a single atomic alias update rather than a deployment.

Phase 1 — Prepare OpenSearch Cluster

# docker-compose.yml — Local OpenSearch 2.x for migration testing
# For production: use the official Helm chart or AWS OpenSearch Service

version: "3.8"
services:
  opensearch-node1:
    image: opensearchproject/opensearch:2.13.0
    environment:
      - cluster.name=migration-cluster
      - node.name=opensearch-node1
      - discovery.seed_hosts=opensearch-node1,opensearch-node2
      - cluster.initial_cluster_manager_nodes=opensearch-node1
      - bootstrap.memory_lock=true
      - "OPENSEARCH_JAVA_OPTS=-Xms2g -Xmx2g"
      - DISABLE_SECURITY_PLUGIN=true   # enable only after mapping ES security config
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - opensearch-data1:/usr/share/opensearch/data
    ports:
      - "9201:9200"    # offset port to avoid conflict with existing ES on 9200

  opensearch-node2:
    image: opensearchproject/opensearch:2.13.0
    environment:
      - cluster.name=migration-cluster
      - node.name=opensearch-node2
      - discovery.seed_hosts=opensearch-node1,opensearch-node2
      - cluster.initial_cluster_manager_nodes=opensearch-node1
      - bootstrap.memory_lock=true
      - "OPENSEARCH_JAVA_OPTS=-Xms2g -Xmx2g"
      - DISABLE_SECURITY_PLUGIN=true
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - opensearch-data2:/usr/share/opensearch/data

  opensearch-dashboards:
    image: opensearchproject/opensearch-dashboards:2.13.0
    ports:
      - "5602:5601"
    environment:
      OPENSEARCH_HOSTS: '["http://opensearch-node1:9200"]'

volumes:
  opensearch-data1:
  opensearch-data2:

Phase 2 — Migrate Index Templates and Pipelines

Before reindexing data, replicate all index templates and ingest pipelines to OpenSearch. The script below exports them from Elasticsearch and applies them to OpenSearch, translating any deprecated Elasticsearch-specific settings.

#!/usr/bin/env python3
# migrate_templates.py — Export ES index templates and apply to OpenSearch
# pip install elasticsearch opensearch-py

from elasticsearch import Elasticsearch
from opensearchpy import OpenSearch
import json

ES_HOST = "https://localhost:9200"
OS_HOST = "https://localhost:9201"

es = Elasticsearch(ES_HOST, basic_auth=("elastic", "changeme"), verify_certs=False)
os_client = OpenSearch(OS_HOST, http_auth=("admin", "admin"), verify_certs=False)


def migrate_component_templates():
    templates = es.cluster.get_component_template()
    for item in templates.get("component_templates", []):
        name = item["name"]
        body = item["component_template"]
        # Remove ES-specific lifecycle keys not supported in OS 2.x
        body.get("template", {}).pop("lifecycle", None)
        os_client.cluster.put_component_template(name=name, body=body)
        print(f"  [component_template] {name}")


def migrate_index_templates():
    templates = es.indices.get_index_template()
    for item in templates.get("index_templates", []):
        name = item["name"]
        body = item["index_template"]
        os_client.indices.put_index_template(name=name, body=body)
        print(f"  [index_template] {name}")


def migrate_ingest_pipelines():
    pipelines = es.ingest.get_pipeline()
    for pipeline_id, pipeline_body in pipelines.items():
        # The 'set_security_user' processor is X-Pack only — remove it
        processors = pipeline_body.get("processors", [])
        cleaned = [p for p in processors if "set_security_user" not in p]
        pipeline_body["processors"] = cleaned
        os_client.ingest.put_pipeline(id=pipeline_id, body=pipeline_body)
        print(f"  [pipeline] {pipeline_id}")


def migrate_ilm_as_ism():
    """
    ILM (Index Lifecycle Management) policies from ES become ISM
    (Index State Management) policies in OpenSearch.
    This is a structural translation — not a 1:1 mapping.
    For complex policies, translate manually using the ISM docs.
    """
    ilm_policies = es.ilm.get_lifecycle()
    print(f"  [ILM] Found {len(ilm_policies)} policies — translate to ISM manually:")
    for name in ilm_policies:
        print(f"    - {name}")


if __name__ == "__main__":
    print("Migrating component templates...")
    migrate_component_templates()

    print("\nMigrating index templates...")
    migrate_index_templates()

    print("\nMigrating ingest pipelines...")
    migrate_ingest_pipelines()

    print("\nILM -> ISM translation (manual step):")
    migrate_ilm_as_ism()

    print("\nTemplate migration complete.")

Phase 3 — Remote Reindex

With templates in place, use OpenSearch's remote reindex API to pull data from the live Elasticsearch cluster. OpenSearch connects to Elasticsearch as the source — no downtime required on the ES side.

#!/usr/bin/env python3
# remote_reindex.py — Trigger remote reindex from ES to OpenSearch
# Requires: opensearch.yml whitelist entry for the ES host

import time
import requests
from opensearchpy import OpenSearch

OS_HOST = "https://localhost:9201"
ES_HOST = "https://es-cluster.internal:9200"
ES_USER = "elastic"
ES_PASS = "changeme"

os_client = OpenSearch(OS_HOST, http_auth=("admin", "admin"), verify_certs=False)


def reindex_index(source_index: str, dest_index: str, batch_size: int = 1000) -> dict:
    """
    Initiate an async remote reindex from ES to OpenSearch.
    Returns the task ID for monitoring.
    """
    body = {
        "source": {
            "remote": {
                "host": ES_HOST,
                "username": ES_USER,
                "password": ES_PASS,
                "socket_timeout": "60s",
                "connect_timeout": "30s",
            },
            "index": source_index,
            "size": batch_size,
        },
        "dest": {
            "index": dest_index,
            "op_type": "index",    # 'create' to skip existing docs
        },
        "conflicts": "proceed",    # don't abort on version conflicts
    }

    response = os_client.reindex(body=body, wait_for_completion=False, timeout="1m")
    return response


def monitor_task(task_id: str, poll_interval: int = 10) -> bool:
    """Poll a reindex task until completion, printing progress."""
    print(f"Monitoring task {task_id}...")
    while True:
        task = os_client.tasks.get(task_id=task_id)
        status = task["task"]["status"]
        created = status.get("created", 0)
        total = status.get("total", 0)
        pct = (created / total * 100) if total > 0 else 0
        print(f"  Progress: {created}/{total} docs ({pct:.1f}%)")

        if task.get("completed"):
            failures = task.get("response", {}).get("failures", [])
            if failures:
                print(f"  Completed with {len(failures)} failures")
                return False
            print("  Completed successfully")
            return True

        time.sleep(poll_interval)


def reindex_all_indices(indices: list[str]) -> None:
    for index in indices:
        print(f"\nReindexing: {index}")
        result = reindex_index(source_index=index, dest_index=index)
        task_id = result["task"]
        success = monitor_task(task_id)
        if not success:
            print(f"  WARNING: reindex of {index} had failures — inspect task API")


if __name__ == "__main__":
    # List indices to reindex — skip system indices (.)
    import sys

    indices_to_migrate = [
        "logs-app-2025",
        "logs-app-2026",
        "products",
        "users",
        "events",
    ]

    reindex_all_indices(indices_to_migrate)

# opensearch.yml — Required whitelist for remote reindex
# Add to /usr/share/opensearch/config/opensearch.yml on each OpenSearch node

reindex.remote.whitelist:
  - "es-cluster.internal:9200"
  - "es-cluster.internal:443"

# For ES clusters with self-signed certs, also add:
reindex.ssl.verification_mode: none

Phase 4 — Alias Cutover

The cutover is the moment read traffic moves from Elasticsearch to OpenSearch. With aliases in place, this is a single atomic API call — applications never see a routing change, only an alias update.

#!/usr/bin/env python3
# cutover.py — Zero-downtime alias-based cutover from ES to OpenSearch
# Run this after remote reindex completes and doc counts match

from elasticsearch import Elasticsearch
from opensearchpy import OpenSearch
import sys

ES_HOST = "https://localhost:9200"
OS_HOST = "https://localhost:9201"

es = Elasticsearch(ES_HOST, basic_auth=("elastic", "changeme"), verify_certs=False)
os_client = OpenSearch(OS_HOST, http_auth=("admin", "admin"), verify_certs=False)

# Indices to cut over — each has a read alias and a write alias
INDICES = [
    {"index": "products", "read_alias": "products-read", "write_alias": "products-write"},
    {"index": "users",    "read_alias": "users-read",    "write_alias": "users-write"},
    {"index": "events",   "read_alias": "events-read",   "write_alias": "events-write"},
]


def verify_doc_counts() -> bool:
    """Verify OpenSearch doc counts are within 0.1% of Elasticsearch."""
    print("Verifying document counts...")
    all_ok = True
    for item in INDICES:
        idx = item["index"]
        es_count = es.count(index=idx)["count"]
        os_count = os_client.count(index=idx)["count"]
        delta = abs(es_count - os_count) / max(es_count, 1)
        status = "OK" if delta < 0.001 else "MISMATCH"
        print(f"  {idx}: ES={es_count:,} OS={os_count:,} delta={delta:.4%} [{status}]")
        if status != "OK":
            all_ok = False
    return all_ok


def cutover_aliases() -> None:
    """
    Atomically move aliases from Elasticsearch indices to OpenSearch indices.
    Step 1: Add aliases to OpenSearch indices.
    Step 2: Remove aliases from Elasticsearch indices.
    Applications using aliases see no disruption.
    """
    print("\nAdding aliases to OpenSearch indices...")
    for item in INDICES:
        actions = [
            {"add": {"index": item["index"], "alias": item["read_alias"]}},
            {"add": {"index": item["index"], "alias": item["write_alias"]}},
        ]
        os_client.indices.update_aliases(body={"actions": actions})
        print(f"  Added aliases to OpenSearch: {item['index']}")

    print("\nRemoving aliases from Elasticsearch indices...")
    for item in INDICES:
        actions = [
            {"remove": {"index": item["index"], "alias": item["read_alias"]}},
            {"remove": {"index": item["index"], "alias": item["write_alias"]}},
        ]
        try:
            es.indices.update_aliases(body={"actions": actions})
            print(f"  Removed aliases from Elasticsearch: {item['index']}")
        except Exception as e:
            print(f"  WARNING: Could not remove ES alias for {item['index']}: {e}")

    print("\nCutover complete. Traffic is now routed to OpenSearch.")


if __name__ == "__main__":
    if not verify_doc_counts():
        print("\nAborting cutover — doc count mismatch detected")
        sys.exit(1)

    print("\nDoc counts verified. Proceeding with alias cutover...")
    cutover_aliases()

Client SDK Migration

The OpenSearch client libraries are forks of the official Elasticsearch clients. They share the same API surface for core operations but use different package names and connection classes. Migrating is mechanical — search your codebase for the old import paths and replace them.

Python

# Before (Elasticsearch)
# pip install elasticsearch
from elasticsearch import Elasticsearch

client = Elasticsearch(
    "https://es-cluster:9200",
    basic_auth=("elastic", "password"),
    verify_certs=True,
    ca_certs="/path/to/ca.crt",
)

# After (OpenSearch)
# pip install opensearch-py
from opensearchpy import OpenSearch

client = OpenSearch(
    hosts=[{"host": "opensearch-cluster", "port": 9200}],
    http_auth=("admin", "password"),
    use_ssl=True,
    verify_certs=True,
    ca_certs="/path/to/ca.crt",
    ssl_show_warn=False,
)

# All standard query operations are identical:
response = client.search(
    index="products",
    body={
        "query": {
            "match": {"name": "widget"},
        },
        "size": 10,
    },
)
hits = response["hits"]["hits"]

Node.js / TypeScript

// Before (Elasticsearch)
// npm install @elastic/elasticsearch
import { Client as ElasticsearchClient } from "@elastic/elasticsearch";

const esClient = new ElasticsearchClient({
  node: "https://es-cluster:9200",
  auth: { username: "elastic", password: "password" },
});

// After (OpenSearch)
// npm install @opensearch-project/opensearch
import { Client as OpenSearchClient } from "@opensearch-project/opensearch";

const osClient = new OpenSearchClient({
  node: "https://opensearch-cluster:9200",
  auth: { username: "admin", password: "password" },
  ssl: {
    rejectUnauthorized: true,
    ca: readFileSync("/path/to/ca.crt"),
  },
});

// Search API is identical
const response = await osClient.search({
  index: "products",
  body: {
    query: { match: { name: "widget" } },
    size: 10,
  },
});

const hits = response.body.hits.hits;

Logstash

# logstash.conf — Before (Elasticsearch output)
output {
  elasticsearch {
    hosts => ["https://es-cluster:9200"]
    user => "logstash_writer"
    password => "${LOGSTASH_PASSWORD}"
    index => "logs-%{+YYYY.MM.dd}"
    ssl => true
    cacert => "/etc/logstash/certs/ca.crt"
  }
}

# logstash.conf — After (OpenSearch output)
# gem install logstash-output-opensearch
output {
  opensearch {
    hosts => ["https://opensearch-cluster:9200"]
    user => "admin"
    password => "${OPENSEARCH_PASSWORD}"
    index => "logs-%{+YYYY.MM.dd}"
    ssl => true
    cacert => "/etc/logstash/certs/ca.crt"
    ssl_certificate_verification => true
  }
}

Security Plugin Migration — X-Pack to OpenSearch Security

Security configuration is the most labour-intensive part of the migration. X-Pack Security (Elasticsearch) and OpenSearch Security both implement TLS, role-based access control, and audit logging, but the configuration files, API paths, and role DSL differ substantially.

# opensearch-security/roles.yml — OpenSearch Security role definition
# Equivalent to X-Pack role with index privileges

my_app_read_role:
  description: "Read-only access to application indices"
  cluster_permissions:
    - "cluster:monitor/health"
    - "cluster:monitor/state"
  index_permissions:
    - index_patterns:
        - "products*"
        - "users*"
      allowed_actions:
        - "read"
        - "get"
        - "search"
        - "indices:data/read/*"
  tenant_permissions: []

my_app_write_role:
  description: "Write access to application indices"
  cluster_permissions:
    - "cluster:monitor/health"
    - "indices:data/write/bulk"
  index_permissions:
    - index_patterns:
        - "products*"
        - "users*"
        - "logs*"
      allowed_actions:
        - "write"
        - "create"
        - "delete"
        - "crud"
        - "indices:data/write/*"
  tenant_permissions: []

# opensearch-security/roles_mapping.yml — Map roles to users/backends

my_app_read_role:
  backend_roles:
    - "app-readers"     # LDAP/SAML group name
  users:
    - "service-account-readonly"
  hosts: []

my_app_write_role:
  backend_roles:
    - "app-writers"
  users:
    - "logstash"
    - "service-account-writer"
  hosts: []

# opensearch.yml — TLS configuration for OpenSearch Security
# Mirrors X-Pack xpack.security.* settings

plugins.security.ssl.transport.pemcert_filepath: certs/node.pem
plugins.security.ssl.transport.pemkey_filepath: certs/node-key.pem
plugins.security.ssl.transport.pemtrustedcas_filepath: certs/ca.pem
plugins.security.ssl.transport.enforce_hostname_verification: false

plugins.security.ssl.http.enabled: true
plugins.security.ssl.http.pemcert_filepath: certs/node.pem
plugins.security.ssl.http.pemkey_filepath: certs/node-key.pem
plugins.security.ssl.http.pemtrustedcas_filepath: certs/ca.pem

plugins.security.allow_unsafe_democertificates: false
plugins.security.allow_default_init_securityindex: true

plugins.security.authcz.admin_dn:
  - "CN=admin,OU=ops,O=myorg,C=US"

plugins.security.audit.type: internal_opensearch
plugins.security.enable_snapshot_restore_privilege: true

Post-Migration Verification

After the alias cutover, run this verification suite before decommissioning the Elasticsearch cluster. Keep the ES cluster alive for at least 48 hours post-cutover as a rollback target.

#!/usr/bin/env python3
# verify_migration.py — Post-cutover verification suite

import json
import sys
from opensearchpy import OpenSearch

OS_HOST = "https://localhost:9201"
os_client = OpenSearch(OS_HOST, http_auth=("admin", "admin"), verify_certs=False)

CHECKS = []
FAILURES = []


def check(name: str, fn) -> None:
    try:
        result = fn()
        if result is True:
            CHECKS.append(f"  [PASS] {name}")
        else:
            CHECKS.append(f"  [FAIL] {name}: {result}")
            FAILURES.append(name)
    except Exception as e:
        CHECKS.append(f"  [ERROR] {name}: {e}")
        FAILURES.append(name)


# Cluster health
check("Cluster is green", lambda: os_client.cluster.health()["status"] == "green" or "yellow")
check("No relocating shards", lambda: os_client.cluster.health()["relocating_shards"] == 0)
check("No unassigned shards", lambda: os_client.cluster.health()["unassigned_shards"] == 0)

# Index aliases are active
def check_aliases():
    aliases = os_client.cat.aliases(format="json")
    alias_names = {a["alias"] for a in aliases}
    required = {"products-read", "products-write", "users-read", "users-write"}
    missing = required - alias_names
    return True if not missing else f"Missing aliases: {missing}"

check("Required aliases exist", check_aliases)

# Search returns results
def check_search():
    r = os_client.search(index="products-read", body={"query": {"match_all": {}}, "size": 1})
    return r["hits"]["total"]["value"] > 0

check("Search returns results", check_search)

# Write succeeds
def check_write():
    r = os_client.index(
        index="products-write",
        body={"name": "_migration_probe", "ts": "2026-05-17"},
        id="_migration_probe",
        refresh="wait_for",
    )
    os_client.delete(index="products-write", id="_migration_probe")
    return r["result"] in ("created", "updated")

check("Write roundtrip succeeds", check_write)

# ISM policies are active
def check_ism():
    r = os_client.transport.perform_request("GET", "/_plugins/_ism/policies")
    return len(r.get("policies", [])) > 0 or "No ISM policies (acceptable if none migrated)"

check("ISM policies present", check_ism)

# Print results
print("Migration Verification Report")
print("=" * 40)
for line in CHECKS:
    print(line)

print(f"\n{len(CHECKS) - len(FAILURES)}/{len(CHECKS)} checks passed")

if FAILURES:
    print(f"\nFailed checks: {FAILURES}")
    sys.exit(1)
else:
    print("\nAll checks passed — migration verified")

Rollback Plan

A migration without a rollback plan is a gamble. Because the alias cutover is atomic and Elasticsearch remains intact, rolling back is a matter of reversing the alias update — provided writes during the OpenSearch period can be reconciled.

#!/usr/bin/env python3
# rollback.py — Revert alias cutover from OpenSearch back to Elasticsearch
# Run only if post-migration verification fails within the rollback window

from elasticsearch import Elasticsearch
from opensearchpy import OpenSearch
import sys

ES_HOST = "https://localhost:9200"
OS_HOST = "https://localhost:9201"

es = Elasticsearch(ES_HOST, basic_auth=("elastic", "changeme"), verify_certs=False)
os_client = OpenSearch(OS_HOST, http_auth=("admin", "admin"), verify_certs=False)

INDICES = [
    {"index": "products", "read_alias": "products-read", "write_alias": "products-write"},
    {"index": "users",    "read_alias": "users-read",    "write_alias": "users-write"},
    {"index": "events",   "read_alias": "events-read",   "write_alias": "events-write"},
]


def rollback() -> None:
    print("ROLLING BACK: restoring aliases to Elasticsearch...")

    # Step 1: add aliases back to Elasticsearch
    for item in INDICES:
        actions = [
            {"add": {"index": item["index"], "alias": item["read_alias"]}},
            {"add": {"index": item["index"], "alias": item["write_alias"]}},
        ]
        es.indices.update_aliases(body={"actions": actions})
        print(f"  Restored aliases on Elasticsearch: {item['index']}")

    # Step 2: remove aliases from OpenSearch
    for item in INDICES:
        actions = [
            {"remove": {"index": item["index"], "alias": item["read_alias"]}},
            {"remove": {"index": item["index"], "alias": item["write_alias"]}},
        ]
        try:
            os_client.indices.update_aliases(body={"actions": actions})
            print(f"  Removed aliases from OpenSearch: {item['index']}")
        except Exception as e:
            print(f"  WARNING: Could not remove OS alias for {item['index']}: {e}")

    print("\nRollback complete. Traffic is back on Elasticsearch.")
    print("Next step: run remote reindex again to sync writes that happened during OS window.")


if __name__ == "__main__":
    confirm = input("Are you sure you want to roll back? (yes/no): ")
    if confirm.strip().lower() != "yes":
        print("Rollback cancelled.")
        sys.exit(0)
    rollback()

Common Gotchas and Breaking Changes

Further Reading

• OpenSearch — Migrating from Elasticsearch — official migration guide covering version compatibility, snapshot/restore, and rolling upgrades
• OpenSearch Security — Configuration Guide — TLS setup, roles.yml, roles_mapping.yml, and backend authentication
• Index State Management (ISM) — OpenSearch Docs — ISM policy DSL reference and ILM-to-ISM translation guidance
• opensearch-py — GitHub — official Python client with async support and connection pool configuration
• opensearch-js — GitHub — official Node.js/TypeScript client with TypeScript type definitions
• Amazon OpenSearch Service — AWS Big Data Blog — managed service overview and migration from Amazon Elasticsearch Service