lib/promscrape: reduce latency for k8s GetLabels (#2454)

replaces internStringMap with sync.Map - it greatly reduces lock contention
concurently reload scrape work for api watcher - each object labels added by dedicated CPU

changes can be tested with following script https://gist.github.com/f41gh7/6f8f8d8719786aff1f18a85c23aebf70
This commit is contained in:
Nikolay 2022-04-20 16:09:40 +03:00 committed by Aliaksandr Valialkin
parent 9dbfd99777
commit 429848a67d
No known key found for this signature in database
GPG Key ID: A72BEC6CD3D0DED1
2 changed files with 44 additions and 18 deletions

View File

@ -9,6 +9,7 @@ import (
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil" "github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@ -1221,24 +1222,27 @@ func internLabelStrings(labels []prompbmarshal.Label) {
} }
func internString(s string) string { func internString(s string) string {
if sInterned, ok := internStringsMap.Load(s); ok {
return sInterned.(string)
}
isc := atomic.LoadUint64(&internStringCount)
if isc > 100e3 {
internStringsMapLock.Lock() internStringsMapLock.Lock()
defer internStringsMapLock.Unlock() internStringsMap = sync.Map{}
atomic.AddUint64(&internStringCount, ^(isc - 1))
if sInterned, ok := internStringsMap[s]; ok { internStringsMapLock.Unlock()
return sInterned
} }
// Make a new copy for s in order to remove references from possible bigger string s refers to. // Make a new copy for s in order to remove references from possible bigger string s refers to.
sCopy := string(append([]byte{}, s...)) sCopy := string(append([]byte{}, s...))
internStringsMap[sCopy] = sCopy internStringsMap.Store(sCopy, sCopy)
if len(internStringsMap) > 100e3 { atomic.AddUint64(&internStringCount, 1)
internStringsMap = make(map[string]string, 100e3)
}
return sCopy return sCopy
} }
var ( var (
internStringCount = uint64(0)
internStringsMapLock sync.Mutex internStringsMapLock sync.Mutex
internStringsMap = make(map[string]string, 100e3) internStringsMap = sync.Map{}
) )
func getParamsFromLabels(labels []prompbmarshal.Label, paramsOrig map[string][]string) map[string][]string { func getParamsFromLabels(labels []prompbmarshal.Label, paramsOrig map[string][]string) map[string][]string {

View File

@ -16,6 +16,7 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger" "github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth" "github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/metrics" "github.com/VictoriaMetrics/metrics"
@ -522,15 +523,36 @@ func (uw *urlWatcher) reloadScrapeWorksForAPIWatchersLocked(awsMap map[*apiWatch
for i := range aws { for i := range aws {
swosByKey[i] = make(map[string][]interface{}) swosByKey[i] = make(map[string][]interface{})
} }
// update swos concurrently,
// it must decrease reload time for high number of records at promscrape file
maxConcurrent := cgroup.AvailableCPUs() - 2
if maxConcurrent < 1 {
maxConcurrent = 1
}
limit := make(chan struct{}, maxConcurrent)
var (
mu sync.Mutex
wg sync.WaitGroup
)
for key, o := range uw.objectsByKey { for key, o := range uw.objectsByKey {
limit <- struct{}{}
wg.Add(1)
// update swsos for each target at separate CPU
go func(key string, o object) {
labels := o.getTargetLabels(uw.gw) labels := o.getTargetLabels(uw.gw)
for i, aw := range aws { for i, aw := range aws {
swos := aw.getScrapeWorkObjectsForLabels(labels) swos := aw.getScrapeWorkObjectsForLabels(labels)
if len(swos) > 0 { if len(swos) > 0 {
mu.Lock()
swosByKey[i][key] = swos swosByKey[i][key] = swos
mu.Unlock()
} }
} }
wg.Done()
<-limit
}(key, o)
} }
wg.Wait()
for i, aw := range aws { for i, aw := range aws {
aw.reloadScrapeWorks(uw, swosByKey[i]) aw.reloadScrapeWorks(uw, swosByKey[i])
} }