1. Introduction — Why Log Aggregation Matters

Modern distributed systems generate an extraordinary volume of log data. A mid-size SaaS company running hundreds of microservices across Kubernetes clusters can easily produce 10 terabytes or more of log data every single day. Every HTTP request, database query, cache miss, background job execution, and infrastructure event leaves behind a trail of structured or semi-structured log entries that collectively tell the story of what happened inside the system and when.

The problem is not the existence of logs — nearly every application writes them. The problem is that logs are scattered across thousands of containers, ephemeral pods, virtual machines, load balancers, and serverless functions. When a production incident strikes at 3 AM and an on-call engineer needs to understand why latency spiked for a particular customer cohort, manually SSH-ing into dozens of machines to grep through local log files is not just inefficient — it is unacceptable. The time-to-insight must be measured in seconds, not hours.

This is precisely the problem that log aggregation platforms solve. Systems like the Elastic Stack (Elasticsearch, Logstash, Kibana — collectively known as ELK), Splunk, Datadog, Grafana Loki, and Sumo Logic provide centralized collection, indexing, search, and visualization of log data from heterogeneous sources. They serve as the backbone of modern observability strategies, complementing metrics and distributed tracing to give engineering teams a complete picture of system behavior.

In this article, we will design a log aggregation platform from first principles. We will start with requirements and capacity planning, then work through the entire system — from collection agents running on edge nodes, through the ingestion pipeline with its parsing and enrichment stages, into the storage and indexing layer, and finally up to the search, alerting, and visualization tiers. We will address real-world concerns like multi-tenancy, compliance, anomaly detection, and multi-region deployment. Along the way, we will include architectural diagrams, database schemas, API contracts, and a substantial C# implementation that demonstrates the core components in production-quality code.

Whether you are preparing for a system design interview at a top-tier tech company, architecting a logging platform for your organization, or simply seeking to deepen your understanding of distributed systems, this guide will give you the knowledge and mental models to reason about log aggregation at any scale. We will reference the design decisions made by Elasticsearch, Splunk, and other industry leaders, analyzing why they made specific trade-offs and how those choices affect performance, cost, and operational complexity.

2. Functional & Non-Functional Requirements

Functional Requirements

1. Log Ingestion: Accept log data from containers, Kubernetes pods, VMs, load balancers, databases, and cloud services via TCP, UDP, HTTP, and agent-based push models.
2. Log Parsing & Normalization: Parse Apache access logs, JSON, syslog, and free-text into a normalized schema with consistent fields (timestamp, source, severity, message).
3. Full-Text Search: Search billions of log entries using keywords, field filters, boolean operators, wildcards, and regex — returning results within seconds.
4. Time-Range Queries: Efficiently narrow results to specified time windows with intuitive time-range selection in the UI.
5. Real-Time Alerting: Define alerts based on threshold breaches, absence of events, anomalies, or complex conditions with notifications via email, Slack, PagerDuty, and webhooks.
6. Dashboards & Visualization: Pre-built and custom dashboards with auto-refresh, drill-down from overviews to individual log entries.
7. Log Retention & Archival: Configurable retention per source, per tenant, per compliance requirement with automatic archival to cheaper storage.
8. Multi-Tenancy: Complete data isolation between organizations sharing the same infrastructure with per-tenant resource tracking.

Non-Functional Requirements

3. Capacity Estimation & Back-of-Envelope Math

Ingestion Volume

Assume 500 microservices each generating 500 log lines/second. Sustained: 250,000 lines/sec. Peak: 500,000 lines/sec. Each line averages 500 bytes. Sustained throughput: 250K x 500 bytes = 125 MB/s = 10.8 TB/day. With 7:1 compression: ~1.5 TB/day on disk.

Storage Calculation

90-day retention at 1.5 TB/day compressed = 135 TB compressed. Adding 25% index overhead yields ~170 TB. For a 30-day hot tier (45 TB), 30-day warm (45 TB), and 30-day cold (45 TB), total footprint is roughly 200 TB including all structures.

Cluster Sizing

50 data nodes (64 GB RAM, 2 TB NVMe each), 3-5 master nodes (16 GB RAM), 10 coordinating nodes (32 GB RAM). Total: ~65 nodes.

4. Data Model — Logs, Sources, Fields, Indices

Core Entities

Log Entry: Atomic unit — unique ID, timestamp, source reference, severity, raw message, parsed fields. Log Source: Origin of data (service, host, namespace) with parsing config, retention policy, and ACL. Log Index: Logical grouping by time range for efficient partitioning. Alert Rule: User-defined condition with query, evaluation frequency, and notification targets. Dashboard: Collection of saved visualizations backed by queries with drill-down and auto-refresh.

Entity Relationships

Sample Log Document

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "timestamp": "2026-07-01T10:15:32.456Z",
  "received_at": "2026-07-01T10:15:32.789Z",
  "source_id": "svc-payment-api",
  "severity": "ERROR",
  "service": "payment-api",
  "host": "k8s-node-42",
  "namespace": "production",
  "message": "Payment processing failed for order_9876: timeout after 30s",
  "request_id": "req-abc-123",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "http_method": "POST",
  "http_status_code": 504,
  "duration_ms": 30000,
  "tags": ["payment", "timeout", "critical"]
}

5. API Design

Ingest Logs

POST /api/v1/ingest
Content-Type: application/json
Authorization: Bearer {api_key}

{
  "source": "payment-api",
  "logs": [
    {
      "timestamp": "2026-07-01T10:15:32.456Z",
      "level": "ERROR",
      "message": "Payment processing failed",
      "fields": {"order_id": "order_9876", "duration_ms": 30000}
    }
  ]
}
Response: 202 Accepted
{ "batch_id": "batch_abc123", "accepted_count": 1 }

Search Logs

POST /api/v1/search
{
  "query": "service:payment-api AND severity:ERROR",
  "time_range": {"from": "2026-07-01T00:00:00Z", "to": "2026-07-01T23:59:59Z"},
  "fields": ["timestamp", "message", "severity", "trace_id"],
  "sort": [{"timestamp": "desc"}],
  "from": 0, "size": 50
}

Response: 200 OK
{ "total": 1247, "took_ms": 342, "hits": [ ... ] }

Create Alert

POST /api/v1/alerts
{
  "name": "Payment API Error Spike",
  "query": "service:payment-api AND severity:ERROR",
  "condition": {"type": "threshold", "metric": "count",
    "window_minutes": 5, "operator": "gt", "value": 500},
  "evaluation_interval_minutes": 1,
  "notifications": [
    {"type": "slack", "channel": "#incidents"},
    {"type": "pagerduty", "service_key": "xxx"}
  ]
}

6. High-Level Architecture

The overall architecture follows a pipeline pattern: collection, transport, ingestion, storage, and serving. Each stage is independently scalable and failure-isolated.

Component Responsibilities Deep-Dive

Collection Layer: Lightweight agents deployed alongside application workloads that tail log files, capture stdout/stderr from containers, and forward structured events to the transport layer. The agents handle local buffering, retry logic, and back-pressure management to prevent data loss during network disruptions. Each agent maintains a persistent registry file that tracks the byte offset of the last successfully forwarded log entry, ensuring exactly-once delivery semantics from the agent to the transport layer even across agent restarts and node reboots.

Transport Layer: Apache Kafka serves as the durable buffer between collection and ingestion. It decouples producers from consumers, absorbs traffic spikes that occur during deployment storms or incident-driven debugging sessions, and provides configurable retention guarantees. Log entries are partitioned by source identifier to ensure ordering within a single source while enabling parallel consumption across thousands of sources. Kafka's replication factor is typically set to 3, meaning each log entry exists on three separate brokers across different availability zones, providing durability even if an entire data center goes offline.

