Hot Take: Elasticsearch 8.15 Is Too Heavy for Log Aggregation – Use Grafana Loki 3.0 and Promtail 2.9 Instead

import time
import json
import random
import string
import logging
from elasticsearch import Elasticsearch, helpers
from loki_client import LokiClient  # pip install loki-client
import argparse
import sys

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

def generate_log_line(size_kb: int = 1) -> str:
    """Generate a realistic application log line of ~size_kb KB."""
    base_log = {
        "timestamp": time.time_ns(),
        "level": random.choice(["INFO", "WARN", "ERROR", "DEBUG"]),
        "service": random.choice(["api-gateway", "auth-service", "payment-svc", "inventory-svc"]),
        "trace_id": ''.join(random.choices(string.hexdigits.lower(), k=32)),
        "message": "User action completed successfully" if random.random() > 0.1 else "Failed to process request"
    }
    # Pad to reach target size
    padding = " " * (size_kb * 1024 - len(json.dumps(base_log)))
    base_log["padding"] = padding[:max(0, len(padding))]
    return json.dumps(base_log)

def benchmark_elasticsearch(es_client: Elasticsearch, index_name: str, num_logs: int, batch_size: int = 500):
    """Benchmark Elasticsearch 8.15 log ingest performance."""
    logger.info(f"Starting Elasticsearch benchmark: {num_logs} logs, batch size {batch_size}")
    start_time = time.perf_counter()
    success_count = 0
    fail_count = 0

    # Prepare bulk ingest payload
    actions = []
    for i in range(num_logs):
        log_line = generate_log_line()
        action = {
            "_index": index_name,
            "_source": json.loads(log_line)
        }
        actions.append(action)
        if len(actions) >= batch_size:
            try:
                success, failed = helpers.bulk(es_client, actions, stats_only=True)
                success_count += success
                fail_count += failed
                actions = []
            except Exception as e:
                logger.error(f"Elasticsearch bulk ingest failed: {e}")
                fail_count += len(actions)
                actions = []

    # Ingest remaining logs
    if actions:
        try:
            success, failed = helpers.bulk(es_client, actions, stats_only=True)
            success_count += success
            fail_count += failed
        except Exception as e:
            logger.error(f"Elasticsearch final ingest failed: {e}")
            fail_count += len(actions)

    elapsed = time.perf_counter() - start_time
    logger.info(f"Elasticsearch benchmark complete: {success_count} success, {fail_count} failed, {elapsed:.2f}s")
    return success_count / elapsed if elapsed > 0 else 0

def benchmark_loki(loki_client: LokiClient, num_logs: int, batch_size: int = 500):
    """Benchmark Grafana Loki 3.0 log ingest performance."""
    logger.info(f"Starting Loki benchmark: {num_logs} logs, batch size {batch_size}")
    start_time = time.perf_counter()
    success_count = 0
    fail_count = 0

    batch = []
    for i in range(num_logs):
        log_line = generate_log_line()
        log_entry = {
            "stream": {
                "service": json.loads(log_line)["service"],
                "level": json.loads(log_line)["level"]
            },
            "values": [
                [str(time.time_ns()), log_line]
            ]
        }
        batch.append(log_entry)
        if len(batch) >= batch_size:
            try:
                loki_client.push(batch)
                success_count += len(batch)
                batch = []
            except Exception as e:
                logger.error(f"Loki push failed: {e}")
                fail_count += len(batch)
                batch = []

    # Push remaining logs
    if batch:
        try:
            loki_client.push(batch)
            success_count += len(batch)
        except Exception as e:
            logger.error(f"Loki final push failed: {e}")
            fail_count += len(batch)

    elapsed = time.perf_counter() - start_time
    logger.info(f"Loki benchmark complete: {success_count} success, {fail_count} failed, {elapsed:.2f}s")
    return success_count / elapsed if elapsed > 0 else 0

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Benchmark log ingest performance: Elasticsearch 8.15 vs Loki 3.0")
    parser.add_argument("--es-host", default="http://localhost:9200", help="Elasticsearch host URL")
    parser.add_argument("--loki-host", default="http://localhost:3100", help="Loki host URL")
    parser.add_argument("--num-logs", type=int, default=100000, help="Number of logs to ingest per benchmark")
    parser.add_argument("--index-name", default="benchmark-logs", help="Elasticsearch index name")
    args = parser.parse_args()

    # Initialize clients with error handling
    try:
        es = Elasticsearch(args.es_host, verify_certs=False, request_timeout=30)
        if not es.ping():
            logger.error("Failed to connect to Elasticsearch")
            sys.exit(1)
    except Exception as e:
        logger.error(f"Elasticsearch client init failed: {e}")
        sys.exit(1)

    try:
        loki = LokiClient(base_url=args.loki_host, timeout=30)
    except Exception as e:
        logger.error(f"Loki client init failed: {e}")
        sys.exit(1)

    # Run benchmarks
    es_rate = benchmark_elasticsearch(es, args.index_name, args.num_logs)
    loki_rate = benchmark_loki(loki, args.num_logs)

    # Print results
    print("\n=== Benchmark Results ===")
    print(f"Elasticsearch 8.15 Ingest Rate: {es_rate:.2f} logs/sec")
    print(f"Loki 3.0 Ingest Rate: {loki_rate:.2f} logs/sec")
    print(f"Loki is {loki_rate/es_rate:.2f}x faster than Elasticsearch")

