Elasticsearch 9 vs. OpenSearch 3: Full-Text Search Latency and Storage Costs for 10TB Log Datasets

import json
import time
import random
import logging
from datetime import datetime, timezone
from elasticsearch import Elasticsearch, helpers
from elasticsearch.exceptions import ConnectionError, RequestError, TransportError

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

# Benchmark configuration - matched to OpenSearch ingestion script
ES_HOST = "https://es9-cluster.example.com:9200"
ES_API_KEY = "VnVhQ2ZHY0JDZ0VxQ0JQa0lqV0t6ZG9JOkxTQ0xQV0RKV1BJUmRFSjQ0T1JU"
BATCH_SIZE = 2000  # Optimal for 10TB log throughput per benchmark
MAX_RETRIES = 5
RETRY_DELAY = 2  # Seconds between retries
INDEX_NAME = "logs-10tb-2024.05"
LOG_FILE_PATH = "/data/10tb-app-logs.ndjson"  # NDJSON with 1KB average log lines

def generate_log_batch(file_handle, batch_size):
    """Yield batches of log documents from NDJSON file for bulk ingest"""
    batch = []
    for line_num, line in enumerate(file_handle, 1):
        if not line.strip():
            continue
        try:
            log_doc = json.loads(line)
            # Add benchmark metadata to track ingestion source
            log_doc["@ingestion_ts"] = datetime.now(timezone.utc).isoformat()
            log_doc["@benchmark_run"] = "es9-10tb-ingest-001"
            batch.append(log_doc)
            if len(batch) >= batch_size:
                yield batch
                batch = []
        except json.JSONDecodeError as e:
            logger.warning(f"Skipping invalid JSON at line {line_num}: {str(e)}")
    if batch:  # Yield remaining partial batch
        yield batch

def ingest_to_elasticsearch():
    """Bulk ingest 10TB of logs to Elasticsearch 9 with retry logic and error handling"""
    # Initialize ES client with benchmark-optimized settings
    es_client = Elasticsearch(
        ES_HOST,
        api_key=ES_API_KEY,
        request_timeout=30,
        retry_on_timeout=True,
        max_retries=3
    )

    # Verify cluster health before ingestion
    try:
        health = es_client.cluster.health(index=INDEX_NAME, wait_for_status="yellow")
        logger.info(f"Cluster health: {health['status']}, active shards: {health['active_shards']}")
    except ConnectionError:
        logger.error("Failed to connect to Elasticsearch cluster")
        return
    except RequestError as e:
        if e.error == "index_not_found_exception":
            # Create index with benchmark-matching settings (same as OpenSearch)
            es_client.indices.create(
                index=INDEX_NAME,
                body={
                    "settings": {
                        "number_of_shards": 12,
                        "number_of_replicas": 1,
                        "refresh_interval": "30s",  # Match OpenSearch config
                        "codec": "best_compression"  # ZSTD compression for fair comparison
                    },
                    "mappings": {
                        "properties": {
                            "timestamp": {"type": "date"},
                            "message": {"type": "text", "analyzer": "standard"},
                            "level": {"type": "keyword"},
                            "service": {"type": "keyword"}
                        }
                    }
                }
            )
            logger.info(f"Created index {INDEX_NAME} with benchmark settings")
        else:
            logger.error(f"Index setup failed: {str(e)}")
            return

    total_ingested = 0
    start_time = time.time()

    with open(LOG_FILE_PATH, "r") as log_file:
        for batch_num, batch in enumerate(generate_log_batch(log_file, BATCH_SIZE), 1):
            retry_count = 0
            while retry_count < MAX_RETRIES:
                try:
                    # Prepare bulk ingest actions
                    actions = [
                        {"_index": INDEX_NAME, "_source": doc}
                        for doc in batch
                    ]
                    # Use helpers.bulk for optimized ingestion
                    success, failed = helpers.bulk(
                        es_client,
                        actions,
                        stats_only=True,
                        raise_on_error=False
                    )
                    total_ingested += success
                    logger.info(f"Batch {batch_num}: Ingested {success} docs, Failed {len(failed)}")
                    if failed:
                        logger.warning(f"Failed docs sample: {failed[:3]}")
                    break  # Exit retry loop on success
                except (ConnectionError, TransportError) as e:
                    retry_count += 1
                    logger.warning(f"Batch {batch_num} failed (attempt {retry_count}/{MAX_RETRIES}): {str(e)}")
                    time.sleep(RETRY_DELAY * retry_count)  # Exponential backoff
                except RequestError as e:
                    logger.error(f"Batch {batch_num} failed with request error: {str(e)}")
                    break  # Non-retryable error
            else:
                logger.error(f"Batch {batch_num} failed after {MAX_RETRIES} retries")

    elapsed_time = time.time() - start_time
    throughput = total_ingested / elapsed_time
    logger.info(f"Ingestion complete. Total docs: {total_ingested}, Time: {elapsed_time:.2f}s, Throughput: {throughput:.2f} docs/s")

