Prometheus TSDB

Gorilla TSDB

架构

类似方案对比

OpenTSDB：

• 基于 Hbase
• 不做  time roll up aggregation for older data（对比较老对数据进行精度压缩，比如按小时为粒度存储）
• a richer data model for identifying time series

Whisper (Graphite)：

• stores time series data on local disk in the Whisper format, a Round Robin Database (RRD) style database
• 需要时序数据固定 interval

InfluxDB：

• A even richer data model than OpenTSDB. 由于其灵活性，存储磁盘消耗更大。

作者认为以上方案的通用问题：从磁盘查不够快。

数据压缩

时间戳压缩：

1. 计算  delta of delta: D = (t3 - t2) - (t2 - t1)
2. 根据 D 的分布范围进行类似霍夫曼编码，比如 D 的访问可以分成5类 0，-63~64, -255~256, -2047~2048, 其他
3. 5个类别的出现概览由高到底，这样就可以进行编码压缩，压缩实验显示 96% 的时间戳都可以压缩到 1 bit 即 0

数值(double)压缩:

1. 存第一个 value，后面对value 对前面对value 进行 xor 操作
2. 如果 xor 为0 则存 0 （1bit）（这种情况占了 59% ）
3. 如果 xor 不为0，则看 xor 的结果里面前面的 0 的数量 和末尾的 0 的数量哪个多，根据 两种 condition 由两种存储方式。【 这种情况的压缩效率并不高，有优化空间。但是考虑到整体的 point 的存储空间已经从 16 bytes （int64 + double） 压缩到了 1.37 byte，这部分优化就不是那么重要了 】。

存储和查询

内存中的数据结构

• 里面比较重要的是 data block 的概念，closed 的datablock 不可变，而 open 的 datablock 只会 append 压缩后的数据。
• 由于特殊的压缩方式，查询时需要把整个 block 拷贝出来，返回给 client，解压缩步骤可以在 client 端完成。
• 由于特殊的压缩方式，查询数据需要查整个 block，那么 block 越小越好，而同时 block 越大，压缩效率则越高，最终选定 2 小时的为 block 的存储时常，这个压缩效率为 1.37 byte / 8 byte

磁盘上的数据结构

为了保证Gorilla 能从 crash 中恢复

• 有四种数据：Key lists（索引）, append-only logs（日志）, complete block files, and checkpoint files （保证写入的 block file 的可靠性）
• 要把 closed block 数据存储到 disk
• 同时保存一个 log，这个 log 是当前写入的数据，从这个 log 里面可以恢复 open data block 
• 当保存了closed block 之后，closed block 相关的 log 就可以删除，节约磁盘

高可用

• 磁盘数据结构保证 crash 能够恢复，基本不丢失数据，磁盘数据存在 GlusterFS，存 3 副本
• 所有 Gorilla  都有两个副本，当一个 down 了之后可以从另一个读取数据
• Gorilla node 前面有一个 ShardManager 进行数据分片管理 （每个 node 只存一个或者多个 shard，数据用 unique string keys 进行分片，所以每个时序数据都可以对应一个 Gorilla host）
• Unhealthy node 能自动转发请求到 healthy node
• 长期存储用 Hbase 存储

Prometheus TSDB

问题

时序数据的格式

读写的问题

写一般是对比如1000个series 同时进行采集，所以他是纵向的。磁盘随机写效率低、SSD 写放大问题  →  顺序写 和 批量写 是必然选择 （sequential and batched writes are the ideal write pattern for spinning disks and SSDs alike. A simple rule to stick）

读的问题更加复杂：可能是读一段时间一个 series 的数据，也可能是 读一个时间点 10000 个 series 的数据，所以读 是在 two-dimensional plane, 读既不是完全横向也不是纵向的，而是一个两者组合的矩形.

Series Churn 

为什么一个 series 一个文件不行：在微服务环境中 series 会非常多，和存在多时间可能很短 (作者命名为 series churn )

• 同时读写几万个文件效率低，频繁打开关闭几万个文件效率低
• 文件系统 inodes 不够用
• 清理过期数据成本高

筛选查询问题

series churn 带来另一个问题：由于 series 数量总是在不停膨胀的，所以从千万个 series 查询出满足条件的 series 的查询效率需要优化

资源消耗问题

series churn increases the usage of memory, CPU, and disk IO every time we scale an application or do a rolling update. 如果 单个文件存储一个 series 这个消耗非常高。

总结

总结一下，主要要解决的几个问题：

