vmsingle: move storage flags to vmstorage/main.go

- storage-only flags are moved to vmstorage/main.go
- vmsingle-specific initializations are moved to the very end of Init() func
- make force-merge goroutine return in case of error

These changes reduce diff with cluster vmstorage.

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

.github/workflows/codeql-analysis-go.yml

@@ -50,14 +50,14 @@ jobs:
           restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@68bde559dea0fdcac2102bfdf6230c5f70eb485e  # v4.35.4
+        uses: github/codeql-action/init@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
         with:
           languages: go
 
       - name: Autobuild
-        uses: github/codeql-action/autobuild@68bde559dea0fdcac2102bfdf6230c5f70eb485e  # v4.35.4
+        uses: github/codeql-action/autobuild@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@68bde559dea0fdcac2102bfdf6230c5f70eb485e  # v4.35.4
+        uses: github/codeql-action/analyze@e46ed2cbd01164d986452f91f178727624ae40d7  # v4.35.3
         with:
           category: 'language:go'

Makefile

@@ -485,8 +485,8 @@ apptest-legacy: victoria-metrics-race vmbackup-race vmrestore-race
 		curl --output-dir /tmp -LO $${URL}/$${VMSINGLE} && tar xzf /tmp/$${VMSINGLE} -C $${DIR} && \
 		curl --output-dir /tmp -LO $${URL}/$${VMCLUSTER} && tar xzf /tmp/$${VMCLUSTER} -C $${DIR} \
 	); \
-	VMSINGLE_V1_132_0_PATH=$${DIR}/victoria-metrics-prod \
-	VMSTORAGE_V1_132_0_PATH=$${DIR}/vmstorage-prod \
+	VM_LEGACY_VMSINGLE_PATH=$${DIR}/victoria-metrics-prod \
+	VM_LEGACY_VMSTORAGE_PATH=$${DIR}/vmstorage-prod \
 	go test ./apptest/tests -run="^TestLegacySingle.*"
 
 benchmark:

app/victoria-metrics/main.go

