Design a Metrics Monitoring and Alerting System

A metrics monitoring system collects numeric measurements (CPU usage, request count, latency) from infrastructure and applications, stores them as time series, shows them on dashboards, and sends alerts when values cross thresholds. This article designs an internal system for a large fleet, similar in scope to Prometheus-based or Datadog-style platforms. Background on metric types and alerting practice is in observability: logs, metrics and traces.

Requirements

Functional

• Collect metrics from servers, containers, databases, and applications.
• Store metrics with labels (tags) such as host, region, service.
• Query metrics over time ranges with aggregation (sum, average, percentiles, rate) and grouping by label.
• Visualize on dashboards.
• Define alert rules and notify through email, chat, or paging systems.

Non-functional

• Scale. about 1 million data points per second (60 million per minute) across about 10 million distinct series (see estimation).
• Write availability. Metric ingestion must keep working during incidents, precisely when load is unusual. Loss of a few data points is acceptable; loss of an hour is not.
• Query latency. Dashboards load in seconds. Alert evaluation completes within its interval.
• Reliability of alerting. The alerting path must be more reliable than the systems it watches and must not depend on them.
• Retention. Full resolution for days to weeks, downsampled data for a year or more.
• Data characteristics. Write-heavy, append-only, recent data is queried most, old data is never updated.

Estimation

Assumptions: 1,000 server pools with 100 servers each, so 100,000 servers; 100 metrics per server; a data point every 10 seconds.

• Series: 100,000 × 100 = 10,000,000 time series.
• Write rate: 10,000,000 / 10 s = 1,000,000 data points per second (60 million per minute).
• Data point size: a timestamp and a value. Uncompressed about 16 bytes; compressed time-series formats reach roughly 1-2 bytes per point for regular data. Uncompressed daily volume: 1,000,000 × 86,400 × 16 B ≈ 1.4 TB per day. With around 1.5 bytes per point, about 130 GB per day.
• Retention example: 7 days at full resolution ≈ 1 TB compressed; if 1-minute rollups were kept for 1 year, one aggregate would take 10,000,000 series × 525,600 points × ~2 bytes ≈ 10 TB, and several aggregates per point (min, max, sum, count) multiply that. 1-hour rollups take 8,760 points per series per year: 10,000,000 × 8,760 × ~2 bytes ≈ 175 GB per aggregate. This is why the retention table below keeps 1-minute data for 30 days and 1-hour data for a year.
• Read rate: dashboards and alert rules produce far fewer operations than writes, but each query may scan thousands of series.
• Ingest bandwidth: 1,000,000 points/s × about 100 bytes on the wire (name and labels are amortized by batching, uncompressed JSON is far larger) ≈ 100 MB/s ≈ 0.8 Gbit/s. Sending compressed batches of 500 to 1,000 points cuts this several times.
• Alert load: 10,000 rules every 30 s ≈ 330 queries/s, each touching tens to thousands of series.
• 1 MB of RAM per 1,000 active series is a rough planning figure for a Prometheus-style head block (about 1 to 3 KB per series including index), so 10,000,000 series need about 10 to 30 GB of memory across the cluster, before replication.

Implications: 1 million points/s exceeds one node (a well-tuned single TSDB node handles on the order of a few hundred thousand to a million points/s), so shard ingestion at least 10 ways (about 100,000 points/s and 1,000,000 series per shard) with 2 to 3 replicas. Compression is what makes retention affordable (130 GB/day versus 1.4 TB/day), and downsampling is required beyond weeks. Series count, not point rate, bounds memory, so cardinality control is a first-class requirement.

API design

Write (agents push, or the system scrapes; see deep dive):

POST /v1/write
{ "series": [ { "name": "cpu_usage", "labels": {"host": "web-17", "region": "eu"}, "points": [[1780000000, 0.42]] } ] }

Query:

GET /v1/query_range?query=avg(cpu_usage{service="api"}) by (region)&start=&end=&step=60

Alert rules (declarative, stored in a rules service):

alert: HighErrorRate
expr: sum(rate(http_errors[5m])) / sum(rate(http_requests[5m])) > 0.05
for: 10m
notify: pager-team-api

The for clause requires the condition to hold continuously for a duration before firing, which suppresses brief spikes.

Data model

A time series is identified by a metric name plus a set of labels. Each series is a sequence of (timestamp, value) pairs.

cpu_usage{host="web-17", region="eu"}  ->  (t1, 0.42), (t2, 0.45), ...

