InfluxDB详解之TSM存储引擎解析（二）

tsdb.Store
    -- map[string]*tsdb.DatabaseIndex
        -- map[string]*tsdb.Measurement
        -- map[string]*tsdb.Series
    -- map[uint64]*tsdb.Shard
        -- Engine // 抽象接口，可插拔设计，目前是 tsm1 存储引擎
            -- tsm1.WAL
            -- tsm1.Cache
            -- tsm1.Compactor
            -- tsm1.FileStore

type Store struct {
    path string         // 数据库文件在磁盘上的存储路径

    // 数据库索引文件，key 为数据库名
    databaseIndexes map[string]*DatabaseIndex

    // 所有 shards 的索引，key 为其 shard ID
    shards map[uint64]*Shard
}

type Shard struct {
    index   *DatabaseIndex      // 所在数据库的索引对象
    path    string              // shard 在磁盘上的路径
    walPath string              // 对应的 wal 文件所在目录
    id      uint64              // shard ID，就是在磁盘上的文件名
    database        string      // 所在数据库名
    retentionPolicy string      // 对应存储策略名
    engine  Engine              // 存储引擎
}

type Engine interface {
    LoadMetadataIndex(shardID uint64, index *DatabaseIndex) error

    Backup(w io.Writer, basePath string, since time.Time) error
    Restore(r io.Reader, basePath string) error

    CreateIterator(opt influxql.IteratorOptions) (influxql.Iterator, error)
    WritePoints(points []models.Point) error
    ContainsSeries(keys []string) (map[string]bool, error)
    DeleteSeries(keys []string) error
    DeleteSeriesRange(keys []string, min, max int64) error
    DeleteMeasurement(name string, seriesKeys []string) error
    SeriesCount() (n int, err error)
    MeasurementFields(measurement string) *MeasurementFields
    CreateSnapshot() (string, error)
}

type Engine struct {
    path      string
    index             *tsdb.DatabaseIndex                 // 数据库索引信息，目前没和存储引擎放在一起，看起来后续会更改设计作为存储引擎的一部分
    measurementFields map[string]*tsdb.MeasurementFields  // 所有 measurement 对应的 fields 对象
    WAL            *WAL                 // WAL 文件对象
    Cache          *Cache               // WAL 文件在内存中的缓存
    Compactor      *Compactor           // 压缩合并管理对象
    CompactionPlan CompactionPlanner
    FileStore      *FileStore           // 数据文件对象
    MaxPointsPerBlock int               // 每个 block 中最多存储的 Points 数量

    // Cache 超过指定大小后内容会被写入一个新的 TSM 文件
    CacheFlushMemorySizeThreshold uint64

    // Cache 超过多长时间后还没有数据写入，会将内容写入新的 TSM 文件
    CacheFlushWriteColdDuration time.Duration
}

func (h *Handler) serveWrite(w http.ResponseWriter, r *http.Request, user *meta.UserInfo) {
    ...
    err := h.PointsWriter.WritePoints(database, r.URL.Query().Get("rp"), consistency, points)
    ...
}

func (w *PointsWriter) WritePoints(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error {
    ...
    // 将要写入的 Points 按照时间划分到要写入的 shard，返回一个 Point 和 shard 之间的映射关系
    shardMappings, err := w.MapShards(&WritePointsRequest{Database: database, RetentionPolicy: retentionPolicy, Points: points})
    ...
    
    // 每一个 shard 都有一个独立的协程负责写入，只要有一个出错，就立即返回错误信息
    ch := make(chan error, len(shardMappings.Points))
    for shardID, points := range shardMappings.Points {
        go func(shard *meta.ShardInfo, database, retentionPolicy string, points []models.Point) {
            ch <- w.writeToShard(shard, database, retentionPolicy, points)
        }(shardMappings.Shards[shardID], database, retentionPolicy, points)
    }
}

// PointsWirte.writeToShard -> TSDBStore.WriteToShard
func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPolicy string, points []models.Point) error {
    ...
    err := w.TSDBStore.WriteToShard(shard.ID, points)
    ...
}

// TSDBStore.WriteToShard -> shard.WritePoints
func (s *Store) WriteToShard(shardID uint64, points []models.Point) error {
    ...
    sh, ok := s.shards[shardID]
    if !ok {
        return ErrShardNotFound
    }
    return sh.WritePoints(points)
}

// shard.WritePoints -> engine.WritePoints
func (s *Shard) WritePoints(points []models.Point) error {
    ...
    err := s.engine.WritePoints(points)
    ...
}

func (e *Engine) WritePoints(points []models.Point) error {
    values := map[string][]Value{}
    for _, p := range points {
        // 一条插入语句中一个 series 对应的多个 value 会被拆分出来，形成多条数据
        for k, v := range p.Fields() {
            // 这里的 key 是 seriesKey + 分隔符 + filedName
            key := string(p.Key()) + keyFieldSeparator + k
            values[key] = append(values[key], NewValue(p.Time().UnixNano(), v))
        }    
    }    
    e.mu.RLock()
    defer e.mu.RUnlock()

    // 向 cache 中写入 value 数据，如果超过了内存阀值上限，返回错误
    err := e.Cache.WriteMulti(values)
    if err != nil {
        return err
    }    

    // 将数据写入 wal 文件中
    _, err = e.WAL.WritePoints(values)
    return err
}

