Design a Logging and Monitoring System

Functional Requirements

• ·Log ingestion: Collect logs from 700+ microservices running across AWS (EC2, containers)
• ·Structured logging: All logs emitted as structured JSON with standard fields (timestamp, service, severity, trace_id, message, metadata)
• ·Distributed tracing: Trace a single user request across 20+ microservices (e.g., Zuul gateway -> API -> Recommendation -> Model Serving -> Feature Store)
• ·Search: Full-text search across logs with filters (service, severity, time range, trace_id)
• ·Alerting: Trigger alerts based on log patterns, metric thresholds, and anomaly detection
• ·Dashboards: Real-time visualization of metrics, log aggregates, and trace waterfalls
• ·Retention: Configurable retention per log type and severity (7 days hot, 30 days warm, 90 days cold)

Non-Functional Requirements

• ·Throughput: Ingest 10M+ log events/sec across all microservices
• ·Latency: Logs searchable within 30 seconds of emission
• ·Scale: Petabytes of log data; hundreds of billions of time-series data points in Atlas
• ·Durability: No loss of ERROR/FATAL logs; best-effort for DEBUG
• ·Availability: 99.9% for search/dashboards; 99.99% for ingestion and alerting (must monitor even during outages)

Constraints

• ·Netflix runs entirely on AWS (no GCP/Azure) -- all storage and compute choices must be AWS-native or self-managed on EC2
• ·700+ microservices with heterogeneous tech stacks (Java, Node.js, Python)
• ·Chaos engineering culture (Chaos Monkey) means monitoring must be resilient to random instance termination
• ·Read-heavy for dashboards (thousands of engineers viewing dashboards simultaneously) but write-heavy for ingestion

Log Storage: Elasticsearch (hot) + S3 (cold)

Why Elasticsearch for hot tier? Full-text search on log messages is the primary query pattern. ES provides near-real-time indexing (~1s refresh) and powerful query DSL. Netflix engineers need to search by arbitrary text patterns during incidents.

Metrics: Atlas (Netflix Custom TSDB)

Why Atlas (custom) over Prometheus? Netflix’s scale (2B+ data points/sec) exceeds single-node Prometheus capacity. Atlas is designed for dimensional time-series at Netflix’s scale, with a custom query language (Atlas Stack Language) that supports real-time streaming evaluation.

Trace Storage: Cassandra

Why Cassandra? Write-optimized for high-throughput span ingestion. Trace queries are always by trace_id (partition key), making Cassandra’s key-value lookup pattern ideal. Netflix already operates Cassandra at massive scale across AWS.

Alert Rules: PostgreSQL (RDS)

Data Flow: Log Ingestion

1. Each microservice emits structured JSON logs to stdout/stderr

2. Sidecar agent (Fluentd) on each host reads logs, buffers locally (disk-backed)

3. Agent batches logs (every 5 seconds or 1000 events) and forwards to Kafka log topics

4. Kafka partitions logs by service name for processing locality

5. Indexer workers (Kafka consumers) bulk-index logs into Elasticsearch (batch of 5000 docs)

6. Archiver (separate consumer group) writes logs in Parquet format to S3 for cold storage

7. Alert Evaluator (streaming consumer) evaluates log-based alert rules in real-time

Data Flow: Distributed Trace

1. User request hits Zuul API gateway, which generates a trace_id (UUID) and first span

2. trace_id propagated via HTTP headers (X-Netflix-TraceId) through all downstream service calls

3. Each service emits span data: { trace_id, span_id, parent_span_id, service, operation, duration_ms }

4. Spans published to Kafka trace topic

5. Mantis stream processor aggregates spans by trace_id

6. Completed traces written to Cassandra (partitioned by trace_id)

7. Edgar UI queries Cassandra to render waterfall visualization

Data Flow: Metrics

1. Application code instruments metrics using Atlas client library: registry.counter(“server.requests”).increment()

2. Atlas agent on each host aggregates metrics locally (1-minute windows)

3. Aggregated data pushed to Atlas backend cluster (in-memory time-series store)

4. Atlas backend serves real-time dashboard queries with sub-second latency

5. Older data (6h+) flushed to SSD; data older than 24h archived to S3 with downsampling

Deep Dive 1: Log Ingestion Pipeline

Sidecar Agent (per host):

• ·Fluentd with Netflix-specific plugins runs as a sidecar on every EC2 instance / container
• ·Local disk buffer (2 GB) prevents data loss if Kafka is temporarily unavailable
• ·Agent enriches logs with instance metadata: instance_id, availability_zone, asg_name, ami_id (from EC2 instance metadata service)
• ·Batching: send every 5 seconds or 1000 events, whichever comes first

Kafka cluster design:

• ·Dedicated Kafka cluster for logging (separate from application Kafka clusters)
• ·Topics: logs.{service} (one topic per service for isolation)
• ·Partitions: 100+ per topic for high-throughput services, 10 for lower-throughput services
• ·Retention: 72 hours (allows replay if indexer falls behind or needs reprocessing)
• ·Replication factor: 3 across AZs for durability

Indexer workers:

• ·Stateless Kafka consumers running on EC2 Auto Scaling Groups
• ·Bulk index to Elasticsearch: 5000 documents per request
• ·Back-pressure: if ES bulk request latency > 5s, consumer pauses (Kafka handles buffering)
• ·Idempotent writes using Kafka offset as ES document version

Sampling strategy (critical for managing 10M events/sec):

