From 097a4c10dd0dc95d9287c9a651351fd660bc6f4f Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Tue, 29 Sep 2020 21:47:40 +0300 Subject: [PATCH] app/vmstorage: add metrics for determining whether background merges need additional disk space to complete These metrics are: * vm_small_merge_need_free_disk_space * vm_big_merge_need_free_disk_space Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/686 --- app/vmstorage/main.go | 8 ++++++ lib/storage/partition.go | 47 +++++++++++++++++++++++++++-------- lib/storage/partition_test.go | 6 ++--- 3 files changed, 48 insertions(+), 13 deletions(-) diff --git a/app/vmstorage/main.go b/app/vmstorage/main.go index 3d9efbbe7b..ea76daf549 100644 --- a/app/vmstorage/main.go +++ b/app/vmstorage/main.go @@ -324,6 +324,14 @@ func registerStorageMetrics(strg *storage.Storage) { return float64(idbm().AssistedMerges) }) + // See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/686 + metrics.NewGauge(`vm_small_merge_need_free_disk_space`, func() float64 { + return float64(tm().SmallMergeNeedFreeDiskSpace) + }) + metrics.NewGauge(`vm_big_merge_need_free_disk_space`, func() float64 { + return float64(tm().BigMergeNeedFreeDiskSpace) + }) + metrics.NewGauge(`vm_pending_rows{type="storage"}`, func() float64 { return float64(tm().PendingRows) }) diff --git a/lib/storage/partition.go b/lib/storage/partition.go index 9bc347a8b8..73d7aba48a 100644 --- a/lib/storage/partition.go +++ b/lib/storage/partition.go @@ -330,6 +330,9 @@ type partitionMetrics struct { SmallPartsRefCount uint64 SmallAssistedMerges uint64 + + SmallMergeNeedFreeDiskSpace uint64 + BigMergeNeedFreeDiskSpace uint64 } // UpdateMetrics updates m with metrics from pt. @@ -388,6 +391,9 @@ func (pt *partition) UpdateMetrics(m *partitionMetrics) { m.SmallRowsDeleted += atomic.LoadUint64(&pt.smallRowsDeleted) m.SmallAssistedMerges += atomic.LoadUint64(&pt.smallAssistedMerges) + + m.SmallMergeNeedFreeDiskSpace = atomic.LoadUint64(&smallMergeNeedFreeDiskSpace) + m.BigMergeNeedFreeDiskSpace = atomic.LoadUint64(&bigMergeNeedFreeDiskSpace) } // AddRows adds the given rows to the partition pt. @@ -990,9 +996,10 @@ func (pt *partition) mergeBigParts(isFinal bool) error { maxRows := maxRowsByPath(pt.bigPartsPath) pt.partsLock.Lock() - pws := getPartsToMerge(pt.bigParts, maxRows, isFinal) + pws, needFreeSpace := getPartsToMerge(pt.bigParts, maxRows, isFinal) pt.partsLock.Unlock() + atomicSetBool(&bigMergeNeedFreeDiskSpace, needFreeSpace) return pt.mergeParts(pws, pt.stopCh) } @@ -1008,14 +1015,28 @@ func (pt *partition) mergeSmallParts(isFinal bool) error { } pt.partsLock.Lock() - pws := getPartsToMerge(pt.smallParts, maxRows, isFinal) + pws, needFreeSpace := getPartsToMerge(pt.smallParts, maxRows, isFinal) pt.partsLock.Unlock() + atomicSetBool(&smallMergeNeedFreeDiskSpace, needFreeSpace) return pt.mergeParts(pws, pt.stopCh) } var errNothingToMerge = fmt.Errorf("nothing to merge") +func atomicSetBool(p *uint64, b bool) { + v := uint64(0) + if b { + v = 1 + } + atomic.StoreUint64(p, v) +} + +var ( + smallMergeNeedFreeDiskSpace uint64 + bigMergeNeedFreeDiskSpace uint64 +) + // mergeParts merges pws. // // Merging is immediately stopped if stopCh is closed. @@ -1242,7 +1263,8 @@ func removeParts(pws []*partWrapper, partsToRemove map[*partWrapper]bool, isBig // getPartsToMerge returns optimal parts to merge from pws. // // The returned rows will contain less than maxRows rows. -func getPartsToMerge(pws []*partWrapper, maxRows uint64, isFinal bool) []*partWrapper { +// The function returns true if pws contains parts, which cannot be merged because of maxRows limit. +func getPartsToMerge(pws []*partWrapper, maxRows uint64, isFinal bool) ([]*partWrapper, bool) { pwsRemaining := make([]*partWrapper, 0, len(pws)) for _, pw := range pws { if !pw.isInMerge { @@ -1251,13 +1273,14 @@ func getPartsToMerge(pws []*partWrapper, maxRows uint64, isFinal bool) []*partWr } maxPartsToMerge := defaultPartsToMerge var pms []*partWrapper + needFreeSpace := false if isFinal { for len(pms) == 0 && maxPartsToMerge >= finalPartsToMerge { - pms = appendPartsToMerge(pms[:0], pwsRemaining, maxPartsToMerge, maxRows) + pms, needFreeSpace = appendPartsToMerge(pms[:0], pwsRemaining, maxPartsToMerge, maxRows) maxPartsToMerge-- } } else { - pms = appendPartsToMerge(pms[:0], pwsRemaining, maxPartsToMerge, maxRows) + pms, needFreeSpace = appendPartsToMerge(pms[:0], pwsRemaining, maxPartsToMerge, maxRows) } for _, pw := range pms { if pw.isInMerge { @@ -1265,15 +1288,16 @@ func getPartsToMerge(pws []*partWrapper, maxRows uint64, isFinal bool) []*partWr } pw.isInMerge = true } - return pms + return pms, needFreeSpace } // appendPartsToMerge finds optimal parts to merge from src, appends // them to dst and returns the result. -func appendPartsToMerge(dst, src []*partWrapper, maxPartsToMerge int, maxRows uint64) []*partWrapper { +// The function returns true if src contains parts, which cannot be merged because of maxRows limit. +func appendPartsToMerge(dst, src []*partWrapper, maxPartsToMerge int, maxRows uint64) ([]*partWrapper, bool) { if len(src) < 2 { // There is no need in merging zero or one part :) - return dst + return dst, false } if maxPartsToMerge < 2 { logger.Panicf("BUG: maxPartsToMerge cannot be smaller than 2; got %d", maxPartsToMerge) @@ -1281,10 +1305,12 @@ func appendPartsToMerge(dst, src []*partWrapper, maxPartsToMerge int, maxRows ui // Filter out too big parts. // This should reduce N for O(N^2) algorithm below. + needFreeSpace := false maxInPartRows := maxRows / 2 tmp := make([]*partWrapper, 0, len(src)) for _, pw := range src { if pw.p.ph.RowsCount > maxInPartRows { + needFreeSpace = true continue } tmp = append(tmp, pw) @@ -1320,6 +1346,7 @@ func appendPartsToMerge(dst, src []*partWrapper, maxPartsToMerge int, maxRows ui } if rowsSum > maxRows { // There is no need in verifying remaining parts with higher number of rows + needFreeSpace = true break } m := float64(rowsSum) / float64(a[len(a)-1].p.ph.RowsCount) @@ -1337,9 +1364,9 @@ func appendPartsToMerge(dst, src []*partWrapper, maxPartsToMerge int, maxRows ui } if maxM < minM { // There is no sense in merging parts with too small m. - return dst + return dst, needFreeSpace } - return append(dst, pws...) + return append(dst, pws...), needFreeSpace } func openParts(pathPrefix1, pathPrefix2, path string) ([]*partWrapper, error) { diff --git a/lib/storage/partition_test.go b/lib/storage/partition_test.go index 457d9b9d0b..bfe5e3da47 100644 --- a/lib/storage/partition_test.go +++ b/lib/storage/partition_test.go @@ -49,7 +49,7 @@ func TestAppendPartsToMergeManyParts(t *testing.T) { iterationsCount := 0 rowsMerged := uint64(0) for { - pms := appendPartsToMerge(nil, pws, defaultPartsToMerge, maxOutPartRows) + pms, _ := appendPartsToMerge(nil, pws, defaultPartsToMerge, maxOutPartRows) if len(pms) == 0 { break } @@ -99,7 +99,7 @@ func testAppendPartsToMerge(t *testing.T, maxPartsToMerge int, initialRowsCount, pws := newTestPartWrappersForRowsCount(initialRowsCount) // Verify appending to nil. - pms := appendPartsToMerge(nil, pws, maxPartsToMerge, 1e9) + pms, _ := appendPartsToMerge(nil, pws, maxPartsToMerge, 1e9) rowsCount := newTestRowsCountFromPartWrappers(pms) if !reflect.DeepEqual(rowsCount, expectedRowsCount) { t.Fatalf("unexpected rowsCount for maxPartsToMerge=%d, initialRowsCount=%d; got\n%d; want\n%d", @@ -118,7 +118,7 @@ func testAppendPartsToMerge(t *testing.T, maxPartsToMerge int, initialRowsCount, {}, {}, } - pms = appendPartsToMerge(prefix, pws, maxPartsToMerge, 1e9) + pms, _ = appendPartsToMerge(prefix, pws, maxPartsToMerge, 1e9) if !reflect.DeepEqual(pms[:len(prefix)], prefix) { t.Fatalf("unexpected prefix for maxPartsToMerge=%d, initialRowsCount=%d; got\n%+v; want\n%+v", maxPartsToMerge, initialRowsCount, pms[:len(prefix)], prefix)