Architecture of Pinterest’s Time Series Database: Goku

High-scale engineering organizations depend heavily on telemetry to monitor infrastructure, detect service anomalies, safeguard against attacks, and track core business metrics in real time. At Pinterest, infrastructure telemetry generates millions of metric data points every second.

Historically, Pinterest relied on OpenTSDB to ingest, store, and serve these time series metrics. However, as scale exploded, OpenTSDB failed to meet Pinterest’s latency, cost, and reliability requirements. To solve this, Pinterest built their own distributed, in-memory time series database named Goku.

---

1. Why OpenTSDB Failed at Scale

OpenTSDB is an open-source time series database built on top of Apache HBase, which in turn runs on Hadoop (HDFS). While HBase provides horizontally scalable storage, its architecture introduced severe performance and operational bottlenecks when subjected to Pinterest’s write- and scan-heavy workloads:

1. Severe Java Garbage Collection (GC) Pauses: Massive point-in-time ingest throughput triggered aggressive memory allocations in HBase, leading to prolonged Stop-the-World GC pauses and frequent cluster crashes.
2. Inefficient Disk Scans and Random I/O: OpenTSDB relies heavily on disk reads. Metrics are split into row-key ranges and bucketed in HBase tables, resulting in excessive random disk I/O operations across HDFS blocks during wide-range queries.
3. Network-Heavy Scatter-Gather Aggregations: OpenTSDB query nodes read raw, unaggregated time series points from all relevant HBase RegionServers over the network before performing aggregations locally on the coordinator. This saturated cluster network bandwidth and caused coordinator nodes to run out of memory.
4. Inefficient Serialization: OpenTSDB utilized JSON over HTTP for data ingestion and query responses, adding high serialization overhead and bloated payload sizes.

To overcome these issues, Pinterest designed Goku with a simple mandate: eliminate disk I/O on the hot query path, leverage extreme data compression, and push aggregation computation directly to the storage tier.

---

2. Time Series Data Model & Query Semantics

Before diving into Goku’s internal architecture, it is essential to understand the underlying data model it supports.

Metric Structure

A time series data point represents a value recorded at a discrete timestamp. In Goku, a data point consists of:

• Metric Name: A hierarchical, dot-delimited string representing the property being tracked (e.g., tc.prog.stat.cpu.total or service.auth.requests.count).
• Tags (Dimensions): Key-value pairs used for contextual slicing and filtering (e.g., host=web-102.us-east-1, service=auth, env=production).
• Value & Timestamp: A tuple of (timestamp, numeric_value) representing the observed observation.

Key   = Metric Name + Set of Key-Value Tags
Point = (Timestamp, Numeric Value)

Filtering and Wildcards

Tags act as index dimensions. Queries can filter metrics using:

• Exact Match: service=auth
• Wildcard Match: host=web-*
• Regex Match: service=~"payment|checkout"

Aggregations & Downsampling

Users rarely inspect millions of raw data points individually; they view aggregate trends across hosts, zones, or entire fleets. Goku natively supports standard aggregation operators:

• SUM, COUNT, MIN, MAX, AVG, and STDDEV.

Additionally, older or large-range queries require downsampling: collapsing hundreds or thousands of high-resolution points within a defined time window (e.g., 5-minute or 1-hour intervals) into a single representative point using an aggregator (such as the average, minimum, or maximum) to save both rendering time and query compute.

---

3. Key Design Decisions in Goku

To address the design shortcomings of OpenTSDB, Pinterest made four critical architectural choices:

Area	OpenTSDB Bottleneck	Goku Architecture Decision
Storage Medium	Disk/HDFS backed (HBase)	Pure in-memory (RAM) storage for the hot window (last 24 hours)
Memory Footprint	Bloated Java heap structures	Integrated Facebook’s Gorilla time series engine (~12x compression)
Distributed Querying	Scatter-gather pulls raw data to coordinator	Push-down aggregation: Local computation happens on shards
Serialization	Verbose JSON parsing overhead	Binary Apache Thrift protocol for queries and internal communication

3.1 100% In-Memory Hot Tier

The vast majority of operational queries (dashboards, alerts, auto-scalers, and anomaly detectors) access data from the most recent 24 hours. By bounding the in-memory window to 24 hours, Goku completely eliminates disk reads from the hot query path. Inverted indexes and data points reside strictly in RAM.

3.2 Gorilla In-Memory Compression

Keeping 24 hours of raw uncompressed metrics for millions of series in memory would require an astronomical amount of RAM. Goku adopts the compression algorithms pioneered by Facebook’s Gorilla time series database:

• Timestamps: Compressed using delta-of-delta encoding. Since metric reports arrive at fixed intervals (e.g., every 15s or 60s), the delta of deltas is usually zero or very small, requiring only 1 to a few bits instead of a 64-bit timestamp.
• Floating-point Values: Compressed using XOR floating-point encoding. Consecutive float values for metrics like CPU or memory often differ only in their lower-order bits. XORing consecutive values yields leading and trailing zeroes, which are packed compactly.

Together, Gorilla encoding delivers approximately 12x compression, allowing Pinterest to retain 24 hours of operational data entirely in RAM at manageable costs.

3.3 Compute Pushed to Storage

In OpenTSDB’s coordinator-centric scatter-gather model, gigabytes of raw time series points travel over the network to the query node before mathematical operations are applied.

Goku reverses this pipeline:

1. The coordinator breaks down the query into partial query specifications.
2. Shards execute filtering and calculate partial aggregations locally over their local in-memory segments.
3. Only the reduced, downsampled partial results are streamed over the network to the coordinator proxy.
4. The coordinator simply combines the partial aggregations and serves the client.

[OpenTSDB Approach]:
Shards --(All Raw Data Points)--> Query Coordinator --(Heavy Aggregation)--> User