Ingestion Layer: Stateless worker processes that consume from Kafka topic partitions, parse raw log lines into structured documents, enrich them with contextual metadata (geo-IP lookups, service ownership, environment classifications), normalize field names and data formats to a consistent schema, and write bulk batches to Elasticsearch using the _bulk API for maximum throughput. This layer scales horizontally by adding more consumer instances up to the number of Kafka partitions. Workers are designed to be completely stateless, meaning any worker can process any partition, enabling rapid scaling and zero-downtime deployments.

Storage Layer: The Elasticsearch cluster stores all indexed log data in a tiered architecture optimized for the temporal access pattern inherent in log data. Hot nodes with fast NVMe SSDs hold the most recent and frequently queried data. Warm nodes with SATA SSDs hold older data that is read-only but still regularly queried for trend analysis and dashboards. Cold nodes using object storage snapshots provide cost-effective archival with searchable capability. Index Lifecycle Management policies automate data migration between tiers as data ages.

Serving Layer: Coordinating nodes receive incoming search requests from the API and UI layers, parse the query into an execution plan, determine which indices and shards are relevant based on the time range and field filters, fan out parallel requests to the appropriate data nodes, collect partial results from each shard, merge and re-score the combined results, and return the final sorted, paginated response to the caller. The serving layer also hosts the real-time alert evaluation engine, which continuously runs saved queries against the latest indexed data and triggers notifications when conditions are met.

Management Layer: Master nodes manage the global cluster state — index creation and deletion, shard allocation across nodes, node health monitoring, and failover orchestration. A metadata database (PostgreSQL) stores non-log application data: user accounts, tenant configurations, alert rule definitions, dashboard metadata, and audit logs. The management layer ensures that the cluster remains healthy and that all administrative operations are performed atomically and consistently.

7. Log Collection Agents

The collection layer is the first point of contact between log sources and the platform. The agent must be lightweight, reliable, and resource-efficient — handling file rotation, container destruction, disk full conditions, and intermittent network issues.

Deployment Patterns

Agent Configuration

filebeat.autodiscover:
  providers:
    - type: kubernetes
      hints.enabled: true
      hints.default_config:
        type: container
        paths:
          - /var/log/containers/${data.kubernetes.pod.name}.log
        processors:
          - add_kubernetes_metadata:
              host: ${NODE_NAME}

output.kafka:
  hosts: ["kafka-0:9092", "kafka-1:9092", "kafka-2:9092"]
  topic: "logs-${data.kubernetes.labels.app}"
  compression: lz4

processors:
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - decode_json_fields:
      fields: ["message"]
      target: ""

8. Log Ingestion Pipeline

The ingestion pipeline transforms raw log lines into clean, indexed documents. It must handle millions of events per second while maintaining ordering within sources and recovering gracefully from failures at every stage.

Stage 1 - Parsing: Extract structured fields from raw lines using JSON deserialization or grok pattern matching. Stage 2 - Enrichment: Add geo-IP, service metadata, and security classifications. Stage 3 - Normalization: Map varying field names to a consistent schema. Stage 4 - Deduplication: Detect and discard duplicates within a sliding window. Stage 5 - Bulk Writing: Accumulate into batches of 5,000-10,000 documents for Elasticsearch.

Bulk Writer Implementation

public class BulkWriter
{
    private readonly ConcurrentQueue<LogEntry> _buffer;
    private readonly Timer _flushTimer;
    private readonly int _batchSize = 5000;

    public BulkWriter(IElasticClient client)
    {
        _buffer = new ConcurrentQueue<LogEntry>();
        _flushTimer = new Timer(
            _ => FlushAsync().GetAwaiter().GetResult(),
            null, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(5));
    }

    public void Add(LogEntry entry) => _buffer.Enqueue(entry);

    public async Task FlushAsync()
    {
        var batch = new List<LogEntry>();
        while (batch.Count < _batchSize && _buffer.TryDequeue(out var e))
            batch.Add(e);
        if (!batch.Any()) return;

        var ops = batch.Select(e =>
            new BulkIndexOperation<LogEntry>(e)
            { Index = $"logs-{e.Source}-{e.Timestamp:yyyy.MM.dd}" }
        ).ToList();

        var req = new BulkRequest { Operations = ops };
        var resp = await _client.BulkAsync(req);
        if (resp.Errors)
            foreach (var item in resp.ItemsWithErrors)
                await SendToDeadLetterQueueAsync(item.Error);
    }
}

9. Log Parsing & Structured Data Extraction

Log parsing is deceptively complex. The challenge is handling the enormous variety of formats in the real world: Apache/Nginx access logs, application framework logs, syslog, Windows Event Logs, database slow query logs, and cloud audit logs.

Grok Patterns

# Apache Access Log
%{IPORHOST:client_ip} - %{DATA:user_name} \[%{HTTPDATE:access_time}\]
"%{WORD:http_method} %{NOTSPACE:request_uri} HTTP/%{NUMBER:http_version}"
%{NUMBER:response_code} %{NUMBER:body_bytes_sent}
"%{DATA:referrer}" "%{DATA:user_agent}"
%{NUMBER:request_duration:float}

# Application Log
%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:severity}
\[%{DATA:thread_name}\] %{DATA:logger_name} - %{GREEDYDATA:message}

Multi-Line Handling

Stack traces and error dumps span multiple lines. The agent must detect multi-line entries and reassemble them into single logical events before parsing. This is done by identifying log entry starts (usually by timestamp) and buffering subsequent lines until the next entry begins.

Field Extraction Strategies

10. Index Design — Inverted Indices & Time-Based Sharding

Effective index design is the most critical factor in achieving both fast ingestion and fast search. Elasticsearch uses inverted indices that map each unique term to the list of documents containing it, enabling O(1) lookups.

Inverted Index Structure

When searching "error AND timeout", Elasticsearch intersects posting lists to find Doc 5 — dramatically faster than scanning every document.

Index Lifecycle Management

Index Template

PUT _index_template/logs-template
{
  "index_patterns": ["logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "30s",
      "codec": "best_compression",
      "index.lifecycle.name": "logs-policy"
    },
    "mappings": {
      "properties": {
        "timestamp": {"type": "date"},
        "severity": {"type": "keyword"},
        "service": {"type": "keyword"},
        "message": {"type": "text", "analyzer": "standard"},
        "trace_id": {"type": "keyword"},
        "duration_ms": {"type": "long"}
      }
    }
  }
}

11. Storage Tiering — Hot, Warm, Cold, and Frozen

Not all log data has equal access frequency. Recent entries are orders of magnitude more likely to be searched. Storage tiering exploits this to optimize both performance and cost.

ILM Policy Configuration

{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {"max_primary_shard_size": "50gb", "max_age": "1d"},
          "set_priority": {"priority": 100}
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": {"number_of_shards": 1},
          "forcemerge": {"max_num_segments": 1},
          "allocate": {"require": {"data": "warm"}}
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "searchable_snapshot": {"snapshot_repository": "logs-archive"}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {"delete": {}}
      }
    }
  }
}

12. Search Engine — Full-Text Search at Scale

The search engine makes a log aggregation platform genuinely useful. It must handle keyword queries, field filters, boolean logic, wildcards, regex, phrase matching, fuzzy matching, and complex aggregations — all returning results in seconds across petabytes.

Query Execution Flow

Query DSL Example

{
  "query": {
    "bool": {
      "must": [
        {"match": {"message": "connection timeout"}},
        {"term": {"severity": "ERROR"}}
      ],
      "filter": [
        {"range": {"timestamp": {"gte": "2026-07-01T00:00:00Z",
          "lte": "2026-07-01T23:59:59Z"}}},
        {"term": {"service": "payment-api"}}
      ]
    }
  },
  "sort": [{"timestamp": {"order": "desc"}}],
  "aggs": {
    "errors_over_time": {
      "date_histogram": {"field": "timestamp", "fixed_interval": "5m"}
    },
    "top_services": {"terms": {"field": "service", "size": 10}}
  }
}

