Repository: VictoriaMetrics/VictoriaMetrics
Pull request #472 (2020-05-12 12:39:19)

Feat/vmalert add max queue size
Im using vmalert-prod online, replacing our own version

find a littile little question, here:

1. some param name is not same with vm other utils, such as vmagent, we make it same
2. default maxQueueSize is not enough for us as we have thounds of alert rule, so make it as a param

thanks

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=h1) Report
> Merging [#472](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=desc) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/8300cc17affde71a6651cdcca8878dd81cf02372&el=desc) will **decrease** coverage by `0.29%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #472      +/-   ##
==========================================
- Coverage   67.93%   67.63%   -0.30%     
==========================================
  Files         148      148              
  Lines       20416    20417       +1     
==========================================
- Hits        13869    13809      -60     
- Misses       5437     5494      +57     
- Partials     1110     1114       +4     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [app/vmalert/main.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvbWFpbi5nbw==) | `0.00% <0.00%> (ø)` | |
| [lib/mergeset/block\_stream\_reader.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L2Jsb2NrX3N0cmVhbV9yZWFkZXIuZ28=) | `50.96% <0.00%> (-11.62%)` | :arrow_down: |
| [lib/storage/merge.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvbWVyZ2UuZ28=) | `79.33% <0.00%> (-4.96%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `66.47% <0.00%> (-2.85%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `97.05% <0.00%> (-1.48%)` | :arrow_down: |
| [lib/storage/part.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydC5nbw==) | `74.50% <0.00%> (-1.31%)` | :arrow_down: |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `75.09% <0.00%> (-0.67%)` | :arrow_down: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `71.63% <0.00%> (+0.23%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=footer). Last update [8300cc1...82f25fc](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/472?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/app/vmagent/remotewrite/client.go b/app/vmagent/remotewrite/client.go
index 093d4d4d3..1da86d441 100644
--- a/app/vmagent/remotewrite/client.go
+++ b/app/vmagent/remotewrite/client.go
@@ -2,11 +2,9 @@ package remotewrite
 
 import (
 	"crypto/tls"
-	"crypto/x509"
 	"encoding/base64"
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"strings"
 	"sync"
 	"time"
@@ -14,6 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/fasthttp"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -28,6 +27,8 @@ var (
 		"If multiple args are set, then they are applied independently for the corresponding -remoteWrite.url")
 	tlsCAFile = flagutil.NewArray("remoteWrite.tlsCAFile", "Optional path to TLS CA file to use for verifying connections to -remoteWrite.url. "+
 		"By default system CA is used. If multiple args are set, then they are applied independently for the corresponding -remoteWrite.url")
+	tlsServerName = flagutil.NewArray("remoteWrite.tlsServerName", "Optional TLS server name to use for connections to -remoteWrite.url. "+
+		"By default the server name from -remoteWrite.url is used. If multiple args are set, then they are applied independently for the corresponding -remoteWrite.url")
 
 	basicAuthUsername = flagutil.NewArray("remoteWrite.basicAuth.username", "Optional basic auth username to use for -remoteWrite.url. "+
 		"If multiple args are set, then they are applied independently for the corresponding -remoteWrite.url")
@@ -152,37 +153,18 @@ func (c *client) MustStop() {
 }
 
 func getTLSConfig(argIdx int) (*tls.Config, error) {
-	var tlsRootCA *x509.CertPool
-	var tlsCertificate *tls.Certificate
-	certFile := tlsCertFile.GetOptionalArg(argIdx)
-	keyFile := tlsKeyFile.GetOptionalArg(argIdx)
-	if certFile != "" || keyFile != "" {
-		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
-		if err != nil {
-			return nil, fmt.Errorf("cannot load TLS certificate for -remoteWrite.tlsCertFile=%q and -remoteWrite.tlsKeyFile=%q: %s", certFile, keyFile, err)
-		}
-		tlsCertificate = &cert
+	tlsConfig := &promauth.TLSConfig{
+		CAFile:             tlsCAFile.GetOptionalArg(argIdx),
+		CertFile:           tlsCertFile.GetOptionalArg(argIdx),
+		KeyFile:            tlsKeyFile.GetOptionalArg(argIdx),
+		ServerName:         tlsServerName.GetOptionalArg(argIdx),
+		InsecureSkipVerify: *tlsInsecureSkipVerify,
 	}
-	if caFile := tlsCAFile.GetOptionalArg(argIdx); caFile != "" {
-		data, err := ioutil.ReadFile(caFile)
-		if err != nil {
-			return nil, fmt.Errorf("cannot read -remoteWrite.tlsCAFile=%q: %s", caFile, err)
-		}
-		tlsRootCA = x509.NewCertPool()
-		if !tlsRootCA.AppendCertsFromPEM(data) {
-			return nil, fmt.Errorf("cannot parse data -remoteWrite.tlsCAFile=%q", caFile)
-		}
-	}
-	tlsCfg := &tls.Config{
-		RootCAs:            tlsRootCA,
-		ClientSessionCache: tls.NewLRUClientSessionCache(0),
-	}
-	if tlsCertificate != nil {
-		tlsCfg.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
-			return tlsCertificate, nil
-		}
+	cfg, err := promauth.NewConfig(".", nil, "", "", tlsConfig)
+	if err != nil {
+		return nil, fmt.Errorf("cannot populate TLS config: %s", err)
 	}
-	tlsCfg.InsecureSkipVerify = *tlsInsecureSkipVerify
+	tlsCfg := cfg.NewTLSConfig()
 	return tlsCfg, nil
 }
 
diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go
index 6a5df13dc..a15308996 100644
--- a/app/vmagent/remotewrite/remotewrite.go
+++ b/app/vmagent/remotewrite/remotewrite.go
@@ -87,6 +87,8 @@ func Stop() {
 // Push sends wr to remote storage systems set via `-remoteWrite.url`.
 //
 // Each timeseries in wr.Timeseries must contain one sample.
+//
+// Note that wr may be modified by Push due to relabeling.
 func Push(wr *prompbmarshal.WriteRequest) {
 	var rctx *relabelCtx
 	if len(prcsGlobal) > 0 || len(labelsGlobal) > 0 {
@@ -128,6 +130,8 @@ type remoteWriteCtx struct {
 	pss        []*pendingSeries
 	pssNextIdx uint64
 
+	tss []prompbmarshal.TimeSeries
+
 	relabelMetricsDropped *metrics.Counter
 }
 
@@ -181,6 +185,11 @@ func (rwctx *remoteWriteCtx) MustStop() {
 func (rwctx *remoteWriteCtx) Push(tss []prompbmarshal.TimeSeries) {
 	var rctx *relabelCtx
 	if len(rwctx.prcs) > 0 {
+		// Make a copy of tss before applying relabeling in order to prevent
+		// from affecting time series for other remoteWrite.url configs.
+		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/467 for details.
+		rwctx.tss = append(rwctx.tss[:0], tss...)
+		tss = rwctx.tss
 		rctx = getRelabelCtx()
 		tssLen := len(tss)
 		tss = rctx.applyRelabeling(tss, nil, rwctx.prcs)
@@ -191,5 +200,7 @@ func (rwctx *remoteWriteCtx) Push(tss []prompbmarshal.TimeSeries) {
 	pss[idx].Push(tss)
 	if rctx != nil {
 		putRelabelCtx(rctx)
+		// Zero rwctx.tss in order to free up GC references.
+		rwctx.tss = prompbmarshal.ResetTimeSeries(rwctx.tss)
 	}
 }
diff --git a/app/vmalert/README.md b/app/vmalert/README.md
index f4dc3bc1c..4417a6746 100644
--- a/app/vmalert/README.md
+++ b/app/vmalert/README.md
@@ -94,6 +94,9 @@ Usage of vmalert:
         Optional basic auth password for -remotewrite.url
   -remotewrite.basicAuth.username string
         Optional basic auth username for -remotewrite.url
+  -remoteWrite.maxQueueSize
+	    Optional Defines the max number of pending datapoints to remote write endpoint
+        
   -remotewrite.url string
         Optional URL to Victoria Metrics or VMInsert where to persist alerts state in form of timeseries. E.g. http://127.0.0.1:8428
   -rule value
diff --git a/app/vmalert/main.go b/app/vmalert/main.go
index 7960bf04e..f524b9cea 100644
--- a/app/vmalert/main.go
+++ b/app/vmalert/main.go
@@ -37,17 +37,18 @@ absolute path to all .yaml files in root.`)
 	basicAuthUsername = flag.String("datasource.basicAuth.username", "", "Optional basic auth username for -datasource.url")
 	basicAuthPassword = flag.String("datasource.basicAuth.password", "", "Optional basic auth password for -datasource.url")
 
-	remoteWriteURL = flag.String("remotewrite.url", "", "Optional URL to Victoria Metrics or VMInsert where to persist alerts state"+
+	remoteWriteURL = flag.String("remoteWrite.url", "", "Optional URL to Victoria Metrics or VMInsert where to persist alerts state"+
 		" in form of timeseries. E.g. http://127.0.0.1:8428")
-	remoteWriteUsername = flag.String("remotewrite.basicAuth.username", "", "Optional basic auth username for -remotewrite.url")
-	remoteWritePassword = flag.String("remotewrite.basicAuth.password", "", "Optional basic auth password for -remotewrite.url")
+	remoteWriteUsername     = flag.String("remoteWrite.basicAuth.username", "", "Optional basic auth username for -remotewrite.url")
+	remoteWritePassword     = flag.String("remoteWrite.basicAuth.password", "", "Optional basic auth password for -remotewrite.url")
+	remoteWriteMaxQueueSize = flag.Int("remoteWrite.maxQueueSize", 10000, "Optional Defines the max number of pending datapoints to remote write endpoint")
 
-	remoteReadURL = flag.String("remoteread.url", "", "Optional URL to Victoria Metrics or VMSelect that will be used to restore alerts"+
+	remoteReadURL = flag.String("remoteRead.url", "", "Optional URL to Victoria Metrics or VMSelect that will be used to restore alerts"+
 		" state. This configuration makes sense only if `vmalert` was configured with `remotewrite.url` before and has been successfully persisted its state."+
 		" E.g. http://127.0.0.1:8428")
-	remoteReadUsername = flag.String("remoteread.basicAuth.username", "", "Optional basic auth username for -remoteread.url")
-	remoteReadPassword = flag.String("remoteread.basicAuth.password", "", "Optional basic auth password for -remoteread.url")
-	remoteReadLookBack = flag.Duration("remoteread.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries."+
+	remoteReadUsername = flag.String("remoteRead.basicAuth.username", "", "Optional basic auth username for -remoteread.url")
+	remoteReadPassword = flag.String("remoteRead.basicAuth.password", "", "Optional basic auth password for -remoteread.url")
+	remoteReadLookBack = flag.Duration("remoteRead.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries."+
 		" For example, if lookback=1h then range from now() to now()-1h will be scanned.")
 
 	evaluationInterval = flag.Duration("evaluationInterval", time.Minute, "How often to evaluate the rules. Default 1m")
@@ -77,6 +78,7 @@ func main() {
 	if *remoteWriteURL != "" {
 		c, err := remotewrite.NewClient(ctx, remotewrite.Config{
 			Addr:          *remoteWriteURL,
+			MaxQueueSize:  *remoteWriteMaxQueueSize,
 			FlushInterval: *evaluationInterval,
 			BasicAuthUser: *remoteWriteUsername,
 			BasicAuthPass: *remoteWritePassword,
diff --git a/app/vmalert/remotewrite/remotewrite.go b/app/vmalert/remotewrite/remotewrite.go
index 10586b029..812c7a178 100644
--- a/app/vmalert/remotewrite/remotewrite.go
+++ b/app/vmalert/remotewrite/remotewrite.go
@@ -103,7 +103,7 @@ func (c *Client) Push(s prompbmarshal.TimeSeries) error {
 	case c.input <- s:
 		return nil
 	default:
-		return fmt.Errorf("failed to push timeseries - queue is full (%d entries)",
+		return fmt.Errorf("failed to push timeseries - queue is full (%d entries), hint from description and add recommendation to increaseremoteWrite.maxQueueSize",
 			c.maxQueueSize)
 	}
 }
diff --git a/app/vmselect/promql/aggr.go b/app/vmselect/promql/aggr.go
index 8071da3a0..8cf478dc4 100644
--- a/app/vmselect/promql/aggr.go
+++ b/app/vmselect/promql/aggr.go
@@ -43,6 +43,7 @@ var aggrFuncs = map[string]aggrFunc{
 	"bottomk_max":    newAggrFuncRangeTopK(maxValue, true),
 	"bottomk_avg":    newAggrFuncRangeTopK(avgValue, true),
 	"bottomk_median": newAggrFuncRangeTopK(medianValue, true),
+	"any":            newAggrFunc(aggrFuncAny),
 }
 
 type aggrFunc func(afa *aggrFuncArg) ([]*timeseries, error)
@@ -64,7 +65,7 @@ func newAggrFunc(afe func(tss []*timeseries) []*timeseries) aggrFunc {
 		if err := expectTransformArgsNum(args, 1); err != nil {
 			return nil, err
 		}
-		return aggrFuncExt(afe, args[0], &afa.ae.Modifier, false)
+		return aggrFuncExt(afe, args[0], &afa.ae.Modifier, afa.ae.Limit, false)
 	}
 }
 
@@ -80,7 +81,7 @@ func removeGroupTags(metricName *storage.MetricName, modifier *metricsql.Modifie
 	}
 }
 
-func aggrFuncExt(afe func(tss []*timeseries) []*timeseries, argOrig []*timeseries, modifier *metricsql.ModifierExpr, keepOriginal bool) ([]*timeseries, error) {
+func aggrFuncExt(afe func(tss []*timeseries) []*timeseries, argOrig []*timeseries, modifier *metricsql.ModifierExpr, maxSeries int, keepOriginal bool) ([]*timeseries, error) {
 	arg := copyTimeseriesMetricNames(argOrig)
 
 	// Perform grouping.
@@ -92,7 +93,13 @@ func aggrFuncExt(afe func(tss []*timeseries) []*timeseries, argOrig []*timeserie
 		if keepOriginal {
 			ts = argOrig[i]
 		}
-		m[string(bb.B)] = append(m[string(bb.B)], ts)
+		tss := m[string(bb.B)]
+		if tss == nil && maxSeries > 0 && len(m) >= maxSeries {
+			// We already reached time series limit after grouping. Skip other time series.
+			continue
+		}
+		tss = append(tss, ts)
+		m[string(bb.B)] = tss
 	}
 	bbPool.Put(bb)
 
@@ -112,6 +119,10 @@ func aggrFuncExt(afe func(tss []*timeseries) []*timeseries, argOrig []*timeserie
 	return rvs, nil
 }
 
+func aggrFuncAny(tss []*timeseries) []*timeseries {
+	return tss[:1]
+}
+
 func aggrFuncSum(tss []*timeseries) []*timeseries {
 	if len(tss) == 1 {
 		// Fast path - nothing to sum.
@@ -441,7 +452,7 @@ func aggrFuncCountValues(afa *aggrFuncArg) ([]*timeseries, error) {
 		}
 		return rvs
 	}
-	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, false)
+	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, false)
 }
 
 func newAggrFuncTopK(isReverse bool) aggrFunc {
@@ -468,7 +479,7 @@ func newAggrFuncTopK(isReverse bool) aggrFunc {
 			}
 			return removeNaNs(tss)
 		}
-		return aggrFuncExt(afe, args[1], &afa.ae.Modifier, true)
+		return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
 	}
 }
 
@@ -512,7 +523,7 @@ func newAggrFuncRangeTopK(f func(values []float64) float64, isReverse bool) aggr
 			}
 			return removeNaNs(tss)
 		}
-		return aggrFuncExt(afe, args[1], &afa.ae.Modifier, true)
+		return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
 	}
 }
 
