Hot Take: OpenSearch 2.0 Will Replace Elasticsearch in 2026 – Migrate Now for Cost Savings

import sys
import time
from elasticsearch import Elasticsearch as ESClient
from opensearchpy import OpenSearch as OSClient
from opensearchpy.helpers import bulk
import logging
from typing import List, Dict, Any

# Configure logging for migration visibility
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

# Configuration - update these values for your environment
ES_HOST = \"https://es-cluster.example.com:9200\"
ES_USER = \"elastic\"
ES_PASS = \"changeme\"
OS_HOST = \"https://os-cluster.example.com:9200\"
OS_USER = \"admin\"
OS_PASS = \"changeme\"
INDEX_PATTERN = \"logs-*\"
BATCH_SIZE = 1000
RETRY_LIMIT = 3

def create_es_client() -> ESClient:
    \"\"\"Initialize Elasticsearch client with error handling\"\"\"
    try:
        client = ESClient(
            hosts=[ES_HOST],
            http_auth=(ES_USER, ES_PASS),
            verify_certs=False,  # Set to True in production with valid certs
            timeout=30
        )
        # Validate connection
        if not client.ping():
            raise ConnectionError(\"Failed to connect to Elasticsearch cluster\")
        logger.info(\"Successfully connected to Elasticsearch cluster\")
        return client
    except Exception as e:
        logger.error(f\"Elasticsearch client creation failed: {str(e)}\")
        sys.exit(1)

def create_os_client() -> OSClient:
    \"\"\"Initialize OpenSearch client with error handling\"\"\"
    try:
        client = OSClient(
            hosts=[OS_HOST],
            http_auth=(OS_USER, OS_PASS),
            verify_certs=False,  # Set to True in production with valid certs
            timeout=30
        )
        # Validate connection
        if not client.ping():
            raise ConnectionError(\"Failed to connect to OpenSearch cluster\")
        logger.info(\"Successfully connected to OpenSearch cluster\")
        return client
    except Exception as e:
        logger.error(f\"OpenSearch client creation failed: {str(e)}\")
        sys.exit(1)

def get_indexes_to_migrate(es_client: ESClient) -> List[str]:
    \"\"\"Retrieve list of indexes matching the pattern to migrate\"\"\"
    try:
        indexes = es_client.indices.get(index=INDEX_PATTERN)
        index_list = list(indexes.keys())
        logger.info(f\"Found {len(index_list)} indexes matching pattern {INDEX_PATTERN}\")
        return index_list
    except Exception as e:
        logger.error(f\"Failed to retrieve indexes: {str(e)}\")
        sys.exit(1)

def migrate_index(es_client: ESClient, os_client: OSClient, index_name: str) -> None:
    \"\"\"Migrate a single index from Elasticsearch to OpenSearch with retry logic\"\"\"
    retry_count = 0
    while retry_count < RETRY_LIMIT:
        try:
            # Check if index exists in OpenSearch, create if not
            if not os_client.indices.exists(index=index_name):
                # Get index settings and mappings from Elasticsearch
                es_index_settings = es_client.indices.get_settings(index=index_name)
                es_index_mappings = es_client.indices.get_mapping(index=index_name)
                # Create index in OpenSearch with same settings/mappings
                os_client.indices.create(
                    index=index_name,
                    body={
                        \"settings\": es_index_settings[index_name][\"settings\"][\"index\"],
                        \"mappings\": es_index_mappings[index_name][\"mappings\"]
                    }
                )
                logger.info(f\"Created index {index_name} in OpenSearch\")

            # Scroll through Elasticsearch index and bulk index to OpenSearch
            page = es_client.search(
                index=index_name,
                scroll=\"2m\",
                size=BATCH_SIZE,
                body={\"query\": {\"match_all\": {}}}
            )
            scroll_id = page[\"_scroll_id\"]
            hits = page[\"hits\"][\"hits\"]

            while len(hits) > 0:
                # Prepare bulk payload
                actions = [
                    {
                        \"_index\": index_name,
                        \"_id\": hit[\"_id\"],
                        \"_source\": hit[\"_source\"]
                    }
                    for hit in hits
                ]
                # Bulk index to OpenSearch
                success, failed = bulk(os_client, actions)
                logger.info(f\"Indexed {success} docs to {index_name}, {len(failed)} failed\")
                # Get next batch
                page = es_client.scroll(scroll_id=scroll_id, scroll=\"2m\")
                scroll_id = page[\"_scroll_id\"]
                hits = page[\"hits\"][\"hits\"]

            # Clear scroll context
            es_client.scroll.clear(scroll_id=scroll_id)
            logger.info(f\"Successfully migrated index {index_name}\")
            return
        except Exception as e:
            retry_count += 1
            logger.warning(f\"Retry {retry_count}/{RETRY_LIMIT} for index {index_name}: {str(e)}\")
            time.sleep(5 * retry_count)
    logger.error(f\"Failed to migrate index {index_name} after {RETRY_LIMIT} retries\")

def validate_migration(es_client: ESClient, os_client: OSClient, indexes: List[str]) -> None:
    \"\"\"Validate document counts match between ES and OpenSearch\"\"\"
    for index in indexes:
        try:
            es_count = es_client.count(index=index)[\"count\"]
            os_count = os_client.count(index=index)[\"count\"]
            if es_count == os_count:
                logger.info(f\"Validation passed for {index}: {es_count} docs\")
            else:
                logger.error(f\"Validation failed for {index}: ES {es_count} vs OS {os_count}\")
        except Exception as e:
            logger.error(f\"Validation error for {index}: {str(e)}\")

if __name__ == \"__main__\":
    logger.info(\"Starting Elasticsearch to OpenSearch migration\")
    es_client = create_es_client()
    os_client = create_os_client()
    indexes = get_indexes_to_migrate(es_client)
    for idx in indexes:
        logger.info(f\"Migrating index: {idx}\")
        migrate_index(es_client, os_client, idx)
    validate_migration(es_client, os_client, indexes)
    logger.info(\"Migration complete\")

version: \"3.8\"

# 3-node OpenSearch 2.0 cluster with security enabled, resource limits, and health checks
# Optimized for production log analytics workloads up to 50TB storage

services:
  opensearch-node1:
    image: opensearchproject/opensearch:2.0.0
    container_name: opensearch-node1
    environment:
      - cluster.name=opensearch-cluster
      - node.name=opensearch-node1
      - discovery.seed_hosts=opensearch-node1,opensearch-node2,opensearch-node3
      - cluster.initial_cluster_manager_nodes=opensearch-node1,opensearch-node2,opensearch-node3
      - bootstrap.memory_lock=true
      - \"OPENSEARCH_JAVA_OPTS=-Xms16g -Xmx16g\"  # Adjust based on instance memory
      - \"OPENSEARCH_INITIAL_ADMIN_PASSWORD=Changeme123!\"  # Change in production
      - plugins.security.disabled=false
      - plugins.security.ssl.http.enabled=true
      - plugins.security.ssl.http.pemcert_filepath=/usr/share/opensearch/config/certs/node1.pem
      - plugins.security.ssl.http.pemkey_filepath=/usr/share/opensearch/config/certs/node1-key.pem
      - plugins.security.ssl.http.pemtrustedcas_filepath=/usr/share/opensearch/config/certs/root-ca.pem
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
    volumes:
      - opensearch-data1:/usr/share/opensearch/data
      - ./certs:/usr/share/opensearch/config/certs:ro
    ports:
      - 9200:9200
      - 9600:9600
    networks:
      - opensearch-net
    restart: unless-stopped
    healthcheck:
      test: [\"CMD\", \"curl\", \"-k\", \"-u\", \"admin:Changeme123!\", \"https://localhost:9200/_cluster/health\"]
      interval: 30s
      timeout: 10s
      retries: 5
    deploy:
      resources:
        limits:
          cpus: \"4\"
          memory: 32G
        reservations:
          cpus: \"2\"
          memory: 24G

  opensearch-node2:
    image: opensearchproject/opensearch:2.0.0
    container_name: opensearch-node2
    environment:
      - cluster.name=opensearch-cluster
      - node.name=opensearch-node2
      - discovery.seed_hosts=opensearch-node1,opensearch-node2,opensearch-node3
      - cluster.initial_cluster_manager_nodes=opensearch-node1,opensearch-node2,opensearch-node3
      - bootstrap.memory_lock=true
      - \"OPENSEARCH_JAVA_OPTS=-Xms16g -Xmx16g\"
      - \"OPENSEARCH_INITIAL_ADMIN_PASSWORD=Changeme123!\"
      - plugins.security.disabled=false
      - plugins.security.ssl.http.enabled=true
      - plugins.security.ssl.http.pemcert_filepath=/usr/share/opensearch/config/certs/node2.pem
      - plugins.security.ssl.http.pemkey_filepath=/usr/share/opensearch/config/certs/node2-key.pem
      - plugins.security.ssl.http.pemtrustedcas_filepath=/usr/share/opensearch/config/certs/root-ca.pem
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
    volumes:
      - opensearch-data2:/usr/share/opensearch/data
      - ./certs:/usr/share/opensearch/config/certs:ro
    networks:
      - opensearch-net
    restart: unless-stopped
    healthcheck:
      test: [\"CMD\", \"curl\", \"-k\", \"-u\", \"admin:Changeme123!\", \"https://localhost:9200/_cluster/health\"]
      interval: 30s
      timeout: 10s
      retries: 5
    deploy:
      resources:
        limits:
          cpus: \"4\"
          memory: 32G
        reservations:
          cpus: \"2\"
          memory: 24G

  opensearch-node3:
    image: opensearchproject/opensearch:2.0.0
    container_name: opensearch-node3
    environment:
      - cluster.name=opensearch-cluster
      - node.name=opensearch-node3
      - discovery.seed_hosts=opensearch-node1,opensearch-node2,opensearch-node3
      - cluster.initial_cluster_manager_nodes=opensearch-node1,opensearch-node2,opensearch-node3
      - bootstrap.memory_lock=true
      - \"OPENSEARCH_JAVA_OPTS=-Xms16g -Xmx16g\"
      - \"OPENSEARCH_INITIAL_ADMIN_PASSWORD=Changeme123!\"
      - plugins.security.disabled=false
      - plugins.security.ssl.http.enabled=true
      - plugins.security.ssl.http.pemcert_filepath=/usr/share/opensearch/config/certs/node3.pem
      - plugins.security.ssl.http.pemkey_filepath=/usr/share/opensearch/config/certs/node3-key.pem
      - plugins.security.ssl.http.pemtrustedcas_filepath=/usr/share/opensearch/config/certs/root-ca.pem
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
    volumes:
      - opensearch-data3:/usr/share/opensearch/data
      - ./certs:/usr/share/opensearch/config/certs:ro
    networks:
      - opensearch-net
    restart: unless-stopped
    healthcheck:
      test: [\"CMD\", \"curl\", \"-k\", \"-u\", \"admin:Changeme123!\", \"https://localhost:9200/_cluster/health\"]
      interval: 30s
      timeout: 10s
      retries: 5
    deploy:
      resources:
        limits:
          cpus: \"4\"
          memory: 32G
        reservations:
          cpus: \"2\"
          memory: 24G

  opensearch-dashboards:
    image: opensearchproject/opensearch-dashboards:2.0.0
    container_name: opensearch-dashboards
    environment:
      - OPENSEARCH_HOSTS=[\"https://opensearch-node1:9200\",\"https://opensearch-node2:9200\",\"https://opensearch-node3:9200\"]
      - OPENSEARCH_USERNAME=admin
      - OPENSEARCH_PASSWORD=Changeme123!
      - OPENSEARCH_SSL_VERIFICATIONMODE=none
    ports:
      - 5601:5601
    networks:
      - opensearch-net
    restart: unless-stopped
    depends_on:
      - opensearch-node1
      - opensearch-node2
      - opensearch-node3
    healthcheck:
      test: [\"CMD\", \"curl\", \"-f\", \"http://localhost:5601/api/status\"]
      interval: 30s
      timeout: 10s
      retries: 5

volumes:
  opensearch-data1:
  opensearch-data2:
  opensearch-data3:

networks:
  opensearch-net:
    driver: bridge

package main

import (
    \"context\"
    \"crypto/tls\"
    \"fmt\"
    \"log\"
    \"net/http\"
    \"os\"
    \"sync\"
    \"time\"

    \"github.com/opensearch-project/opensearch-go/v2\"
    \"github.com/elastic/go-elasticsearch/v8\"
)

// Config holds connection details for both clusters
type Config struct {
    ESHost   string
    ESUser   string
    ESPass   string
    OSHost   string
    OSUser   string
    OSPass   string
    Index    string
    Query    map[string]interface{}
    Runs     int
}

// BenchmarkResult holds latency metrics for a cluster
type BenchmarkResult struct {
    ClusterName string
    P50Latency  time.Duration
    P99Latency  time.Duration
    ErrorRate   float64
}

func main() {
    // Load config from environment variables
    cfg := Config{
        ESHost:   getEnv(\"ES_HOST\", \"https://localhost:9200\"),
        ESUser:   getEnv(\"ES_USER\", \"elastic\"),
        ESPass:   getEnv(\"ES_PASS\", \"changeme\"),
        OSHost:   getEnv(\"OS_HOST\", \"https://localhost:9201\"),
        OSUser:   getEnv(\"OS_USER\", \"admin\"),
        OSPass:   getEnv(\"OS_PASS\", \"changeme\"),
        Index:    getEnv(\"INDEX\", \"logs-2024-05\"),
        Runs:     1000,
        Query: map[string]interface{}{
            \"query\": map[string]interface{}{
                \"match\": map[string]interface{}{
                    \"message\": \"error\",
                },
            },
        },
    }

    // Run benchmarks in parallel
    var wg sync.WaitGroup
    results := make(chan BenchmarkResult, 2)

    wg.Add(1)
    go func() {
        defer wg.Done()
        res, err := runBenchmark(\"Elasticsearch\", cfg.ESHost, cfg.ESUser, cfg.ESPass, cfg.Index, cfg.Query, cfg.Runs)
        if err != nil {
            log.Printf(\"Elasticsearch benchmark failed: %v\", err)
            return
        }
        results <- res
    }()

    wg.Add(1)
    go func() {
        defer wg.Done()
        res, err := runBenchmark(\"OpenSearch\", cfg.OSHost, cfg.OSUser, cfg.OSPass, cfg.Index, cfg.Query, cfg.Runs)
        if err != nil {
            log.Printf(\"OpenSearch benchmark failed: %v\", err)
            return
        }
        results <- res
    }()

    // Wait for benchmarks to complete
    go func() {
        wg.Wait()
        close(results)
    }()

    // Print results
    fmt.Println(\"=== Search Latency Benchmark Results ===\")
    for res := range results {
        fmt.Printf(\"\\nCluster: %s\\n\", res.ClusterName)
        fmt.Printf(\"P50 Latency: %v\\n\", res.P50Latency)
        fmt.Printf(\"P99 Latency: %v\\n\", res.P99Latency)
        fmt.Printf(\"Error Rate: %.2f%%\\n\", res.ErrorRate*100)
    }
}

func runBenchmark(clusterName, host, user, pass, index string, query map[string]interface{}, runs int) (BenchmarkResult, error) {
    var client interface{}
    var httpClient *http.Client

    // Create HTTP client with TLS config (disable cert verification for testing)
    tlsConfig := &tls.Config{InsecureSkipVerify: true}
    httpClient = &http.Client{
        Transport: &http.Transport{TLSClientConfig: tlsConfig},
        Timeout:   10 * time.Second,
    }

    // Initialize client based on cluster type
    if clusterName == \"Elasticsearch\" {
        esClient, err := elasticsearch.NewClient(elasticsearch.Config{
            Addresses: []string{host},
            Username:  user,
            Password:  pass,
            Transport: httpClient.Transport,
        })
        if err != nil {
            return BenchmarkResult{}, fmt.Errorf(\"failed to create ES client: %w\", err)
        }
        client = esClient
    } else {
        osClient, err := opensearch.NewClient(opensearch.Config{
            Addresses: []string{host},
            Username:  user,
            Password:  pass,
            Transport: httpClient.Transport,
        })
        if err != nil {
            return BenchmarkResult{}, fmt.Errorf(\"failed to create OS client: %w\", err)
        }
        client = osClient
    }

    // Collect latency metrics
    latencies := make([]time.Duration, 0, runs)
    errors := 0

    for i := 0; i < runs; i++ {
        start := time.Now()
        var err error

        if clusterName == \"Elasticsearch\" {
            _, err = client.(*elasticsearch.Client).Search(
                client.(*elasticsearch.Client).Search.WithContext(context.Background()),
                client.(*elasticsearch.Client).Search.WithIndex(index),
                client.(*elasticsearch.Client).Search.WithBody(wrapQuery(query)),
            )
        } else {
            _, err = client.(*opensearch.Client).Search(
                client.(*opensearch.Client).Search.WithContext(context.Background()),
                client.(*opensearch.Client).Search.WithIndex(index),
                client.(*opensearch.Client).Search.WithBody(wrapQuery(query)),
            )
        }

        latency := time.Since(start)
        if err != nil {
            errors++
        } else {
            latencies = append(latencies, latency)
        }
    }

    // Calculate percentiles (simplified, for demo purposes)
    // In production use a proper percentile library like github.com/influxdata/go-percentile
    p50 := calculatePercentile(latencies, 50)
    p99 := calculatePercentile(latencies, 99)
    errorRate := float64(errors) / float64(runs)

    return BenchmarkResult{
        ClusterName: clusterName,
        P50Latency:  p50,
        P99Latency:  p99,
        ErrorRate:   errorRate,
    }, nil
}

func getEnv(key, defaultVal string) string {
    if val, ok := os.LookupEnv(key); ok {
        return val
    }
    return defaultVal
}

func wrapQuery(query map[string]interface{}) *map[string]interface{} {
    return &query
}

func calculatePercentile(latencies []time.Duration, percentile int) time.Duration {
    // Simplified percentile calculation: sort and pick index
    // Note: Use a proper library for production benchmarks
    n := len(latencies)
    if n == 0 {
        return 0
    }
    // Sort latencies (ascending)
    for i := 0; i < n; i++ {
        for j := i + 1; j < n; j++ {
            if latencies[i] > latencies[j] {
                latencies[i], latencies[j] = latencies[j], latencies[i]
            }
        }
    }
    idx := int(float64(percentile)/100.0*float64(n)) - 1
    if idx < 0 {
        idx = 0
    }
    if idx >= n {
        idx = n - 1
    }
    return latencies[idx]
}

docker run -it --rm opensearchproject/opensearch-migration-assistant:2.0.0 \
  --source-host https://es-cluster.example.com:9200 \
  --source-user elastic \
  --source-pass changeme \
  --report-format json \
  --output-file migration-report.json

curl -k -u admin:changeme -X PUT \"https://os-cluster.example.com:9200/_cluster/settings\" \
  -H \"Content-Type: application/json\" \
  -d '{
    \"persistent\": {
      \"indices.breaker.total.limit\": \"75%\",
      \"indices.breaker.fielddata.limit\": \"30%\",
      \"network.breaker.inflight_requests.limit\": \"40%\"
    }
  }'

curl -k -u admin:changeme -X PUT \"https://os-cluster.example.com:9200/_snapshot/minio-backup\" \
  -H \"Content-Type: application/json\" \
  -d '{
    \"type\": \"s3\",
    \"settings\": {
      \"endpoint\": \"minio.example.com:9000\",
      \"bucket\": \"opensearch-snapshots\",
      \"access_key\": \"minioadmin\",
      \"secret_key\": \"miniosecret\",
      \"protocol\": \"http\"
    }
  }'