fix: same target from two different jobs missing in targetallocator (#4066)

* initial changes

* fix test

* remove print statements

* address feedback

* cleanup

* Fix linter warnings

* Fix more linter warnings

---------

Co-authored-by: Mikołaj Świątek <[email protected]>

Author: gracewehner

.chloggen/labelshash.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'bug_fix'
+
+# The name of the component, or a single word describing the area of concern, (e.g. collector, target allocator, auto-instrumentation, opamp, github action)
+component: 'target allocator'
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: 'Fixes an issue where the same target from two different jobs was being allocated for only one job'
+
+# One or more tracking issues related to the change
+issues: [4044]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

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

@@ -74,12 +74,12 @@ func TestDiscovery(t *testing.T) {
 	defer cancelFunc()
 
 	go func() {
-		err := d.Run()
-		assert.Error(t, err)
+		runErr := d.Run()
+		assert.Error(t, runErr)
 	}()
 	go func() {
-		err := manager.Run()
-		assert.NoError(t, err)
+		runErr := manager.Run()
+		assert.NoError(t, runErr)
 	}()
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -341,6 +341,106 @@ func TestDiscovery_ScrapeConfigHashing(t *testing.T) {
 	}
 }
 
+func TestDiscoveryTargetHashing(t *testing.T) {
+	tests := []struct {
+		description string
+		cfg         *promconfig.Config
+	}{
+		{
+			description: "same targets in two different jobs",
+			cfg: &promconfig.Config{
+				ScrapeConfigs: []*promconfig.ScrapeConfig{
+					{
+						JobName:         "prometheus",
+						HonorTimestamps: true,
+						ScrapeInterval:  model.Duration(30 * time.Second),
+						ScrapeProtocols: defaultScrapeProtocols,
+						ScrapeTimeout:   model.Duration(30 * time.Second),
+						MetricsPath:     "/metrics",
+						Scheme:          "http",
+						ServiceDiscoveryConfigs: discovery.Configs{
+							discovery.StaticConfig{
+								{
+									Targets: []model.LabelSet{
+										{"__address__": "prom.domain:9001"},
+										{"__address__": "prom.domain:9002"},
+										{"__address__": "prom.domain:9003"},
+									},
+								},
+							},
+						},
+					},
+					{
+						JobName:         "prometheus2",
+						HonorTimestamps: true,
+						ScrapeInterval:  model.Duration(30 * time.Second),
+						ScrapeProtocols: defaultScrapeProtocols,
+						ScrapeTimeout:   model.Duration(30 * time.Second),
+						MetricsPath:     "/metrics2",
+						Scheme:          "http",
+						ServiceDiscoveryConfigs: discovery.Configs{
+							discovery.StaticConfig{
+								{
+									Targets: []model.LabelSet{
+										{"__address__": "prom.domain:9001"},
+										{"__address__": "prom.domain:9002"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	scu := &mockScrapeConfigUpdater{}
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	registry := prometheus.NewRegistry()
+	sdMetrics, err := discovery.CreateAndRegisterSDMetrics(registry)
+	require.NoError(t, err)
+	d := discovery.NewManager(ctx, config.NopLogger, registry, sdMetrics)
+	results := make(chan []*Item)
+	manager := NewDiscoverer(ctrl.Log.WithName("test"), d, nil, scu, func(targets []*Item) {
+		var result []*Item
+		result = append(result, targets...)
+		results <- result
+	})
+
+	defer manager.Close()
+	defer cancelFunc()
+
+	go func() {
+		runErr := d.Run()
+		assert.Error(t, runErr)
+	}()
+	go func() {
+		runErr := manager.Run()
+		assert.NoError(t, runErr)
+	}()
+
+	for _, tt := range tests {
+		t.Run(tt.description, func(t *testing.T) {
+			assert.NoError(t, err)
+			assert.True(t, len(tt.cfg.ScrapeConfigs) > 0)
+			err = manager.ApplyConfig(allocatorWatcher.EventSourcePrometheusCR, tt.cfg.ScrapeConfigs)
+			assert.NoError(t, err)
+
+			gotTargets := <-results
+
+			// Verify that all targets have different hashes
+			targetHashes := make(map[ItemHash]bool)
+			for _, target := range gotTargets {
+				hash := target.Hash()
+				if _, exists := targetHashes[hash]; exists {
+					t.Errorf("Duplicate hash %d found for target %s (%s)", hash, target.TargetURL, target.JobName)
+				}
+				targetHashes[hash] = true
+			}
+			assert.Equal(t, len(gotTargets), len(targetHashes), "Number of unique hashes should match number of targets")
+		})
+	}
+}
+
 func TestDiscovery_NoConfig(t *testing.T) {
 	scu := &mockScrapeConfigUpdater{mockCfg: map[string]*promconfig.ScrapeConfig{}}
 	ctx, cancelFunc := context.WithCancel(context.Background())

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

@@ -4,6 +4,7 @@
 package target
 
 import (
+	"github.com/cespare/xxhash/v2"
 	"github.com/prometheus/prometheus/model/labels"
 )
 
@@ -33,7 +34,7 @@ type Item struct {
 
 func (t *Item) Hash() ItemHash {
 	if t.hash == 0 {
-		t.hash = ItemHash(t.Labels.Hash())
+		t.hash = ItemHash(LabelsHashWithJobName(t.Labels, t.JobName))
 	}
 	return t.hash
 }
@@ -64,3 +65,41 @@ func NewItem(jobName string, targetURL string, labels labels.Labels, collectorNa
 		CollectorName: collectorName,
 	}
 }
+
+// LabelsHashWithJobName computes a hash of the labels and the job name.
+// Same logic as Prometheus labels.Hash: https://github.com/prometheus/prometheus/blob/8fd46f74aa0155e4d5aa30654f9c02e564e03743/model/labels/labels.go#L72
+// but adds in the job name since this is not in the labelset from the discovery manager.
+// The scrape manager adds it later. Address is already included in the labels, so it is not needed here.
+func LabelsHashWithJobName(ls labels.Labels, jobName string) uint64 {
+	var sep byte = '\xff'
+	var seps = []byte{sep}
+
+	// Use xxhash.Sum64(b) for fast path as it's faster.
+	b := make([]byte, 0, 1024)
+
+	// Differs from Prometheus implementation by adding job name.
+	b = append(b, jobName...)
+	b = append(b, sep)
+
+	for i, v := range ls {
+		if len(b)+len(v.Name)+len(v.Value)+2 >= cap(b) {
+			// If labels entry is 1KB+ do not allocate whole entry.
+			h := xxhash.New()
+			_, _ = h.Write(b)
+			for _, v := range ls[i:] {
+				_, _ = h.WriteString(v.Name)
+				_, _ = h.Write(seps)
+				_, _ = h.WriteString(v.Value)
+				_, _ = h.Write(seps)
+			}
+			return h.Sum64()
+		}
+
+		b = append(b, v.Name...)
+		b = append(b, sep)
+		b = append(b, v.Value...)
+		b = append(b, sep)
+	}
+
+	return xxhash.Sum64(b)
+}