Design a time series database for distributed environments.

Question

Anonymous · Accepted Answer

A distributed time series database (TSDB) must handle continuous, high-volume
append operations, heavy data compression, efficient time-range scans across
multidimensional labels, and automated retention management.

1. Requirements & System Characteristics

Functional Requirements

* Ingestion: High-throughput ingestion of time series metrics ($>1	ext{M}$ to
   $10	ext{M}+$ data points/sec).

* Querying: Low-latency aggregation and filtering (e.g., avg, rate, sum, p99)
   across multiple tag dimensions over specified time windows.

* Downsampling & Retention: Automated rollups (e.g., 10s raw $	o$ 5m
   downsampled) and eviction policies based on TTL.

Non-Functional Requirements

* Write-Heavy Workload: $>90\%$ writes, with sequential time order and
   occasional late-arriving (out-of-order) data.

* Availability & Partition Tolerance: Target AP semantics under the CAP theorem
   with eventual consistency for querying recent data points.

* Storage Efficiency: Heavy compression to minimize raw disk footprints
   (target: 1–2 bytes per data point).

2. Data Model & Architecture Overview

A single time series point is modeled as:

$$	ext{Metric Name} + \{	ext{Key}_i = 	ext{Value}_i\} + 	ext{Timestamp} +
	ext{Value}$$

* Series ID: A 64-bit fingerprint hash of the metric name and sorted tag
   key-value pairs:
   
   $$	ext{SeriesID} = 	ext{Hash}(	ext{metric\_name}, 	ext{tag}_1 = v_1,
   \dots, 	ext{tag}_k = v_k)$$

* Data Point: (Timestamp, Value) where Timestamp is an epoch integer and Value
   is a 64-bit float/integer.

+-------------------------+

|  Clients / Agent Nodes  |

+------------+------------+

|

v

+-------------------------+

| Load Balancer / Ingress |

+------------+------------+

|

+----------------------+----------------------+

|                                             |

v                                             v

+------------------+                          +------------------+

| Ingestion Router |                          |   Query Engine   |

+--------+---------+                          +--------+---------+

| (Consistent Hashing)                        | (Scatter-Gather)

v                                             v

+------------------+                          +------------------+

| Storage Node A   |<========================>| Storage Node B   |

| - Inverted Index |                          | - Inverted Index |

| - MemTable / WAL |                          | - MemTable / WAL |

| - Immutable Chunks                          | - Immutable Chunks

+------------------+                          +------------------+

Component Roles

* Ingestion Router: Stateless proxy receiving writes. It computes the SeriesID
   and uses consistent hashing to route points to primary and replica storage
   nodes.

* Query Engine: Parses PromQL/SQL-like range queries, executes distributed
   scatter-gather requests across relevant storage nodes, and computes aggregate
   transformations.

* Storage Nodes: Manage both the metadata inverted index and the
   on-disk/in-memory time series chunks.

* Consensus / Cluster Coordinator: Lightweight coordination layer (e.g., etcd,
   Raft) maintaining cluster membership, shard maps, and partition leases.

3. Storage Engine: In-Memory & On-Disk Layout

To satisfy high append throughput and range queries, a Time-Structured
Merge-Tree (TSM) or modified LSM-tree structure is standard.

Incoming Data Point

|

+------+------+

|             |

v             v

+-------+     +-------+

|  WAL  |     | Mem-  |

| (Disk)|     | Table |

+-------+     +---+---+

| (Flush every 2h window)

v

+-----------------+

| Immutable Chunk |

| (Columnar File) |

+--------+--------+

| (Compaction & Downsampling)

v

+-----------------+

| Cold Chunk / S3 |

+-----------------+

1. Write Ahead Log (WAL) & MemTable

* WAL: Point appends are appended sequentially to disk to provide crash
   recovery and durability.

* MemTable: An in-memory cache sorted by (SeriesID, Timestamp).

* Flushing: Once the MemTable reaches a threshold or a fixed time boundary
   (e.g., 2 hours), it flushes to disk as an immutable, chunked columnar file.

2. Time-Bucket Compression Algorithms

TSDB data compresses significantly because timestamps are monotonically
increasing and values change gradually:

* Timestamp Encoding (Gorilla Double-Delta):

