Weight-class configuration for LeastWeighted target allocation strategy

.chloggen/target-allocation-weight-classes.yaml

@@ -0,0 +1,19 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# 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: Add weight-based target allocation for the least-weighted strategy via pod annotations
+
+# One or more tracking issues related to the change
+issues: [3128]
+
+# (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: |
+  Targets can be assigned a weight (1-100) via the pod annotation `opentelemetry.io/target-allocation-weight`.
+  The least-weighted strategy uses these weights to balance load across collectors.
+  Targets without the annotation default to weight 1.

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

@@ -9,6 +9,7 @@ import (
 	"maps"
 	"runtime"
 	"slices"
+	"strconv"
 	"sync"
 	"time"
 
@@ -17,6 +18,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 
+	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/internal/config"
 	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/internal/diff"
 	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/internal/target"
 )
@@ -111,6 +113,18 @@ func (a *allocator) SetFallbackStrategy(strategy Strategy) {
 	a.strategy.SetFallbackStrategy(strategy)
 }
 
+// getTargetWeight returns the numeric weight for a target.
+// Resolution order: pod annotation meta label > default (1).
+func (*allocator) getTargetWeight(tg *target.Item) int {
+	// Check pod annotation (exposed as meta label by K8s SD)
+	if labelVal := tg.Labels.Get(config.WeightAnnotationMetaLabel); labelVal != "" {
+		if w, err := strconv.Atoi(labelVal); err == nil && w >= 1 && w <= 100 {
+			return w
+		}
+	}
+	return config.DefaultWeight
+}
+
 // 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.