Optimization Techniques

• Time-range pre-filtering: Skip indices outside query range entirely, reducing shard count by 97% for 24-hour queries against 30-day retention
• Request caching: Cache identical query results for 5 minutes by default, with automatic invalidation when new data is indexed in the queried time range
• Preference routing: Route repeated searches for the same service or tenant to the same shard replica, maximizing filesystem cache hit rates and reducing disk I/O
• Doc-value fields: Columnar storage optimized for sequential scans during aggregations, dramatically faster than the inverted index for numerical computations
• Runtime fields: Compute fields at search time rather than indexing, trading search performance for reduced index storage and maximum flexibility when query patterns are unpredictable
• Scroll and search_after: For large result sets that exceed a single page, use search_after with a composite sort key instead of deep pagination via offset, avoiding the quadratic cost of skip-and-take operations
• Index sorting: Configure indices to be sorted by timestamp at index time, enabling Elasticsearch to skip entire segments during time-range queries that only cover recent data
• Minimum_should_match: For OR queries across many terms, set minimum_should_match to avoid scoring every possible combination, reducing CPU usage for broad searches

Search Relevance and Scoring

Elasticsearch uses the BM25 algorithm (Okapi BM25) by default for scoring text relevance. BM25 considers term frequency (how often the search term appears in a document), inverse document frequency (how rare the term is across all documents), and field length (shorter documents score higher for the same term density). For log search, relevance scoring is less critical than for web search — most queries are filtered by time and severity, and the user typically wants the most recent matching entries rather than the most semantically relevant ones. However, when searching for a specific error message across millions of similar entries, BM25 scoring helps surface the most relevant matches at the top of the results.