[Goku Approach]:
Shards --(Local Partial Aggregation)--> Goku Proxy --(Merge Intermediate)--> User

---

4. Shard Storage Internals: The Two-Tiered Lifecycle

Metrics are partitioned across a cluster of Goku storage nodes (shards) by metric name and tag hash. Each Goku instance operates autonomously and stores metrics partitioned temporally into discrete buckets.

graph TD
    Incoming[Incoming Metric Writes] --> Shard[Goku Shard Node]
    
    subgraph Shard Architecture
        Shard --> BMap[Bucket Map: 24-Hour Sliding Window]
        
        subgraph 2-Hour Bucket Window
            BMap --> Mutable[Bucket Time Series Object
(Mutable In-Memory Hash Map)]
            Mutable -- After window closes --> Immutable[Bucket Storage
(Immutable Inverted Index & Gorilla Blocks)]
        end
        
        Immutable -. Periodic Flush .-> Disk[(Persistent Disk Storage)]
    end

4.1 Temporal Bucket Map

To bound memory footprint and organize time-bounded queries, a Goku shard organizes its 24-hour retention window into smaller discrete windows—typically 2-hour buckets (e.g., 00:00-02:00, 02:00-04:00, etc.).

Each 2-hour window contains two operational states:

4.2 The Mutable Buffer (Bucket Time Series Object)

Network jitter and asynchronous pipelines mean metric data points frequently arrive slightly out of order or with short delays. Incoming points are routed to an in-memory mutable structure called the Bucket Time Series Object.

• Implemented as a high-throughput in-memory hash map.
• Permits updates, append operations, and out-of-order writes within the active window.

4.3 The Immutable Segment (Bucket Storage)

Once the time window expires (e.g., after 2 hours), the bucket transitions to an immutable state called Bucket Storage:

• No further writes are permitted.
• Data is reorganized and finalized into a columnar, read-optimized format.
• An in-memory inverted index maps tags to compressed time series block IDs for fast tag-based lookups.
• Points are tightly compressed using Gorilla compression.

4.4 Disk Persistence

While Goku executes all queries against RAM, it ensures durability against process restarts or hardware crashes. Goku periodically checkpoints and flushes active buckets to local disk (e.g., every 5 minutes). Once a 2-hour bucket becomes immutable, it is written out in its final compressed form to disk.

---

5. Distributed Query Execution Workflow

When a dashboard (such as Grafana or an internal Pinterest metrics UI) requests data, the query lifecycle flows through a dedicated proxy layer.

sequenceDiagram
    autonumber
    actor Client as Analytics / Dashboard
    participant Proxy as Goku Proxy
    participant ShardA as Goku Shard A
    participant ShardB as Goku Shard B
    
    Client->>Proxy: Query (Metric, Tags, Range, Aggregator: SUM)
    Proxy->>Proxy: Determine target shards based on metric routing
    par Parallel Distributed Scan
        Proxy->>ShardA: Partial Query (Local Range, Filter, Local SUM)
        Proxy->>ShardB: Partial Query (Local Range, Filter, Local SUM)
        Note over ShardA: Scan Inverted Index in RAM<br/>Decompress Gorilla blocks<br/>Execute local downsampling & SUM
        Note over ShardB: Scan Inverted Index in RAM<br/>Decompress Gorilla blocks<br/>Execute local downsampling & SUM
        ShardA-->>Proxy: Thrift Response (Partial SUM per interval)
        ShardB-->>Proxy: Thrift Response (Partial SUM per interval)
    end
    Proxy->>Proxy: Merge partial sums across shards
    Proxy-->>Client: Final Time Series Response (Thrift / JSON)

Step-by-Step Query Execution:

1. Query Ingestion: The client issues a query requesting metrics matching specific tag predicates, over a defined time range, specifying a downsample interval (e.g., 1 minute) and aggregation type (e.g., SUM).
2. Routing: The Goku Proxy inspects the metric key and fans out the query concurrently to the exact Goku shards holding the relevant partitions.
3. Local In-Memory Scan: Each shard inspects its 2-hour Bucket Maps that intersect with the queried time range:
   • Queries against the recent, active window hit the mutable buffer.
   • Historical queries (up to 24 hours) hit the immutable bucket storage, leveraging the inverted index to resolve tag filters.
4. Shard-Level Aggregation: The shard decompresses only the matched Gorilla time series blocks, runs the downsampling function, and aggregates all matching local streams into intermediate results.
5. Thrift Transport: Shards serialize the intermediate aggregate results into compact Apache Thrift binary payloads and return them to the Goku Proxy.
6. Final Merge: The Goku Proxy combines the intermediate aggregates (e.g., summing partial sums across shards) into the final time series and returns it to the client.

---

6. Summary of Architectural Takeaways

1. Specialize for the Access Pattern: Operational telemetry has a clear temporal bias: recent data (last 24 hours) is queried orders of magnitude more frequently than historical data. Bounding the hot tier to 24 hours in RAM eliminates disk I/O bottlenecks.
2. Standing on the Shoulders of Giants: Rather than reinventing low-level bit packing from scratch, Pinterest utilized Facebook’s Gorilla compression algorithm inside Goku, achieving an immediate 12x reduction in memory consumption.
3. Decouple Ingestion from Query Optimization: Goku’s two-tiered lifecycle—a mutable hash map for handling out-of-order writes followed by an immutable columnar inverted index—balances high-throughput ingest with sub-millisecond query performance.
4. Bring Compute to the Data: Scatter-gather architectures that ship raw points across the network are non-viable at million-point-per-second scale. Pushing down aggregation, downsampling, and filtering directly to individual storage nodes reduces cluster network traffic by multiple orders of magnitude.