Design a Distributed Metrics Monitoring & Alerting System

1. Problem Statement & Scope Clarification

System Mission

Design a planetary-scale, multi-tenant distributed Metrics Monitoring, Time-Series Storage, and Alerting Platform (equivalent to Prometheus/Thanos, Datadog, or Amazon CloudWatch) capable of ingesting tens of millions of metric points per second, executing real-time alert rules with sub-minute evaluation cycles, and delivering sub-second analytical dashboard queries over multi-year time horizons.

Functional Requirements

1. High-Throughput Metric Ingestion (PushMetric / BatchIngest): Continuously ingest structured time-series data points comprising metric name, 64-bit millisecond timestamp, 64-bit IEEE 754 float value, and multi-dimensional string key-value tag labels (e.g., cpu.util{env="prod", service="payment", region="us-east-1"}).
2. Multi-Dimensional Tag Filtering & Analytical Queries: Support flexible label slicing and time-series aggregation functions (avg, sum, rate, p95, p99) executed over custom sliding time windows.
3. Real-Time Sliding-Window Alert Evaluation: Continuously evaluate user-defined alert conditions (e.g., rate(http_requests_5xx[5m]) > 0.05) every 15–30 seconds, dispatching deduplicated notifications to Amazon SNS, PagerDuty, and Slack.
4. Automated Downsampling & Tiered Retention: Automatically roll up 1-second raw metrics into 1-minute and 1-hour summaries to optimize storage costs over a 5-year retention horizon.
5. High-Cardinality Defense: Detect and strip unbounded dynamic tag dimensions (e.g., user_id or order_id injected into metric tags) before they exhaust inverted index memory.

Non-Functional Requirements (SLAs & SLOs)

• Ingestion Throughput: $10,000,000\text{ metric points/sec}$ sustained average ($10\text{M points/sec}$), with peak ingestion headroom of $30,000,000\text{ points/sec}$ ($30\text{M points/sec}$).
• Query Latency (P95):
  • Dashboard Queries on Recent Data ($< 2\text{ hours}$): $P95 < 50\text{ ms}$.
  • Range Queries on Historical Data ($30\text{ days}$): $P95 < 500\text{ ms}$.
• High Availability: $99.999\%$ ("five nines") uptime SLA for ingestion and alerting. Monitoring systems must survive the very datacenter failures they observe.
• Compression Efficiency: Achieve $\ge 90\%$ storage reduction versus raw 16-byte samples using Facebook's Gorilla Time-Series Compression Algorithm (Delta-of-deltas timestamp encoding + XOR float64 compression); the paper's production figure is $1.37\text{ bytes/point}$, i.e. $91.4\%$.
• Data Durability: $99.999999999\%$ (11 9s) durability across tiered storage backends.

---

2. Capacity & Scale Estimation (Back-of-the-Envelope Math)

Ingestion Scale & Network Bandwidth

• Sustained Ingest Rate: $10,000,000\text{ data points/sec}$ ($10\text{M points/sec}$).
• Peak Ingest Rate ($3\times$ burst headroom): $30,000,000\text{ points/sec}$ ($30\text{M points/sec}$).
• Raw Metric Point Size:
  • metric_name pointer ($8\text{ bytes}$), timestamp ($8\text{ bytes}$ int64), value ($8\text{ bytes}$ float64), tag metadata pointers ($40\text{ bytes}$) $\approx \mathbf{64\text{ bytes/point}}$.
• Peak Raw Network Ingress Bandwidth: $\text{Peak Ingress} = 30 \times 10^6\text{ pts/sec} \times 64\text{ bytes} = 1.92\text{ GB/sec} \approx \mathbf{15.36\text{ Gbps}}$ Distributed across internal AWS Network Load Balancers (NLB) and an ECS Ingestion Gateway fleet.

Gorilla Compression Mathematics & Storage Derivations

Standard uncompressed time-series storage requires $16\text{ bytes}$ per point ($8\text{B timestamp} + 8\text{B float value}$). The Gorilla compression algorithm compresses this drastically:

1. Timestamp Compression (Delta-of-Deltas):
   • For regular scrape intervals (e.g., every 10 seconds), $D = t_i - t_{i-1} = 10\text{s}$.
   • Delta-of-deltas: $D' = (t_i - t_{i-1}) - (t_{i-1} - t_{i-2}) = 10 - 10 = 0$.
   • A delta-of-deltas of $0$ encodes as a single bit 0.
   • Over $96\%$ of all timestamps in production compress into a single 0 bit.
2. Float64 Value Compression (XOR Encoding):
   • Successive floating-point measurements (e.g., CPU utilization fluctuating between $42.15\%$ and $42.18\%$) share identical IEEE 754 sign and exponent bits.
   • XORing current and previous floats yields many leading and trailing zero bits.
   • If value is unchanged ($V_i \oplus V_{i-1} = 0$), encode as a single bit 0 ($59\%$ of metrics in production).
   • If changed, encode control bits and only the meaningful XOR bits.

• Empirical Gorilla Compression Factor: $\text{Average Compressed Size} \approx \mathbf{1.37\text{ bytes/point}} \quad (\text{vs } 16\text{ bytes raw} \implies 91.4\% \text{ reduction})$
• Daily Compressed Ingestion Footprint: $\text{Daily Volume} = 10 \times 10^6\text{ pts/sec} \times 86,400\text{ sec/day} \times 1.37\text{ bytes} \approx 1.183 \times 10^{12}\text{ bytes} \approx \mathbf{1.18\text{ TB/day}}$

Tiered Storage Footprint (Multi-Year Horizon)