if __name__ == "__main__":
    ingest_to_elasticsearch()

import json
import time
import random
import logging
from datetime import datetime, timezone
from opensearchpy import OpenSearch, helpers
from opensearchpy.exceptions import ConnectionError, RequestError, TransportError

# Configure logging for ingestion metrics (matched to ES9 script)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Benchmark configuration - identical to Elasticsearch script for fair comparison
OS_HOST = "https://os3-cluster.example.com:9200"
OS_USER = "admin"
OS_PASSWORD = "OpenSearch_Admin_2024!"
BATCH_SIZE = 2000  # Matched to ES9 batch size
MAX_RETRIES = 5
RETRY_DELAY = 2  # Seconds between retries
INDEX_NAME = "logs-10tb-2024.05"  # Same index name as ES9
LOG_FILE_PATH = "/data/10tb-app-logs.ndjson"  # Identical log dataset

def generate_log_batch(file_handle, batch_size):
    """Yield batches of log documents from NDJSON file for bulk ingest (identical to ES9)"""
    batch = []
    for line_num, line in enumerate(file_handle, 1):
        if not line.strip():
            continue
        try:
            log_doc = json.loads(line)
            # Add benchmark metadata to track ingestion source
            log_doc["@ingestion_ts"] = datetime.now(timezone.utc).isoformat()
            log_doc["@benchmark_run"] = "os3-10tb-ingest-001"
            batch.append(log_doc)
            if len(batch) >= batch_size:
                yield batch
                batch = []
        except json.JSONDecodeError as e:
            logger.warning(f"Skipping invalid JSON at line {line_num}: {str(e)}")
    if batch:  # Yield remaining partial batch
        yield batch

