時序數據庫(TSDB: Time Series Database)大多數時候都是為了滿足監控場景的需求，這里先介紹兩個概念：

• 數據點(Point): 時序數據的數據點是一個包含 (Timestamp:int64, Value:float64) 的二元組。
• 時間線(Series): 不同標簽(Label)的組合稱為不同的時間線，如

1. series1: {"__name__": "netspeed", "host": "localhost", "iface": "eth0"} 
2. series2: {"__name__": "netspeed", "host": "localhost", "iface": "eth1"} 

Prometheus, InfluxDB, M3, TimescaleDB 都是時下流行的 TSDB。時序數據的壓縮算法很大程度上決定了 TSDB 的性能，以上幾個項目的實現都參考了 Fackbook 2015 年發表的論文《Gorilla: A fast, scalable, in-memory time series database》(http://www.vldb.org/pvldb/vol8/p1816-teller.pdf) 中提到的差值算法，該算法平均可以將 16 字節的數據點壓縮成 1.37 字節。

Who's mando?

• Din Djarin, also known as "the Mandalorian" or simply "Mando," was a human male Mandalorian who worked as a famous bounty hunter during the New Republic Era.

What's mandodb?

mandodb(https://github.com/chenjiandongx/mandodb) 是我在學習過程中實現的一個最小化的 TSDB，從概念上來講它還算不上是一個完整的 TSDB，因為它：

• 沒有實現自己的查詢引擎(實現難度大)
• 缺少磁盤歸檔文件 Compact 操作(有空的話會實現)
• 沒有 WAL 作為災備保證高可用(心情好的話會實現)

mandodb 主要受到了兩個項目的啟發。本項目僅限于學習用途，未經生產環境測試驗證!

• nakabonne/tstorage
• prometheus/prometheus

prometheus 的核心開發者 Fabian Reinartz 寫了一篇文章 《Writing a Time Series Database from Scratch》(https://fabxc.org/tsdb/) 來介紹 prometheus TSDB 的演變過程，非常值得一讀，強烈推薦。

 數據模型 & API 文檔

 

數據模型定義

1. // Point 表示一個數據點 (ts, value) 二元組 
2. type Point struct { 
3.  Ts    int64 // in seconds 
4.  Value float64 
5. } 
6.  
7. // Label 代表一個標簽組合 
8. type Label struct { 
9.  Name  string 
10.  Value string 
11. } 
12.  
13. // Row 一行時序數據 包括數據點和標簽組合 
14. type Row struct { 
15.  Metric string 
16.  Labels LabelSet 
17.  Point  Point 
18. } 
19.  
20. // LabelSet 表示 Label 組合 
21. type LabelSet []Label 
22.  
23. // LabelMatcher Label 匹配器 支持正則 
24. type LabelMatcher struct { 
25.  Name   string 
26.  Value  string 
27.  IsRegx bool 
28. } 
29.  
30. // LabelMatcherSet 表示 LabelMatcher 組合 
31. type LabelMatcherSet []LabelMatcher 

API

1. // InsertRows 寫數據 
2. InsertRows(rows []*Row) error  
3.  
4. // QueryRange 查詢時序數據點 
5. QueryRange(metric string, lms LabelMatcherSet, start, end int64) ([]MetricRet, error) 
6.  
7. // QuerySeries 查詢時序序列組合 
8. QuerySeries(lms LabelMatcherSet, start, end int64) ([]map[string]string, error) 
9.  
10. // QueryLabelValues 查詢標簽值 
11. QueryLabelValues(label string, start, end int64) []string 

 配置選項

 

配置項在初始化 TSDB 的時候設置。

1. // WithMetaSerializerType 設置 Metadata 數據的序列化類型 
2. // 目前只提供了 BinaryMetaSerializer 
3. WithMetaSerializerType(t MetaSerializerType) Option  
4.  
5. // WithMetaBytesCompressorType 設置字節數據的壓縮算法 
6. // 目前提供了 
7. // * 不壓縮: NoopBytesCompressor（默認） 
8. // * ZSTD: ZstdBytesCompressor 
9. // * Snappy: SnappyBytesCompressor 
10. WithMetaBytesCompressorType(t BytesCompressorType) Option 
11.  
12. // WithOnlyMemoryMode 設置是否默認只存儲在內存中 
13. // 默認為 false 
14. WithOnlyMemoryMode(memoryMode bool) Option 
15.  
16. // WithEnabledOutdated 設置是否支持亂序寫入 此特性會增加資源開銷 但會提升數據完整性 
17. // 默認為 true 
18. WithEnabledOutdated(outdated bool) Option 
19.  
20. // WithMaxRowsPerSegment 設置單 Segment 最大允許存儲的點數 
21. // 默認為 19960412（夾雜私貨 ） 
22. WithMaxRowsPerSegment(n int64) Option 
23.  
24. // WithDataPath 設置 Segment 持久化存儲文件夾 
25. // 默認為 "." 
26. WithDataPath(d string) Option 
27.  
28. // WithRetention 設置 Segment 持久化數據保存時長 
29. // 默認為 7d 
30. WithRetention(t time.Duration) Option 
31.  
32. // WithWriteTimeout 設置寫入超時閾值 
33. // 默認為 30s 
34. WithWriteTimeout(t time.Duration) Option 
35.  
36. // WithLoggerConfig 設置日志配置項 
37. // logger: github.com/chenjiandongx/logger 
38. WithLoggerConfig(opt *logger.Options) Option 

 用法示例

 

1. package main 
2.  
3. import ( 
4.  "fmt" 
5.  "time" 
6.  
7.  "github.com/chenjiandongx/mandodb" 
8. ) 
9.  
10. func main() { 
11.  store := mandodb.OpenTSDB( 
12.   mandodb.WithOnlyMemoryMode(true), 
13.   mandodb.WithWriteTimeout(10*time.Second), 
14.  ) 
15.  defer store.Close() 
16.  
17.  // 插入數據 
18.  _ = store.InsertRows([]*mandodb.Row{ 
19.   { 
20.    Metric: "cpu.busy", 
21.    Labels: []mandodb.Label{ 
22.     {Name: "node", Value: "vm1"}, 
23.     {Name: "dc", Value: "gz-idc"}, 
24.    }, 
25.    Point: mandodb.Point{Ts: 1600000001, Value: 0.1}, 
26.   }, 
27.   { 
28.    Metric: "cpu.busy", 
29.    Labels: []mandodb.Label{ 
30.     {Name: "node", Value: "vm2"}, 
31.     {Name: "dc", Value: "sz-idc"}, 
32.    }, 
33.    Point: mandodb.Point{Ts: 1600000001, Value: 0.1}, 
34.   }, 
35.  }) 
36.  
37.  time.Sleep(time.Millisecond) 
38.  
39.  // 時序數據查詢 
40.  data, _ := store.QueryRange("cpu.busy", nil, 1600000000, 1600000002) 
41.  fmt.Printf("data: %+v\n", data) 
42.  // output: 
43.  // data: [{Labels:{__name__="cpu.busy", dc="gz-idc", node="vm1"} Points:[{Ts:1600000001 Value:0.1}]}] 
44.  
45.  // 查詢 Series 
46.  // __name__ 是 metric 名稱在 TSDB 中的 Label Key 
47.  ser, _ := store.QuerySeries( 
48.         mandodb.LabelMatcherSet{{Name: "__name__", Value: "cpu.busy"}}, 1600000000, 1600000002) 
49.  for _, d := range ser { 
50.   fmt.Printf("data: %+v\n", d) 
51.  } 
52.  // output: 
53.  // data: map[__name__:cpu.busy dc:gz-idc node:vm1] 
54.  // data: map[__name__:cpu.busy dc:sz-idc node:vm2] 
55.  
56.  // 查詢標簽值 
57.  lvs := store.QueryLabelValues("node", 1600000000, 1600000002) 
58.  fmt.Printf("data: %+v\n", lvs) 
59.  // output: 
60.  // data: [vm1 vm2] 
61. } 

下面是我對這段時間學習內容的整理，嘗試完整介紹如何從零開始實現一個小型的 TSDB。

我本身并沒有數據庫開發的背景，某些描述可能并不那么準確，所以歡迎 實名 diss 指正。

 Gorilla 差值算法

 

Gorilla 論文 4.1 小節介紹了壓縮算法，先整體看一下壓縮方案，T/V 是緊挨存儲的，'0'/'10'/'11' 表示控制位。

Figure: Gorilla 壓縮算法

Timestamp DOD 壓縮：

在時序的場景中，每個時序點都有一個對應的 Timestamp，一條時序序列中相鄰數據點的間隔是有規律可循的。一般來講，監控數據的采集都是會以固定的時間間隔進行的，所以就可以用差值來記錄時間間隔，更進一步，我們可以用差值的差值來記錄以此來減少存儲空間。

1. t1: 1627401800; t2: 1627401810; t3: 1627401820; t4: 1627401830 
2. -------------------------------------------------------------- 
3. // 差值：delta 
4. t1: 1627401800; (t2-t1)d1: 10; (t3-t2)d2: 10; (t4-t3)d3: 10;  
5. -------------------------------------------------------------- 
6. // 差值的差值：delta of delta 
7. t1: 1627401800; dod1: 0; dod2: 0; dod3: 0;  

實際環境中當然不可能每個間隔都這么均勻，由于網絡延遲等其他原因，差值會有波動。

Value XOR 壓縮：

Figure: IEEE 浮點數以及 XOR 計算結果

當兩個數據點數值值比較接近的話，通過異或操作計算出來的結果是比較相似的，利用這點就可以通過記錄前置零和后置零個數以及數值部分來達到壓縮空間的目的。

下面通過算法實現來介紹，代碼來自項目 dgryski/go-tsz。代碼完全按照論文中給出的步驟來實現。

1. // New 初始化 block 這里會將第一個原始時間戳寫入到 block 中 
2. func New(t0 uint32) *Series { 
3.  s := Series{ 
4.   T0:      t0, 
5.   leading: ^uint8(0), 
6.  } 
7.  
8.  s.bw.writeBits(uint64(t0), 32) 
9.  return &s 
10. } 
11.  
12. // Push 負責寫入時序數據 
13. func (s *Series) Push(t uint32, v float64) { 
14.  // .... 
15.  // 如果是第一個數據點的話寫入原始數據后直接返回 
16.  if s.t == 0 { 
17.   s.t = t 
18.   s.val = v 
19.   s.tDelta = t - s.T0 // 實際上這里為 0 
20.  
21.   // The block header stores the starting time stamp, t-1（前一個時間戳）, 
22.   // which is aligned to a two hour window; the first time 
23.   // stamp, t0, in the block is stored as a delta from t−1 in 14 bits. 
24.          
25.   // 用 14 個 bit 寫入時間戳差值 
26.   s.bw.writeBits(uint64(s.tDelta), 14) 
27.   // 原始數據點完整寫入 
28.   s.bw.writeBits(math.Float64bits(v), 64) 
29.   return 
30.  } 
31.  
32.  tDelta := t - s.t 
33.  dod := int32(tDelta - s.tDelta) // 計算差值的差值 Detla of Delta 
34.  
35.  // 下面開始就處理非第一個數據點的情況了 
36.  switch { 
37.   // If D is zero, then store a single ‘0’ bit 
38.   // 如果是零的話 那直接用 '0' 一個字節就可以直接表示 
39.  case dod == 0: 
40.   s.bw.writeBit(zero) 
41.  
42.   //  If D is between [-63, 64], store ‘10’ followed by the value (7 bits) 
43.  case -63 <= dod && dod <= 64: 
44.   s.bw.writeBits(0x02, 2) // 控制位 '10' 
45.   s.bw.writeBits(uint64(dod), 7) // 7bits 可以表示 [-63, 64] 的范圍 
46.  
47.   // If D is between [-255, 256], store ‘110’ followed by the value (9 bits) 
48.  case -255 <= dod && dod <= 256: 
49.   s.bw.writeBits(0x06, 3) // 控制位 '110' 
50.   s.bw.writeBits(uint64(dod), 9) 
51.  
52.   // if D is between [-2047, 2048], store ‘1110’ followed by the value (12 bits) 
53.  case -2047 <= dod && dod <= 2048: 
54.   s.bw.writeBits(0x0e, 4) // 控制位 '1110' 
55.   s.bw.writeBits(uint64(dod), 12) 
56.  
57.   // Otherwise store ‘1111’ followed by D using 32 bits 
58.  default: 
59.   s.bw.writeBits(0x0f, 4) // 其余情況控制位均用 '1111' 
60.   s.bw.writeBits(uint64(dod), 32) 
61.  } 
62.  
63.  // 到這里 (T, V) 中的時間戳已經寫入完畢了 接下來是寫 V 部分 
64.  
65.  // 先計算兩個值的異或結果 
66.  vDelta := math.Float64bits(v) ^ math.Float64bits(s.val) 
67.  
68.  // If XOR with the previous is zero (same value), store single ‘0’ bit 
69.  // 如果前后兩個值相等的話 直接用 '0' 1 個 bit 就可以表示 
70.  // 所以如果上報的時序數據是 1 或者 0 這種的話 占用的內存會非常少 
71.  
72.  // zero = '0'; one = '1' 
73.  if vDelta == 0 { 
74.   s.bw.writeBit(zero) 
75.  } else {    // 非 0 情況那就要把控制位置為 1 
76.   s.bw.writeBit(one) 
77.  
78.   // 計算前置 0 和后置 0 
79.   leading := uint8(bits.LeadingZeros64(vDelta)) 
80.   trailing := uint8(bits.TrailingZeros64(vDelta)) 
81.  
82.   // clamp number of leading zeros to avoid overflow when encoding 
83.   if leading >= 32 { 
84.    leading = 31 
85.   } 
86.  
87.   // (Control bit ‘0’) If the block of meaningful bits 
88.   // falls within the block of previous meaningful bits, 
89.   // i.e., there are at least as many leading zeros and 
90.   // as many trailing zeros as with the previous value, 
91.   // use that information for the block position and 
92.   // just store the meaningful XORed value. 
93.  
94.   // 如果前置 0 不小于上一個值計算的異或結果的前置 0 且后置 0 也不小于上一個值計算的異或結果的后置 0 
95.   if s.leading != ^uint8(0) && leading >= s.leading && trailing >= s.trailing { // => 控制位 '10' 
96.    s.bw.writeBit(zero) 
97.    // 記錄異或值非零部分 
98.    s.bw.writeBits(vDelta>>s.trailing, 64-int(s.leading)-int(s.trailing)) 
99.   } else { // => 控制位 '11' 
100.  
101.    // (Control bit ‘1’) Store the length of the number 
102.    // of leading zeros in the next 5 bits, then store the 
103.    // length of the meaningful XORed value in the next 
104.    // 6 bits. Finally store the meaningful bits of the XORed value. 
105.    s.leading, s.trailing = leading, trailing 
106.  
107.    // 其他情況控制位置為 1 并用接下來的 5bits 記錄前置 0 個數 
108.    s.bw.writeBit(one) 
109.    s.bw.writeBits(uint64(leading), 5) 
110.  
111.    // 然后用接下來的 6bits 記錄異或差值中的非零部分 
112.    sigbits := 64 - leading - trailing 
113.    s.bw.writeBits(uint64(sigbits), 6) 
114.    s.bw.writeBits(vDelta>>trailing, int(sigbits)) 
115.   } 
116.  } 
117.  
118.  // 狀態更新 至此（T, V）均已被壓縮寫入到內存中 
119.  s.tDelta = tDelta 
120.  s.t = t 
121.  s.val = v 
122. } 
123.  
124. // 每個 block 的結尾會使用特殊標記用于標識 
125. func finish(w *bstream) { 
126.  // write an end-of-stream record 
127.  w.writeBits(0x0f, 4) 
128.  w.writeBits(0xffffffff, 32) 
129.  w.writeBit(zero) 
130. } 

論文給出了不同 case 的 buckets 占比分布。

Figure: Timestamp buckets distribution

Figure: Value buckets distribution

Timestamp buckets 中，前后兩個時間戳差值相同的比例高達 96.39%，而在 Value buckets 中只用一個控制位的占比也達到了 59.06%，可見其壓縮比之高。

論文還給出了一個重要結論，數據壓縮比隨著時間的增長而增長，并在 120 個點的時候開始收斂到一個最佳值。

Figure: 壓縮率曲線

Gorilla 差值算法也應用于我的另外一個項目 chenjiandongx/tszlist，一種時序數據線程安全鏈表。

 數據寫入

 

時序數據具有「垂直寫，水平查」的特性，即同一時刻有多條時間線的數據不斷被追加。但查詢的時候往往是查某條時間線持續一段時間內的數據點。

1. series 
2.   ^    
3.   │   . . . . . . . . . . . . . . . . .   . . . . .   {__name__="request_total", method="GET"} 
4.   │     . . . . . . . . . . . . . . . . . . . . . .   {__name__="request_total", method="POST"} 
5.   │         . . . . . . . 
6.   │       . . .     . . . . . . . . . . . . . . . .                  ...  
7.   │     . . . . . . . . . . . . . . . . .   . . . .    
8.   │     . . . . . . . . . .   . . . . . . . . . . .   {__name__="errors_total", method="POST"} 
9.   │           . . .   . . . . . . . . .   . . . . .   {__name__="errors_total", method="GET"} 
10.   │         . . . . . . . . .       . . . . . 
11.   │       . . .     . . . . . . . . . . . . . . . .                  ...  
12.   │     . . . . . . . . . . . . . . . .   . . . .  
13.   v 
14.     <-------------------- time ---------------------> 

時序數據跟時間是強相關的(不然還叫時序數據?)，即大多數查詢其實只會查詢最近時刻的數據，這里的「最近」是個相對概念。所以沒必要維護一條時間線的完整生命周期，特別是在 Kubernetes 這種云原生場景，Pod 隨時有可能會被擴縮容，也就意味著一條時間線的生命周期可能會很短。如果我們一直記錄著所有的時間線的索引信息，那么隨著時間的推移，數據庫里的時間線的數量會呈現一個線性增長的趨勢 ，會極大地影響查詢效率。

這里引入一個概念「序列分流」，這個概念描述的是一組時間序列變得不活躍，即不再接收數據點，取而代之的是有一組新的活躍的序列出現的場景。

1. series 
2.   ^ 
3.   │   . . . . . . 
4.   │   . . . . . . 
5.   │   . . . . . . 
6.   │               . . . . . . . 
7.   │               . . . . . . . 
8.   │               . . . . . . . 
9.   │                             . . . . . . 
10.   │                             . . . . . . 
11.   │                                         . . . . . 
12.   │                                         . . . . . 
13.   │                                         . . . . . 
14.   v 
15.     <-------------------- time ---------------------> 

我們將多條時間線的數據按一定的時間跨度切割成多個小塊，每個小塊本質就是一個獨立小型的數據庫，這種做法另外一個優勢是清除過期操作的時候非常方便，只要將整個塊給刪了就行 (梭哈是一種智慧)。內存中保留最近兩個小時的熱數據(Memory Segment)，其余數據持久化到磁盤(Disk Segment)。

Figure: 序列分塊

DiskSegment 使用的是 AVL Tree 實現的列表，可在插入時排序。為什么不用更加高大上的紅黑樹?因為不好實現...

當 Memory Segment 達到歸檔條件的時候，會創建一個新的內存塊并異步將剛歸檔的塊寫入到磁盤，同時會使用 mmap 將磁盤文件句柄映射到內存中。代碼實現如下。

1. func (tsdb *TSDB) getHeadPartition() (Segment, error) { 
2.  tsdb.mut.Lock() 
3.  defer tsdb.mut.Unlock() 
4.  
5.  if tsdb.segs.head.Frozen() { 
6.   head := tsdb.segs.head 
7.  
8.   go func() { 
9.    tsdb.wg.Add(1) 
10.    defer tsdb.wg.Done() 
11.  
12.    tsdb.segs.Add(head) 
13.  
14.    t0 := time.Now() 
15.    dn := dirname(head.MinTs(), head.MaxTs()) 
16.  
17.    if err := writeToDisk(head.(*memorySegment)); err != nil { 
18.     logger.Errorf("failed to flush data to disk, %v", err) 
19.     return 
20.    } 
21.  
22.    fname := path.Join(dn, "data") 
23.    mf, err := mmap.OpenMmapFile(fname) 
24.    if err != nil { 
25.     logger.Errorf("failed to make a mmap file %s, %v", fname, err) 
26.     return 
27.    } 
28.  
29.    tsdb.segs.Remove(head) 
30.    tsdb.segs.Add(newDiskSegment(mf, dn, head.MinTs(), head.MaxTs())) 
31.    logger.Infof("write file %s take: %v", fname, time.Since(t0)) 
32.   }() 
33.  
34.   tsdb.segs.head = newMemorySegment() 
35.  } 
36.  
37.  return tsdb.segs.head, nil 
38. } 

Figure: Memory Segment 兩部分數據

寫入的時候支持數據時間回撥，也就是支持有限的亂序數據寫入，實現方案是在內存中對還沒歸檔的每條時間線維護一個鏈表(同樣使用 AVL Tree 實現)，當數據點的時間戳不是遞增的時候存儲到鏈表中，查詢的時候會將兩部分數據合并查詢，持久化的時候也會將兩者合并寫入。

 Mmap 內存映射

 

mmap 是一種將磁盤文件映射到進程的虛擬地址空間來實現對文件讀取和修改操作的技術。

從 Linux 角度來看，操作系統的內存空間被分為「內核空間」和「用戶空間」兩大部分，其中內核空間和用戶空間的空間大小、操作權限以及核心功能都不相同。這里的內核空間是指操作系統本身使用的內存空間，而用戶空間則是提供給各個進程使用的內存空間。由于用戶進程不具有訪問內核資源的權限，例如訪問硬件資源，因此當一個用戶進程需要使用內核資源的時候，就需要通過 系統調用 來完成。

虛擬內存細節可以閱讀 《虛擬內存精粹》 這篇文章。

Figure: 常規文件操作和 mmap 操作的區別

常規文件操作

讀文件： 用戶進程首先執行 read(2) 系統調用，會進行系統上下文環境切換，從用戶態切換到內核態，之后由 DMA 將文件數據從磁盤讀取到內核緩沖區，再將內核空間緩沖區的數據復制到用戶空間的緩沖區中，最后 read(2) 系統調用返回，進程從內核態切換到用戶態，整個過程結束。

寫文件： 用戶進程發起 write(2) 系統調用，從用戶態切換到內核態，將數據從用戶空間緩沖區復制到內核空間緩沖區，接著 write(2) 系統調用返回，同時進程從內核態切換到用戶態，數據從內核緩沖區寫入到磁盤，整個過程結束。

mmap 操作

mmap 內存映射的實現過程，總的來說可以分為三個階段：

進程啟動映射過程，并在虛擬地址空間中為映射創建虛擬映射區域。

執行內核空間的系統調用函數 mmap，建立文件物理地址和進程虛擬地址的一一映射關系。

進程發起對這片映射空間的訪問，引發缺頁異常，實現文件內容到物理內存的拷貝。

 小結

常規文件操作為了提高讀寫效率和保護磁盤，使用了頁緩存機制。這樣造成讀文件時需要先將文件頁從磁盤拷貝到頁緩存中，由于頁緩存處在內核空間，不能被用戶進程直接尋址，所以還需要將頁緩存中數據頁再次拷貝到內存對應的用戶空間中。這樣，通過了兩次數據拷貝過程，才能完成進程對文件內容的獲取任務。寫操作也是一樣，待寫入的 buffer 在內核空間不能直接訪問，必須要先拷貝至內核空間對應的主存，再寫回磁盤中(延遲寫回)，也是需要兩次數據拷貝。

而使用 mmap 操作文件，創建新的虛擬內存區域和建立文件磁盤地址和虛擬內存區域映射這兩步，沒有任何文件拷貝操作。而之后訪問數據時發現內存中并無數據而發起的缺頁異常過程，可以通過已經建立好的映射關系，只使用一次數據拷貝，就從磁盤中將數據傳入內存的用戶空間中，供進程使用。

總而言之，常規文件操作需要從磁盤到頁緩存再到用戶主存的兩次數據拷貝。而 mmap 操控文件只需要從磁盤到用戶主存的一次數據拷貝過程。mmap 的關鍵點是實現了「用戶空間」和「內核空間」的數據直接交互而省去了不同空間數據復制的開銷。

 索引設計

 

TSDB 的查詢，是通過 Label 組合來鎖定到具體的時間線進而確定分塊偏移檢索出數據。

Sid(MetricHash/-/LabelHash) 是一個 Series 的唯一標識。

Label(Name/-/Value) => vm="node1"; vm="node2"; iface="eth0"。

在傳統的關系型數據庫，索引設計可能是這樣的。

時序數據是 NoSchema 的，沒辦法提前建表和定義數據模型 ，因為我們要支持用戶上報任意 Label 組合的數據，這樣的話就沒辦法進行動態的擴展了?；蛟S你會靈光一現 ?，既然這樣，那把 Labels 放一個字段拼接起來不就可以無限擴展啦，比如下面這個樣子。

喲嚯，乍一看沒毛病，靚仔竊喜。

不對，有問題，要定位到其中的某條時間線，那我是不是得全表掃描一趟。而且這種設計還有另外一個弊病，就是會導致內存激增，Label 的 Name 和 Value 都可能是特別長的字符串。

那怎么辦呢(靚仔沉默...)，剎那間我的腦中閃過一個帥氣的身影，沒錯，就是你，花澤類「只要倒立眼淚就不會流出來」。

我悟了!要學會逆向思維，把 Label 當做主鍵，Sid 當做其字段不就好了。這其實有點類似于 ElasticSearch 中的倒排索引，主鍵為 Keyword，字段為 DocumentID。索引設計如下。

Label 作為主鍵時會建立索引(Hashkey)，查找的效率可視為 O(1)，再根據鎖定的 Label 來最終確定想要的 Sid。舉個例子，我們想要查找 {vm="node1", iface="eth0"} 的時間線的話就可以快速定位到 Sids(忽略其他 ... sid)。

1. sid1; sid2; sid3 
2. sid2; sid3; sid5 

兩者求一個交集，就可以得到最終要查詢的 Sid 為 sid2 和 sid3。Nice!

假設我們的查詢只支持相等匹配的話，格局明顯就小了。查詢條件是 {vm=~"node*", iface="eth0"} 腫么辦?對 label1、label2、label3 和 label4 一起求一個并集嗎?顯然不是，因為這樣算的話那結果就是 sid3。

厘清關系就不難看出，只要對相同的 Label Name 做并集然后再對不同的 Label Name 求交集就可以了。這樣算的正確結果就是 sid3 和 sid5。實現的時候用到了 Roaring Bitmap，一種優化的位圖算法。

Memory Segment 索引匹配

1. func (mim *memoryIndexMap) MatchSids(lvs *labelValueSet, lms LabelMatcherSet) []string { 
2.  // ... 
3.  sids := newMemorySidSet() 
4.  var got bool 
5.  for i := len(lms) - 1; i >= 0; i-- { 
6.   tmp := newMemorySidSet() 
7.   vs := lvs.Match(lms[i]) 
8.   // 對相同的 Label Name 求并集 
9.   for _, v := range vs { 
10.    midx := mim.idx[joinSeparator(lms[i].Name, v)] 
11.    if midx == nil || midx.Size() <= 0 { 
12.     continue 
13.    } 
14.  
15.    tmp.Union(midx.Copy()) 
16.   } 
17.  
18.   if tmp == nil || tmp.Size() <= 0 { 
19.    return nil 
20.   } 
21.  
22.   if !got { 
23.    sids = tmp 
24.    got = true 
25.    continue 
26.   } 
27.  
28.   // 對不同的 Label Name 求交集 
29.   sids.Intersection(tmp.Copy()) 
30.  } 
31.  
32.  return sids.List() 
33. } 

Disk Segment 索引匹配

1. func (dim *diskIndexMap) MatchSids(lvs *labelValueSet, lms LabelMatcherSet) []uint32 { 
2.  // ... 
3.  
4.  lst := make([]*roaring.Bitmap, 0) 
5.  for i := len(lms) - 1; i >= 0; i-- { 
6.   tmp := make([]*roaring.Bitmap, 0) 
7.   vs := lvs.Match(lms[i]) 
8.  
9.   // 對相同的 Label Name 求并集 
10.   for _, v := range vs { 
11.    didx := dim.label2sids[joinSeparator(lms[i].Name, v)] 
12.    if didx == nil || didx.set.IsEmpty() { 
13.     continue 
14.    } 
15.  
16.    tmp = append(tmp, didx.set) 
17.   } 
18.  
19.   union := roaring.ParOr(4, tmp...) 
20.   if union.IsEmpty() { 
21.    return nil 
22.   } 
23.  
24.   lst = append(lst, union) 
25.  } 
26.  
27.  // 對不同的 Label Name 求交集 
28.  return roaring.ParAnd(4, lst...).ToArray() 
29. } 

然而，確定相同的 LabelName 也是一個問題，因為 Label 本身就代表著 Name:Value，難不成我還要遍歷所有 label 才能確定嘛，這不就又成了全表掃描???

沒有什么問題是一個索引解決不了的，如果有，那就再增加一個索引。--- 魯迅。

只要我們保存 Label 的 Name 對應的 Value 列表的映射關系即可高效解決這個問題。

還是上面的 {vm=~"node1|node2", iface="eth0"} 查詢，第一步通過正則匹配確定匹配到 node1, node2，第二步匹配到 eth0，再將 LabelName 和 LabelValue 一拼裝，Label 就出來了，完事!

橋豆麻袋!還有一個精彩的正則匹配優化算法沒介紹。

fastRegexMatcher 是一種優化的正則匹配器，算法來自 Prometheus。

1. // 思路就是盡量先執行前綴匹配和后綴匹配 能不用正則就不用正則 
2. // 如 label 表達式為 {vm="node*"} 
3. // 而我們此時內存中有 vm=node1, vm=node2, vm=foo, vm=bar，那這個時候只需要前綴匹配就能直接把 vm=foo,vm=bar 給過濾了 
4. // 畢竟前綴匹配和后綴匹配的執行效率還是比正則高不少的 
5. type fastRegexMatcher struct { 
6.  re       *regexp.Regexp 
7.  prefix   string 
8.  suffix   string 
9.  contains string 
10. } 
11.  
12. func newFastRegexMatcher(v string) (*fastRegexMatcher, error) { 
13.  re, err := regexp.Compile("^(?:" + v + ")$") 
14.  if err != nil { 
15.   return nil, err 
16.  } 
17.  
18.  parsed, err := syntax.Parse(v, syntax.Perl) 
19.  if err != nil { 
20.   return nil, err 
21.  } 
22.  
23.  m := &fastRegexMatcher{ 
24.   re: re, 
25.  } 
26.  
27.  if parsed.Op == syntax.OpConcat { 
28.   m.prefix, m.suffix, m.contains = optimizeConcatRegex(parsed) 
29.  } 
30.  
31.  return m, nil 
32. } 
33.  
34. // optimizeConcatRegex returns literal prefix/suffix text that can be safely 
35. // checked against the label value before running the regexp matcher. 
36. func optimizeConcatRegex(r *syntax.Regexp) (prefix, suffix, contains string) { 
37.  sub := r.Sub 
38.  
39.  // We can safely remove begin and end text matchers respectively 
40.  // at the beginning and end of the regexp. 
41.  if len(sub) > 0 && sub[0].Op == syntax.OpBeginText { 
42.   sub = sub[1:] 
43.  } 
44.  if len(sub) > 0 && sub[len(sub)-1].Op == syntax.OpEndText { 
45.   sub = sub[:len(sub)-1] 
46.  } 
47.  
48.  if len(sub) == 0 { 
49.   return 
50.  } 
51.  
52.  // Given Prometheus regex matchers are always anchored to the begin/end 
53.  // of the text, if the first/last operations are literals, we can safely 
54.  // treat them as prefix/suffix. 
55.  if sub[0].Op == syntax.OpLiteral && (sub[0].Flags&syntax.FoldCase) == 0 { 
56.   prefix = string(sub[0].Rune) 
57.  } 
58.  if last := len(sub) - 1; sub[last].Op == syntax.OpLiteral && (sub[last].Flags&syntax.FoldCase) == 0 { 
59.   suffix = string(sub[last].Rune) 
60.  } 
61.  
62.  // If contains any literal which is not a prefix/suffix, we keep the 
63.  // 1st one. We do not keep the whole list of literals to simplify the 
64.  // fast path. 
65.  for i := 1; i < len(sub)-1; i++ { 
66.   if sub[i].Op == syntax.OpLiteral && (sub[i].Flags&syntax.FoldCase) == 0 { 
67.    contains = string(sub[i].Rune) 
68.    break 
69.   } 
70.  } 
71.  
72.  return 
73. } 
74.  
75. func (m *fastRegexMatcher) MatchString(s string) bool { 
76.  if m.prefix != "" && !strings.HasPrefix(s, m.prefix) { 
77.   return false 
78.  } 
79.  
80.  if m.suffix != "" && !strings.HasSuffix(s, m.suffix) { 
81.   return false 
82.  } 
83.  
84.  if m.contains != "" && !strings.Contains(s, m.contains) { 
85.   return false 
86.  } 
87.  return m.re.MatchString(s) 
88. } 

存儲布局

 

既然是數據庫，那么自然少不了數據持久化的特性。了解完索引的設計，再看看落到磁盤的存儲布局就很清晰了。先跑個示例程序寫入一些數據熱熱身。

1. package main 
2.  
3. import ( 
4.  "fmt" 
5.  "math/rand" 
6.  "strconv" 
7.  "time" 
8.  
9.  "github.com/chenjiandongx/mandodb" 
10.  "github.com/satori/go.uuid" 
11. ) 
12.  
13. // 模擬一些監控指標 
14. var metrics = []string{ 
15.  "cpu.busy", "cpu.load1", "cpu.load5", "cpu.load15", "cpu.iowait", 
16.  "disk.write.ops", "disk.read.ops", "disk.used", 
17.  "net.in.bytes", "net.out.bytes", "net.in.packages", "net.out.packages", 
18.  "mem.used", "mem.idle", "mem.used.bytes", "mem.total.bytes", 
19. } 
20.  
21. // 增加 Label 數量 
22. var uid1, uid2, uid3 []string 
23.  
24. func init() { 
25.  for i := 0; i < len(metrics); i++ { 
26.   uid1 = append(uid1, uuid.NewV4().String()) 
27.   uid2 = append(uid2, uuid.NewV4().String()) 
28.   uid3 = append(uid3, uuid.NewV4().String()) 
29.  } 
30. } 
31.  
32. func genPoints(ts int64, node, dc int) []*mandodb.Row { 
33.  points := make([]*mandodb.Row, 0) 
34.  for idx, metric := range metrics { 
35.   points = append(points, &mandodb.Row{ 
36.    Metric: metric, 
37.    Labels: []mandodb.Label{ 
38.     {Name: "node", Value: "vm" + strconv.Itoa(node)}, 
39.     {Name: "dc", Value: strconv.Itoa(dc)}, 
40.     {Name: "foo", Value: uid1[idx]}, 
41.     {Name: "bar", Value: uid2[idx]}, 
42.     {Name: "zoo", Value: uid3[idx]}, 
43.    }, 
44.    Point: mandodb.Point{Ts: ts, Value: float64(rand.Int31n(60))}, 
45.   }) 
46.  } 
47.  
48.  return points 
49. } 
50.  
51. func main() { 
52.  store := mandodb.OpenTSDB() 
53.  defer store.Close() 
54.  
55.  now := time.Now().Unix() - 36000 // 10h ago 
56.  
57.  for i := 0; i < 720; i++ { 
58.   for n := 0; n < 5; n++ { 
59.    for j := 0; j < 1024; j++ { 
60.     _ = store.InsertRows(genPoints(now, n, j)) 
61.    } 
62.   } 
63.  
64.   now += 60 //1min 
65.  } 
66.  
67.  fmt.Println("finished") 
68.  
69.  select {} 
70. } 

每個分塊保存在名字為 seg-${mints}-${maxts} 文件夾里，每個文件夾含有 data 和 meta.json 兩個文件。

• data: 存儲了一個 Segment 的所有數據，包括數據點和索引信息。
• meta.json: 描述了分塊的時間線數量，數據點數量以及該塊的數據時間跨度。

1. ?  tree -h seg-* 
2. seg-1627709713-1627716973 
3. ├── [ 28M]  data 
4. └── [ 110]  meta.json 
5. seg-1627716973-1627724233 
6. ├── [ 28M]  data 
7. └── [ 110]  meta.json 
8. seg-1627724233-1627731493 
9. ├── [ 28M]  data 
10. └── [ 110]  meta.json 
11. seg-1627731493-1627738753 
12. ├── [ 28M]  data 
13. └── [ 110]  meta.json 
14. seg-1627738753-1627746013 
15. ├── [ 28M]  data 
16. └── [ 110]  meta.json 
17.  
18. 0 directories, 10 files 
19.  
20. ?  cat seg-1627709713-1627716973/meta.json -p 
21. { 
22.     "seriesCount": 81920, 
23.     "dataPointsCount": 9912336, 
24.     "maxTs": 1627716973, 
25.     "minTs": 1627709713 
26. } 

存儲 8 萬條時間線共接近 1 千萬的數據點的數據塊占用磁盤 28M。實際上在寫入的時候，一條數據是這個樣子的。

1. {__name__="cpu.busy", node="vm0", dc="0", foo="bdac463d-8805-4cbe-bc9a-9bf495f87bab", bar="3689df1d-cbf3-4962-abea-6491861e62d2", zoo="9551010d-9726-4b3b-baf3-77e50655b950"} 1627710454 41 

這樣一條數據按照 JSON 格式進行網絡通信的話，大概是 200Byte，初略計算一下。

200 * 9912336 = 1982467200Byte = 1890M

可以選擇 ZSTD 或者 Snappy 算法進行二次壓縮(默認不開啟)。還是上面的示例代碼，不過在 TSDB 啟動的時候指定了壓縮算法。

ZstdBytesCompressor

1. func main() { 
2.  store := mandodb.OpenTSDB(mandodb.WithMetaBytesCompressorType(mandodb.ZstdBytesCompressor)) 
3.  defer store.Close() 
4.  // ... 
5. } 
6.  
7. // 壓縮效果 28M -> 25M 
8. ?  ll seg-1627711905-1627719165 
9. Permissions Size User          Date Modified Name 
10. .rwxr-xr-x   25M chenjiandongx  1 Aug 00:13  data 
11. .rwxr-xr-x   110 chenjiandongx  1 Aug 00:13  meta.json 

SnappyBytesCompressor

1. func main() { 
2.  store := mandodb.OpenTSDB(mandodb.WithMetaBytesCompressorType(mandodb.SnappyBytesCompressor)) 
3.  defer store.Close() 
4.  // ... 
5. } 
6.  
7. // 壓縮效果 28M -> 26M 
8. ?  ll seg-1627763918-1627771178 
9. Permissions Size User          Date Modified Name 
10. .rwxr-xr-x   26M chenjiandongx  1 Aug 14:39  data 
11. .rwxr-xr-x   110 chenjiandongx  1 Aug 14:39  meta.json 

多多少少還是有點效果的 ...

壓縮是有成本的，壓縮體積的同時會增大 CPU 開銷(mbp 可以煎雞蛋了)，減緩寫入速率。

敲黑板，接下來就要來好好講講 data 文件到底寫了什么東西。 data 存儲布局如下。

Figure: Segment Stroage

TOC 描述了 Data Block 和 Meta Block(Series Block + Labels Block)的體積，用于后面對 data 進行解析讀取。Data Block 存儲了每條時間線具體的數據點，時間線之間數據緊挨存儲。DataContent 就是使用 Gorilla 差值算法壓縮的 block。

Figure: Data Block

Labels Block 記錄了具體的 Label 值以及對應 Label 與哪些 Series 相關聯。

Figure: Labels Block

Series Block 記錄了每條時間線的元數據，字段解釋如下。

• SidLength: Sid 的長度。
• Sid: 時間線的唯一標識。
• StartOffset: 時間線數據塊在 Data Block 中的起始偏移。
• EndOffset: 時間線數據塊在 Data Block 中的終止偏移。
• LabelCount: 時間線包含的 Label 數量。
• Labels: 標簽在 Labels Block 中的序號(僅記錄序號，不記錄具體值)。
• Figure: Series Block

了解完設計，再看看 Meta Block 編碼和解編碼的代碼實現，binaryMetaSerializer 實現了 MetaSerializer 接口。

1. type MetaSerializer interface { 
2.  Marshal(Metadata) ([]byte, error) 
3.  Unmarshal([]byte, *Metadata) error 
4. } 

編碼 Metadata

1. const ( 
2.  endOfBlock uint16 = 0xffff 
3.  uint16Size        = 2 
4.  uint32Size        = 4 
5.  uint64Size        = 8 
6.  
7.  magic = "https://github.com/chenjiandongx/mandodb" 
8. ) 
9.  
10. func (s *binaryMetaSerializer) Marshal(meta Metadata) ([]byte, error) { 
11.  encf := newEncbuf() 
12.  
13.  // labels block 
14.  labelOrdered := make(map[string]int) 
15.  for idx, row := range meta.Labels { 
16.   labelOrdered[row.Name] = idx 
17.   encf.MarshalUint16(uint16(len(row.Name))) 
18.   encf.MarshalString(row.Name) 
19.   encf.MarshalUint32(uint32(len(row.Sids))) 
20.   encf.MarshalUint32(row.Sids...) 
21.  } 
22.  encf.MarshalUint16(endOfBlock) 
23.  
24.  // series block 
25.  for idx, series := range meta.Series { 
26.   encf.MarshalUint16(uint16(len(series.Sid))) 
27.   encf.MarshalString(series.Sid) 
28.   encf.MarshalUint64(series.StartOffset, series.EndOffset) 
29.  
30.   rl := meta.sidRelatedLabels[idx] 
31.   encf.MarshalUint32(uint32(rl.Len())) 
32.  
33.   lids := make([]uint32, 0, rl.Len()) 
34.   for _, lb := range rl { 
35.    lids = append(lids, uint32(labelOrdered[lb.MarshalName()])) 
36.   } 
37.  
38.   sort.Slice(lids, func(i, j int) bool { 
39.    return lids[i] < lids[j] 
40.   }) 
41.   encf.MarshalUint32(lids...) 
42.  } 
43.  encf.MarshalUint16(endOfBlock) 
44.  
45.  encf.MarshalUint64(uint64(meta.MinTs)) 
46.  encf.MarshalUint64(uint64(meta.MaxTs)) 
47.  encf.MarshalString(magic)   // <-- magic here 
48.  
49.  return ByteCompress(encf.Bytes()), nil 
50. } 

解碼 Metadata

1. func (s *binaryMetaSerializer) Unmarshal(data []byte, meta *Metadata) error { 
2.  data, err := ByteDecompress(data) 
3.  if err != nil { 
4.   return ErrInvalidSize 
5.  } 
6.  
7.  if len(data) < len(magic) { 
8.   return ErrInvalidSize 
9.  } 
10.  
11.  decf := newDecbuf() 
12.  // 檢驗數據完整性 
13.  if decf.UnmarshalString(data[len(data)-len(magic):]) != magic { 
14.   return ErrInvalidSize 
15.  } 
16.  
17.  // labels block 
18.  offset := 0 
19.  labels := make([]seriesWithLabel, 0) 
20.  for { 
21.   var labelName string 
22.   labelLen := decf.UnmarshalUint16(data[offset : offset+uint16Size]) 
23.   offset += uint16Size 
24.  
25.   if labelLen == endOfBlock { 
26.    break 
27.   } 
28.  
29.   labelName = decf.UnmarshalString(data[offset : offset+int(labelLen)]) 
30.   offset += int(labelLen) 
31.   sidCnt := decf.UnmarshalUint32(data[offset : offset+uint32Size]) 
32.   offset += uint32Size 
33.  
34.   sidLst := make([]uint32, sidCnt) 
35.   for i := 0; i < int(sidCnt); i++ { 
36.    sidLst[i] = decf.UnmarshalUint32(data[offset : offset+uint32Size]) 
37.    offset += uint32Size 
38.   } 
39.   labels = append(labels, seriesWithLabel{Name: labelName, Sids: sidLst}) 
40.  } 
41.  meta.Labels = labels 
42.  
43.  // series block 
44.  rows := make([]metaSeries, 0) 
45.  for { 
46.   series := metaSeries{} 
47.   sidLen := decf.UnmarshalUint16(data[offset : offset+uint16Size]) 
48.   offset += uint16Size 
49.  
50.   if sidLen == endOfBlock { 
51.    break 
52.   } 
53.  
54.   series.Sid = decf.UnmarshalString(data[offset : offset+int(sidLen)]) 
55.   offset += int(sidLen) 
56.   series.StartOffset = decf.UnmarshalUint64(data[offset : offset+uint64Size]) 
57.   offset += uint64Size 
58.   series.EndOffset = decf.UnmarshalUint64(data[offset : offset+uint64Size]) 
59.   offset += uint64Size 
60.   labelCnt := decf.UnmarshalUint32(data[offset : offset+uint32Size]) 
61.   offset += uint32Size 
62.  
63.   labelLst := make([]uint32, labelCnt) 
64.   for i := 0; i < int(labelCnt); i++ { 
65.    labelLst[i] = decf.UnmarshalUint32(data[offset : offset+uint32Size]) 
66.    offset += uint32Size 
67.   } 
68.   series.Labels = labelLst 
69.   rows = append(rows, series) 
70.  } 
71.  meta.Series = rows 
72.  
73.  meta.MinTs = int64(decf.UnmarshalUint64(data[offset : offset+uint64Size])) 
74.  offset += uint64Size 
75.  meta.MaxTs = int64(decf.UnmarshalUint64(data[offset : offset+uint64Size])) 
76.  offset += uint64Size 
77.  
78.  return decf.Err() 
79. } 

至此，對 mandodb 的索引和存儲整體設計是不是就了然于胸。

【編輯推薦】https://mp.weixin.qq.com/s/PqVHjGLLu5dXxjHubPbXYA