• 读写效率问题
• series churn 如何存储的问题
• 快速筛选查询 series 的问题
• 资源消耗问题

如何解决

文件结构设计

1. 把文件存储按照时间划分成多个 “little database” （block，每个 block 2小时），这样的好处是：
   1. 查询时序数据一般为一个时间范围，这样就可以忽略时间访问不在查询时间内的 block，同时也解决了 series churn 查询 的问题，因为需要检索的数据从一开始减少了
   2. 写完一个 block 之后，从内存持久化到文件中，这样就是批量顺序写，提高了写效率
   3. 最近的 chunks 是被查询最多的，始终在内存中，提高热数据查询效率
   4. 删除旧数据变得非常简单 → 只要删除一个文件夹就可以了
   5. 同时每个 block 中的同一个 series 的数据是 连续一段存储空间，提高查询效率
2. 使用 mmap 提高读写效率 （缓存的工作交给操作系统），这意味着我可以把整个 database 里面的文件都当成在内存里面进行操作，并且在适当的时候缓存和清理缓存。
3. Compaction: 由于单个 block 比较小（2小时），所以需要定期将小 block 定期合并成大 block 以方便后续的合并或者其他操作。 同时可以修改数据：dropping deleted data, or restructuring our sample chunks for improved query performance.

索引

使用倒序索引，series ID 是 forward index. 对每种 label name/value 对都建立倒序索引，比如 app=nginx → ID 1，4，5，ID保持有序，这样多个 label 组合的时候查询效率也是 O(n)

总结

• 在内存中 batching，使用 wal 跟踪，定期 flush 到磁盘，这是当今数据库类程序常见的操作
• 使用在横向划分多个 block 定期 Compaction/Retention 的方式提高读写/删除效率
• 使用倒序索引提高筛选查询效率

文件结构详细

基于 2.10 

meta.json

meta.json 例子

在下面这个例子里面：

• 存储了 54 小时的数据 maxTime-minTime/ 3600,000; 时间单位一般都是 ms 
• 27 个 block 合并而成，原始的 block 都是 2小时，level 为 4表示合并了多次，原始的 27 个block id 都在 sources 里面
• 共计 1050682376 个数据点，351207 个 series，9083900 个 chunk （大概是 series * 54/2 ）

Index

【索引这块看上去有不小的优化空间，观察一个 100 M 左右的 block (kubernetes 环境的 prometheus)，index 文件24 M左右，占比很高】

【另：按照时间分片的文件结构也带来不少问题，一个问题是构建 cluster 变得很困难（同时因为prometheus 是主动 scrape ），如何在多节点上进行分片是一个问题，参考 Gorilla 的做一个 shard manager （按照 series id/name 进行分片 ）是一个办法，前端抓取之后，后端分片存储到多个prometheus instance，查询的时候再做 merge】

label 中用的strings，集中存储能减少 index 大小series 数据，按照 label set 排序，同时 id = offset/16一个 series 数据的索引信息，包括 labels, chunks的开始结束时间，以及真实chuck 索引一个 label对 的所有倒序索引，即这个 label 对匹配的series idsPosting 的索引，对一个 label 对，其对应的posting 位置，这里比较奇怪的是 name 和 value 直接存储并没有索引到 Symbol table整体布局的索引

Chunks

chunks/ 文件夹下面的内容，最大单文件大小512 M

┌──────────────────────────────┐
│  magic(0x85BD40DD) <4 byte>  │
├──────────────────────────────┤
│    version(1) <1 byte>       │
├──────────────────────────────┤
│    padding(0) <3 byte>       │
├──────────────────────────────┤
│ ┌──────────────────────────┐ │
│ │         Chunk 1          │ │
│ ├──────────────────────────┤ │
│ │          ...             │ │
│ ├──────────────────────────┤ │
│ │         Chunk N          │ │
│ └──────────────────────────┘ │
└──────────────────────────────┘

chuck

┌───────────────┬───────────────────┬──────────────┬────────────────┐
│ len <uvarint> │ encoding <1 byte> │ data <bytes> │ CRC32 <4 byte> │
└───────────────┴───────────────────┴──────────────┴────────────────┘

Tombstones

被删除的 数据

┌────────────────────────────┬─────────────────────┐
│ magic(0x0130BA30) <4b>     │ version(1) <1 byte> │
├────────────────────────────┴─────────────────────┤
│ ┌──────────────────────────────────────────────┐ │
│ │                Tombstone 1                   │ │
│ ├──────────────────────────────────────────────┤ │
│ │                      ...                     │ │
│ ├──────────────────────────────────────────────┤ │
│ │                Tombstone N                   │ │
│ ├──────────────────────────────────────────────┤ │
│ │                  CRC<4b>                     │ │
│ └──────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────┘