type DatabaseIndex struct {
    measurements map[string]*Measurement // 该数据库下所有 Measurement 对象
    series       map[string]*Series      // 所有 Series 对象，SeriesKey = measurement + tags
    name string // 数据库名
}

type Measurement struct {
    Name       string `json:"name,omitempty"`
    fieldNames map[string]struct{}      // 此 measurement 中的所有 filedNames

    // 内存中的索引信息
    // id 以及其对应的 series 信息，主要是为了在 seriesByTagKeyValue 中存储Id节约内存
    seriesByID          map[uint64]*Series              // lookup table for series by their id

    // 根据 tagk 和 tagv 的双重索引，保存排好序的 SeriesID 数组
    // 这个 map 用于在查询操作时，可以根据 tags 来快速过滤出要查询的所有 SeriesID，之后根据 SeriesKey 以及时间范围从文件中读取相应内容
    seriesByTagKeyValue map[string]map[string]SeriesIDs // map from tag key to value to sorted set of series ids

    // 此 measurement 中所有 series 的 id，按照 id 排序
    seriesIDs           SeriesIDs                       // sorted list of series IDs in this measurement
}

type Series struct {
    Key         string              // series key
    Tags        map[string]string   // tags
    id          uint64              // id
    measurement *Measurement        // 所属 measurement
    // 在哪些 shard 中存在
    shardIDs    map[uint64]bool // shards that have this series defined
}

SHOW MEASUREMENTS
SHOW TAG KEYS FROM "measurement_name"
SHOW TAG VALUES FROM "measurement_name" WITH KEY = "tag_key"

SHOW TAG VALUES FROM "cpu_usage" WITH KEY = "host"

`SELECT value FROM "cpu_usage" WHERE host='server01' AND time > now() - 1h`

type indirectIndex struct {
    b []byte                // 下层详细索引的字节流
    offsets []int32         // 偏移量数组，记录了一个 key 在 b 中的偏移量
    minKey, maxKey string   
    minTime, maxTime int64  // 此文件中的最小时间和最大时间，根据这个可以快速判断要查询的数据在此文件中是否存在，是否有必要读取这个文件
    tombstones map[string][]TimeRange   // 用于记录哪些 key 在指定范围内的数据是已经被删除的
}

type indexEntries struct {
    Type    byte 
    entries []IndexEntry
}

type IndexEntry struct {
    // 一个 block 中的 point 都在这个最小和最大的时间范围内
    MinTime, MaxTime int64

    // block 在 tsm 文件中偏移量
    Offset int64

    // block 的具体大小
    Size uint32
}

type cursor interface {
    next() (t int64, v interface{})
}

type floatCursor interface {
    cursor
    nextFloat() (t int64, v float64)
}

// 底层主要是 KeyCursor，每次读取一个 block 的数据
type floatAscendingCursor struct {
    // 内存中的 value 对象
    cache struct {
        values Values
        pos    int
    }

    tsm struct {
        tdec      TimeDecoder   // 时间序列化对象
        vdec      FloatDecoder  // value 序列化对象
        buf       []FloatValue
        values    []FloatValue  // 从 tsm 文件中读取到的 FloatValue 的缓存
        pos       int
        keyCursor *KeyCursor
    }
}

type Compactor struct {
    Dir  string
    Size int
    FileStore interface {
        NextGeneration() int
    }
}

type CompactionPlanner interface {
    Plan(lastWrite time.Time) []CompactionGroup
    PlanLevel(level int) []CompactionGroup
    PlanOptimize() []CompactionGroup
}

// 默认的压缩合并计划
type DefaultPlanner struct {
    FileStore interface {
        Stats() []FileStat
        LastModified() time.Time
        BlockCount(path string, idx int) int
    }

    // 如果一个 shard 对应的 wal 文件超过指定时间一直没有数据写入
    // 存储引擎将会将此 shard 中的 tsm 文件进行一次全量压缩合并
    CompactFullWriteColdDuration time.Duration

    // 如果为 true，表示此 shard 中的 tsm 文件要么只有一个，要么已经处于单个文件最大值
    lastPlanCompactedFull bool

    // lastPlanCheck is the last time Plan was called
    lastPlanCheck time.Time
}

func (e *Engine) SetCompactionsEnabled(enabled bool) {
    if enabled {
        ...
        e.done = make(chan struct{})
        // 启用压缩合并功能
        e.Compactor.Open()

        // 启用另外的协程去定期检测是否需要进行压缩合并
        e.wg.Add(5)
        // 将 cache 中的内容刷新到磁盘上的新的 tsm 文件中
        go e.compactCache()
        // 下面是定期合并不同 level 的 tsm 文件
        go e.compactTSMFull()
        go e.compactTSMLevel(true, 1)
        go e.compactTSMLevel(true, 2)
        go e.compactTSMLevel(false, 3)
    }
    ...
}

func (e *Engine) compactCache() {
    defer e.wg.Done()
    // 每秒钟检查一次
    for {
        select {
        case <-e.done:
            return

        default:
            // 更新缓存据上一次快照时间的间隔时间
            e.Cache.UpdateAge()
            if e.ShouldCompactCache(e.WAL.LastWriteTime()) {
                start := time.Now()
                err := e.WriteSnapshot()
                ...
            }
        }
        time.Sleep(time.Second)
    }
}