fix time ranges, use 10k series

Signed-off-by: Artem Fetishev <rtm@victoriametrics.com>

app/vmselect/netstorage/tmp_blocks_file.go

@@ -62,9 +62,6 @@ type tmpBlocksFile struct {
 	r *fs.ReaderAt
 
 	offset uint64
-
-	// err stores the first error occurred while writing the temporary blocks file.
-	err error
 }
 
 func getTmpBlocksFile() *tmpBlocksFile {
@@ -85,7 +82,6 @@ func putTmpBlocksFile(tbf *tmpBlocksFile) {
 	tbf.f = nil
 	tbf.r = nil
 	tbf.offset = 0
-	tbf.err = nil
 	tmpBlocksFilePool.Put(tbf)
 }
 
@@ -113,10 +109,6 @@ var (
 // and this must be handled.
 func (tbf *tmpBlocksFile) WriteBlockRefData(b []byte) (tmpBlockAddr, error) {
 	var addr tmpBlockAddr
-	if tbf.err != nil {
-		// Do not write anything to the tbf after the first failed write
-		return addr, tbf.err
-	}
 	addr.offset = tbf.offset
 	addr.size = len(b)
 	tbf.offset += uint64(addr.size)
@@ -130,8 +122,7 @@ func (tbf *tmpBlocksFile) WriteBlockRefData(b []byte) (tmpBlockAddr, error) {
 	if tbf.f == nil {
 		f, err := os.CreateTemp(tmpBlocksDir, "")
 		if err != nil {
-			tbf.err = fmt.Errorf("cannot create temporary blocks file at %q: %w", tmpBlocksDir, err)
-			return addr, tbf.err
+			return addr, err
 		}
 		tbf.f = f
 		tmpBlocksFilesCreated.Inc()
@@ -139,9 +130,7 @@ func (tbf *tmpBlocksFile) WriteBlockRefData(b []byte) (tmpBlockAddr, error) {
 	_, err := tbf.f.Write(tbf.buf)
 	tbf.buf = append(tbf.buf[:0], b...)
 	if err != nil {
-		// The blocks buffered at tbf.buf could be partially lost, mark the tbf as unusable.
-		tbf.err = fmt.Errorf("cannot write block to %q: %w", tbf.f.Name(), err)
-		return addr, tbf.err
+		return addr, fmt.Errorf("cannot write block to %q: %w", tbf.f.Name(), err)
 	}
 	return addr, nil
 }
@@ -152,16 +141,12 @@ func (tbf *tmpBlocksFile) Len() uint64 {
 }
 
 func (tbf *tmpBlocksFile) Finalize() error {
-	if tbf.err != nil {
-		return tbf.err
-	}
 	if tbf.f == nil {
 		return nil
 	}
 	fname := tbf.f.Name()
 	if _, err := tbf.f.Write(tbf.buf); err != nil {
-		tbf.err = fmt.Errorf("cannot write the remaining %d bytes to %q: %w", len(tbf.buf), fname, err)
-		return tbf.err
+		return fmt.Errorf("cannot write the remaining %d bytes to %q: %w", len(tbf.buf), fname, err)
 	}
 	tbf.buf = tbf.buf[:0]
 	r := fs.NewReaderAt(tbf.f)
@@ -181,10 +166,6 @@ func (tbf *tmpBlocksFile) Finalize() error {
 }
 
 func (tbf *tmpBlocksFile) MustReadBlockRefAt(partRef storage.PartRef, addr tmpBlockAddr) storage.BlockRef {
-	if tbf.err != nil {
-		// This should never happen, since Finalize() already returns the error for such a tbf.
-		logger.Panicf("BUG: cannot read block at %s from the temporary blocks file with the failed write: %s", addr, tbf.err)
-	}
 	var buf []byte
 	if tbf.r == nil {
 		buf = tbf.buf[addr.offset : addr.offset+uint64(addr.size)]

docs/victoriametrics/changelog/CHANGELOG.md

@@ -33,7 +33,6 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fix infinite loop in the OpenTelemetry Firehose ingestion endpoint (`/opentelemetry/api/v1/push`) when receiving a malformed record with an incomplete varint in the `data` field. Previously this caused the goroutine to spin forever, permanently consuming CPU until the process was restarted.
 * BUGFIX: [vmalert-tool](https://docs.victoriametrics.com/victoriametrics/vmalert-tool/): reuse connections to `-remoteWrite.url` when writing the results of recording rules and alerts. Previously every series was sent over a new connection, which left a lot of sockets in `TIME_WAIT` state and could exhaust the ephemeral port range. The number of idle connections can be tuned via the new `-remoteWrite.maxIdleConnections` command-line flag. Thanks @evkuzin for contribution.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): prevent process crash in `sort_by_label_numeric()` and `sort_by_label_numeric_desc()` when a label value contains a number with 309 or more digits. See [#11423](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11423).
-* BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fail the query request directly when there is not enough disk space to store temporary search results. Previously, such queries could lead to vmselect crash. See [#4688](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4688).
 
 ## [v1.150.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.150.0)
 

lib/storage/search_timing_test.go

@@ -42,8 +42,8 @@ func benchmarkSearchData(b *testing.B, s *Storage, tr TimeRange, mrs []MetricRow
 	}
 
 	var mn MetricName
-	got := make([]MetricRow, len(mrs))
-	for i, mb := range mbs {
+	var got []MetricRow
+	for _, mb := range mbs {
 		rb := newTestRawBlock(mb.Block, tr)
 		if err := mn.Unmarshal(mb.MetricName); err != nil {
 			b.Fatalf("cannot unmarshal MetricName %v: %v", string(mb.MetricName), err)
@@ -55,7 +55,7 @@ func benchmarkSearchData(b *testing.B, s *Storage, tr TimeRange, mrs []MetricRow
 				Timestamp:     timestamp,
 				Value:         rb.Values[j],
 			}
-			got[i] = mr
+			got = append(got, mr)
 		}
 	}
 	testSortMetricRows(got)

lib/storage/storage_timing_test.go

@@ -538,12 +538,18 @@ func benchmarkSearchMetricNames(b *testing.B, s *Storage, tr TimeRange, mrs []Me
 		got[i] = string(mn.MetricGroup)
 	}
 	slices.Sort(got)
-	want := make([]string, len(mrs))
-	for i, mr := range mrs {
+
+	seen := make(map[string]bool)
+	var want []string
+	for _, mr := range mrs {
 		if err := mn.UnmarshalRaw(mr.MetricNameRaw); err != nil {
 			b.Fatalf("could not unmarshal metric row: %v", err)
 		}
-		want[i] = string(mn.MetricGroup)
+		v := string(mn.MetricGroup)
+		if !seen[v] {
+			want = append(want, v)
+			seen[v] = true
+		}
 	}
 	slices.Sort(want)
 	if diff := cmp.Diff(want, got); diff != "" {
@@ -929,3 +935,187 @@ func variableTimeRange() []dataConfig {
 	}
 	return cfgs
 }
+
+func BenchmarkSearchTimeRanges_Data(b *testing.B) {
+	benchmarkSearchTimeRanges(b, benchmarkSearchData)
+}
+
+func BenchmarkSearchTimeRanges_MetricNames(b *testing.B) {
+	benchmarkSearchTimeRanges(b, benchmarkSearchMetricNames)
+}
+
+func benchmarkSearchTimeRanges(b *testing.B, op func(b *testing.B, s *Storage, tr TimeRange, mrs []MetricRow)) {
+	type cfg struct {
+		name   string
+		tr     TimeRange
+		numTRs int64
+	}
+	tr1h := cfg{
+		name: "1h",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 1, 1, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 1,
+	}
+	tr2h := cfg{
+		name: "2h",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 1, 2, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 2,
+	}
+	tr3h := cfg{
+		name: "3h",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 1, 3, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 3,
+	}
+	tr6h := cfg{
+		name: "6h",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 1, 6, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 6,
+	}
+	tr12h := cfg{
+		name: "12h",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 12,
+	}
+	tr24h := cfg{
+		name: "24h",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 1, 24, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 24,
+	}
+	tr1d := cfg{
+		name: "1d",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 2, 0, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 1,
+	}
+	tr2d := cfg{
+		name: "2d",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 3, 0, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 2,
+	}
+	tr4d := cfg{
+		name: "4d",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 5, 0, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 4,
+	}
+	tr8d := cfg{
+		name: "8d",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 9, 0, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 8,
+	}
+	tr15d := cfg{
+		name: "15d",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 16, 0, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 15,
+	}
+	tr30d := cfg{
+		name: "30d",
+		tr: TimeRange{
+			MinTimestamp: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2025, 1, 31, 0, 0, 0, 0, time.UTC).UnixMilli(),
+		},
+		numTRs: 30,
+	}
+
+	const seriesPerHour = 10_000
+	for _, seriesRepeatEveryHour := range []bool{false, true} {
+		for _, cfg := range []cfg{tr1h, tr2h, tr3h, tr6h, tr12h, tr24h} {
+			name := fmt.Sprintf("seriesPerHour=%d/seriesRepeatEveryHour=%t/%s", seriesPerHour, seriesRepeatEveryHour, cfg.name)
+			b.Run(name, func(b *testing.B) {
+				benchmarkSearchTimeRange(b, seriesPerHour, cfg.tr, cfg.numTRs, seriesRepeatEveryHour, op)
+			})
+		}
+	}
+
+	const seriesPerDay = 10_000
+	for _, seriesRepeatEveryDay := range []bool{false, true} {
+		for _, cfg := range []cfg{tr1d, tr2d, tr4d, tr8d, tr15d, tr30d} {
+			name := fmt.Sprintf("seriesPerDay=%d/seriesRepeatEveryDay=%t/%s", seriesPerDay, seriesRepeatEveryDay, cfg.name)
+			b.Run(name, func(b *testing.B) {
+				benchmarkSearchTimeRange(b, seriesPerDay, cfg.tr, cfg.numTRs, seriesRepeatEveryDay, op)
+			})
+		}
+	}
+}
+
+func benchmarkSearchTimeRange(b *testing.B, numSeries int, tr TimeRange, numTRs int64, sameSeries bool, search func(b *testing.B, s *Storage, tr TimeRange, mrs []MetricRow)) {
+	b.Helper()
+	genRows := func(n int, tr TimeRange, trSeqNum int64) []MetricRow {
+		mrs := make([]MetricRow, n)
+		if n == 0 {
+			return mrs
+		}
+		if sameSeries {
+			trSeqNum = 0
+		}
+		step := (tr.MaxTimestamp - tr.MinTimestamp) / int64(n)
+		for i := range n {
+			name := fmt.Sprintf("metric_%09d_%09d", trSeqNum, i)
+			labelName := fmt.Sprintf("label_%09d_%09d", trSeqNum, i)
+			labelValue := fmt.Sprintf("value_%09d_%09d", trSeqNum, i)
+			mn := MetricName{
+				MetricGroup: []byte(name),
+				Tags: []Tag{
+					{[]byte(labelName), []byte("value")},
+					{[]byte("label"), []byte(labelValue)},
+				},
+			}
+			timestamp := tr.MinTimestamp + int64(i)*step
+			value := float64(timestamp)
+			mrs[i].MetricNameRaw = mn.marshalRaw(nil)
+			mrs[i].Timestamp = timestamp
+			mrs[i].Value = value
+		}
+		return mrs
+	}
+
+	var mrs []MetricRow
+	trLen := (tr.MaxTimestamp - tr.MinTimestamp) / numTRs
+	for i := range numTRs {
+		subTR := TimeRange{
+			MinTimestamp: tr.MinTimestamp + trLen*i,
+			MaxTimestamp: tr.MinTimestamp + trLen*(i+1),
+		}
+		mrs = append(mrs, genRows(numSeries, subTR, i)...)
+	}
+
+	s := MustOpenStorage(b.Name(), OpenOptions{})
+	s.AddRows(mrs, 64)
+	s.DebugFlush()
+
+	tr.MaxTimestamp -= 1
+	search(b, s, tr, mrs)
+
+	s.MustClose()
+	_ = os.RemoveAll(b.Name())
+}