Delay hashing scrape targets

Instead of hashing scrape targets when they're received from service
discovery, wait until the allocator actually needs the hash. This lets
us avoid calculating hashes for targets filtered out by relabeling.

Author: swiatekm

cmd/otel-allocator/internal/allocation/allocator.go

@@ -73,20 +73,24 @@ func (a *allocator) SetFallbackStrategy(strategy Strategy) {
 // SetTargets accepts a list of targets that will be used to make
 // load balancing decisions. This method should be called when there are
 // new targets discovered or existing targets are shutdown.
-func (a *allocator) SetTargets(targets map[string]*target.Item) {
+func (a *allocator) SetTargets(targets []*target.Item) {
 	timer := prometheus.NewTimer(TimeToAssign.WithLabelValues("SetTargets", a.strategy.GetName()))
 	defer timer.ObserveDuration()
 
 	if a.filter != nil {
 		targets = a.filter.Apply(targets)
 	}
 	RecordTargetsKept(targets)
+	targetMap := make(map[string]*target.Item, len(targets))
+	for _, tg := range targets {
+		targetMap[tg.Hash()] = tg
+	}
 
 	a.m.Lock()
 	defer a.m.Unlock()
 
 	// Check for target changes
-	targetsDiff := diff.Maps(a.targetItems, targets)
+	targetsDiff := diff.Maps(a.targetItems, targetMap)
 	// If there are any additions or removals
 	if len(targetsDiff.Additions()) != 0 || len(targetsDiff.Removals()) != 0 {
 		a.handleTargets(targetsDiff)

cmd/otel-allocator/internal/allocation/allocator_test.go

@@ -174,10 +174,7 @@ func TestAllocationCollision(t *testing.T) {
 		firstTarget := target.NewItem("sample-name", "0.0.0.0:8000", firstLabels, "")
 		secondTarget := target.NewItem("sample-name", "0.0.0.0:8000", secondLabels, "")
 
-		targetList := map[string]*target.Item{
-			firstTarget.Hash():  firstTarget,
-			secondTarget.Hash(): secondTarget,
-		}
+		targetList := []*target.Item{firstTarget, secondTarget}
 
 		// test that targets and collectors are added properly
 		allocator.SetTargets(targetList)

cmd/otel-allocator/internal/allocation/least_weighted_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"slices"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -128,7 +129,7 @@ func TestCollectorBalanceWhenAddingAndRemovingAtRandom(t *testing.T) {
 	for index := range targets {
 		shouldDelete := rand.Intn(toDelete) //nolint:gosec
 		if counter < shouldDelete {
-			delete(targets, index)
+			targets = slices.Delete(targets, index, index)
 		}
 		counter++
 	}
@@ -144,9 +145,7 @@ func TestCollectorBalanceWhenAddingAndRemovingAtRandom(t *testing.T) {
 		assert.InDelta(t, i.NumTargets, count, math.Round(percent))
 	}
 	// adding targets at 'random'
-	for _, item := range MakeNNewTargets(13, 3, 100) {
-		targets[item.Hash()] = item
-	}
+	targets = append(targets, MakeNNewTargets(13, 3, 100)...)
 	s.SetTargets(targets)
 
 	targetItemLen = len(s.TargetItems())

cmd/otel-allocator/internal/allocation/per_node_test.go

@@ -56,11 +56,11 @@ func TestAllocationPerNode(t *testing.T) {
 	thirdTarget := target.NewItem("sample-name", "0.0.0.0:8000", thirdLabels, "")
 	fourthTarget := target.NewItem("sample-name", "0.0.0.0:8000", fourthLabels, "")
 
-	targetList := map[string]*target.Item{
-		firstTarget.Hash():  firstTarget,
-		secondTarget.Hash(): secondTarget,
-		thirdTarget.Hash():  thirdTarget,
-		fourthTarget.Hash(): fourthTarget,
+	targetList := []*target.Item{
+		firstTarget,
+		secondTarget,
+		thirdTarget,
+		fourthTarget,
 	}
 
 	// test that targets and collectors are added properly
@@ -74,16 +74,16 @@ func TestAllocationPerNode(t *testing.T) {
 	assert.Len(t, actualItems, expectedTargetLen)
 
 	// verify allocation to nodes
-	for targetHash, item := range targetList {
-		actualItem, found := actualItems[targetHash]
+	for _, item := range targetList {
+		actualItem, found := actualItems[item.Hash()]
 		// if third target, should be skipped
 		assert.True(t, found, "target with hash %s not found", item.Hash())
 
 		// only the first two targets should be allocated
 		itemsForCollector := s.GetTargetsForCollectorAndJob(actualItem.CollectorName, actualItem.JobName)
 
 		// first two should be assigned one to each collector; if third target, should not be assigned
-		if targetHash == thirdTarget.Hash() {
+		if item.Hash() == thirdTarget.Hash() {
 			assert.Len(t, itemsForCollector, 0)
 			continue
 		}
@@ -123,11 +123,11 @@ func TestAllocationPerNodeUsingFallback(t *testing.T) {
 	thirdTarget := target.NewItem("sample-name", "0.0.0.0:8000", thirdLabels, "")
 	fourthTarget := target.NewItem("sample-name", "0.0.0.0:8000", fourthLabels, "")
 
-	targetList := map[string]*target.Item{
-		firstTarget.Hash():  firstTarget,
-		secondTarget.Hash(): secondTarget,
-		thirdTarget.Hash():  thirdTarget,
-		fourthTarget.Hash(): fourthTarget,
+	targetList := []*target.Item{
+		firstTarget,
+		secondTarget,
+		thirdTarget,
+		fourthTarget,
 	}
 
 	// test that targets and collectors are added properly
@@ -141,8 +141,8 @@ func TestAllocationPerNodeUsingFallback(t *testing.T) {
 	assert.Len(t, actualItems, expectedTargetLen)
 
 	// verify allocation to nodes
-	for targetHash, item := range targetList {
-		actualItem, found := actualItems[targetHash]
+	for _, item := range targetList {
+		actualItem, found := actualItems[item.Hash()]
 
 		assert.True(t, found, "target with hash %s not found", item.Hash())
 
@@ -151,7 +151,7 @@ func TestAllocationPerNodeUsingFallback(t *testing.T) {
 		// first two should be assigned one to each collector; if third target, it should be assigned
 		// according to the fallback strategy which may assign it to the otherwise empty collector or
 		// one of the others, depending on the strategy and collector loop order
-		if targetHash == thirdTarget.Hash() {
+		if item.Hash() == thirdTarget.Hash() {
 			assert.Empty(t, item.GetNodeName())
 			assert.NotZero(t, len(itemsForCollector))
 			continue

cmd/otel-allocator/internal/allocation/strategy.go

@@ -50,7 +50,7 @@ var (
 type Option func(Allocator)
 
 type Filter interface {
-	Apply(map[string]*target.Item) map[string]*target.Item
+	Apply([]*target.Item) []*target.Item
 }
 
 func WithFilter(filter Filter) Option {
@@ -69,7 +69,7 @@ func WithFallbackStrategy(fallbackStrategy string) Option {
 	}
 }
 
-func RecordTargetsKept(targets map[string]*target.Item) {
+func RecordTargetsKept(targets []*target.Item) {
 	TargetsRemaining.Set(float64(len(targets)))
 }
 
@@ -90,7 +90,7 @@ func GetRegisteredAllocatorNames() []string {
 
 type Allocator interface {
 	SetCollectors(collectors map[string]*Collector)
-	SetTargets(targets map[string]*target.Item)
+	SetTargets(targets []*target.Item)
 	TargetItems() map[string]*target.Item
 	Collectors() map[string]*Collector
 	GetTargetsForCollectorAndJob(collector string, job string) []*target.Item

cmd/otel-allocator/internal/allocation/testutils.go

@@ -24,16 +24,16 @@ func colIndex(index, numCols int) int {
 	return index % numCols
 }
 
-func MakeNNewTargets(n int, numCollectors int, startingIndex int) map[string]*target.Item {
-	toReturn := map[string]*target.Item{}
+func MakeNNewTargets(n int, numCollectors int, startingIndex int) []*target.Item {
+	toReturn := []*target.Item{}
 	for i := startingIndex; i < n+startingIndex; i++ {
 		collector := fmt.Sprintf("collector-%d", colIndex(i, numCollectors))
 		label := labels.Labels{
 			{Name: "i", Value: strconv.Itoa(i)},
 			{Name: "total", Value: strconv.Itoa(n + startingIndex)},
 		}
 		newTarget := target.NewItem(fmt.Sprintf("test-job-%d", i), fmt.Sprintf("test-url-%d", i), label, collector)
-		toReturn[newTarget.Hash()] = newTarget
+		toReturn = append(toReturn, newTarget)
 	}
 	return toReturn
 }
@@ -51,16 +51,16 @@ func MakeNCollectors(n int, startingIndex int) map[string]*Collector {
 	return toReturn
 }
 
-func MakeNNewTargetsWithEmptyCollectors(n int, startingIndex int) map[string]*target.Item {
-	toReturn := map[string]*target.Item{}
+func MakeNNewTargetsWithEmptyCollectors(n int, startingIndex int) []*target.Item {
+	toReturn := []*target.Item{}
 	for i := startingIndex; i < n+startingIndex; i++ {
 		label := labels.Labels{
 			{Name: "i", Value: strconv.Itoa(i)},
 			{Name: "total", Value: strconv.Itoa(n + startingIndex)},
 			{Name: "__meta_kubernetes_pod_node_name", Value: "node-0"},
 		}
 		newTarget := target.NewItem(fmt.Sprintf("test-job-%d", i), fmt.Sprintf("test-url-%d", i), label, "")
-		toReturn[newTarget.Hash()] = newTarget
+		toReturn = append(toReturn, newTarget)
 	}
 	return toReturn
 }

cmd/otel-allocator/internal/prehook/prehook.go

@@ -11,7 +11,7 @@ import (
 )
 
 type Hook interface {
-	Apply(map[string]*target.Item) map[string]*target.Item
+	Apply([]*target.Item) []*target.Item
 	SetConfig(map[string][]*relabel.Config)
 	GetConfig() map[string][]*relabel.Config
 }

cmd/otel-allocator/internal/prehook/relabel.go

@@ -26,29 +26,34 @@ func newRelabelConfigTargetFilter(log logr.Logger) Hook {
 	}
 }
 
-func (tf *relabelConfigTargetFilter) Apply(targets map[string]*target.Item) map[string]*target.Item {
+func (tf *relabelConfigTargetFilter) Apply(targets []*target.Item) []*target.Item {
 	numTargets := len(targets)
 
 	// need to wait until relabelCfg is set
 	if len(tf.relabelCfg) == 0 {
 		return targets
 	}
 
+	var newTargets []*target.Item
+
 	// Note: jobNameKey != tItem.JobName (jobNameKey is hashed)
-	for jobNameKey, tItem := range targets {
-		var keepTarget bool
+	for _, tItem := range targets {
+		keepTarget := true
 		lset := tItem.Labels
 		for _, cfg := range tf.relabelCfg[tItem.JobName] {
 			lset, keepTarget = relabel.Process(lset, cfg)
 			if !keepTarget {
-				delete(targets, jobNameKey)
 				break // inner loop
 			}
 		}
+
+		if keepTarget {
+			newTargets = append(newTargets, tItem)
+		}
 	}
 
 	tf.log.V(2).Info("Filtering complete", "seen", numTargets, "kept", len(targets))
-	return targets
+	return newTargets
 }
 
 func (tf *relabelConfigTargetFilter) SetConfig(cfgs map[string][]*relabel.Config) {

cmd/otel-allocator/internal/prehook/relabel_test.go

@@ -167,9 +167,9 @@ func colIndex(index, numCols int) int {
 	return index % numCols
 }
 
-func makeNNewTargets(rCfgs []relabelConfigObj, n int, numCollectors int, startingIndex int) (map[string]*target.Item, int, map[string]*target.Item, map[string][]*relabel.Config) {
-	toReturn := map[string]*target.Item{}
-	expectedMap := make(map[string]*target.Item)
+func makeNNewTargets(rCfgs []relabelConfigObj, n int, numCollectors int, startingIndex int) ([]*target.Item, int, []*target.Item, map[string][]*relabel.Config) {
+	toReturn := []*target.Item{}
+	expected := []*target.Item{}
 	numItemsRemaining := n
 	relabelConfig := make(map[string][]*relabel.Config)
 	for i := startingIndex; i < n+startingIndex; i++ {
@@ -189,15 +189,14 @@ func makeNNewTargets(rCfgs []relabelConfigObj, n int, numCollectors int, startin
 
 		relabelConfig[jobName] = rCfgs[index].cfg
 
-		targetKey := newTarget.Hash()
 		if relabelConfigs[index].isDrop {
 			numItemsRemaining--
 		} else {
-			expectedMap[targetKey] = newTarget
+			expected = append(expected, newTarget)
 		}
-		toReturn[targetKey] = newTarget
+		toReturn = append(toReturn, newTarget)
 	}
-	return toReturn, numItemsRemaining, expectedMap, relabelConfig
+	return toReturn, numItemsRemaining, expected, relabelConfig
 }
 
 func TestApply(t *testing.T) {

cmd/otel-allocator/internal/server/mocks_test.go

@@ -17,7 +17,7 @@ type mockAllocator struct {
 }
 
 func (m *mockAllocator) SetCollectors(_ map[string]*allocation.Collector)               {}
-func (m *mockAllocator) SetTargets(_ map[string]*target.Item)                           {}
+func (m *mockAllocator) SetTargets(_ []*target.Item)                                    {}
 func (m *mockAllocator) Collectors() map[string]*allocation.Collector                   { return nil }
 func (m *mockAllocator) GetTargetsForCollectorAndJob(_ string, _ string) []*target.Item { return nil }
 func (m *mockAllocator) SetFilter(_ allocation.Filter)                                  {}

cmd/otel-allocator/internal/server/server_test.go

@@ -160,8 +160,12 @@ func TestServer_TargetsHandler(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			listenAddr := ":8080"
 			s := NewServer(logger, tt.args.allocator, listenAddr)
+			targets := []*target.Item{}
+			for _, item := range tt.args.cMap {
+				targets = append(targets, item)
+			}
 			tt.args.allocator.SetCollectors(map[string]*allocation.Collector{"test-collector": {Name: "test-collector"}})
-			tt.args.allocator.SetTargets(tt.args.cMap)
+			tt.args.allocator.SetTargets(targets)
 			request := httptest.NewRequest("GET", fmt.Sprintf("/jobs/%s/targets?collector_id=%s", tt.args.job, tt.args.collector), nil)
 			w := httptest.NewRecorder()
 

cmd/otel-allocator/internal/target/discovery.go

@@ -54,7 +54,7 @@ type Discoverer struct {
 	scrapeConfigsUpdater   scrapeConfigsUpdater
 	targetSets             map[string][]*targetgroup.Group
 	triggerReload          chan struct{}
-	processTargetsCallBack func(targets map[string]*Item)
+	processTargetsCallBack func(targets []*Item)
 	mtxTargets             sync.Mutex
 }
 
@@ -66,7 +66,7 @@ type scrapeConfigsUpdater interface {
 	UpdateScrapeConfigResponse(map[string]*promconfig.ScrapeConfig) error
 }
 
-func NewDiscoverer(log logr.Logger, manager *discovery.Manager, hook discoveryHook, scrapeConfigsUpdater scrapeConfigsUpdater, setTargets func(targets map[string]*Item)) *Discoverer {
+func NewDiscoverer(log logr.Logger, manager *discovery.Manager, hook discoveryHook, scrapeConfigsUpdater scrapeConfigsUpdater, setTargets func(targets []*Item)) *Discoverer {
 	return &Discoverer{
 		log:                    log,
 		manager:                manager,
@@ -163,19 +163,25 @@ func (m *Discoverer) reloader() {
 func (m *Discoverer) Reload() {
 	m.mtxScrape.Lock()
 	var wg sync.WaitGroup
-	targets := map[string]*Item{}
 	timer := prometheus.NewTimer(processTargetsDuration)
 	defer timer.ObserveDuration()
 
+	// count targets and preallocate
+	targetCount := 0
+	for _, groups := range m.targetSets {
+		for _, group := range groups {
+			targetCount += len(group.Targets)
+		}
+	}
+	targets := make([]*Item, 0, targetCount)
+
 	for jobName, groups := range m.targetSets {
 		wg.Add(1)
 		// Run the sync in parallel as these take a while and at high load can't catch up.
 		go func(jobName string, groups []*targetgroup.Group) {
 			processedTargets := m.processTargetGroups(jobName, groups)
 			m.mtxTargets.Lock()
-			for k, v := range processedTargets {
-				targets[k] = v
-			}
+			targets = append(targets, processedTargets...)
 			m.mtxTargets.Unlock()
 			wg.Done()
 		}(jobName, groups)
@@ -186,10 +192,17 @@ func (m *Discoverer) Reload() {
 }
 
 // processTargetGroups processes the target groups and returns a map of targets.
-func (m *Discoverer) processTargetGroups(jobName string, groups []*targetgroup.Group) map[string]*Item {
+func (m *Discoverer) processTargetGroups(jobName string, groups []*targetgroup.Group) []*Item {
 	builder := labels.NewBuilder(labels.Labels{})
 	timer := prometheus.NewTimer(processTargetGroupsDuration.WithLabelValues(jobName))
-	targets := map[string]*Item{}
+
+	// count targets and preallocate
+	targetCount := 0
+	for _, group := range groups {
+		targetCount += len(group.Targets)
+	}
+	targets := make([]*Item, 0, targetCount)
+
 	defer timer.ObserveDuration()
 	var count float64 = 0
 	for _, tg := range groups {
@@ -205,7 +218,7 @@ func (m *Discoverer) processTargetGroups(jobName string, groups []*targetgroup.G
 				builder.Set(string(ln), string(lv))
 			}
 			item := NewItem(jobName, string(t[model.AddressLabel]), builder.Labels(), "")
-			targets[item.Hash()] = item
+			targets = append(targets, item)
 		}
 	}
 	targetsDiscovered.WithLabelValues(jobName).Set(count)

cmd/otel-allocator/internal/target/discovery_test.go

@@ -63,7 +63,7 @@ func TestDiscovery(t *testing.T) {
 	require.NoError(t, err)
 	d := discovery.NewManager(ctx, gokitlog.NewNopLogger(), registry, sdMetrics)
 	results := make(chan []string)
-	manager := NewDiscoverer(ctrl.Log.WithName("test"), d, nil, scu, func(targets map[string]*Item) {
+	manager := NewDiscoverer(ctrl.Log.WithName("test"), d, nil, scu, func(targets []*Item) {
 		var result []string
 		for _, t := range targets {
 			result = append(result, t.TargetURL)