# Promtail 2.9 Configuration for High-Volume Log Aggregation
# Validated against Promtail v2.9.2 on Linux x86_64
# Run validation: promtail --config.file=promtail-config.yaml --dry-run

server:
  http_listen_port: 9080
  grpc_listen_port: 9095

positions:
  filename: /var/lib/promtail/positions.yaml  # Persist read positions to avoid re-ingesting logs
  sync_period: 10s  # Sync positions to disk every 10 seconds

clients:
  - url: http://loki:3100/loki/api/v1/push  # Loki 3.0 push endpoint
    batchwait: 5s  # Wait up to 5 seconds to batch logs
    batchsize: 102400  # Batch size in bytes (100KB)
    timeout: 30s  # Push timeout per batch
    backoff_config:
      min_period: 500ms
      max_period: 5m
      max_retries: 10  # Retry failed pushes up to 10 times before dropping logs

scrape_configs:
  # Scrape Docker container logs
  - job_name: docker-containers
    docker_sd_configs:
      - host: unix:///var/run/docker.sock  # Connect to local Docker daemon
        refresh_interval: 30s  # Refresh container list every 30 seconds
    relabel_configs:
      - source_labels: [__meta_docker_container_name]
        target_label: container_name
      - source_labels: [__meta_docker_container_id]
        target_label: container_id
      - source_labels: [__meta_docker_container_log_path]
        target_label: log_path
    pipeline_stages:
      - json:  # Parse JSON log lines
          expressions:
            level: level
            service: service
            trace_id: trace_id
      - labels:  # Extract labels from log fields
          level:
          service:
          trace_id:
      - timestamp:  # Parse custom timestamp field
          source: timestamp
          format: unix_ns  # Elasticsearch-style nanosecond timestamps
      - output:
          source: message  # Only send the message field to Loki, drop other fields to save space

  # Scrape systemd journal logs
  - job_name: systemd-journal
    journal:
      path: /var/log/journal  # Read from persistent journal directory
      max_age: 24h  # Only read logs from last 24 hours on startup
      labels:
        job: systemd-journal
    relabel_configs:
      - source_labels: [__journal__systemd_unit]
        target_label: systemd_unit
      - source_labels: [__journal__hostname]
        target_label: hostname
    pipeline_stages:
      - regex:  # Parse systemd log format
          expression: '^(?P\w+ \d+ \d+:\d+:\d+) (?P\S+) (?P\S+): (?P.*)$'
      - labels:
          service:
          hostname:
      - timestamp:
          source: timestamp
          format: "Jan 2 15:04:05"  # Systemd timestamp format

  # Scrape application log files
  - job_name: app-logs
    static_configs:
      - targets:
          - localhost
        labels:
          job: app-logs
          __path__: /var/log/app/*.log  # Match all .log files in /var/log/app
    pipeline_stages:
      - multiline:  # Handle multi-line stack traces
          firstline: '^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}'
          max_wait_time: 3s  # Wait up to 3s for next line before flushing
      - regex:
          expression: '^(?P\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d+Z) (?P\w+) (?P\S+) (?P.*)$'
      - labels:
          level:
          service:
      - timestamp:
          source: timestamp
          format: RFC3339

import time
import json
import psutil  # pip install psutil
import argparse
from elasticsearch import Elasticsearch
from loki_client import LokiClient
import logging
import sys

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def get_process_memory():
    """Return current process memory usage in MB."""
    process = psutil.Process()
    return process.memory_info().rss / 1024 / 1024

def query_elasticsearch(es: Elasticsearch, index: str, query: dict, start_time: str, end_time: str):
    """Execute query against Elasticsearch 8.15 and return results + metrics."""
    logger.info(f"Querying Elasticsearch index {index} for time range {start_time} to {end_time}")
    start_mem = get_process_memory()
    start = time.perf_counter()
    try:
        response = es.search(
            index=index,
            body={
                "query": {
                    "bool": {
                        "must": [
                            {"range": {"@timestamp": {"gte": start_time, "lte": end_time}}},
                            query
                        ]
                    }
                },
                "size": 0  # We only care about aggregation results, not raw logs
            },
            request_timeout=60
        )
        elapsed = time.perf_counter() - start
        end_mem = get_process_memory()
        mem_used = end_mem - start_mem
        hit_count = response["hits"]["total"]["value"]
        logger.info(f"Elasticsearch query returned {hit_count} hits in {elapsed:.2f}s, used {mem_used:.2f}MB memory")
        return {
            "hit_count": hit_count,
            "latency_sec": elapsed,
            "memory_mb": mem_used
        }
    except Exception as e:
        logger.error(f"Elasticsearch query failed: {e}")
        return None

def query_loki(loki: LokiClient, query: str, start_time: str, end_time: str):
    """Execute query against Loki 3.0 and return results + metrics."""
    logger.info(f"Querying Loki for time range {start_time} to {end_time}")
    start_mem = get_process_memory()
    start = time.perf_counter()
    try:
        # Convert time to Unix nanoseconds for Loki
        start_ns = int(time.mktime(time.strptime(start_time, "%Y-%m-%dT%H:%M:%S")) * 1e9)
        end_ns = int(time.mktime(time.strptime(end_time, "%Y-%m-%dT%H:%M:%S")) * 1e9)
        response = loki.query_range(
            query=query,
            start=start_ns,
            end=end_ns,
            limit=1000
        )
        elapsed = time.perf_counter() - start
        end_mem = get_process_memory()
        mem_used = end_mem - start_mem
        # Loki returns streams, count total entries
        hit_count = sum(len(stream["values"]) for stream in response.get("data", {}).get("result", []))
        logger.info(f"Loki query returned {hit_count} hits in {elapsed:.2f}s, used {mem_used:.2f}MB memory")
        return {
            "hit_count": hit_count,
            "latency_sec": elapsed,
            "memory_mb": mem_used
        }
    except Exception as e:
        logger.error(f"Loki query failed: {e}")
        return None

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Compare query performance: Elasticsearch 8.15 vs Loki 3.0")
    parser.add_argument("--es-host", default="http://localhost:9200", help="Elasticsearch host")
    parser.add_argument("--loki-host", default="http://localhost:3100", help="Loki host")
    parser.add_argument("--index", default="benchmark-logs", help="Elasticsearch index name")
    parser.add_argument("--start-time", default="2024-05-01T00:00:00", help="Query start time (ISO 8601)")
    parser.add_argument("--end-time", default="2024-05-01T01:00:00", help="Query end time (ISO 8601)")
    parser.add_argument("--query-level", default="ERROR", help="Log level to query")
    args = parser.parse_args()

    # Initialize clients
    try:
        es = Elasticsearch(args.es_host, verify_certs=False, request_timeout=30)
        if not es.ping():
            logger.error("Elasticsearch connection failed")
            sys.exit(1)
    except Exception as e:
        logger.error(f"Elasticsearch init failed: {e}")
        sys.exit(1)

    try:
        loki = LokiClient(base_url=args.loki_host, timeout=30)
    except Exception as e:
        logger.error(f"Loki init failed: {e}")
        sys.exit(1)

    # Define query (find all ERROR logs from payment-svc)
    es_query = {
        "match": {
            "level": args.query_level
        }
    }
    loki_query = f'{{level="{args.query_level}"}}'  # Loki LogQL query

    # Execute queries
    es_result = query_elasticsearch(es, args.index, es_query, args.start_time, args.end_time)
    loki_result = query_loki(loki, loki_query, args.start_time, args.end_time)

    # Print comparison
    if es_result and loki_result:
        print("\n=== Query Performance Comparison ===")
        print(f"Elasticsearch 8.15: {es_result['latency_sec']:.2f}s, {es_result['memory_mb']:.2f}MB, {es_result['hit_count']} hits")
        print(f"Loki 3.0: {loki_result['latency_sec']:.2f}s, {loki_result['memory_mb']:.2f}MB, {loki_result['hit_count']} hits")
        print(f"Loki is {es_result['latency_sec']/loki_result['latency_sec']:.2f}x faster, uses {es_result['memory_mb']/loki_result['memory_mb']:.2f}x less memory")

storage_config:
  aws:
    s3: s3://us-east-1/loki-logs-prod
    bucketnames: loki-logs-prod
    region: us-east-1
    access_key_id: ${S3_ACCESS_KEY}
    secret_access_key: ${S3_SECRET_KEY}
  boltdb_shipper:
    active_index_directory: /loki/index
    cache_location: /loki/index-cache

pipeline_stages:
  - json:
      expressions:
        user_id: user_id  # Extract but do not label
        service: service
        level: level
  - labels:
      service:
      level:
  - drop:
      source: user_id  # Drop from labels if accidentally added
  - output:
      source: message  # Only send message to Loki

./es2loki \
  --es-host http://es-prod:9200 \
  --es-index-pattern "logs-*" \
  --loki-host http://loki-prod:3100 \
  --start-time 2024-04-01T00:00:00 \
  --end-time 2024-04-07T23:59:59 \
  --batch-size 1000