lib/jwt: fix tests after cb540aaa662511063af3171a370ea17f17121e60

Allow skipping "vm_access" claim validation in order to use claims match based routing. Previously, that required to
modify the token and add an artificial "vm_access: {}" value which is inconvinient.

app/victoria-metrics/main.go

@@ -34,21 +34,8 @@ var (
 		"This can be changed with -promscrape.config.strictParse=false command-line flag")
 	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.")
-	vmselectMaxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
-		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
-		"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
-	vmselectMaxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests "+
-		"limit is reached; see also -search.maxQueryDuration")
 )
 
-func getDefaultMaxConcurrentRequests() int {
-	// A single request can saturate all the CPU cores, so there is no sense
-	// in allowing higher number of concurrent requests - they will just contend
-	// for unavailable CPU time.
-	n := min(cgroup.AvailableCPUs()*2, 16)
-	return n
-}
-
 func main() {
 	// VictoriaMetrics is optimized for reduced memory allocations,
 	// so it can run with the reduced GOGC in order to reduce the used memory,
@@ -89,8 +76,8 @@ func main() {
 	}
 	logger.Infof("starting VictoriaMetrics at %q...", listenAddrs)
 	startTime := time.Now()
-	vmstorage.Init(*vmselectMaxConcurrentRequests, promql.ResetRollupResultCacheIfNeeded)
-	vmselect.Init(*vmselectMaxConcurrentRequests, *vmselectMaxQueueDuration)
+	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
+	vmselect.Init()
 	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)
 	vminsert.Init()
 

app/victoria-metrics/self_scraper.go

@@ -93,7 +93,7 @@ func selfScraper(scrapeInterval time.Duration) {
 				mr.Value = r.Value
 			}
 		}
-		if err := vmstorage.VMInsertAPI.WriteRows(mrs); err != nil {
+		if err := vmstorage.AddRows(mrs); err != nil {
 			logger.Errorf("cannot store self-scraped metrics: %s", err)
 		}
 		if len(metadataRows.Rows) > 0 {
@@ -105,7 +105,7 @@ func selfScraper(scrapeInterval time.Duration) {
 					Type:             mm.Type,
 				})
 			}
-			if err := vmstorage.VMInsertAPI.WriteMetadata(mms); err != nil {
+			if err := vmstorage.AddMetadataRows(mms); err != nil {
 				logger.Errorf("cannot store self-scraped metrics metadata: %s", err)
 			}
 		}

app/vmalert-tool/unittest/input.go

@@ -52,7 +52,7 @@ func writeInputSeries(input []series, interval *promutil.Duration, startStamp ti
 	data := testutil.Compress(r)
 	// write input series to vm
 	httpWrite(dst, bytes.NewBuffer(data))
-	vmstorage.DebugFlush()
+	vmstorage.Storage.DebugFlush()
 	return nil
 }
 

app/vmalert-tool/unittest/unittest.go

@@ -108,9 +108,7 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	storagePath = tmpFolder
 	processFlags()
 	vminsert.Init()
-	const maxConcurrentRequests = 4
-	maxQueueDuration := 5 * time.Second
-	vmselect.Init(maxConcurrentRequests, maxQueueDuration)
+	vmselect.Init()
 	// storagePath will be created again when closing vmselect, so remove it again.
 	defer fs.MustRemoveDir(storagePath)
 	defer vminsert.Stop()
@@ -281,8 +279,7 @@ func processFlags() {
 }
 
 func setUp() {
-	const maxConcurrentRequests = 4
-	vmstorage.Init(maxConcurrentRequests, promql.ResetRollupResultCacheIfNeeded)
+	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	readyCheckFunc := func() bool {
@@ -387,7 +384,7 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 				}
 			}
 			// flush series after each group evaluation
-			vmstorage.DebugFlush()
+			vmstorage.Storage.DebugFlush()
 		}
 
 		// check alert_rule_test case at every eval time

app/vmauth/auth_config.go

@@ -906,14 +906,15 @@ func reloadAuthConfigData(data []byte) (bool, error) {
 		return false, fmt.Errorf("failed to parse auth config: %w", err)
 	}
 
-	jui, oidcDP, err := parseJWTUsers(ac)
+	jui, oidcDP, hasUsersWithSkipVMAccessValidation, err := parseJWTUsers(ac)
 	if err != nil {
 		return false, fmt.Errorf("failed to parse JWT users from auth config: %w", err)
 	}
 	oidcDP.startDiscovery()
 	jwtc := &jwtCache{
-		users:  jui,
-		oidcDP: oidcDP,
+		users:                 jui,
+		oidcDP:                oidcDP,
+		enforceVMAccessClaims: !hasUsersWithSkipVMAccessValidation,
 	}
 
 	m, err := parseAuthConfigUsers(ac)

app/vmauth/example_config.yml

@@ -140,6 +140,17 @@ users:
   - "ProjectID: {{.MetricsProjectID}}"
   url_prefix: "http://vminsert:8480/insert/prometheus"
 
+# JWT-based routing that relies solely on custom claims.
+# skip_vm_access_validation accepts tokens that don't carry a `vm_access` claim,
+# e.g. {"role": "admin"}.
+- name: jwt-no-vm-access
+  jwt:
+    skip_vm_access_validation: true
+    skip_verify: true
+    match_claims:
+      role: admin
+  url_prefix: "http://vmselect-admin:8481/select/0/prometheus"
+
 # Requests without Authorization header are proxied according to `unauthorized_user` section.
 # Requests are proxied in round-robin fashion between `url_prefix` backends.
 # The deny_partial_response query arg is added to all the proxied requests.

app/vmauth/jwt.go

@@ -52,18 +52,22 @@ var urlPathPlaceHolders = []string{
 
 type jwtCache struct {
 	// users contain UserInfo`s from AuthConfig with JWTConfig set
-	users []*UserInfo
-
+	users  []*UserInfo
 	oidcDP *oidcDiscovererPool
+
+	// enforcement of vm_access claim is enabled if there are no users with "skip_vm_access_validation=true"
+	// used for fast rejection path in case of missing "vm_access" claim
+	enforceVMAccessClaims bool
 }
 
 type JWTConfig struct {
-	PublicKeys        []string          `yaml:"public_keys,omitempty"`
-	PublicKeyFiles    []string          `yaml:"public_key_files,omitempty"`
-	SkipVerify        bool              `yaml:"skip_verify,omitempty"`
-	OIDC              *oidcConfig       `yaml:"oidc,omitempty"`
-	MatchClaims       map[string]string `yaml:"match_claims,omitempty"`
-	parsedMatchClaims []*jwt.Claim
+	PublicKeys             []string          `yaml:"public_keys,omitempty"`
+	PublicKeyFiles         []string          `yaml:"public_key_files,omitempty"`
+	SkipVerify             bool              `yaml:"skip_verify,omitempty"`
+	SkipVMAccessValidation bool              `yaml:"skip_vm_access_validation,omitempty"`
+	OIDC                   *oidcConfig       `yaml:"oidc,omitempty"`
+	MatchClaims            map[string]string `yaml:"match_claims,omitempty"`
+	parsedMatchClaims      []*jwt.Claim
 
 	// verifierPool is used to verify JWT tokens.
 	// It is initialized from PublicKeys and/or PublicKeyFiles.
@@ -72,9 +76,10 @@ type JWTConfig struct {
 	verifierPool atomic.Pointer[jwt.VerifierPool]
 }
 
-func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
+func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, bool, error) {
 	jui := make([]*UserInfo, 0, len(ac.Users))
 	oidcDP := &oidcDiscovererPool{}
+	hasUsersWithSkipVMAccessValidation := false
 
 	uniqClaims := make(map[string]*UserInfo)
 	var sortedClaims []string
@@ -85,10 +90,10 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 		}
 
 		if ui.AuthToken != "" || ui.BearerToken != "" || ui.Username != "" || ui.Password != "" {
-			return nil, nil, fmt.Errorf("auth_token, bearer_token, username and password cannot be specified if jwt is set")
+			return nil, nil, false, fmt.Errorf("auth_token, bearer_token, username and password cannot be specified if jwt is set")
 		}
 		if len(jwtToken.PublicKeys) == 0 && len(jwtToken.PublicKeyFiles) == 0 && !jwtToken.SkipVerify && jwtToken.OIDC == nil {
-			return nil, nil, fmt.Errorf("jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true")
+			return nil, nil, false, fmt.Errorf("jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true")
 		}
 		var claimsString string
 		sortedClaims = sortedClaims[:0]
@@ -97,7 +102,7 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 			sortedClaims = append(sortedClaims, fmt.Sprintf("%s=%s", ck, cv))
 			pc, err := jwt.NewClaim(ck, cv)
 			if err != nil {
-				return nil, nil, fmt.Errorf("incorrect match claim, key=%q, value regex=%q: %w", ck, cv, err)
+				return nil, nil, false, fmt.Errorf("incorrect match claim, key=%q, value regex=%q: %w", ck, cv, err)
 			}
 			parsedClaims = append(parsedClaims, pc)
 		}
@@ -106,7 +111,7 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 		claimsString = strings.Join(sortedClaims, ",")
 
 		if oldUI, ok := uniqClaims[claimsString]; ok {
-			return nil, nil, fmt.Errorf("duplicate match claims=%q found for name=%q at idx=%d; the previous one is set for name=%q", claimsString, ui.Name, idx, oldUI.Name)
+			return nil, nil, false, fmt.Errorf("duplicate match claims=%q found for name=%q at idx=%d; the previous one is set for name=%q", claimsString, ui.Name, idx, oldUI.Name)
 		}
 		uniqClaims[claimsString] = &ui
 		if len(jwtToken.PublicKeys) > 0 || len(jwtToken.PublicKeyFiles) > 0 {
@@ -115,7 +120,7 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 			for i := range jwtToken.PublicKeys {
 				k, err := jwt.ParseKey([]byte(jwtToken.PublicKeys[i]))
 				if err != nil {
-					return nil, nil, err
+					return nil, nil, false, err
 				}
 				keys = append(keys, k)
 			}
@@ -123,52 +128,56 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 			for _, filePath := range jwtToken.PublicKeyFiles {
 				keyData, err := os.ReadFile(filePath)
 				if err != nil {
-					return nil, nil, fmt.Errorf("cannot read public key from file %q: %w", filePath, err)
+					return nil, nil, false, fmt.Errorf("cannot read public key from file %q: %w", filePath, err)
 				}
 				k, err := jwt.ParseKey(keyData)
 				if err != nil {
-					return nil, nil, fmt.Errorf("cannot parse public key from file %q: %w", filePath, err)
+					return nil, nil, false, fmt.Errorf("cannot parse public key from file %q: %w", filePath, err)
 				}
 				keys = append(keys, k)
 			}
 
 			vp, err := jwt.NewVerifierPool(keys)
 			if err != nil {
-				return nil, nil, err
+				return nil, nil, false, err
 			}
 
 			jwtToken.verifierPool.Store(vp)
 		}
 		if jwtToken.OIDC != nil {
 			if len(jwtToken.PublicKeys) > 0 || len(jwtToken.PublicKeyFiles) > 0 || jwtToken.SkipVerify {
-				return nil, nil, fmt.Errorf("jwt with oidc cannot contain public keys or have skip_verify=true")
+				return nil, nil, false, fmt.Errorf("jwt with oidc cannot contain public keys or have skip_verify=true")
 			}
 
 			if jwtToken.OIDC.Issuer == "" {
-				return nil, nil, fmt.Errorf("oidc issuer cannot be empty")
+				return nil, nil, false, fmt.Errorf("oidc issuer cannot be empty")
 			}
 			isserURL, err := url.Parse(jwtToken.OIDC.Issuer)
 			if err != nil {
-				return nil, nil, fmt.Errorf("oidc issuer %q must be a valid URL", jwtToken.OIDC.Issuer)
+				return nil, nil, false, fmt.Errorf("oidc issuer %q must be a valid URL", jwtToken.OIDC.Issuer)
 			}
 			if isserURL.Scheme != "https" && isserURL.Scheme != "http" {
-				return nil, nil, fmt.Errorf("oidc issuer %q must have http or https scheme", jwtToken.OIDC.Issuer)
+				return nil, nil, false, fmt.Errorf("oidc issuer %q must have http or https scheme", jwtToken.OIDC.Issuer)
 			}
 
 			oidcDP.createOrAdd(ui.JWT.OIDC.Issuer, &ui.JWT.verifierPool)
 		}
 
 		if err := parseJWTPlaceholdersForUserInfo(&ui, true); err != nil {
-			return nil, nil, err
+			return nil, nil, false, err
 		}
 
 		if err := ui.initURLs(); err != nil {
-			return nil, nil, err
+			return nil, nil, false, err
+		}
+
+		if ui.JWT.SkipVMAccessValidation {
+			hasUsersWithSkipVMAccessValidation = true
 		}
 
 		metricLabels, err := ui.getMetricLabels()
 		if err != nil {
-			return nil, nil, fmt.Errorf("cannot parse metric_labels: %w", err)
+			return nil, nil, false, fmt.Errorf("cannot parse metric_labels: %w", err)
 		}
 		ui.requests = ac.ms.GetOrCreateCounter(`vmauth_user_requests_total` + metricLabels)
 		ui.requestErrors = ac.ms.GetOrCreateCounter(`vmauth_user_request_errors_total` + metricLabels)
@@ -187,7 +196,7 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 
 		rt, err := newRoundTripper(ui.TLSCAFile, ui.TLSCertFile, ui.TLSKeyFile, ui.TLSServerName, ui.TLSInsecureSkipVerify)
 		if err != nil {
-			return nil, nil, fmt.Errorf("cannot initialize HTTP RoundTripper: %w", err)
+			return nil, nil, false, fmt.Errorf("cannot initialize HTTP RoundTripper: %w", err)
 		}
 		ui.rt = rt
 