* Store the base timestamp $t_0$.

* For subsequent timestamps, store the delta of deltas:
     
     $$D = (t_i - t_{i-1}) - (t_{i-1} - t_{i-2})$$

* If $D = 0$, emit a single bit 0. If $D$ is within $[-63, 64]$, emit 10
     followed by 7 bits, scaling up for larger jitter.

* Float Value Encoding (Gorilla XOR):

* XOR the current floating-point value with the previous one:
     
     $$\Delta v = v_i \oplus v_{i-1}$$

* If $\Delta v = 0$, store bit 0.

* If $\Delta v 
eq 0$, store bit 1, followed by control bits representing
     the count of leading/trailing zeros, and the significant XOR bits.

3. Inverted Index for Tags

To resolve queries like cpu.usage{host="server-1", env="prod"} quickly without
full scans:

* Maintain a reverse mapping:
   
   $$	ext{Key:Value} \longrightarrow 	ext{Roaring Bitmap of SeriesIDs}$$

* Intersections and unions of tag sets are performed via bitmap bitwise AND/OR
   operations, identifying the target SeriesIDs in microseconds.

4. Distributed Sharding & Query Execution

Sharding Strategy: 2D Composite Partitioning

To prevent both write hotspots on recent data and wide scatter-gather latency,
TSDB partitions data across two dimensions:

Shard Dimension

Implementation

Trade-off / Benefit

Series Hash (Space)

Hash(SeriesID) mod $N$ or via Consistent Hash Ring

Evenly distributes metrics across storage nodes; prevents single-device
hotspots.

Time Bucketing (Time)

Chunks divided into deterministic windows (e.g., 2 hours or 1 day)

Isolates write workloads to active shards and makes drop/eviction operations
$O(1)$.

Time Windows   Shard 1 (Hash 00-3F)   Shard 2 (Hash 40-7F)   Shard 3 (Hash
80-FF)

[10:00 - 12:00]       Node A                 Node B                 Node C

[12:00 - 14:00]       Node B                 Node C                 Node A

Query Execution (Scatter-Gather)

1. Routing: The Query Engine parses the query range $[T_{	ext{start}},
    T_{	ext{end}}]$ and tag filters.

2. Pruning: Resolves target time buckets to prune shards outside the requested
    time window.

3. Index Evaluation: Relevant storage nodes use their local inverted index to
    resolve matching SeriesIDs.

4. Chunk Retrieval: Storage nodes decompress only the relevant time blocks for
    those series IDs.

5. Aggregation & Merge: Nodes compute partial aggregations locally; the Query
    Engine merges and streams final sorted series back to the client.

5. Replication, Consistency & Lifecycle Management

Replication & Quorum

* Replication Factor (RF = 3): Writes are replicated across 3 nodes on the hash
   ring.

* Tunable Consistency:

* Writes use $W=2$ for fast acknowledgments with high availability.

* Reads use $R=2$ with read-repair to resolve discrepancies between replicas.

* Sloppy Quorum & Hinted Handoff: If a replica node is unreachable during
   ingestion, hints are temporarily written to a neighboring node and replayed
   when the replica recovers.

Late-Arriving & Out-of-Order Data

* A bounded out-of-order grace period (e.g., 1 hour) is supported:

* Points within the grace window insert directly into the current/recent
     MemTable or trigger a merge with the most recent immutable segment.

* Points beyond the grace window are routed directly to an asynchronous
     cold-path compaction pipeline to prevent degrading write cache performance.

Tiered Storage & Retention

* Hot Tier (NVMe / RAM): Keeps the last 24–48 hours of raw data for
   high-frequency dashboard queries and alerting.

* Warm Tier (Standard SSD / Compacted Disk): Keeps downsampled series (e.g.,
   5-minute rollups) for medium-range queries (30–90 days).

* Cold Tier (Object Storage like S3): Historical data past 90 days downsampled
   to 1-hour rollups and stored as Parquet/columnar files.

* Data Dropping: Expired data is dropped simply by unlinking entire time-window
   segment directories, avoiding per-record DELETE overhead.

Design a time series database for distributed environments.

Practice this question with AI

Go Premium

Community Answers

Unlock Full Access

Practice More Questions