System Design: Design a Distributed Logging System

When a distributed system has hundreds of services producing logs, grepping files on individual machines stops being viable. You need a centralized logging system that collects, indexes, and makes all logs searchable from a single interface. Systems like the ELK stack (Elasticsearch, Logstash, Kibana), Splunk, and Grafana Loki solve this problem at different scales and trade-off points.

Functional Requirements

• Collect logs from all application instances and infrastructure components.
• Support structured (JSON) and unstructured (plaintext) log formats.
• Provide full-text search and field-based filtering across all logs.
• Support time-range queries (show me all errors from the payment service in the last hour).
• Retain logs for configurable periods (7 days hot, 90 days warm, 1 year cold).
• Support alerting on log patterns (trigger an alert when error rate exceeds a threshold).

Non-Functional Requirements

• Throughput: ingest millions of log lines per second.
• Latency: logs should be searchable within 30 seconds of emission.
• Durability: once a log line is acknowledged, it must not be lost.
• Availability: the query interface must remain available even during ingestion spikes.

High-Level Architecture

The system has four stages:

1. Collection — agents running on each host ship logs to the ingestion layer.
2. Ingestion — a message queue buffers and decouples collection from processing.
3. Processing — workers parse, enrich, and index logs.
4. Storage and query — an indexed store serves search queries.

[App] -> [Agent] -> [Message Queue] -> [Processor] -> [Index Store] -> [Query API] -> [Dashboard]

Log Collection

Each application host runs a lightweight log agent (like Filebeat or Fluentd). The agent tails log files, applies minimal parsing, and ships log entries to the ingestion layer over TCP or HTTP.

Agent design principles:

• Durability: track the current read position in each log file using a checkpoint file. If the agent restarts, it resumes from the last checkpoint, preventing data loss.
• Backpressure: if the downstream ingestion layer is slow, the agent buffers locally (in memory, then spilling to disk) rather than dropping logs.
• Low overhead: the agent must consume minimal CPU and memory. It should not compete with the application for resources.

Structured logging. Encourage application teams to emit structured JSON logs rather than free-form text. A structured log line includes fields like timestamp, level, service, trace_id, and message. This eliminates the need for expensive regex parsing downstream.

Ingestion Layer

Place a message queue (Kafka) between the agents and the processing layer. This provides three benefits:

1. Buffering: absorbs ingestion spikes without overloading the processors.
2. Decoupling: agents and processors can scale independently.
3. Replay: if a processor bug corrupts indexes, you can replay logs from Kafka to reindex.

Partition Kafka topics by service name or a hash of the log source. This ensures logs from the same service land in the same partition, enabling ordered processing per service.

Set a retention period on Kafka that covers your reprocessing window (for example, 72 hours). This is your safety net for reindexing.

Processing Pipeline

Processing workers consume from Kafka and prepare logs for indexing. The pipeline has several stages:

Parsing. For structured logs, validate the JSON and extract fields. For unstructured logs, apply grok patterns or regex to extract timestamp, level, and message.

Enrichment. Add metadata that is not in the original log: hostname-to-service mapping, datacenter, Kubernetes pod name, or GeoIP data for access logs.

Normalization. Standardize field names and timestamp formats across services. One service might log ts while another logs @timestamp. Normalize to a single schema.

Filtering. Drop or sample low-value logs (debug-level logs in production, health check access logs) to reduce storage costs.

Batching. Accumulate processed logs into batches (by size or time window) before writing to the index store. Bulk writes are dramatically faster than individual writes for most storage engines.

Storage and Indexing

The storage engine must support two access patterns: fast writes (millions of log lines per second) and fast reads (full-text search and field filtering).

Inverted index approach (Elasticsearch). Each log line is a document. An inverted index maps each token (word, field value) to the list of documents containing it. This enables fast full-text search. Documents are stored in time-based indices (one index per day), making retention simple — delete old indices.

Label-based approach (Loki). Instead of indexing the full log content, index only metadata labels (service, level, pod). Store log lines in compressed chunks grouped by label set. Queries first filter by labels (fast index lookup) and then grep through the matching chunks. This trades query flexibility for dramatically lower storage and indexing costs.

Choose based on your query patterns. If users need ad-hoc full-text search across arbitrary fields, use an inverted index. If most queries filter by service and time range, the label-based approach is more cost-effective.

Tiered Storage

Logs have a steep access curve: most queries target the last few hours. Older logs are rarely searched but must be retained for compliance.

Hot tier (0-7 days): SSDs, fully indexed, fast queries. This is your primary working set.

Warm tier (7-90 days): HDDs or cheaper SSDs, fully indexed but slower. Queries take longer but still return results.

Cold tier (90 days - 1 year): object storage (S3). Logs are compressed and stored as archived files. Querying requires rehydrating the data, which takes minutes. Suitable for compliance and forensic investigation.

Implement lifecycle policies that automatically migrate indices from hot to warm to cold based on age. In Elasticsearch, Index Lifecycle Management (ILM) handles this natively.

Query Engine

The query API accepts requests like “show me all logs from payment-service where level = ERROR and message contains 'timeout' in the last 2 hours.”

Query execution:

1. Determine which indices or chunks cover the requested time range.
2. Apply field filters to narrow down candidate documents.
3. Apply full-text search within the filtered set.
4. Sort by timestamp (descending) and return the top N results.

Pagination. For large result sets, use cursor-based pagination (search_after in Elasticsearch) rather than offset-based pagination. Offset pagination degrades badly for deep pages.

Aggregations. Support count, histogram, and top-N aggregations for analytics. “How many errors per minute from the payment service in the last 24 hours?” These power dashboards and trend analysis.

Alerting

Alerting bridges logging and monitoring. Define alert rules like “if more than 100 ERROR logs from payment-service in any 5-minute window, notify the on-call engineer.”

Run alerting as a separate component that periodically executes queries against the index store. Use a sliding window or tumbling window to evaluate conditions. When a condition triggers, deduplicate (do not alert every 5 minutes for an ongoing incident) and send notifications via configured channels (Slack, PagerDuty, email).

Scaling Considerations

• Ingestion scaling: add more Kafka partitions and processing workers. Kafka scales linearly with partitions.
• Storage scaling: Elasticsearch scales by adding nodes to the cluster and distributing shards across them. Monitor shard count per node (keep it under 1000).
• Query isolation: separate read and write traffic. Use dedicated coordinating nodes for queries so that heavy searches do not slow down ingestion.
• Sampling: at extreme scale, sample verbose log levels (DEBUG, TRACE) rather than ingesting every line. Keep all ERROR and WARN logs.

Summary

A distributed logging system is a pipeline: collect at the edge, buffer through a message queue, process and index in parallel, store in tiers, and query through an API. The most impactful decisions are whether to index full text or just labels (cost vs. query flexibility), how to tier storage (hot/warm/cold), and how to handle backpressure through the pipeline so that ingestion spikes do not cause data loss.