@@ -200,7 +209,7 @@ func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
 		return len(jui[i].JWT.MatchClaims) > len(jui[j].JWT.MatchClaims)
 	})
 
-	return jui, oidcDP, nil
+	return jui, oidcDP, hasUsersWithSkipVMAccessValidation, nil
 }
 
 var tokenPool sync.Pool
@@ -239,6 +248,12 @@ func getJWTUserInfo(ats []string) (*UserInfo, *jwt.Token) {
 			}
 			continue
 		}
+		if js.enforceVMAccessClaims && !tkn.HasVMAccess() {
+			if *logInvalidAuthTokens {
+				logger.Infof("cannot parse jwt token: %s", jwt.ErrVMAccessFieldMissing)
+			}
+			continue
+		}
 		if tkn.IsExpired(time.Now()) {
 			if *logInvalidAuthTokens {
 				// TODO: add more context:
@@ -259,6 +274,10 @@ func getJWTUserInfo(ats []string) (*UserInfo, *jwt.Token) {
 
 func getUserInfoByJWTToken(tkn *jwt.Token, users []*UserInfo) *UserInfo {
 	for _, ui := range users {
+		if !ui.JWT.SkipVMAccessValidation && !tkn.HasVMAccess() {
+			continue
+		}
+
 		if !tkn.MatchClaims(ui.JWT.parsedMatchClaims) {
 			continue
 		}
@@ -433,7 +452,6 @@ func validateJWTPlaceholdersForURL(up *URLPrefix, isAllowed bool) error {
 			}
 			if strings.Contains(p, placeholderPrefix) {
 				return fmt.Errorf("invalid placeholder found in URL request path: %q, supported values are: %s", bu.Path, strings.Join(allPlaceholders, ", "))
-
 			}
 		}
 		for param, values := range bu.Query() {
@@ -488,7 +506,6 @@ func hasAnyPlaceholders(u *url.URL) bool {
 				return true
 			}
 		}
-
 	}
 	return false
 }

app/vmauth/jwt_test.go

@@ -39,7 +39,7 @@ XOtclIk1uhc03oL9nOQ=
 			}
 			return
 		}
-		users, oidcDP, err := parseJWTUsers(ac)
+		users, oidcDP, _, err := parseJWTUsers(ac)
 		if err == nil {
 			t.Fatalf("expecting non-nil error; got %v", users)
 		}
@@ -326,7 +326,7 @@ XOtclIk1uhc03oL9nOQ=
 			t.Fatalf("unexpected error: %s", err)
 		}
 
-		jui, oidcDP, err := parseJWTUsers(ac)
+		jui, oidcDP, _, err := parseJWTUsers(ac)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}

app/vmauth/main_test.go

@@ -739,6 +739,12 @@ users:
 		"vm_access": map[string]any{},
 	}, false)
 