def ingest_to_opensearch():
    """Bulk ingest 10TB of logs to OpenSearch 3 with retry logic and error handling"""
    # Initialize OpenSearch client with benchmark-optimized settings
    os_client = OpenSearch(
        hosts=[OS_HOST],
        http_auth=(OS_USER, OS_PASSWORD),
        use_ssl=True,
        verify_certs=False,  # Internal benchmark cluster, disable for throughput
        request_timeout=30,
        retry_on_timeout=True,
        max_retries=3
    )

    # Verify cluster health before ingestion
    try:
        health = os_client.cluster.health(index=INDEX_NAME, wait_for_status="yellow")
        logger.info(f"Cluster health: {health['status']}, active shards: {health['active_shards']}")
    except ConnectionError:
        logger.error("Failed to connect to OpenSearch cluster")
        return
    except RequestError as e:
        if e.error == "index_not_found_exception":
            # Create index with identical settings to Elasticsearch 9
            os_client.indices.create(
                index=INDEX_NAME,
                body={
                    "settings": {
                        "number_of_shards": 12,
                        "number_of_replicas": 1,
                        "refresh_interval": "30s",  # Matched to ES9
                        "codec": "best_compression"  # ZSTD compression for fair comparison
                    },
                    "mappings": {
                        "properties": {
                            "timestamp": {"type": "date"},
                            "message": {"type": "text", "analyzer": "standard"},
                            "level": {"type": "keyword"},
                            "service": {"type": "keyword"}
                        }
                    }
                }
            )
            logger.info(f"Created index {INDEX_NAME} with benchmark settings")
        else:
            logger.error(f"Index setup failed: {str(e)}")
            return

    total_ingested = 0
    start_time = time.time()

    with open(LOG_FILE_PATH, "r") as log_file:
        for batch_num, batch in enumerate(generate_log_batch(log_file, BATCH_SIZE), 1):
            retry_count = 0
            while retry_count < MAX_RETRIES:
                try:
                    # Prepare bulk ingest actions
                    actions = [
                        {"_index": INDEX_NAME, "_source": doc}
                        for doc in batch
                    ]
                    # Use helpers.bulk for optimized ingestion
                    success, failed = helpers.bulk(
                        os_client,
                        actions,
                        stats_only=True,
                        raise_on_error=False
                    )
                    total_ingested += success
                    logger.info(f"Batch {batch_num}: Ingested {success} docs, Failed {len(failed)}")
                    if failed:
                        logger.warning(f"Failed docs sample: {failed[:3]}")
                    break  # Exit retry loop on success
                except (ConnectionError, TransportError) as e:
                    retry_count += 1
                    logger.warning(f"Batch {batch_num} failed (attempt {retry_count}/{MAX_RETRIES}): {str(e)}")
                    time.sleep(RETRY_DELAY * retry_count)  # Exponential backoff
                except RequestError as e:
                    logger.error(f"Batch {batch_num} failed with request error: {str(e)}")
                    break  # Non-retryable error
            else:
                logger.error(f"Batch {batch_num} failed after {MAX_RETRIES} retries")

    elapsed_time = time.time() - start_time
    throughput = total_ingested / elapsed_time
    logger.info(f"Ingestion complete. Total docs: {total_ingested}, Time: {elapsed_time:.2f}s, Throughput: {throughput:.2f} docs/s")

if __name__ == "__main__":
    ingest_to_opensearch()

import time
import json
import logging
import random
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch
from opensearchpy import OpenSearch
from statistics import mean, median, pstdev
import csv

# Benchmark configuration
ES_HOST = "https://es9-cluster.example.com:9200"
ES_API_KEY = "VnVhQ2ZHY0JDZ0VxQ0JQa0lqV0t6ZG9JOkxTQ0xQV0RKV1BJUmRFSjQ0T1JU"
OS_HOST = "https://os3-cluster.example.com:9200"
OS_USER = "admin"
OS_PASSWORD = "OpenSearch_Admin_2024!"
INDEX_NAME = "logs-10tb-2024.05"
QUERY_COUNT = 10000  # Total queries per run
CONCURRENT_WORKERS = 8  # Matches production log search concurrency
OUTPUT_CSV = "search_benchmark_results.csv"

# 10 representative full-text search queries for log datasets
SEARCH_QUERIES = [
    {"query": {"match": {"message": "timeout connection pool"}}},
    {"query": {"match": {"message": "500 internal server error"}}},
    {"query": {"match": {"message": "user authentication failed"}}},
    {"query": {"match": {"message": "database connection refused"}}},
    {"query": {"match": {"message": "rate limit exceeded"}}},
    {"query": {"match_phrase": {"message": "failed to process request"}}},
    {"query": {"bool": {"must": [{"match": {"level": "ERROR"}}, {"match": {"service": "payment-service"}}]}}},
    {"query": {"range": {"@timestamp": {"gte": "now-1h"}}}},
    {"query": {"match": {"message": "cache miss for key"}}},
    {"query": {"match": {"message": "ssl certificate expired"}}}
]

def setup_es_client():
    """Initialize Elasticsearch client with benchmark settings"""
    return Elasticsearch(
        ES_HOST,
        api_key=ES_API_KEY,
        request_timeout=10,
        retry_on_timeout=False  # We measure latency including retries
    )

