Victoria Metrics 写入流程分析（上篇）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

d := s.nextRetentionSeconds()
if d >= 3600 {
	// Fast path: 只在索引轮换前最后一个小时内进行预写
	return nil
}

for i := range rows {
	r := &rows[i]
	// 当 d=[3600 ... 0] 时，概率从 0% 到 100% 成比例增加。
	// 这也符合直觉，因为越新的数据，越可能出现在第二天
	pMin := float64(d) / 3600
	p := float64(uint32(fastHashUint64(r.TSID.MetricID))) / (1 << 32)
	if p < pMin {
		// Fast path: it is too early to pre-fill indexes for the given MetricID.
		continue
	}

	// 在缓存中判断是否已经预写过，这里的 generation 是 idbNext 的
	if s.dateMetricIDCache.Has(generation, date, metricID) {
		continue
	}

	// 接着在 idbNext 中判断，如果有，则更新缓存，这样下次就就可以直接通过缓存判断
	if isNext.hasDateMetricIDNoExtDB(date, metricID) {
		s.dateMetricIDCache.Set(generation, date, metricID)
		continue
	}

	// 创建为 idbNext 的 Global 与 PerDay 中的索引
	createAllIndexesForMetricName(isNext, mn, &r.TSID, date)
	genTSID.TSID = r.TSID
	genTSID.generation = generation
	// 更新 tsidCache 与 dateMetricIDCache
	s.putSeriesToCache(metricNameRaw, &genTSID, date)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

hm := s.currHourMetricIDs.Load()
hmPrev := s.prevHourMetricIDs.Load()
// 这里首先把 ts 转成 hour，然后再减去 23，在正数时有效，那么 pMin 的取值变化就是 [0...1]
// 也就是说预写的概率越来越大
pMin := (float64(ts%(3600*24)) / 3600) - 23
for i := range rows {
	r := &rows[i]
	date = uint64(r.Timestamp) / msecPerDay
	hour = uint64(r.Timestamp) / msecPerHour
	if hour == hm.hour { // 属于当前小时，从 currHour 缓存中查找
		if hm.m.Has(metricID) { // Fast path，从当前缓存中找到
			if pMin > 0 { // hour 大于 23 时才开始预写下一天数据
				p := float64(uint32(fastHashUint64(metricID))) / (1 << 32)
				if p < pMin && !nextDayMetricIDs.Has(metricID) {
					pendingDateMetricIDs = append(pendingDateMetricIDs, pendingDateMetricID{
						date: date + 1, // 加 1 表示预写下一天的索引
						tsid: &r.TSID,
						mr:   mrs[i],
					})
					pendingNextDayMetricIDs = append(pendingNextDayMetricIDs, metricID)
				}
			}
			continue
		}
		pendingHourEntries = append(pendingHourEntries, metricID)
		// 已经在当前的前一个小时写入过了
		if date == hmPrevDate && hmPrev.m.Has(metricID) {
			continue
		}
	}

	// Slow path 从全局缓存中查找
	if s.dateMetricIDCache.Has(generation, date, metricID) {
		continue
	}

	// Slow path: 在 indexdb 插入 (date, metricID)
	pendingDateMetricIDs = append(pendingDateMetricIDs, pendingDateMetricID{
		date: date,
		tsid: &r.TSID,
		mr:   mrs[i],
	})

	for _, dmid := range pendingDateMetricIDs {
		if !is.hasDateMetricIDNoExtDB(date, metricID) {
			mn.sortTags()
			// 创建天级别的索引
			is.createPerDayIndexes(date, dmid.tsid, mn)
		}
		dateMetricIDsForCache = append(dateMetricIDsForCache, dateMetricID{
			date:     date,
			metricID: metricID,
		})
	}
	// 在索引更新成功后，再把 (date, metricID) 对插入到缓存中
	s.dateMetricIDCache.Store(generation, dateMetricIDsForCache)
}