+	// token without vm_access claim, but with a custom claim usable for routing
+	roleToken := genToken(t, map[string]any{
+		"exp":  time.Now().Add(10 * time.Minute).Unix(),
+		"role": "admin",
+	}, true)
+
 	fullToken := genToken(t, map[string]any{
 		"exp": time.Now().Add(10 * time.Minute).Unix(),
 		"vm_access": map[string]any{
@@ -779,6 +785,39 @@ statusCode=401
 Unauthorized`
 	f(simpleCfgStr, request, responseExpected)
 
+	// token without vm_access claim is rejected even with a matching custom claim
+	request = httptest.NewRequest(`GET`, "http://some-host.com/abc", nil)
+	request.Header.Set(`Authorization`, `Bearer `+roleToken)
+	responseExpected = `
+statusCode=401
+Unauthorized`
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+    match_claims:
+      role: admin
+  url_prefix: {BACKEND}/foo`, string(publicKeyPEM)), request, responseExpected)
+
+	// token without vm_access claim is accepted when skip_vm_access_validation is set
+	request = httptest.NewRequest(`GET`, "http://some-host.com/abc", nil)
+	request.Header.Set(`Authorization`, `Bearer `+roleToken)
+	responseExpected = `
+statusCode=200
+path: /foo/abc
+query:
+headers:`
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+    skip_vm_access_validation: true
+    match_claims:
+      role: admin
+  url_prefix: {BACKEND}/foo`, string(publicKeyPEM)), request, responseExpected)
+
 	// expired token
 	request = httptest.NewRequest(`GET`, "http://some-host.com/abc", nil)
 	request.Header.Set(`Authorization`, `Bearer `+expiredToken)

app/vminsert/common/insert_ctx.go

@@ -184,7 +184,7 @@ func (ctx *InsertCtx) WriteMetadata(mmpbs []prompb.MetricMetadata) error {
 	}
 	ctx.mms = mms
 
-	err := vmstorage.VMInsertAPI.WriteMetadata(mms)
+	err := vmstorage.AddMetadataRows(mms)
 	if err != nil {
 		return &httpserver.ErrorWithStatusCode{
 			Err:        fmt.Errorf("cannot store metrics metadata: %w", err),
@@ -209,7 +209,7 @@ func (ctx *InsertCtx) WritePromMetadata(mmps []prometheus.Metadata) error {
 	}
 	ctx.mms = mms
 
-	err := vmstorage.VMInsertAPI.WriteMetadata(mms)
+	err := vmstorage.AddMetadataRows(mms)
 	if err != nil {
 		return &httpserver.ErrorWithStatusCode{
 			Err:        fmt.Errorf("cannot store prometheus metrics metadata: %w", err),
@@ -278,7 +278,7 @@ func (ctx *InsertCtx) FlushBufs() error {
 	// since the number of concurrent FlushBufs() calls should be already limited via writeconcurrencylimiter
 	// used at every stream.Parse() call under lib/protoparser/*
 
-	err := vmstorage.VMInsertAPI.WriteRows(ctx.mrs)
+	err := vmstorage.AddRows(ctx.mrs)
 	ctx.Reset(0)
 	if err == nil {
 		return nil

app/vminsert/common/streamaggr.go

@@ -283,7 +283,7 @@ func pushAggregateSeries(tss []prompb.TimeSeries) {
 	}
 	// There is no need in limiting the number of concurrent calls to vmstorage.AddRows() here,
 	// since the number of concurrent pushAggregateSeries() calls should be already limited by lib/streamaggr.
-	if err := vmstorage.VMInsertAPI.WriteRows(ctx.mrs); err != nil {
+	if err := vmstorage.AddRows(ctx.mrs); err != nil {
 		logger.Errorf("cannot flush aggregate series: %s", err)
 	}
 }

app/vmselect/graphite/metrics_api.go

@@ -1,6 +1,7 @@
 package graphite
 
 import (
+	"flag"
 	"fmt"
 	"math"
 	"net/http"
@@ -20,6 +21,8 @@ import (
 	"github.com/VictoriaMetrics/metricsql"
 )
 
+var maxTagValueSuffixes = flag.Int("search.maxTagValueSuffixesPerSearch", 100e3, "The maximum number of tag value suffixes returned from /metrics/find")
+
 // MetricsFindHandler implements /metrics/find handler.
 //
 // See https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find
@@ -219,11 +222,10 @@ func MetricsIndexHandler(startTime time.Time, w http.ResponseWriter, r *http.Req
 
 // metricsFind searches for label values that match the given qHead and qTail.
 func metricsFind(tr storage.TimeRange, label, qHead, qTail string, delimiter byte, isExpand bool, deadline searchutil.Deadline) ([]string, error) {
-	maxSuffixes := 0 // let vmstorage use its maxTagValueSuffixesPerSearch limit
 	n := strings.IndexAny(qTail, "*{[")
 	if n < 0 {
 		query := qHead + qTail
-		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, maxSuffixes, deadline)
+		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, *maxTagValueSuffixes, deadline)
 		if err != nil {
 			return nil, err
 		}
@@ -243,7 +245,7 @@ func metricsFind(tr storage.TimeRange, label, qHead, qTail string, delimiter byt
 	}
 	if n == len(qTail)-1 && strings.HasSuffix(qTail, "*") {
 		query := qHead + qTail[:len(qTail)-1]
-		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, maxSuffixes, deadline)
+		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, *maxTagValueSuffixes, deadline)
 		if err != nil {
 			return nil, err
 		}

app/vmselect/graphite/tags_api.go

@@ -138,9 +138,7 @@ func registerMetrics(startTime time.Time, w http.ResponseWriter, r *http.Request
 		mr.MetricNameRaw = storage.MarshalMetricNameRaw(mr.MetricNameRaw[:0], labels)
 		mr.Timestamp = ct
 	}
-	if err := vmstorage.VMSelectAPI.RegisterMetricNames(nil, mrs, 0); err != nil {
-		return err
-	}
+	vmstorage.RegisterMetricNames(nil, mrs)
 
 	// Return response
 	contentType := "text/plain; charset=utf-8"

app/vmselect/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/stats"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -35,6 +36,12 @@ var (
 	deleteAuthKey                = flagutil.NewPassword("deleteAuthKey", "authKey for metrics' deletion via /api/v1/admin/tsdb/delete_series and /tags/delSeries. It could be passed via authKey query arg. It overrides -httpAuth.*")
 	metricNamesStatsResetAuthKey = flagutil.NewPassword("metricNamesStatsResetAuthKey", "authKey for resetting metric names usage cache via /api/v1/admin/status/metric_names_stats/reset. It overrides -httpAuth.*. "+
 		"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#track-ingested-metrics-usage")
+
+	maxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
+		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
+		"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
+	maxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests "+
+		"limit is reached; see also -search.maxQueryDuration")
 	resetCacheAuthKey    = flagutil.NewPassword("search.resetCacheAuthKey", "Optional authKey for resetting rollup cache via /internal/resetRollupResultCache call. It could be passed via authKey query arg. It overrides -httpAuth.*")
 	logSlowQueryDuration = flag.Duration("search.logSlowQueryDuration", 5*time.Second, "Log queries with execution time exceeding this value. Zero disables slow query logging. "+
 		"See also -search.logQueryMemoryUsage")
@@ -43,17 +50,23 @@ var (
 
 var slowQueries = metrics.NewCounter(`vm_slow_queries_total`)
 
+func getDefaultMaxConcurrentRequests() int {
+	// A single request can saturate all the CPU cores, so there is no sense
+	// in allowing higher number of concurrent requests - they will just contend
+	// for unavailable CPU time.
+	n := min(cgroup.AvailableCPUs()*2, 16)
+	return n
+}
+
 // Init initializes vmselect
-func Init(vmselectMaxConcurrentRequests int, vmselectMaxQueueDuration time.Duration) {
+func Init() {
 	tmpDirPath := vmstorage.DataPath() + "/tmp"
 	fs.MustRemoveDirContents(tmpDirPath)
 	netstorage.InitTmpBlocksDir(tmpDirPath)
 	promql.InitRollupResultCache(vmstorage.DataPath() + "/cache/rollupResult")
+	prometheus.InitMaxUniqueTimeseries(*maxConcurrentRequests)
 
-	maxConcurrentRequests = vmselectMaxConcurrentRequests
-	maxQueueDuration = vmselectMaxQueueDuration
-	concurrencyLimitCh = make(chan struct{}, maxConcurrentRequests)
-
+	concurrencyLimitCh = make(chan struct{}, *maxConcurrentRequests)
 	initVMUIConfig()
 	initVMAlertProxy()
 
@@ -65,11 +78,7 @@ func Stop() {
 	promql.StopRollupResultCache()
 }
 
-var (
-	maxConcurrentRequests int
-	maxQueueDuration      time.Duration
-	concurrencyLimitCh    chan struct{}
-)
+var concurrencyLimitCh chan struct{}
 
 var (
 	concurrencyLimitReached = metrics.NewCounter(`vm_concurrent_select_limit_reached_total`)
@@ -81,6 +90,9 @@ var (
 	_ = metrics.NewGauge(`vm_concurrent_select_current`, func() float64 {
 		return float64(len(concurrencyLimitCh))
 	})
+	_ = metrics.NewGauge(`vm_search_max_unique_timeseries`, func() float64 {
+		return float64(prometheus.GetMaxUniqueTimeSeries())
+	})
 )
 
 //go:embed vmui
@@ -119,12 +131,12 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	default:
 		// Sleep for a while until giving up. This should resolve short bursts in requests.
 		concurrencyLimitReached.Inc()
-		d := min(searchutil.GetMaxQueryDuration(r), maxQueueDuration)
+		d := min(searchutil.GetMaxQueryDuration(r), *maxQueueDuration)
 		t := timerpool.Get(d)
 		select {
 		case concurrencyLimitCh <- struct{}{}:
 			timerpool.Put(t)
-			qt.Printf("wait in queue because -search.maxConcurrentRequests=%d concurrent requests are executed", maxConcurrentRequests)
+			qt.Printf("wait in queue because -search.maxConcurrentRequests=%d concurrent requests are executed", *maxConcurrentRequests)
 			defer func() { <-concurrencyLimitCh }()
 		case <-r.Context().Done():
 			timerpool.Put(t)
@@ -140,7 +152,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 				Err: fmt.Errorf("couldn't start executing the request in %.3f seconds, since -search.maxConcurrentRequests=%d concurrent requests "+
 					"are executed. Possible solutions: to reduce query load; to add more compute resources to the server; "+
 					"to increase -search.maxQueueDuration=%s; to increase -search.maxQueryDuration; to increase -search.maxConcurrentRequests",
-					d.Seconds(), maxConcurrentRequests, maxQueueDuration),
+					d.Seconds(), *maxConcurrentRequests, maxQueueDuration),
 				StatusCode: http.StatusTooManyRequests,
 			}
 			w.Header().Add("Retry-After", "10")

app/vmselect/netstorage/netstorage.go

@@ -27,6 +27,10 @@ import (
 )
 
 var (
+	maxTagKeysPerSearch = flag.Int("search.maxTagKeys", 100e3, "The maximum number of tag keys returned from /api/v1/labels . "+
+		"See also -search.maxLabelsAPISeries and -search.maxLabelsAPIDuration")
+	maxTagValuesPerSearch = flag.Int("search.maxTagValues", 100e3, "The maximum number of tag values returned from /api/v1/label/<label_name>/values . "+
+		"See also -search.maxLabelsAPISeries and -search.maxLabelsAPIDuration")
 	maxSamplesPerSeries = flag.Int("search.maxSamplesPerSeries", 30e6, "The maximum number of raw samples a single query can scan per each time series. This option allows limiting memory usage")
 	maxSamplesPerQuery  = flag.Int("search.maxSamplesPerQuery", 1e9, "The maximum number of raw samples a single query can process across all time series. "+
 		"This protects from heavy queries, which select unexpectedly high number of raw samples. See also -search.maxSamplesPerSeries")
@@ -76,7 +80,7 @@ func (rss *Results) Cancel() {
 }
 
 func (rss *Results) mustClose() {
-	vmstorage.PutSearch(rss.sr)
+	putStorageSearch(rss.sr)
 	rss.sr = nil
 	putTmpBlocksFile(rss.tbf)
 	rss.tbf = nil
@@ -754,7 +758,12 @@ var sbhPool sync.Pool
 func DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline searchutil.Deadline) (int, error) {
 	qt = qt.NewChild("delete series: %s", sq)
 	defer qt.Done()
-	return vmstorage.VMSelectAPI.DeleteSeries(qt, sq, deadline.Deadline())
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return 0, err
+	}
+	return vmstorage.DeleteSeries(qt, tfss, sq.MaxMetrics)
 }
 
 // LabelNames returns label names matching the given sq until the given deadline.
@@ -764,7 +773,15 @@ func LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames i
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	labels, err := vmstorage.VMSelectAPI.LabelNames(qt, sq, maxLabelNames, deadline.Deadline())
+	if maxLabelNames > *maxTagKeysPerSearch || maxLabelNames <= 0 {
+		maxLabelNames = *maxTagKeysPerSearch
+	}
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+	labels, err := vmstorage.SearchLabelNames(qt, tfss, tr, maxLabelNames, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during labels search on time range: %w", err)
 	}
@@ -824,7 +841,15 @@ func LabelValues(qt *querytracer.Tracer, labelName string, sq *storage.SearchQue
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	labelValues, err := vmstorage.VMSelectAPI.LabelValues(qt, sq, labelName, maxLabelValues, deadline.Deadline())
+	if maxLabelValues > *maxTagValuesPerSearch || maxLabelValues <= 0 {
+		maxLabelValues = *maxTagValuesPerSearch
+	}
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+	labelValues, err := vmstorage.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during label values search on time range for labelName=%q: %w", labelName, err)
 	}
@@ -839,10 +864,7 @@ func GetMetricsMetadata(qt *querytracer.Tracer, limit int, metricName string) ([
 	qt = qt.NewChild("get metrics metadata: limit=%d, metric_name=%q", limit, metricName)
 	defer qt.Done()
 
-	metadata, err := vmstorage.VMSelectAPI.GetMetadataRecords(qt, nil, limit, metricName, 0)
-	if err != nil {
-		return nil, err
-	}
+	metadata := vmstorage.Storage.GetMetadataRows(qt, limit, metricName)
 
 	sort.Slice(metadata, func(i, j int) bool {
 		return string(metadata[i].MetricFamilyName) < string(metadata[j].MetricFamilyName)
@@ -890,11 +912,16 @@ func TagValueSuffixes(qt *querytracer.Tracer, tr storage.TimeRange, tagKey, tagV
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	suffixes, err := vmstorage.VMSelectAPI.TagValueSuffixes(qt, 0, 0, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline.Deadline())
+	suffixes, err := vmstorage.SearchTagValueSuffixes(qt, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during search for suffixes for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s: %w",
 			tagKey, tagValuePrefix, delimiter, tr.String(), err)
 	}
+	if len(suffixes) >= maxSuffixes {
+		return nil, fmt.Errorf("more than -search.maxTagValueSuffixesPerSearch=%d tag value suffixes found for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s; "+
+			"either narrow down the query or increase -search.maxTagValueSuffixesPerSearch command-line flag value",
+			maxSuffixes, tagKey, tagValuePrefix, delimiter, tr.String())
+	}
 	return suffixes, nil
 }
 
@@ -907,7 +934,13 @@ func TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel stri
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	status, err := vmstorage.VMSelectAPI.TSDBStatus(qt, sq, focusLabel, topN, deadline.Deadline())
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+	date := uint64(tr.MinTimestamp) / (3600 * 24 * 1000)
+	status, err := vmstorage.GetTSDBStatus(qt, tfss, date, focusLabel, topN, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during tsdb status request: %w", err)
 	}
@@ -921,13 +954,28 @@ func SeriesCount(qt *querytracer.Tracer, deadline searchutil.Deadline) (uint64,
 	if deadline.Exceeded() {
 		return 0, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	n, err := vmstorage.VMSelectAPI.SeriesCount(qt, 0, 0, deadline.Deadline())
+	n, err := vmstorage.GetSeriesCount(deadline.Deadline())
 	if err != nil {
 		return 0, fmt.Errorf("error during series count request: %w", err)
 	}
 	return n, nil
 }
 
+func getStorageSearch() *storage.Search {
+	v := ssPool.Get()
+	if v == nil {
+		return &storage.Search{}
+	}
+	return v.(*storage.Search)
+}
+
+func putStorageSearch(sr *storage.Search) {
+	sr.MustClose()
+	ssPool.Put(sr)
+}
+
+var ssPool sync.Pool
+
 // ExportBlocks searches for time series matching sq and calls f for each found block.
 //
 // f is called in parallel from multiple goroutines.
@@ -941,13 +989,18 @@ func ExportBlocks(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline sear
 	if deadline.Exceeded() {
 		return fmt.Errorf("timeout exceeded before starting data export: %s", deadline.String())
 	}
-
 	tr := sq.GetTimeRange()
-	sr, _, err := vmstorage.GetSearch(qt, sq, deadline.Deadline())
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
 	if err != nil {
 		return err
 	}
-	defer vmstorage.PutSearch(sr)
+
+	vmstorage.WG.Add(1)
+	defer vmstorage.WG.Done()
+
+	sr := getStorageSearch()
+	defer putStorageSearch(sr)
+	sr.Init(qt, vmstorage.Storage, tfss, tr, sq.MaxMetrics, deadline.Deadline())
 
 	// Start workers that call f in parallel on available CPU cores.
 	workCh := make(chan *exportWork, gomaxprocs*8)
@@ -1040,7 +1093,14 @@ func SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline
 		return nil, fmt.Errorf("timeout exceeded before starting to search metric names: %s", deadline.String())
 	}
 
-	metricNames, err := vmstorage.VMSelectAPI.SearchMetricNames(qt, sq, deadline.Deadline())
+	// Setup search.
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+
+	metricNames, err := vmstorage.SearchMetricNames(qt, tfss, tr, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("cannot find metric names: %w", err)
 	}
@@ -1059,11 +1119,18 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
 
-	sr, maxSeriesCount, err := vmstorage.GetSearch(qt, sq, deadline.Deadline())
+	// Setup search.
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
 	if err != nil {
 		return nil, err
 	}
 
+	vmstorage.WG.Add(1)
+	defer vmstorage.WG.Done()
+
+	sr := getStorageSearch()
+	maxSeriesCount := sr.Init(qt, vmstorage.Storage, tfss, tr, sq.MaxMetrics, deadline.Deadline())
 	type blockRefs struct {
 		brs []blockRef
 	}
@@ -1101,7 +1168,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		blocksRead++
 		if deadline.Exceeded() {
 			putTmpBlocksFile(tbf)
-			vmstorage.PutSearch(sr)
+			putStorageSearch(sr)
 			return nil, fmt.Errorf("timeout exceeded while fetching data block #%d from storage: %s", blocksRead, deadline.String())
 		}
 		br := sr.MetricBlockRef.BlockRef
@@ -1113,7 +1180,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		samples += br.RowsCount()
 		if *maxSamplesPerQuery > 0 && samples > *maxSamplesPerQuery {
 			putTmpBlocksFile(tbf)
-			vmstorage.PutSearch(sr)
+			putStorageSearch(sr)
 			return nil, fmt.Errorf("cannot select more than -search.maxSamplesPerQuery=%d samples; possible solutions: increase the -search.maxSamplesPerQuery; "+
 				"reduce time range for the query; use more specific label filters in order to select fewer series", *maxSamplesPerQuery)
 		}
@@ -1122,7 +1189,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		addr, err := tbf.WriteBlockRefData(buf)
 		if err != nil {
 			putTmpBlocksFile(tbf)
-			vmstorage.PutSearch(sr)
+			putStorageSearch(sr)
 			return nil, fmt.Errorf("cannot write %d bytes to temporary file: %w", len(buf), err)
 		}
 
@@ -1180,7 +1247,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 
 	if err := sr.Error(); err != nil {
 		putTmpBlocksFile(tbf)
-		vmstorage.PutSearch(sr)
+		putStorageSearch(sr)
 		if errors.Is(err, storage.ErrDeadlineExceeded) {
 			return nil, fmt.Errorf("timeout exceeded during the query: %s", deadline.String())
 		}
@@ -1188,13 +1255,13 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 	}
 	if err := tbf.Finalize(); err != nil {
 		putTmpBlocksFile(tbf)
-		vmstorage.PutSearch(sr)
+		putStorageSearch(sr)
 		return nil, fmt.Errorf("cannot finalize temporary file: %w", err)
 	}
 	qt.Printf("fetch unique series=%d, blocks=%d, samples=%d, bytes=%d", len(m), blocksRead, samples, tbf.Len())
 
 	var rss Results
-	rss.tr = sq.GetTimeRange()
+	rss.tr = tr
 	rss.deadline = deadline
 	pts := make([]packedTimeseries, len(orderedMetricNames))
 	for i, metricName := range orderedMetricNames {
@@ -1235,6 +1302,35 @@ func getBlockRefsEnd(a []blockRef) uintptr {
 	return uintptr(unsafe.Pointer(unsafe.SliceData(a))) + uintptr(len(a))*unsafe.Sizeof(blockRef{})
 }
 
+func setupTfss(qt *querytracer.Tracer, tr storage.TimeRange, tagFilterss [][]storage.TagFilter, maxMetrics int, deadline searchutil.Deadline) ([]*storage.TagFilters, error) {
+	tfss := make([]*storage.TagFilters, 0, len(tagFilterss))
+	for _, tagFilters := range tagFilterss {
+		tfs := storage.NewTagFilters()
+		for i := range tagFilters {
+			tf := &tagFilters[i]
+			if string(tf.Key) == "__graphite__" {
+				query := tf.Value
+				paths, err := vmstorage.SearchGraphitePaths(qt, tr, query, maxMetrics, deadline.Deadline())
+				if err != nil {
+					return nil, fmt.Errorf("error when searching for Graphite paths for query %q: %w", query, err)
+				}
+				if len(paths) >= maxMetrics {
+					return nil, fmt.Errorf("more than %d time series match Graphite query %q; "+
+						"either narrow down the query or increase the corresponding -search.max* command-line flag value; "+
+						"see https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#resource-usage-limits", maxMetrics, query)
+				}
+				tfs.AddGraphiteQuery(query, paths, tf.IsNegative)
+				continue
+			}
+			if err := tfs.Add(tf.Key, tf.Value, tf.IsNegative, tf.IsRegexp); err != nil {
+				return nil, fmt.Errorf("cannot parse tag filter %s: %w", tf, err)
+			}
+		}
+		tfss = append(tfss, tfs)
+	}
+	return tfss, nil
+}
+
 func applyGraphiteRegexpFilter(filter string, ss []string) ([]string, error) {
 	// Anchor filter regexp to the beginning of the string as Graphite does.
 	// See https://github.com/graphite-project/graphite-web/blob/3ad279df5cb90b211953e39161df416e54a84948/webapp/graphite/tags/localdatabase.py#L157
@@ -1261,12 +1357,13 @@ const maxFastAllocBlockSize = 32 * 1024
 func GetMetricNamesStats(qt *querytracer.Tracer, limit, le int, matchPattern string) (metricnamestats.StatsResult, error) {
 	qt = qt.NewChild("get metric names usage statistics with limit: %d, less or equal to: %d, match pattern=%q", limit, le, matchPattern)
 	defer qt.Done()
-	return vmstorage.VMSelectAPI.GetMetricNamesUsageStats(qt, nil, limit, le, matchPattern, 0)
+	return vmstorage.GetMetricNamesStats(qt, limit, le, matchPattern)
 }
 
 // ResetMetricNamesStats resets state of metric names usage
 func ResetMetricNamesStats(qt *querytracer.Tracer) error {
 	qt = qt.NewChild("reset metric names usage stats")
 	defer qt.Done()
-	return vmstorage.VMSelectAPI.ResetMetricNamesUsageStats(qt, 0)
+	vmstorage.ResetMetricNamesStats(qt)
+	return nil
 }

app/vmselect/prometheus/prometheus.go

@@ -28,6 +28,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
@@ -48,6 +50,9 @@ var (
 		"If set to true, the query model becomes closer to InfluxDB data model. If set to true, then -search.maxLookback and -search.maxStalenessInterval are ignored")
 	maxStepForPointsAdjustment = flag.Duration("search.maxStepForPointsAdjustment", time.Minute, "The maximum step when /api/v1/query_range handler adjusts "+
 		"points with timestamps closer than -search.latencyOffset to the current time. The adjustment is needed because such points may contain incomplete data")
+
+	maxUniqueTimeseries = flag.Int("search.maxUniqueTimeseries", 0, "The maximum number of unique time series, which can be selected during /api/v1/query and /api/v1/query_range queries. This option allows limiting memory usage. "+
+		"When set to zero, the limit is automatically calculated based on -search.maxConcurrentRequests (inversely proportional) and memory available to the process (proportional).")
 	maxFederateSeries       = flag.Int("search.maxFederateSeries", 1e6, "The maximum number of time series, which can be returned from /federate. This option allows limiting memory usage")
 	maxExportSeries         = flag.Int("search.maxExportSeries", 10e6, "The maximum number of time series, which can be returned from /api/v1/export* APIs. This option allows limiting memory usage")
 	maxTSDBStatusSeries     = flag.Int("search.maxTSDBStatusSeries", 10e6, "The maximum number of time series, which can be processed during the call to /api/v1/status/tsdb. This option allows limiting memory usage")
@@ -868,7 +873,7 @@ func QueryHandler(qt *querytracer.Tracer, startTime time.Time, w http.ResponseWr
 		End:                 start,
 		Step:                step,
 		MaxPointsPerSeries:  *maxPointsPerTimeseries,
-		MaxSeries:           0, // let vmstorage use maxUniqueTimeseries by default
+		MaxSeries:           GetMaxUniqueTimeSeries(),
 		QuotedRemoteAddr:    httpserver.GetQuotedRemoteAddr(r),
 		Deadline:            deadline,
 		MayCache:            mayCache,
@@ -979,7 +984,7 @@ func queryRangeHandler(qt *querytracer.Tracer, startTime time.Time, w http.Respo
 		End:                 end,
 		Step:                step,
 		MaxPointsPerSeries:  *maxPointsPerTimeseries,
-		MaxSeries:           0, // let vmstorage use maxUniqueTimeseries by default
+		MaxSeries:           GetMaxUniqueTimeSeries(),
 		QuotedRemoteAddr:    httpserver.GetQuotedRemoteAddr(r),
 		Deadline:            deadline,
 		MayCache:            mayCache,
@@ -1315,6 +1320,43 @@ func (sw *scalableWriter) flush() error {
 	return sw.bw.Flush()
 }
 
+var (
+	maxUniqueTimeseriesValueOnce sync.Once
+	maxUniqueTimeseriesValue     int
+)
+
+// InitMaxUniqueTimeseries init the max metrics limit calculated by available resources.
+// The calculation is split into calculateMaxUniqueTimeSeriesForResource for unit testing.
+func InitMaxUniqueTimeseries(maxConcurrentRequests int) {
+	maxUniqueTimeseriesValueOnce.Do(func() {
+		maxUniqueTimeseriesValue = *maxUniqueTimeseries
+		if maxUniqueTimeseriesValue <= 0 {
+			maxUniqueTimeseriesValue = calculateMaxUniqueTimeSeriesForResource(maxConcurrentRequests, memory.Remaining())
+		}
+	})
+}
+
+// calculateMaxUniqueTimeSeriesForResource calculate the max metrics limit calculated by available resources.
+func calculateMaxUniqueTimeSeriesForResource(maxConcurrentRequests, remainingMemory int) int {
+	if maxConcurrentRequests <= 0 {
+		// This line should NOT be reached unless the user has set an incorrect `search.maxConcurrentRequests`.
+		// In such cases, fallback to unlimited.
+		logger.Warnf("limiting -search.maxUniqueTimeseries to %v because -search.maxConcurrentRequests=%d.", 2e9, maxConcurrentRequests)
+		return 2e9
+	}
+
+	// Calculate the max metrics limit for a single request in the worst-case concurrent scenario.
+	// The approximate size of 1 unique series that could occupy in the vmstorage is 200 bytes.
+	mts := remainingMemory / 200 / maxConcurrentRequests
+	logger.Infof("limiting -search.maxUniqueTimeseries to %d according to -search.maxConcurrentRequests=%d and remaining memory=%d bytes. To increase the limit, reduce -search.maxConcurrentRequests or increase memory available to the process.", mts, maxConcurrentRequests, remainingMemory)
+	return mts
+}
+
+// GetMaxUniqueTimeSeries returns the max metrics limit calculated by available resources.
+func GetMaxUniqueTimeSeries() int {
+	return maxUniqueTimeseriesValue
+}
+
 // copied from https://github.com/prometheus/common/blob/adea6285c1c7447fcb7bfdeb6abfc6eff893e0a7/model/metric.go#L483
 // it's not possible to use direct import due to increased binary size
 func unescapePrometheusLabelName(name string) string {

app/vmselect/prometheus/prometheus_test.go

@@ -4,6 +4,7 @@ import (
 	"math"
 	"net/http"
 	"reflect"
+	"runtime"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
@@ -229,3 +230,29 @@ func TestGetLatencyOffsetMillisecondsFailure(t *testing.T) {
 	}
 	f("http://localhost?latency_offset=foobar")
 }
+
+func TestCalculateMaxMetricsLimitByResource(t *testing.T) {
+	f := func(maxConcurrentRequest, remainingMemory, expect int) {
+		t.Helper()
+		maxMetricsLimit := calculateMaxUniqueTimeSeriesForResource(maxConcurrentRequest, remainingMemory)
+		if maxMetricsLimit != expect {
+			t.Fatalf("unexpected max metrics limit: got %d, want %d", maxMetricsLimit, expect)
+		}
+	}
+
+	// Skip when GOARCH=386
+	if runtime.GOARCH != "386" {
+		// 8 CPU & 32 GiB
+		f(16, int(math.Round(32*1024*1024*1024*0.4)), 4294967)
+		// 4 CPU & 32 GiB
+		f(8, int(math.Round(32*1024*1024*1024*0.4)), 8589934)
+	}
+
+	// 2 CPU & 4 GiB
+	f(4, int(math.Round(4*1024*1024*1024*0.4)), 2147483)
+
+	// other edge cases
+	f(0, int(math.Round(4*1024*1024*1024*0.4)), 2e9)
+	f(4, 0, 0)
+
+}

app/vmstorage/main.go

@@ -1,6 +1,7 @@
 package vmstorage
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -8,10 +9,12 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/VictoriaMetrics/metrics"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
@@ -20,9 +23,11 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vminsertapi"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
 )
 
 var (
@@ -34,8 +39,11 @@ var (
 	snapshotAuthKey   = flagutil.NewPassword("snapshotAuthKey", "authKey, which must be passed in query string to /snapshot* pages. It overrides -httpAuth.*")
 	forceMergeAuthKey = flagutil.NewPassword("forceMergeAuthKey", "authKey, which must be passed in query string to /internal/force_merge pages. It overrides -httpAuth.*")
 	forceFlushAuthKey = flagutil.NewPassword("forceFlushAuthKey", "authKey, which must be passed in query string to /internal/force_flush pages. It overrides -httpAuth.*")
+	snapshotsMaxAge   = flagutil.NewRetentionDuration("snapshotsMaxAge", "3d", "Automatically delete snapshots older than -snapshotsMaxAge if it is set to non-zero duration. Make sure that backup process has enough time to finish the backup before the corresponding snapshot is automatically deleted")
 	_                 = flag.Duration("snapshotCreateTimeout", 0, "Deprecated: this flag does nothing")
 
+	precisionBits = flag.Int("precisionBits", 64, "The number of precision bits to store per each value. Lower precision bits improves data compression at the cost of precision loss")
+
 	_ = flag.Duration("finalMergeDelay", 0, "Deprecated: this flag does nothing")
 	_ = flag.Int("bigMergeConcurrency", 0, "Deprecated: this flag does nothing")
 	_ = flag.Int("smallMergeConcurrency", 0, "Deprecated: this flag does nothing")
@@ -109,7 +117,11 @@ func DataPath() string {
 }
 
 // Init initializes vmstorage.
-func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []storage.MetricRow)) {
+func Init(resetCacheIfNeeded func(mrs []storage.MetricRow)) {
+	if err := encoding.CheckPrecisionBits(uint8(*precisionBits)); err != nil {
+		logger.Fatalf("invalid `-precisionBits`: %s", err)
+	}
+
 	storage.SetDedupInterval(*minScrapeInterval)
 	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
 	storage.LegacySetRetentionTimezoneOffset(*retentionTimezoneOffset)
@@ -153,7 +165,7 @@ func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []stora
 		LogNewSeries:                *logNewSeries,
 	}
 	strg := storage.MustOpenStorage(*storageDataPath, opts)
-	vmStorage = newVMStorageSingleNode(strg, vmselectMaxConcurrentRequests, resetCacheIfNeeded)
+	initStaleSnapshotsRemover(strg)
 
 	var m storage.Metrics
 	strg.UpdateMetrics(&m)
@@ -167,32 +179,151 @@ func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []stora
 
 	// register storage metrics
 	storageMetrics = metrics.NewSet()
-	storageMetrics.RegisterMetricsWriter(vmStorage.writeStorageMetrics)
+	storageMetrics.RegisterMetricsWriter(func(w io.Writer) {
+		writeStorageMetrics(w, strg)
+	})
 	metrics.RegisterSet(storageMetrics)
 
-	VMInsertAPI = vmStorage
-	VMSelectAPI = vmStorage
-	GetSearch = vmStorage.GetSearch
-	PutSearch = vmStorage.PutSearch
-	RequestHandler = vmStorage.requestHandler
-	DebugFlush = vmStorage.vms.s.DebugFlush
+	WG = syncwg.WaitGroup{}
+	resetResponseCacheIfNeeded = resetCacheIfNeeded
+	Storage = strg
 }
 
 var storageMetrics *metrics.Set
 
-var (
-	// vmStorageSingleNode is an instance of vmstorage used by vminsert and
-	// vmselect for writing and reading data.
-	vmStorage      *VMStorageSingleNode
-	VMInsertAPI    vminsertapi.API
-	VMSelectAPI    vmselectapi.API
-	GetSearch      func(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error)
-	PutSearch      func(sr *storage.Search)
-	RequestHandler func(w http.ResponseWriter, r *http.Request) bool
+// Storage is a storage.
+//
+// Every storage call must be wrapped into WG.Add(1) ... WG.Done()
+// for proper graceful shutdown when Stop is called.
+var Storage *storage.Storage
 
-	// TODO(@rtm0): Remove this dependency from vmalert-tool unit tests.
-	DebugFlush func()
-)
+// WG must be incremented before Storage call.
+//
+// Use syncwg instead of sync, since Add is called from concurrent goroutines.
+var WG syncwg.WaitGroup
+
+// resetResponseCacheIfNeeded is a callback for automatic resetting of response cache if needed.
+var resetResponseCacheIfNeeded func(mrs []storage.MetricRow)
+
+// AddRows adds mrs to the storage.
+//
+// The caller should limit the number of concurrent calls to AddRows() in order to limit memory usage.
+func AddRows(mrs []storage.MetricRow) error {
+	if Storage.IsReadOnly() {
+		return errReadOnly
+	}
+	resetResponseCacheIfNeeded(mrs)
+	WG.Add(1)
+	Storage.AddRows(mrs, uint8(*precisionBits))
+	WG.Done()
+	return nil
+}
+
+// AddMetadataRows adds mrs to the storage.
+//
+// The caller should limit the number of concurrent calls to AddMetadataRows() in order to limit memory usage.
+func AddMetadataRows(mms []metricsmetadata.Row) error {
+	if Storage.IsReadOnly() {
+		return errReadOnly
+	}
+	WG.Add(1)
+	Storage.AddMetadataRows(mms)
+	WG.Done()
+	return nil
+}
+
+var errReadOnly = errors.New("the storage is in read-only mode; check -storage.minFreeDiskSpaceBytes command-line flag value")
+
+// RegisterMetricNames registers all the metrics from mrs in the storage.
+func RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow) {
+	WG.Add(1)
+	Storage.RegisterMetricNames(qt, mrs)
+	WG.Done()
+}
+
+// DeleteSeries deletes series matching tfss.
+//
+// Returns the number of deleted series.
+func DeleteSeries(qt *querytracer.Tracer, tfss []*storage.TagFilters, maxMetrics int) (int, error) {
+	WG.Add(1)
+	n, err := Storage.DeleteSeries(qt, tfss, maxMetrics)
+	WG.Done()
+	return n, err
+}
+
+// GetMetricNamesStats returns metric names usage stats with give limit and lte predicate
+func GetMetricNamesStats(qt *querytracer.Tracer, limit, le int, matchPattern string) (metricnamestats.StatsResult, error) {
+	WG.Add(1)
+	r := Storage.GetMetricNamesStats(qt, limit, le, matchPattern)
+	WG.Done()
+	return r, nil
+}
+
+// ResetMetricNamesStats resets state for metric names usage tracker
+func ResetMetricNamesStats(qt *querytracer.Tracer) {
+	WG.Add(1)
+	Storage.ResetMetricNamesStats(qt)
+	WG.Done()
+}
+
+// SearchMetricNames returns metric names for the given tfss on the given tr.
+func SearchMetricNames(qt *querytracer.Tracer, tfss []*storage.TagFilters, tr storage.TimeRange, maxMetrics int, deadline uint64) ([]string, error) {
+	WG.Add(1)
+	metricNames, err := Storage.SearchMetricNames(qt, tfss, tr, maxMetrics, deadline)
+	WG.Done()
+	return metricNames, err
+}
+
+// SearchLabelNames searches for tag keys matching the given tfss on tr.
+func SearchLabelNames(qt *querytracer.Tracer, tfss []*storage.TagFilters, tr storage.TimeRange, maxTagKeys, maxMetrics int, deadline uint64) ([]string, error) {
+	WG.Add(1)
+	labelNames, err := Storage.SearchLabelNames(qt, tfss, tr, maxTagKeys, maxMetrics, deadline)
+	WG.Done()
+	return labelNames, err
+}
+
+// SearchLabelValues searches for label values for the given labelName, tfss and
+// tr.
+func SearchLabelValues(qt *querytracer.Tracer, labelName string, tfss []*storage.TagFilters, tr storage.TimeRange, maxLabelValues, maxMetrics int, deadline uint64) ([]string, error) {
+	WG.Add(1)
+	labelValues, err := Storage.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, maxMetrics, deadline)
+	WG.Done()
+	return labelValues, err
+}
+
+// SearchTagValueSuffixes returns all the tag value suffixes for the given tagKey and tagValuePrefix on the given tr.
+//
+// This allows implementing https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find or similar APIs.
+func SearchTagValueSuffixes(qt *querytracer.Tracer, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, maxTagValueSuffixes int, deadline uint64) ([]string, error) {
+	WG.Add(1)
+	suffixes, err := Storage.SearchTagValueSuffixes(qt, tr, tagKey, tagValuePrefix, delimiter, maxTagValueSuffixes, deadline)
+	WG.Done()
+	return suffixes, err
+}
+
+// SearchGraphitePaths returns all the metric names matching the given Graphite query.
+func SearchGraphitePaths(qt *querytracer.Tracer, tr storage.TimeRange, query []byte, maxPaths int, deadline uint64) ([]string, error) {
+	WG.Add(1)
+	paths, err := Storage.SearchGraphitePaths(qt, tr, query, maxPaths, deadline)
+	WG.Done()
+	return paths, err
+}
+
+// GetTSDBStatus returns TSDB status for given filters on the given date.
+func GetTSDBStatus(qt *querytracer.Tracer, tfss []*storage.TagFilters, date uint64, focusLabel string, topN, maxMetrics int, deadline uint64) (*storage.TSDBStatus, error) {
+	WG.Add(1)
+	status, err := Storage.GetTSDBStatus(qt, tfss, date, focusLabel, topN, maxMetrics, deadline)
+	WG.Done()
+	return status, err
+}
+
+// GetSeriesCount returns the number of time series in the storage.
+func GetSeriesCount(deadline uint64) (uint64, error) {
+	WG.Add(1)
+	n, err := Storage.GetSeriesCount(deadline)
+	WG.Done()
+	return n, err
+}
 
 // Stop stops the vmstorage
 func Stop() {
@@ -202,22 +333,17 @@ func Stop() {
 
 	logger.Infof("gracefully closing the storage at %s", *storageDataPath)
 	startTime := time.Now()
-	vmStorage.Stop()
+	WG.WaitAndBlock()
+	stopStaleSnapshotsRemover()
+	Storage.MustClose()
 	logger.Infof("successfully closed the storage in %.3f seconds", time.Since(startTime).Seconds())
 
 	fs.MustStopDirRemover()
-	logger.Infof("the vmstorage has been stopped")
+	logger.Infof("the storage has been stopped")
 }
 
-func (vmssn *VMStorageSingleNode) requestHandler(w http.ResponseWriter, r *http.Request) bool {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.requestHandler(w, r)
-}
-
-// requestHandler is a storage request handler.
-// TODO(@rtm0): Move to a separate file, request_handler.go
-func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) bool {
+// RequestHandler is a storage request handler.
+func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	path := r.URL.Path
 	if path == "/internal/force_merge" {
 		if !httpserver.CheckAuthFlag(w, r, forceMergeAuthKey) {
@@ -230,7 +356,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 			defer activeForceMerges.Dec()
 			logger.Infof("forced merge for partition_prefix=%q has been started", partitionNamePrefix)
 			startTime := time.Now()
-			if err := vms.s.ForceMergePartitions(partitionNamePrefix); err != nil {
+			if err := Storage.ForceMergePartitions(partitionNamePrefix); err != nil {
 				logger.Errorf("error in forced merge for partition_prefix=%q: %s", partitionNamePrefix, err)
 				return
 			}
@@ -243,7 +369,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 			return true
 		}
 		logger.Infof("flushing storage to make pending data available for reading")
-		vms.s.DebugFlush()
+		Storage.DebugFlush()
 		return true
 	}
 
@@ -263,7 +389,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 		}
 		logger.Infof("enabling logging of new series for the next %s. This may increase resource usage during this period.", time.Duration(dealine)*time.Second)
 		endTime := fasttime.UnixTimestamp() + uint64(dealine)
-		vms.s.SetLogNewSeriesUntil(endTime)
+		Storage.SetLogNewSeriesUntil(endTime)
 		fmt.Fprintf(w, `{"status":"success","data":{"logEndTime":%q}}`, time.Unix(int64(endTime), 0))
 		return true
 	}
@@ -285,13 +411,13 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 	case "/create":
 		snapshotsCreateTotal.Inc()
 		w.Header().Set("Content-Type", "application/json")
-		snapshotName := vms.s.MustCreateSnapshot()
+		snapshotName := Storage.MustCreateSnapshot()
 
 		// Verify whether the client already closed the connection.
 		// In this case it is better to drop the created snapshot, since the client isn't interested in it.
 		if err := r.Context().Err(); err != nil {
 			logger.Infof("deleting already created snapshot at %s because the client canceled the request", snapshotName)
-			if err := vms.deleteSnapshot(snapshotName); err != nil {
+			if err := deleteSnapshot(snapshotName); err != nil {
 				logger.Infof("cannot delete just created snapshot: %s", err)
 				return true
 			}
@@ -307,7 +433,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 	case "/list":
 		snapshotsListTotal.Inc()
 		w.Header().Set("Content-Type", "application/json")
-		snapshots := vms.s.MustListSnapshots()
+		snapshots := Storage.MustListSnapshots()
 		fmt.Fprintf(w, `{"status":"ok","snapshots":[`)
 		if len(snapshots) > 0 {
 			for _, snapshot := range snapshots[:len(snapshots)-1] {
@@ -321,7 +447,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 		snapshotsDeleteTotal.Inc()
 		w.Header().Set("Content-Type", "application/json")
 		snapshotName := r.FormValue("snapshot")
-		if err := vms.deleteSnapshot(snapshotName); err != nil {
+		if err := deleteSnapshot(snapshotName); err != nil {
 			jsonResponseError(w, err)
 			snapshotsDeleteErrorsTotal.Inc()
 			return true
@@ -331,9 +457,9 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 	case "/delete_all":
 		snapshotsDeleteAllTotal.Inc()
 		w.Header().Set("Content-Type", "application/json")
-		snapshots := vms.s.MustListSnapshots()
+		snapshots := Storage.MustListSnapshots()
 		for _, snapshotName := range snapshots {
-			if err := vms.s.DeleteSnapshot(snapshotName); err != nil {
+			if err := Storage.DeleteSnapshot(snapshotName); err != nil {
 				err = fmt.Errorf("cannot delete snapshot %q: %w", snapshotName, err)
 				jsonResponseError(w, err)
 				snapshotsDeleteAllErrorsTotal.Inc()
@@ -347,6 +473,50 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 	}
 }
 
+func deleteSnapshot(snapshotName string) error {
+	snapshots := Storage.MustListSnapshots()
+	for _, snName := range snapshots {
+		if snName == snapshotName {
+			if err := Storage.DeleteSnapshot(snName); err != nil {
+				return fmt.Errorf("cannot delete snapshot %q: %w", snName, err)
+			}
+			return nil
+		}
+	}
+	return fmt.Errorf("cannot find snapshot %q", snapshotName)
+}
+
+func initStaleSnapshotsRemover(strg *storage.Storage) {
+	staleSnapshotsRemoverCh = make(chan struct{})
+	if snapshotsMaxAge.Duration() <= 0 {
+		return
+	}
+	snapshotsMaxAgeDur := snapshotsMaxAge.Duration()
+	staleSnapshotsRemoverWG.Go(func() {
+		d := timeutil.AddJitterToDuration(time.Second * 11)
+		t := time.NewTicker(d)
+		defer t.Stop()
+		for {
+			select {
+			case <-staleSnapshotsRemoverCh:
+				return
+			case <-t.C:
+			}
+			strg.MustDeleteStaleSnapshots(snapshotsMaxAgeDur)
+		}
+	})
+}
+
+func stopStaleSnapshotsRemover() {
+	close(staleSnapshotsRemoverCh)
+	staleSnapshotsRemoverWG.Wait()
+}
+
+var (
+	staleSnapshotsRemoverCh chan struct{}
+	staleSnapshotsRemoverWG sync.WaitGroup
+)
+
 var (
 	activeForceMerges = metrics.NewCounter("vm_active_force_merges")
 
@@ -361,16 +531,7 @@ var (
 	snapshotsDeleteAllErrorsTotal = metrics.NewCounter(`vm_http_request_errors_total{path="/snapshot/delete_all"}`)
 )
 
-// TODO(@rtm0): Move to metrics.go.
-func (vmssn *VMStorageSingleNode) writeStorageMetrics(w io.Writer) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	vmssn.vms.writeStorageMetrics(w)
-}
-
-// TODO(@rtm0): Move to metrics.go.
-func (vms *VMStorage) writeStorageMetrics(w io.Writer) {
-	strg := vms.s
+func writeStorageMetrics(w io.Writer, strg *storage.Storage) {
 	var m storage.Metrics
 	strg.UpdateMetrics(&m)
 	tm := &m.TableMetrics
@@ -594,8 +755,6 @@ func (vms *VMStorage) writeStorageMetrics(w io.Writer) {
 	metrics.WriteGaugeUint64(w, `vm_downsampling_partitions_scheduled`, tm.ScheduledDownsamplingPartitions)
 	metrics.WriteGaugeUint64(w, `vm_downsampling_partitions_scheduled_size_bytes`, tm.ScheduledDownsamplingPartitionsSize)
 
-	metrics.WriteGaugeUint64(w, `vm_search_max_unique_timeseries`, uint64(vms.maxUniqueTimeSeriesCalculated))
-
 	metrics.WriteGaugeUint64(w, `vm_metrics_metadata_storage_items`, m.MetadataStorageItemsCurrent)
 	metrics.WriteCounterUint64(w, `vm_metrics_metadata_storage_size_bytes`, m.MetadataStorageCurrentSizeBytes)
 	metrics.WriteCounterUint64(w, `vm_metrics_metadata_storage_max_size_bytes`, m.MetadataStorageMaxSizeBytes)

app/vmstorage/vmstorage.go

@@ -1,391 +0,0 @@
-package vmstorage
-
-import (
-	"flag"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
-)
-
-var (
-	precisionBits = flag.Int("precisionBits", 64, "The number of precision bits to store per each value. Lower precision bits improves data compression "+
-		"at the cost of precision loss")
-	maxUniqueTimeseries = flag.Int("search.maxUniqueTimeseries", 0, "The maximum number of unique time series, which can be scanned during every query. "+
-		"This allows protecting against heavy queries, which select unexpectedly high number of series. When set to zero, the limit is automatically calculated based on -search.maxConcurrentRequests (inversely proportional) and memory available to the process (proportional). See also -search.max* command-line flags at vmselect")
-	maxTagKeys = flag.Int("search.maxTagKeys", 100e3, "The maximum number of tag keys returned per search. "+
-		"See also -search.maxLabelsAPISeries and -search.maxLabelsAPIDuration")
-	maxTagValues = flag.Int("search.maxTagValues", 100e3, "The maximum number of tag values returned per search. "+
-		"See also -search.maxLabelsAPISeries and -search.maxLabelsAPIDuration")
-	maxTagValueSuffixesPerSearch = flag.Int("search.maxTagValueSuffixesPerSearch", 100e3, "The maximum number of tag value suffixes returned from /metrics/find")
-	snapshotsMaxAge              = flagutil.NewRetentionDuration("snapshotsMaxAge", "3d", "Automatically delete snapshots older than -snapshotsMaxAge if it is set to non-zero duration. Make sure that backup process has enough time to finish the backup before the corresponding snapshot is automatically deleted")
-)
-
-// newVMStorage creates a new instance of of VMStorage.
-//
-// The created VMStorage instance takes ownership of s.
-func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int) *VMStorage {
-	if err := encoding.CheckPrecisionBits(uint8(*precisionBits)); err != nil {
-		logger.Fatalf("invalid -precisionBits=%d: %s", *precisionBits, err)
-	}
-
-	maxUniqueTimeseriesCalculated := *maxUniqueTimeseries
-	if maxUniqueTimeseriesCalculated <= 0 {
-		maxUniqueTimeseriesCalculated = calculateMaxUniqueTimeseries(vmselectMaxConcurrentRequests, memory.Remaining())
-	}
-
-	vms := &VMStorage{
-		s:                             s,
-		maxUniqueTimeseries:           *maxUniqueTimeseries,
-		maxUniqueTimeSeriesCalculated: maxUniqueTimeseriesCalculated,
-		staleSnapshotsRemoverCh:       make(chan struct{}),
-	}
-	vms.initStaleSnapshotsRemover()
-	return vms
-}
-
-// calculateMaxUniqueTimeseries calculates the maxUniqueTimeseries based on the
-// available system resources.
-func calculateMaxUniqueTimeseries(maxConcurrentRequests, remainingMemory int) int {
-	if maxConcurrentRequests <= 0 {
-		// This line should NOT be reached unless the user has set an incorrect `search.maxConcurrentRequests`.
-		// In such cases, fallback to unlimited.
-		logger.Warnf("limiting -search.maxUniqueTimeseries to %v because -search.maxConcurrentRequests=%d.", 2e9, maxConcurrentRequests)
-		return 2e9
-	}
-
-	// Calculate the max metrics limit for a single request in the worst-case concurrent scenario.
-	// The approximate size of 1 unique series that could occupy in the vmstorage is 200 bytes.
-	mts := remainingMemory / 200 / maxConcurrentRequests
-	logger.Infof("limiting -search.maxUniqueTimeseries to %d according to -search.maxConcurrentRequests=%d and remaining memory=%d bytes. To increase the limit, reduce -search.maxConcurrentRequests or increase memory available to the process.", mts, maxConcurrentRequests, remainingMemory)
-	return mts
-}
-
-// VMStorage impelements vmselectapi.API and vminsertapi.API.
-type VMStorage struct {
-	s                             *storage.Storage
-	maxUniqueTimeseries           int
-	maxUniqueTimeSeriesCalculated int
-	staleSnapshotsRemoverCh       chan struct{}
-	staleSnapshotsRemoverWG       sync.WaitGroup
-}
-
-func (vms *VMStorage) initStaleSnapshotsRemover() {
-	if snapshotsMaxAge.Duration() <= 0 {
-		return
-	}
-	snapshotsMaxAgeDuration := snapshotsMaxAge.Duration()
-	vms.staleSnapshotsRemoverWG.Go(func() {
-		d := timeutil.AddJitterToDuration(time.Second * 11)
-		t := time.NewTicker(d)
-		defer t.Stop()
-		for {
-			select {
-			case <-vms.staleSnapshotsRemoverCh:
-				return
-			case <-t.C:
-			}
-			vms.s.MustDeleteStaleSnapshots(snapshotsMaxAgeDuration)
-		}
-	})
-}
-
-func (vms *VMStorage) Stop() {
-	close(vms.staleSnapshotsRemoverCh)
-	vms.staleSnapshotsRemoverWG.Wait()
-	vms.s.MustClose()
-}
-
-// WriteRows writes metric rows to the storage.
-//
-// The caller should limit the number of concurrent calls to WriteRows() in
-// order to limit memory usage.
-func (vms *VMStorage) WriteRows(rows []storage.MetricRow) error {
-	vms.s.AddRows(rows, uint8(*precisionBits))
-	return nil
-}
-
-// WriteMetadata writes metrics metadata to storage.
-//
-// The caller should limit the number of concurrent calls to WriteMetadata() in
-// order to limit memory usage.
-func (vms *VMStorage) WriteMetadata(rows []metricsmetadata.Row) error {
-	vms.s.AddMetadataRows(rows)
-	return nil
-}
-
-// IsReadOnly returns true is the storage is in read-only mode.
-func (vms *VMStorage) IsReadOnly() bool {
-	return vms.s.IsReadOnly()
-}
-
-func (vms *VMStorage) InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (vmselectapi.BlockIterator, error) {
-	tr := sq.GetTimeRange()
-	maxMetrics := vms.getMaxMetrics(sq.MaxMetrics)
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		return nil, err
-	}
-	if len(tfss) == 0 {
-		return nil, fmt.Errorf("missing tag filters")
-	}
-	bi := getBlockIterator()
-	bi.sr.Init(qt, vms.s, tfss, tr, maxMetrics, deadline)
-	if err := bi.sr.Error(); err != nil {
-		bi.MustClose()
-		return nil, err
-	}
-	return bi, nil
-}
-
-func (vms *VMStorage) getMaxMetrics(searchQueryLimit int) int {
-	if searchQueryLimit <= 0 {
-		return vms.maxUniqueTimeSeriesCalculated
-	}
-	// searchQueryLimit cannot exceed `-search.maxUniqueTimeseries`
-	if vms.maxUniqueTimeseries != 0 && searchQueryLimit > vms.maxUniqueTimeseries {
-		searchQueryLimit = vms.maxUniqueTimeseries
-	}
-	return searchQueryLimit
-}
-
-// blockIterator implements vmselectapi.BlockIterator
-type blockIterator struct {
-	sr storage.Search
-	mb storage.MetricBlock
-}
-
-var blockIteratorsPool sync.Pool
-
-func (bi *blockIterator) MustClose() {
-	bi.sr.MustClose()
-	bi.mb.MetricName = nil
-	bi.mb.Block.Reset()
-	blockIteratorsPool.Put(bi)
-}
-
-func getBlockIterator() *blockIterator {
-	v := blockIteratorsPool.Get()
-	if v == nil {
-		v = &blockIterator{}
-	}
-	return v.(*blockIterator)
-}
-
-func (bi *blockIterator) NextBlock(dst []byte) ([]byte, bool) {
-	if !bi.sr.NextMetricBlock() {
-		return dst, false
-	}
-	mb := bi.mb
-	mb.MetricName = bi.sr.MetricBlockRef.MetricName
-	bi.sr.MetricBlockRef.BlockRef.MustReadBlock(&mb.Block)
-	dst = mb.Marshal(dst[:0])
-	return dst, true
-}
-
-func (bi *blockIterator) Error() error {
-	return bi.sr.Error()
-}
-
-// SearchMetricNames returns metric names for the given tfss on the given tr.
-func (vms *VMStorage) SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error) {
-	tr := sq.GetTimeRange()
-	maxMetrics := sq.MaxMetrics
-	if maxMetrics <= 0 {
-		// fallback to maxUniqueTimeSeries if no limit is provided,
-		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7857
-		maxMetrics = vms.maxUniqueTimeSeriesCalculated
-	}
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		return nil, err
-	}
-	if len(tfss) == 0 {
-		return nil, fmt.Errorf("missing tag filters")
-	}
-	return vms.s.SearchMetricNames(qt, tfss, tr, maxMetrics, deadline)
-}
-
-// SearchLabelValues searches for label values for the given labelName, tfss and
-// tr.
-func (vms *VMStorage) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error) {
-	tr := sq.GetTimeRange()
-	if maxLabelValues <= 0 || maxLabelValues > *maxTagValues {
-		maxLabelValues = *maxTagValues
-	}
-	maxMetrics := sq.MaxMetrics
-	if maxMetrics <= 0 {
-		// fallback to maxUniqueTimeSeries if no limit is provided,
-		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7857
-		maxMetrics = vms.maxUniqueTimeSeriesCalculated
-	}
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		return nil, err
-	}
-	return vms.s.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, maxMetrics, deadline)
-}
-
-// TagValueSuffixes returns all the tag value suffixes for the given tagKey and
-// tagValuePrefix on the given tr.
-//
-// This allows implementing
-// https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find or
-// similar APIs.
-func (vms *VMStorage) TagValueSuffixes(qt *querytracer.Tracer, _, _ uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte,
-	maxSuffixes int, deadline uint64) ([]string, error) {
-	if maxSuffixes <= 0 || maxSuffixes > *maxTagValueSuffixesPerSearch {
-		maxSuffixes = *maxTagValueSuffixesPerSearch
-	}
-	suffixes, err := vms.s.SearchTagValueSuffixes(qt, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline)
-	if err != nil {
-		return nil, err
-	}
-	if len(suffixes) >= maxSuffixes {
-		return nil, fmt.Errorf("more than -search.maxTagValueSuffixesPerSearch=%d suffixes returned; "+
-			"either narrow down the search or increase -search.maxTagValueSuffixesPerSearch command-line flag value", maxSuffixes)
-	}
-	return suffixes, nil
-}
-
-// SearchLabelNames searches for tag keys matching the given tfss on tr.
-func (vms *VMStorage) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames int, deadline uint64) ([]string, error) {
-	tr := sq.GetTimeRange()
-	if maxLabelNames <= 0 || maxLabelNames > *maxTagKeys {
-		maxLabelNames = *maxTagKeys
-	}
-	maxMetrics := sq.MaxMetrics
-	if maxMetrics <= 0 {
-		// fallback to maxUniqueTimeSeries if no limit is provided,
-		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7857
-		maxMetrics = vms.maxUniqueTimeSeriesCalculated
-	}
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		return nil, err
-	}
-	return vms.s.SearchLabelNames(qt, tfss, tr, maxLabelNames, maxMetrics, deadline)
-}
-
-func (vms *VMStorage) SeriesCount(_ *querytracer.Tracer, _, _ uint32, deadline uint64) (uint64, error) {
-	return vms.s.GetSeriesCount(deadline)
-}
-
-func (vms *VMStorage) Tenants(_ *querytracer.Tracer, _ storage.TimeRange, _ uint64) ([]string, error) {
-	return nil, nil
-}
-
-// GetTSDBStatus returns TSDB status for given filters on the given date.
-func (vms *VMStorage) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error) {
-	tr := sq.GetTimeRange()
-	maxMetrics := sq.MaxMetrics
-	if maxMetrics <= 0 {
-		// fallback to maxUniqueTimeSeries if no limit is provided,
-		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7857
-		maxMetrics = vms.maxUniqueTimeSeriesCalculated
-	}
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		return nil, err
-	}
-	date := uint64(sq.MinTimestamp) / (24 * 3600 * 1000)
-	return vms.s.GetTSDBStatus(qt, tfss, date, focusLabel, topN, maxMetrics, deadline)
-}
-
-// DeleteSeries deletes series matching tfss.
-//
-// Returns the number of deleted series.
-func (vms *VMStorage) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error) {
-	tr := sq.GetTimeRange()
-	maxMetrics := sq.MaxMetrics
-	if maxMetrics <= 0 {
-		// fallback to maxUniqueTimeSeries if no limit is provided,
-		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7857
-		maxMetrics = vms.maxUniqueTimeSeriesCalculated
-	}
-	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		return 0, err
-	}
-	if len(tfss) == 0 {
-		return 0, fmt.Errorf("missing tag filters")
-	}
-	return vms.s.DeleteSeries(qt, tfss, maxMetrics)
-}
-
-func (vms *VMStorage) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, _ uint64) error {
-	vms.s.RegisterMetricNames(qt, mrs)
-	return nil
-}
-
-// GetMetricNamesUsageStats returns metric name usage stats.
-func (vms *VMStorage) GetMetricNamesUsageStats(qt *querytracer.Tracer, _ *storage.TenantToken, limit, le int, matchPattern string, _ uint64) (metricnamestats.StatsResult, error) {
-	return vms.s.GetMetricNamesStats(qt, limit, le, matchPattern), nil
-}
-
-// ResetMetricNamesStats resets state for metric names usage tracker
-func (vms *VMStorage) ResetMetricNamesUsageStats(qt *querytracer.Tracer, _ uint64) error {
-	vms.s.ResetMetricNamesStats(qt)
-	return nil
-}
-
-func (vms *VMStorage) setupTfss(qt *querytracer.Tracer, sq *storage.SearchQuery, tr storage.TimeRange, maxMetrics int, deadline uint64) ([]*storage.TagFilters, error) {
-	tfss := make([]*storage.TagFilters, 0, len(sq.TagFilterss))
-	for _, tagFilters := range sq.TagFilterss {
-		tfs := storage.NewTagFilters()
-		for i := range tagFilters {
-			tf := &tagFilters[i]
-			if string(tf.Key) == "__graphite__" {
-				query := tf.Value
-				qtChild := qt.NewChild("searching for series matching __graphite__=%q", query)
-				paths, err := vms.s.SearchGraphitePaths(qtChild, tr, query, maxMetrics, deadline)
-				qtChild.Donef("found %d series", len(paths))
-				if err != nil {
-					return nil, fmt.Errorf("error when searching for Graphite paths for query %q: %w", query, err)
-				}
-				if len(paths) >= maxMetrics {
-					return nil, fmt.Errorf("more than %d time series match Graphite query %q; "+
-						"either narrow down the query or increase the corresponding -search.max* command-line flag value at vmselect nodes; "+
-						"see https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#resource-usage-limits", maxMetrics, query)
-				}
-				tfs.AddGraphiteQuery(query, paths, tf.IsNegative)
-				continue
-			}
-			if err := tfs.Add(tf.Key, tf.Value, tf.IsNegative, tf.IsRegexp); err != nil {
-				return nil, fmt.Errorf("cannot parse tag filter %s: %w", tf, err)
-			}
-		}
-		tfss = append(tfss, tfs)
-	}
-	return tfss, nil
-}
-
-func (vms *VMStorage) GetMetadataRecords(qt *querytracer.Tracer, _ *storage.TenantToken, limit int, metricName string, _ uint64) ([]*metricsmetadata.Row, error) {
-	return vms.s.GetMetadataRows(qt, limit, metricName), nil
-}
-
-// deleteSnapshot deletes a snapshot by its name.
-//
-// Callers must wrap the call with wg.Add(1)...wg.Done().
-func (vms *VMStorage) deleteSnapshot(snapshotName string) error {
-	snapshots := vms.s.MustListSnapshots()
-	for _, snName := range snapshots {
-		if snName == snapshotName {
-			if err := vms.s.DeleteSnapshot(snName); err != nil {
-				return fmt.Errorf("cannot delete snapshot %q: %w", snName, err)
-			}
-			return nil
-		}
-	}
-	return fmt.Errorf("cannot find snapshot %q", snapshotName)
-}

app/vmstorage/vmstorage_single_node.go

@@ -1,213 +0,0 @@
-package vmstorage
-
-import (
-	"errors"
-	"fmt"
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
-)
-
-// newVMStorageSingleNode creates a new instance of of VMStorage for vmsingle.
-func newVMStorageSingleNode(s *storage.Storage, maxConcurrentRequests int, resetCacheIfNeeded func(mrs []storage.MetricRow)) *VMStorageSingleNode {
-	vms := newVMStorage(s, maxConcurrentRequests)
-	return &VMStorageSingleNode{
-		vms:                vms,
-		wg:                 syncwg.WaitGroup{},
-		resetCacheIfNeeded: resetCacheIfNeeded,
-	}
-}
-
-type VMStorageSingleNode struct {
-	vms *VMStorage
-
-	// wg is used to wrap every storage call into wg.Add(1) ... wg.Done()
-	// for proper graceful shutdown when Stop is called.
-	//
-	// Use syncwg instead of sync, since Add is called from concurrent
-	// goroutines.
-	wg syncwg.WaitGroup
-
-	// resetCacheIfNeeded is a callback for automatic resetting of response
-	// cache if needed.
-	resetCacheIfNeeded func(mrs []storage.MetricRow)
-}
-
-func (vmssn *VMStorageSingleNode) Stop() {
-	vmssn.wg.WaitAndBlock()
-	vmssn.vms.Stop()
-}
-
-// WriteRows writes metric rows to the storage.
-//
-// Returns an error if the storage is in read-only mode.
-//
-// The caller should limit the number of concurrent calls to WriteRows() in
-// order to limit memory usage.
-func (vmssn *VMStorageSingleNode) WriteRows(rows []storage.MetricRow) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-
-	if vmssn.vms.IsReadOnly() {
-		return errReadOnly
-	}
-	vmssn.resetCacheIfNeeded(rows)
-	return vmssn.vms.WriteRows(rows)
-}
-
-// WriteMetadata writes metrics metadata to storage.
-//
-// Returns an error if the storage is in read-only mode.
-//
-// The caller should limit the number of concurrent calls to WriteMetadata() in
-// order to limit memory usage.
-func (vmssn *VMStorageSingleNode) WriteMetadata(rows []metricsmetadata.Row) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-
-	if vmssn.vms.IsReadOnly() {
-		return errReadOnly
-	}
-	return vmssn.vms.WriteMetadata(rows)
-}
-
-var errReadOnly = errors.New("the storage is in read-only mode; check -storage.minFreeDiskSpaceBytes command-line flag value")
-
-func (vmssn *VMStorageSingleNode) IsReadOnly() bool {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.IsReadOnly()
-}
-
-func (vmssn *VMStorageSingleNode) InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (vmselectapi.BlockIterator, error) {
-	return nil, fmt.Errorf("not implemented in vmsingle")
-}
-
-// GetSearch sets up an instance of storage search and returns it to the caller
-// along with the max series count that the search can return.
-//
-// This method is not part of the vmselectapi.API and must only be used by
-// vmsingle HTTP handlers.
-//
-// Callers of this method must call PutSearch() once the search instance is not
-// needed anymore.
-func (vmssn *VMStorageSingleNode) GetSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error) {
-	vmssn.wg.Add(1)
-
-	tr := sq.GetTimeRange()
-	maxMetrics := vmssn.vms.getMaxMetrics(sq.MaxMetrics)
-	tfss, err := vmssn.vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		vmssn.wg.Done()
-		return nil, 0, err
-	}
-
-	sr := getSearch()
-	maxSeriesCount := sr.Init(qt, vmssn.vms.s, tfss, tr, sq.MaxMetrics, deadline)
-	return sr, maxSeriesCount, nil
-}
-
-// PutSearch resets the search once it is not needed anymore and puts it aside
-// for future reuse.
-//
-// This method is not part of the vmselectapi.API and must only be used by
-// vmsingle HTTP handlers.
-//
-// The method must only be used on search instances that have been created with
-// GetSearch().
-func (vmssn *VMStorageSingleNode) PutSearch(sr *storage.Search) {
-	putSearch(sr)
-	vmssn.wg.Done()
-}
-
-func getSearch() *storage.Search {
-	v := ssPool.Get()
-	if v == nil {
-		return &storage.Search{}
-	}
-	return v.(*storage.Search)
-}
-
-func putSearch(sr *storage.Search) {
-	sr.MustClose()
-	ssPool.Put(sr)
-}
-
-var ssPool sync.Pool
-
-func (vmssn *VMStorageSingleNode) SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.SearchMetricNames(qt, sq, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.LabelValues(qt, sq, labelName, maxLabelValues, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) TagValueSuffixes(qt *querytracer.Tracer, accountID, projectID uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, maxSuffixes int, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.TagValueSuffixes(qt, accountID, projectID, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames int, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.LabelNames(qt, sq, maxLabelNames, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) SeriesCount(qt *querytracer.Tracer, accountID, projectID uint32, deadline uint64) (uint64, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.SeriesCount(qt, accountID, projectID, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) Tenants(qt *querytracer.Tracer, tr storage.TimeRange, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.Tenants(qt, tr, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.TSDBStatus(qt, sq, focusLabel, topN, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.DeleteSeries(qt, sq, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, deadline uint64) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.RegisterMetricNames(qt, mrs, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) GetMetricNamesUsageStats(qt *querytracer.Tracer, tt *storage.TenantToken, limit, le int, matchPattern string, deadline uint64) (metricnamestats.StatsResult, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.GetMetricNamesUsageStats(qt, tt, limit, le, matchPattern, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) ResetMetricNamesUsageStats(qt *querytracer.Tracer, deadline uint64) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.ResetMetricNamesUsageStats(qt, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) GetMetadataRecords(qt *querytracer.Tracer, tt *storage.TenantToken, limit int, metricName string, deadline uint64) ([]*metricsmetadata.Row, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.GetMetadataRecords(qt, tt, limit, metricName, deadline)
-}

app/vmstorage/vmstorage_test.go

@@ -1,62 +0,0 @@
-package vmstorage
-
-import (
-	"math"
-	"strconv"
-	"testing"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-)
-
-func TestCalculateMaxMetricsLimitByResource(t *testing.T) {
-	f := func(maxConcurrentRequest, remainingMemory, expect int) {
-		t.Helper()
-		maxMetricsLimit := calculateMaxUniqueTimeseries(maxConcurrentRequest, remainingMemory)
-		if maxMetricsLimit != expect {
-			t.Fatalf("unexpected max metrics limit: got %d, want %d", maxMetricsLimit, expect)
-		}
-	}
-
-	// 64-bit architectures support memory sizes > 4GB.
-	if strconv.IntSize == 64 {
-		// 8 CPU & 32 GiB
-		f(16, int(math.Round(32*1024*1024*1024*0.4)), 4294967)
-		// 4 CPU & 32 GiB
-		f(8, int(math.Round(32*1024*1024*1024*0.4)), 8589934)
-	}
-
-	// 2 CPU & 4 GiB
-	f(4, int(math.Round(4*1024*1024*1024*0.4)), 2147483)
-
-	// other edge cases
-	f(0, int(math.Round(4*1024*1024*1024*0.4)), 2e9)
-	f(4, 0, 0)
-
-}
-
-func TestGetMaxMetrics(t *testing.T) {
-	originalMaxUniqueTimeSeries := *maxUniqueTimeseries
-	defer func() {
-		*maxUniqueTimeseries = originalMaxUniqueTimeSeries
-		fs.MustRemoveDir(t.Name())
-	}()
-
-	maxConcurrentRequests := 2 * cgroup.AvailableCPUs()
-	f := func(searchQueryLimit, storageMaxUniqueTimeseries, expect int) {
-		t.Helper()
-		*maxUniqueTimeseries = storageMaxUniqueTimeseries
-		s := storage.MustOpenStorage(t.Name(), storage.OpenOptions{})
-		vms := newVMStorage(s, maxConcurrentRequests)
-		defer vms.Stop()
-		maxMetrics := vms.getMaxMetrics(searchQueryLimit)
-		if maxMetrics != expect {
-			t.Fatalf("unexpected max metrics: got %d, want %d", maxMetrics, expect)
-		}
-	}
-
-	f(0, 1e6, 1e6)
-	f(2e6, 0, 2e6)
-	f(2e6, 1e6, 1e6)
-}

docs/victoriametrics/changelog/CHANGELOG.md

@@ -33,6 +33,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/) : introduce `vmagent_remotewrite_kafka_outbuf_latency_seconds` and  `vmagent_remotewrite_kafka_rtt_seconds` metrics for [kafka integration](https://docs.victoriametrics.com/victoriametrics/integrations/kafka/). The metrics could help identify throughput bottlenecks. See [#10730](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10730).
 * FEATURE: [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/): properly log user information when a missing route error occurs. See [#11052](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11052).
 * FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl/): add the ability to migrate data from Mimir object storage to VictoriaMetrics. See [#7717](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7717). 
+* FEATURE: [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/): add `skip_vm_access_validation` option for [JWT authorization](https://docs.victoriametrics.com/victoriametrics/vmauth/#jwt-token-auth-proxy) to accept tokens without the mandatory `vm_access` claim. This is useful when routing is built solely on [JWT claim matching](https://docs.victoriametrics.com/victoriametrics/vmauth/#jwt-claim-matching) using other token claims. See [#11054](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11054).
 
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): fix the `Notifiers` page in web UI appearing blank despite the API returning notifier data correctly. See [#11035](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11035).
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): reset the group evaluation timestamp if it exceeds the current host time. Previously, vmalert could use future timestamps for evaluations if the system clock was shifted backward. See [#10985](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10985).

docs/victoriametrics/vmauth.md

@@ -270,7 +270,8 @@ users:
   url_prefix: "http://victoria-metrics:8428/"
 ```
 
-JWT tokens must contain a `"vm_access": {}` claim, more on that in [JWT claim-based request templating](https://docs.victoriametrics.com/victoriametrics/vmauth/#jwt-claim-based-request-templating)
+JWT tokens must contain a `"vm_access": {}` claim, more on that in [JWT claim-based request templating](https://docs.victoriametrics.com/victoriametrics/vmauth/#jwt-claim-based-request-templating).
+This requirement can be relaxed per user with `skip_vm_access_validation`, see [Optional vm_access claim](https://docs.victoriametrics.com/victoriametrics/vmauth/#optional-vm_access-claim).
 
 For testing, skip signature verification with `skip_verify: true` (not recommended for production).
 
@@ -311,6 +312,33 @@ If the OIDC provider is temporarily unavailable during a key refresh, `vmauth` c
 If no keys have been fetched yet (e.g., on startup when the provider is unreachable), the config section is skipped during authentication.
 
 
+#### Optional vm_access claim
+
+By default, `vmauth` rejects JWT tokens that don't contain a `vm_access` claim. When routing is built solely on
+[JWT claim matching](https://docs.victoriametrics.com/victoriametrics/vmauth/#jwt-claim-matching) using other token claims,
+the `vm_access` claim is redundant. Set `skip_vm_access_validation: true`{{% available_from "#" %}} on the `jwt` user
+to accept tokens without a `vm_access` claim:
+
+```yaml
+users:
+- jwt:
+    public_keys:
+    - |
+      -----BEGIN PUBLIC KEY-----
+      MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
+      -----END PUBLIC KEY-----
+    skip_vm_access_validation: true
+    match_claims:
+      role: admin
+  url_prefix: "http://victoria-metrics-admin:8428/"
+```
+
+`skip_vm_access_validation` only relaxes the requirement that the claim is present - the token signature is still verified,
+and a `vm_access` claim is still parsed and applied when present (e.g. for [request templating](https://docs.victoriametrics.com/victoriametrics/vmauth/#jwt-claim-based-request-templating)).
+The setting is per user, so tokens without `vm_access` are accepted only for the matched user that opts in.
+When the claim is absent, the default tenant `0:0` is assumed for any `vm_access`-based placeholders. See [#11054](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11054).
+
+
 #### JWT claim matching
 
 `vmauth` can route requests to different backends depending on the claims contained

lib/jwt/jwt.go

@@ -105,6 +105,10 @@ type body struct {
 	Scope         string `json:"scope,omitempty"`
 	vmAccessClaim VMAccessClaim
 
+	// hasVMAccess is set to true when the token body contains a `vm_access` claim.
+	// Presence enforcement is left to the caller via Token.HasVMAccess.
+	hasVMAccess bool
+
 	buf []byte
 	p   *fastjson.Parser
 
@@ -121,7 +125,6 @@ type body struct {
 }
 
 func (b *body) parse(src string) error {
-
 	var err error
 	b.buf, err = decodeB64(b.buf[:0], src)
 	if err != nil {
@@ -132,6 +135,9 @@ func (b *body) parse(src string) error {
 	if err != nil {
 		return err
 	}
+	if jv.Type() != fastjson.TypeObject {
+		return fmt.Errorf("unexpected non json object; type: %q", jv.Type())
+	}
 	if expObject := jv.Get("exp"); expObject != nil {
 		b.Exp, err = expObject.Int64()
 		if err != nil {
@@ -153,30 +159,31 @@ func (b *body) parse(src string) error {
 	}
 
 	vaObject := jv.Get("vm_access")
-	if vaObject == nil {
-		return ErrVMAccessFieldMissing
-	}
-	// some IDPs encode custom claims as a string
-	// try parsing as an object and fallback to a string
-	switch vaObject.Type() {
-	case fastjson.TypeObject:
-		if err := b.vmAccessClaim.parseFrom(vaObject); err != nil {
-			return err
-		}
-	case fastjson.TypeString:
-		b.claimsParser = parserPool.Get()
-		va, err := b.claimsParser.ParseBytes(vaObject.GetStringBytes())
-		if err != nil {
-			return fmt.Errorf("cannot parse `vm_access` string json: %w", err)
-		}
-		if err := b.vmAccessClaim.parseFrom(va); err != nil {
-			return fmt.Errorf("cannot parse `vm_access` values from string json: %w", err)
-		}
-		b.vmAccessClaimObject = va
-	case fastjson.TypeNull:
-		return ErrVMAccessFieldMissing
+	switch {
+	case vaObject == nil || vaObject.Type() == fastjson.TypeNull:
+		b.hasVMAccess = false
 	default:
-		return fmt.Errorf("unexpected type for `vm_access` field; got: %q, want object {}", vaObject.Type())
+		// some IDPs encode custom claims as a string
+		// try parsing as an object and fallback to a string
+		switch vaObject.Type() {
+		case fastjson.TypeObject:
+			if err := b.vmAccessClaim.parseFrom(vaObject); err != nil {
+				return err
+			}
+		case fastjson.TypeString:
+			b.claimsParser = parserPool.Get()
+			va, err := b.claimsParser.ParseBytes(vaObject.GetStringBytes())
+			if err != nil {
+				return fmt.Errorf("cannot parse `vm_access` string json: %w", err)
+			}
+			if err := b.vmAccessClaim.parseFrom(va); err != nil {
+				return fmt.Errorf("cannot parse `vm_access` values from string json: %w", err)
+			}
+			b.vmAccessClaimObject = va
+		default:
+			return fmt.Errorf("unexpected type for `vm_access` field; got: %q, want object {}", vaObject.Type())
+		}
+		b.hasVMAccess = true
 	}
 	b.Jti = bytesutil.ToUnsafeString(jv.GetStringBytes("jti"))
 
@@ -218,6 +225,7 @@ func (b *body) reset() {
 	b.buf = b.buf[:0]
 	b.allClaims = nil
 	b.vmAccessClaim.reset()
+	b.hasVMAccess = false
 	if b.p != nil {
 		parserPool.Put(b.p)
 		b.p = nil
@@ -229,11 +237,9 @@ func (b *body) reset() {
 	if b.vmAccessClaimObject != nil {
 		b.vmAccessClaimObject = nil
 	}
-
 }
 
 // Parse parses JWT token from given source string
-//
 // Token field is valid until src is reachable
 func (t *Token) Parse(src string, enforceAuthPrefix bool) error {
 	if enforceAuthPrefix && (len(src) < len(prefix) || !strings.EqualFold(src[:len(prefix)], prefix)) {
@@ -268,6 +274,11 @@ func (t *Token) Parse(src string, enforceAuthPrefix bool) error {
 	return nil
 }
 
+// HasVMAccess reports whether the parsed token contains a `vm_access` claim.
+func (t *Token) HasVMAccess() bool {
+	return t.body.hasVMAccess
+}
+
 // Issuer returns `iss` claim value from token body
 func (t *Token) Issuer() string {
 	return t.body.Iss
@@ -425,7 +436,6 @@ func (vac *VMAccessClaim) reset() {
 }
 
 func (vac *VMAccessClaim) parseFrom(jv *fastjson.Value) error {
-
 	if err := vac.Tenant.parseFrom(jv); err != nil {
 		return err
 	}
@@ -569,6 +579,9 @@ func NewToken(auth string, enforceAuthPrefix bool) (*Token, error) {
 	if err := t.parse(jwt[0], jwt[1], jwt[2]); err != nil {
 		return nil, err
 	}
+	if !t.body.hasVMAccess {
+		return nil, ErrVMAccessFieldMissing
+	}
 	return &t, nil
 }
 

lib/jwt/jwt_test.go

@@ -168,17 +168,10 @@ func TestParseJWTBody_Failure(t *testing.T) {
 		true,
 	)
 
-	// invalid body type json
+	// non-object body type
 	f(
 		`[]`,
-		"missing `vm_access` claim",
-		true,
-	)
-
-	// missing vm_access claim
-	f(
-		`{}`,
-		"missing `vm_access` claim",
+		`unexpected non json object; type: "array"`,
 		true,
 	)
 
@@ -189,13 +182,6 @@ func TestParseJWTBody_Failure(t *testing.T) {
 		true,
 	)
 
-	// vm_access claim null
-	f(
-		`{"vm_access": null}`,
-		"missing `vm_access` claim",
-		true,
-	)
-
 	// invalid vm_access: account_id type mismatch
 	f(
 		`{"vm_access": {"tenant_id": {"account_id": "1", "project_id": 5}}}`,
@@ -555,6 +541,33 @@ func TestParseJWTBody_Success(t *testing.T) {
 	)
 }
 
+func TestParseJWTBody_VMAccessPresence(t *testing.T) {
+	f := func(data string, wantHasVMAccess bool) {
+		t.Helper()
+
+		encodedLen := base64.RawURLEncoding.EncodedLen(len(data))
+		encoded := make([]byte, encodedLen)
+		base64.RawURLEncoding.Encode(encoded, []byte(data))
+
+		var b body
+		if err := b.parse(string(encoded)); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if b.hasVMAccess != wantHasVMAccess {
+			t.Fatalf("unexpected hasVMAccess; got %v; want %v", b.hasVMAccess, wantHasVMAccess)
+		}
+	}
+
+	// vm_access claim is present
+	f(`{"vm_access": {}}`, true)
+	f(`{"vm_access": {"metrics_account_id": 1}}`, true)
+
+	// vm_access claim is absent or null - parsing must succeed with hasVMAccess=false
+	f(`{}`, false)
+	f(`{"vm_access": null}`, false)
+	f(`{"role": "admin"}`, false)
+}
+
 func TestNewTokenFromRequest_Failure(t *testing.T) {
 	f := func(r *http.Request) {
 		t.Helper()
@@ -866,7 +879,6 @@ func TestNewTokenFromRequest_Success(t *testing.T) {
 }
 
 func TestTokenMatchClaims(t *testing.T) {
-
 	/*
 		{
 		  "iss": "https://login.microsoftonline.com/-6691-4868-a77b-1b0f9bbe5f43/v2.0",

lib/storage/search.go

@@ -90,77 +90,6 @@ type MetricBlockRef struct {
 	BlockRef *BlockRef
 }
 
-// MetricBlock is a time series block for a single metric.
-type MetricBlock struct {
-	// MetricName is metric name for the given Block.
-	MetricName []byte
-
-	// Block is a block for the given MetricName
-	Block Block
-}
-
-// Marshal marshals MetricBlock to dst
-func (mb *MetricBlock) Marshal(dst []byte) []byte {
-	dst = encoding.MarshalBytes(dst, mb.MetricName)
-	return MarshalBlock(dst, &mb.Block)
-}
-
-// CopyFrom copies src to mb.
-func (mb *MetricBlock) CopyFrom(src *MetricBlock) {
-	mb.MetricName = append(mb.MetricName[:0], src.MetricName...)
-	mb.Block.CopyFrom(&src.Block)
-}
-
-// MarshalBlock marshals b to dst.
-//
-// b.MarshalData must be called on b before calling MarshalBlock.
-func MarshalBlock(dst []byte, b *Block) []byte {
-	dst = b.bh.Marshal(dst)
-	dst = encoding.MarshalBytes(dst, b.timestampsData)
-	dst = encoding.MarshalBytes(dst, b.valuesData)
-	return dst
-}
-
-// Unmarshal unmarshals MetricBlock from src
-func (mb *MetricBlock) Unmarshal(src []byte) ([]byte, error) {
-	mb.Block.Reset()
-	mn, nSize := encoding.UnmarshalBytes(src)
-	if nSize <= 0 {
-		return src, fmt.Errorf("cannot unmarshal MetricName")
-	}
-	src = src[nSize:]
-	mb.MetricName = append(mb.MetricName[:0], mn...)
-
-	return UnmarshalBlock(&mb.Block, src)
-}
-
-// UnmarshalBlock unmarshal Block from src to dst.
-//
-// dst.UnmarshalData isn't called on the block.
-func UnmarshalBlock(dst *Block, src []byte) ([]byte, error) {
-	tail, err := dst.bh.Unmarshal(src)
-	if err != nil {
-		return tail, fmt.Errorf("cannot unmarshal blockHeader: %w", err)
-	}
-	src = tail
-
-	tds, nSize := encoding.UnmarshalBytes(src)
-	if nSize <= 0 {
-		return tail, fmt.Errorf("cannot unmarshal timestampsData")
-	}
-	src = src[nSize:]
-	dst.timestampsData = append(dst.timestampsData[:0], tds...)
-
-	vd, nSize := encoding.UnmarshalBytes(src)
-	if nSize <= 0 {
-		return tail, fmt.Errorf("cannot unmarshal valuesData")
-	}
-	src = src[nSize:]
-	dst.valuesData = append(dst.valuesData[:0], vd...)
-
-	return src, nil
-}
-
 // Search is a search for time series.
 type Search struct {
 	// MetricBlockRef is updated with each Search.NextMetricBlock call.
@@ -361,24 +290,6 @@ func NewSearchQuery(start, end int64, tagFilterss [][]TagFilter, maxMetrics int)
 	}
 }
 
-// TenantToken represents a tenant (accountID, projectID) pair.
-type TenantToken struct {
-	AccountID uint32
-	ProjectID uint32
-}
-
-// String returns string representation of t.
-func (t *TenantToken) String() string {
-	return fmt.Sprintf("{accountID=%d, projectID=%d}", t.AccountID, t.ProjectID)
-}
-
-// Marshal appends marshaled t to dst and returns the result.
-func (t *TenantToken) Marshal(dst []byte) []byte {
-	dst = encoding.MarshalUint32(dst, t.AccountID)
-	dst = encoding.MarshalUint32(dst, t.ProjectID)
-	return dst
-}
-
 // TagFilter represents a single tag filter from SearchQuery.
 type TagFilter struct {
 	Key        []byte

lib/vminsertapi/api.go

@@ -1,30 +0,0 @@
-package vminsertapi
-
-import (
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-)
-
-// RPCCall defines rpc call from vminsert to vmstorage
-type RPCCall struct {
-	Name          string
-	VersionedName string
-}
-
-var (
-	MetricRowsRpcCall = RPCCall{
-		Name:          "metric_rows",
-		VersionedName: "writeRows_v1",
-	}
-	MetricMetadataRpcCall = RPCCall{
-		Name:          "metricmetadata_rows",
-		VersionedName: "writeMetadata_v1",
-	}
-)
-
-// API must implement vminsert API.
-type API interface {
-	WriteRows(rows []storage.MetricRow) error
-	WriteMetadata(mrs []metricsmetadata.Row) error
-	IsReadOnly() bool
-}

lib/vmselectapi/api.go

@@ -1,68 +0,0 @@
-package vmselectapi
-
-import (
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-)
-
-// API must implement vmselect API.
-type API interface {
-	// InitSearch initialize series search for the given sq.
-	//
-	// The returned BlockIterator must be closed with MustClose to free up resources when it is no longer needed.
-	InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (BlockIterator, error)
-
-	// SearchMetricNames returns metric names matching the given sq.
-	SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error)
-
-	// LabelValues returns values for labelName label acorss series matching the given sq.
-	LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error)
-
-	// TagValueSuffixes returns tag value suffixes for the given args.
-	TagValueSuffixes(qt *querytracer.Tracer, accountID, projectID uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, maxSuffixes int, deadline uint64) ([]string, error)
-
-	// LabelNames returns lable names for series matching the given sq.
-	LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLableNames int, deadline uint64) ([]string, error)
-
-	// SeriesCount returns the number of series for the given (accountID, projectID).
-	SeriesCount(qt *querytracer.Tracer, accountID, projectID uint32, deadline uint64) (uint64, error)
-
-	// TSDBStatus returns tsdb status for the given sq.
-	TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error)
-
-	// DeleteSeries deletes series matching the given sq.
-	DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error)
-
-	// RegisterMetricNames registers the given mrs in the storage.
-	RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, deadline uint64) error
-
-	// Tenants returns list of tenants in the storage on the given tr.
-	Tenants(qt *querytracer.Tracer, tr storage.TimeRange, deadline uint64) ([]string, error)
-
-	// GetMetricNamesUsageStats returns statistics for metric names
-	GetMetricNamesUsageStats(qt *querytracer.Tracer, tt *storage.TenantToken, limit, le int, matchPattern string, deadline uint64) (metricnamestats.StatsResult, error)
-
-	// ResetMetricNamesUsageStats resets internal state of metric names tracker
-	ResetMetricNamesUsageStats(qt *querytracer.Tracer, deadline uint64) error
-
-	// GetMetadataRecords returns metrics metadata.
-	GetMetadataRecords(qt *querytracer.Tracer, tt *storage.TenantToken, limit int, metricName string, deadline uint64) ([]*metricsmetadata.Row, error)
-}
-
-// BlockIterator must iterate through series blocks found by VMSelect.InitSearch.
-//
-// MustClose must be called in order to free up allocated resources when BlockIterator is no longer needed.
-type BlockIterator interface {
-	// NextBlock marshals next storage.MetricBlock into dst.
-	//
-	// It returns true on success, false on error or if no blocks to read.
-	NextBlock(dst []byte) ([]byte, bool)
-
-	// MustClose frees up resources allocated by BlockIterator.
-	MustClose()
-
-	// Error returns the last error occurred in NextBlock(), which returns false.
-	Error() error
-}