前文我们介绍了 VictorialMetrics 中是如何接收和传输数据的,接下来我们来分析下当vmstorage接收到数据后是如何保存监控指标的。 现在我们使用csv来导入一行指标数据,直接使用下面的请求即可:
curl -d "GOOG,1.23,4.56,NYSE" 'http://127.0.0.1:8480/insert/0/prometheus/api/v1/import/csv?format=2:metric:ask,3:metric:bid,1:label:ticker,4:label:market'
执行上面的请求后,在vmstorage组件下面会收到如下所示的一些日志信息: 同时在数据目录vmstorage-data下面也多了一个cache目录,而且data下面的small目录和indexdb目录下面也生成了一些文件,这些文件就是用来存储指标数据的。 接下来我们就来仔细分析下这些文件是干什么的,以及这些文件的存储格式是怎样的。 要想弄明白vmstorage是如何去存储数据的,首先我们要先弄明白几个概念。
存储格式
下图是 VictoriaMetrics 支持的 Prometheus 协议的一个写入示例。 VM 在收到写入请求时,会对请求中包含的时序数据做转换处理。首先根据包含 metric 和 labels 的 MetricName 生成一个唯一标识TSID,然后metric(指标名称__name__) + labels + TSID作为索引 index,TSID + timestamp + value作为数据 data,最后索引 index 和数据 data 分别进行存储和检索。 因此 VM 的数据整体上分成索引和数据两个部分,因此文件格式整体上会有两个部分,其中索引部分主要是用于支持按照 label 或者 tag 进行多维检索,数据存储时,先将数据按 TSID 进行分组,然后每个 TSID 包含的数据点各自使用列式压缩存储。
TSID
VictoriaMetrics 的MetricName的结构如下所示,包含MetricGroup(指标名称 __name__)和 Tag 数组,其中,Tags 是可选的,每个 Tag 由 Key 和 Value 等字节数组构成。 为了规范,Tags 必须按标签 Key 排序,使用 sortTags 方法。 VictoriaMetrics 的 TSID 的结构如下所示,包含MetricGroupID、JobID、InstanceID、MetricID等几个字段,其中除了 MetricID 外,其他字段都是可选的。这个几个 ID 的生成方法如下:
-
MetricGroupID是根据MetricName中的MetricGroup使用xxhash的 sum64 算法生成。
JobID和InstanceID分别由MetricName中的第一个 tag 和第二个 tag 使用xxhash的 sum64 算法生成。为什么使用第一个 tag 和第二个 tag?这是因为 VictoriaMetrics 在写入时,将写入请求中的 JobID 和 InstanceID 放在了 Tag 数组的第一个和第二个位置。
MetricID,使用 VictoriaMetrics 进程启动时的系统纳秒时间戳自增生成。
// lib/storage/tsid.go // TSID 是一个时间序列的唯一 ID,实际上就是唯一标识一个时间序列的结构体。 // // 时间序列会根据 TSID 进行排序。 // // 除了 MetricID 之外其他属性都是可选的。 它们的存在仅仅是为了更好地对相关指标进行分组。 // 如果它们的含义与它们的命名不同,那也没关系。 type TSID struct { AccountID uint32 ProjectID uint32 // 下面分析的时候可以暂时忽略这两个属性,用于多租户标识的属性 // MetricGroupID(指标组ID)对于指定的(AccountID, ProjectID)必须是唯一的。 // // Metric Group 包含具有相同名称的指标,例如 “memory_usage”、“http_requests”,但具有不同的标签。 // 例如,下面的这些指标属于 memory_usage 这个指标组: // // memory_usage{datacenter="foo1", job="bar1", instance="baz1:1234"} // memory_usage{datacenter="foo1", job="bar1", instance="baz2:1234"} // memory_usage{datacenter="foo1", job="bar2", instance="baz1:1234"} // memory_usage{datacenter="foo2", job="bar1", instance="baz2:1234"} MetricGroupID uint64 // JobID 是给定项目的单个作业(又名服务)的 ID。 // // JobID 对于指定的(AccountID, ProjectID)必须是唯一的。 // // 一个 Job 任务可能由多个实例组成。 // See https://prometheus.io/docs/concepts/jobs_instances/ for details. JobID uint32 // InstanceID 是实例(进程)ID,对于特定的(AccountID, ProjectID)必须是唯一的。 InstanceID uint32 // MetricID 是指标(时间序列)的唯一ID。 // // 其他所有的 TSID 字段都可以通过 MetricID 获取。 MetricID uint64 }
因为 TSID 中除了 MetricID 外,其他字段都是可选的,因此 TSID 中可以始终作为有效信息的只有 MetricID,因此 VictoriaMetrics 的在构建 tag 到 TSID 的字典过程中,是直接存储的 tag 到 MetricID 的字典。 以写入http_requests_total{status=”200″, method=”GET”}为例,则 MetricName 为http_requests_total{status=”200″, method=”GET”},假设生成的 TSID 为{metricGroupID=0, jobID=0, instanceID=0, metricID=51106185174286},则 VictoriaMetrics 在写入时就构建了如下几种类型的索引 item,其他类型的索引 item 是在后台或者查询时构建的。
-
metricName -> TSID, 即http_requests_total{status=”200″, method=”GET”} -> {metricGroupID=0, jobID=0, instanceID=0, metricID=51106185174286}。
metricID -> metricName,即51106185174286 -> http_requests_total{status=”200″, method=”GET”}。
metricID -> TSID,即51106185174286 -> {metricGroupID=0, jobID=0, instanceID=0, metricID=51106185174286}。
tag -> metricID,即status=”200″ -> 51106185174286、method=”GET” -> 51106185174286、”__name__” = http_requests_total -> 51106185174286(其实还有一个联合索引)。
有了这些索引的 item 后,就可以支持基于 tag 的多维检索了,在当给定查询条件http_requests_total{status=”200″}时,VictoriaMetrics 先根据给定的 tag 条件,找出每个 tag 的 metricID 列表,然后计算所有 tag 的 metricID 列表的交集,然后根据交集中的 metricID,再到索引文件中检索出 TSID,根据 TSID 就可以到数据文件中查询数据了,在返回结果之前,再根据 TSID 中的 metricID,到索引文件中检索出对应的写入时的原始 MetircName。 但是由于 VictoriaMetrics 的 tag 到 metricID 的字典,没有将相同 tag 的所有 metricID 放在一起存储,在检索时,一个 tag 可能需要查询多次才能得到完整的 metricID 列表。另外查询出 metricID 后,还要再到索引文件中去检索 TSID 才能去数据文件查询数据,又增加了一次 IO 开销。这样来看的话,VictoriaMetrics 的索引文件在检索时,如果命中的时间线比较多的情况下,其 IO 开销会比较大,查询延迟也会比较高。 这里我们了解了TSID这个非常重要的概念,还有几个结构体需要我们了解下,比如rawRow表示一个原始的时间序列行,MetricRow表示插入到存储中的指标数据:
// lib/storage/raw_row.go // rawRow 表示一个原始的时间序列行 type rawRow struct { TSID TSID // 时间序列ID Timestamp int64 // 时间戳 Value float64 // 给定时间戳的时间序列值 // PrecisionBits是要存储的值中的有效位数,可能值为 [1..64] // 1 表示最大. 50% error, 2 - 25%, 3 - 12.5%, 64 没有错误, i.e. // 存储的值不会丢失精度 PrecisionBits uint8 } // libe/storage/storage.go // MetricRow 插入到存储中的指标 type MetricRow struct { // MetricNameRaw 包含原始的指标名称,必须使用 metricne.UnmarshalRaw 对其进行解码。 MetricNameRaw []byte Timestamp int64 Value float64 }
插入指标
有了上面几个概念的认识,现在我们回过头再去看下vmstorage中对vminsert请求的处理:
// app/vmstorage/transport/server.go func (s *Server) processVMInsertConn(bc *handshake.BufferedConn) error { return clusternative.ParseStream(bc, func(rows []storage.MetricRow) error { vminsertMetricsRead.Add(len(rows)) return s.storage.AddRows(rows, uint8(*precisionBits)) }, s.storage.IsReadOnly) }
当vmstorage节点接收到数据后,最后会通过回调执行s.storage.AddRows(rows, uint8(*precisionBits)),该函数将数据添加到底层存储去:
// lib/storage/storage.go // AddRows 添加 mrs 集合到存储 s func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error { if len(mrs) == 0 { return nil } // 限制可能向存储添加行的并发 goroutine 数量 // 当太多的 goroutine 调用 AddRows 时,这应该可以防止内存不足错误和 CPU 抖动。 select { // 如果写入 channel 成功,说明并发小于 CPU 最大核数,然后就可以走插入逻辑 // 如果没写入成功(也就是满了),则执行default case case addRowsConcurrencyCh <- struct{}{}: default: // 如果插入 channel 失败,说明某个 insert 操作的协程被阻塞,这时需要通知 select 协程去让出。 atomic.AddUint64(&s.addRowsConcurrencyLimitReached, 1) t := timerpool.Get(addRowsTimeout) // 获取一个30s超时的timer // 数据摄取优先级高于并发搜索 // pacelimiter(步长限制器)中有个原子累加的变量,表示有多少个 insert 操作在等待 // 走到这里证明有一个 insert 操作被阻塞了,调用 Inc,表示需要(Search操作)等待 storagepacelimiter.Search.Inc() select { // 写入不成功或者还未超时就会阻塞在这里了 // 在超时的30s时间内,尝试去写入 channel 队列 case addRowsConcurrencyCh <- struct{}{}: timerpool.Put(t) // 把 timer 放回对象池,减少 GC // 可以成功写入 channel 了,那么可以执行 insert 操作了,则执行限制器的 Dec 操作,减一 storagepacelimiter.Search.Dec() // 当限制器的等待数量为0的时候,会调用 cond.Broadcast() 去通知 select 协程开始工作。 case <-t.C: // 到30s超时时间了 // 把 timer 放回对象池,减少 GC timerpool.Put(t) // 超时了那么当前的 insert 就报错了,等待的数量就可以减一了 storagepacelimiter.Search.Dec() atomic.AddUint64(&s.addRowsConcurrencyLimitTimeout, 1) // 记录下超时次数 atomic.AddUint64(&s.addRowsConcurrencyDroppedRows, uint64(len(mrs))) // 记录没有被插入成功的 mr 数量 // 等待了30秒仍然没有CPU资源,只能报错 return fmt.Errorf("cannot add %d rows to storage in %s, since it is overloaded with %d concurrent writers; add more CPUs or reduce load", len(mrs), addRowsTimeout, cap(addRowsConcurrencyCh)) } } // 下面是插入逻辑 // 一次插入不要太大 var firstErr error ic := getMetricRowsInsertCtx() maxBlockLen := len(ic.rrs) for len(mrs) > 0 { mrsBlock := mrs // 如果要插入的 mrs 超过了最大长度 if len(mrs) > maxBlockLen { // 则先插入最大长度的 mrs mrsBlock = mrs[:maxBlockLen] // 剩下的 mrs 下次循环去处理 mrs = mrs[maxBlockLen:] } else { mrs = nil } // 执行真正的 add 操作 if err := s.add(ic.rrs, ic.tmpMrs, mrsBlock, precisionBits); err != nil { if firstErr == nil { firstErr = err } continue } // 记录下插入成功的 mrs 数量 atomic.AddUint64(&rowsAddedTotal, uint64(len(mrsBlock))) } // 放回对象池 putMetricRowsInsertCtx(ic) <-addRowsConcurrencyCh // insert 逻辑执行完成后,出队 return firstErr }
该函数的实现非常经典,会限制可能向存储添加数据的并发 goroutine 数量,当太多的 goroutine 调用 AddRows 时,可以防止内存不足错误和 CPU 抖动。这里实现了插入比查询更高的优先级,当资源不足时,查询操作会挂起让出资源给到插入操作使用。
获取 TSID
真正实现添加数据是下面的add函数,其中rawRow是原始的时序数据行,MetricRow是要插入到存储中的行数据,该函数的核心就是要生成指标序列的 TSID 数据,如下所示:
// lib/storage/storage.go func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, precisionBits uint8) error { // 当前使用的索引 idb := s.idb() j := 0 var ( // 这些变量用于加速同一 metricName 的多个相邻行的批量导入。 prevTSID TSID prevMetricNameRaw []byte ) var pmrs *pendingMetricRows // 获取该数据块的最小时间和最大时间 minTimestamp, maxTimestamp := s.tb.getMinMaxTimestamps() // 带有第几代索引信息的 TSID 对象 var genTSID generationTSID // 只返回第一个错误,因为它返回所有错误没有意义 var firstWarn error // 循环数据行,其实就是填充 rawRow 中的 TSID 数据 for i := range mrs { mr := &mrs[i] if math.IsNaN(mr.Value) { // 值为 NaN if !decimal.IsStaleNaN(mr.Value) { // 跳过 Prometheus staleness 标记以外的 NaN // 因为底层编码不知道如何使用它们。 continue } } // 如果指标的时间戳小于最小的时间戳 // 则跳过保留期外时间戳过小的行 if mr.Timestamp < minTimestamp { ...... continue } // 同样跳过超过最大时间戳的数据 if mr.Timestamp > maxTimestamp { ...... continue } dstMrs[j] = mr r := &rows[j] j++ r.Timestamp = mr.Timestamp r.Value = mr.Value r.PrecisionBits = precisionBits // 快速路径 - 当前 mr 包含与前一 mr 相同的指标名称,因此它包含相同的 TSID。 if string(mr.MetricNameRaw) == string(prevMetricNameRaw) { // 当许多行包含相同的 MetricNameRaw 时,应在批量导入时触发此路径。 r.TSID = prevTSID continue } // 判断 TSID 是否在缓存中(命中缓存) if s.getTSIDFromCache(&genTSID, mr.MetricNameRaw) { r.TSID = genTSID.TSID // 跳过该行,因为已超出唯一序列数的限制。 if s.isSeriesCardinalityExceeded(r.TSID.MetricID, mr.MetricNameRaw) { j-- continue } // 快速路径 - 给定 MetricNameRaw 的 TSID 已在缓存中找到,并且未删除。 // 不需要检查 r.TSID.MetricID 是否已删除,因为 tsidCache 不包含已删除时间序列的 MetricName -> TSID 条目,可以查看 Storage.DeleteMetrics 的代码 prevTSID = r.TSID // 设置前一个 TSID 的值 prevMetricNameRaw = mr.MetricNameRaw // 设置前一个 MetricNameRaw 的值 // 找到的TSID不是当前代的索引(来自上一代缓存下来的索引) if genTSID.generation != idb.generation { // 索引需要尝试使用该 TSID 重新填充当前代的索引数据 // https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1401 created, err := idb.maybeCreateIndexes(&genTSID.TSID, mr.MetricNameRaw) if err != nil { return fmt.Errorf("cannot create indexes in the current indexdb: %w", err) } if created { // 如果填充成功,则将当前的 TSID 设置为当前代索引 genTSID.generation = idb.generation // 重新将该 TSID -> MetricNameRaw 数据放回缓存,方便后面的序列处理 s.putTSIDToCache(&genTSID, mr.MetricNameRaw) } } continue } // 慢速路径 - 缓存中缺少TSID // 在下面的循环中推迟搜索 j-- if pmrs == nil { // 初始化 pendingMetricRows pmrs = getPendingMetricRows() } // 将 mr 数据添加到 pendingMetricRows 中去待处理 if err := pmrs.addRow(mr); err != nil { // 错误时不要停止添加数据 - 只需跳过无效行即可。 // 这保证了无效行不会阻止将有效行添加到存储中去。 if firstWarn == nil { firstWarn = err } continue } } // 有指标的 TSID 没有在缓存中(上面的慢速路径) if pmrs != nil { // 按指标名称对 pendingMetricRows 进行排序,以便通过下面循环中的 “is” 加快搜索速度。 pendingMetricRows := pmrs.pmrs sort.Slice(pendingMetricRows, func(i, j int) bool { return string(pendingMetricRows[i].MetricName) < string(pendingMetricRows[j].MetricName) }) // is := idb.getIndexSearch(0, 0, noDeadline) prevMetricNameRaw = nil // 接收前一个 MetricNameRaw var slowInsertsCount uint64 for i := range pendingMetricRows { pmr := &pendingMetricRows[i] mr := pmr.mr // MetricRaw dstMrs[j] = mr r := &rows[j] j++ r.Timestamp = mr.Timestamp r.Value = mr.Value r.PrecisionBits = precisionBits // 快速路径 - 当前 mr 包含与前一个 mr 相同的指标名称,因此它包含相同的 TSID。 if string(mr.MetricNameRaw) == string(prevMetricNameRaw) { // 当许多行包含相同的 MetricNameRaw 时,在批量导入时会触发该路径。 r.TSID = prevTSID if s.isSeriesCardinalityExceeded(r.TSID.MetricID, mr.MetricNameRaw) { // 跳过该行,因为已超出唯一序列数的限制 j-- continue } continue } // 慢速路径 slowInsertsCount++ // 记录慢插入次数 // 通过 MetricName 去获取(没有就创建)TSID 数据 if err := is.GetOrCreateTSIDByName(&r.TSID, pmr.MetricName); err != nil { if firstWarn == nil { firstWarn = fmt.Errorf("cannot obtain or create TSID for MetricName %q: %w", pmr.MetricName, err) } j-- continue } // 设置 genTSID 为当前生成的 TSID genTSID.generation = idb.generation genTSID.TSID = r.TSID // 返回缓存 s.putTSIDToCache(&genTSID, mr.MetricNameRaw) // 缓存当前的 TSID 和 MetricNameRaw,方便下一条序列快速处理 prevTSID = r.TSID prevMetricNameRaw = mr.MetricNameRaw if s.isSeriesCardinalityExceeded(r.TSID.MetricID, mr.MetricNameRaw) { // 跳过该行,因为已超出唯一序列数的限制 j-- continue } } // 回收对象 idb.putIndexSearch(is) putPendingMetricRows(pmrs) atomic.AddUint64(&s.slowRowInserts, slowInsertsCount) } // 提示错误信息 if firstWarn != nil { logger.WithThrottler("storageAddRows", 5*time.Second).Warnf("warn occurred during rows addition: %s", firstWarn) } dstMrs = dstMrs[:j] rows = rows[:j] // TSID 填充完成,可以插入数据了 var firstError error if err := s.tb.AddRows(rows); err != nil { firstError = fmt.Errorf("cannot add rows to table: %w", err) } if err := s.updatePerDateData(rows, dstMrs); err != nil && firstError == nil { firstError = fmt.Errorf("cannot update per-date data: %w", err) } if firstError != nil { return fmt.Errorf("error occurred during rows addition: %w", firstError) } return nil }
首先循环数据,把时间戳过小或过大的都过滤掉,然后就是想办法尽可能快地获取到指标的 TSID:
-
快速路径- 当前 MetricRow 包含与前一 MetricRow 相同的指标名称,因此它们具有相同的 TSID,所以直接将当前对象的 TSID 设置成前一个 TSID,这是最快的方式。
如果和前一个指标名称不一样,则去查看 genTSID 是否在缓存中(命中缓存)。
-
如果命中缓存则 genTSID 中的 TSID 就是我们需要的,同时也将其设置为前一个 prevTSID。如果该 TSID 不是当代的索引(来自上一代缓存下来的索引),则需要尝试使用该 TSID 重新填充当代的索引数据,这和索引轮换有关,后面会详细说明。
如果没有命中缓存,则属于慢速路径,将当前数据添加到pendingMetricRows中去待处理。
-
循环了所有指标数据后,接下来需要处理pendingMetricRows中的数据,也就是缓存中没有对应的 TSID,此时就需要我们去生成对应的 TSID 数据。
快速路径- 同样是当前 MetricRow 与前一个 MetricRow 的指标名称相同,因此它包含相同的 TSID,直接设置成前一个 TSID 即可。
慢速路径- 走到这个分支则只能去创建 TSID 了,通过 MetricName 去获取(没有就创建)TSID 数据,也就是上面GetOrCreateTSIDByName函数。获取后记得放到缓存中去。
上面费了很大的功夫就是为了获取时间序列对应的 TSID 数据的,这也是插入数据过程中最可能出现慢插入的地方,因为该过程涉及到索引,比较耗时间,如果你插入的数据出现大量的高基数序列(比如包含一些随机生成的 ID 作为标签),则会大大降低vmstorage的插入性能。 我们可以去查看下GetOrCreateTSIDByName函数的实现。
// lib/storage/index_db.go // GetOrCreateTSIDByName 使用指定 metricName 的 TSID 填充 dst。 func (is *indexSearch) GetOrCreateTSIDByName(dst *TSID, metricName []byte) error { // hack:在多次连续未命中后跳过 TSID 的搜索 // 这将提高大批量新时间序列的插入性能。 if is.tsidByNameMisses < 100 { err := is.getTSIDByMetricName(dst, metricName) if err == nil { is.tsidByNameMisses = 0 return nil } if err != io.EOF { return fmt.Errorf("cannot search TSID by MetricName %q: %w", metricName, err) } is.tsidByNameMisses++ } else { is.tsidByNameSkips++ if is.tsidByNameSkips > 10000 { is.tsidByNameSkips = 0 is.tsidByNameMisses = 0 } } // 找不到给定名称的 TSID,创建它。 // 如果 mn 的重复 TSID 是由并发 goroutines 创建的,那么这也是可以的。 // 指标结果将在表搜索 TableSearch 后由 mn 合并。 if err := is.db.createTSIDByName(dst, metricName); err != nil { return fmt.Errorf("cannot create TSID by MetricName %q: %w", metricName, err) } return nil } // 根据 metricName 去搜索 TSID func (is *indexSearch) getTSIDByMetricName(dst *TSID, metricName []byte) error { dmis := is.db.s.getDeletedMetricIDs() ts := &is.ts // TableSearch kb := &is.kb kb.B = append(kb.B[:0], nsPrefixMetricNameToTSID) // MetricName -> TSID 的前缀 kb.B = append(kb.B, metricName...) kb.B = append(kb.B, kvSeparatorChar) ts.Seek(kb.B) // Seek 查找 ts 中大于或等于 k 的第一项 for ts.NextItem() { // 循环查找 if !bytes.HasPrefix(ts.Item, kb.B) { // ts.Item 不是以 kb.B 为前缀 // 没找到 return io.EOF } v := ts.Item[len(kb.B):] // 获得尾部的值 tail, err := dst.Unmarshal(v) // 填充dst if err != nil { return fmt.Errorf("cannot unmarshal TSID: %w", err) } if len(tail) > 0 { // 尾部还有值 return fmt.Errorf("unexpected non-empty tail left after unmarshaling TSID: %X", tail) } if dmis.Len() > 0 { // 有标记删除的 MetricID 列表 // 验证 dst 是否标记为已删除。 if dmis.Has(dst.MetricID) { // dst 被删除了,继续搜索。 continue } } // 找到了有效的 dst return nil } if err := ts.Error(); err != nil { return fmt.Errorf("error when searching TSID by metricName; searchPrefix %q: %w", kb.B, err) } // 什么都没发现 return io.EOF }
该函数会获取metricName对应的 TSID,但是可能会出现多次连续未命中的情况,为了提高性能,这里做了一点 hack,如果连续未查询到 TSID 100 次则跳过搜索,就只能去创建 TSID 了,如果跳过了 10000 次则又重置可以重新去搜索。 搜索 TSID 是通过下面的getTSIDByMetricName函数来实现的,创建 TSID 是通过createTSIDByName函数实现的。 TSID 的生成方法如下所示:
// lib/storage/index_db.go // 根据指定的 metricName 创建 TSID func (db *indexDB) createTSIDByName(dst *TSID, metricName []byte) error { mn := GetMetricName() defer PutMetricName(mn) if err := mn.Unmarshal(metricName); err != nil { return fmt.Errorf("cannot unmarshal metricName %q: %w", metricName, err) } // 创建 TSID created, err := db.getOrCreateTSID(dst, metricName, mn) if err != nil { return fmt.Errorf("cannot generate TSID: %w", err) } // TSID 创建后要创建索引,这一步是最耗时的 if err := db.createIndexes(dst, mn); err != nil { return fmt.Errorf("cannot create indexes: %w", err) } // 不需要使 tag 缓存无效,因为它在 db 上无效,tb 通过传递给 OpenTable 的invalidateTagFiltersCache flushCallback 刷新。 if created { // 仅当 indexDB 中未找到 tsid 时,才增加 newTimeseriesCreated 计数器 atomic.AddUint64(&db.newTimeseriesCreated, 1) if logNewSeries { logger.Infof("new series created: %s", mn.String()) } } return nil } // getOrCreateTSID 在 db.extDB 中查找指定 metricName 的 TSID // 如果找不到任何内容,则创建新的 TSID // // 如果 TSID 已创建,则返回 true;如果 TSID 在 extDB 中,则返回 false func (db *indexDB) getOrCreateTSID(dst *TSID, metricName []byte, mn *MetricName) (bool, error) { // 在外部存储中搜索 TSID // 这个 db 通常来自上一个时期 var err error // 相当于去上一个索引 db 中查找 TSID if db.doExtDB(func(extDB *indexDB) { err = extDB.getTSIDByNameNoCreate(dst, metricName) }) { if err == nil { // 已在外部存储中找到 TSID return false, nil } if err != io.EOF { return false, fmt.Errorf("external search failed: %w", err) } } // 在外部存储中找不到 TSID,在本地生成。 generateTSID(dst, mn) return true, nil } // 生成 TSID 数据 func generateTSID(dst *TSID, mn *MetricName) { dst.AccountID = mn.AccountID dst.ProjectID = mn.ProjectID // 根据 MetricName 中的 MetricGroup 使用 xxhash 的 sum64 算法生成。 dst.MetricGroupID = xxhash.Sum64(mn.MetricGroup) // 假设 job-like metric 放在 mn.Tags[0],而 instance-like metric 放在 mn.Tags[1] // 这个假设是正确的,因为 mn.Tags 必须在调用 generateTSID() 函数之前使用 mn.sortTags() 进行排序。 // 这允许对磁盘上彼此靠近的相同(job、instance)的数据块进行分组。 // 当从磁盘读取相同 job 和/或 instance 的数据块时,这会减少磁盘寻道和磁盘读取 IO。 // 例如,与 `process_resident_memory_bytes{job="vmstorage"}` 匹配的时间序列的数据块在磁盘上是物理相邻的。 if len(mn.Tags) > 0 { dst.JobID = uint32(xxhash.Sum64(mn.Tags[0].Value)) // 第一个Tag规定为 JobID } if len(mn.Tags) > 1 { dst.InstanceID = uint32(xxhash.Sum64(mn.Tags[1].Value)) // 第二个Tag规定为 InstanceID } dst.MetricID = generateUniqueMetricID() // 生成唯一的指标ID }
MetricID通过generateUniqueMetricID()生成, 在重启时,nextUniqueMetricID被赋值为当时的时间戳, 随后每次新的TSID的创建都会在此基础之上+1。
// lib/storage/index_db.go // 生成唯一的 MetricID func generateUniqueMetricID() uint64 { // 期望的是从此函数返回的 metricID 必须是密集的。 // 如果它们是稀疏的,那么这可能会损害 metric_ids 与 uint64set.Set 的交集性能。 return atomic.AddUint64(&nextUniqueMetricID, 1) } // 该数在重新启动时不能倒退,否则可能会发生 metricID 冲突。 // 所以不要在 VictoriaMetrics 重新启动期间更改服务器上的时间。 var nextUniqueMetricID = uint64(time.Now().UnixNano())
但是我们可能在这里看不懂 TSID 是如何去搜索或者创建的,这就需要我们去了解下 VM 中的倒排索引了。
倒排索引
当创建完TSID后, 需要建立一系列的索引供查找时使用。在 VM 中不同类型的索引都是通过KV关系来描述,在代码中称为Item,Item的结构如下: 在 VM 中 Item 的整体上是一个 KV 结构的字节数组,共计有 7 种类型,每种类型的 Item 通过固定前缀来区分,前缀类型如下图所示。 在storage/index_db.go: createIndexes函数中,去分别建立了各个索引,生成Items,代码如下所示:
// lib/storage/index_db.go // 创建索引 func (db *indexDB) createIndexes(tsid *TSID, mn *MetricName) error { // 索引 items 的顺序很重要,它保证了索引的一致性。 ii := getIndexItems() defer putIndexItems(ii) // 创建 MetricName -> TSID 的索引。 ii.B = append(ii.B, nsPrefixMetricNameToTSID) // 前缀 ii.B = mn.Marshal(ii.B) ii.B = append(ii.B, kvSeparatorChar) // 分隔符 ii.B = tsid.Marshal(ii.B) ii.Next() // 创建 MetricID -> MetricName 索引。 ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricIDToMetricName, mn.AccountID, mn.ProjectID) ii.B = encoding.MarshalUint64(ii.B, tsid.MetricID) ii.B = mn.Marshal(ii.B) ii.Next() // 创建 MetricID -> TSID 索引 ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricIDToTSID, mn.AccountID, mn.ProjectID) ii.B = encoding.MarshalUint64(ii.B, tsid.MetricID) ii.B = tsid.Marshal(ii.B) ii.Next() // 创建 Tag -> MetricID 索引 prefix := kbPool.Get() prefix.B = marshalCommonPrefix(prefix.B[:0], nsPrefixTagToMetricIDs, mn.AccountID, mn.ProjectID) ii.registerTagIndexes(prefix.B, mn, tsid.MetricID) kbPool.Put(prefix) // 将 Items 添加到 Table 中去 return db.tb.AddItems(ii.Items) }
对于ask{market=”NYSE”,ticker=”GOOG”} 1.23的时序指标,对应的MetricName为AccountID=0, ProjectID=0, ask{market=”NYSE”,ticker=”GOOG”},假设生成的TSID为:
{ AccountID: 0 ProjectID: 0 MetricGroupID: 6661248876682682060 JobID: 3817370224 InstanceID: 4166188337 MetricID: 1654132102944898001 }
则生成的索引 Item 逻辑结构如下图所示: 上图为构建的MetricName -> TSID的索引,前缀为nsPrefixMetricNameToTSID=0,整个索引项就是一个key: value的形式,key 为MetricName编码后的值,value 为TSID编码后的值,中间通过一个kvSeparator的分隔符进行连接,当然这些值真正的存储形式都是[]byte。除了上图的这个索引之外还有几个其他的索引:MetricID -> MetricName、MetricID -> TSID、Tag -> MetricID,方式都是一样的,只是要注意每种索引的前缀是不一样的。最后得到的索引就是上面构建的几种索引的集合数组。 索引构建完成后又是如何去持久化数据的呢?保存的数据又是怎样的格式呢?
转载请注明:IT运维空间 » 运维技术 » VictorialMetrics存储原理之索引
发表评论