Storage requirements: very high write throughput, time-ordered reads by series and range, compact storage, and cheap deletion of old data. A general-purpose relational database, or a document store with per-point rows, wastes space and write capacity. A time-series database (TSDB) is designed for this:

• Data is written to an in-memory buffer per series, with a write-ahead log for durability, then flushed to immutable time-partitioned blocks on disk (Prometheus, InfluxDB and similar systems work this way).
• Timestamps are compressed with delta-of-delta encoding: regular intervals produce deltas of deltas equal to zero, which encode in one bit or so.
• Values are compressed by XOR with the previous value (Gorilla-style encoding): similar consecutive floats share leading and trailing bits.
• Old blocks are deleted whole once past retention, which is cheaper than deleting rows.
• An inverted index maps each label=value pair to the list of series ids that have it, so a query such as service="api" finds matching series quickly (see inverted index).

Options: operate an existing TSDB (Prometheus with long-term storage such as Thanos or Mimir, VictoriaMetrics, InfluxDB), or build on a wide-column store. Building a TSDB from scratch is rarely justified.

High-level design

flowchart LR
    T["Targets: servers, services, containers"] --> COL["Collectors"]
    COL --> Q[("Queue")]
    Q --> ING["Ingesters"]
    ING --> TSDB[("Time-series storage")]
    TSDB --> QS["Query service"]
    QS --> CACHE[("Query cache")]
    QS --> DASH["Dashboards"]
    QS --> AE["Alert evaluator"]
    AE --> AM["Alert manager"]
    AM --> N["Email / chat / paging"]
    RULES[("Alert rules")] --> AE

Component responsibilities:

• Collectors. Gather metrics from targets and forward them.
• Queue (Kafka). Buffers writes so ingesters and storage can be slow or restarted without dropping data, and decouples collection from storage. It also lets the alerting path consume the same data (see message queues).
• Ingesters. Consume from the queue, batch, and write to the TSDB, sharded by series.
• Time-series storage. Recent data in memory and local disk, older blocks in object storage, downsampled data for long ranges.
• Query service. Parses the query language, resolves labels via the index, reads the relevant series, aggregates, and returns results. Caches recent results.
• Alert evaluator. Periodically runs each rule's query and tracks state.
• Alert manager. Deduplicates, groups, silences, routes, and rate limits notifications.

Deep dive

1. Collection: pull versus push

Large systems use both. A common design has collectors scale horizontally by assigning each collector a subset of targets using consistent hashing over the target list, and a coordination service (etcd or ZooKeeper) to reassign targets when a collector dies.

2. Storage sharding, replication, and the write path

The system shards series by hashing (metric name, labels), so all points for one series go to one shard and are compressed together. Each shard is replicated (typically 2 or 3 copies) for durability and to survive node loss. The write path is: collector to queue, ingester to WAL and memory buffer, periodic flush to a block, block uploaded to object storage.

Choices and numbers. Shard key: the full series identity hash, giving an even spread and single-shard compression. The alternative, sharding by time, sends all current writes to one shard (a hot shard), and sharding by metric name alone concentrates high-volume metrics such as http_requests. Sharding by service or tenant keeps a tenant's queries local but lets one noisy tenant overload a shard; use it with per-tenant quotas. Consistency: replicate asynchronously or write to 2 of 3 replicas (quorum) and have queries merge and deduplicate results from replicas; an immutable append-only stream needs no conflict resolution, only deduplication by (series, timestamp). Queue: Kafka partitioned by series hash gives per-series ordering; ingesters use at-least-once consumption with idempotent writes (the same timestamp and value is a no-op), and a retention of 1 to 3 days lets ingesters replay after an outage. Choose a queue over direct writes because an ingester restart otherwise drops 1 million points/s of data; choose direct push-to-ingester for the alert path where seconds of extra latency matter. The WAL is fsynced in batches (for example every 100 ms to 1 s), trading a small loss window on a machine crash for write throughput, with replicas covering it.

To keep queries efficient, storage partitions data in time blocks (for example, 2 hours). A query for a time range opens only the blocks that overlap it.

3. Downsampling and retention

Full-resolution data is expensive to keep and unnecessary for old ranges. Policy example:

A background job rolls raw points into aggregates. For each interval, it stores several aggregates (min, max, sum, count), because average alone cannot be recombined correctly, and max is needed to see spikes. The query service chooses the resolution based on the requested range and step: a 30-day dashboard reads the 1-minute or 1-hour data rather than raw points.

The alternative is to keep everything raw in cheaper object storage and compute at query time; that costs about 10 to 60 times more storage and query time for old ranges, so rollups win. The price is lost detail: a 1-hour rollup cannot show a 30-second spike except through max.

