Hot Take: You Should Ditch Elasticsearch for OpenSearch 2.12 in 2026 – Save Licensing Costs Now

import json
import logging
from elasticsearch import Elasticsearch as ESClient
from opensearchpy import OpenSearch as OSClient
from opensearchpy.helpers import bulk
import sys

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

# Configuration - replace with your actual cluster endpoints
ES_ENDPOINT = \"https://es-cluster.example.com:9200\"
OS_ENDPOINT = \"https://os-cluster.example.com:9200\"
ES_INDEX = \"product_catalog_v3\"
OS_INDEX = \"product_catalog_v3\"
BATCH_SIZE = 1000

def create_es_client():
    \"\"\"Initialize Elasticsearch client with error handling for auth/connectivity\"\"\"
    try:
        client = ESClient(
            [ES_ENDPOINT],
            basic_auth=(\"es_admin\", \"es_password\"),  # Use env vars in prod
            verify_certs=True,
            ca_certs=\"/path/to/es_ca.pem\"  # Optional for self-signed certs
        )
        # Test connection
        if not client.ping():
            raise ConnectionError(\"Elasticsearch cluster is unreachable\")
        logger.info(f\"Connected to Elasticsearch index: {ES_INDEX}\")
        return client
    except Exception as e:
        logger.error(f\"ES client init failed: {str(e)}\")
        sys.exit(1)

def create_os_client():
    \"\"\"Initialize OpenSearch client with matching auth config\"\"\"
    try:
        client = OSClient(
            [OS_ENDPOINT],
            basic_auth=(\"os_admin\", \"os_password\"),  # Use env vars in prod
            verify_certs=True,
            ca_certs=\"/path/to/os_ca.pem\"
        )
        if not client.ping():
            raise ConnectionError(\"OpenSearch cluster is unreachable\")
        logger.info(f\"Connected to OpenSearch index: {OS_INDEX}\")
        return client
    except Exception as e:
        logger.error(f\"OS client init failed: {str(e)}\")
        sys.exit(1)

def migrate_index(es_client, os_client):
    \"\"\"Reindex all documents from ES to OS with batch processing and error handling\"\"\"
    try:
        # Get total doc count for progress tracking
        total_docs = es_client.count(index=ES_INDEX)[\"count\"]
        logger.info(f\"Starting migration of {total_docs} documents from {ES_INDEX} to {OS_INDEX}\")

        # Scroll through ES index in batches
        page = es_client.search(
            index=ES_INDEX,
            scroll=\"2m\",
            size=BATCH_SIZE,
            body={\"query\": {\"match_all\": {}}}
        )
        scroll_id = page[\"_scroll_id\"]
        hits = page[\"hits\"][\"hits\"]

        migrated_count = 0
        while len(hits) > 0:
            # Prepare batch for OpenSearch bulk insert
            actions = [
                {
                    \"_index\": OS_INDEX,
                    \"_id\": doc[\"_id\"],
                    \"_source\": doc[\"_source\"]
                }
                for doc in hits
            ]

            # Bulk insert with error handling
            success, failed = bulk(os_client, actions, stats_only=True)
            migrated_count += success
            logger.info(f\"Migrated {migrated_count}/{total_docs} documents (batch success: {success})\")

            if failed > 0:
                logger.warning(f\"Failed to migrate {failed} documents in current batch\")

            # Get next batch
            page = es_client.scroll(scroll_id=scroll_id, scroll=\"2m\")
            scroll_id = page[\"_scroll_id\"]
            hits = page[\"hits\"][\"hits\"]

        logger.info(f\"Migration complete: {migrated_count} documents migrated to {OS_INDEX}\")
        return migrated_count
    except Exception as e:
        logger.error(f\"Migration failed: {str(e)}\")
        sys.exit(1)
    finally:
        if \"scroll_id\" in locals():
            es_client.clear_scroll(scroll_id=scroll_id)

if __name__ == \"__main__\":
    logger.info(\"Starting Elasticsearch to OpenSearch 2.12 migration\")
    es_client = create_es_client()
    os_client = create_os_client()
    # Create OS index with matching mapping (optional, OS auto-creates)
    if not os_client.indices.exists(OS_INDEX):
        # Copy mapping from ES
        es_mapping = es_client.indices.get_mapping(index=ES_INDEX)
        os_client.indices.create(index=OS_INDEX, body=es_mapping[ES_INDEX])
        logger.info(f\"Created OpenSearch index {OS_INDEX} with ES mapping\")
    migrate_index(es_client, os_client)

const { Client: ESClient } = require('@elastic/elasticsearch');
const { Client: OSClient } = require('@opensearch-project/opensearch');
const { performance } = require('perf_hooks');

// Configuration
const ES_NODE = 'https://es-cluster.example.com:9200';
const OS_NODE = 'https://os-cluster.example.com:9200';
const INDEX = 'product_catalog_v3';
const QUERY = {
  query: {
    multi_match: {
      query: 'wireless headphones',
      fields: ['name', 'description', 'category']
    }
  },
  size: 20
};
const ITERATIONS = 1000;

// Initialize clients with connection pooling
const esClient = new ESClient({
  node: ES_NODE,
  auth: { username: 'es_admin', password: 'es_password' },
  tls: { rejectUnauthorized: true }
});

const osClient = new OSClient({
  node: OS_NODE,
  auth: { username: 'os_admin', password: 'os_password' },
  tls: { rejectUnauthorized: true }
});

/**
 * Run a single query against a client and return latency in ms
 * @param {Object} client - ES or OS client
 * @returns {number} Latency in milliseconds
 */
async function runQuery(client) {
  const start = performance.now();
  try {
    const response = await client.search({
      index: INDEX,
      body: QUERY
    });
    if (response.meta.statusCode !== 200) {
      throw new Error(`Query failed with status ${response.meta.statusCode}`);
    }
    return performance.now() - start;
  } catch (error) {
    console.error(`Query error: ${error.message}`);
    return -1; // Indicate failure
  }
}

/**
 * Run benchmark for a given client type
 * @param {string} clientType - 'elasticsearch' or 'opensearch'
 * @returns {Object} Benchmark stats
 */
async function runBenchmark(clientType) {
  const client = clientType === 'elasticsearch' ? esClient : osClient;
  const latencies = [];
  let successCount = 0;
  let errorCount = 0;

  console.log(`Running ${ITERATIONS} iterations for ${clientType}...`);

  for (let i = 0; i < ITERATIONS; i++) {
    const latency = await runQuery(client);
    if (latency >= 0) {
      latencies.push(latency);
      successCount++;
    } else {
      errorCount++;
    }

    // Log progress every 100 iterations
    if (i % 100 === 0 && i > 0) {
      console.log(`Progress: ${i}/${ITERATIONS} (${successCount} success, ${errorCount} errors)`);
    }
  }

  // Calculate stats
  latencies.sort((a, b) => a - b);
  const avg = latencies.reduce((sum, val) => sum + val, 0) / latencies.length;
  const p50 = latencies[Math.floor(latencies.length * 0.5)];
  const p95 = latencies[Math.floor(latencies.length * 0.95)];
  const p99 = latencies[Math.floor(latencies.length * 0.99)];

  return { clientType, avg, p50, p95, p99, successCount, errorCount };
}

async function main() {
  try {
    // Verify connections
    await esClient.ping();
    console.log('Elasticsearch connection verified');
    await osClient.ping();
    console.log('OpenSearch 2.12 connection verified');

    // Run benchmarks
    const esBenchmark = await runBenchmark('elasticsearch');
    const osBenchmark = await runBenchmark('opensearch');

    // Print results
    console.log('\n=== Benchmark Results (1000 iterations) ===');
    console.log('Elasticsearch 8.14:');
    console.log(`  Avg: ${esBenchmark.avg.toFixed(2)}ms`);
    console.log(`  P50: ${esBenchmark.p50.toFixed(2)}ms`);
    console.log(`  P95: ${esBenchmark.p95.toFixed(2)}ms`);
    console.log(`  P99: ${esBenchmark.p99.toFixed(2)}ms`);
    console.log(`  Success Rate: ${(esBenchmark.successCount / ITERATIONS * 100).toFixed(2)}%`);

    console.log('\nOpenSearch 2.12:');
    console.log(`  Avg: ${osBenchmark.avg.toFixed(2)}ms`);
    console.log(`  P50: ${osBenchmark.p50.toFixed(2)}ms`);
    console.log(`  P95: ${osBenchmark.p95.toFixed(2)}ms`);
    console.log(`  P99: ${osBenchmark.p99.toFixed(2)}ms`);
    console.log(`  Success Rate: ${(osBenchmark.successCount / ITERATIONS * 100).toFixed(2)}%`);

    console.log(`\nOpenSearch 2.12 P99 latency is ${(osBenchmark.p99 / esBenchmark.p99 * 100).toFixed(1)}% of Elasticsearch P99`);
  } catch (error) {
    console.error(`Benchmark failed: ${error.message}`);
    process.exit(1);
  } finally {
    await esClient.close();
    await osClient.close();
  }
}

main();

package main

import (
    \"context\"
    \"encoding/json\"
    \"fmt\"
    \"log\"
    \"os\"
    \"time\"

    \"github.com/opensearch-project/opensearch-go/v2\"
    \"github.com/opensearch-project/opensearch-go/v2/opensearchapi\"
)

const (
    openSearchEndpoint = \"https://localhost:9200\"
    indexName          = \"product_embeddings\"
    vectorDim          = 384 // All-MiniLM-L6-v2 embedding dimension
)

// ProductEmbedding represents a product with its vector embedding
type ProductEmbedding struct {
    ID          string    `json:\"id\"`
    Name        string    `json:\"name\"`
    Description string    `json:\"description\"`
    Embedding   []float32 `json:\"embedding\"`
    Price       float64   `json:\"price\"`
    Category    string    `json:\"category\"`
}

func main() {
    // Initialize OpenSearch client with security config
    client, err := opensearch.NewClient(opensearch.Config{
        Addresses: []string{openSearchEndpoint},
        Username:  \"admin\",
        Password:  \"admin\", // Use env vars in production
        TLS: opensearch.TLSConfig{
            VerifyCerts: false, // Set to true with proper CA in prod
        },
    })
    if err != nil {
        log.Fatalf(\"Failed to create OpenSearch client: %v\", err)
    }

    // Verify cluster health
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    healthResp, err := client.Cluster.Health(
        client.Cluster.Health.WithContext(ctx),
    )
    if err != nil {
        log.Fatalf(\"Failed to check cluster health: %v\", err)
    }
    defer healthResp.Body.Close()
    if healthResp.StatusCode != 200 {
        log.Fatalf(\"Cluster health check failed: status %d\", healthResp.StatusCode)
    }
    log.Println(\"OpenSearch 2.12 cluster health verified\")

    // Create index with vector search mapping
    createIndexResp, err := client.Indices.Create(
        indexName,
        client.Indices.Create.WithBody(&map[string]interface{}{
            \"mappings\": map[string]interface{}{
                \"properties\": map[string]interface{}{
                    \"embedding\": map[string]interface{}{
                        \"type\":      \"knn_vector\",
                        \"dimension\": vectorDim,
                        \"method\": map[string]interface{}{
                            \"name\":       \"hnsw\",
                            \"space_type\": \"cosine\",
                            \"engine\":     \"nmslib\",
                        },
                    },
                    \"name\":       map[string]interface{}{\"type\": \"text\"},
                    \"description\": map[string]interface{}{\"type\": \"text\"},
                    \"price\":      map[string]interface{}{\"type\": \"double\"},
                    \"category\":   map[string]interface{}{\"type\": \"keyword\"},
                },
            },
        }),
    )
    if err != nil {
        log.Fatalf(\"Failed to create index: %v\", err)
    }
    defer createIndexResp.Body.Close()
    log.Printf(\"Created index %s with knn_vector mapping\", indexName)

    // Index sample product embeddings
    sampleProducts := []ProductEmbedding{
        {
            ID:          \"prod_001\",
            Name:        \"Wireless Noise Cancelling Headphones\",
            Description: \"Over-ear wireless headphones with active noise cancellation\",
            Embedding:   generateRandomEmbedding(vectorDim), // In prod, use actual model
            Price:       199.99,
            Category:    \"Electronics\",
        },
        {
            ID:          \"prod_002\",
            Name:        \"True Wireless Earbuds\",
            Description: \"Compact in-ear earbuds with 24h battery life\",
            Embedding:   generateRandomEmbedding(vectorDim),
            Price:       89.99,
            Category:    \"Electronics\",
        },
    }

    for _, product := range sampleProducts {
        productJSON, err := json.Marshal(product)
        if err != nil {
            log.Printf(\"Failed to marshal product %s: %v\", product.ID, err)
            continue
        }

        indexResp, err := client.Index(
            indexName,
            &productJSON,
            client.Index.WithDocumentID(product.ID),
            client.Index.WithRefresh(\"true\"),
        )
        if err != nil {
            log.Printf(\"Failed to index product %s: %v\", product.ID, err)
            continue
        }
        defer indexResp.Body.Close()
        log.Printf(\"Indexed product %s: %s\", product.ID, product.Name)
    }

    // Run k-NN vector search
    queryEmbedding := generateRandomEmbedding(vectorDim) // In prod, embed query text
    searchResp, err := client.Search(
        client.Search.WithIndex(indexName),
        client.Search.WithBody(&map[string]interface{}{
            \"query\": map[string]interface{}{
                \"knn\": map[string]interface{}{
                    \"embedding\": map[string]interface{}{
                        \"vector\":       queryEmbedding,
                        \"k\":            2,
                        \"similarity\":   0.7,
                    },
                },
            },
        }),
    )
    if err != nil {
        log.Fatalf(\"Vector search failed: %v\", err)
    }
    defer searchResp.Body.Close()

    var searchResult map[string]interface{}
    if err := json.NewDecoder(searchResp.Body).Decode(&searchResult); err != nil {
        log.Fatalf(\"Failed to decode search response: %v\", err)
    }

    hits := searchResult[\"hits\"].(map[string]interface{})[\"hits\"].([]interface{})
    log.Printf(\"Found %d matching products for vector query\", len(hits))
    for _, hit := range hits {
        hitMap := hit.(map[string]interface{})
        source := hitMap[\"_source\"].(map[string]interface{})
        log.Printf(\"  - %s (score: %.2f)\", source[\"name\"], hitMap[\"_score\"])
    }
}

// generateRandomEmbedding creates a random embedding for demo purposes
func generateRandomEmbedding(dim int) []float32 {
    embedding := make([]float32, dim)
    for i := range embedding {
        embedding[i] = float32(i) / float32(dim) // Deterministic for demo
    }
    return embedding
}

opensearch-compatibility-checker \
  --es-endpoint https://es-cluster.example.com:9200 \
  --es-username es_admin \
  --es-password es_password \
  --report-format json \
  --output es_compat_report.json

from langchain.vectorstores import OpenSearchVectorSearch
from langchain.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name=\"all-MiniLM-L6-v2\")
vector_store = OpenSearchVectorSearch(
    opensearch_url=\"https://os-cluster.example.com:9200\",
    index_name=\"rag_documents\",
    embedding_function=embeddings,
    use_ssl=True,
    verify_certs=True
)

curl -X PUT \"https://os-cluster.example.com:9200/_plugins/_notifications/monitors\" \
  -H \"Content-Type: application/json\" \
  -u admin:admin \
  -d '{
    \"name\": \"High Search Latency Alert\",
    \"enabled\": true,
    \"schedule\": { \"period\": { \"interval\": 1, \"unit\": \"MINUTES\" } },
    \"inputs\": [{
      \"search\": {
        \"indices\": [\"product_catalog_v3\"],
        \"query\": { \"range\": { \"latency_ms\": { \"gt\": 200 } } }
      }
    }],
    \"triggers\": [{
      \"name\": \"Latency Trigger\",
      \"severity\": \"CRITICAL\",
      \"condition\": { \"script\": { \"source\": \"ctx.results[0].hits.total.value > 0\" } },
      \"actions\": [{
        \"name\": \"Slack Alert\",
        \"destination\": { \"slack\": { \"url\": \"https://hooks.slack.com/services/xxx/yyy/zzz\" } },
        \"message\": { \"text\": \"High search latency detected: {{ctx.results[0].hits.total.value}} requests over 200ms\" }
      }]
    }]
  }'