Dynamic sampling: If error rate for a service exceeds 2x its baseline, automatically increase INFO/DEBUG sampling to 100% for that service for 15 minutes. Implemented via Mantis stream processing rule that pushes sampling config updates to sidecar agents.

Deep Dive 2: Search Architecture + Elasticsearch Cluster Design

Elasticsearch cluster topology (across 3 AWS AZs):

Index Lifecycle Management (ILM):

Search optimization:

• ·Time-range filtering applied first (most queries during incidents include time range)
• ·Index-per-day-per-service pattern means searches only hit relevant indices
• ·Keyword fields for exact match (service, severity, host) vs. text fields for full-text (message)
• ·Use routing by trace_id so all logs for a trace land on the same shard (faster trace-based queries)
• ·ES query circuit breaker: reject queries that would scan > 500 shards (force users to add time/service filters)

Edgar integration (Netflix’s distributed tracing UI):

• ·Edgar queries both Cassandra (trace spans) and Elasticsearch (logs) by trace_id
• ·Unified view: trace waterfall + correlated log lines for each span
• ·Critical for incident debugging: "Why is this request slow?" -> trace shows which service, logs show why

Deep Dive 3: Alerting Engine + Chaos-Resilient Monitoring

Architecture: Dual evaluation pipeline for resilience

Alert types:

Deduplication and routing:

• ·Group alerts by rule_id + service + severity
• ·Configurable cooldown: alert at most once per 15 minutes per group
• ·Escalation chain: Warning -> Slack #alerts-{team} -> if unacknowledged for 10 min -> PagerDuty -> on-call engineer
• ·Auto-resolve: if alert condition clears for 5 minutes, send resolution notification

Chaos resilience (monitoring must survive Chaos Monkey):

• ·Alert Evaluator runs across 3 AZs with no single points of failure
• ·Kafka consumer group rebalancing handles evaluator instance termination
• ·Atlas backend uses consistent hashing — loss of one node redistributes time-series to surviving nodes
• ·Notification Service has fallback path: if PagerDuty API is down, fall back to direct SMS via AWS SNS
• ·"Watchdog" alert: a synthetic heartbeat metric that should always fire. If the watchdog alert stops firing, the alerting system itself is broken — monitored by an independent, minimal system running in a separate AWS account

Bottlenecks

1. Elasticsearch indexing throughput: 10M events/sec at 500B each = 5 GB/s ingestion

• ·Mitigation: Sampling reduces actual indexing to ~1-2 GB/s; bulk indexing with 5000-doc batches; index-per-service-per-day limits shard count per index

1. Kafka throughput for log topics: Single Kafka cluster handling 5 GB/s of log data

• ·Mitigation: Dedicated logging Kafka cluster separate from application Kafka; 100+ partitions for high-throughput services; compression (lz4) reduces network bandwidth

1. Trace storage write amplification: 5M spans/sec to Cassandra

• ·Mitigation: Trace sampling at 10% of requests (keep 100% for errored requests); Cassandra's write-optimized LSM architecture handles high write throughput; TTL auto-deletes old traces

1. Hot service log volume skew: A few services (e.g., Zuul, API gateway) produce disproportionate log volume

• ·Mitigation: Per-service Kafka topics prevent noisy-neighbor; per-service ES indices allow independent scaling; more aggressive sampling for high-volume services

1. Alert storm during outages: A cascading failure triggers thousands of alerts simultaneously

• ·Mitigation: Alert grouping and deduplication; top-level "service down" alerts suppress downstream alerts; rate-limiting notification delivery

Key Tradeoffs

Monitoring & Alerting (monitor the monitoring system)

• ·Kafka consumer lag: Alert if log indexer falls behind > 5 minutes (log freshness SLA violated)
• ·ES cluster health: RED = potential data loss, immediate page; YELLOW = reduced redundancy, warning
• ·ES indexing rate: Sudden drop = agent failure or network partition; sudden spike = incident in progress
• ·ES search latency p99: Alert if > 10 seconds (dashboard experience degraded)
• ·Atlas publish errors: Alert if any service fails to publish metrics (silent monitoring gap)
• ·Alert firing rate: Sudden spike = real incident; sustained high = alert fatigue, review thresholds
• ·Trace completeness: Percentage of requests with full trace chain — alert if < 90% (indicates broken propagation)
• ·Watchdog alert: Synthetic always-firing alert; if it stops, the alerting pipeline itself is broken

Failure Modes

• ·Kafka broker failure: Replication factor 3 ensures no data loss; producers retry to other brokers; consumer group rebalances automatically
• ·Elasticsearch node failure: Replica shards on surviving nodes serve reads; cluster auto-recovers when node returns; shard allocation throttling prevents recovery storm
• ·Atlas backend node failure: Consistent hashing redistributes time-series; in-memory data for failed node lost (acceptable: metrics are re-published every minute)
• ·Cassandra node failure: Replication factor 3 with LOCAL_QUORUM reads/writes ensures availability; hinted handoff repairs data when node returns
• ·Full AZ outage: All components (Kafka, ES, Atlas, Cassandra) deployed across 3 AZs; survives loss of any one AZ with degraded capacity; auto-scaling replaces capacity in surviving AZs
• ·Alerting pipeline failure: Watchdog detection triggers fallback alerting from independent system in separate AWS account; Kafka retains unprocessed events for replay after recovery
• ·Chaos Monkey kills monitoring instance: All monitoring components are stateless or replicated; instance termination triggers auto-scaling group replacement; no single instance is critical