@@ -618,7 +629,7 @@ func aggrFuncLimitK(afa *aggrFuncArg) ([]*timeseries, error) {
 		}
 		return tss
 	}
-	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, true)
+	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
 }
 
 func aggrFuncQuantile(afa *aggrFuncArg) ([]*timeseries, error) {
@@ -631,7 +642,7 @@ func aggrFuncQuantile(afa *aggrFuncArg) ([]*timeseries, error) {
 		return nil, err
 	}
 	afe := newAggrQuantileFunc(phis)
-	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, false)
+	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, false)
 }
 
 func aggrFuncMedian(afa *aggrFuncArg) ([]*timeseries, error) {
@@ -641,7 +652,7 @@ func aggrFuncMedian(afa *aggrFuncArg) ([]*timeseries, error) {
 	}
 	phis := evalNumber(afa.ec, 0.5)[0].Values
 	afe := newAggrQuantileFunc(phis)
-	return aggrFuncExt(afe, args[0], &afa.ae.Modifier, false)
+	return aggrFuncExt(afe, args[0], &afa.ae.Modifier, afa.ae.Limit, false)
 }
 
 func newAggrQuantileFunc(phis []float64) func(tss []*timeseries) []*timeseries {
diff --git a/app/vmselect/promql/aggr_incremental.go b/app/vmselect/promql/aggr_incremental.go
index f2599ab27..c9d058b3e 100644
--- a/app/vmselect/promql/aggr_incremental.go
+++ b/app/vmselect/promql/aggr_incremental.go
@@ -48,6 +48,11 @@ var incrementalAggrFuncCallbacksMap = map[string]*incrementalAggrFuncCallbacks{
 		mergeAggrFunc:    mergeAggrGeomean,
 		finalizeAggrFunc: finalizeAggrGeomean,
 	},
+	"any": {
+		updateAggrFunc:   updateAggrAny,
+		mergeAggrFunc:    mergeAggrAny,
+		finalizeAggrFunc: finalizeAggrCommon,
+	},
 }
 
 type incrementalAggrFuncContext struct {
@@ -81,6 +86,10 @@ func (iafc *incrementalAggrFuncContext) updateTimeseries(ts *timeseries, workerI
 	bb.B = marshalMetricNameSorted(bb.B[:0], &ts.MetricName)
 	iac := m[string(bb.B)]
 	if iac == nil {
+		if iafc.ae.Limit > 0 && len(m) >= iafc.ae.Limit {
+			// Skip this time series, since the limit on the number of output time series has been already reached.
+			return
+		}
 		tsAggr := &timeseries{
 			Values:     make([]float64, len(ts.Values)),
 			Timestamps: ts.Timestamps,
@@ -106,6 +115,10 @@ func (iafc *incrementalAggrFuncContext) finalizeTimeseries() []*timeseries {
 		for k, iac := range m {
 			iacGlobal := mGlobal[k]
 			if iacGlobal == nil {
+				if iafc.ae.Limit > 0 && len(mGlobal) >= iafc.ae.Limit {
+					// Skip this time series, since the limit on the number of output time series has been already reached.
+					continue
+				}
 				mGlobal[k] = iac
 				continue
 			}
@@ -450,3 +463,25 @@ func finalizeAggrGeomean(iac *incrementalAggrContext) {
 		dstValues[i] = math.Pow(dstValues[i], 1/v)
 	}
 }
+
+func updateAggrAny(iac *incrementalAggrContext, values []float64) {
+	dstCounts := iac.values
+	if dstCounts[0] > 0 {
+		return
+	}
+	for i := range values {
+		dstCounts[i] = 1
+	}
+	iac.ts.Values = append(iac.ts.Values[:0], values...)
+}
+
+func mergeAggrAny(dst, src *incrementalAggrContext) {
+	srcValues := src.ts.Values
+	srcCounts := src.values
+	dstCounts := dst.values
+	if dstCounts[0] > 0 {
+		return
+	}
+	dstCounts[0] = srcCounts[0]
+	dst.ts.Values = append(dst.ts.Values[:0], srcValues...)
+}
diff --git a/app/vmselect/promql/eval.go b/app/vmselect/promql/eval.go
index c26cef927..a56e4ed1b 100644
--- a/app/vmselect/promql/eval.go
+++ b/app/vmselect/promql/eval.go
@@ -663,12 +663,17 @@ func evalRollupFuncWithMetricExpr(ec *EvalConfig, name string, rf rollupFunc,
 	pointsPerTimeseries := 1 + (ec.End-ec.Start)/ec.Step
 	timeseriesLen := rssLen
 	if iafc != nil {
-		// Incremental aggregates require hold only GOMAXPROCS timeseries in memory.
+		// Incremental aggregates require holding only GOMAXPROCS timeseries in memory.
 		timeseriesLen = runtime.GOMAXPROCS(-1)
 		if iafc.ae.Modifier.Op != "" {
-			// Increase the number of timeseries for non-empty group list: `aggr() by (something)`,
-			// since each group can have own set of time series in memory.
-			timeseriesLen *= 1000
+			if iafc.ae.Limit > 0 {
+				// There is an explicit limit on the number of output time series.
+				timeseriesLen *= iafc.ae.Limit
+			} else {
+				// Increase the number of timeseries for non-empty group list: `aggr() by (something)`,
+				// since each group can have own set of time series in memory.
+				timeseriesLen *= 1000
+			}
 		}
 		// The maximum number of output time series is limited by rssLen.
 		if timeseriesLen > rssLen {
diff --git a/docs/MetricsQL.md b/docs/MetricsQL.md
index af440fdef..dea5008a8 100644
--- a/docs/MetricsQL.md
+++ b/docs/MetricsQL.md
@@ -22,6 +22,8 @@ Feel free [filing a feature request](https://github.com/VictoriaMetrics/Victoria
 This functionality can be tried at [an editable Grafana dashboard](http://play-grafana.victoriametrics.com:3000/d/4ome8yJmz/node-exporter-on-victoriametrics-demo).
 
 - [`WITH` templates](https://play.victoriametrics.com/promql/expand-with-exprs). This feature simplifies writing and managing complex queries. Go to [`WITH` templates playground](https://victoriametrics.com/promql/expand-with-exprs) and try it.
+- All the aggregate functions support optional `limit N` suffix in order to limit the number of output series. For example, `sum(x) by (y) limit 10` limits
+  the number of output time series after the aggregation to 10. All the other time series are dropped.
 - Metric names and metric labels may contain escaped chars. For instance, `foo\-bar{baz\=aa="b"}` is valid expression. It returns time series with name `foo-bar` containing label `baz=aa` with value `b`. Additionally, `\xXX` escape sequence is supported, where `XX` is hexadecimal representation of escaped char.
 - `offset`, range duration and step value for range vector may refer to the current step aka `$__interval` value from Grafana.
   For instance, `rate(metric[10i] offset 5i)` would return per-second rate over a range covering 10 previous steps with the offset of 5 steps.
@@ -72,6 +74,8 @@ This functionality can be tried at [an editable Grafana dashboard](http://play-g
 - `median_over_time(m[d])` - calculates median values for `m` over `d` time window. Shorthand to `quantile_over_time(0.5, m[d])`.
 - `median(q)` - median aggregate. Shorthand to `quantile(0.5, q)`.
 - `limitk(k, q)` - limits the number of time series returned from `q` to `k`.
+- `any(q) by (x)` - returns any time series from `q` for each group in `x`. Note that `any()` removes all the labels except of those listed in `by (x)`.
+  Use `limitk(1, q)` if you need retaining all the labels from `q`.
 - `keep_last_value(q)` - fills missing data (gaps) in `q` with the previous non-empty value.
 - `keep_next_value(q)` - fills missing data (gaps) in `q` with the next non-empty value.
 - `distinct_over_time(m[d])` - returns distinct number of values for `m` data points over `d` duration.
diff --git a/go.mod b/go.mod
index bc4c9f540..b5bbfb5fd 100644
--- a/go.mod
+++ b/go.mod
@@ -9,7 +9,7 @@ require (
 	// like https://github.com/valyala/fasthttp/commit/996610f021ff45fdc98c2ce7884d5fa4e7f9199b
 	github.com/VictoriaMetrics/fasthttp v1.0.1
 	github.com/VictoriaMetrics/metrics v1.11.2
-	github.com/VictoriaMetrics/metricsql v0.1.0
+	github.com/VictoriaMetrics/metricsql v0.2.1
 	github.com/aws/aws-sdk-go v1.30.25
 	github.com/cespare/xxhash/v2 v2.1.1
 	github.com/golang/protobuf v1.4.1 // indirect
diff --git a/go.sum b/go.sum
index 7e9a94ba4..af2244af8 100644
--- a/go.sum
+++ b/go.sum
@@ -45,8 +45,8 @@ github.com/VictoriaMetrics/fasthttp v1.0.1 h1:I7YdbswTIW63WxoFoUOSNxeOEGB46rdKUL
 github.com/VictoriaMetrics/fasthttp v1.0.1/go.mod h1:BqgsieH90PR7x97c89j+eqZDloKkDhAEQTwhLw6jw/4=
 github.com/VictoriaMetrics/metrics v1.11.2 h1:t/ceLP6SvagUqypCKU7cI7+tQn54+TIV/tGoxihHvx8=
 github.com/VictoriaMetrics/metrics v1.11.2/go.mod h1:LU2j9qq7xqZYXz8tF3/RQnB2z2MbZms5TDiIg9/NHiQ=
-github.com/VictoriaMetrics/metricsql v0.1.0 h1:IoyG84PCwFY15rNuxpr2nQ+YZBYIhnd7zTiaGo5BNpc=
-github.com/VictoriaMetrics/metricsql v0.1.0/go.mod h1:UIjd9S0W1UnTWlJdM0wLS+2pfuPqjwqKoK8yTos+WyE=
+github.com/VictoriaMetrics/metricsql v0.2.1 h1:OI/W2QCFiQiFULVN3ZiC/iCqZFt25rXp/O7P2NiAwYU=
+github.com/VictoriaMetrics/metricsql v0.2.1/go.mod h1:UIjd9S0W1UnTWlJdM0wLS+2pfuPqjwqKoK8yTos+WyE=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/aws/aws-sdk-go v1.30.25 h1:89NXJkfpjnMEnsxkP8MVX+LDsoiLCSqevraLb5y4Mjk=
diff --git a/lib/fs/reader_at.go b/lib/fs/reader_at.go
index 23ceb91cb..3a9c5f439 100644
--- a/lib/fs/reader_at.go
+++ b/lib/fs/reader_at.go
@@ -10,7 +10,10 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-var disableMmap = flag.Bool("fs.disableMmap", false, "Whether to use pread() instead of mmap() for reading data files")
+var disableMmap = flag.Bool("fs.disableMmap", is32BitPtr, "Whether to use pread() instead of mmap() for reading data files. "+
+	"By default mmap() is used for 64-bit arches and pread() is used for 32-bit arches, since they cannot data files bigger than 2^32 bytes in memory")
+
+const is32BitPtr = (^uintptr(0) >> 32) == 0
 
 // MustReadAtCloser is rand-access read interface.
 type MustReadAtCloser interface {
diff --git a/lib/promauth/config.go b/lib/promauth/config.go
index 4c3cd82fe..76c305967 100644
--- a/lib/promauth/config.go
+++ b/lib/promauth/config.go
@@ -69,9 +69,8 @@ func (ac *Config) NewTLSConfig() *tls.Config {
 		ClientSessionCache: tls.NewLRUClientSessionCache(0),
 	}
 	if ac.TLSCertificate != nil {
-		tlsCfg.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
-			return ac.TLSCertificate, nil
-		}
+		// Do not set tlsCfg.GetClientCertificate, since tlsCfg.Certificates should work OK.
+		tlsCfg.Certificates = []tls.Certificate{*ac.TLSCertificate}
 	}
 	tlsCfg.ServerName = ac.TLSServerName
 	tlsCfg.InsecureSkipVerify = ac.TLSInsecureSkipVerify
diff --git a/lib/prompbmarshal/util.go b/lib/prompbmarshal/util.go
index 72d45e954..2dcbc5f81 100644
--- a/lib/prompbmarshal/util.go
+++ b/lib/prompbmarshal/util.go
@@ -21,10 +21,15 @@ func MarshalWriteRequest(dst []byte, wr *WriteRequest) []byte {
 
 // ResetWriteRequest resets wr.
 func ResetWriteRequest(wr *WriteRequest) {
-	for i := range wr.Timeseries {
-		ts := wr.Timeseries[i]
+	wr.Timeseries = ResetTimeSeries(wr.Timeseries)
+}
+
+// ResetTimeSeries clears all the GC references from tss and returns an empty tss ready for further use.
+func ResetTimeSeries(tss []TimeSeries) []TimeSeries {
+	for i := range tss {
+		ts := tss[i]
 		ts.Labels = nil
 		ts.Samples = nil
 	}
-	wr.Timeseries = wr.Timeseries[:0]
+	return tss[:0]
 }
diff --git a/lib/promscrape/discovery/kubernetes/api.go b/lib/promscrape/discovery/kubernetes/api.go
index 142a9e468..55da6b240 100644
--- a/lib/promscrape/discovery/kubernetes/api.go
+++ b/lib/promscrape/discovery/kubernetes/api.go
@@ -50,7 +50,7 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		tlsConfig := promauth.TLSConfig{
 			CAFile: "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
 		}
-		acNew, err := promauth.NewConfig("/", nil, "", "/var/run/secrets/kubernetes.io/serviceaccount/token", &tlsConfig)
+		acNew, err := promauth.NewConfig(".", nil, "", "/var/run/secrets/kubernetes.io/serviceaccount/token", &tlsConfig)
 		if err != nil {
 			return nil, fmt.Errorf("cannot initialize service account auth: %s; probably, `kubernetes_sd_config->api_server` is missing in Prometheus configs?", err)
 		}
diff --git a/lib/storage/tag_filters.go b/lib/storage/tag_filters.go
index 83b5197a4..00cd59c67 100644
--- a/lib/storage/tag_filters.go
+++ b/lib/storage/tag_filters.go
@@ -479,7 +479,16 @@ func getOptimizedReMatchFuncExt(reMatch func(b []byte) bool, sre *syntax.Regexp)
 				literals = append(literals, []byte(string(sub.Rune)))
 			}
 		}
+		var suffix []byte
+		if isLiteral(sre.Sub[len(sre.Sub)-1]) {
+			suffix = literals[len(literals)-1]
+			literals = literals[:len(literals)-1]
+		}
 		return func(b []byte) bool {
+			if len(suffix) > 0 && !bytes.HasSuffix(b, suffix) {
+				// Fast path - b has no the given suffix
+				return false
+			}
 			bOrig := b
 			for _, literal := range literals {
 				n := bytes.Index(b, literal)
diff --git a/vendor/github.com/VictoriaMetrics/metricsql/README.md b/vendor/github.com/VictoriaMetrics/metricsql/README.md
index 28af4ac44..26d6db266 100644
--- a/vendor/github.com/VictoriaMetrics/metricsql/README.md
+++ b/vendor/github.com/VictoriaMetrics/metricsql/README.md
@@ -1,7 +1,5 @@
-[![Build Status](https://github.com/VictoriaMetrics/metricsql/workflows/main/badge.svg)](https://github.com/VictoriaMetrics/metricsql/actions)
 [![GoDoc](https://godoc.org/github.com/VictoriaMetrics/metricsql?status.svg)](http://godoc.org/github.com/VictoriaMetrics/metricsql)
 [![Go Report](https://goreportcard.com/badge/github.com/VictoriaMetrics/metricsql)](https://goreportcard.com/report/github.com/VictoriaMetrics/metricsql)
-[![codecov](https://codecov.io/gh/VictoriaMetrics/metricsql/branch/master/graph/badge.svg)](https://codecov.io/gh/VictoriaMetrics/metricsql)
 
 
 # metricsql
diff --git a/vendor/github.com/VictoriaMetrics/metricsql/aggr.go b/vendor/github.com/VictoriaMetrics/metricsql/aggr.go
index 66a8f8b87..5cd88de20 100644
--- a/vendor/github.com/VictoriaMetrics/metricsql/aggr.go
+++ b/vendor/github.com/VictoriaMetrics/metricsql/aggr.go
@@ -33,6 +33,7 @@ var aggrFuncs = map[string]bool{
 	"bottomk_max":    true,
 	"bottomk_avg":    true,
 	"bottomk_median": true,
+	"any":            true,
 }
 
 func isAggrFunc(s string) bool {
diff --git a/vendor/github.com/VictoriaMetrics/metricsql/parser.go b/vendor/github.com/VictoriaMetrics/metricsql/parser.go
index 10febd145..5ad0552e6 100644
--- a/vendor/github.com/VictoriaMetrics/metricsql/parser.go
+++ b/vendor/github.com/VictoriaMetrics/metricsql/parser.go
@@ -535,12 +535,10 @@ func (p *parser) parseAggrFuncExpr() (*AggrFuncExpr, error) {
 	if isIdentPrefix(p.lex.Token) {
 		goto funcPrefixLabel
 	}
-	switch p.lex.Token {
-	case "(":
+	if p.lex.Token == "(" {
 		goto funcArgsLabel
-	default:
-		return nil, fmt.Errorf(`AggrFuncExpr: unexpected token %q; want "("`, p.lex.Token)
 	}
+	return nil, fmt.Errorf(`AggrFuncExpr: unexpected token %q; want "("`, p.lex.Token)
 
 funcPrefixLabel:
 	{
@@ -550,7 +548,6 @@ funcPrefixLabel:
 		if err := p.parseModifierExpr(&ae.Modifier); err != nil {
 			return nil, err
 		}
-		goto funcArgsLabel
 	}
 
 funcArgsLabel:
@@ -562,11 +559,25 @@ funcArgsLabel:
 		ae.Args = args
 
 		// Verify whether func suffix exists.
-		if ae.Modifier.Op != "" || !isAggrFuncModifier(p.lex.Token) {
-			return &ae, nil
+		if ae.Modifier.Op == "" && isAggrFuncModifier(p.lex.Token) {
+			if err := p.parseModifierExpr(&ae.Modifier); err != nil {
+				return nil, err
+			}
 		}
-		if err := p.parseModifierExpr(&ae.Modifier); err != nil {
-			return nil, err
+
+		// Check for optional limit.
+		if strings.ToLower(p.lex.Token) == "limit" {
+			if err := p.lex.Next(); err != nil {
+				return nil, err
+			}
+			limit, err := strconv.Atoi(p.lex.Token)
+			if err != nil {
+				return nil, fmt.Errorf("cannot parse limit %q: %s", p.lex.Token, err)
+			}
+			if err := p.lex.Next(); err != nil {
+				return nil, err
+			}
+			ae.Limit = limit
 		}
 		return &ae, nil
 	}
@@ -640,6 +651,7 @@ func expandWithExpr(was []*withArgExpr, e Expr) (Expr, error) {
 			Name:     t.Name,
 			Args:     args,
 			Modifier: t.Modifier,
+			Limit:    t.Limit,
 		}
 		ae.Modifier.Args = modifierArgs
 		return ae, nil
@@ -1495,6 +1507,12 @@ type AggrFuncExpr struct {
 
 	// Modifier is optional modifier such as `by (...)` or `without (...)`.
 	Modifier ModifierExpr
+
+	// Optional limit for the number of output time series.
+	// This is MetricsQL extension.
+	//
+	// Example: `sum(...) by (...) limit 10` would return maximum 10 time series.
+	Limit int
 }
 
 // AppendString appends string representation of ae to dst and returns the result.
@@ -1505,6 +1523,10 @@ func (ae *AggrFuncExpr) AppendString(dst []byte) []byte {
 		dst = append(dst, ' ')
 		dst = ae.Modifier.AppendString(dst)
 	}
+	if ae.Limit > 0 {
+		dst = append(dst, " limit "...)
+		dst = strconv.AppendInt(dst, int64(ae.Limit), 10)
+	}
 	return dst
 }
 
diff --git a/vendor/modules.txt b/vendor/modules.txt
index c78589590..70f693a5a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -18,7 +18,7 @@ github.com/VictoriaMetrics/fasthttp/fasthttputil
 github.com/VictoriaMetrics/fasthttp/stackless
 # github.com/VictoriaMetrics/metrics v1.11.2
 github.com/VictoriaMetrics/metrics
-# github.com/VictoriaMetrics/metricsql v0.1.0
+# github.com/VictoriaMetrics/metricsql v0.2.1
 github.com/VictoriaMetrics/metricsql
 github.com/VictoriaMetrics/metricsql/binaryop
 # github.com/aws/aws-sdk-go v1.30.25


Test patch:
diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index a297a6cee..c7289e6a2 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -3496,6 +3496,21 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1, r2}
 		f(q, resultExpected)
 	})
+	t.Run(`sum(multi-vector) by (known-tag) limit 1`, func(t *testing.T) {
+		t.Parallel()
+		q := `sum(label_set(10, "foo", "bar") or label_set(time()/100, "baz", "sss")) by (foo) limit 1`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{10, 10, 10, 10, 10, 10},
+			Timestamps: timestampsExpected,
+		}
+		r.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("foo"),
+			Value: []byte("bar"),
+		}}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`sum(multi-vector) by (known-tags)`, func(t *testing.T) {
 		t.Parallel()
 		q := `sum(label_set(10, "foo", "bar", "baz", "sss", "x", "y") or label_set(time()/100, "baz", "sss", "foo", "bar")) by (foo, baz, foo)`
@@ -3768,6 +3783,17 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1, r2}
 		f(q, resultExpected)
 	})
+	t.Run(`any()`, func(t *testing.T) {
+		t.Parallel()
+		q := `any(label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss"))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{10, 10, 10, 10, 10, 10},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`topk(-1)`, func(t *testing.T) {
 		t.Parallel()
 		q := `sort(topk(-1, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss")))`
diff --git a/lib/storage/tag_filters_test.go b/lib/storage/tag_filters_test.go
index a935ff2aa..a0d2baf99 100644
--- a/lib/storage/tag_filters_test.go
+++ b/lib/storage/tag_filters_test.go
@@ -68,6 +68,7 @@ func TestGetRegexpFromCache(t *testing.T) {
 	f(".*foo.*bar", nil, []string{"foobar", "xfoobar", "xfooxbar", "fooxbar"}, []string{"", "foobarx", "afoobarx", "aaa"})
 	f("foo.*bar", nil, []string{"foobar", "fooxbar"}, []string{"xfoobar", "", "foobarx", "aaa"})
 	f("foo.*bar.*", nil, []string{"foobar", "fooxbar", "foobarx", "fooxbarx"}, []string{"", "afoobarx", "aaa", "afoobar"})
+	f("foo.*bar.*baz", nil, []string{"foobarbaz", "fooxbarxbaz", "foobarxbaz", "fooxbarbaz"}, []string{"", "afoobarx", "aaa", "afoobar", "foobarzaz"})
 	f(".+foo.+(b|c).+", nil, []string{"xfooxbar", "xfooxca"}, []string{"", "foo", "foob", "xfooc", "xfoodc"})
 
 	f("(?i)foo", nil, []string{"foo", "Foo", "FOO"}, []string{"xfoo", "foobar", "xFOObar"})