Repository: VictoriaMetrics/VictoriaMetrics
Pull request #3476 (2022-12-12 17:49:09)

Add Anomaly Detection to enterprise features list


Patch:
diff --git a/app/vmalert/README.md b/app/vmalert/README.md
index d5a49ac21..842fc42b6 100644
--- a/app/vmalert/README.md
+++ b/app/vmalert/README.md
@@ -213,7 +213,8 @@ The following variables are available in templating:
 | $labels or .Labels                 | The list of labels of the current alert. Use as ".Labels.<label_name>".                                   | {% raw %}Too high number of connections for {{ .Labels.instance }}{% endraw %}                                                                                         |
 | $alertID or .AlertID               | The current alert's ID generated by vmalert.                                                              | {% raw %}Link: vmalert/alert?group_id={{.GroupID}}&alert_id={{.AlertID}}{% endraw %}                                                                                   |
 | $groupID or .GroupID               | The current alert's group ID generated by vmalert.                                                        | {% raw %}Link: vmalert/alert?group_id={{.GroupID}}&alert_id={{.AlertID}}{% endraw %}                                                                                   |
-| $expr or .Expr                     | Alert's expression. Can be used for generating links to Grafana or other systems.                         | {% raw %}/api/v1/query?query={{ $expr&#124;queryEscape }}{% endraw %}                                                                                |
+| $expr or .Expr                     | Alert's expression. Can be used for generating links to Grafana or other systems.                         | {% raw %}/api/v1/query?query={{ $expr&#124;queryEscape }}{% endraw %}                                                                                                  |
+| $for or .For                       | Alert's configured for param.                                                                             | {% raw %}Number of connections is too high for more than {{ .For }}{% endraw %}                                                                                        |
 | $externalLabels or .ExternalLabels | List of labels configured via `-external.label` command-line flag.                                        | {% raw %}Issues with {{ $labels.instance }} (datacenter-{{ $externalLabels.dc }}){% endraw %}                                                                          |
 | $externalURL or .ExternalURL       | URL configured via `-external.url` command-line flag. Used for cases when vmalert is hidden behind proxy. | {% raw %}Visit {{ $externalURL }} for more details{% endraw %}                                                                                                         |
 
diff --git a/app/vmalert/alerting.go b/app/vmalert/alerting.go
index 18221c7d5..368785105 100644
--- a/app/vmalert/alerting.go
+++ b/app/vmalert/alerting.go
@@ -456,6 +456,7 @@ func (ar *AlertingRule) newAlert(m datasource.Metric, ls *labelSet, start time.T
 		Value:    m.Values[0],
 		ActiveAt: start,
 		Expr:     ar.Expr,
+		For:      ar.For,
 	}
 	a.Annotations, err = a.ExecTemplate(qFn, ls.origin, ar.Annotations)
 	return a, err
diff --git a/app/vmalert/notifier/alert.go b/app/vmalert/notifier/alert.go
index fa9699048..35a68c54c 100644
--- a/app/vmalert/notifier/alert.go
+++ b/app/vmalert/notifier/alert.go
@@ -45,6 +45,8 @@ type Alert struct {
 	ID uint64
 	// Restored is true if Alert was restored after restart
 	Restored bool
+	// For defines for how long Alert needs to be active to become StateFiring
+	For time.Duration
 }
 
 // AlertState type indicates the Alert state