1. Hot Tier (In-Memory / NVMe on Amazon EKS - 2 Hours): $\text{Hot RAM Volume} = 10 \times 10^6\text{ pts/sec} \times 7,200\text{ sec} \times 1.37\text{ bytes} \approx \mathbf{98.64\text{ GB RAM}}$ Series cardinality (the number people forget): at a 10-second scrape interval, $10\text{M points/sec}$ means $10 \times 10^6 \times 10 = 100\text{M active time series}$. Each series carries a Gorilla block header, a Roaring Bitmap posting entry per label, and a label-set string (roughly $200\text{ bytes}$ in total): $\text{Series Index RAM} = 100 \times 10^6 \times 200\text{ bytes} \approx \mathbf{20\text{ GB RAM}}$ Sample data ($98.6\text{ GB}$) plus index ($20\text{ GB}$) with a $2\times$ safety margin needs $\approx 240\text{ GB}$ across the hot tier. Because series are hash-sharded (not replicated) across nodes, a 3-node r6i.4xlarge cluster ($128\text{ GB RAM}$ each, $384\text{ GB}$ total) holds it with headroom; three r6i.2xlarge nodes ($192\text{ GB}$ total) would not survive a single node loss. Durability during a node loss comes from the NVMe WAL replay plus the Kinesis retention window, not from RAM replication.
2. Warm Tier (Amazon Timestream / ClickHouse Columnar - 30 Days): $\text{Warm Volume} = 1.18\text{ TB/day} \times 30\text{ days} \approx \mathbf{35.4\text{ TB}}$
3. Cold Tier (Downsampled Amazon S3 Parquet Lakehouse - 5 Years):
   • Raw 1-second metrics are progressively downsampled:
     • 1-minute rollups ($60\times$ volume reduction) for days 31–180.
     • 1-hour rollups ($3,600\times$ volume reduction) for days 181–1,825 (5 years). $\text{Annual Cold Volume} \approx \frac{1.18\text{ TB/day} \times 365\text{ days}}{60} \approx \mathbf{7.18\text{ TB/year}}$ $\text{5-Year Cold Storage} = 7.18\text{ TB/year} \times 5 \approx \mathbf{35.9\text{ TB}}$

---

3. High-Level Architecture & AWS Component Mapping

Synthesizing vector architecture diagram...

Data Flow Walkthrough

1. Telemetry Ingress: Agents collect metrics locally and push batched Protobuf payloads via persistent gRPC HTTP/2 connections to an internal AWS NLB.
2. Consistent Partitioning: Ingest gateways inspect metric names and tags, compute a 64-bit MurmurHash3 shard key, and write records to Amazon Kinesis Data Streams.
3. In-Memory Gorilla Ingestion: Stateful EKS worker pods consume from assigned Kinesis shards. Each series writes to an active 2-Hour Gorilla Block in RAM while appending raw bytes to an NVMe Write-Ahead Log (WAL) for crash recovery.
4. Roaring Bitmap Indexing: Tags are indexed into an in-memory inverted index utilizing Roaring Bitmaps. Queries filtering on multiple labels execute bitwise AND intersections in microseconds.
5. Real-Time Alert Evaluation: The Alert Engine queries the Hot TSDB every 15 seconds. If an alert condition breaches its threshold across consecutive windows, the engine checks the Redis debounce ring (SET NX EX 900). If new, a critical incident payload is published to Amazon SNS.

Concrete Step-by-Step Request Walkthrough: Ingesting & Querying a Metric Point

---

4. API Interface Design & Wire Protocol

1. gRPC Ingestion Protocol (metrics_service.proto)

protobufCopy
syntax = "proto3";

package hispeeddesign.metrics.v1;

service MetricsIngestService {
  rpc PushMetrics (PushMetricsRequest) returns (PushMetricsResponse);
}

message PushMetricsRequest {
  string tenant_id = 1;
  repeated TimeSeries points = 2;
}

message TimeSeries {
  string metric_name = 1;
  map<string, string> labels = 2;
  repeated MetricSample samples = 3;
}

message MetricSample {
  int64 timestamp_ms = 1;
  double value = 2;
}

message PushMetricsResponse {
  uint32 points_ingested = 1;
  uint32 points_rejected = 2;
  string error_message = 3;
}

2. Analytical Range Query REST API

httpCopy
POST /v1/query_range HTTP/1.1
Host: tsdb-query.production.aws.internal
Content-Type: application/json

{
  "query": "avg(rate(http_requests_total{env=\"prod\", status=~\"5..\"}[5m])) by (service)",
  "start_epoch_sec": 1767222000,
  "end_epoch_sec": 1767225600,
  "step_seconds": 60
}

Response: 200 OK
Content-Type: application/json

{
  "status": "success",
  "data": {
    "resultType": "matrix",
    "result": [
      {
        "metric": { "service": "payment-api" },
        "values": [
          [ 1767222000, "0.012" ],
          [ 1767222060, "0.015" ],
          [ 1767222120, "0.084" ]
        ]
      }
    ]
  }
}

---

5. Data Models & Storage Architecture

1. Gorilla Block Physical Memory Layout

Time-series points are organized into discrete 2-Hour In-Memory Slabs:

Each slab starts with a fixed header:

After the header comes the compressed bitstream. Each point writes a timestamp code and then a value code; the first bits of each code (the prefix) say how many bits follow:

2. Inverted Index Schema: Roaring Bitmaps

Labels are mapped to monotonically increasing 32-bit Series IDs:

3. DynamoDB Alert Rules Table (AlertRulesCatalog)

• Partition Key (PK): TENANT#<tenant_id>
• Sort Key (SK): RULE#<rule_id>
• Attributes: query_expression, evaluation_interval_sec, for_duration_sec, severity, notification_sns_topic_arn, status.

---