Prometheus時序數據庫-數據的插入

作者：alchemystarlzy 2021-03-08 10:18:55

筆者詳細的闡述了Prometheus時序數據庫在內存和磁盤中的存儲結構。有了前面的鋪墊，筆者就可以在本篇文章闡述下數據的插入過程。

前言

在之前的文章里，筆者詳細的闡述了Prometheus時序數據庫在內存和磁盤中的存儲結構。有了前面的鋪墊，筆者就可以在本篇文章闡述下數據的插入過程。

監控數據的插入

在這里，筆者并不會去討論Promtheus向各個Endpoint抓取數據的過程。而是僅僅圍繞著數據是如何插入Prometheus的過程做下闡述。對應方法:

func (a *headAppender) Add(lset labels.Labels, t int64, v float64) (uint64, error) { 
    ...... 
    // 如果lset對應的series沒有，則建一個。同時把新建的series放入倒排Posting映射里面 
    s, created := a.head.getOrCreate(lset.Hash(), lset)  
    if created { // 如果新創建了一個，則將新建的也放到a.series里面 
        a.series = append(a.series, record.RefSeries{ 
            Ref:    s.ref, 
            Labels: lset, 
        }) 
    } 
    return s.ref, a.AddFast(s.ref, t, v) 
}

我們就以下面的add函數調用為例:

app.Add(labels.FromStrings("foo", "bar"), 0, 0)

首先是getOrCreate,顧名思義，不存在則創建一個。創建的過程包含了seriesHashMap/Postings(倒排索引)/LabelIndex的維護。如下圖所示:

然后是AddFast方法

func (a *headAppender) AddFast(ref uint64, t int64, v float64) error{ 
        // 拿出對應的memSeries 
        s := a.head.series.getByID(ref) 
        ...... 
        // 設置為等待提交狀態 
        s.pendingCommit=true 
        ...... 
        // 為了事務概念，放入temp存儲，等待真正commit時候再寫入memSeries 
        a.samples = append(a.samples, record.RefSample{Ref: ref,T:   t,V:   v,}) 
        //  
}

Prometheus在add數據點的時候并沒有直接add到memSeries(也就是query所用到的結構體里),而是加入到一個臨時的samples切片里面。同時還將這個數據點對應的memSeries同步增加到另一個sampleSeries里面。

事務可見性

為什么要這么做呢?就是為了實現commit語義，只有commit過后數據才可見(能被查詢到)。否則，無法見到這些數據。而commit的動作主要就是WAL(Write Ahead Log)以及將headerAppender.samples數據寫到其對應的memSeries中。這樣，查詢就可見這些數據了，如下圖所示:

WAL

由于Prometheus最近的數據是保存在內存里面的，未防止服務器宕機丟失數據。其在commit之前先寫了日志WAL。等服務重啟的時候，再從WAL日志里面獲取信息并重放。

為了性能，Prometheus了另一個goroutine去做文件的sync操作，所以并不能保證WAL不丟。進而也不能保證監控數據完全不丟。這點也是監控業務的特性決定的。

寫入代碼為:

commit() 
|=> 
func (a *headAppender) log() error { 
    ...... 
    // 往WAL寫入對應的series信息 
    if len(a.series) > 0 { 
        rec = enc.Series(a.series, buf) 
        buf = rec[:0] 
 
        if err := a.head.wal.Log(rec); err != nil { 
            return errors.Wrap(err, "log series") 
        } 
    } 
    ...... 
    // 往WAL寫入真正的samples 
    if len(a.samples) > 0 { 
        rec = enc.Samples(a.samples, buf) 
        buf = rec[:0] 
 
        if err := a.head.wal.Log(rec); err != nil { 
            return errors.Wrap(err, "log samples") 
        } 
    } 
}

對應的WAL日志格式為:

Series records

┌────────────────────────────────────────────┐ 
│ 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

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