Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1095 (2021-02-26 14:58:14)

adds enforced tag filters into cache key


Patch:
diff --git a/app/vmselect/promql/rollup_result_cache.go b/app/vmselect/promql/rollup_result_cache.go
index 8ef54398e..e518324f6 100644
--- a/app/vmselect/promql/rollup_result_cache.go
+++ b/app/vmselect/promql/rollup_result_cache.go
@@ -178,7 +178,7 @@ func (rrc *rollupResultCache) Get(ec *EvalConfig, expr metricsql.Expr, window in
 	bb := bbPool.Get()
 	defer bbPool.Put(bb)
 
-	bb.B = marshalRollupResultCacheKey(bb.B[:0], expr, window, ec.Step)
+	bb.B = marshalRollupResultCacheKey(bb.B[:0], expr, window, ec.Step, ec.EnforcedTagFilters)
 	metainfoBuf := rrc.c.Get(nil, bb.B)
 	if len(metainfoBuf) == 0 {
 		return nil, ec.Start
@@ -198,7 +198,7 @@ func (rrc *rollupResultCache) Get(ec *EvalConfig, expr metricsql.Expr, window in
 	if len(compressedResultBuf.B) == 0 {
 		mi.RemoveKey(key)
 		metainfoBuf = mi.Marshal(metainfoBuf[:0])
-		bb.B = marshalRollupResultCacheKey(bb.B[:0], expr, window, ec.Step)
+		bb.B = marshalRollupResultCacheKey(bb.B[:0], expr, window, ec.Step, ec.EnforcedTagFilters)
 		rrc.c.Set(bb.B, metainfoBuf)
 		return nil, ec.Start
 	}
@@ -301,7 +301,7 @@ func (rrc *rollupResultCache) Put(ec *EvalConfig, expr metricsql.Expr, window in
 	bb.B = key.Marshal(bb.B[:0])
 	rrc.c.SetBig(bb.B, compressedResultBuf.B)
 
-	bb.B = marshalRollupResultCacheKey(bb.B[:0], expr, window, ec.Step)
+	bb.B = marshalRollupResultCacheKey(bb.B[:0], expr, window, ec.Step, ec.EnforcedTagFilters)
 	metainfoBuf := rrc.c.Get(nil, bb.B)
 	var mi rollupResultCacheMetainfo
 	if len(metainfoBuf) > 0 {
@@ -331,11 +331,14 @@ var tooBigRollupResults = metrics.NewCounter("vm_too_big_rollup_results_total")
 // Increment this value every time the format of the cache changes.
 const rollupResultCacheVersion = 7
 
-func marshalRollupResultCacheKey(dst []byte, expr metricsql.Expr, window, step int64) []byte {
+func marshalRollupResultCacheKey(dst []byte, expr metricsql.Expr, window, step int64, filters []storage.TagFilter) []byte {
 	dst = append(dst, rollupResultCacheVersion)
 	dst = encoding.MarshalInt64(dst, window)
 	dst = encoding.MarshalInt64(dst, step)
 	dst = expr.AppendString(dst)
+	for _, f := range filters {
+		dst = f.Marshal(dst)
+	}
 	return dst
 }
 
diff --git a/app/vmselect/promql/transform.go b/app/vmselect/promql/transform.go
index a893ce2e4..714f3cf80 100644
--- a/app/vmselect/promql/transform.go
+++ b/app/vmselect/promql/transform.go
@@ -343,6 +343,11 @@ func transformBucketsLimit(tfa *transformFuncArg) ([]*timeseries, error) {
 	if limit <= 0 {
 		return nil, nil
 	}
+	if limit < 3 {
+		// Preserve the first and the last bucket for better accuracy,
+		// since these buckets are usually `[0...leMin]` and `(leMax ... +Inf]`
+		limit = 3
+	}
 	tss := vmrangeBucketsToLE(args[1])
 	if len(tss) == 0 {
 		return nil, nil
@@ -404,15 +409,18 @@ func transformBucketsLimit(tfa *transformFuncArg) ([]*timeseries, error) {
 			}
 		}
 		for len(leGroup) > limit {
+			// Preserve the first and the last bucket for better accuracy,
+			// since these buckets are usually `[0...leMin]` and `(leMax ... +Inf]`
 			xxMinIdx := 0
-			for i, xx := range leGroup {
+			for i, xx := range leGroup[1 : len(leGroup)-1] {
 				if xx.hits < leGroup[xxMinIdx].hits {
 					xxMinIdx = i
 				}
 			}
+			xxMinIdx++
 			// Merge the leGroup[xxMinIdx] bucket with the smallest adjacent bucket in order to preserve
 			// the maximum accuracy.
-			if xxMinIdx+1 == len(leGroup) || (xxMinIdx > 0 && leGroup[xxMinIdx-1].hits < leGroup[xxMinIdx+1].hits) {
+			if xxMinIdx > 1 && leGroup[xxMinIdx-1].hits < leGroup[xxMinIdx+1].hits {
 				xxMinIdx--
 			}
 			leGroup[xxMinIdx+1].hits += leGroup[xxMinIdx].hits
@@ -578,7 +586,6 @@ func transformHistogramShare(tfa *transformFuncArg) ([]*timeseries, error) {
 	m := groupLeTimeseries(tss)
 
 	// Calculate share for les
-
 	share := func(i int, les []float64, xss []leTimeseries) (q, lower, upper float64) {
 		leReq := les[i]
 		if math.IsNaN(leReq) || len(xss) == 0 {
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index b9ce762de..336422d41 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -15,12 +15,14 @@
 * FEATURE: vmagent: export `vm_promscrape_target_relabel_duration_seconds` metric, which can be used for monitoring the time spend on relabeling for discovered targets.
 * FEATURE: vmagent: optimize [relabeling](https://victoriametrics.github.io/vmagent.html#relabeling) performance for common cases.
 * FEATURE: add `increase_pure(m[d])` function to MetricsQL. It works the same as `increase(m[d])` except of various edge cases. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/962) for details.
+* FEATURE: increase accuracy for `buckets_limit(limit, buckets)` results for small `limit` values. See [MetricsQL docs](https://victoriametrics.github.io/MetricsQL.html) for details.
 
 
 * BUGFIX: vmagent: properly perform graceful shutdown on `SIGINT` and `SIGTERM` signals. The graceful shutdown has been broken in `v1.54.0`. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1065
 * BUGFIX: reduce the probability of `duplicate time series` errors when querying Kubernetes metrics.
 * BUGFIX: properly calculate `histogram_quantile()` over time series with only a single non-zero bucket with `{le="+Inf"}`. Previously `NaN` was returned, now the value for the last bucket before `{le="+Inf"}` is returned like Prometheus does.
 * BUGFIX: vmselect: do not cache partial query results on timeout when receiving data from `vmstorage` nodes. See https://github.com/VictoriaMetrics/VictoriaMetrics/pull/1085
+* BUGFIX: properly handle `stale NFS file handle` error.
 
 
 # [v1.54.1](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.54.1)
diff --git a/lib/fs/dir_remover.go b/lib/fs/dir_remover.go
index aefcd77eb..fae5701bb 100644
--- a/lib/fs/dir_remover.go
+++ b/lib/fs/dir_remover.go
@@ -13,7 +13,7 @@ import (
 
 func mustRemoveAll(path string, done func()) bool {
 	err := os.RemoveAll(path)
-	if err == nil {
+	if err == nil || isStaleNFSFileHandleError(err) {
 		// Make sure the parent directory doesn't contain references
 		// to the current directory.
 		mustSyncParentDirIfExists(path)
@@ -87,6 +87,11 @@ func dirRemover() {
 	}
 }
 
+func isStaleNFSFileHandleError(err error) bool {
+	errStr := err.Error()
+	return strings.Contains(errStr, "stale NFS file handle")
+}
+
 func isTemporaryNFSError(err error) bool {
 	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/61 for details.
 	errStr := err.Error()
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index b41bcc463..641adf8f1 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -6,12 +6,15 @@ import (
 	"io/ioutil"
 	"net/url"
 	"path/filepath"
+	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envtemplate"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
@@ -510,6 +513,8 @@ func getScrapeWorkConfig(sc *ScrapeConfig, baseDir string, globalCfg *GlobalConf
 		disableKeepAlive:     sc.DisableKeepAlive,
 		streamParse:          sc.StreamParse,
 		scrapeAlignInterval:  sc.ScrapeAlignInterval,
+
+		cache: newScrapeWorkCache(),
 	}
 	return swc, nil
 }
@@ -533,6 +538,57 @@ type scrapeWorkConfig struct {
 	disableKeepAlive     bool
 	streamParse          bool
 	scrapeAlignInterval  time.Duration
+
+	cache *scrapeWorkCache
+}
+
+type scrapeWorkCache struct {
+	mu              sync.Mutex
+	m               map[string]*scrapeWorkEntry
+	lastCleanupTime uint64
+}
+
+type scrapeWorkEntry struct {
+	sw             *ScrapeWork
+	lastAccessTime uint64
+}
+
+func newScrapeWorkCache() *scrapeWorkCache {
+	return &scrapeWorkCache{
+		m: make(map[string]*scrapeWorkEntry),
+	}
+}
+
+func (swc *scrapeWorkCache) Get(key string) *ScrapeWork {
+	currentTime := fasttime.UnixTimestamp()
+	swc.mu.Lock()
+	swe := swc.m[key]
+	if swe != nil {
+		swe.lastAccessTime = currentTime
+	}
+	swc.mu.Unlock()
+	if swe == nil {
+		return nil
+	}
+	return swe.sw
+}
+
+func (swc *scrapeWorkCache) Set(key string, sw *ScrapeWork) {
+	currentTime := fasttime.UnixTimestamp()
+	swc.mu.Lock()
+	swc.m[key] = &scrapeWorkEntry{
+		sw:             sw,
+		lastAccessTime: currentTime,
+	}
+	if currentTime > swc.lastCleanupTime+10*60 {
+		for k, swe := range swc.m {
+			if currentTime > swe.lastAccessTime+2*60 {
+				delete(swc.m, k)
+			}
+		}
+		swc.lastCleanupTime = currentTime
+	}
+	swc.mu.Unlock()
 }
 
 type targetLabelsGetter interface {
@@ -562,7 +618,7 @@ func appendScrapeWorkForTargetLabels(dst []*ScrapeWork, swc *scrapeWorkConfig, t
 		go func() {
 			for metaLabels := range workCh {
 				target := metaLabels["__address__"]
-				sw, err := getScrapeWork(swc, target, nil, metaLabels)
+				sw, err := swc.getScrapeWork(target, nil, metaLabels)
 				if err != nil {
 					err = fmt.Errorf("skipping %s target %q for job_name %q because of error: %w", discoveryType, target, swc.jobName, err)
 				}
@@ -643,7 +699,7 @@ func (stc *StaticConfig) appendScrapeWork(dst []*ScrapeWork, swc *scrapeWorkConf
 			logger.Errorf("`static_configs` target for `job_name` %q cannot be empty; skipping it", swc.jobName)
 			continue
 		}
-		sw, err := getScrapeWork(swc, target, stc.Labels, metaLabels)
+		sw, err := swc.getScrapeWork(target, stc.Labels, metaLabels)
 		if err != nil {
 			// Do not return this error, since other targets may be valid
 			logger.Errorf("error when parsing `static_configs` target %q for `job_name` %q: %s; skipping it", target, swc.jobName, err)
@@ -656,7 +712,42 @@ func (stc *StaticConfig) appendScrapeWork(dst []*ScrapeWork, swc *scrapeWorkConf
 	return dst
 }
 
-func getScrapeWork(swc *scrapeWorkConfig, target string, extraLabels, metaLabels map[string]string) (*ScrapeWork, error) {
+func (swc *scrapeWorkConfig) getScrapeWork(target string, extraLabels, metaLabels map[string]string) (*ScrapeWork, error) {
+	key := getScrapeWorkKey(extraLabels, metaLabels)
+	if sw := swc.cache.Get(key); sw != nil {
+		return sw, nil
+	}
+	sw, err := swc.getScrapeWorkReal(target, extraLabels, metaLabels)
+	if err != nil {
+		swc.cache.Set(key, sw)
+	}
+	return sw, err
+}
+
+func getScrapeWorkKey(extraLabels, metaLabels map[string]string) string {
+	var b []byte
+	b = appendSortedKeyValuePairs(b, extraLabels)
+	b = appendSortedKeyValuePairs(b, metaLabels)
+	return string(b)
+}
+
+func appendSortedKeyValuePairs(dst []byte, m map[string]string) []byte {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		dst = strconv.AppendQuote(dst, k)
+		dst = append(dst, ':')
+		dst = strconv.AppendQuote(dst, m[k])
+		dst = append(dst, ',')
+	}
+	dst = append(dst, '\n')
+	return dst
+}
+
+func (swc *scrapeWorkConfig) getScrapeWorkReal(target string, extraLabels, metaLabels map[string]string) (*ScrapeWork, error) {
 	labels := mergeLabels(swc.jobName, swc.scheme, target, swc.metricsPath, extraLabels, swc.externalLabels, metaLabels, swc.params)
 	var originalLabels []prompbmarshal.Label
 	if !*dropOriginalLabels {
diff --git a/lib/promscrape/discovery/kubernetes/api.go b/lib/promscrape/discovery/kubernetes/api.go
index 816d5e194..a11e528c7 100644
--- a/lib/promscrape/discovery/kubernetes/api.go
+++ b/lib/promscrape/discovery/kubernetes/api.go
@@ -16,6 +16,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
@@ -109,6 +110,7 @@ type WatchEvent struct {
 // object is any Kubernetes object.
 type object interface {
 	key() string
+	getTargetLabels(aw *apiWatcher) []map[string]string
 }
 
 // parseObjectFunc must parse object from the given data.
@@ -142,7 +144,7 @@ type apiWatcher struct {
 
 	// The last time the apiWatcher was queried for cached objects.
 	// It is used for stopping unused watchers.
-	lastAccessTime time.Time
+	lastAccessTime uint64
 }
 
 func newAPIWatcher(client *http.Client, apiServer, authorization string, namespaces []string, selectors []Selector) *apiWatcher {
@@ -155,10 +157,29 @@ func newAPIWatcher(client *http.Client, apiServer, authorization string, namespa
 
 		watchersByURL: make(map[string]*urlWatcher),
 
-		lastAccessTime: time.Now(),
+		lastAccessTime: fasttime.UnixTimestamp(),
 	}
 }
 
+// getLabelsForRole returns all the sets of labels for the given role.
+func (aw *apiWatcher) getLabelsForRole(role string) []map[string]string {
+	var ms []map[string]string
+	aw.mu.Lock()
+	for _, uw := range aw.watchersByURL {
+		if uw.role != role {
+			continue
+		}
+		uw.mu.Lock()
+		for _, labels := range uw.labelsByKey {
+			ms = append(ms, labels...)
+		}
+		uw.mu.Unlock()
+	}
+	aw.lastAccessTime = fasttime.UnixTimestamp()
+	aw.mu.Unlock()
+	return ms
+}
+
 // getObjectByRole returns an object with the given (namespace, name) key and the given role.
 func (aw *apiWatcher) getObjectByRole(role, namespace, name string) object {
 	if aw == nil {
@@ -179,31 +200,11 @@ func (aw *apiWatcher) getObjectByRole(role, namespace, name string) object {
 			break
 		}
 	}
-	aw.lastAccessTime = time.Now()
+	aw.lastAccessTime = fasttime.UnixTimestamp()
 	aw.mu.Unlock()
 	return o
 }
 
-// getObjectsByRole returns all the objects for the given role.
-func (aw *apiWatcher) getObjectsByRole(role string) []object {
-	aw.startWatchersForRole(role)
-	var os []object
-	aw.mu.Lock()
-	for _, uw := range aw.watchersByURL {
-		if uw.role != role {
-			continue
-		}
-		uw.mu.Lock()
-		for _, o := range uw.objectsByKey {
-			os = append(os, o)
-		}
-		uw.mu.Unlock()
-	}
-	aw.lastAccessTime = time.Now()
-	aw.mu.Unlock()
-	return os
-}
-
 func (aw *apiWatcher) startWatchersForRole(role string) {
 	parseObject, parseObjectList := getObjectParsersForRole(role)
 	paths := getAPIPaths(role, aw.namespaces, aw.selectors)
@@ -235,7 +236,7 @@ func (aw *apiWatcher) startWatcherForURL(role, apiURL string, parseObject parseO
 func (aw *apiWatcher) needStop() bool {
 	aw.mu.Lock()
 	defer aw.mu.Unlock()
-	return time.Since(aw.lastAccessTime) > 5*time.Minute
+	return fasttime.UnixTimestamp() > aw.lastAccessTime+5*60
 }
 
 // doRequest performs http request to the given requestURL.
@@ -258,11 +259,12 @@ type urlWatcher struct {
 	parseObject     parseObjectFunc
 	parseObjectList parseObjectListFunc
 
-	// mu protects objectsByKey
+	// mu protects objectsByKey and labelsByKey
 	mu sync.Mutex
 
 	// objectsByKey contains the latest state for objects obtained from apiURL
 	objectsByKey map[string]object
+	labelsByKey  map[string][]map[string]string
 
 	// the parent apiWatcher
 	aw *apiWatcher
@@ -277,6 +279,7 @@ func (aw *apiWatcher) newURLWatcher(role, apiURL string, parseObject parseObject
 		parseObjectList: parseObjectList,
 
 		objectsByKey: make(map[string]object),
+		labelsByKey:  make(map[string][]map[string]string),
 
 		aw: aw,
 	}
@@ -394,9 +397,14 @@ func (uw *urlWatcher) readObjectUpdateStream(r io.Reader) error {
 			uw.mu.Lock()
 			uw.objectsByKey[key] = o
 			uw.mu.Unlock()
+			labels := o.getTargetLabels(uw.aw)
+			uw.mu.Lock()
+			uw.labelsByKey[key] = labels
+			uw.mu.Unlock()
 		case "DELETED":
 			uw.mu.Lock()
 			delete(uw.objectsByKey, key)
+			delete(uw.labelsByKey, key)
 			uw.mu.Unlock()
 		default:
 			return fmt.Errorf("unexpected WatchEvent type %q for role %q", we.Type, uw.role)
diff --git a/lib/promscrape/discovery/kubernetes/endpoints.go b/lib/promscrape/discovery/kubernetes/endpoints.go
index 2b6881140..003cceca5 100644
--- a/lib/promscrape/discovery/kubernetes/endpoints.go
+++ b/lib/promscrape/discovery/kubernetes/endpoints.go
@@ -7,25 +7,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getEndpointsLabels returns labels for k8s endpoints obtained from the given cfg.
-func getEndpointsLabels(cfg *apiConfig) []map[string]string {
-	epss := getEndpoints(cfg)
-	var ms []map[string]string
-	for _, eps := range epss {
-		ms = eps.appendTargetLabels(ms, cfg.aw)
-	}
-	return ms
-}
-
-func getEndpoints(cfg *apiConfig) []*Endpoints {
-	os := cfg.aw.getObjectsByRole("endpoint")
-	epss := make([]*Endpoints, len(os))
-	for i, o := range os {
-		epss[i] = o.(*Endpoints)
-	}
-	return epss
-}
-
 func (eps *Endpoints) key() string {
 	return eps.Metadata.key()
 }
@@ -104,15 +85,16 @@ type EndpointPort struct {
 	Protocol    string
 }
 
-// appendTargetLabels appends labels for each endpoint in eps to ms and returns the result.
+// getTargetLabels returns labels for each endpoint in eps.
 //
 // See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#endpoints
-func (eps *Endpoints) appendTargetLabels(ms []map[string]string, aw *apiWatcher) []map[string]string {
+func (eps *Endpoints) getTargetLabels(aw *apiWatcher) []map[string]string {
 	var svc *Service
 	if o := aw.getObjectByRole("service", eps.Metadata.Namespace, eps.Metadata.Name); o != nil {
 		svc = o.(*Service)
 	}
 	podPortsSeen := make(map[*Pod][]int)
+	var ms []map[string]string
 	for _, ess := range eps.Subsets {
 		for _, epp := range ess.Ports {
 			ms = appendEndpointLabelsForAddresses(ms, aw, podPortsSeen, eps, ess.Addresses, epp, svc, "true")
diff --git a/lib/promscrape/discovery/kubernetes/endpointslices.go b/lib/promscrape/discovery/kubernetes/endpointslices.go
index 2246be105..cec87ce13 100644
--- a/lib/promscrape/discovery/kubernetes/endpointslices.go
+++ b/lib/promscrape/discovery/kubernetes/endpointslices.go
@@ -8,26 +8,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getEndpointSlicesLabels returns labels for k8s endpointSlices obtained from the given cfg.
-func getEndpointSlicesLabels(cfg *apiConfig) []map[string]string {
-	epss := getEndpointSlices(cfg)
-	var ms []map[string]string
-	for _, eps := range epss {
-		ms = eps.appendTargetLabels(ms, cfg.aw)
-	}
-	return ms
-}
-
-// getEndpointSlices retrieves endpointSlice with given apiConfig
-func getEndpointSlices(cfg *apiConfig) []*EndpointSlice {
-	os := cfg.aw.getObjectsByRole("endpointslices")
-	epss := make([]*EndpointSlice, len(os))
-	for i, o := range os {
-		epss[i] = o.(*EndpointSlice)
-	}
-	return epss
-}
-
 func (eps *EndpointSlice) key() string {
 	return eps.Metadata.key()
 }
@@ -52,14 +32,16 @@ func parseEndpointSlice(data []byte) (object, error) {
 	return &eps, nil
 }
 
-// appendTargetLabels injects labels for endPointSlice to slice map
-// follows TargetRef for enrich labels with pod and service metadata
-func (eps *EndpointSlice) appendTargetLabels(ms []map[string]string, aw *apiWatcher) []map[string]string {
+// getTargetLabels returns labels for eps.
+//
+// See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#endpointslices
+func (eps *EndpointSlice) getTargetLabels(aw *apiWatcher) []map[string]string {
 	var svc *Service
 	if o := aw.getObjectByRole("service", eps.Metadata.Namespace, eps.Metadata.Name); o != nil {
 		svc = o.(*Service)
 	}
 	podPortsSeen := make(map[*Pod][]int)
+	var ms []map[string]string
 	for _, ess := range eps.Endpoints {
 		var p *Pod
 		if o := aw.getObjectByRole("pod", ess.TargetRef.Namespace, ess.TargetRef.Name); o != nil {
diff --git a/lib/promscrape/discovery/kubernetes/ingress.go b/lib/promscrape/discovery/kubernetes/ingress.go
index f69c6673d..38aef79d3 100644
--- a/lib/promscrape/discovery/kubernetes/ingress.go
+++ b/lib/promscrape/discovery/kubernetes/ingress.go
@@ -5,25 +5,6 @@ import (
 	"fmt"
 )
 
-// getIngressesLabels returns labels for k8s ingresses obtained from the given cfg.
-func getIngressesLabels(cfg *apiConfig) []map[string]string {
-	igs := getIngresses(cfg)
-	var ms []map[string]string
-	for _, ig := range igs {
-		ms = ig.appendTargetLabels(ms)
-	}
-	return ms
-}
-
-func getIngresses(cfg *apiConfig) []*Ingress {
-	os := cfg.aw.getObjectsByRole("ingress")
-	igs := make([]*Ingress, len(os))
-	for i, o := range os {
-		igs[i] = o.(*Ingress)
-	}
-	return igs
-}
-
 func (ig *Ingress) key() string {
 	return ig.Metadata.key()
 }
@@ -101,16 +82,17 @@ type HTTPIngressPath struct {
 	Path string
 }
 
-// appendTargetLabels appends labels for Ingress ig to ms and returns the result.
+// getTargetLabels returns labels for ig.
 //
 // See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#ingress
-func (ig *Ingress) appendTargetLabels(ms []map[string]string) []map[string]string {
+func (ig *Ingress) getTargetLabels(aw *apiWatcher) []map[string]string {
 	tlsHosts := make(map[string]bool)
 	for _, tls := range ig.Spec.TLS {
 		for _, host := range tls.Hosts {
 			tlsHosts[host] = true
 		}
 	}
+	var ms []map[string]string
 	for _, r := range ig.Spec.Rules {
 		paths := getIngressRulePaths(r.HTTP.Paths)
 		scheme := "http"
diff --git a/lib/promscrape/discovery/kubernetes/kubernetes.go b/lib/promscrape/discovery/kubernetes/kubernetes.go
index b6c7a99c4..504fef616 100644
--- a/lib/promscrape/discovery/kubernetes/kubernetes.go
+++ b/lib/promscrape/discovery/kubernetes/kubernetes.go
@@ -44,18 +44,8 @@ func (sdc *SDConfig) GetLabels(baseDir string) ([]map[string]string, error) {
 		return nil, fmt.Errorf("cannot create API config: %w", err)
 	}
 	switch sdc.Role {
-	case "node":
-		return getNodesLabels(cfg), nil
-	case "pod":
-		return getPodsLabels(cfg), nil
-	case "service":
-		return getServicesLabels(cfg), nil
-	case "endpoints":
-		return getEndpointsLabels(cfg), nil
-	case "endpointslices":
-		return getEndpointSlicesLabels(cfg), nil
-	case "ingress":
-		return getIngressesLabels(cfg), nil
+	case "node", "pod", "service", "endpoints", "endpointslices", "ingress":
+		return cfg.aw.getLabelsForRole(sdc.Role), nil
 	default:
 		return nil, fmt.Errorf("unexpected `role`: %q; must be one of `node`, `pod`, `service`, `endpoints`, `endpointslices` or `ingress`; skipping it", sdc.Role)
 	}
diff --git a/lib/promscrape/discovery/kubernetes/node.go b/lib/promscrape/discovery/kubernetes/node.go
index 720a58f04..653a99e73 100644
--- a/lib/promscrape/discovery/kubernetes/node.go
+++ b/lib/promscrape/discovery/kubernetes/node.go
@@ -8,24 +8,6 @@ import (
 )
 
 // getNodesLabels returns labels for k8s nodes obtained from the given cfg
-func getNodesLabels(cfg *apiConfig) []map[string]string {
-	nodes := getNodes(cfg)
-	var ms []map[string]string
-	for _, n := range nodes {
-		ms = n.appendTargetLabels(ms)
-	}
-	return ms
-}
-
-func getNodes(cfg *apiConfig) []*Node {
-	os := cfg.aw.getObjectsByRole("node")
-	ns := make([]*Node, len(os))
-	for i, o := range os {
-		ns[i] = o.(*Node)
-	}
-	return ns
-}
-
 func (n *Node) key() string {
 	return n.Metadata.key()
 }
@@ -89,14 +71,14 @@ type NodeDaemonEndpoints struct {
 	KubeletEndpoint DaemonEndpoint
 }
 
-// appendTargetLabels appends labels for the given Node n to ms and returns the result.
+// getTargetLabels returs labels for the given n.
 //
 // See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#node
-func (n *Node) appendTargetLabels(ms []map[string]string) []map[string]string {
+func (n *Node) getTargetLabels(aw *apiWatcher) []map[string]string {
 	addr := getNodeAddr(n.Status.Addresses)
 	if len(addr) == 0 {
 		// Skip node without address
-		return ms
+		return nil
 	}
 	addr = discoveryutils.JoinHostPort(addr, n.Status.DaemonEndpoints.KubeletEndpoint.Port)
 	m := map[string]string{
@@ -114,8 +96,7 @@ func (n *Node) appendTargetLabels(ms []map[string]string) []map[string]string {
 		ln := discoveryutils.SanitizeLabelName(a.Type)
 		m["__meta_kubernetes_node_address_"+ln] = a.Address
 	}
-	ms = append(ms, m)
-	return ms
+	return []map[string]string{m}
 }
 
 func getNodeAddr(nas []NodeAddress) string {
diff --git a/lib/promscrape/discovery/kubernetes/pod.go b/lib/promscrape/discovery/kubernetes/pod.go
index cb2b00877..80864d25e 100644
--- a/lib/promscrape/discovery/kubernetes/pod.go
+++ b/lib/promscrape/discovery/kubernetes/pod.go
@@ -9,25 +9,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getPodsLabels returns labels for k8s pods obtained from the given cfg
-func getPodsLabels(cfg *apiConfig) []map[string]string {
-	pods := getPods(cfg)
-	var ms []map[string]string
-	for _, p := range pods {
-		ms = p.appendTargetLabels(ms)
-	}
-	return ms
-}
-
-func getPods(cfg *apiConfig) []*Pod {
-	os := cfg.aw.getObjectsByRole("pod")
-	ps := make([]*Pod, len(os))
-	for i, o := range os {
-		ps[i] = o.(*Pod)
-	}
-	return ps
-}
-
 func (p *Pod) key() string {
 	return p.Metadata.key()
 }
@@ -111,14 +92,15 @@ type PodCondition struct {
 	Status string
 }
 
-// appendTargetLabels appends labels for each port of the given Pod p to ms and returns the result.
+// getTargetLabels returns labels for each port of the given p.
 //
 // See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#pod
-func (p *Pod) appendTargetLabels(ms []map[string]string) []map[string]string {
+func (p *Pod) getTargetLabels(aw *apiWatcher) []map[string]string {
 	if len(p.Status.PodIP) == 0 {
 		// Skip pod without IP
-		return ms
+		return nil
 	}
+	var ms []map[string]string
 	ms = appendPodLabels(ms, p, p.Spec.Containers, "false")
 	ms = appendPodLabels(ms, p, p.Spec.InitContainers, "true")
 	return ms
diff --git a/lib/promscrape/discovery/kubernetes/service.go b/lib/promscrape/discovery/kubernetes/service.go
index c23569bab..c129816e6 100644
--- a/lib/promscrape/discovery/kubernetes/service.go
+++ b/lib/promscrape/discovery/kubernetes/service.go
@@ -7,27 +7,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getServicesLabels returns labels for k8s services obtained from the given cfg
-func getServicesLabels(cfg *apiConfig) []map[string]string {
-	svcs := getServices(cfg)
-	var ms []map[string]string
-	for _, svc := range svcs {
-		ms = svc.appendTargetLabels(ms)
-	}
-	return ms
-}
-
-func getServices(cfg *apiConfig) []*Service {
-	os := cfg.aw.getObjectsByRole("service")
-	svcs := make([]*Service, len(os))
-	for i, o := range os {
-		svcs[i] = o.(*Service)
-	}
-	return svcs
-}
-
-func (svc *Service) key() string {
-	return svc.Metadata.key()
+func (s *Service) key() string {
+	return s.Metadata.key()
 }
 
 func parseServiceList(data []byte) (map[string]object, ListMeta, error) {
@@ -36,18 +17,18 @@ func parseServiceList(data []byte) (map[string]object, ListMeta, error) {
 		return nil, sl.Metadata, fmt.Errorf("cannot unmarshal ServiceList from %q: %w", data, err)
 	}
 	objectsByKey := make(map[string]object)
-	for _, svc := range sl.Items {
-		objectsByKey[svc.key()] = svc
+	for _, s := range sl.Items {
+		objectsByKey[s.key()] = s
 	}
 	return objectsByKey, sl.Metadata, nil
 }
 
 func parseService(data []byte) (object, error) {
-	var svc Service
-	if err := json.Unmarshal(data, &svc); err != nil {
+	var s Service
+	if err := json.Unmarshal(data, &s); err != nil {
 		return nil, err
 	}
-	return &svc, nil
+	return &s, nil
 }
 
 // ServiceList is k8s service list.
@@ -85,11 +66,12 @@ type ServicePort struct {
 	Port     int
 }
 
-// appendTargetLabels appends labels for each port of the given Service s to ms and returns the result.
+// getTargetLabels returns labels for each port of the given s.
 //
 // See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#service
-func (s *Service) appendTargetLabels(ms []map[string]string) []map[string]string {
+func (s *Service) getTargetLabels(aw *apiWatcher) []map[string]string {
 	host := fmt.Sprintf("%s.%s.svc", s.Metadata.Name, s.Metadata.Namespace)
+	var ms []map[string]string
 	for _, sp := range s.Spec.Ports {
 		addr := discoveryutils.JoinHostPort(host, sp.Port)
 		m := map[string]string{


Test patch:
diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index 0bc3efc56..9a3664064 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -3337,14 +3337,14 @@ func TestExecSuccess(t *testing.T) {
 		)))`
 		r1 := netstorage.Result{
 			MetricName: metricNameExpected,
-			Values:     []float64{52, 52, 52, 52, 52, 52},
+			Values:     []float64{9, 9, 9, 9, 9, 9},
 			Timestamps: timestampsExpected,
 		}
 		r1.MetricName.MetricGroup = []byte("metric")
 		r1.MetricName.Tags = []storage.Tag{
 			{
 				Key:   []byte("le"),
-				Value: []byte("200"),
+				Value: []byte("10"),
 			},
 			{
 				Key:   []byte("x"),
@@ -3353,11 +3353,27 @@ func TestExecSuccess(t *testing.T) {
 		}
 		r2 := netstorage.Result{
 			MetricName: metricNameExpected,
-			Values:     []float64{100, 100, 100, 100, 100, 100},
+			Values:     []float64{98, 98, 98, 98, 98, 98},
 			Timestamps: timestampsExpected,
 		}
 		r2.MetricName.MetricGroup = []byte("metric")
 		r2.MetricName.Tags = []storage.Tag{
+			{
+				Key:   []byte("le"),
+				Value: []byte("300"),
+			},
+			{
+				Key:   []byte("x"),
+				Value: []byte("y"),
+			},
+		}
+		r3 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{100, 100, 100, 100, 100, 100},
+			Timestamps: timestampsExpected,
+		}
+		r3.MetricName.MetricGroup = []byte("metric")
+		r3.MetricName.Tags = []storage.Tag{
 			{
 				Key:   []byte("le"),
 				Value: []byte("inf"),
@@ -3367,7 +3383,7 @@ func TestExecSuccess(t *testing.T) {
 				Value: []byte("y"),
 			},
 		}
-		resultExpected := []netstorage.Result{r1, r2}
+		resultExpected := []netstorage.Result{r1, r2, r3}
 		f(q, resultExpected)
 	})
 	t.Run(`prometheus_buckets(missing-vmrange)`, func(t *testing.T) {
@@ -4175,11 +4191,11 @@ func TestExecSuccess(t *testing.T) {
 	})
 	t.Run(`sum(histogram_over_time) by (vmrange)`, func(t *testing.T) {
 		t.Parallel()
-		q := `sort_desc(
+		q := `sort_by_label(
 			buckets_limit(
 				3,
 				sum(histogram_over_time(alias(label_set(rand(0)*1.3+1.1, "foo", "bar"), "xxx")[200s:5s])) by (vmrange)
-			)
+			), "le"
 		)`
 		r1 := netstorage.Result{
 			MetricName: metricNameExpected,
@@ -4194,24 +4210,24 @@ func TestExecSuccess(t *testing.T) {
 		}
 		r2 := netstorage.Result{
 			MetricName: metricNameExpected,
-			Values:     []float64{24, 22, 26, 25, 24, 24},
+			Values:     []float64{0, 0, 0, 0, 0, 0},
 			Timestamps: timestampsExpected,
 		}
 		r2.MetricName.Tags = []storage.Tag{
 			{
 				Key:   []byte("le"),
-				Value: []byte("1.896e+00"),
+				Value: []byte("1.000e+00"),
 			},
 		}
 		r3 := netstorage.Result{
 			MetricName: metricNameExpected,
-			Values:     []float64{11, 12, 11, 7, 11, 13},
+			Values:     []float64{40, 40, 40, 40, 40, 40},
 			Timestamps: timestampsExpected,
 		}
 		r3.MetricName.Tags = []storage.Tag{
 			{
 				Key:   []byte("le"),
-				Value: []byte("1.468e+00"),
+				Value: []byte("2.448e+00"),
 			},
 		}
 		resultExpected := []netstorage.Result{r1, r2, r3}
diff --git a/lib/promscrape/discovery/kubernetes/endpoints_test.go b/lib/promscrape/discovery/kubernetes/endpoints_test.go
index bd00ebdcf..05a74183b 100644
--- a/lib/promscrape/discovery/kubernetes/endpoints_test.go
+++ b/lib/promscrape/discovery/kubernetes/endpoints_test.go
@@ -88,9 +88,7 @@ func TestParseEndpointsListSuccess(t *testing.T) {
 		t.Fatalf("unexpected resource version; got %s; want %s", meta.ResourceVersion, expectedResourceVersion)
 	}
 
-	sortedLabelss := getSortedLabelss(objectsByKey, func(o object) []map[string]string {
-		return o.(*Endpoints).appendTargetLabels(nil, nil)
-	})
+	sortedLabelss := getSortedLabelss(objectsByKey)
 	expectedLabelss := [][]prompbmarshal.Label{
 		discoveryutils.GetSortedLabels(map[string]string{
 			"__address__": "172.17.0.2:8443",
diff --git a/lib/promscrape/discovery/kubernetes/endpointslices_test.go b/lib/promscrape/discovery/kubernetes/endpointslices_test.go
index a89eccf1b..d587fd93f 100644
--- a/lib/promscrape/discovery/kubernetes/endpointslices_test.go
+++ b/lib/promscrape/discovery/kubernetes/endpointslices_test.go
@@ -185,9 +185,7 @@ func TestParseEndpointSliceListSuccess(t *testing.T) {
 	if meta.ResourceVersion != expectedResourceVersion {
 		t.Fatalf("unexpected resource version; got %s; want %s", meta.ResourceVersion, expectedResourceVersion)
 	}
-	sortedLabelss := getSortedLabelss(objectsByKey, func(o object) []map[string]string {
-		return o.(*EndpointSlice).appendTargetLabels(nil, nil)
-	})
+	sortedLabelss := getSortedLabelss(objectsByKey)
 	expectedLabelss := [][]prompbmarshal.Label{
 		discoveryutils.GetSortedLabels(map[string]string{
 			"__address__": "172.18.0.2:6443",
diff --git a/lib/promscrape/discovery/kubernetes/ingress_test.go b/lib/promscrape/discovery/kubernetes/ingress_test.go
index 8007c6376..8046d8a24 100644
--- a/lib/promscrape/discovery/kubernetes/ingress_test.go
+++ b/lib/promscrape/discovery/kubernetes/ingress_test.go
@@ -79,9 +79,7 @@ func TestParseIngressListSuccess(t *testing.T) {
 	if meta.ResourceVersion != expectedResourceVersion {
 		t.Fatalf("unexpected resource version; got %s; want %s", meta.ResourceVersion, expectedResourceVersion)
 	}
-	sortedLabelss := getSortedLabelss(objectsByKey, func(o object) []map[string]string {
-		return o.(*Ingress).appendTargetLabels(nil)
-	})
+	sortedLabelss := getSortedLabelss(objectsByKey)
 	expectedLabelss := [][]prompbmarshal.Label{
 		discoveryutils.GetSortedLabels(map[string]string{
 			"__address__": "foobar",
diff --git a/lib/promscrape/discovery/kubernetes/node_test.go b/lib/promscrape/discovery/kubernetes/node_test.go
index deed59fc4..e63cbe4b4 100644
--- a/lib/promscrape/discovery/kubernetes/node_test.go
+++ b/lib/promscrape/discovery/kubernetes/node_test.go
@@ -235,9 +235,7 @@ func TestParseNodeListSuccess(t *testing.T) {
 	if meta.ResourceVersion != expectedResourceVersion {
 		t.Fatalf("unexpected resource version; got %s; want %s", meta.ResourceVersion, expectedResourceVersion)
 	}
-	sortedLabelss := getSortedLabelss(objectsByKey, func(o object) []map[string]string {
-		return o.(*Node).appendTargetLabels(nil)
-	})
+	sortedLabelss := getSortedLabelss(objectsByKey)
 	expectedLabelss := [][]prompbmarshal.Label{
 		discoveryutils.GetSortedLabels(map[string]string{
 			"instance":                    "m01",
@@ -283,10 +281,10 @@ func TestParseNodeListSuccess(t *testing.T) {
 	}
 }
 
-func getSortedLabelss(objectsByKey map[string]object, getLabelss func(o object) []map[string]string) [][]prompbmarshal.Label {
+func getSortedLabelss(objectsByKey map[string]object) [][]prompbmarshal.Label {
 	var result [][]prompbmarshal.Label
 	for _, o := range objectsByKey {
-		labelss := getLabelss(o)
+		labelss := o.getTargetLabels(nil)
 		for _, labels := range labelss {
 			result = append(result, discoveryutils.GetSortedLabels(labels))
 		}
diff --git a/lib/promscrape/discovery/kubernetes/pod_test.go b/lib/promscrape/discovery/kubernetes/pod_test.go
index 74c5456ca..204a8f102 100644
--- a/lib/promscrape/discovery/kubernetes/pod_test.go
+++ b/lib/promscrape/discovery/kubernetes/pod_test.go
@@ -236,9 +236,7 @@ func TestParsePodListSuccess(t *testing.T) {
 	if meta.ResourceVersion != expectedResourceVersion {
 		t.Fatalf("unexpected resource version; got %s; want %s", meta.ResourceVersion, expectedResourceVersion)
 	}
-	sortedLabelss := getSortedLabelss(objectsByKey, func(o object) []map[string]string {
-		return o.(*Pod).appendTargetLabels(nil)
-	})
+	sortedLabelss := getSortedLabelss(objectsByKey)
 	expectedLabelss := [][]prompbmarshal.Label{
 		discoveryutils.GetSortedLabels(map[string]string{
 			"__address__": "172.17.0.2:1234",
diff --git a/lib/promscrape/discovery/kubernetes/service_test.go b/lib/promscrape/discovery/kubernetes/service_test.go
index 4718fe6ed..56508e5f3 100644
--- a/lib/promscrape/discovery/kubernetes/service_test.go
+++ b/lib/promscrape/discovery/kubernetes/service_test.go
@@ -97,9 +97,7 @@ func TestParseServiceListSuccess(t *testing.T) {
 	if meta.ResourceVersion != expectedResourceVersion {
 		t.Fatalf("unexpected resource version; got %s; want %s", meta.ResourceVersion, expectedResourceVersion)
 	}
-	sortedLabelss := getSortedLabelss(objectsByKey, func(o object) []map[string]string {
-		return o.(*Service).appendTargetLabels(nil)
-	})
+	sortedLabelss := getSortedLabelss(objectsByKey)
 	expectedLabelss := [][]prompbmarshal.Label{
 		discoveryutils.GetSortedLabels(map[string]string{
 			"__address__":                             "kube-dns.kube-system.svc:53",