@@ -258,6 +272,7 @@ func (a *allocator) addTargetToTargetItems(tg *target.Item) error {
 	tg.CollectorName = colOwner.Name
 	a.addCollectorTargetItemMapping(tg)
 	a.collectors[colOwner.Name].NumTargets++
+	a.collectors[colOwner.Name].WeightedLoad += a.getTargetWeight(tg)
 	a.collectors[colOwner.Name].TargetsPerJob[tg.JobName]++
 	a.targetsPerCollector.Record(context.Background(), int64(a.collectors[colOwner.String()].NumTargets), metric.WithAttributes(attribute.String("collector_name", colOwner.String()), attribute.String("strategy", a.strategy.GetName())))
 	return nil
@@ -274,6 +289,7 @@ func (a *allocator) unassignTargetItem(item *target.Item) {
 		return
 	}
 	c.NumTargets--
+	c.WeightedLoad -= a.getTargetWeight(item)
 	c.TargetsPerJob[item.JobName]--
 	if c.TargetsPerJob[item.JobName] == 0 {
 		delete(c.TargetsPerJob, item.JobName)

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

@@ -34,9 +34,9 @@ func (*leastWeightedStrategy) GetCollectorForTarget(collectors map[string]*Colle
 	jobName := item.JobName
 	for _, v := range collectors {
 		// If the initial collector is empty, set the initial collector to the first element of map
-		if col == nil || v.NumTargets < col.NumTargets {
+		if col == nil || v.WeightedLoad < col.WeightedLoad {
 			col = v
-		} else if v.NumTargets == col.NumTargets {
+		} else if v.WeightedLoad == col.WeightedLoad {
 			vPerJob := v.TargetsPerJob[jobName]
 			colPerJob := col.TargetsPerJob[jobName]
 			// Tiebreaker: prefer collector with fewer targets from this job

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

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
+	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/internal/config"
 	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/internal/target"
 )
 
@@ -203,6 +204,92 @@ func TestLeastWeightedJobDistribution(t *testing.T) {
 	}
 }
 
+// TestWeightedLoadBalancing verifies that heavy and light targets are distributed
+// so that WeightedLoad is balanced across collectors rather than just target count.
+func TestWeightedLoadBalancing(t *testing.T) {
+	s, _ := New("least-weighted", logger)
+
+	numCols := 3
+	cols := MakeNCollectors(numCols, 0)
+	s.SetCollectors(cols)
+
+	// Create 3 heavy targets (weight=10 each) and 30 light targets (weight=1 each)
+	// Total weight = 3*10 + 30*1 = 60, expect ~20 per collector
+	heavyTargets := MakeNTargetsWithWeight(3, "heavy-job", 0, config.WeightAnnotationMetaLabel, "10")
+	lightTargets := MakeNTargetsWithWeight(30, "light-job", 100, config.WeightAnnotationMetaLabel, "1")
+
+	allTargets := append(heavyTargets, lightTargets...)
+	s.SetTargets(allTargets)
+
+	collectors := s.Collectors()
+
+	// Verify heavy targets are spread across collectors (not all on one)
+	heavyPerCollector := map[string]int{}
+	for _, col := range collectors {
+		targets := s.GetTargetsForCollectorAndJob(col.Name, "heavy-job")
+		heavyPerCollector[col.Name] = len(targets)
+	}
+	t.Logf("Heavy targets per collector: %v", heavyPerCollector)
+	// Each collector should have exactly 1 heavy target (3 heavy / 3 collectors)
+	for colName, count := range heavyPerCollector {
+		assert.Equal(t, 1, count, "collector %s should have exactly 1 heavy target, got %d", colName, count)
+	}
+
+	// Verify WeightedLoad is balanced across collectors
+	var loads []int
+	for _, col := range collectors {
+		loads = append(loads, col.WeightedLoad)
+	}
+	t.Logf("WeightedLoad per collector: %v", loads)
+
+	// Expected total weight = 60, expected per collector = 20
+	expectedPerCollector := 60.0 / float64(numCols)
+	for _, col := range collectors {
+		assert.InDelta(t, expectedPerCollector, col.WeightedLoad, expectedPerCollector*0.5,
+			"collector %s WeightedLoad should be ~%.0f, got %d", col.Name, expectedPerCollector, col.WeightedLoad)
+	}
+}
+
+// TestWeightedLoadUnlabeledTargets verifies that targets without a weight annotation
+// get the default weight of 1, so WeightedLoad equals NumTargets.
+func TestWeightedLoadUnlabeledTargets(t *testing.T) {
+	s, _ := New("least-weighted", logger)
+
+	cols := MakeNCollectors(3, 0)
+	s.SetCollectors(cols)
+
+	targets := MakeNNewTargets(9, 3, 0)
+	s.SetTargets(targets)
+
+	for _, col := range s.Collectors() {
+		assert.Equal(t, col.NumTargets, col.WeightedLoad,
+			"unlabeled targets should have weight 1, so WeightedLoad equals NumTargets for collector %s", col.Name)
+	}
+}
+
+// TestWeightedLoadInvalidAnnotation verifies that targets with an invalid weight annotation
+// (non-numeric, zero, negative, or >100) use the default weight (1).
+func TestWeightedLoadInvalidAnnotation(t *testing.T) {
+	invalidValues := []string{"notanumber", "0", "-5", "101", "999"}
+	for _, val := range invalidValues {
+		t.Run(val, func(t *testing.T) {
+			s, _ := New("least-weighted", logger)
+
+			cols := MakeNCollectors(1, 0)
+			s.SetCollectors(cols)
+
+			targets := MakeNTargetsWithWeight(2, "test-job", 0, config.WeightAnnotationMetaLabel, val)
+			s.SetTargets(targets)
+
+			for _, col := range s.Collectors() {
+				assert.Equal(t, 2, col.WeightedLoad,
+					"annotation %q should use default weight (1)", val)
+				assert.Equal(t, 2, col.NumTargets)
+			}
+		})
+	}
+}
+
 func TestTargetsWithNoCollectorsLeastWeighted(t *testing.T) {
 	s, _ := New("least-weighted", logger)
 

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

@@ -89,6 +89,7 @@ type Collector struct {
 	Name          string
 	NodeName      string
 	NumTargets    int
+	WeightedLoad  int // sum of weights of assigned targets; equals NumTargets when no weight annotation is set
 	TargetsPerJob map[string]int
 }
 

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

@@ -79,6 +79,21 @@ func MakeNTargetsForJob(n int, jobName string, startingIndex int) []*target.Item
 	return toReturn
 }
 
+// MakeNTargetsWithWeight creates n targets with a specific weight annotation.
+func MakeNTargetsWithWeight(n int, jobName string, startingIndex int, weightAnnotation string, weight string) []*target.Item {
+	toReturn := []*target.Item{}
+	for i := startingIndex; i < n+startingIndex; i++ {
+		label := labels.New(
+			labels.Label{Name: "i", Value: strconv.Itoa(i)},
+			labels.Label{Name: "total", Value: strconv.Itoa(n + startingIndex)},
+			labels.Label{Name: weightAnnotation, Value: weight},
+		)
+		newTarget := target.NewItem(jobName, fmt.Sprintf("test-url-%d", i), label, "")
+		toReturn = append(toReturn, newTarget)
+	}
+	return toReturn
+}
+
 func RunForAllStrategies(t *testing.T, f func(t *testing.T, allocator Allocator)) {
 	allocatorNames := GetRegisteredAllocatorNames()
 	logger := logf.Log.WithName("unit-tests")

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

@@ -76,6 +76,17 @@ type Config struct {
 	CollectorNotReadyGracePeriod time.Duration         `yaml:"collector_not_ready_grace_period,omitempty"`
 }
 
+const (
+	// WeightAnnotation is the pod annotation name used to specify a target's weight.
+	WeightAnnotation = "opentelemetry.io/target-allocation-weight"
+
+	// WeightAnnotationMetaLabel is the Kubernetes SD meta label that surfaces the weight annotation.
+	WeightAnnotationMetaLabel = "__meta_kubernetes_pod_annotation_opentelemetry_io_target_allocation_weight"
+
+	// DefaultWeight is the weight used when no weight is specified.
+	DefaultWeight = 1
+)
+
 type PrometheusCRConfig struct {
 	Enabled                         bool                          `yaml:"enabled,omitempty"`
 	AllowNamespaces                 []string                      `yaml:"allow_namespaces,omitempty"`