Custom scoring can be implemented using function_score queries that boost results based on recency (newer entries scored higher), severity (ERROR entries scored above INFO), or specific field values (entries from the user's own services boosted higher). This is particularly useful for dashboards where the user wants to see the most impactful entries first, not just the most recently indexed ones.

Aggregation Pipeline

Aggregations are the foundation of dashboards and analytics. They compute statistics over search results without returning individual documents. Bucket aggregations group data by field values (terms aggregation for top-N services, date_histogram for time-series trends, histogram for numeric distributions). Metric aggregations compute statistics (avg, percentiles, percentiles_ranks, cardinality for unique counts). Pipeline aggregations chain results from other aggregations (moving averages over time-series buckets, cumulative sums). Understanding aggregation execution is critical for performance — a terms aggregation on a high-cardinality field (like user_id with millions of unique values) can consume significant memory and should be limited with a size parameter.

13. Real-Time Alerting Engine

Alerting transforms a reactive investigation tool into a proactive monitoring system. The engine continuously evaluates conditions and triggers notifications when problems are detected.

Alert Types

Alert Pipeline

Alert State Machine

Alert states: OK (condition not met), FIRING (condition met, notification sent), ACKNOWLEDGED (human acknowledged), RESOLVED (condition cleared, recovery notification sent). Flapping prevention uses hysteresis — the condition must be consistently met for N evaluations before firing.

Alert Evaluator

public class AlertEvaluator
{
    private readonly ISearchEngine _search;
    private readonly INotificationService _notifier;
    private readonly IAlertStateStore _stateStore;

    public async Task EvaluateAsync(AlertRule rule)
    {
        var query = BuildQuery(rule);
        var result = await _search.ExecuteAsync(query);
        var currentValue = ExtractMetric(result, rule.Condition.Metric);
        var thresholdMet = EvaluateCondition(
            currentValue, rule.Condition.Operator, rule.Condition.Value);

        var state = await _stateStore.GetStateAsync(rule.Id);

        if (thresholdMet && (state?.Status != AlertStatus.FIRING))
        {
            var firing = new AlertState
            {
                RuleId = rule.Id, Status = AlertStatus.FIRING,
                FiredAt = DateTime.UtcNow, CurrentValue = currentValue
            };
            await _stateStore.SaveStateAsync(firing);
            await NotifyAsync(rule, firing);
        }
        else if (!thresholdMet && state?.Status == AlertStatus.FIRING)
        {
            state.Status = AlertStatus.OK;
            state.ResolvedAt = DateTime.UtcNow;
            await _stateStore.SaveStateAsync(state);
            await NotifyResolutionAsync(rule, state);
        }
    }

    private bool EvaluateCondition(double value, string op, double threshold) => op switch
    {
        "gt" => value > threshold,
        "gte" => value >= threshold,
        "lt" => value < threshold,
        "lte" => value <= threshold,
        _ => false
    };
}

14. Dashboard & Visualization Layer

Dashboards are the primary interface for most users. A well-designed dashboard provides at-a-glance system health understanding.

Drill-down navigation allows clicking a bar in a chart to filter the entire dashboard, and clicking a time-series point opens a search for that time range. Cross-linking between dashboards preserves filter context.

15. Distributed Tracing Integration

Logs tell you what happened. Traces tell you how a request flowed. The real power emerges when you can jump between them.

OpenTelemetry provides the standard mechanism for propagating trace context. Each service includes trace_id and span_id in log entries through automatic injection or explicit inclusion. The log platform indexes these fields, enabling users to search for all logs associated with a specific trace and jump directly to trace visualization.

16. Deriving Metrics from Logs

Logs contain a wealth of information that can be aggregated into useful metrics, especially when applications do not export native metrics.

• Counter metrics: Count of matching logs per interval (e.g., ERROR count/min/service)
• Rate metrics: 5xx error rate as percentage of total responses
• Histogram metrics: Distribution of duration_ms values
• Cardinality metrics: Unique user_ids experiencing errors

Log-based SLIs (Service Level Indicators) capture actual user experience — counting successful HTTP responses, computing latency percentiles, and tracking error rates directly from access logs to feed SLO tracking and error budget calculations.

17. Log Retention & Compliance

PII Redaction Rules

{
  "redaction_rules": [
    {
      "field": "message",
      "pattern": "\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b",
      "replacement": "[EMAIL_REDACTED]"
    },
    {
      "field": "message",
      "pattern": "\\b\\d{4}[- ]?\\d{4}[- ]?\\d{4}[- ]?\\d{4}\\b",
      "replacement": "[CC_REDACTED]"
    },
    {
      "field": "user_id",
      "type": "hash",
      "algorithm": "sha256"
    }
  ]
}

18. Multi-Tenancy Architecture

Three isolation strategies: Index-per-tenant (strong isolation, index proliferation), Shared indices with tenant filtering (efficient, relies on correct query construction), Hybrid (small tenants shared, large tenants dedicated).

Resource quotas per tenant: ingestion rate (events/sec), storage quota, search concurrency, API rate limits. RBAC within each tenant: admin (manage all), viewer (search + dashboards), restricted (specific services only). Token bucket rate limiting at the API gateway layer.

19. Anomaly Detection with Machine Learning

Seasonal pattern detection requires 2-4 weeks of training data. The platform implements two-phase detection: build a baseline model, then evaluate new points using point anomaly detection (individual deviations) and contextual anomalies (deviations given time and recent trend).

Log template mining algorithms (Drain, LenMa) extract invariant patterns from variable log messages, enabling pattern-level analysis. For example, "Connection refused to 10.0.0.1:5432" and "Connection refused to 10.0.0.2:5432" share template "Connection refused to {ip}:{port}".

Practical Anomaly Detection Architecture

The anomaly detection subsystem runs as a separate processing pipeline that consumes from the same log stream as the indexing pipeline. It maintains a sliding window of recent log metrics (error counts, latency distributions, unique error types) and compares current values against the learned baseline. The architecture consists of three components: a feature extraction pipeline that computes numerical features from raw logs, a model training service that periodically retrains models on historical data, and a scoring service that evaluates new data points against the current model.

Feature extraction operates on a configurable window (typically 1-minute or 5-minute buckets) and produces numerical vectors representing the log stream characteristics: total event count, error rate, warning rate, unique log template count, latency percentiles, unique source count, and custom user-defined features. These vectors serve as input to both training and scoring.

The model training service runs daily (or on-demand) and trains separate models for each metric dimension. For simple metrics with clear seasonal patterns, ARIMA or Prophet models work well. For complex multi-dimensional patterns, isolation forests or autoencoders capture relationships between metrics that univariate models miss. The training service stores model artifacts in a model registry and evaluates model quality using held-out test data before deploying new models to production.

The scoring service evaluates each new feature vector against the current model, computing an anomaly score. If the score exceeds a configurable threshold (tuned to balance precision and recall for the specific use case), the system generates an anomaly event that can trigger alerts, populate anomaly markers on dashboards, or feed into downstream incident management workflows. The scoring service is designed for low latency (under 10 milliseconds per evaluation) to support real-time alerting on anomalous log patterns.

20. Database & Metadata Schema

CREATE TABLE tenants (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    slug VARCHAR(100) UNIQUE NOT NULL,
    plan VARCHAR(50) NOT NULL DEFAULT 'free',
    settings JSONB DEFAULT '{}',
    ingestion_rate_limit INT DEFAULT 10000,
    storage_quota_bytes BIGINT DEFAULT 1099511627776,
    retention_days INT DEFAULT 30,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID REFERENCES tenants(id),
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    role VARCHAR(50) NOT NULL DEFAULT 'viewer',
    api_key_hash VARCHAR(255),
    last_login_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE log_sources (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID REFERENCES tenants(id),
    name VARCHAR(255) NOT NULL,
    source_type VARCHAR(50) NOT NULL,
    parsing_config JSONB DEFAULT '{}',
    retention_policy JSONB DEFAULT '{}',
    enabled BOOLEAN DEFAULT true
);

CREATE TABLE alert_rules (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID REFERENCES tenants(id),
    name VARCHAR(255) NOT NULL,
    query TEXT NOT NULL,
    condition JSONB NOT NULL,
    notifications JSONB DEFAULT '[]',
    evaluation_interval_minutes INT DEFAULT 1,
    enabled BOOLEAN DEFAULT true,
    last_status VARCHAR(20) DEFAULT 'OK',
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE alert_states (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    rule_id UUID REFERENCES alert_rules(id),
    status VARCHAR(20) NOT NULL DEFAULT 'OK',
    fired_at TIMESTAMPTZ,
    resolved_at TIMESTAMPTZ,
    current_value DOUBLE PRECISION,
    incident_count INT DEFAULT 0,
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE dashboards (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID REFERENCES tenants(id),
    name VARCHAR(255) NOT NULL,
    panel_config JSONB NOT NULL DEFAULT '[]',
    refresh_interval_seconds INT DEFAULT 30,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE audit_logs (
    id BIGSERIAL PRIMARY KEY,
    tenant_id UUID REFERENCES tenants(id),
    user_id UUID REFERENCES users(id),
    action VARCHAR(100) NOT NULL,
    resource_type VARCHAR(100),
    resource_id UUID,
    details JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_users_tenant ON users(tenant_id);
CREATE INDEX idx_sources_tenant ON log_sources(tenant_id);
CREATE INDEX idx_alerts_tenant ON alert_rules(tenant_id);
CREATE INDEX idx_audit_tenant_time ON audit_logs(tenant_id, created_at DESC);

UUIDs for primary keys support distributed ID generation. JSONB columns store semi-structured configuration. Soft deletes preserve data for recovery. Composite indices ensure tenant-scoped queries are efficient.

21. Caching Strategy

The filesystem cache is most impactful. Allocating 50% of each node's memory for OS page cache keeps hot segments in RAM, dramatically reducing search latency for recent data.

22. Multi-Region Design

Data Routing: Log data routes to its source region for data locality and GDPR compliance. Cross-region replication is NOT performed for primary log data (prohibitively expensive at petabyte scale).

Federated Search: Cross-region queries fan out to all clusters in parallel, merge results, and return unified responses. Metadata (alert rules, dashboards, users) replicates across all regions using multi-region PostgreSQL (CockroachDB or YugabyteDB).

Disaster Recovery: If one region fails, surviving regions absorb its traffic with increased latency. Metadata remains available through cross-region replication. Ingestion endpoints auto-failover to nearest healthy region.

23. Cost Estimation

Cost Optimization

• Compression: ZSTD achieves 7:1-10:1 ratios on log data
• Aggressive tiering: Move to warm after 3 days instead of 7 for 50% hot-tier savings
• Field reduction: Index only queried fields for 30-50% index size reduction
• Spot instances: Cold-tier and ingestion workers on spot for 60-70% savings
• Reserved instances: 1-year commitments reduce compute 30-40%

24. Interview Q&A

Q1: How would you design the index for a log aggregation system?

Time-based indices (one per day) with 3 primary shards and 1 replica. ILM moves data from hot (NVMe) after 7 days, to warm (SATA SSD) after 30 days, to cold (S3 snapshots) after 90 days, then deletes. Keyword type for filtered fields, text for the message, numeric for metrics. The key insight is that log data is almost always queried within a time range, so time-based partitioning eliminates the need to scan irrelevant data. Each daily index is approximately 200 GB with compression, and with 3 shards per index, each shard is around 67 GB — slightly above the ideal 10-50 GB range but acceptable for write-once time-based indices that are never updated after initial indexing.

Q2: How do you handle hot/warm/cold efficiently?

Driven by access patterns. Hot nodes use NVMe for write/read throughput. Warm nodes are read-only with force-merged segments for optimal reads on cheaper SATA. Cold uses S3 searchable snapshots — metadata local, data in S3 loaded on-demand. Cost proportional to access frequency. The key trade-off is between query latency and storage cost: hot tier queries return in under 1 millisecond, warm tier in 1-5 milliseconds, cold tier in 5-50 milliseconds, and frozen tier in 100-1000 milliseconds. By tuning the phase transition thresholds based on actual access patterns, you can optimize the cost-performance curve for your specific workload. For most organizations, the default 7-day hot, 30-day warm, 90-day cold, then delete policy works well, but organizations with compliance requirements may need 7-year retention with most of that data in the frozen tier.

Q3: How do you ensure reliable ingestion without data loss?

Three layers: agent persistent registry tracks read offsets for restart recovery. Kafka provides durable replicated storage with 72-hour retention. Workers use at-least-once delivery — offset committed only after successful Elasticsearch write. Deduplication handles duplicates from retries. The critical design decision is the Kafka retention period — it must be long enough to survive a complete ingestion pipeline outage. If your pipeline takes 4 hours to recover from a worst-case failure, 72-hour retention provides an 18x safety margin. The agent's persistent registry is equally critical — without it, an agent restart would either re-deliver all buffered data (if the buffer is in memory) or miss data written since the last flush (if the buffer is only on disk). The registry file tracks the exact byte offset of the last successfully forwarded entry, enabling precise resume after any restart scenario.

Q4: How would you implement search across billions of entries?

Inverted indices map terms to document IDs. Boolean queries intersect posting lists in O(1) per term. Time-based partitioning skips irrelevant indices. Request cache stores identical queries for 5 minutes. Doc-value fields provide columnar storage for aggregations. Coordinating nodes fan out in parallel. The key optimization is the interaction between time-range filtering and inverted index lookups. When a user searches for "error AND timeout" in the last hour, Elasticsearch first identifies which daily indices contain data from the last hour (likely just today's index), then within that index, it intersects the posting lists for "error" and "timeout" to find matching documents. This means the search only touches a tiny fraction of the total data, even though the cluster contains petabytes of logs. The inverted index makes this intersection operation extremely efficient — typically returning results in single-digit milliseconds even against billions of documents.

Q5: How do you handle multi-tenancy?

Shared indices with mandatory tenant_id filtering for small tenants. Dedicated indices for large enterprises. API gateway enforces isolation by injecting tenant_id into every query. RBAC at the application layer. Resource quotas prevent noisy-neighbor effects. The hybrid approach is recommended: tenants with fewer than 1 million log entries per day share indices grouped by plan tier (free, pro, enterprise), while tenants with higher volume get dedicated indices. This balances resource efficiency with isolation. Resource quotas must be enforced at multiple levels: API rate limiting prevents abuse, ingestion rate limiting prevents storage flooding, and search concurrency limits prevent one tenant's complex queries from starving others of search capacity.

Q6: How do you handle parsing failures?

Failed entries go to a dead letter queue (separate Kafka topic). Main pipeline is never blocked. DLQ depth is monitored with alerts. Reprocessing pipeline consumes from DLQ after parser fixes. Raw entries preserved for investigation. The DLQ pattern is essential because parsing failures are inevitable at scale — new log formats are deployed, log lines get truncated by network issues, and encoding problems cause garbled bytes. Without a DLQ, these entries would either block the pipeline (if the parser retries indefinitely) or be silently dropped (if the parser gives up). The DLQ preserves the raw entries for investigation while keeping the main pipeline flowing. A common operational pattern is to alert when DLQ depth exceeds 0.1% of total ingestion volume, investigate the root cause, deploy a parser fix, and reprocess from the DLQ.

Q7: How would you design the alerting system?

Periodic evaluator runs each rule's query, compares to the condition. State machine tracks OK/FIRING/ACKNOWLEDGED/RESOLVED transitions. Flapping prevention via hysteresis. State persisted in PostgreSQL for restart recovery. Multi-channel notifications. The evaluator must handle several edge cases: what happens when the query fails due to a search timeout (treat as no-data, do not fire), what happens when the evaluator is behind schedule (skip late evaluations rather than catching up, to avoid burst notifications), and what happens during Elasticsearch maintenance windows (pause evaluation and send a "monitoring paused" notification to the on-call team).

Q8: How do you scale the ingestion pipeline?

Horizontally by adding Kafka partitions and consumer instances. Kafka topics partitioned by source for ordering. Each worker processes parsing, enrichment, normalization in parallel. Bulk writes to Elasticsearch. Stateless workers enable auto-scaling. The scaling limit is the number of Kafka partitions — if you have 1,000 Kafka partitions, you can have at most 1,000 consumer instances in a single consumer group. For most deployments, 100-200 partitions per topic is sufficient. If you need more parallelism, you can create multiple topics (one per source category) or increase the batch size per bulk write. Elasticsearch bulk write performance depends on the refresh_interval — a 30-second refresh interval allows batching writes for 30 seconds before they become searchable, significantly improving write throughput compared to a 1-second refresh interval.

Q9: How do you keep search fast at petabyte scale?

Time-based partitioning, tiered storage, force-merging to single segments, field reduction, 10-50 GB shard sizing, filesystem cache for hot segments, request caching for repeated queries, and preference routing for cache affinity. At petabyte scale, the most impactful optimization is reducing the amount of data scanned per query. A well-designed log aggregation system scans less than 1% of total stored data for a typical query, thanks to time-range filtering on daily indices. Combined with the inverted index for keyword lookups, the effective scan volume for a 24-hour query is typically under 100 GB even on a 200 TB cluster. The filesystem cache ensures that this 100 GB is served from RAM rather than disk, providing sub-second response times for the majority of queries.

Q10: How do you handle schema evolution?

Elasticsearch dynamic mapping auto-detects new fields. Index templates define expected fields per source. New fields without templates use auto-detection. Old entries have null for new fields — backward compatible. Breaking changes require coordinated parser and template updates. The practical approach is to version your index templates and maintain backward-compatible field additions. For example, if you need to rename a field from "level" to "severity", you add "severity" as a new field, update the parser to write both fields, and gradually migrate dashboards and queries to use the new field name. Once all consumers are migrated, you can stop writing to the old field. This "expand and contract" pattern avoids breaking existing queries during field renames.

Q11: How do you implement distributed tracing integration?

Every log entry includes trace_id and span_id, injected by OpenTelemetry SDK. Fields indexed as keywords for efficient lookup. "View Trace" action queries the trace store and returns the waterfall. Cross-linking enables bidirectional navigation. The key challenge is maintaining correlation across heterogeneous services. Not all services use the same tracing library or version, and some legacy services may not be instrumented at all. The recommended approach is to use OpenTelemetry auto-instrumentation for all supported languages, providing zero-code tracing for most services, and manual instrumentation only for legacy or custom protocol services. The trace context must be propagated through HTTP headers (W3C Trace Context standard), message queue attributes, and database query comments to maintain correlation across the entire request path.

Q12: How do you handle GDPR right to erasure?

user_id field must be indexed. Delete By Query API removes matching documents across all time-based indices. Audit log records the deletion request and completion. Retention system ensures no backups retain deleted data beyond a recovery window. The practical challenge is performance — Delete By Query on a 90-day cluster with hundreds of indices can take hours and consume significant I/O. A better approach is to implement soft deletion at index time: mark records as deleted in a "deleted" boolean field, filter them out at query time, and let ILM handle physical deletion during normal index rotation. For urgent erasure requests, use the Delete By Query API on specific time-range indices rather than the entire cluster. Additionally, ensure that any downstream systems (S3 archives, backup snapshots) also honor the deletion request, which may require explicit snapshot invalidation.

25. Full C# Implementation

Core Domain Models

using System;
using System.Collections.Generic;

namespace LogAggregation.Core.Models
{
    public enum LogLevel { TRACE, DEBUG, INFO, WARN, ERROR, FATAL, CRITICAL }
    public enum AlertStatus { OK, FIRING, ACKNOWLEDGED, RESOLVED }
    public enum SourceType { Application, Infrastructure, Security, Audit, Network }

    public class LogEntry
    {
        public string Id { get; set; } = Guid.NewGuid().ToString();
        public DateTime Timestamp { get; set; } = DateTime.UtcNow;
        public DateTime ReceivedAt { get; set; } = DateTime.UtcNow;
        public string SourceId { get; set; } = string.Empty;
        public SourceType SourceType { get; set; }
        public LogLevel Severity { get; set; }
        public string Service { get; set; } = string.Empty;
        public string Host { get; set; } = string.Empty;
        public string Message { get; set; } = string.Empty;
        public string? StackTrace { get; set; }
        public string? TraceId { get; set; }
        public string? SpanId { get; set; }
        public Dictionary<string, object> Fields { get; set; } = new();
        public List<string> Tags { get; set; } = new();
        public string IndexPattern { get; set; } = string.Empty;
    }

    public class LogSource
    {
        public string Id { get; set; } = Guid.NewGuid().ToString();
        public string TenantId { get; set; } = string.Empty;
        public string Name { get; set; } = string.Empty;
        public SourceType Type { get; set; }
        public ParsingConfig ParsingConfig { get; set; } = new();
        public RetentionPolicy RetentionPolicy { get; set; } = new();
        public bool Enabled { get; set; } = true;
    }

    public class ParsingConfig
    {
        public string Format { get; set; } = "json";
        public Dictionary<string, string> GrokPatterns { get; set; } = new();
        public List<RedactionRule> RedactionRules { get; set; } = new();
    }

    public class RedactionRule
    {
        public string Field { get; set; } = string.Empty;
        public string? Pattern { get; set; }
        public string Replacement { get; set; } = "[REDACTED]";
        public string? Type { get; set; }
    }

    public class RetentionPolicy
    {
        public int HotDays { get; set; } = 7;
        public int WarmDays { get; set; } = 30;
        public int ColdDays { get; set; } = 90;
        public int RetentionDays { get; set; } = 90;
    }

    public class AlertRule
    {
        public string Id { get; set; } = Guid.NewGuid().ToString();
        public string TenantId { get; set; } = string.Empty;
        public string Name { get; set; } = string.Empty;
        public string Query { get; set; } = string.Empty;
        public AlertCondition Condition { get; set; } = new();
        public List<NotificationTarget> Notifications { get; set; } = new();
        public int EvaluationIntervalMinutes { get; set; } = 1;
        public bool Enabled { get; set; } = true;
    }

    public class AlertCondition
    {
        public string Metric { get; set; } = "count";
        public string? Field { get; set; }
        public int WindowMinutes { get; set; } = 5;
        public string Operator { get; set; } = "gt";
        public double Value { get; set; }
    }

    public class NotificationTarget
    {
        public string Type { get; set; } = string.Empty;
        public string Endpoint { get; set; } = string.Empty;
    }

    public class AlertState
    {
        public string RuleId { get; set; } = string.Empty;
        public AlertStatus Status { get; set; } = AlertStatus.OK;
        public DateTime? FiredAt { get; set; }
        public DateTime? ResolvedAt { get; set; }
        public double CurrentValue { get; set; }
        public int IncidentCount { get; set; }
    }

    public class SearchQuery
    {
        public string Query { get; set; } = string.Empty;
        public DateTime TimeFrom { get; set; }
        public DateTime TimeTo { get; set; }
        public List<string> Fields { get; set; } = new();
        public List<SortClause> Sort { get; set; } = new();
        public int From { get; set; }
        public int Size { get; set; } = 50;
        public List<FilterClause> Filters { get; set; } = new();
        public Dictionary<string, AggSpec> Aggregations { get; set; } = new();
    }

    public class SortClause
    {
        public string Field { get; set; } = string.Empty;
        public string Order { get; set; } = "desc";
    }

    public class FilterClause
    {
        public string Field { get; set; } = string.Empty;
        public string Operator { get; set; } = "eq";
        public object Value { get; set; } = null!;
    }

    public class AggSpec
    {
        public string Type { get; set; } = "terms";
        public string Field { get; set; } = string.Empty;
        public int Size { get; set; } = 10;
    }

    public class SearchResult
    {
        public long Total { get; set; }
        public int TookMs { get; set; }
        public List<LogEntry> Hits { get; set; } = new();
        public Dictionary<string, object> Aggregations { get; set; } = new();
    }
}

Ingestion Service

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;

namespace LogAggregation.Core.Services
{
    public class LogParser
    {
        private readonly Dictionary<string, Regex> _patterns;
        private readonly ILogger<LogParser> _logger;

        public LogParser(LogParserConfig config)
        {
            _patterns = new Dictionary<string, Regex>();
            foreach (var p in config.Patterns)
                _patterns[p.Key] = new Regex(p.Value, RegexOptions.Compiled);
            _logger = LoggerFactory.Create(b => b.AddConsole())
                .CreateLogger<LogParser>();
        }

        public ParseResult Parse(LogEntry raw, LogSource source)
        {
            if (string.IsNullOrWhiteSpace(raw.Message))
                return ParseResult.Failed("Empty message");

            if (raw.Message.TrimStart().StartsWith("{"))
            {
                try
                {
                    var fields = JsonSerializer.Deserialize
                        <Dictionary<string, object>>(raw.Message);
                    if (fields != null)
                    {
                        foreach (var f in fields)
                            raw.Fields[f.Key] = f.Value;
                        return ParseResult.Success(raw, "json");
                    }
                }
                catch (JsonException) { }
            }

            foreach (var pat in _patterns)
            {
                var match = pat.Value.Match(raw.Message);
                if (match.Success)
                {
                    foreach (string name in match.Groups.Keys)
                    {
                        if (name != "0")
                            raw.Fields[name] = match.Groups[name].Value;
                    }
                    return ParseResult.Success(raw, pat.Key);
                }
            }

            raw.Fields["unparsed"] = true;
            return ParseResult.Partial(raw, "raw_fallback");
        }
    }

    public class ParseResult
    {
        public bool Success { get; set; }
        public LogEntry? Entry { get; set; }
        public string Pattern { get; set; } = string.Empty;
        public string? ErrorMessage { get; set; }

        public static ParseResult Success(LogEntry entry, string pattern)
            => new() { Success = true, Entry = entry, Pattern = pattern };
        public static ParseResult Failed(string error)
            => new() { Success = false, ErrorMessage = error };
        public static ParseResult Partial(LogEntry entry, string pattern)
            => new() { Success = true, Entry = entry, Pattern = pattern };
    }

    public class LogEnricher
    {
        private readonly Dictionary<string, LogSource> _sourceCache;

        public LogEnricher()
        {
            _sourceCache = new Dictionary<string, LogSource>();
        }

        public LogEntry Enrich(LogEntry entry, LogSource source)
        {
            entry.SourceId = source.Id;
            entry.SourceType = source.Type;
            if (string.IsNullOrEmpty(entry.Service))
                entry.Service = source.Name;

            if (!entry.Tags.Contains(source.Type.ToString().ToLower()))
                entry.Tags.Add(source.Type.ToString().ToLower());

            if (entry.Fields.TryGetValue("remote_addr", out var addr))
            {
                var geo = LookupGeoIp(addr?.ToString() ?? "");
                if (geo != null)
                {
                    entry.Fields["geo_country"] = geo.Country;
                    entry.Fields["geo_city"] = geo.City;
                }
            }

            return entry;
        }

        private GeoInfo? LookupGeoIp(string ip)
        {
            if (string.IsNullOrEmpty(ip)) return null;
            return new GeoInfo { Country = "US", City = "Unknown" };
        }
    }

    public class GeoInfo
    {
        public string Country { get; set; } = "";
        public string City { get; set; } = "";
    }

    public class IngestionService
    {
        private readonly LogParser _parser;
        private readonly LogEnricher _enricher;
        private readonly IBulkWriter _writer;
        private readonly ILogger<IngestionService> _logger;
        private long _totalIngested;
        private long _totalFailed;

        public IngestionService(
            LogParser parser,
            LogEnricher enricher,
            IBulkWriter writer,
            ILogger<IngestionService> logger)
        {
            _parser = parser;
            _enricher = enricher;
            _writer = writer;
            _logger = logger;
        }

        public async Task<IngestionResult> IngestAsync(
            string sourceId, IEnumerable<LogEntry> entries, LogSource source)
        {
            var result = new IngestionResult { SourceId = sourceId };
            foreach (var entry in entries)
            {
                try
                {
                    var parseResult = _parser.Parse(entry, source);
                    if (!parseResult.Success || parseResult.Entry == null)
                    {
                        Interlocked.Increment(ref _totalFailed);
                        result.FailedCount++;
                        continue;
                    }

                    var enriched = _enricher.Enrich(parseResult.Entry, source);
                    enriched.IndexPattern =
                        $"logs-{source.Name}-{enriched.Timestamp:yyyy.MM.dd}";

                    ApplyRedactions(enriched, source.ParsingConfig.RedactionRules);
                    await _writer.WriteAsync(enriched);

                    Interlocked.Increment(ref _totalIngested);
                    result.SuccessCount++;
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Ingestion failed for {Source}", sourceId);
                    Interlocked.Increment(ref _totalFailed);
                    result.FailedCount++;
                }
            }
            return result;
        }

        private void ApplyRedactions(LogEntry entry, List<RedactionRule> rules)
        {
            if (rules == null) return;
            foreach (var rule in rules)
            {
                if (rule.Type == "hash" &&
                    entry.Fields.TryGetValue(rule.Field, out var val))
                {
                    using var sha = System.Security.Cryptography.SHA256.Create();
                    var bytes = Encoding.UTF8.GetBytes(val?.ToString() ?? "");
                    entry.Fields[rule.Field] = Convert.ToBase64String(
                        sha.ComputeHash(bytes));
                }
                else if (!string.IsNullOrEmpty(rule.Pattern) &&
                    entry.Fields.TryGetValue(rule.Field, out var textVal))
                {
                    entry.Fields[rule.Field] = Regex.Replace(
                        textVal?.ToString() ?? "", rule.Pattern, rule.Replacement);
                }
            }
        }

        public long GetTotalIngested() => Interlocked.Read(ref _totalIngested);
        public long GetTotalFailed() => Interlocked.Read(ref _totalFailed);
    }

    public class IngestionResult
    {
        public string SourceId { get; set; } = "";
        public int SuccessCount { get; set; }
        public int FailedCount { get; set; }
    }

    public interface IBulkWriter
    {
        Task WriteAsync(LogEntry entry);
        Task FlushAsync();
    }

    public class LogParserConfig
    {
        public Dictionary<string, string> Patterns { get; set; } = new();
    }
}

Search Engine

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using LogAggregation.Core.Models;

namespace LogAggregation.Core.Services
{
    public interface ISearchEngine
    {
        Task<SearchResult> SearchAsync(SearchQuery query, string tenantId);
        Task<List<LogEntry>> GetByTraceIdAsync(
            string traceId, string tenantId);
    }

    public class SearchEngine : ISearchEngine
    {
        private readonly IElasticClientWrapper _client;
        private readonly ILogger<SearchEngine> _logger;

        public SearchEngine(
            IElasticClientWrapper client,
            ILogger<SearchEngine> logger)
        {
            _client = client;
            _logger = logger;
        }

        public async Task<SearchResult> SearchAsync(
            SearchQuery query, string tenantId)
        {
            var sw = Stopwatch.StartNew();
            var boolQuery = BuildBoolQuery(query, tenantId);
            var indexPattern = $"logs-{tenantId}-*";

            var searchRequest = new
            {
                query = boolQuery,
                sort = query.Sort.Any()
                    ? query.Sort.Select(s =>
                        new Dictionary<string, object>
                        { { s.Field, new { order = s.Order } } }).ToArray()
                    : new[] { new Dictionary<string, object>
                        { { "timestamp", new { order = "desc" } } } },
                _source = query.Fields.Any()
                    ? (object)query.Fields.ToArray() : new[] { "*" },
                from = query.From,
                size = query.Size,
                aggs = BuildAggs(query.Aggregations)
            };

            var response = await _client.SearchAsync<LogEntry>(
                indexPattern, searchRequest);

            sw.Stop();

            return new SearchResult
            {
                Total = response.TotalHits,
                TookMs = (int)sw.ElapsedMilliseconds,
                Hits = response.Documents.ToList(),
                Aggregations = response.Aggs ?? new()
            };
        }

        public async Task<List<LogEntry>> GetByTraceIdAsync(
            string traceId, string tenantId)
        {
            var query = new SearchQuery
            {
                Query = $"trace_id:{traceId}",
                TimeFrom = DateTime.UtcNow.AddHours(-24),
                TimeTo = DateTime.UtcNow,
                Sort = new() { new() { Field = "timestamp", Order = "asc" } },
                Size = 1000
            };
            var result = await SearchAsync(query, tenantId);
            return result.Hits;
        }

        private object BuildBoolQuery(SearchQuery query, string tenantId)
        {
            var must = new List<object>();
            var filter = new List<object>
            {
                new { term = new { tenant_id = tenantId } },
                new { range = new
                {
                    timestamp = new
                    {
                        gte = query.TimeFrom.ToString("o"),
                        lte = query.TimeTo.ToString("o")
                    }
                }}
            };

            if (!string.IsNullOrWhiteSpace(query.Query))
                must.Add(new { query_string = new
                    { query = query.Query, default_field = "message" } });

            foreach (var f in query.Filters)
                filter.Add(BuildFilter(f));

            return new { @bool = new
            {
                must = must.Any()
                    ? must.ToArray()
                    : new[] { new { match_all = new { } } },
                filter = filter.ToArray()
            }};
        }

        private object BuildFilter(FilterClause f) => f.Operator switch
        {
            "eq" => new { term = new Dictionary<string, object>
                { { f.Field, f.Value } } },
            "gt" => new { range = new Dictionary<string, object>
                { { f.Field, new { gt = f.Value } } } },
            "lt" => new { range = new Dictionary<string, object>
                { { f.Field, new { lt = f.Value } } } },
            "contains" => new { match_phrase = new Dictionary<string, object>
                { { f.Field, new { query = f.Value } } } },
            _ => throw new ArgumentException($"Unknown op: {f.Operator}")
        };

        private object BuildAggs(Dictionary<string, AggSpec> specs)
        {
            var aggs = new Dictionary<string, object>();
            foreach (var s in specs)
            {
                aggs[s.Key] = s.Value.Type switch
                {
                    "terms" => new { terms = new
                        { field = s.Value.Field, size = s.Value.Size } },
                    "date_histogram" => new { date_histogram = new
                        { field = s.Value.Field, fixed_interval = "5m" } },
                    "avg" => new { avg = new { field = s.Value.Field } },
                    _ => new { terms = new
                        { field = s.Value.Field, size = s.Value.Size } }
                };
            }
            return aggs;
        }
    }

    public interface IElasticClientWrapper
    {
        Task<ElasticResponse> SearchAsync<T>(
            string index, object request);
    }

    public class ElasticResponse
    {
        public long TotalHits { get; set; }
        public List<LogEntry> Documents { get; set; } = new();
        public Dictionary<string, object>? Aggs { get; set; }
    }
}

Alert Evaluator

using System;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using LogAggregation.Core.Models;

namespace LogAggregation.Core.Services
{
    public interface IAlertEvaluator
    {
        Task EvaluateAsync(AlertRule rule, string tenantId);
    }

    public class AlertEvaluator : IAlertEvaluator
    {
        private readonly ISearchEngine _search;
        private readonly INotificationService _notifier;
        private readonly IAlertStateStore _stateStore;
        private readonly ILogger<AlertEvaluator> _logger;

        public AlertEvaluator(
            ISearchEngine search,
            INotificationService notifier,
            IAlertStateStore stateStore,
            ILogger<AlertEvaluator> logger)
        {
            _search = search;
            _notifier = notifier;
            _stateStore = stateStore;
            _logger = logger;
        }

        public async Task EvaluateAsync(AlertRule rule, string tenantId)
        {
            _logger.LogDebug("Evaluating alert: {Name}", rule.Name);

            var query = new SearchQuery
            {
                Query = rule.Query,
                TimeFrom = DateTime.UtcNow.AddMinutes(
                    -rule.Condition.WindowMinutes),
                TimeTo = DateTime.UtcNow
            };

            var result = await _search.SearchAsync(query, tenantId);
            var currentValue = (double)result.Total;
            var thresholdMet = EvaluateCondition(
                currentValue,
                rule.Condition.Operator,
                rule.Condition.Value);

            var state = await _stateStore.GetAsync(rule.Id);

            if (thresholdMet)
            {
                if (state?.Status != AlertStatus.FIRING)
                {
                    var newState = new AlertState
                    {
                        RuleId = rule.Id,
                        Status = AlertStatus.FIRING,
                        FiredAt = DateTime.UtcNow,
                        CurrentValue = currentValue,
                        IncidentCount = 1
                    };
                    await _stateStore.SaveAsync(newState);
                    await _notifier.NotifyAsync(rule,
                        $"FIRING: {rule.Name} - value {currentValue} " +
                        $"{rule.Condition.Operator} {rule.Condition.Value}");
                    _logger.LogWarning(
                        "Alert fired: {Name} (value={Value})",
                        rule.Name, currentValue);
                }
                else
                {
                    state.CurrentValue = currentValue;
                    state.IncidentCount++;
                    await _stateStore.SaveAsync(state);
                }
            }
            else if (state?.Status == AlertStatus.FIRING)
            {
                state.Status = AlertStatus.OK;
                state.ResolvedAt = DateTime.UtcNow;
                await _stateStore.SaveAsync(state);
                await _notifier.NotifyAsync(rule,
                    $"RESOLVED: {rule.Name}");
                _logger.LogInformation(
                    "Alert resolved: {Name}", rule.Name);
            }
        }

        private bool EvaluateCondition(
            double value, string op, double threshold) => op switch
        {
            "gt" => value > threshold,
            "gte" => value >= threshold,
            "lt" => value < threshold,
            "lte" => value <= threshold,
            "eq" => Math.Abs(value - threshold) < 0.001,
            _ => false
        };
    }

    public interface INotificationService
    {
        Task NotifyAsync(AlertRule rule, string message);
    }

    public interface IAlertStateStore
    {
        Task<AlertState?> GetAsync(string ruleId);
        Task SaveAsync(AlertState state);
    }
}

Dashboard Service

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using LogAggregation.Core.Models;

namespace LogAggregation.Core.Services
{
    public class DashboardService
    {
        private readonly IDashboardRepository _repo;
        private readonly ISearchEngine _search;

        public DashboardService(
            IDashboardRepository repo, ISearchEngine search)
        {
            _repo = repo;
            _search = search;
        }

        public async Task<DashboardData> GetDashboardDataAsync(
            string dashboardId, string tenantId, DateTime? timeFrom = null,
            DateTime? timeTo = null)
        {
            var dashboard = await _repo.GetByIdAsync(dashboardId, tenantId);
            if (dashboard == null)
                throw new KeyNotFoundException(
                    $"Dashboard {dashboardId} not found");

            var from = timeFrom ?? DateTime.UtcNow.AddHours(-24);
            var to = timeTo ?? DateTime.UtcNow;

            var panelDataTasks = new List<Task<PanelData>>();
            foreach (var panel in dashboard.Panels)
            {
                panel.TimeFrom = from;
                panel.TimeTo = to;
                panelDataTasks.Add(
                    GetPanelDataAsync(panel, tenantId));
            }

            var panelResults = await Task.WhenAll(panelDataTasks);
            return new DashboardData
            {
                Dashboard = dashboard,
                Panels = panelResults,
                TimeRange = new() { From = from, To = to }
            };
        }

        private async Task<PanelData> GetPanelDataAsync(
            DashboardPanel panel, string tenantId)
        {
            var query = panel.Query;
            query.TimeFrom = panel.TimeFrom;
            query.TimeTo = panel.TimeTo;

            var searchResult = await _search.SearchAsync(query, tenantId);
            return new PanelData
            {
                PanelId = panel.Id,
                Title = panel.Title,
                Type = panel.Type,
                TotalHits = searchResult.Total,
                Hits = searchResult.Hits,
                Aggregations = searchResult.Aggregations,
                TookMs = searchResult.TookMs
            };
        }
    }

    public class DashboardData
    {
        public Dashboard Dashboard { get; set; } = null!;
        public PanelData[] Panels { get; set; } = Array.Empty<PanelData>();
        public TimeRange TimeRange { get; set; } = new();
    }

    public class PanelData
    {
        public string PanelId { get; set; } = "";
        public string Title { get; set; } = "";
        public string Type { get; set; } = "";
        public long TotalHits { get; set; }
        public List<LogEntry> Hits { get; set; } = new();
        public Dictionary<string, object> Aggregations { get; set; } = new();
        public int TookMs { get; set; }
    }

    public class TimeRange
    {
        public DateTime From { get; set; }
        public DateTime To { get; set; }
    }

    public interface IDashboardRepository
    {
        Task<Dashboard?> GetByIdAsync(string id, string tenantId);
        Task<List<Dashboard>> GetByTenantAsync(string tenantId);
        Task SaveAsync(Dashboard dashboard);
        Task DeleteAsync(string id, string tenantId);
    }

    public class DashboardPanel
    {
        public string Id { get; set; } = Guid.NewGuid().ToString();
        public string Type { get; set; } = "line_chart";
        public string Title { get; set; } = "";
        public SearchQuery Query { get; set; } = new();
        public DateTime TimeFrom { get; set; }
        public DateTime TimeTo { get; set; }
    }
}

26. Conclusion

Designing a log aggregation platform like ELK or Splunk requires careful consideration of every layer in the stack — from lightweight collection agents to durable message queues, from sophisticated parsing engines to tiered storage architectures, and from powerful search engines to real-time alerting systems. The key design principles that emerge throughout this guide are:

Decouple aggressively. Every stage of the pipeline — collection, transport, ingestion, storage, and serving — should be independently scalable and failure-isolated. Kafka in the transport layer is the single most impactful architectural decision for achieving this decoupling.

Optimize for the access pattern. Log data is write-once-read-many with extreme temporal locality. Time-based index partitioning, tiered storage, inverted indices, and filesystem caching all exploit this fundamental characteristic.

Accept imperfection gracefully. Parsing failures will happen. Network partitions will occur. Nodes will crash. The system must handle every failure mode without losing data or blocking the pipeline. Dead letter queues, at-least-once delivery, and idempotent writes are essential patterns.

Make the common case fast. Most searches cover the last 24 hours. Most users search the same services repeatedly. Most dashboard queries are identical within the refresh interval. Optimize the hot path — caching, index design, and shard placement should prioritize the 95% case.

Control costs proactively. At petabyte scale, storage and compute costs can spiral without active management. Compression, aggressive tiering, field reduction, spot instances, and reserved pricing must be built into the architecture from day one, not bolted on later.

The complete C# implementation in this guide demonstrates production-quality patterns for ingestion, parsing, search, alerting, and dashboard services. While the specific technologies may vary — Elasticsearch vs. ClickHouse for storage, Kafka vs. Pulsar for transport, Kibana vs. Grafana for visualization — the fundamental architectural patterns remain consistent across implementations.

Whether you are building a logging platform from scratch, evaluating commercial solutions, or preparing for a system design interview, the mental models and trade-off analysis in this guide will help you reason about log aggregation at any scale. The journey from 100 log lines per second to 1 million is not about finding the perfect architecture — it is about making the right trade-offs at each scale boundary and knowing which knobs to turn when the next order of magnitude arrives.

At small scale (thousands of log lines per second), a simple Elasticsearch cluster with Filebeat agents is sufficient. The operational overhead is minimal, and the cost is manageable even with a single-node deployment. As you grow to medium scale (hundreds of thousands of lines per second), you need Kafka for buffering, a dedicated ingestion pipeline with multiple worker instances, and careful index management with ILM policies. The team will need dedicated platform engineers to manage the Elasticsearch cluster health, tuning, and upgrades. At large scale (millions of lines per second), you need a fully distributed architecture with multi-region deployment, federated search, ML-based anomaly detection, and aggressive cost optimization through storage tiering, compression, and spot instance usage.

The most successful log aggregation platforms share several characteristics: they treat logs as a first-class data product rather than an afterthought, they invest heavily in parsing and normalization to ensure data quality and consistency across heterogeneous sources, they provide self-service capabilities for product teams to onboard their own log sources and create their own dashboards without requiring platform team involvement, and they maintain a clear cost attribution model so that each team understands the cost implications of their logging volume, retention choices, and query patterns. Building these platforms is a significant engineering investment, but the return — faster incident resolution, deeper system visibility, and the ability to debug complex distributed systems with confidence — is transformational for any engineering organization that operates at scale.