chuck

┌────────────────┬─────────────────┬────────────────┐
│ref <uvarint64> │ mint <varint64> │ maxt <varint64>│
└────────────────┴─────────────────┴────────────────┘

Wal

默认单文件大小限制为 128 M

Series records encode the labels that identifies a series and its unique ID.
┌────────────────────────────────────────────┐
│ type = 1 <1b>                              │
├────────────────────────────────────────────┤
│ ┌─────────┬──────────────────────────────┐ │
│ │ id <8b> │ n = len(labels) <uvarint>    │ │
│ ├─────────┴────────────┬─────────────────┤ │
│ │ len(str_1) <uvarint> │ str_1 <bytes>   │ │
│ ├──────────────────────┴─────────────────┤ │
│ │  ...                                   │ │
│ ├───────────────────────┬────────────────┤ │
│ │ len(str_2n) <uvarint> │ str_2n <bytes> │ │
│ └───────────────────────┴────────────────┘ │
│                  . . .                     │
└────────────────────────────────────────────┘

Sample records encode samples as a list of triples (series_id, timestamp, value).

┌──────────────────────────────────────────────────────────────────┐
│ type = 2 <1b>                                                    │
├──────────────────────────────────────────────────────────────────┤
│ ┌────────────────────┬───────────────────────────┐               │
│ │ id <8b>            │ timestamp <8b>            │               │
│ └────────────────────┴───────────────────────────┘               │
│ ┌────────────────────┬───────────────────────────┬─────────────┐ │
│ │ id_delta <uvarint> │ timestamp_delta <uvarint> │ value <8b>  │ │
│ └────────────────────┴───────────────────────────┴─────────────┘ │
│                              . . .                               │
└──────────────────────────────────────────────────────────────────┘

其他 tsdb 相关参考

• 比较标准的LSM tree是实现。参考
• The NoSQL Ecosystem
• Key-Value stores: a practical overview

tsdb 部分源码分析

compact 流程

• db.go 后台线程 db.run()
  • db.Compact(): chan notify with backoff
    • db.compactor.Write(db.dir, rangeHead:head, ...)
      • LeveledCompactor.write(dest, meta, BlockReader:rangeHead:b)
      • LeveledCompactor.populateBlock(BlockReader:rangeHead:blocks, meta, index.Writer, chucks.Writer)
        • index.Writer.AddSymbol
        • index.Writer.ensureStage(idxStageSymbols): startSymbols
        • chucks.Writer.WriteChunks: level=1, writer 为 instrumentedChunkWriter（只是多记录了一些 metrics，其他一样）
        • index.Writer.AddSeries
        • index.Writer.ensureStage(idxStageSeries): finishSymbols
        • index.Writer.Close()
        • index.Writer.ensureStage(idxStageDone)
        • writePostingsToTmpFiles
        • writeLabelIndices
        • writePostings
        • writeLabelIndexesOffsetTable
        • writePostingsOffsetTable
        • writeTOC
      • writeMetaFile
      • tombstones.WriteFile
    • db.reload(): reload blocks and trigger head truncation if new blocks appeared.Blocks that are obsolete due to replacement or retention will be deleted.
      • openBlocks(db.dir, db.blocks...)
        • readMetaFile
        • getBlock, if not open, OpenBlock
        • readMetaFile
        • return Block with index.Reader, chunks.Reader, tombstones.Reader
      • db.deleteBlocks(deletable)
    • db.compactor.Plan(): Plan returns a set of directories that can be compacted concurrently.
    • db.compactor.Compact(): Compact runs compaction against the provided directories.
      • compactBlockMetas
      • db.compactor(LeveledCompactor).write(dest, meta, blocks...) 和保存head里面的数据到文件的不同之处只是 BlockMeta.Compaction.Level 不同
    • db.reload()
    • runtime.GC(): 多次运行

query 流程

• DB.Querier(mint, maxt int64): append(rangeHeader, blocks)
• tsdb.querier.Select(ms ...*labels.Matcher)
  • for b in querier.blocks: b.Select(ms...)
    • blockQuerier.LookupChunkSeries
      • PostingsForMatchers
        • for m in labels.Matchers: postingsForMatcher
        • IndexReader.Postings(labelName, labelValue)
      • return baseChunkSeries{index.Postings, IndexReader, MemTombstones}