@@ -22,7 +22,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 )
 
 var (
@@ -30,21 +29,11 @@ var (
 	useProxyProtocol = flagutil.NewArrayBool("httpListenAddr.useProxyProtocol", "Whether to use proxy protocol for connections accepted at the corresponding -httpListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt . "+
 		"With enabled proxy protocol http server cannot serve regular /metrics endpoint. Use -pushmetrics.url for metrics pushing")
-	minScrapeInterval = flag.Duration("dedup.minScrapeInterval", 0, "Leave only the last sample in every time series per each discrete interval "+
-		"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
 	dryRun = flag.Bool("dryRun", false, "Whether to check config files without running VictoriaMetrics. The following config files are checked: "+
 		"-promscrape.config, -relabelConfig and -streamAggr.config. Unknown config entries aren't allowed in -promscrape.config by default. "+
 		"This can be changed with -promscrape.config.strictParse=false command-line flag")
-	inmemoryDataFlushInterval = flag.Duration("inmemoryDataFlushInterval", 5*time.Second, "The interval for guaranteed saving of in-memory data to disk. "+
-		"The saved data survives unclean shutdowns such as OOM crash, hardware reset, SIGKILL, etc. "+
-		"Bigger intervals may help increase the lifetime of flash storage with limited write cycles (e.g. Raspberry PI). "+
-		"Smaller intervals increase disk IO load. Minimum supported value is 1s")
 	maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
 		"By default there are no limits on samples ingestion rate.")
-	finalDedupScheduleInterval = flag.Duration("storage.finalDedupScheduleCheckInterval", time.Hour, "The interval for checking when final deduplication process should be started."+
-		"Storage unconditionally adds 25% jitter to the interval value on each check evaluation."+
-		" Changing the interval to the bigger values may delay downsampling, deduplication for historical data."+
-		" See also https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
 )
 
 func main() {
@@ -87,12 +76,6 @@ func main() {
 	}
 	logger.Infof("starting VictoriaMetrics at %q...", listenAddrs)
 	startTime := time.Now()
-	storage.SetDedupInterval(*minScrapeInterval)
-	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
-	if *finalDedupScheduleInterval < time.Hour {
-		logger.Fatalf("-dedup.finalDedupScheduleCheckInterval cannot be smaller than 1 hour; got %s", *finalDedupScheduleInterval)
-	}
-	storage.SetFinalDedupScheduleInterval(*finalDedupScheduleInterval)
 	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
 	vmselect.Init()
 	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)

app/vmstorage/main.go

@@ -51,6 +51,12 @@ var (
 	retentionTimezoneOffset = flag.Duration("retentionTimezoneOffset", 0, "The offset for performing indexdb rotation. "+
 		"If set to 0, then the indexdb rotation is performed at 4am UTC time per each -retentionPeriod. "+
 		"If set to 2h, then the indexdb rotation is performed at 4am EET time (the timezone with +2h offset)")
+	minScrapeInterval = flag.Duration("dedup.minScrapeInterval", 0, "Leave only the last sample in every time series per each discrete interval "+
+		"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
+	inmemoryDataFlushInterval = flag.Duration("inmemoryDataFlushInterval", 5*time.Second, "The interval for guaranteed saving of in-memory data to disk. "+
+		"The saved data survives unclean shutdowns such as OOM crash, hardware reset, SIGKILL, etc. "+
+		"Bigger intervals may help increase the lifetime of flash storage with limited write cycles (e.g. Raspberry PI). "+
+		"Smaller intervals increase disk IO load. Minimum supported value is 1s")
 
 	logNewSeries = flag.Bool("logNewSeries", false, "Whether to log new series. This option is for debug purposes only. It can lead to performance issues "+
 		"when big number of new series are ingested into VictoriaMetrics")
@@ -68,6 +74,11 @@ var (
 
 	minFreeDiskSpaceBytes = flagutil.NewBytes("storage.minFreeDiskSpaceBytes", 100e6, "The minimum free disk space at -storageDataPath after which the storage stops accepting new data")
 
+	finalDedupScheduleInterval = flag.Duration("storage.finalDedupScheduleCheckInterval", time.Hour, "The interval for checking when final deduplication process should be started."+
+		"Storage unconditionally adds 25% jitter to the interval value on each check evaluation."+
+		" Changing the interval to the bigger values may delay downsampling, deduplication for historical data."+
+		" See also https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
+
 	cacheSizeStorageTSID = flagutil.NewBytes("storage.cacheSizeStorageTSID", 0, "Overrides max size for storage/tsid cache. "+
 		"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#cache-tuning")
 	cacheSizeStorageMetricName = flagutil.NewBytes("storage.cacheSizeStorageMetricName", 0, "Overrides max size for storage/metricName cache. "+
@@ -111,11 +122,16 @@ func Init(resetCacheIfNeeded func(mrs []storage.MetricRow)) {
 		logger.Fatalf("invalid `-precisionBits`: %s", err)
 	}
 
-	resetResponseCacheIfNeeded = resetCacheIfNeeded
+	storage.SetDedupInterval(*minScrapeInterval)
+	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
 	storage.LegacySetRetentionTimezoneOffset(*retentionTimezoneOffset)
 	storage.SetFreeDiskSpaceLimit(minFreeDiskSpaceBytes.N)
 	storage.SetTSIDCacheSize(cacheSizeStorageTSID.IntN())
 	storage.SetTagFiltersCacheSize(cacheSizeIndexDBTagFilters.IntN())
+	if *finalDedupScheduleInterval < time.Hour {
+		logger.Fatalf("-storage.finalDedupScheduleCheckInterval cannot be smaller than 1 hour; got %s", *finalDedupScheduleInterval)
+	}
+	storage.SetFinalDedupScheduleInterval(*finalDedupScheduleInterval)
 	storage.SetMetricNamesStatsCacheSize(cacheSizeMetricNamesStats.IntN())
 	storage.SetMetricNameCacheSize(cacheSizeStorageMetricName.IntN())
 	storage.SetMetadataStorageSize(metadataStorageSize.IntN())
@@ -134,9 +150,9 @@ func Init(resetCacheIfNeeded func(mrs []storage.MetricRow)) {
 	if *idbPrefillStart > 23*time.Hour {
 		logger.Panicf("-storage.idbPrefillStart cannot exceed 23 hours; got %s", idbPrefillStart)
 	}
+	fs.RegisterPathFsMetrics(*storageDataPath)
 	logger.Infof("opening storage at %q with -retentionPeriod=%s", *storageDataPath, retentionPeriod)
 	startTime := time.Now()
-	WG = syncwg.WaitGroup{}
 	opts := storage.OpenOptions{
 		Retention:                   retentionPeriod.Duration(),
 		FutureRetention:             futureRetention.Duration(),
@@ -149,7 +165,6 @@ func Init(resetCacheIfNeeded func(mrs []storage.MetricRow)) {
 		LogNewSeries:                *logNewSeries,
 	}
 	strg := storage.MustOpenStorage(*storageDataPath, opts)
-	Storage = strg
 	initStaleSnapshotsRemover(strg)
 
 	var m storage.Metrics
@@ -168,7 +183,10 @@ func Init(resetCacheIfNeeded func(mrs []storage.MetricRow)) {
 		writeStorageMetrics(w, strg)
 	})
 	metrics.RegisterSet(storageMetrics)
-	fs.RegisterPathFsMetrics(*storageDataPath)
+
+	WG = syncwg.WaitGroup{}
+	resetResponseCacheIfNeeded = resetCacheIfNeeded
+	Storage = strg
 }
 
 var storageMetrics *metrics.Set
@@ -340,6 +358,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 			startTime := time.Now()
 			if err := Storage.ForceMergePartitions(partitionNamePrefix); err != nil {
 				logger.Errorf("error in forced merge for partition_prefix=%q: %s", partitionNamePrefix, err)
+				return
 			}
 			logger.Infof("forced merge for partition_prefix=%q has been successfully finished in %.3f seconds", partitionNamePrefix, time.Since(startTime).Seconds())
 		}()
@@ -353,6 +372,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		Storage.DebugFlush()
 		return true
 	}
+
 	if path == "/internal/log_new_series" {
 		if !httpserver.CheckAuthFlag(w, r, logNewSeriesAuthKey) {
 			return true

apptest/app.go

@@ -22,7 +22,6 @@ var (
 	vminsertAddrRE              = regexp.MustCompile(`accepting vminsert conns at (.*:\d{1,5})$`)
 	vminsertClusterNativeAddrRE = regexp.MustCompile(`started TCP clusternative server at "(.*:\d{1,5})"`)
 	vmselectAddrRE              = regexp.MustCompile(`accepting vmselect conns at (.*:\d{1,5})$`)
-	vmauthHttpListenAddrRE      = regexp.MustCompile(`pprof handlers are exposed at http://(.*:\d{1,5})/debug/pprof/`)
 )
 
 // app represents an instance of some VictoriaMetrics server (such as vmstorage,

apptest/testcase.go

@@ -88,11 +88,19 @@ func (tc *TestCase) MustStartDefaultVmsingle() *Vmsingle {
 }
 
 // MustStartVmsingle is a test helper function that starts an instance of
-// vmsingle (latest version) and fails the test if the app fails to start.
+// vmsingle located at ../../bin/victoria-metrics-race and fails the test if the app
+// fails to start.
 func (tc *TestCase) MustStartVmsingle(instance string, flags []string) *Vmsingle {
 	tc.t.Helper()
+	return tc.MustStartVmsingleAt(instance, "../../bin/victoria-metrics-race", flags)
+}
 
-	app, err := StartVmsingle(instance, flags, tc.cli, tc.output)
+// MustStartVmsingleAt is a test helper function that starts an instance of
+// vmsingle and fails the test if the app fails to start.
+func (tc *TestCase) MustStartVmsingleAt(instance, binary string, flags []string) *Vmsingle {
+	tc.t.Helper()
+
+	app, err := StartVmsingleAt(instance, binary, flags, tc.cli, tc.output)
 	if err != nil {
 		tc.t.Fatalf("Could not start %s: %v", instance, err)
 	}
@@ -101,11 +109,19 @@ func (tc *TestCase) MustStartVmsingle(instance string, flags []string) *Vmsingle
 }
 
 // MustStartVmstorage is a test helper function that starts an instance of
-// vmstorage (latest version) and fails the test if the app fails to start.
+// vmstorage located at ../../bin/vmstorage-race and fails the test if the app fails
+// to start.
 func (tc *TestCase) MustStartVmstorage(instance string, flags []string) *Vmstorage {
 	tc.t.Helper()
+	return tc.MustStartVmstorageAt(instance, "../../bin/vmstorage-race", flags)
+}
 
-	app, err := StartVmstorage(instance, flags, tc.cli, tc.output)
+// MustStartVmstorageAt is a test helper function that starts an instance of
+// vmstorage and fails the test if the app fails to start.
+func (tc *TestCase) MustStartVmstorageAt(instance string, binary string, flags []string) *Vmstorage {
+	tc.t.Helper()
+
+	app, err := StartVmstorageAt(instance, binary, flags, tc.cli, tc.output)
 	if err != nil {
 		tc.t.Fatalf("Could not start %s: %v", instance, err)
 	}
@@ -114,7 +130,7 @@ func (tc *TestCase) MustStartVmstorage(instance string, flags []string) *Vmstora
 }
 
 // MustStartVmselect is a test helper function that starts an instance of
-// vmselect (latest version) and fails the test if the app fails to start.
+// vmselect and fails the test if the app fails to start.
 func (tc *TestCase) MustStartVmselect(instance string, flags []string) *Vmselect {
 	tc.t.Helper()
 
@@ -274,8 +290,10 @@ func (tc *TestCase) MustStartDefaultCluster() *Vmcluster {
 // tests usually come paired with corresponding vmsingle tests.
 type ClusterOptions struct {
 	Vmstorage1Instance string
+	Vmstorage1Binary   string
 	Vmstorage1Flags    []string
 	Vmstorage2Instance string
+	Vmstorage2Binary   string
 	Vmstorage2Flags    []string
 	VminsertInstance   string
 	VminsertFlags      []string
@@ -287,8 +305,15 @@ type ClusterOptions struct {
 func (tc *TestCase) MustStartCluster(opts *ClusterOptions) *Vmcluster {
 	tc.t.Helper()
 
-	vmstorage1 := tc.MustStartVmstorage(opts.Vmstorage1Instance, opts.Vmstorage1Flags)
-	vmstorage2 := tc.MustStartVmstorage(opts.Vmstorage2Instance, opts.Vmstorage2Flags)
+	if opts.Vmstorage1Binary == "" {
+		opts.Vmstorage1Binary = "../../bin/vmstorage-race"
+	}
+	vmstorage1 := tc.MustStartVmstorageAt(opts.Vmstorage1Instance, opts.Vmstorage1Binary, opts.Vmstorage1Flags)
+
+	if opts.Vmstorage2Binary == "" {
+		opts.Vmstorage2Binary = "../../bin/vmstorage-race"
+	}
+	vmstorage2 := tc.MustStartVmstorageAt(opts.Vmstorage2Instance, opts.Vmstorage2Binary, opts.Vmstorage2Flags)
 
 	opts.VminsertFlags = append(opts.VminsertFlags, []string{
 		"-storageNode=" + vmstorage1.VminsertAddr() + "," + vmstorage2.VminsertAddr(),

apptest/testcase_legacy.go

@@ -1,50 +0,0 @@
-package apptest
-
-// MustStartVmsingle_v1_132_0 is a test helper function that starts an instance
-// of vmsingle-v1.132.0 (last version that uses legacy index) and fails the test
-// if the app fails to start.
-func (tc *TestCase) MustStartVmsingle_v1_132_0(instance string, flags []string) *Vmsingle {
-	tc.t.Helper()
-
-	app, err := StartVmsingle_v1_132_0(instance, flags, tc.cli, tc.output)
-	if err != nil {
-		tc.t.Fatalf("Could not start %s: %v", instance, err)
-	}
-	tc.addApp(instance, app)
-	return app
-}
-
-// MustStartVmstorage_v1_132_0 is a test helper function that starts an instance
-// of vmstorage-v1.132.0 (last version that uses legacy index)  and fails the
-// test if the app fails to start.
-func (tc *TestCase) MustStartVmstorage_v1_132_0(instance string, flags []string) *Vmstorage {
-	tc.t.Helper()
-
-	app, err := StartVmstorage_v1_132_0(instance, flags, tc.cli, tc.output)
-	if err != nil {
-		tc.t.Fatalf("Could not start %s: %v", instance, err)
-	}
-	tc.addApp(instance, app)
-	return app
-}
-
-// MustStartCluster_v1_132_0 starts a cluster with vmstorage-v1.132.0 with
-// custom flags.
-func (tc *TestCase) MustStartCluster_v1_132_0(opts *ClusterOptions) *Vmcluster {
-	tc.t.Helper()
-
-	vmstorage1 := tc.MustStartVmstorage_v1_132_0(opts.Vmstorage1Instance, opts.Vmstorage1Flags)
-	vmstorage2 := tc.MustStartVmstorage_v1_132_0(opts.Vmstorage2Instance, opts.Vmstorage2Flags)
-
-	opts.VminsertFlags = append(opts.VminsertFlags, []string{
-		"-storageNode=" + vmstorage1.VminsertAddr() + "," + vmstorage2.VminsertAddr(),
-	}...)
-	vminsert := tc.MustStartVminsert(opts.VminsertInstance, opts.VminsertFlags)
-
-	opts.VmselectFlags = append(opts.VmselectFlags, []string{
-		"-storageNode=" + vmstorage1.VmselectAddr() + "," + vmstorage2.VmselectAddr(),
-	}...)
-	vmselect := tc.MustStartVmselect(opts.VmselectInstance, opts.VmselectFlags)
-
-	return &Vmcluster{vminsert, vmselect, []*Vmstorage{vmstorage1, vmstorage2}}
-}

apptest/tests/legacy_indexdb_test.go

@@ -2,6 +2,7 @@ package tests
 
 import (
 	"fmt"
+	"os"
 	"path/filepath"
 	"slices"
 	"testing"
@@ -10,6 +11,11 @@ import (
 	at "github.com/VictoriaMetrics/VictoriaMetrics/apptest"
 )
 
+var (
+	legacyVmsinglePath  = os.Getenv("VM_LEGACY_VMSINGLE_PATH")
+	legacyVmstoragePath = os.Getenv("VM_LEGACY_VMSTORAGE_PATH")
+)
+
 type testLegacyDeleteSeriesOpts struct {
 	startLegacySUT func() at.PrometheusWriteQuerier
 	startNewSUT    func() at.PrometheusWriteQuerier
@@ -25,7 +31,7 @@ func TestLegacySingleDeleteSeries(t *testing.T) {
 
 	opts := testLegacyDeleteSeriesOpts{
 		startLegacySUT: func() at.PrometheusWriteQuerier {
-			return tc.MustStartVmsingle_v1_132_0("vmsingle-legacy", []string{
+			return tc.MustStartVmsingleAt("vmsingle-legacy", legacyVmsinglePath, []string{
 				"-storageDataPath=" + storageDataPath,
 				"-retentionPeriod=100y",
 				"-search.maxStalenessInterval=1m",
@@ -58,13 +64,15 @@ func TestLegacyClusterDeleteSeries(t *testing.T) {
 
 	opts := testLegacyDeleteSeriesOpts{
 		startLegacySUT: func() at.PrometheusWriteQuerier {
-			return tc.MustStartCluster_v1_132_0(&at.ClusterOptions{
+			return tc.MustStartCluster(&at.ClusterOptions{
 				Vmstorage1Instance: "vmstorage1-legacy",
+				Vmstorage1Binary:   legacyVmstoragePath,
 				Vmstorage1Flags: []string{
 					"-storageDataPath=" + storage1DataPath,
 					"-retentionPeriod=100y",
 				},
 				Vmstorage2Instance: "vmstorage2-legacy",
+				Vmstorage2Binary:   legacyVmstoragePath,
 				Vmstorage2Flags: []string{
 					"-storageDataPath=" + storage2DataPath,
 					"-retentionPeriod=100y",
@@ -247,7 +255,7 @@ func TestLegacySingleBackupRestore(t *testing.T) {
 
 	opts := testLegacyBackupRestoreOpts{
 		startLegacySUT: func() at.PrometheusWriteQuerier {
-			return tc.MustStartVmsingle_v1_132_0("vmsingle-legacy", []string{
+			return tc.MustStartVmsingleAt("vmsingle-legacy", legacyVmsinglePath, []string{
 				"-storageDataPath=" + storageDataPath,
 				"-retentionPeriod=100y",
 				"-search.disableCache=true",
@@ -290,13 +298,15 @@ func TestLegacyClusterBackupRestore(t *testing.T) {
 
 	opts := testLegacyBackupRestoreOpts{
 		startLegacySUT: func() at.PrometheusWriteQuerier {
-			return tc.MustStartCluster_v1_132_0(&at.ClusterOptions{
+			return tc.MustStartCluster(&at.ClusterOptions{
 				Vmstorage1Instance: "vmstorage1-legacy",
+				Vmstorage1Binary:   legacyVmstoragePath,
 				Vmstorage1Flags: []string{
 					"-storageDataPath=" + storage1DataPath,
 					"-retentionPeriod=100y",
 				},
 				Vmstorage2Instance: "vmstorage2-legacy",
+				Vmstorage2Binary:   legacyVmstoragePath,
 				Vmstorage2Flags: []string{
 					"-storageDataPath=" + storage2DataPath,
 					"-retentionPeriod=100y",
@@ -573,7 +583,7 @@ func TestLegacySingleDowngrade(t *testing.T) {
 
 	opts := testLegacyDowngradeOpts{
 		startLegacySUT: func() at.PrometheusWriteQuerier {
-			return tc.MustStartVmsingle_v1_132_0("vmsingle-legacy", []string{
+			return tc.MustStartVmsingleAt("vmsingle-legacy", legacyVmsinglePath, []string{
 				"-storageDataPath=" + storageDataPath,
 				"-retentionPeriod=100y",
 				"-search.disableCache=true",
@@ -608,13 +618,15 @@ func TestLegacyClusterDowngrade(t *testing.T) {
 
 	opts := testLegacyDowngradeOpts{
 		startLegacySUT: func() at.PrometheusWriteQuerier {
-			return tc.MustStartCluster_v1_132_0(&at.ClusterOptions{
+			return tc.MustStartCluster(&at.ClusterOptions{
 				Vmstorage1Instance: "vmstorage1-legacy",
+				Vmstorage1Binary:   legacyVmstoragePath,
 				Vmstorage1Flags: []string{
 					"-storageDataPath=" + storage1DataPath,
 					"-retentionPeriod=100y",
 				},
 				Vmstorage2Instance: "vmstorage2-legacy",
+				Vmstorage2Binary:   legacyVmstoragePath,
 				Vmstorage2Flags: []string{
 					"-storageDataPath=" + storage2DataPath,
 					"-retentionPeriod=100y",

apptest/vmagent.go

@@ -16,63 +16,43 @@ import (
 	"github.com/golang/snappy"
 )
 
-// StartVmagent starts the latest version of vmagent.
-//
-// The path to the binary can be provided via VMAGENT_PATH environment
-// variable. If the variable is not set, ../../bin/vmagent-race will be
-// used.
+// Vmagent holds the state of a vmagent app and provides vmagent-specific functions
+type Vmagent struct {
+	*app
+	*metricsClient
+
+	httpListenAddr string
+
+	cli *Client
+}
+
+// StartVmagent starts an instance of vmagent with the given flags. It also
+// sets the default flags and populates the app instance state with runtime
+// values extracted from the application log (such as httpListenAddr)
 func StartVmagent(instance string, flags []string, cli *Client, promScrapeConfigFilePath string, output io.Writer) (*Vmagent, error) {
-	binary := os.Getenv("VMAGENT_PATH")
-	if binary == "" {
-		binary = "../../bin/vmagent-race"
+	extractREs := []*regexp.Regexp{
+		httpListenAddrRE,
 	}
-	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
+
+	app, stderrExtracts, err := startApp(instance, "../../bin/vmagent-race", flags, &appOptions{
 		defaultFlags: map[string]string{
 			"-httpListenAddr":          "127.0.0.1:0",
 			"-promscrape.config":       promScrapeConfigFilePath,
 			"-remoteWrite.tmpDataPath": fmt.Sprintf("%s/%s-%d", os.TempDir(), instance, time.Now().UnixNano()),
 		},
-		extractREs: []*regexp.Regexp{
-			httpListenAddrRE,
-		},
-		output: output,
+		extractREs: extractREs,
+		output:     output,
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	return newVmagent(app, cli, vmagentRuntimeValues{
-		httpListenAddr: stderrExtracts[0],
-	}), nil
-}
-
-type vmagentRuntimeValues struct {
-	httpListenAddr string
-}
-
-func newVmagent(app *app, cli *Client, rt vmagentRuntimeValues) *Vmagent {
 	return &Vmagent{
 		app:            app,
+		metricsClient:  newMetricsClient(cli, stderrExtracts[0]),
+		httpListenAddr: stderrExtracts[0],
 		cli:            cli,
-		metricsClient:  newMetricsClient(cli, rt.httpListenAddr),
-		httpListenAddr: rt.httpListenAddr,
-	}
-}
-
-// Vmagent holds the state of a vmagent app and provides vmagent-specific
-// functions.
-type Vmagent struct {
-	*app
-	*metricsClient
-
-	cli            *Client
-	httpListenAddr string
-}
-
-// HTTPAddr returns the address at which the vmagent process is listening
-// for http connections.
-func (app *Vmagent) HTTPAddr() string {
-	return app.httpListenAddr
+	}, nil
 }
 
 // APIV1ImportPrometheus is a test helper function that inserts a
@@ -223,6 +203,12 @@ func (app *Vmagent) PrometheusAPIV1Write(t *testing.T, wr prompb.WriteRequest, o
 	})
 }
 
+// HTTPAddr returns the address at which the vmagent process is listening
+// for http connections.
+func (app *Vmagent) HTTPAddr() string {
+	return app.httpListenAddr
+}
+
 // sendBlocking sends the data to vmstorage by executing `send` function and
 // waits until the data is actually sent.
 //

apptest/vmauth.go

@@ -2,7 +2,6 @@ package apptest
 
 import (
 	"io"
-	"os"
 	"regexp"
 	"syscall"
 	"testing"
@@ -11,48 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 )
 
-// StartVmauth starts the latest version of vmauth.
-//
-// The path to the binary can be provided via VMAUTH_PATH environment
-// variable. If the variable is not set, ../../bin/vmauth-race will be
-// used.
-func StartVmauth(instance string, flags []string, cli *Client, configFilePath string, output io.Writer) (*Vmauth, error) {
-	binary := os.Getenv("VMAUTH_PATH")
-	if binary == "" {
-		binary = "../../bin/vmauth-race"
-	}
-	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
-		defaultFlags: map[string]string{
-			"-httpListenAddr": "127.0.0.1:0",
-			"-auth.config":    configFilePath,
-		},
-		extractREs: []*regexp.Regexp{
-			vmauthHttpListenAddrRE,
-		},
-		output: output,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return newVmauth(app, cli, configFilePath, vmauthRuntimeValues{
-		httpListenAddr: stderrExtracts[0],
-	}), nil
-}
-
-type vmauthRuntimeValues struct {
-	httpListenAddr string
-}
-
-func newVmauth(app *app, cli *Client, configFilePath string, rt vmauthRuntimeValues) *Vmauth {
-	return &Vmauth{
-		app:            app,
-		metricsClient:  newMetricsClient(cli, rt.httpListenAddr),
-		httpListenAddr: rt.httpListenAddr,
-		configFilePath: configFilePath,
-		cli:            cli,
-	}
-}
+var httpBuilitinListenAddrRE = regexp.MustCompile(`pprof handlers are exposed at http://(.*:\d{1,5})/debug/pprof/`)
 
 // Vmauth holds the state of a vmauth app and provides vmauth-specific
 // functions.
@@ -60,14 +18,38 @@ type Vmauth struct {
 	*app
 	*metricsClient
 
-	cli            *Client
 	httpListenAddr string
 	configFilePath string
+	cli            *Client
 }
 
-// GetHTTPListenAddr returns listen http addr
-func (app *Vmauth) GetHTTPListenAddr() string {
-	return app.httpListenAddr
+// StartVmauth starts an instance of vmauth with the given flags. It also
+// sets the default flags and populates the app instance state with runtime
+// values extracted from the application log (such as httpListenAddr)
+func StartVmauth(instance string, flags []string, cli *Client, configFilePath string, output io.Writer) (*Vmauth, error) {
+	extractREs := []*regexp.Regexp{
+		httpBuilitinListenAddrRE,
+	}
+
+	app, stderrExtracts, err := startApp(instance, "../../bin/vmauth-race", flags, &appOptions{
+		defaultFlags: map[string]string{
+			"-httpListenAddr": "127.0.0.1:0",
+			"-auth.config":    configFilePath,
+		},
+		extractREs: extractREs,
+		output:     output,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Vmauth{
+		app:            app,
+		metricsClient:  newMetricsClient(cli, stderrExtracts[0]),
+		httpListenAddr: stderrExtracts[0],
+		configFilePath: configFilePath,
+		cli:            cli,
+	}, nil
 }
 
 // UpdateConfiguration updates the vmauth configuration file with the provided YAML content,
@@ -97,3 +79,8 @@ func (app *Vmauth) UpdateConfiguration(t *testing.T, configFileYAML string) {
 
 	t.Fatalf("config were not reloaded after SIGHUP signal; previous total: %d, current total: %d", prevTotal, currTotal)
 }
+
+// GetHTTPListenAddr returns listen http addr
+func (app *Vmauth) GetHTTPListenAddr() string {
+	return app.httpListenAddr
+}

apptest/vmbackup.go

@@ -1,26 +1,15 @@
 package apptest
 
-import (
-	"io"
-	"os"
-)
+import "io"
 
-// StartVmbackup starts the latest version of vmbackup with the given flags and
-// waits until it exits.
-//
-// The path to the binary can be provided via VMBACKUP_PATH environment
-// variable. If the variable is not set, ../../bin/vmbackup-race will be
-// used.
+// StartVmbackup starts an instance of vmbackup with the given flags and waits
+// until it exits.
 func StartVmbackup(instance, storageDataPath, snapshotCreateURL, dst string, output io.Writer) error {
-	binary := os.Getenv("VMBACKUP_PATH")
-	if binary == "" {
-		binary = "../../bin/vmbackup-race"
-	}
 	flags := []string{
 		"-storageDataPath=" + storageDataPath,
 		"-snapshot.createURL=" + snapshotCreateURL,
 		"-dst=" + dst,
 	}
-	_, _, err := startApp(instance, binary, flags, &appOptions{wait: true, output: output})
+	_, _, err := startApp(instance, "../../bin/vmbackup-race", flags, &appOptions{wait: true, output: output})
 	return err
 }

apptest/vmctl.go

@@ -1,23 +1,9 @@
 package apptest
 
-import (
-	"io"
-	"os"
-)
+import "io"
 
 // StartVmctl starts an instance of vmctl cli with the given flags
-
-// StartVmctl starts the latest version of vmctl with the given flags and
-// waits until it exits.
-//
-// The path to the binary can be provided via VMCTL_PATH environment
-// variable. If the variable is not set, ../../bin/vmctl-race will be
-// used.
 func StartVmctl(instance string, flags []string, output io.Writer) error {
-	binary := os.Getenv("VMCTL_PATH")
-	if binary == "" {
-		binary = "../../bin/vmctl-race"
-	}
-	_, _, err := startApp(instance, binary, flags, &appOptions{wait: true, output: output})
+	_, _, err := startApp(instance, "../../bin/vmctl-race", flags, &appOptions{wait: true, output: output})
 	return err
 }

apptest/vminsert.go

@@ -3,13 +3,23 @@ package apptest
 import (
 	"fmt"
 	"io"
-	"os"
 	"regexp"
 	"strings"
 	"testing"
 	"time"
 )
 
+// Vminsert holds the state of a vminsert app and provides vminsert-specific
+// functions.
+type Vminsert struct {
+	*app
+	*metricsClient
+	*vminsertClient
+
+	httpListenAddr          string
+	clusternativeListenAddr string
+}
+
 // storageNodes returns the storage node addresses passed to vminsert via
 // -storageNode command line flag.
 func storageNodes(flags []string) []string {
@@ -21,11 +31,9 @@ func storageNodes(flags []string) []string {
 	return nil
 }
 
-// StartVminsert starts the latest version of vminsert.
-//
-// The path to the binary can be provided via VMINSERT_PATH environment
-// variable. If the variable is not set, ../../bin/vminsert-race will be
-// used.
+// StartVminsert starts an instance of vminsert with the given flags. It also
+// sets the default flags and populates the app instance state with runtime
+// values extracted from the application log (such as httpListenAddr)
 func StartVminsert(instance string, flags []string, cli *Client, output io.Writer) (*Vminsert, error) {
 	extractREs := []*regexp.Regexp{
 		httpListenAddrRE,
@@ -40,15 +48,11 @@ func StartVminsert(instance string, flags []string, cli *Client, output io.Write
 		extractREs = append(extractREs, regexp.MustCompile(logRecord))
 	}
 
-	binary := os.Getenv("VMINSERT_PATH")
-	if binary == "" {
-		binary = "../../bin/vminsert-race"
-	}
-	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
+	app, stderrExtracts, err := startApp(instance, "../../bin/vminsert-race", flags, &appOptions{
 		defaultFlags: map[string]string{
 			"-httpListenAddr":                              "127.0.0.1:0",
 			"-clusternativeListenAddr":                     "127.0.0.1:0",
-			"-graphiteListenAddr":                          "127.0.0.1:0",
+			"-graphiteListenAddr":                          ":0",
 			"-opentsdbListenAddr":                          "127.0.0.1:0",
 			"-clusternative.vminsertConnsShutdownDuration": "1ms",
 		},
@@ -59,56 +63,27 @@ func StartVminsert(instance string, flags []string, cli *Client, output io.Write
 		return nil, err
 	}
 
-	return newVminsert(app, cli, vminsertRuntimeValues{
+	metricsClient := newMetricsClient(cli, stderrExtracts[0])
+	return &Vminsert{
+		app:           app,
+		metricsClient: metricsClient,
+		vminsertClient: &vminsertClient{
+			vminsertCli: cli,
+			url: func(op, path string, opts QueryOpts) string {
+				return getClusterPath(stderrExtracts[0], op, path, opts)
+			},
+			openTSDBURL: func(op, path string, opts QueryOpts) string {
+				return getClusterPath(stderrExtracts[3], op, path, opts)
+			},
+			graphiteListenAddr: stderrExtracts[2],
+			sendBlocking: func(t *testing.T, numRecordsToSend int, send func()) {
+				t.Helper()
+				sendBlocking(t, metricsClient, numRecordsToSend, send)
+			},
+		},
 		httpListenAddr:          stderrExtracts[0],
 		clusternativeListenAddr: stderrExtracts[1],
-		graphiteListenAddr:      stderrExtracts[2],
-		openTSDBListenAddr:      stderrExtracts[3],
-	}), nil
-}
-
-type vminsertRuntimeValues struct {
-	httpListenAddr          string
-	clusternativeListenAddr string
-	graphiteListenAddr      string
-	openTSDBListenAddr      string
-}
-
-func newVminsert(app *app, cli *Client, rt vminsertRuntimeValues) *Vminsert {
-	metricsClient := newMetricsClient(cli, rt.httpListenAddr)
-	vminsertClient := &vminsertClient{
-		vminsertCli: cli,
-		url: func(op, path string, opts QueryOpts) string {
-			return getClusterPath(rt.httpListenAddr, op, path, opts)
-		},
-		openTSDBURL: func(op, path string, opts QueryOpts) string {
-			return getClusterPath(rt.openTSDBListenAddr, op, path, opts)
-		},
-		graphiteListenAddr: rt.graphiteListenAddr,
-		sendBlocking: func(t *testing.T, numRecordsToSend int, send func()) {
-			t.Helper()
-			sendBlocking(t, metricsClient, numRecordsToSend, send)
-		},
-	}
-
-	return &Vminsert{
-		app:                     app,
-		metricsClient:           metricsClient,
-		vminsertClient:          vminsertClient,
-		httpListenAddr:          rt.httpListenAddr,
-		clusternativeListenAddr: rt.clusternativeListenAddr,
-	}
-}
-
-// Vminsert holds the state of a vminsert app and provides vminsert-specific
-// functions.
-type Vminsert struct {
-	*app
-	*metricsClient
-	*vminsertClient
-
-	httpListenAddr          string
-	clusternativeListenAddr string
+	}, nil
 }
 
 // ClusternativeListenAddr returns the address at which the vminsert process is

apptest/vmrestore.go

@@ -1,25 +1,14 @@
 package apptest
 
-import (
-	"io"
-	"os"
-)
+import "io"
 
-// StartVmrestore starts the latest version of vmrestore with the given flags
-// and waits until it exits.
-//
-// The path to the binary can be provided via VMRESTORE_PATH environment
-// variable. If the variable is not set, ../../bin/vmrestore-race will be
-// used.
+// StartVmrestore starts an instance of vmrestore with the given flags and waits
+// until it exits.
 func StartVmrestore(instance, src, storageDataPath string, output io.Writer) error {
-	binary := os.Getenv("VMRESTORE_PATH")
-	if binary == "" {
-		binary = "../../bin/vmrestore-race"
-	}
 	flags := []string{
 		"-src=" + src,
 		"-storageDataPath=" + storageDataPath,
 	}
-	_, _, err := startApp(instance, binary, flags, &appOptions{wait: true, output: output})
+	_, _, err := startApp(instance, "../../bin/vmrestore-race", flags, &appOptions{wait: true, output: output})
 	return err
 }

apptest/vmselect.go

@@ -3,21 +3,26 @@ package apptest
 import (
 	"fmt"
 	"io"
-	"os"
 	"regexp"
 )
 
-// StartVmselect starts the latest version of vmselect.
-//
-// The path to the binary can be provided via VMSELECT_PATH environment
-// variable. If the variable is not set, ../../bin/vmselect-race will be
-// used.
+// Vmselect holds the state of a vmselect app and provides vmselect-specific
+// functions.
+type Vmselect struct {
+	*app
+	*metricsClient
+	*vmselectClient
+
+	httpListenAddr          string
+	clusternativeListenAddr string
+	cli                     *Client
+}
+
+// StartVmselect starts an instance of vmselect with the given flags. It also
+// sets the default flags and populates the app instance state with runtime
+// values extracted from the application log (such as httpListenAddr)
 func StartVmselect(instance string, flags []string, cli *Client, output io.Writer) (*Vmselect, error) {
-	binary := os.Getenv("VMSELECT_PATH")
-	if binary == "" {
-		binary = "../../bin/vmselect-race"
-	}
-	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
+	app, stderrExtracts, err := startApp(instance, "../../bin/vmselect-race", flags, &appOptions{
 		defaultFlags: map[string]string{
 			"-httpListenAddr":          "127.0.0.1:0",
 			"-clusternativeListenAddr": "127.0.0.1:0",
@@ -32,43 +37,21 @@ func StartVmselect(instance string, flags []string, cli *Client, output io.Write
 		return nil, err
 	}
 
-	return newVmselect(app, cli, vmselectRuntimeValues{
-		httpListenAddr:          stderrExtracts[0],
-		clusternativeListenAddr: stderrExtracts[1],
-	}), nil
-}
-
-type vmselectRuntimeValues struct {
-	httpListenAddr          string
-	clusternativeListenAddr string
-}
-
-func newVmselect(app *app, cli *Client, rt vmselectRuntimeValues) *Vmselect {
 	return &Vmselect{
 		app:           app,
-		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
+		metricsClient: newMetricsClient(cli, stderrExtracts[0]),
 		vmselectClient: &vmselectClient{
 			vmselectCli: cli,
 			url: func(op, path string, opts QueryOpts) string {
-				return getClusterPath(rt.httpListenAddr, op, path, opts)
+				return getClusterPath(stderrExtracts[0], op, path, opts)
 			},
-			metricNamesStatsResetURL: fmt.Sprintf("http://%s/admin/api/v1/admin/status/metric_names_stats/reset", rt.httpListenAddr),
-			tenantsURL:               fmt.Sprintf("http://%s/admin/tenants", rt.httpListenAddr),
+			metricNamesStatsResetURL: fmt.Sprintf("http://%s/admin/api/v1/admin/status/metric_names_stats/reset", stderrExtracts[0]),
+			tenantsURL:               fmt.Sprintf("http://%s/admin/tenants", stderrExtracts[0]),
 		},
-		httpListenAddr:          rt.httpListenAddr,
-		clusternativeListenAddr: rt.clusternativeListenAddr,
-	}
-}
-
-// Vmselect holds the state of a vmselect app and provides vmselect-specific
-// functions.
-type Vmselect struct {
-	*app
-	*metricsClient
-	*vmselectClient
-
-	httpListenAddr          string
-	clusternativeListenAddr string
+		httpListenAddr:          stderrExtracts[0],
+		clusternativeListenAddr: stderrExtracts[1],
+		cli:                     cli,
+	}, nil
 }
 
 // ClusternativeListenAddr returns the address at which the vmselect process is

apptest/vmsingle.go

@@ -9,85 +9,6 @@ import (
 	"time"
 )
 
-// StartVmsingle starts the latest version of vmsingle.
-//
-// The path to the binary can be provided via VMSINGLE_PATH environment
-// variable. If the variable is not set, ../../bin/victoria-metrics-race will be
-// used.
-func StartVmsingle(instance string, flags []string, cli *Client, output io.Writer) (*Vmsingle, error) {
-	binary := os.Getenv("VMSINGLE_PATH")
-	if binary == "" {
-		binary = "../../bin/victoria-metrics-race"
-	}
-	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
-		defaultFlags: map[string]string{
-			"-storageDataPath":    fmt.Sprintf("%s/%s-%d", os.TempDir(), instance, time.Now().UnixNano()),
-			"-httpListenAddr":     "127.0.0.1:0",
-			"-graphiteListenAddr": "127.0.0.1:0",
-			"-opentsdbListenAddr": "127.0.0.1:0",
-		},
-		extractREs: []*regexp.Regexp{
-			storageDataPathRE,
-			httpListenAddrRE,
-			graphiteListenAddrRE,
-			openTSDBListenAddrRE,
-		},
-		output: output,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return newVmsingle(app, cli, vmsingleRuntimeValues{
-		storageDataPath:    stderrExtracts[0],
-		httpListenAddr:     stderrExtracts[1],
-		graphiteListenAddr: stderrExtracts[2],
-		openTSDBListenAddr: stderrExtracts[3],
-	}), nil
-}
-
-type vmsingleRuntimeValues struct {
-	storageDataPath    string
-	httpListenAddr     string
-	graphiteListenAddr string
-	openTSDBListenAddr string
-}
-
-func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
-	return &Vmsingle{
-		app:           app,
-		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
-		vmstorageClient: &vmstorageClient{
-			vmstorageCli:   cli,
-			httpListenAddr: rt.httpListenAddr,
-		},
-		vmselectClient: &vmselectClient{
-			vmselectCli: cli,
-			url: func(op, path string, opts QueryOpts) string {
-				return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
-			},
-			metricNamesStatsResetURL: fmt.Sprintf("http://%s/api/v1/admin/status/metric_names_stats/reset", rt.httpListenAddr),
-			tenantsURL:               "vmsingle-does-not-serve-tenants",
-		},
-		vminsertClient: &vminsertClient{
-			vminsertCli: cli,
-			url: func(_, path string, _ QueryOpts) string {
-				return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
-			},
-			openTSDBURL: func(_, path string, _ QueryOpts) string {
-				return fmt.Sprintf("http://%s/%s", rt.openTSDBListenAddr, path)
-			},
-			graphiteListenAddr: rt.graphiteListenAddr,
-			sendBlocking: func(t *testing.T, _ int, send func()) {
-				t.Helper()
-				send()
-			},
-		},
-		storageDataPath: rt.storageDataPath,
-		httpListenAddr:  rt.httpListenAddr,
-	}
-}
-
 // Vmsingle holds the state of a vmsingle app and provides vmsingle-specific
 // functions.
 type Vmsingle struct {
@@ -101,6 +22,63 @@ type Vmsingle struct {
 	httpListenAddr  string
 }
 
+// StartVmsingleAt starts an instance of vmsingle with the given flags. It also
+// sets the default flags and populates the app instance state with runtime
+// values extracted from the application log (such as httpListenAddr).
+func StartVmsingleAt(instance, binary string, flags []string, cli *Client, output io.Writer) (*Vmsingle, error) {
+	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
+		defaultFlags: map[string]string{
+			"-storageDataPath":    fmt.Sprintf("%s/%s-%d", os.TempDir(), instance, time.Now().UnixNano()),
+			"-httpListenAddr":     "127.0.0.1:0",
+			"-graphiteListenAddr": ":0",
+			"-opentsdbListenAddr": "127.0.0.1:0",
+		},
+		extractREs: []*regexp.Regexp{
+			storageDataPathRE,
+			httpListenAddrRE,
+			graphiteListenAddrRE,
+			openTSDBListenAddrRE,
+		},
+		output: output,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Vmsingle{
+		app:           app,
+		metricsClient: newMetricsClient(cli, stderrExtracts[1]),
+		vmstorageClient: &vmstorageClient{
+			vmstorageCli:   cli,
+			httpListenAddr: stderrExtracts[1],
+		},
+		vmselectClient: &vmselectClient{
+			vmselectCli: cli,
+			url: func(op, path string, opts QueryOpts) string {
+				return fmt.Sprintf("http://%s/%s", stderrExtracts[1], path)
+			},
+			metricNamesStatsResetURL: fmt.Sprintf("http://%s/api/v1/admin/status/metric_names_stats/reset", stderrExtracts[1]),
+			tenantsURL:               "vmsingle-does-not-serve-tenants",
+		},
+		vminsertClient: &vminsertClient{
+			vminsertCli: cli,
+			url: func(_, path string, _ QueryOpts) string {
+				return fmt.Sprintf("http://%s/%s", stderrExtracts[1], path)
+			},
+			openTSDBURL: func(_, path string, _ QueryOpts) string {
+				return fmt.Sprintf("http://%s/%s", stderrExtracts[3], path)
+			},
+			graphiteListenAddr: stderrExtracts[2],
+			sendBlocking: func(t *testing.T, _ int, send func()) {
+				t.Helper()
+				send()
+			},
+		},
+		storageDataPath: stderrExtracts[0],
+		httpListenAddr:  stderrExtracts[1],
+	}, nil
+}
+
 // HTTPAddr returns the address at which the vminsert process is
 // listening for incoming HTTP requests.
 func (app *Vmsingle) HTTPAddr() string {

apptest/vmsingle_legacy.go

@@ -1,43 +0,0 @@
-package apptest
-
-import (
-	"fmt"
-	"io"
-	"os"
-	"regexp"
-	"time"
-)
-
-// StartVmsingle_v1_132_0 starts vmsingle-v1.132.0 (the last version that uses
-// legacy index).
-//
-// The path to the binary must be provided via VMSINGLE_V1_132_0_PATH
-// environment variable.
-func StartVmsingle_v1_132_0(instance string, flags []string, cli *Client, output io.Writer) (*Vmsingle, error) {
-	binary := os.Getenv("VMSINGLE_V1_132_0_PATH")
-	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
-		defaultFlags: map[string]string{
-			"-storageDataPath":    fmt.Sprintf("%s/%s-%d", os.TempDir(), instance, time.Now().UnixNano()),
-			"-httpListenAddr":     "127.0.0.1:0",
-			"-graphiteListenAddr": "127.0.0.1:0",
-			"-opentsdbListenAddr": "127.0.0.1:0",
-		},
-		extractREs: []*regexp.Regexp{
-			storageDataPathRE,
-			httpListenAddrRE,
-			graphiteListenAddrRE,
-			openTSDBListenAddrRE,
-		},
-		output: output,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return newVmsingle(app, cli, vmsingleRuntimeValues{
-		storageDataPath:    stderrExtracts[0],
-		httpListenAddr:     stderrExtracts[1],
-		graphiteListenAddr: stderrExtracts[2],
-		openTSDBListenAddr: stderrExtracts[3],
-	}), nil
-}

apptest/vmstorage.go

@@ -8,22 +8,23 @@ import (
 	"time"
 )
 
-// StartVmstorage starts the latest version of vmstorage.
-//
-// The path to the binary can be provided via VMSTORAGE_PATH environment
-// variable. If the variable is not set, ../../bin/vmstorage-race will be used.
-func StartVmstorage(instance string, flags []string, cli *Client, output io.Writer) (*Vmstorage, error) {
-	binary := os.Getenv("VMSTORAGE_PATH")
-	if binary == "" {
-		binary = "../../bin/vmstorage-race"
-	}
-	return startVmstorage(instance, binary, flags, cli, output)
+// Vmstorage holds the state of a vmstorage app and provides vmstorage-specific
+// functions.
+type Vmstorage struct {
+	*app
+	*metricsClient
+	*vmstorageClient
+
+	storageDataPath string
+	httpListenAddr  string
+	vminsertAddr    string
+	vmselectAddr    string
 }
 
-// startVmstorage starts an instance of vmstorage with the given flags. It also
+// StartVmstorageAt starts an instance of vmstorage with the given flags. It also
 // sets the default flags and populates the app instance state with runtime
 // values extracted from the application log (such as httpListenAddr)
-func startVmstorage(instance, binary string, flags []string, cli *Client, output io.Writer) (*Vmstorage, error) {
+func StartVmstorageAt(instance, binary string, flags []string, cli *Client, output io.Writer) (*Vmstorage, error) {
 	app, stderrExtracts, err := startApp(instance, binary, flags, &appOptions{
 		defaultFlags: map[string]string{
 			"-storageDataPath": fmt.Sprintf("%s/%s-%d", os.TempDir(), instance, time.Now().UnixNano()),
@@ -43,47 +44,18 @@ func startVmstorage(instance, binary string, flags []string, cli *Client, output
 		return nil, err
 	}
 
-	return newVmstorage(app, cli, vmstorageRuntimeValues{
+	return &Vmstorage{
+		app:           app,
+		metricsClient: newMetricsClient(cli, stderrExtracts[1]),
+		vmstorageClient: &vmstorageClient{
+			vmstorageCli:   cli,
+			httpListenAddr: stderrExtracts[1],
+		},
 		storageDataPath: stderrExtracts[0],
 		httpListenAddr:  stderrExtracts[1],
 		vminsertAddr:    stderrExtracts[2],
 		vmselectAddr:    stderrExtracts[3],
-	}), nil
-}
-
-type vmstorageRuntimeValues struct {
-	storageDataPath string
-	httpListenAddr  string
-	vminsertAddr    string
-	vmselectAddr    string
-}
-
-func newVmstorage(app *app, cli *Client, rt vmstorageRuntimeValues) *Vmstorage {
-	return &Vmstorage{
-		app:           app,
-		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
-		vmstorageClient: &vmstorageClient{
-			vmstorageCli:   cli,
-			httpListenAddr: rt.httpListenAddr,
-		},
-		storageDataPath: rt.storageDataPath,
-		httpListenAddr:  rt.httpListenAddr,
-		vminsertAddr:    rt.vminsertAddr,
-		vmselectAddr:    rt.vmselectAddr,
-	}
-}
-
-// Vmstorage holds the state of a vmstorage app and provides vmstorage-specific
-// functions.
-type Vmstorage struct {
-	*app
-	*metricsClient
-	*vmstorageClient
-
-	storageDataPath string
-	httpListenAddr  string
-	vminsertAddr    string
-	vmselectAddr    string
+	}, nil
 }
 
 // VminsertAddr returns the address at which the vmstorage process is listening

apptest/vmstorage_legacy.go

@@ -1,16 +0,0 @@
-package apptest
-
-import (
-	"io"
-	"os"
-)
-
-// StartVmstorage_v1_132_0 starts vmstorage-v1.132.0 (the last version that uses
-// legacy index).
-//
-// The path to the binary must be provided via VMSTORAGE_V1_132_0_PATH
-// environment variable.
-func StartVmstorage_v1_132_0(instance string, flags []string, cli *Client, output io.Writer) (*Vmstorage, error) {
-	binary := os.Getenv("VMSTORAGE_V1_132_0_PATH")
-	return startVmstorage(instance, binary, flags, cli, output)
-}