Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1249 (2021-04-27 06:16:29)

changes vmalert Querier with per rule querier
it allows to changes some parametrs based on rule setting
for instance - alert type, tenant for cluster version or event endpoint url.


Patch:
diff --git a/app/vmalert/alerting.go b/app/vmalert/alerting.go
index 9eff009bc..c454d66bc 100644
--- a/app/vmalert/alerting.go
+++ b/app/vmalert/alerting.go
@@ -29,6 +29,8 @@ type AlertingRule struct {
 	GroupID     uint64
 	GroupName   string
 
+	q datasource.Querier
+
 	// guard status fields
 	mu sync.RWMutex
 	// stores list of active alerts
@@ -49,7 +51,7 @@ type alertingRuleMetrics struct {
 	active  *gauge
 }
 
-func newAlertingRule(group *Group, cfg config.Rule) *AlertingRule {
+func newAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule) *AlertingRule {
 	ar := &AlertingRule{
 		Type:        cfg.Type,
 		RuleID:      cfg.ID,
@@ -60,6 +62,7 @@ func newAlertingRule(group *Group, cfg config.Rule) *AlertingRule {
 		Annotations: cfg.Annotations,
 		GroupID:     group.ID(),
 		GroupName:   group.Name,
+		q:           qb.BuildWithParams(datasource.QuerierParams{DataSourceType: &cfg.Type}),
 		alerts:      make(map[uint64]*notifier.Alert),
 		metrics:     &alertingRuleMetrics{},
 	}
@@ -121,8 +124,8 @@ func (ar *AlertingRule) ID() uint64 {
 
 // Exec executes AlertingRule expression via the given Querier.
 // Based on the Querier results AlertingRule maintains notifier.Alerts
-func (ar *AlertingRule) Exec(ctx context.Context, q datasource.Querier, series bool) ([]prompbmarshal.TimeSeries, error) {
-	qMetrics, err := q.Query(ctx, ar.Expr, ar.Type)
+func (ar *AlertingRule) Exec(ctx context.Context, series bool) ([]prompbmarshal.TimeSeries, error) {
+	qMetrics, err := ar.q.Query(ctx, ar.Expr)
 	ar.mu.Lock()
 	defer ar.mu.Unlock()
 
@@ -139,7 +142,7 @@ func (ar *AlertingRule) Exec(ctx context.Context, q datasource.Querier, series b
 		}
 	}
 
-	qFn := func(query string) ([]datasource.Metric, error) { return q.Query(ctx, query, ar.Type) }
+	qFn := func(query string) ([]datasource.Metric, error) { return ar.q.Query(ctx, query) }
 	updated := make(map[uint64]struct{})
 	// update list of active alerts
 	for _, m := range qMetrics {
@@ -407,7 +410,7 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
 		return fmt.Errorf("querier is nil")
 	}
 
-	qFn := func(query string) ([]datasource.Metric, error) { return q.Query(ctx, query, ar.Type) }
+	qFn := func(query string) ([]datasource.Metric, error) { return ar.q.Query(ctx, query) }
 
 	// account for external labels in filter
 	var labelsFilter string
@@ -420,7 +423,7 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
 	// remote write protocol which is used for state persistence in vmalert.
 	expr := fmt.Sprintf("last_over_time(%s{alertname=%q%s}[%ds])",
 		alertForStateMetricName, ar.Name, labelsFilter, int(lookback.Seconds()))
-	qMetrics, err := q.Query(ctx, expr, ar.Type)
+	qMetrics, err := q.Query(ctx, expr)
 	if err != nil {
 		return err
 	}
diff --git a/app/vmalert/datasource/datasource.go b/app/vmalert/datasource/datasource.go
index 4625d09ee..fbdd36abc 100644
--- a/app/vmalert/datasource/datasource.go
+++ b/app/vmalert/datasource/datasource.go
@@ -4,11 +4,16 @@ import (
 	"context"
 )
 
+// QuerierBuilder builds Querier with given params.
+type QuerierBuilder interface {
+	BuildWithParams(params QuerierParams) Querier
+}
+
 // Querier interface wraps Query method which
 // executes given query and returns list of Metrics
 // as result
 type Querier interface {
-	Query(ctx context.Context, query string, engine Type) ([]Metric, error)
+	Query(ctx context.Context, query string) ([]Metric, error)
 }
 
 // Metric is the basic entity which should be return by datasource
diff --git a/app/vmalert/datasource/init.go b/app/vmalert/datasource/init.go
index 9d500fa38..7d27b9d55 100644
--- a/app/vmalert/datasource/init.go
+++ b/app/vmalert/datasource/init.go
@@ -28,7 +28,7 @@ var (
 )
 
 // Init creates a Querier from provided flag values.
-func Init() (Querier, error) {
+func Init() (QuerierBuilder, error) {
 	if *addr == "" {
 		return nil, fmt.Errorf("datasource.url is empty")
 	}
diff --git a/app/vmalert/datasource/vm.go b/app/vmalert/datasource/vm.go
index e06b3af3d..6e92db2c1 100644
--- a/app/vmalert/datasource/vm.go
+++ b/app/vmalert/datasource/vm.go
@@ -81,6 +81,7 @@ type VMStorage struct {
 	appendTypePrefix bool
 	lookBack         time.Duration
 	queryStep        time.Duration
+	dataSourceType   Type
 }
 
 const queryPath = "/api/v1/query"
@@ -89,6 +90,38 @@ const graphitePath = "/render"
 const prometheusPrefix = "/prometheus"
 const graphitePrefix = "/graphite"
 
+// QuerierParams params for Querier.
+type QuerierParams struct {
+	DataSourceType *Type
+}
+
+// Clone makes clone of VMStorage, shares http client.
+func (s *VMStorage) Clone() *VMStorage {
+	return &VMStorage{
+		c:                s.c,
+		datasourceURL:    s.datasourceURL,
+		basicAuthUser:    s.basicAuthUser,
+		basicAuthPass:    s.basicAuthPass,
+		lookBack:         s.lookBack,
+		queryStep:        s.queryStep,
+		appendTypePrefix: s.appendTypePrefix,
+		dataSourceType:   s.dataSourceType,
+	}
+}
+
+// ApplyParams - changes given querier params.
+func (s *VMStorage) ApplyParams(params QuerierParams) *VMStorage {
+	if params.DataSourceType != nil {
+		s.dataSourceType = *params.DataSourceType
+	}
+	return s
+}
+
+// BuildWithParams - implements interface.
+func (s *VMStorage) BuildWithParams(params QuerierParams) Querier {
+	return s.Clone().ApplyParams(params)
+}
+
 // NewVMStorage is a constructor for VMStorage
 func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, lookBack time.Duration, queryStep time.Duration, appendTypePrefix bool, c *http.Client) *VMStorage {
 	return &VMStorage{
@@ -99,18 +132,19 @@ func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, lookBack time.Du
 		appendTypePrefix: appendTypePrefix,
 		lookBack:         lookBack,
 		queryStep:        queryStep,
+		dataSourceType:   NewPrometheusType(),
 	}
 }
 
 // Query reads metrics from datasource by given query and type
-func (s *VMStorage) Query(ctx context.Context, query string, dataSourceType Type) ([]Metric, error) {
-	switch dataSourceType.name {
+func (s *VMStorage) Query(ctx context.Context, query string) ([]Metric, error) {
+	switch s.dataSourceType.name {
 	case "", prometheusType:
 		return s.queryDataSource(ctx, query, s.setPrometheusReqParams, parsePrometheusResponse)
 	case graphiteType:
 		return s.queryDataSource(ctx, query, s.setGraphiteReqParams, parseGraphiteResponse)
 	default:
-		return nil, fmt.Errorf("engine not found: %q", dataSourceType)
+		return nil, fmt.Errorf("engine not found: %q", s.dataSourceType.name)
 	}
 }
 
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index e8a73b25f..f9ded5d86 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -49,7 +49,7 @@ func newGroupMetrics(name, file string) *groupMetrics {
 	return m
 }
 
-func newGroup(cfg config.Group, defaultInterval time.Duration, labels map[string]string) *Group {
+func newGroup(cfg config.Group, qb datasource.QuerierBuilder, defaultInterval time.Duration, labels map[string]string) *Group {
 	g := &Group{
 		Type:        cfg.Type,
 		Name:        cfg.Name,
@@ -81,17 +81,17 @@ func newGroup(cfg config.Group, defaultInterval time.Duration, labels map[string
 			}
 			r.Labels[k] = v
 		}
-		rules[i] = g.newRule(r)
+		rules[i] = g.newRule(qb, r)
 	}
 	g.Rules = rules
 	return g
 }
 
-func (g *Group) newRule(rule config.Rule) Rule {
+func (g *Group) newRule(qb datasource.QuerierBuilder, rule config.Rule) Rule {
 	if rule.Alert != "" {
-		return newAlertingRule(g, rule)
+		return newAlertingRule(qb, g, rule)
 	}
-	return newRecordingRule(g, rule)
+	return newRecordingRule(qb, g, rule)
 }
 
 // ID return unique group ID that consists of
@@ -106,7 +106,7 @@ func (g *Group) ID() uint64 {
 }
 
 // Restore restores alerts state for group rules
-func (g *Group) Restore(ctx context.Context, q datasource.Querier, lookback time.Duration, labels map[string]string) error {
+func (g *Group) Restore(ctx context.Context, qb datasource.QuerierBuilder, lookback time.Duration, labels map[string]string) error {
 	for _, rule := range g.Rules {
 		rr, ok := rule.(*AlertingRule)
 		if !ok {
@@ -115,6 +115,7 @@ func (g *Group) Restore(ctx context.Context, q datasource.Querier, lookback time
 		if rr.For < 1 {
 			continue
 		}
+		q := qb.BuildWithParams(datasource.QuerierParams{})
 		if err := rr.Restore(ctx, q, lookback, labels); err != nil {
 			return fmt.Errorf("error while restoring rule %q: %w", rule, err)
 		}
@@ -189,7 +190,7 @@ func (g *Group) close() {
 
 var skipRandSleepOnGroupStart bool
 
-func (g *Group) start(ctx context.Context, querier datasource.Querier, nts []notifier.Notifier, rw *remotewrite.Client) {
+func (g *Group) start(ctx context.Context, nts []notifier.Notifier, rw *remotewrite.Client) {
 	defer func() { close(g.finishedCh) }()
 
 	// Spread group rules evaluation over time in order to reduce load on VictoriaMetrics.
@@ -213,7 +214,7 @@ func (g *Group) start(ctx context.Context, querier datasource.Querier, nts []not
 	}
 
 	logger.Infof("group %q started; interval=%v; concurrency=%d", g.Name, g.Interval, g.Concurrency)
-	e := &executor{querier, nts, rw}
+	e := &executor{nts, rw}
 	t := time.NewTicker(g.Interval)
 	defer t.Stop()
 	for {
@@ -256,7 +257,6 @@ func (g *Group) start(ctx context.Context, querier datasource.Querier, nts []not
 }
 
 type executor struct {
-	querier   datasource.Querier
 	notifiers []notifier.Notifier
 	rw        *remotewrite.Client
 }
@@ -310,7 +310,7 @@ func (e *executor) exec(ctx context.Context, rule Rule, returnSeries bool, inter
 		execDuration.UpdateDuration(execStart)
 	}()
 
-	tss, err := rule.Exec(ctx, e.querier, returnSeries)
+	tss, err := rule.Exec(ctx, returnSeries)
 	if err != nil {
 		execErrors.Inc()
 		return fmt.Errorf("rule %q: failed to execute: %w", rule, err)
diff --git a/app/vmalert/main.go b/app/vmalert/main.go
index a76fe5869..a2a5ff75a 100644
--- a/app/vmalert/main.go
+++ b/app/vmalert/main.go
@@ -140,10 +140,10 @@ func newManager(ctx context.Context) (*manager, error) {
 	}
 
 	manager := &manager{
-		groups:    make(map[uint64]*Group),
-		querier:   q,
-		notifiers: nts,
-		labels:    map[string]string{},
+		groups:         make(map[uint64]*Group),
+		querierBuilder: q,
+		notifiers:      nts,
+		labels:         map[string]string{},
 	}
 	rw, err := remotewrite.Init(ctx)
 	if err != nil {
diff --git a/app/vmalert/manager.go b/app/vmalert/manager.go
index baaf0effb..cd248f5bd 100644
--- a/app/vmalert/manager.go
+++ b/app/vmalert/manager.go
@@ -15,11 +15,12 @@ import (
 
 // manager controls group states
 type manager struct {
-	querier   datasource.Querier
-	notifiers []notifier.Notifier
+	querierBuilder datasource.QuerierBuilder
+	notifiers      []notifier.Notifier
 
 	rw *remotewrite.Client
-	rr datasource.Querier
+	// remote read builder.
+	rr datasource.QuerierBuilder
 
 	wg     sync.WaitGroup
 	labels map[string]string
@@ -74,7 +75,7 @@ func (m *manager) startGroup(ctx context.Context, group *Group, restore bool) {
 	m.wg.Add(1)
 	id := group.ID()
 	go func() {
-		group.start(ctx, m.querier, m.notifiers, m.rw)
+		group.start(ctx, m.notifiers, m.rw)
 		m.wg.Done()
 	}()
 	m.groups[id] = group
@@ -89,7 +90,7 @@ func (m *manager) update(ctx context.Context, path []string, validateTpl, valida
 
 	groupsRegistry := make(map[uint64]*Group)
 	for _, cfg := range groupsCfg {
-		ng := newGroup(cfg, *evaluationInterval, m.labels)
+		ng := newGroup(cfg, m.querierBuilder, *evaluationInterval, m.labels)
 		groupsRegistry[ng.ID()] = ng
 	}
 
diff --git a/app/vmalert/recording.go b/app/vmalert/recording.go
index 5e8f780f0..698c19b86 100644
--- a/app/vmalert/recording.go
+++ b/app/vmalert/recording.go
@@ -25,6 +25,8 @@ type RecordingRule struct {
 	Labels  map[string]string
 	GroupID uint64
 
+	q datasource.Querier
+
 	// guard status fields
 	mu sync.RWMutex
 	// stores last moment of time Exec was called
@@ -52,7 +54,7 @@ func (rr *RecordingRule) ID() uint64 {
 	return rr.RuleID
 }
 
-func newRecordingRule(group *Group, cfg config.Rule) *RecordingRule {
+func newRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule) *RecordingRule {
 	rr := &RecordingRule{
 		Type:    cfg.Type,
 		RuleID:  cfg.ID,
@@ -61,6 +63,7 @@ func newRecordingRule(group *Group, cfg config.Rule) *RecordingRule {
 		Labels:  cfg.Labels,
 		GroupID: group.ID(),
 		metrics: &recordingRuleMetrics{},
+		q:       qb.BuildWithParams(datasource.QuerierParams{DataSourceType: &cfg.Type}),
 	}
 
 	labels := fmt.Sprintf(`recording=%q, group=%q, id="%d"`, rr.Name, group.Name, rr.ID())
@@ -82,12 +85,12 @@ func (rr *RecordingRule) Close() {
 }
 
 // Exec executes RecordingRule expression via the given Querier.
-func (rr *RecordingRule) Exec(ctx context.Context, q datasource.Querier, series bool) ([]prompbmarshal.TimeSeries, error) {
+func (rr *RecordingRule) Exec(ctx context.Context, series bool) ([]prompbmarshal.TimeSeries, error) {
 	if !series {
 		return nil, nil
 	}
 
-	qMetrics, err := q.Query(ctx, rr.Expr, rr.Type)
+	qMetrics, err := rr.q.Query(ctx, rr.Expr)
 	rr.mu.Lock()
 	defer rr.mu.Unlock()
 
diff --git a/app/vmalert/remoteread/init.go b/app/vmalert/remoteread/init.go
index 4a2153c10..3900989b4 100644
--- a/app/vmalert/remoteread/init.go
+++ b/app/vmalert/remoteread/init.go
@@ -26,7 +26,7 @@ var (
 
 // Init creates a Querier from provided flag values.
 // Returns nil if addr flag wasn't set.
-func Init() (datasource.Querier, error) {
+func Init() (datasource.QuerierBuilder, error) {
 	if *addr == "" {
 		return nil, nil
 	}
diff --git a/app/vmalert/rule.go b/app/vmalert/rule.go
index 293e929bf..7aa3ddff0 100644
--- a/app/vmalert/rule.go
+++ b/app/vmalert/rule.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
@@ -18,7 +17,7 @@ type Rule interface {
 	// Exec executes the rule with given context
 	// and Querier. If returnSeries is true, Exec
 	// may return TimeSeries as result of execution
-	Exec(ctx context.Context, q datasource.Querier, returnSeries bool) ([]prompbmarshal.TimeSeries, error)
+	Exec(ctx context.Context, returnSeries bool) ([]prompbmarshal.TimeSeries, error)
 	// UpdateWith performs modification of current Rule
 	// with fields of the given Rule.
 	UpdateWith(Rule) error
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index bfd636495..5c0270c62 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -9,6 +9,7 @@ sort: 15
 * FEATURE: vmauth: add ability to set madatory query args in `url_prefix`. For example, `url_prefix: http://vm:8428/?extra_label=team=dev` would add `extra_label=team=dev` query arg to all the incoming requests. See [the example](https://docs.victoriametrics.com/vmauth.html#auth-config) for more details.
 * FEATURE: add OpenTSDB migration option to vmctl. See more details [here](https://docs.victoriametrics.com/vmctl#migrating-data-from-opentsdb).
 Thanks to @johnseekins!
+* FEATURE: improved new time series registration speed on systems with many CPU cores. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1244). Thanks to @waldoweng for the idea and [draft implementation](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/1243).
 
 * BUGFIX: vmagent: properly update `role: endpoints` and `role: endpointslices` scrape targets if the underlying service changes in `kubernetes_sd_config`. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1240).
 * BUGFIX: vmagent: apply `scrape_timeout` on receiving the first response byte from `stream_parse: true` scrape targets. Previously it was applied to receiving and *processing* the full response stream. This could result in false timeout errors when scrape target exposes millions of metrics as described [here](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1017#issuecomment-767235047).
diff --git a/lib/mergeset/table.go b/lib/mergeset/table.go
index a5ce4d4ff..ddd0bac40 100644
--- a/lib/mergeset/table.go
+++ b/lib/mergeset/table.go
@@ -103,9 +103,10 @@ type Table struct {
 	partsLock sync.Mutex
 	parts     []*partWrapper
 
-	rawItemsBlocks        []*inmemoryBlock
-	rawItemsLock          sync.Mutex
-	rawItemsLastFlushTime uint64
+	// rawItems contains recently added items that haven't been converted to parts yet.
+	//
+	// rawItems aren't used in search for performance reasons
+	rawItems rawItemsShards
 
 	snapshotLock sync.RWMutex
 
@@ -124,6 +125,97 @@ type Table struct {
 	rawItemsPendingFlushesWG syncwg.WaitGroup
 }
 
+type rawItemsShards struct {
+	shardIdx uint32
+
+	// shards reduce lock contention when adding rows on multi-CPU systems.
+	shards []rawItemsShard
+}
+
+// The number of shards for rawItems per table.
+//
+// Higher number of shards reduces CPU contention and increases the max bandwidth on multi-core systems.
+var rawItemsShardsPerTable = cgroup.AvailableCPUs()
+
+const maxBlocksPerShard = 512
+
+func (riss *rawItemsShards) init() {
+	riss.shards = make([]rawItemsShard, rawItemsShardsPerTable)
+}
+
+func (riss *rawItemsShards) addItems(tb *Table, items [][]byte) error {
+	n := atomic.AddUint32(&riss.shardIdx, 1)
+	shards := riss.shards
+	idx := n % uint32(len(shards))
+	shard := &shards[idx]
+	return shard.addItems(tb, items)
+}
+
+func (riss *rawItemsShards) Len() int {
+	n := 0
+	for i := range riss.shards {
+		n += riss.shards[i].Len()
+	}
+	return n
+}
+
+type rawItemsShard struct {
+	mu            sync.Mutex
+	ibs           []*inmemoryBlock
+	lastFlushTime uint64
+}
+
+func (ris *rawItemsShard) Len() int {
+	ris.mu.Lock()
+	n := 0
+	for _, ib := range ris.ibs {
+		n += len(ib.items)
+	}
+	ris.mu.Unlock()
+	return n
+}
+
+func (ris *rawItemsShard) addItems(tb *Table, items [][]byte) error {
+	var err error
+	var blocksToMerge []*inmemoryBlock
+
+	ris.mu.Lock()
+	ibs := ris.ibs
+	if len(ibs) == 0 {
+		ib := getInmemoryBlock()
+		ibs = append(ibs, ib)
+		ris.ibs = ibs
+	}
+	ib := ibs[len(ibs)-1]
+	for _, item := range items {
+		if !ib.Add(item) {
+			ib = getInmemoryBlock()
+			if !ib.Add(item) {
+				putInmemoryBlock(ib)
+				err = fmt.Errorf("cannot insert an item %q into an empty inmemoryBlock; it looks like the item is too large? len(item)=%d", item, len(item))
+				break
+			}
+			ibs = append(ibs, ib)
+			ris.ibs = ibs
+		}
+	}
+	if len(ibs) >= maxBlocksPerShard {
+		blocksToMerge = ibs
+		ris.ibs = make([]*inmemoryBlock, 0, maxBlocksPerShard)
+		ris.lastFlushTime = fasttime.UnixTimestamp()
+	}
+	ris.mu.Unlock()
+
+	if blocksToMerge == nil {
+		// Fast path.
+		return err
+	}
+
+	// Slow path: merge blocksToMerge.
+	tb.mergeRawItemsBlocks(blocksToMerge)
+	return err
+}
+
 type partWrapper struct {
 	p *part
 
@@ -195,6 +287,7 @@ func OpenTable(path string, flushCallback func(), prepareBlock PrepareBlockCallb
 		flockF:        flockF,
 		stopCh:        make(chan struct{}),
 	}
+	tb.rawItems.init()
 	tb.startPartMergers()
 	tb.startRawItemsFlusher()
 
@@ -314,11 +407,7 @@ func (tb *Table) UpdateMetrics(m *TableMetrics) {
 	m.ItemsMerged += atomic.LoadUint64(&tb.itemsMerged)
 	m.AssistedMerges += atomic.LoadUint64(&tb.assistedMerges)
 
-	tb.rawItemsLock.Lock()
-	for _, ib := range tb.rawItemsBlocks {
-		m.PendingItems += uint64(len(ib.items))
-	}
-	tb.rawItemsLock.Unlock()
+	m.PendingItems += uint64(tb.rawItems.Len())
 
 	tb.partsLock.Lock()
 	m.PartsCount += uint64(len(tb.parts))
@@ -352,42 +441,10 @@ func (tb *Table) UpdateMetrics(m *TableMetrics) {
 
 // AddItems adds the given items to the tb.
 func (tb *Table) AddItems(items [][]byte) error {
-	var err error
-	var blocksToMerge []*inmemoryBlock
-
-	tb.rawItemsLock.Lock()
-	if len(tb.rawItemsBlocks) == 0 {
-		ib := getInmemoryBlock()
-		tb.rawItemsBlocks = append(tb.rawItemsBlocks, ib)
-	}
-	ib := tb.rawItemsBlocks[len(tb.rawItemsBlocks)-1]
-	for _, item := range items {
-		if !ib.Add(item) {
-			ib = getInmemoryBlock()
-			if !ib.Add(item) {
-				putInmemoryBlock(ib)
-				err = fmt.Errorf("cannot insert an item %q into an empty inmemoryBlock on %q; it looks like the item is too large? len(item)=%d",
-					item, tb.path, len(item))
-				break
-			}
-			tb.rawItemsBlocks = append(tb.rawItemsBlocks, ib)
-		}
-	}
-	if len(tb.rawItemsBlocks) >= 512 {
-		blocksToMerge = tb.rawItemsBlocks
-		tb.rawItemsBlocks = nil
-		tb.rawItemsLastFlushTime = fasttime.UnixTimestamp()
-	}
-	tb.rawItemsLock.Unlock()
-
-	if blocksToMerge == nil {
-		// Fast path.
-		return err
+	if err := tb.rawItems.addItems(tb, items); err != nil {
+		return fmt.Errorf("cannot insert data into %q: %w", tb.path, err)
 	}
-
-	// Slow path: merge blocksToMerge.
-	tb.mergeRawItemsBlocks(blocksToMerge)
-	return err
+	return nil
 }
 
 // getParts appends parts snapshot to dst and returns it.
@@ -522,9 +579,25 @@ func (tb *Table) DebugFlush() {
 }
 
 func (tb *Table) flushRawItems(isFinal bool) {
+	tb.rawItems.flush(tb, isFinal)
+}
+
+func (riss *rawItemsShards) flush(tb *Table, isFinal bool) {
 	tb.rawItemsPendingFlushesWG.Add(1)
 	defer tb.rawItemsPendingFlushesWG.Done()
 
+	var wg sync.WaitGroup
+	wg.Add(len(riss.shards))
+	for i := range riss.shards {
+		go func(ris *rawItemsShard) {
+			ris.flush(tb, isFinal)
+			wg.Done()
+		}(&riss.shards[i])
+	}
+	wg.Wait()
+}
+
+func (ris *rawItemsShard) flush(tb *Table, isFinal bool) {
 	mustFlush := false
 	currentTime := fasttime.UnixTimestamp()
 	flushSeconds := int64(rawItemsFlushInterval.Seconds())
@@ -533,14 +606,14 @@ func (tb *Table) flushRawItems(isFinal bool) {
 	}
 	var blocksToMerge []*inmemoryBlock
 
-	tb.rawItemsLock.Lock()
-	if isFinal || currentTime-tb.rawItemsLastFlushTime > uint64(flushSeconds) {
+	ris.mu.Lock()
+	if isFinal || currentTime-ris.lastFlushTime > uint64(flushSeconds) {
 		mustFlush = true
-		blocksToMerge = tb.rawItemsBlocks
-		tb.rawItemsBlocks = nil
-		tb.rawItemsLastFlushTime = currentTime
+		blocksToMerge = ris.ibs
+		ris.ibs = make([]*inmemoryBlock, 0, maxBlocksPerShard)
+		ris.lastFlushTime = currentTime
 	}
-	tb.rawItemsLock.Unlock()
+	ris.mu.Unlock()
 
 	if mustFlush {
 		tb.mergeRawItemsBlocks(blocksToMerge)
diff --git a/lib/promscrape/discovery/kubernetes/api_watcher.go b/lib/promscrape/discovery/kubernetes/api_watcher.go
index fe9fb071d..92a9a6cd6 100644
--- a/lib/promscrape/discovery/kubernetes/api_watcher.go
+++ b/lib/promscrape/discovery/kubernetes/api_watcher.go
@@ -74,6 +74,11 @@ func newAPIWatcher(apiServer string, ac *promauth.Config, sdc *SDConfig, swcFunc
 
 func (aw *apiWatcher) mustStart() {
 	aw.gw.startWatchersForRole(aw.role, aw)
+	if aw.role == "endpoints" || aw.role == "endpointslices" {
+		// endpoints and endpointslices watchers query pod and service objects. So start watchers for these roles as well.
+		aw.gw.startWatchersForRole("pod", nil)
+		aw.gw.startWatchersForRole("service", nil)
+	}
 }
 
 func (aw *apiWatcher) mustStop() {
@@ -280,7 +285,6 @@ func (gw *groupWatcher) getObjectByRole(role, namespace, name string) object {
 
 func (gw *groupWatcher) getCachedObjectByRole(role, namespace, name string) object {
 	key := namespace + "/" + name
-	gw.startWatchersForRole(role, nil)
 	uws := gw.getURLWatchers()
 	for _, uw := range uws {
 		if uw.role != role {
@@ -309,9 +313,6 @@ func (gw *groupWatcher) refreshEndpointsLabels(namespace, key string) {
 }
 
 func (gw *groupWatcher) refreshObjectLabels(role, namespace, key string) {
-	// There is no need in starting url watcher for the given role,
-	// since there is no (namespace, key) object yet for this role.
-	// gw.startWatchersForRole(role, nil)
 	uws := gw.getURLWatchers()
 	for _, uw := range uws {
 		if uw.role != role {
@@ -470,6 +471,10 @@ func (uw *urlWatcher) subscribeAPIWatcher(aw *apiWatcher) {
 
 func (uw *urlWatcher) registerPendingAPIWatchers() {
 	uw.mu.Lock()
+	if len(uw.awsPending) == 0 {
+		uw.mu.Unlock()
+		return
+	}
 	awsPending := make([]*apiWatcher, 0, len(uw.awsPending))
 	for aw := range uw.awsPending {
 		awsPending = append(awsPending, aw)
diff --git a/lib/storage/partition.go b/lib/storage/partition.go
index 0196b916b..4bf6f833b 100644
--- a/lib/storage/partition.go
+++ b/lib/storage/partition.go
@@ -437,54 +437,49 @@ func (pt *partition) AddRows(rows []rawRow) {
 }
 
 type rawRowsShards struct {
-	lock     sync.Mutex
-	shardIdx int
+	shardIdx uint32
 
 	// Shards reduce lock contention when adding rows on multi-CPU systems.
 	shards []rawRowsShard
 }
 
-func (rrs *rawRowsShards) init() {
-	rrs.shards = make([]rawRowsShard, rawRowsShardsPerPartition)
+func (rrss *rawRowsShards) init() {
+	rrss.shards = make([]rawRowsShard, rawRowsShardsPerPartition)
 }
 
-func (rrs *rawRowsShards) addRows(pt *partition, rows []rawRow) {
-	rrs.lock.Lock()
-	rrs.shardIdx++
-	if rrs.shardIdx >= len(rrs.shards) {
-		rrs.shardIdx = 0
-	}
-	shard := &rrs.shards[rrs.shardIdx]
-	rrs.lock.Unlock()
-
+func (rrss *rawRowsShards) addRows(pt *partition, rows []rawRow) {
+	n := atomic.AddUint32(&rrss.shardIdx, 1)
+	shards := rrss.shards
+	idx := n % uint32(len(shards))
+	shard := &shards[idx]
 	shard.addRows(pt, rows)
 }
 
-func (rrs *rawRowsShards) Len() int {
+func (rrss *rawRowsShards) Len() int {
 	n := 0
-	for i := range rrs.shards[:] {
-		n += rrs.shards[i].Len()
+	for i := range rrss.shards[:] {
+		n += rrss.shards[i].Len()
 	}
 	return n
 }
 
 type rawRowsShard struct {
-	lock          sync.Mutex
+	mu            sync.Mutex
 	rows          []rawRow
 	lastFlushTime uint64
 }
 
 func (rrs *rawRowsShard) Len() int {
-	rrs.lock.Lock()
+	rrs.mu.Lock()
 	n := len(rrs.rows)
-	rrs.lock.Unlock()
+	rrs.mu.Unlock()
 	return n
 }
 
 func (rrs *rawRowsShard) addRows(pt *partition, rows []rawRow) {
 	var rrss []*rawRows
 
-	rrs.lock.Lock()
+	rrs.mu.Lock()
 	if cap(rrs.rows) == 0 {
 		rrs.rows = getRawRowsMaxSize().rows
 	}
@@ -506,7 +501,7 @@ func (rrs *rawRowsShard) addRows(pt *partition, rows []rawRow) {
 		rrss = append(rrss, rr)
 		rrs.lastFlushTime = fasttime.UnixTimestamp()
 	}
-	rrs.lock.Unlock()
+	rrs.mu.Unlock()
 
 	for _, rr := range rrss {
 		pt.addRowsPart(rr.rows)
@@ -752,10 +747,16 @@ func (pt *partition) flushRawRows(isFinal bool) {
 	pt.rawRows.flush(pt, isFinal)
 }
 
-func (rrs *rawRowsShards) flush(pt *partition, isFinal bool) {
-	for i := range rrs.shards[:] {
-		rrs.shards[i].flush(pt, isFinal)
+func (rrss *rawRowsShards) flush(pt *partition, isFinal bool) {
+	var wg sync.WaitGroup
+	wg.Add(len(rrss.shards))
+	for i := range rrss.shards {
+		go func(rrs *rawRowsShard) {
+			rrs.flush(pt, isFinal)
+			wg.Done()
+		}(&rrss.shards[i])
 	}
+	wg.Wait()
 }
 
 func (rrs *rawRowsShard) flush(pt *partition, isFinal bool) {
@@ -766,12 +767,12 @@ func (rrs *rawRowsShard) flush(pt *partition, isFinal bool) {
 		flushSeconds = 1
 	}
 
-	rrs.lock.Lock()
+	rrs.mu.Lock()
 	if isFinal || currentTime-rrs.lastFlushTime > uint64(flushSeconds) {
 		rr = getRawRowsMaxSize()
 		rrs.rows, rr.rows = rr.rows, rrs.rows
 	}
-	rrs.lock.Unlock()
+	rrs.mu.Unlock()
 
 	if rr != nil {
 		pt.addRowsPart(rr.rows)
@@ -1363,7 +1364,7 @@ func (pt *partition) removeStaleParts() {
 	pt.snapshotLock.RLock()
 	var removeWG sync.WaitGroup
 	for pw := range m {
-		logger.Infof("removing part %q, since its data is out of the configured retention (%d secs)", pw.p.path, retentionDeadline/1000)
+		logger.Infof("removing part %q, since its data is out of the configured retention (%d secs)", pw.p.path, pt.retentionMsecs/1000)
 		removeWG.Add(1)
 		fs.MustRemoveAllWithDoneCallback(pw.p.path, removeWG.Done)
 	}


Test patch:
diff --git a/app/vmalert/alerting_test.go b/app/vmalert/alerting_test.go
index e5f10cc31..ea5bb51dc 100644
--- a/app/vmalert/alerting_test.go
+++ b/app/vmalert/alerting_test.go
@@ -294,11 +294,12 @@ func TestAlertingRule_Exec(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.rule.Name, func(t *testing.T) {
 			fq := &fakeQuerier{}
+			tc.rule.q = fq
 			tc.rule.GroupID = fakeGroup.ID()
 			for _, step := range tc.steps {
 				fq.reset()
 				fq.add(step...)
-				if _, err := tc.rule.Exec(context.TODO(), fq, false); err != nil {
+				if _, err := tc.rule.Exec(context.TODO(), false); err != nil {
 					t.Fatalf("unexpected err: %s", err)
 				}
 				// artificial delay between applying steps
@@ -410,6 +411,7 @@ func TestAlertingRule_Restore(t *testing.T) {
 		t.Run(tc.rule.Name, func(t *testing.T) {
 			fq := &fakeQuerier{}
 			tc.rule.GroupID = fakeGroup.ID()
+			tc.rule.q = fq
 			fq.add(tc.metrics...)
 			if err := tc.rule.Restore(context.TODO(), fq, time.Hour, nil); err != nil {
 				t.Fatalf("unexpected err: %s", err)
@@ -437,17 +439,18 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
 	fq := &fakeQuerier{}
 	ar := newTestAlertingRule("test", 0)
 	ar.Labels = map[string]string{"job": "test"}
+	ar.q = fq
 
 	// successful attempt
 	fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
-	_, err := ar.Exec(context.TODO(), fq, false)
+	_, err := ar.Exec(context.TODO(), false)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// label `job` will collide with rule extra label and will make both time series equal
 	fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "baz"))
-	_, err = ar.Exec(context.TODO(), fq, false)
+	_, err = ar.Exec(context.TODO(), false)
 	if !errors.Is(err, errDuplicate) {
 		t.Fatalf("expected to have %s error; got %s", errDuplicate, err)
 	}
@@ -456,7 +459,7 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
 
 	expErr := "connection reset by peer"
 	fq.setErr(errors.New(expErr))
-	_, err = ar.Exec(context.TODO(), fq, false)
+	_, err = ar.Exec(context.TODO(), false)
 	if err == nil {
 		t.Fatalf("expected to get err; got nil")
 	}
@@ -544,8 +547,9 @@ func TestAlertingRule_Template(t *testing.T) {
 		t.Run(tc.rule.Name, func(t *testing.T) {
 			fq := &fakeQuerier{}
 			tc.rule.GroupID = fakeGroup.ID()
+			tc.rule.q = fq
 			fq.add(tc.metrics...)
-			if _, err := tc.rule.Exec(context.TODO(), fq, false); err != nil {
+			if _, err := tc.rule.Exec(context.TODO(), false); err != nil {
 				t.Fatalf("unexpected err: %s", err)
 			}
 			for hash, expAlert := range tc.expAlerts {
diff --git a/app/vmalert/datasource/vm_test.go b/app/vmalert/datasource/vm_test.go
index 2c6148133..9ed26ef74 100644
--- a/app/vmalert/datasource/vm_test.go
+++ b/app/vmalert/datasource/vm_test.go
@@ -70,25 +70,25 @@ func TestVMSelectQuery(t *testing.T) {
 	srv := httptest.NewServer(mux)
 	defer srv.Close()
 	am := NewVMStorage(srv.URL, basicAuthName, basicAuthPass, time.Minute, 0, false, srv.Client())
-	if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
+	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected connection error got nil")
 	}
-	if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
+	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected invalid response status error got nil")
 	}
-	if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
+	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected response body error got nil")
 	}
-	if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
+	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected error status got nil")
 	}
-	if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
+	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected unknown status got nil")
 	}
-	if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
+	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected non-vector resultType error  got nil")
 	}
-	m, err := am.Query(ctx, query, NewPrometheusType())
+	m, err := am.Query(ctx, query)
 	if err != nil {
 		t.Fatalf("unexpected %s", err)
 	}
@@ -106,7 +106,10 @@ func TestVMSelectQuery(t *testing.T) {
 		m[0].Labels[0].Name != expected.Labels[0].Name {
 		t.Fatalf("unexpected metric %+v want %+v", m[0], expected)
 	}
-	m, err = am.Query(ctx, queryRender, NewGraphiteType())
+
+	dst := NewGraphiteType()
+	q := am.BuildWithParams(QuerierParams{&dst})
+	m, err = q.Query(ctx, queryRender)
 	if err != nil {
 		t.Fatalf("unexpected %s", err)
 	}
diff --git a/app/vmalert/group_test.go b/app/vmalert/group_test.go
index 093817abf..7f1103278 100644
--- a/app/vmalert/group_test.go
+++ b/app/vmalert/group_test.go
@@ -6,6 +6,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 )
@@ -105,20 +107,32 @@ func TestUpdateWith(t *testing.T) {
 				{Record: "foo5"},
 			},
 		},
+		{
+			"update datasource type",
+			[]config.Rule{
+				{Alert: "foo1", Type: datasource.NewPrometheusType()},
+				{Alert: "foo3", Type: datasource.NewGraphiteType()},
+			},
+			[]config.Rule{
+				{Alert: "foo1", Type: datasource.NewGraphiteType()},
+				{Alert: "foo10", Type: datasource.NewPrometheusType()},
+			},
+		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			g := &Group{Name: "test"}
+			qb := &fakeQuerier{}
 			for _, r := range tc.currentRules {
 				r.ID = config.HashRule(r)
-				g.Rules = append(g.Rules, g.newRule(r))
+				g.Rules = append(g.Rules, g.newRule(qb, r))
 			}
 
 			ng := &Group{Name: "test"}
 			for _, r := range tc.newRules {
 				r.ID = config.HashRule(r)
-				ng.Rules = append(ng.Rules, ng.newRule(r))
+				ng.Rules = append(ng.Rules, ng.newRule(qb, r))
 			}
 
 			err := g.updateWith(ng)
@@ -156,11 +170,11 @@ func TestGroupStart(t *testing.T) {
 		t.Fatalf("failed to parse rules: %s", err)
 	}
 	const evalInterval = time.Millisecond
-	g := newGroup(groups[0], evalInterval, map[string]string{"cluster": "east-1"})
-	g.Concurrency = 2
-
-	fn := &fakeNotifier{}
 	fs := &fakeQuerier{}
+	fn := &fakeNotifier{}
+
+	g := newGroup(groups[0], fs, evalInterval, map[string]string{"cluster": "east-1"})
+	g.Concurrency = 2
 
 	const inst1, inst2, job = "foo", "bar", "baz"
 	m1 := metricWithLabels(t, "instance", inst1, "job", job)
@@ -195,7 +209,7 @@ func TestGroupStart(t *testing.T) {
 	fs.add(m1)
 	fs.add(m2)
 	go func() {
-		g.start(context.Background(), fs, []notifier.Notifier{fn}, nil)
+		g.start(context.Background(), []notifier.Notifier{fn}, nil)
 		close(finished)
 	}()
 
diff --git a/app/vmalert/helpers_test.go b/app/vmalert/helpers_test.go
index 3f879d466..9c31222df 100644
--- a/app/vmalert/helpers_test.go
+++ b/app/vmalert/helpers_test.go
@@ -38,7 +38,11 @@ func (fq *fakeQuerier) add(metrics ...datasource.Metric) {
 	fq.Unlock()
 }
 
-func (fq *fakeQuerier) Query(_ context.Context, _ string, _ datasource.Type) ([]datasource.Metric, error) {
+func (fq *fakeQuerier) BuildWithParams(_ datasource.QuerierParams) datasource.Querier {
+	return fq
+}
+
+func (fq *fakeQuerier) Query(_ context.Context, _ string) ([]datasource.Metric, error) {
 	fq.Lock()
 	defer fq.Unlock()
 	if fq.err != nil {
@@ -160,6 +164,9 @@ func compareAlertingRules(t *testing.T, a, b *AlertingRule) error {
 	if !reflect.DeepEqual(a.Labels, b.Labels) {
 		return fmt.Errorf("expected to have labels %#v; got %#v", a.Labels, b.Labels)
 	}
+	if a.Type.String() != b.Type.String() {
+		return fmt.Errorf("expected to have Type %#v; got %#v", a.Type.String(), b.Type.String())
+	}
 	return nil
 }
 
diff --git a/app/vmalert/manager_test.go b/app/vmalert/manager_test.go
index 8762ef6cb..984505a3a 100644
--- a/app/vmalert/manager_test.go
+++ b/app/vmalert/manager_test.go
@@ -37,9 +37,9 @@ func TestManagerEmptyRulesDir(t *testing.T) {
 // Should be executed with -race flag
 func TestManagerUpdateConcurrent(t *testing.T) {
 	m := &manager{
-		groups:    make(map[uint64]*Group),
-		querier:   &fakeQuerier{},
-		notifiers: []notifier.Notifier{&fakeNotifier{}},
+		groups:         make(map[uint64]*Group),
+		querierBuilder: &fakeQuerier{},
+		notifiers:      []notifier.Notifier{&fakeNotifier{}},
 	}
 	paths := []string{
 		"config/testdata/dir/rules0-good.rules",
@@ -242,7 +242,7 @@ func TestManagerUpdate(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.TODO())
-			m := &manager{groups: make(map[uint64]*Group), querier: &fakeQuerier{}}
+			m := &manager{groups: make(map[uint64]*Group), querierBuilder: &fakeQuerier{}}
 			path := []string{tc.initPath}
 			if err := m.update(ctx, path, true, true, false); err != nil {
 				t.Fatalf("failed to complete initial rules update: %s", err)
diff --git a/app/vmalert/recording_test.go b/app/vmalert/recording_test.go
index 1ab35295c..80877b6ff 100644
--- a/app/vmalert/recording_test.go
+++ b/app/vmalert/recording_test.go
@@ -76,7 +76,8 @@ func TestRecoridngRule_ToTimeSeries(t *testing.T) {
 		t.Run(tc.rule.Name, func(t *testing.T) {
 			fq := &fakeQuerier{}
 			fq.add(tc.metrics...)
-			tss, err := tc.rule.Exec(context.TODO(), fq, true)
+			tc.rule.q = fq
+			tss, err := tc.rule.Exec(context.TODO(), true)
 			if err != nil {
 				t.Fatalf("unexpected Exec err: %s", err)
 			}
@@ -95,8 +96,8 @@ func TestRecoridngRule_ToTimeSeriesNegative(t *testing.T) {
 	fq := &fakeQuerier{}
 	expErr := "connection reset by peer"
 	fq.setErr(errors.New(expErr))
-
-	_, err := rr.Exec(context.TODO(), fq, true)
+	rr.q = fq
+	_, err := rr.Exec(context.TODO(), true)
 	if err == nil {
 		t.Fatalf("expected to get err; got nil")
 	}
@@ -111,7 +112,7 @@ func TestRecoridngRule_ToTimeSeriesNegative(t *testing.T) {
 	fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "foo"))
 	fq.add(metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "bar"))
 
-	_, err = rr.Exec(context.TODO(), fq, true)
+	_, err = rr.Exec(context.TODO(), true)
 	if err == nil {
 		t.Fatalf("expected to get err; got nil")
 	}