@@ -80,6 +82,7 @@ type AlertTplData struct {
 	AlertID  uint64
 	GroupID  uint64
 	ActiveAt time.Time
+	For      time.Duration
 }
 
 var tplHeaders = []string{
@@ -91,6 +94,7 @@ var tplHeaders = []string{
 	"{{ $alertID := .AlertID }}",
 	"{{ $groupID := .GroupID }}",
 	"{{ $activeAt := .ActiveAt }}",
+	"{{ $for := .For }}",
 }
 
 // ExecTemplate executes the Alert template for given
@@ -98,7 +102,15 @@ var tplHeaders = []string{
 // Every alert could have a different datasource, so function
 // requires a queryFunction as an argument.
 func (a *Alert) ExecTemplate(q templates.QueryFn, labels, annotations map[string]string) (map[string]string, error) {
-	tplData := AlertTplData{Value: a.Value, Labels: labels, Expr: a.Expr, AlertID: a.ID, GroupID: a.GroupID, ActiveAt: a.ActiveAt}
+	tplData := AlertTplData{
+		Value:    a.Value,
+		Labels:   labels,
+		Expr:     a.Expr,
+		AlertID:  a.ID,
+		GroupID:  a.GroupID,
+		ActiveAt: a.ActiveAt,
+		For:      a.For,
+	}
 	tmpl, err := templates.GetWithFuncs(templates.FuncsWithQuery(q))
 	if err != nil {
 		return nil, fmt.Errorf("error getting a template: %w", err)
diff --git a/app/vmalert/templates/template.go b/app/vmalert/templates/template.go
index 9c35de1ab..faab29a94 100644
--- a/app/vmalert/templates/template.go
+++ b/app/vmalert/templates/template.go
@@ -361,7 +361,7 @@ func templateFuncs() textTpl.FuncMap {
 			return fmt.Sprintf("%.4g%s", v, prefix), nil
 		},
 
-		// humanizeDuration converts given seconds to a human readable duration
+		// humanizeDuration converts given seconds to a human-readable duration
 		"humanizeDuration": func(i interface{}) (string, error) {
 			v, err := toFloat64(i)
 			if err != nil {
diff --git a/app/vmbackup/README.md b/app/vmbackup/README.md
index 7b3b4d56f..7d62c9793 100644
--- a/app/vmbackup/README.md
+++ b/app/vmbackup/README.md
@@ -21,7 +21,7 @@ creation of hourly, daily, weekly and monthly backups.
 
 * [GCS](https://cloud.google.com/storage/). Example: `gs://<bucket>/<path/to/backup>`
 * [S3](https://aws.amazon.com/s3/). Example: `s3://<bucket>/<path/to/backup>`
-* [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `azblob://<bucket>/<path/to/backup>`
+* [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `azblob://<container>/<path/to/backup>`
 * Any S3-compatible storage such as [MinIO](https://github.com/minio/minio), [Ceph](https://docs.ceph.com/en/pacific/radosgw/s3/) or [Swift](https://platform.swiftstack.com/docs/admin/middleware/s3_middleware.html). See [these docs](#advanced-usage) for details.
 * Local filesystem. Example: `fs://</absolute/path/to/backup>`. Note that `vmbackup` prevents from storing the backup into the directory pointed by `-storageDataPath` command-line flag, since this directory should be managed solely by VictoriaMetrics or `vmstorage`.
 
@@ -187,7 +187,7 @@ See [this article](https://medium.com/@valyala/speeding-up-backups-for-big-time-
   -customS3Endpoint string
      Custom S3 endpoint for use with S3-compatible storages (e.g. MinIO). S3 is used if not set
   -dst string
-     Where to put the backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://bucket/path/to/backup or fs:///path/to/local/backup/dir
+     Where to put the backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://container/path/to/backup or fs:///path/to/local/backup/dir
      -dst can point to the previous backup. In this case incremental backup is performed, i.e. only changed data is uploaded
   -enableTCP6
      Whether to enable IPv6 for listening and dialing. By default only IPv4 TCP and UDP is used
diff --git a/app/vmbackup/main.go b/app/vmbackup/main.go
index 001f1d6d8..57771ca64 100644
--- a/app/vmbackup/main.go
+++ b/app/vmbackup/main.go
@@ -30,7 +30,7 @@ var (
 	snapshotDeleteURL = flag.String("snapshot.deleteURL", "", "VictoriaMetrics delete snapshot url. Optional. Will be generated from -snapshot.createURL if not provided. "+
 		"All created snapshots will be automatically deleted. Example: http://victoriametrics:8428/snapshot/delete")
 	dst = flag.String("dst", "", "Where to put the backup on the remote storage. "+
-		"Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://bucket/path/to/backup or fs:///path/to/local/backup/dir\n"+
+		"Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://container/path/to/backup or fs:///path/to/local/backup/dir\n"+
 		"-dst can point to the previous backup. In this case incremental backup is performed, i.e. only changed data is uploaded")
 	origin            = flag.String("origin", "", "Optional origin directory on the remote storage with old backup for server-side copying when performing full backup. This speeds up full backups")
 	concurrency       = flag.Int("concurrency", 10, "The number of concurrent workers. Higher concurrency may reduce backup duration")
diff --git a/app/vmrestore/README.md b/app/vmrestore/README.md
index 45cf53209..31bf15995 100644
--- a/app/vmrestore/README.md
+++ b/app/vmrestore/README.md
@@ -19,7 +19,7 @@ Run the following command to restore backup from the given `-src` into the given
   `vmrestore` can restore backups from the following storage types:
   * [GCS](https://cloud.google.com/storage/). Example: `-src=gs://<bucket>/<path/to/backup>`
   * [S3](https://aws.amazon.com/s3/). Example: `-src=s3://<bucket>/<path/to/backup>`
-  * [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `-src=azblob://<bucket>/<path/to/backup>`
+  * [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `-src=azblob://<container>/<path/to/backup>`
   * Any S3-compatible storage such as [MinIO](https://github.com/minio/minio), [Ceph](https://docs.ceph.com/en/pacific/radosgw/s3/)
     or [Swift](https://platform.swiftstack.com/docs/admin/middleware/s3_middleware.html). See [these docs](#advanced-usage) for details.
   * Local filesystem. Example: `-src=fs://</absolute/path/to/backup>`. Note that `vmbackup` prevents from storing the backup
@@ -162,7 +162,7 @@ i.e. the end result would be similar to [rsync --delete](https://askubuntu.com/q
   -skipBackupCompleteCheck
      Whether to skip checking for 'backup complete' file in -src. This may be useful for restoring from old backups, which were created without 'backup complete' file
   -src string
-     Source path with backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://bucket/path/to/backup or fs:///path/to/local/backup
+     Source path with backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://container/path/to/backup or fs:///path/to/local/backup
   -storageDataPath string
      Destination path where backup must be restored. VictoriaMetrics must be stopped when restoring from backup. -storageDataPath dir can be non-empty. In this case the contents of -storageDataPath dir is synchronized with -src contents, i.e. it works like 'rsync --delete' (default "victoria-metrics-data")
   -tls
diff --git a/app/vmrestore/main.go b/app/vmrestore/main.go
index 08d07827a..66f72f335 100644
--- a/app/vmrestore/main.go
+++ b/app/vmrestore/main.go
@@ -20,7 +20,7 @@ import (
 var (
 	httpListenAddr = flag.String("httpListenAddr", ":8421", "TCP address for exporting metrics at /metrics page")
 	src            = flag.String("src", "", "Source path with backup on the remote storage. "+
-		"Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://bucket/path/to/backup or fs:///path/to/local/backup")
+		"Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://container/path/to/backup or fs:///path/to/local/backup")
 	storageDataPath = flag.String("storageDataPath", "victoria-metrics-data", "Destination path where backup must be restored. "+
 		"VictoriaMetrics must be stopped when restoring from backup. -storageDataPath dir can be non-empty. In this case the contents of -storageDataPath dir "+
 		"is synchronized with -src contents, i.e. it works like 'rsync --delete'")
diff --git a/app/vmstorage/main.go b/app/vmstorage/main.go
index 2033386ad..13433671a 100644
--- a/app/vmstorage/main.go
+++ b/app/vmstorage/main.go
@@ -558,8 +558,15 @@ func registerStorageMetrics(strg *storage.Storage) {
 	metrics.NewGauge(`vm_assisted_merges_total{type="storage/inmemory"}`, func() float64 {
 		return float64(tm().InmemoryAssistedMerges)
 	})
+	metrics.NewGauge(`vm_assisted_merges_total{type="storage/small"}`, func() float64 {
+		return float64(tm().SmallAssistedMerges)
+	})
+
 	metrics.NewGauge(`vm_assisted_merges_total{type="indexdb/inmemory"}`, func() float64 {
-		return float64(idbm().AssistedInmemoryMerges)
+		return float64(idbm().InmemoryAssistedMerges)
+	})
+	metrics.NewGauge(`vm_assisted_merges_total{type="indexdb/file"}`, func() float64 {
+		return float64(idbm().FileAssistedMerges)
 	})
 
 	metrics.NewGauge(`vm_indexdb_items_added_total`, func() float64 {
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index eb4cbde1f..786e52735 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -15,7 +15,11 @@ The following tip changes can be tested by building VictoriaMetrics components f
 
 ## tip
 
+* FEATURE: [vmalert](https://docs.victoriametrics.com/vmalert.html): support `$for` or `.For` template variables in alert's annotations. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3246).
+
 * BUGFIX: [DataDog protocol parser](https://docs.victoriametrics.com/#how-to-send-data-from-datadog-agent): do not re-use `host` and `device` fields from the previously parsed messages if these fields are missing in the currently parsed message. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3432).
+* BUGFIX: reduce CPU usage when the regex-based relabeling rules are applied to more than 100K unique Graphite metrics. See [this issue](https://docs.victoriametrics.com/CHANGELOG.html#v1820). The issue was introduced in [v1.82.0](https://docs.victoriametrics.com/CHANGELOG.html#v1820).
+* BUGFIX: do not block [merges](https://docs.victoriametrics.com/#storage) of small parts by merges of big parts on hosts with small number of CPU cores. This issue could result in the increasing number of `storage/small` parts while big merge is in progress. This, in turn, could result in increased CPU usage and memory usage during querying, since queries need to inspect bigger number of small parts. The issue has been introduced in [v1.85.0](https://docs.victoriametrics.com/CHANGELOG.html#v1850).
 
 
 ## [v1.85.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.85.0)
diff --git a/docs/enterprise.md b/docs/enterprise.md
index 0090b88ff..85892de5e 100644
--- a/docs/enterprise.md
+++ b/docs/enterprise.md
@@ -44,6 +44,7 @@ plus the following additional features:
 - [mTLS for cluster components](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#mtls-protection).
 - [Kafka integration](https://docs.victoriametrics.com/vmagent.html#kafka-integration).
 - [Multitenant support in vmalert](https://docs.victoriametrics.com/vmalert.html#multitenancy).
+- [Anomaly Detection Service](https://docs.victoriametrics.com/vmanomaly.html)
 
 On top of this, enterprise package of VictoriaMetrics includes the following important Enterprise features:
 
diff --git a/docs/vmalert.md b/docs/vmalert.md
index d1c890aaf..1c81a7921 100644
--- a/docs/vmalert.md
+++ b/docs/vmalert.md
@@ -217,7 +217,8 @@ The following variables are available in templating:
 | $labels or .Labels                 | The list of labels of the current alert. Use as ".Labels.<label_name>".                                   | {% raw %}Too high number of connections for {{ .Labels.instance }}{% endraw %}                                                                                         |
 | $alertID or .AlertID               | The current alert's ID generated by vmalert.                                                              | {% raw %}Link: vmalert/alert?group_id={{.GroupID}}&alert_id={{.AlertID}}{% endraw %}                                                                                   |
 | $groupID or .GroupID               | The current alert's group ID generated by vmalert.                                                        | {% raw %}Link: vmalert/alert?group_id={{.GroupID}}&alert_id={{.AlertID}}{% endraw %}                                                                                   |
-| $expr or .Expr                     | Alert's expression. Can be used for generating links to Grafana or other systems.                         | {% raw %}/api/v1/query?query={{ $expr&#124;queryEscape }}{% endraw %}                                                                                |
+| $expr or .Expr                     | Alert's expression. Can be used for generating links to Grafana or other systems.                         | {% raw %}/api/v1/query?query={{ $expr&#124;queryEscape }}{% endraw %}                                                                                                  |
+| $for or .For                       | Alert's configured for param.                                                                             | {% raw %}Number of connections is too high for more than {{ .For }}{% endraw %}                                                                                        |
 | $externalLabels or .ExternalLabels | List of labels configured via `-external.label` command-line flag.                                        | {% raw %}Issues with {{ $labels.instance }} (datacenter-{{ $externalLabels.dc }}){% endraw %}                                                                          |
 | $externalURL or .ExternalURL       | URL configured via `-external.url` command-line flag. Used for cases when vmalert is hidden behind proxy. | {% raw %}Visit {{ $externalURL }} for more details{% endraw %}                                                                                                         |
 
diff --git a/docs/vmbackup.md b/docs/vmbackup.md
index 90001a93d..65bed947d 100644
--- a/docs/vmbackup.md
+++ b/docs/vmbackup.md
@@ -25,7 +25,7 @@ creation of hourly, daily, weekly and monthly backups.
 
 * [GCS](https://cloud.google.com/storage/). Example: `gs://<bucket>/<path/to/backup>`
 * [S3](https://aws.amazon.com/s3/). Example: `s3://<bucket>/<path/to/backup>`
-* [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `azblob://<bucket>/<path/to/backup>`
+* [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `azblob://<container>/<path/to/backup>`
 * Any S3-compatible storage such as [MinIO](https://github.com/minio/minio), [Ceph](https://docs.ceph.com/en/pacific/radosgw/s3/) or [Swift](https://platform.swiftstack.com/docs/admin/middleware/s3_middleware.html). See [these docs](#advanced-usage) for details.
 * Local filesystem. Example: `fs://</absolute/path/to/backup>`. Note that `vmbackup` prevents from storing the backup into the directory pointed by `-storageDataPath` command-line flag, since this directory should be managed solely by VictoriaMetrics or `vmstorage`.
 
@@ -191,7 +191,7 @@ See [this article](https://medium.com/@valyala/speeding-up-backups-for-big-time-
   -customS3Endpoint string
      Custom S3 endpoint for use with S3-compatible storages (e.g. MinIO). S3 is used if not set
   -dst string
-     Where to put the backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://bucket/path/to/backup or fs:///path/to/local/backup/dir
+     Where to put the backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://container/path/to/backup or fs:///path/to/local/backup/dir
      -dst can point to the previous backup. In this case incremental backup is performed, i.e. only changed data is uploaded
   -enableTCP6
      Whether to enable IPv6 for listening and dialing. By default only IPv4 TCP and UDP is used
diff --git a/docs/vmrestore.md b/docs/vmrestore.md
index eac807e2b..f6981c22d 100644
--- a/docs/vmrestore.md
+++ b/docs/vmrestore.md
@@ -23,7 +23,7 @@ Run the following command to restore backup from the given `-src` into the given
   `vmrestore` can restore backups from the following storage types:
   * [GCS](https://cloud.google.com/storage/). Example: `-src=gs://<bucket>/<path/to/backup>`
   * [S3](https://aws.amazon.com/s3/). Example: `-src=s3://<bucket>/<path/to/backup>`
-  * [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `-src=azblob://<bucket>/<path/to/backup>`
+  * [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/). Example: `-src=azblob://<container>/<path/to/backup>`
   * Any S3-compatible storage such as [MinIO](https://github.com/minio/minio), [Ceph](https://docs.ceph.com/en/pacific/radosgw/s3/)
     or [Swift](https://platform.swiftstack.com/docs/admin/middleware/s3_middleware.html). See [these docs](#advanced-usage) for details.
   * Local filesystem. Example: `-src=fs://</absolute/path/to/backup>`. Note that `vmbackup` prevents from storing the backup
@@ -166,7 +166,7 @@ i.e. the end result would be similar to [rsync --delete](https://askubuntu.com/q
   -skipBackupCompleteCheck
      Whether to skip checking for 'backup complete' file in -src. This may be useful for restoring from old backups, which were created without 'backup complete' file
   -src string
-     Source path with backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://bucket/path/to/backup or fs:///path/to/local/backup
+     Source path with backup on the remote storage. Example: gs://bucket/path/to/backup, s3://bucket/path/to/backup, azblob://container/path/to/backup or fs:///path/to/local/backup
   -storageDataPath string
      Destination path where backup must be restored. VictoriaMetrics must be stopped when restoring from backup. -storageDataPath dir can be non-empty. In this case the contents of -storageDataPath dir is synchronized with -src contents, i.e. it works like 'rsync --delete' (default "victoria-metrics-data")
   -tls
diff --git a/lib/bytesutil/fast_string_matcher.go b/lib/bytesutil/fast_string_matcher.go
index fdbc2b7e2..d2933563b 100644
--- a/lib/bytesutil/fast_string_matcher.go
+++ b/lib/bytesutil/fast_string_matcher.go
@@ -4,6 +4,8 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // FastStringMatcher implements fast matcher for strings.
@@ -11,44 +13,68 @@ import (
 // It caches string match results and returns them back on the next calls
 // without calling the matchFunc, which may be expensive.
 type FastStringMatcher struct {
-	m    atomic.Value
-	mLen uint64
+	lastCleanupTime uint64
+
+	m sync.Map
 
 	matchFunc func(s string) bool
 }
 
+type fsmEntry struct {
+	lastAccessTime uint64
+	ok             bool
+}
+
 // NewFastStringMatcher creates new matcher, which applies matchFunc to strings passed to Match()
 //
 // matchFunc must return the same result for the same input.
 func NewFastStringMatcher(matchFunc func(s string) bool) *FastStringMatcher {
-	var fsm FastStringMatcher
-	fsm.m.Store(&sync.Map{})
-	fsm.matchFunc = matchFunc
-	return &fsm
+	return &FastStringMatcher{
+		lastCleanupTime: fasttime.UnixTimestamp(),
+		matchFunc:       matchFunc,
+	}
 }
 
 // Match applies matchFunc to s and returns the result.
 func (fsm *FastStringMatcher) Match(s string) bool {
-	m := fsm.m.Load().(*sync.Map)
-	v, ok := m.Load(s)
+	ct := fasttime.UnixTimestamp()
+	v, ok := fsm.m.Load(s)
 	if ok {
 		// Fast path - s match result is found in the cache.
-		bp := v.(*bool)
-		return *bp
+		e := v.(*fsmEntry)
+		if atomic.LoadUint64(&e.lastAccessTime)+10 < ct {
+			// Reduce the frequency of e.lastAccessTime update to once per 10 seconds
+			// in order to improve the fast path speed on systems with many CPU cores.
+			atomic.StoreUint64(&e.lastAccessTime, ct)
+		}
+		return e.ok
 	}
 	// Slow path - run matchFunc for s and store the result in the cache.
 	b := fsm.matchFunc(s)
-	bp := &b
+	e := &fsmEntry{
+		lastAccessTime: ct,
+		ok:             b,
+	}
 	// Make a copy of s in order to limit memory usage to the s length,
 	// since the s may point to bigger string.
 	// This also protects from the case when s contains unsafe string, which points to a temporary byte slice.
 	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3227
 	s = strings.Clone(s)
-	m.Store(s, bp)
-	n := atomic.AddUint64(&fsm.mLen, 1)
-	if n > 100e3 {
-		atomic.StoreUint64(&fsm.mLen, 0)
-		fsm.m.Store(&sync.Map{})
+	fsm.m.Store(s, e)
+
+	if atomic.LoadUint64(&fsm.lastCleanupTime)+61 < ct {
+		// Perform a global cleanup for fsm.m by removing items, which weren't accessed
+		// during the last 5 minutes.
+		atomic.StoreUint64(&fsm.lastCleanupTime, ct)
+		m := &fsm.m
+		m.Range(func(k, v interface{}) bool {
+			e := v.(*fsmEntry)
+			if atomic.LoadUint64(&e.lastAccessTime)+5*60 < ct {
+				m.Delete(k)
+			}
+			return true
+		})
 	}
+
 	return b
 }
diff --git a/lib/bytesutil/fast_string_transformer.go b/lib/bytesutil/fast_string_transformer.go
index 2dfa31af0..c131d15e8 100644
--- a/lib/bytesutil/fast_string_transformer.go
+++ b/lib/bytesutil/fast_string_transformer.go
@@ -4,6 +4,8 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // FastStringTransformer implements fast transformer for strings.
@@ -11,30 +13,41 @@ import (
 // It caches transformed strings and returns them back on the next calls
 // without calling the transformFunc, which may be expensive.
 type FastStringTransformer struct {
-	m    atomic.Value
-	mLen uint64
+	lastCleanupTime uint64
+
+	m sync.Map
 
 	transformFunc func(s string) string
 }
 
+type fstEntry struct {
+	lastAccessTime uint64
+	s              string
+}
+
 // NewFastStringTransformer creates new transformer, which applies transformFunc to strings passed to Transform()
 //
 // transformFunc must return the same result for the same input.
 func NewFastStringTransformer(transformFunc func(s string) string) *FastStringTransformer {
-	var fst FastStringTransformer
-	fst.m.Store(&sync.Map{})
-	fst.transformFunc = transformFunc
-	return &fst
+	return &FastStringTransformer{
+		lastCleanupTime: fasttime.UnixTimestamp(),
+		transformFunc:   transformFunc,
+	}
 }
 
 // Transform applies transformFunc to s and returns the result.
 func (fst *FastStringTransformer) Transform(s string) string {
-	m := fst.m.Load().(*sync.Map)
-	v, ok := m.Load(s)
+	ct := fasttime.UnixTimestamp()
+	v, ok := fst.m.Load(s)
 	if ok {
 		// Fast path - the transformed s is found in the cache.
-		sp := v.(*string)
-		return *sp
+		e := v.(*fstEntry)
+		if atomic.LoadUint64(&e.lastAccessTime)+10 < ct {
+			// Reduce the frequency of e.lastAccessTime update to once per 10 seconds
+			// in order to improve the fast path speed on systems with many CPU cores.
+			atomic.StoreUint64(&e.lastAccessTime, ct)
+		}
+		return e.s
 	}
 	// Slow path - transform s and store it in the cache.
 	sTransformed := fst.transformFunc(s)
@@ -48,12 +61,25 @@ func (fst *FastStringTransformer) Transform(s string) string {
 		// which, in turn, can point to bigger string.
 		sTransformed = s
 	}
-	sp := &sTransformed
-	m.Store(s, sp)
-	n := atomic.AddUint64(&fst.mLen, 1)
-	if n > 100e3 {
-		atomic.StoreUint64(&fst.mLen, 0)
-		fst.m.Store(&sync.Map{})
+	e := &fstEntry{
+		lastAccessTime: ct,
+		s:              sTransformed,
+	}
+	fst.m.Store(s, e)
+
+	if atomic.LoadUint64(&fst.lastCleanupTime)+61 < ct {
+		// Perform a global cleanup for fst.m by removing items, which weren't accessed
+		// during the last 5 minutes.
+		atomic.StoreUint64(&fst.lastCleanupTime, ct)
+		m := &fst.m
+		m.Range(func(k, v interface{}) bool {
+			e := v.(*fstEntry)
+			if atomic.LoadUint64(&e.lastAccessTime)+5*60 < ct {
+				m.Delete(k)
+			}
+			return true
+		})
 	}
+
 	return sTransformed
 }
diff --git a/lib/bytesutil/internstring.go b/lib/bytesutil/internstring.go
index 9318bc980..45c9a6328 100644
--- a/lib/bytesutil/internstring.go
+++ b/lib/bytesutil/internstring.go
@@ -1,35 +1,58 @@
 package bytesutil
 
 import (
+	"strings"
 	"sync"
 	"sync/atomic"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // InternString returns interned s.
 //
 // This may be needed for reducing the amounts of allocated memory.
 func InternString(s string) string {
-	m := internStringsMap.Load().(*sync.Map)
-	if v, ok := m.Load(s); ok {
-		sp := v.(*string)
-		return *sp
+	ct := fasttime.UnixTimestamp()
+	if v, ok := internStringsMap.Load(s); ok {
+		e := v.(*ismEntry)
+		if atomic.LoadUint64(&e.lastAccessTime)+10 < ct {
+			// Reduce the frequency of e.lastAccessTime update to once per 10 seconds
+			// in order to improve the fast path speed on systems with many CPU cores.
+			atomic.StoreUint64(&e.lastAccessTime, ct)
+		}
+		return e.s
 	}
 	// Make a new copy for s in order to remove references from possible bigger string s refers to.
-	sCopy := string(append([]byte{}, s...))
-	m.Store(sCopy, &sCopy)
-	n := atomic.AddUint64(&internStringsMapLen, 1)
-	if n > 100e3 {
-		atomic.StoreUint64(&internStringsMapLen, 0)
-		internStringsMap.Store(&sync.Map{})
+	sCopy := strings.Clone(s)
+	e := &ismEntry{
+		lastAccessTime: ct,
+		s:              sCopy,
+	}
+	internStringsMap.Store(sCopy, e)
+
+	if atomic.LoadUint64(&internStringsMapLastCleanupTime)+61 < ct {
+		// Perform a global cleanup for internStringsMap by removing items, which weren't accessed
+		// during the last 5 minutes.
+		atomic.StoreUint64(&internStringsMapLastCleanupTime, ct)
+		m := &internStringsMap
+		m.Range(func(k, v interface{}) bool {
+			e := v.(*ismEntry)
+			if atomic.LoadUint64(&e.lastAccessTime)+5*60 < ct {
+				m.Delete(k)
+			}
+			return true
+		})
 	}
+
 	return sCopy
 }
 
+type ismEntry struct {
+	lastAccessTime uint64
+	s              string
+}
+
 var (
-	internStringsMap    atomic.Value
-	internStringsMapLen uint64
+	internStringsMap                sync.Map
+	internStringsMapLastCleanupTime uint64
 )
-
-func init() {
-	internStringsMap.Store(&sync.Map{})
-}
diff --git a/lib/mergeset/table.go b/lib/mergeset/table.go
index 08b1e4e37..aa921b9e1 100644
--- a/lib/mergeset/table.go
+++ b/lib/mergeset/table.go
@@ -25,8 +25,17 @@ import (
 // maxInmemoryParts is the maximum number of inmemory parts in the table.
 //
 // This number may be reached when the insertion pace outreaches merger pace.
+// If this number is reached, then assisted merges are performed
+// during data ingestion.
 const maxInmemoryParts = 64
 
+// maxFileParts is the maximum number of file parts in the table.
+//
+// This number may be reached when the insertion pace outreaches merger pace.
+// If this number is reached, then assisted merges are performed
+// during data ingestion.
+const maxFileParts = 256
+
 // Default number of parts to merge at once.
 //
 // This number has been obtained empirically - it gives the lowest possible overhead.
@@ -98,7 +107,8 @@ type Table struct {
 	inmemoryItemsMerged uint64
 	fileItemsMerged     uint64
 
-	assistedInmemoryMerges uint64
+	inmemoryAssistedMerges uint64
+	fileAssistedMerges     uint64
 
 	itemsAdded          uint64
 	itemsAddedSizeBytes uint64
@@ -419,7 +429,8 @@ type TableMetrics struct {
 	InmemoryItemsMerged uint64
 	FileItemsMerged     uint64
 
-	AssistedInmemoryMerges uint64
+	InmemoryAssistedMerges uint64
+	FileAssistedMerges     uint64
 
 	ItemsAdded          uint64
 	ItemsAddedSizeBytes uint64
@@ -469,7 +480,8 @@ func (tb *Table) UpdateMetrics(m *TableMetrics) {
 	m.InmemoryItemsMerged += atomic.LoadUint64(&tb.inmemoryItemsMerged)
 	m.FileItemsMerged += atomic.LoadUint64(&tb.fileItemsMerged)
 
-	m.AssistedInmemoryMerges += atomic.LoadUint64(&tb.assistedInmemoryMerges)
+	m.InmemoryAssistedMerges += atomic.LoadUint64(&tb.inmemoryAssistedMerges)
+	m.FileAssistedMerges += atomic.LoadUint64(&tb.fileAssistedMerges)
 
 	m.ItemsAdded += atomic.LoadUint64(&tb.itemsAdded)
 	m.ItemsAddedSizeBytes += atomic.LoadUint64(&tb.itemsAddedSizeBytes)
@@ -739,9 +751,8 @@ func (tb *Table) flushBlocksToParts(ibs []*inmemoryBlock, isFinal bool) {
 
 	flushConcurrencyCh <- struct{}{}
 	tb.assistedMergeForInmemoryParts()
+	tb.assistedMergeForFileParts()
 	<-flushConcurrencyCh
-	// There is no need in assited merge for file parts,
-	// since the bottleneck is possible only at inmemory parts.
 
 	if tb.flushCallback != nil {
 		if isFinal {
@@ -765,10 +776,10 @@ func (tb *Table) assistedMergeForInmemoryParts() {
 
 		// Prioritize assisted merges over searches.
 		storagepacelimiter.Search.Inc()
+		atomic.AddUint64(&tb.inmemoryAssistedMerges, 1)
 		err := tb.mergeInmemoryParts()
 		storagepacelimiter.Search.Dec()
 		if err == nil {
-			atomic.AddUint64(&tb.assistedInmemoryMerges, 1)
 			continue
 		}
 		if errors.Is(err, errNothingToMerge) || errors.Is(err, errForciblyStopped) {
@@ -778,6 +789,30 @@ func (tb *Table) assistedMergeForInmemoryParts() {
 	}
 }
 
+func (tb *Table) assistedMergeForFileParts() {
+	for {
+		tb.partsLock.Lock()
+		ok := getNotInMergePartsCount(tb.fileParts) < maxFileParts
+		tb.partsLock.Unlock()
+		if ok {
+			return
+		}
+
+		// Prioritize assisted merges over searches.
+		storagepacelimiter.Search.Inc()
+		atomic.AddUint64(&tb.fileAssistedMerges, 1)
+		err := tb.mergeExistingParts(false)
+		storagepacelimiter.Search.Dec()
+		if err == nil {
+			continue
+		}
+		if errors.Is(err, errNothingToMerge) || errors.Is(err, errForciblyStopped) || errors.Is(err, errReadOnlyMode) {
+			return
+		}
+		logger.Panicf("FATAL: cannot assist with merging file parts: %s", err)
+	}
+}
+
 func getNotInMergePartsCount(pws []*partWrapper) int {
 	n := 0
 	for _, pw := range pws {
@@ -866,7 +901,10 @@ func newPartWrapperFromInmemoryPart(mp *inmemoryPart, flushToDiskDeadline time.T
 }
 
 func (tb *Table) startMergeWorkers() {
-	for i := 0; i < cap(mergeWorkersLimitCh); i++ {
+	// Start a merge worker per available CPU core.
+	// The actual number of concurrent merges is limited inside mergeWorker() below.
+	workersCount := cgroup.AvailableCPUs()
+	for i := 0; i < workersCount; i++ {
 		tb.wg.Add(1)
 		go func() {
 			tb.mergeWorker()
@@ -940,6 +978,8 @@ func (tb *Table) mergeWorker() {
 	isFinal := false
 	t := time.NewTimer(sleepTime)
 	for {
+		// Limit the number of concurrent calls to mergeExistingParts, since the total number of merge workers
+		// across tables may exceed the the cap(mergeWorkersLimitCh).
 		mergeWorkersLimitCh <- struct{}{}
 		err := tb.mergeExistingParts(isFinal)
 		<-mergeWorkersLimitCh
diff --git a/lib/storage/partition.go b/lib/storage/partition.go
index 020e6a320..77f07a244 100644
--- a/lib/storage/partition.go
+++ b/lib/storage/partition.go
@@ -31,8 +31,15 @@ import (
 const maxBigPartSize = 1e12
 
 // The maximum number of inmemory parts in the partition.
+//
+// If the number of inmemory parts reaches this value, then assisted merge runs during data ingestion.
 const maxInmemoryPartsPerPartition = 32
 
+// The maximum number of small parts in the partition.
+//
+// If the number of small parts reaches this value, then assisted merge runs during data ingestion.
+const maxSmallPartsPerPartition = 128
+
 // Default number of parts to merge at once.
 //
 // This number has been obtained empirically - it gives the lowest possible overhead.
@@ -112,6 +119,7 @@ type partition struct {
 	bigRowsDeleted      uint64
 
 	inmemoryAssistedMerges uint64
+	smallAssistedMerges    uint64
 
 	mergeNeedFreeDiskSpace uint64
 
@@ -338,6 +346,7 @@ type partitionMetrics struct {
 	BigPartsRefCount      uint64
 
 	InmemoryAssistedMerges uint64
+	SmallAssistedMerges    uint64
 
 	MergeNeedFreeDiskSpace uint64
 }
@@ -404,6 +413,7 @@ func (pt *partition) UpdateMetrics(m *partitionMetrics) {
 	m.BigRowsDeleted += atomic.LoadUint64(&pt.bigRowsDeleted)
 
 	m.InmemoryAssistedMerges += atomic.LoadUint64(&pt.inmemoryAssistedMerges)
+	m.SmallAssistedMerges += atomic.LoadUint64(&pt.smallAssistedMerges)
 
 	m.MergeNeedFreeDiskSpace += atomic.LoadUint64(&pt.mergeNeedFreeDiskSpace)
 }
@@ -576,6 +586,7 @@ func (pt *partition) flushRowsToParts(rows []rawRow) {
 
 	flushConcurrencyCh <- struct{}{}
 	pt.assistedMergeForInmemoryParts()
+	pt.assistedMergeForSmallParts()
 	<-flushConcurrencyCh
 	// There is no need in assisted merges for small and big parts,
 	// since the bottleneck is possible only at inmemory parts.
@@ -597,10 +608,10 @@ func (pt *partition) assistedMergeForInmemoryParts() {
 		// Assist with mering inmemory parts.
 		// Prioritize assisted merges over searches.
 		storagepacelimiter.Search.Inc()
+		atomic.AddUint64(&pt.inmemoryAssistedMerges, 1)
 		err := pt.mergeInmemoryParts()
 		storagepacelimiter.Search.Dec()
 		if err == nil {
-			atomic.AddUint64(&pt.inmemoryAssistedMerges, 1)
 			continue
 		}
 		if errors.Is(err, errNothingToMerge) || errors.Is(err, errForciblyStopped) {
@@ -610,6 +621,33 @@ func (pt *partition) assistedMergeForInmemoryParts() {
 	}
 }
 
+func (pt *partition) assistedMergeForSmallParts() {
+	for {
+		pt.partsLock.Lock()
+		ok := getNotInMergePartsCount(pt.smallParts) < maxSmallPartsPerPartition
+		pt.partsLock.Unlock()
+		if ok {
+			return
+		}
+
+		// There are too many unmerged small parts.
+		// This usually means that the app cannot keep up with the data ingestion rate.
+		// Assist with mering small parts.
+		// Prioritize assisted merges over searches.
+		storagepacelimiter.Search.Inc()
+		atomic.AddUint64(&pt.smallAssistedMerges, 1)
+		err := pt.mergeExistingParts(false)
+		storagepacelimiter.Search.Dec()
+		if err == nil {
+			continue
+		}
+		if errors.Is(err, errNothingToMerge) || errors.Is(err, errForciblyStopped) || errors.Is(err, errReadOnlyMode) {
+			return
+		}
+		logger.Panicf("FATAL: cannot merge small parts: %s", err)
+	}
+}
+
 func getNotInMergePartsCount(pws []*partWrapper) int {
 	n := 0
 	for _, pw := range pws {
@@ -981,7 +1019,10 @@ func SetMergeWorkersCount(n int) {
 }
 
 func (pt *partition) startMergeWorkers() {
-	for i := 0; i < cap(mergeWorkersLimitCh); i++ {
+	// Start a merge worker per available CPU core.
+	// The actual number of concurrent merges is limited inside mergeWorker() below.
+	workersCount := cgroup.AvailableCPUs()
+	for i := 0; i < workersCount; i++ {
 		pt.wg.Add(1)
 		go func() {
 			pt.mergeWorker()
@@ -1001,7 +1042,8 @@ func (pt *partition) mergeWorker() {
 	isFinal := false
 	t := time.NewTimer(sleepTime)
 	for {
-		// Limit the number of concurrent calls to mergeExistingParts, cine the number of merge
+		// Limit the number of concurrent calls to mergeExistingParts, since the total number of merge workers
+		// across partitions may exceed the the cap(mergeWorkersLimitCh).
 		mergeWorkersLimitCh <- struct{}{}
 		err := pt.mergeExistingParts(isFinal)
 		<-mergeWorkersLimitCh
@@ -1092,10 +1134,10 @@ func (pt *partition) getMaxBigPartSize() uint64 {
 
 func getMaxOutBytes(path string, workersCount int) uint64 {
 	n := fs.MustGetFreeSpace(path)
-	// Do not substract freeDiskSpaceLimitBytes from n before calculating the maxOutBytes,
+	// Do not subtract freeDiskSpaceLimitBytes from n before calculating the maxOutBytes,
 	// since this will result in sub-optimal merges - e.g. many small parts will be left unmerged.
 
-	// Divide free space by the max number concurrent merges.
+	// Divide free space by the max number of concurrent merges.
 	maxOutBytes := n / uint64(workersCount)
 	if maxOutBytes > maxBigPartSize {
 		maxOutBytes = maxBigPartSize


Test patch:
diff --git a/app/vmalert/notifier/alert_test.go b/app/vmalert/notifier/alert_test.go
index 7032bceab..9999f6f42 100644
--- a/app/vmalert/notifier/alert_test.go
+++ b/app/vmalert/notifier/alert_test.go
@@ -54,14 +54,15 @@ func TestAlert_ExecTemplate(t *testing.T) {
 					"job":      "staging",
 					"instance": "localhost",
 				},
+				For: 5 * time.Minute,
 			},
 			annotations: map[string]string{
 				"summary":     "Too high connection number for {{$labels.instance}} for job {{$labels.job}}",
-				"description": "It is {{ $value }} connections for {{$labels.instance}}",
+				"description": "It is {{ $value }} connections for {{$labels.instance}} for more than {{ .For }}",
 			},
 			expTpl: map[string]string{
 				"summary":     "Too high connection number for localhost for job staging",
-				"description": "It is 10000 connections for localhost",
+				"description": "It is 10000 connections for localhost for more than 5m0s",
 			},
 		},
 		{
@@ -152,7 +153,7 @@ func TestAlert_ExecTemplate(t *testing.T) {
 			},
 		},
 		{
-			name: "ActiveAt custome format",
+			name: "ActiveAt custom format",
 			alert: &Alert{
 				ActiveAt: time.Date(2022, 8, 19, 20, 34, 58, 651387237, time.UTC),
 			},