def setup_os_client():
    """Initialize OpenSearch client with benchmark settings"""
    return OpenSearch(
        hosts=[OS_HOST],
        http_auth=(OS_USER, OS_PASSWORD),
        use_ssl=True,
        verify_certs=False,
        request_timeout=10,
        retry_on_timeout=False
    )

def run_search_benchmark(client, client_name, query_list, query_count):
    """Run full-text search benchmark and return latency metrics"""
    latencies = []
    errors = 0
    start_time = time.time()

    for i in range(query_count):
        query = random.choice(query_list)
        try:
            query_start = time.perf_counter()
            response = client.search(
                index=INDEX_NAME,
                body=query,
                size=10  # Typical log search result size
            )
            query_end = time.perf_counter()
            latency_ms = (query_end - query_start) * 1000
            latencies.append(latency_ms)
            if (i + 1) % 1000 == 0:
                logging.info(f"{client_name}: Completed {i+1}/{query_count} queries")
        except Exception as e:
            errors += 1
            logging.warning(f"{client_name} query failed: {str(e)}")

    total_time = time.time() - start_time
    throughput = query_count / total_time

    # Calculate percentile latencies
    latencies_sorted = sorted(latencies)
    p50 = median(latencies_sorted)
    p95 = latencies_sorted[int(len(latencies_sorted) * 0.95)]
    p99 = latencies_sorted[int(len(latencies_sorted) * 0.99)]
    avg = mean(latencies_sorted)
    stddev = pstdev(latencies_sorted)

    return {
        "client": client_name,
        "total_queries": query_count,
        "successful_queries": len(latencies),
        "errors": errors,
        "avg_latency_ms": round(avg, 2),
        "p50_latency_ms": round(p50, 2),
        "p95_latency_ms": round(p95, 2),
        "p99_latency_ms": round(p99, 2),
        "stddev_ms": round(stddev, 2),
        "throughput_qps": round(throughput, 2)
    }

def main():
    """Run benchmark for both clusters and write results to CSV"""
    logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
    logger = logging.getLogger(__name__)

    logger.info("Starting 10TB log search benchmark for Elasticsearch 9 vs OpenSearch 3")
    logger.info(f"Configuration: {QUERY_COUNT} queries, {CONCURRENT_WORKERS} workers, {len(SEARCH_QUERIES)} query templates")

    # Initialize clients
    es_client = setup_es_client()
    os_client = setup_os_client()

    # Verify indices exist
    try:
        es_count = es_client.count(index=INDEX_NAME)["count"]
        os_count = os_client.count(index=INDEX_NAME)["count"]
        logger.info(f"Document counts - ES9: {es_count}, OS3: {os_count}")
        if abs(es_count - os_count) > 1000:
            logger.error("Document count mismatch, aborting benchmark")
            return
    except Exception as e:
        logger.error(f"Failed to verify document counts: {str(e)}")
        return

    # Run benchmarks
    es_results = run_search_benchmark(es_client, "Elasticsearch 9.0.1", SEARCH_QUERIES, QUERY_COUNT)
    os_results = run_search_benchmark(os_client, "OpenSearch 3.0.0", SEARCH_QUERIES, QUERY_COUNT)

    # Write results to CSV
    with open(OUTPUT_CSV, "w", newline="") as csvfile:
        fieldnames = es_results.keys()
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerow(es_results)
        writer.writerow(os_results)

    logger.info(f"Benchmark complete. Results written to {OUTPUT_CSV}")
    logger.info(f"Elasticsearch 9 p99 latency: {es_results['p99_latency_ms']}ms")
    logger.info(f"OpenSearch 3 p99 latency: {os_results['p99_latency_ms']}ms")
    logger.info(f"Latency difference: {es_results['p99_latency_ms'] - os_results['p99_latency_ms']}ms")

if __name__ == "__main__":
    main()

PUT /logs-10tb-2024.05/_settings
{
  "index.refresh_interval": "30s"
}

PUT /logs-10tb-2024.05/_settings
{
  "index.codec": "best_compression"
}

telemetry:
  metrics:
    prometheus:
      enabled: true
      port: 9201
      host: "0.0.0.0"
      prefix: "opensearch"