Percentiles cannot be recomputed from downsampled percentiles. Store histograms (counts per latency bucket) as the raw metric, so percentiles can be estimated after aggregation across hosts.

4. Alert evaluation

The alert evaluator loads rules and runs each rule's query every evaluation interval (for example, 30 seconds). Each rule instance is a state machine:

stateDiagram-v2
    [*] --> Inactive
    Inactive --> Pending: condition true
    Pending --> Inactive: condition false
    Pending --> Firing: true for the full for-duration
    Firing --> Inactive: condition false, resolved

Design points:

• Scale. Shard rules across evaluator instances by rule id. Give each rule a single owner at a time to avoid duplicate alerts, or allow duplicates and deduplicate downstream.
• The alert manager groups alerts (100 hosts down in one rack become one notification), deduplicates repeats, supports silences during maintenance, and routes by label to the right team.
• Reliability. Run the alert path in more than one zone, and add a dead man's switch: an always-firing alert whose absence at an external service indicates that the monitoring system itself is broken. The monitoring system must not be hosted only on the infrastructure it monitors.
• Missing data. Decide per rule whether no data means healthy or alerting; a crashed target produces no data.

5. Cardinality

Cardinality is the number of distinct series. It is multiplicative: a metric with 5 endpoints × 4 status codes × 100 hosts is 2,000 series, and adding a user_id label with 1 million values makes it 2 billion. Each unique label combination creates a series, so a label with unbounded values (user id, request id, full URL) multiplies series count and can exhaust memory and index space. Mitigations: reject or drop high-cardinality labels at ingestion, enforce per-tenant series limits, and use logs or traces for per-request detail instead of metrics.

Bottlenecks and scaling

• Write throughput. Shard by series hash; batch writes; use the queue to absorb spikes.
• Query cost. Queries that fan out over many series are the main load. Use the query cache, pre-compute expensive expressions as recording rules, limit query time ranges and series count, and query downsampled data for long ranges.
• Index memory. Bound cardinality and shard the index with the data.
• Hot shard. A metric with many labels can concentrate on one node if sharding uses only the metric name; hash the full label set.
• Overload during incidents. Apply per-tenant rate limits and shed low-priority queries before dropping writes.
• Multi-region. Run a collector and short-term store per region and a global query layer that fans out to regions; this keeps a region's alerting independent of cross-region links.

Failure modes

The overriding rule is that a monitoring system degrades by losing resolution or dashboards first, never by silently losing alerts. At 10x (10 million points/s, 100 million series, 1.3 TB/day compressed): a single head index per shard no longer fits, so shard 100 or more ways, use a horizontally scaled TSDB (Mimir or VictoriaMetrics style), push recording rules to pre-aggregate at the edge, and move toward tiered storage in object storage with 2-hour blocks compacted to 24 hours. Queries must be sharded and parallelized by time range, and cardinality quotas become mandatory.

Trade-offs and follow-up questions

• Why a queue in front of storage? Absorbs bursts and outages of ingesters, at the cost of extra latency (seconds) and another system to operate. Alert latency-sensitive paths can read directly from collectors when needed.
• Availability versus consistency. Monitoring favors availability of writes and reads over strict consistency; a missing or slightly stale point is acceptable.
• Push versus pull is decided by discovery, firewall, and short-lived-job requirements, as in the table above.
• Alert fatigue. Alert on symptoms (error rate, latency) tied to user impact rather than every cause; use for durations and grouping.
• How is the monitoring system itself monitored? A separate, minimal instance and a dead man's switch.
• What if a region loses connectivity to the global layer? Local collectors, storage, and alerting keep working; the global view is partial and labeled so.
• What if a deploy creates 1 million new series at once (pod churn)? Series churn hurts the index even if the active count is stable; set limits on new series per minute and expire inactive series after a stale timeout (for example 5 minutes without data).
• What if clocks differ across hosts? Stamp at the collector or reject points more than a few minutes in the future or past; out-of-order writes within a window (for example 1 hour) can be accepted if the TSDB supports it.
• What if a user runs a query over 1 year at 10-second step? Reject or force the step up to what the range allows (about 1,000 to 11,000 points per line is a sensible cap) and read rollups.
• Would you use a relational database instead? At 1 million inserts/s with per-row index overhead, no; the TSDB stores about 1.5 bytes per point versus tens of bytes per row.
• Follow-ups: anomaly detection over seasonal data; multi-tenancy and quotas; exemplars linking a metric spike to traces.