Allow setting target allocator via label

.github/workflows/e2e.yaml

@@ -51,6 +51,8 @@ jobs:
            kube-version: "1.29"
          - group: e2e-targetallocator
            setup: "enable-targetallocator-cr prepare-e2e"
+         - group: e2e-targetallocator-cr
+           setup: "enable-targetallocator-cr prepare-e2e"
     steps:
     - name: Check out code into the Go module directory
       uses: actions/checkout@v4

Makefile

@@ -327,6 +327,11 @@ e2e-prometheuscr: chainsaw
 e2e-targetallocator: chainsaw
 	$(CHAINSAW) test --test-dir ./tests/e2e-targetallocator
 
+# Target allocator CR end-to-tests
+.PHONY: e2e-targetallocator-cr
+e2e-targetallocator-cr: chainsaw
+	$(CHAINSAW) test --test-dir ./tests/e2e-targetallocator-cr
+
 .PHONY: add-certmanager-permissions
 add-certmanager-permissions: 
 	# Kustomize only allows patches in the folder where the kustomization is located

controllers/opentelemetrycollector_controller.go

@@ -38,6 +38,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
+	"github.com/open-telemetry/opentelemetry-operator/apis/v1alpha1"
 	"github.com/open-telemetry/opentelemetry-operator/apis/v1beta1"
 	"github.com/open-telemetry/opentelemetry-operator/internal/autodetect/openshift"
 	"github.com/open-telemetry/opentelemetry-operator/internal/autodetect/prometheus"
@@ -50,6 +51,8 @@ import (
 	"github.com/open-telemetry/opentelemetry-operator/pkg/featuregate"
 )
 
+const collectorTargetAllocatorLabelName = "opentelemetry.io/target-allocator"
+
 var (
 	ownedClusterObjectTypes = []client.Object{
 		&rbacv1.ClusterRole{},
@@ -168,7 +171,7 @@ func (r *OpenTelemetryCollectorReconciler) getConfigMapsToRemove(configVersionsT
 	return ownedConfigMaps
 }
 
-func (r *OpenTelemetryCollectorReconciler) GetParams(instance v1beta1.OpenTelemetryCollector) (manifests.Params, error) {
+func (r *OpenTelemetryCollectorReconciler) GetParams(ctx context.Context, instance v1beta1.OpenTelemetryCollector) (manifests.Params, error) {
 	p := manifests.Params{
 		Config:   r.config,
 		Client:   r.Client,
@@ -179,14 +182,28 @@ func (r *OpenTelemetryCollectorReconciler) GetParams(instance v1beta1.OpenTeleme
 	}
 
 	// generate the target allocator CR from the collector CR
-	targetAllocator, err := collector.TargetAllocator(p)
+	targetAllocator, err := r.getTargetAllocator(ctx, p)
 	if err != nil {
 		return p, err
 	}
 	p.TargetAllocator = targetAllocator
 	return p, nil
 }
 
+func (r *OpenTelemetryCollectorReconciler) getTargetAllocator(ctx context.Context, params manifests.Params) (*v1alpha1.TargetAllocator, error) {
+	otelcol := params.OtelCol
+	if taName, ok := otelcol.GetLabels()[collectorTargetAllocatorLabelName]; ok {
+		targetAllocator := &v1alpha1.TargetAllocator{}
+		taKey := client.ObjectKey{Name: taName, Namespace: otelcol.GetNamespace()}
+		err := r.Client.Get(ctx, taKey, targetAllocator)
+		if err != nil {
+			return nil, err
+		}
+		return targetAllocator, nil
+	}
+	return collector.TargetAllocator(params)
+}
+
 // NewReconciler creates a new reconciler for OpenTelemetryCollector objects.
 func NewReconciler(p Params) *OpenTelemetryCollectorReconciler {
 	r := &OpenTelemetryCollectorReconciler{
@@ -230,7 +247,7 @@ func (r *OpenTelemetryCollectorReconciler) Reconcile(ctx context.Context, req ct
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	params, err := r.GetParams(instance)
+	params, err := r.GetParams(ctx, instance)
 	if err != nil {
 		log.Error(err, "Failed to create manifest.Params")
 		return ctrl.Result{}, err

controllers/targetallocator_controller.go

@@ -98,7 +98,24 @@ func (r *TargetAllocatorReconciler) getCollector(ctx context.Context, instance v
 		return &collector, nil
 	}
 
-	return nil, nil
+	var collectors v1beta1.OpenTelemetryCollectorList
+	listOpts := []client.ListOption{
+		client.InNamespace(instance.GetNamespace()),
+		client.MatchingLabels{
+			collectorTargetAllocatorLabelName: instance.GetName(),
+		},
+	}
+	err := r.List(ctx, &collectors, listOpts...)
+	if err != nil {
+		return nil, err
+	}
+	if len(collectors.Items) == 0 {
+		return nil, nil
+	} else if len(collectors.Items) > 1 {
+		return nil, fmt.Errorf("found multiple OpenTelemetry collectors annotated with the same Target Allocator: %s/%s", instance.GetNamespace(), instance.GetName())
+	}
+
+	return &collectors.Items[0], nil
 }
 
 // NewTargetAllocatorReconciler creates a new reconciler for TargetAllocator objects.
@@ -195,6 +212,25 @@ func (r *TargetAllocatorReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		),
 	)
 
+	// watch collectors which have the target allocator label
+	collectorSelector := metav1.LabelSelector{
+		MatchExpressions: []metav1.LabelSelectorRequirement{
+			{
+				Key:      collectorTargetAllocatorLabelName,
+				Operator: metav1.LabelSelectorOpExists,
+			},
+		},
+	}
+	selectorPredicate, err := predicate.LabelSelectorPredicate(collectorSelector)
+	if err != nil {
+		return err
+	}
+	ctrlBuilder.Watches(
+		&v1beta1.OpenTelemetryCollector{},
+		handler.EnqueueRequestsFromMapFunc(getTargetAllocatorRequestsFromLabel),
+		builder.WithPredicates(selectorPredicate),
+	)
+
 	return ctrlBuilder.Complete(r)
 }
 
@@ -208,3 +244,17 @@ func getTargetAllocatorForCollector(_ context.Context, collector client.Object)
 		},
 	}
 }
+
+func getTargetAllocatorRequestsFromLabel(_ context.Context, collector client.Object) []reconcile.Request {
+	if taName, ok := collector.GetLabels()[collectorTargetAllocatorLabelName]; ok {
+		return []reconcile.Request{
+			{
+				NamespacedName: types.NamespacedName{
+					Name:      taName,
+					Namespace: collector.GetNamespace(),
+				},
+			},
+		}
+	}
+	return []reconcile.Request{}
+}

controllers/targetallocator_reconciler_test.go

@@ -55,7 +55,10 @@ func init() {
 func TestTargetAllocatorReconciler_GetCollector(t *testing.T) {
 	testCollector := &v1beta1.OpenTelemetryCollector{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: "my-instance-collector",
+			Name: "test",
+			Labels: map[string]string{
+				collectorTargetAllocatorLabelName: "label-ta",
+			},
 		},
 	}
 	fakeClient := fake.NewFakeClient(testCollector)
@@ -105,6 +108,36 @@ func TestTargetAllocatorReconciler_GetCollector(t *testing.T) {
 		assert.Nil(t, collector)
 		assert.Errorf(t, err, "error getting owner for TargetAllocator default/test: opentelemetrycollectors.opentelemetry.io \"non_existent\" not found")
 	})
+	t.Run("collector attached by label", func(t *testing.T) {
+		ta := v1alpha1.TargetAllocator{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "label-ta",
+			},
+		}
+		collector, err := reconciler.getCollector(context.Background(), ta)
+		require.NoError(t, err)
+		assert.Equal(t, testCollector, collector)
+	})
+	t.Run("multiple collectors attached by label", func(t *testing.T) {
+		testCollector2 := testCollector.DeepCopy()
+		testCollector2.SetName("test2")
+		fakeClient := fake.NewFakeClient(testCollector, testCollector2)
+		reconciler := NewTargetAllocatorReconciler(
+			fakeClient,
+			testScheme,
+			record.NewFakeRecorder(10),
+			config.New(),
+			testLogger,
+		)
+		ta := v1alpha1.TargetAllocator{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "label-ta",
+			},
+		}
+		collector, err := reconciler.getCollector(context.Background(), ta)
+		assert.Nil(t, collector)
+		assert.Errorf(t, err, "found multiple OpenTelemetry collectors annotated with the same Target Allocator: %s/%s", ta.Namespace, ta.Name)
+	})
 }
 
 func TestGetTargetAllocatorForCollector(t *testing.T) {
@@ -123,3 +156,23 @@ func TestGetTargetAllocatorForCollector(t *testing.T) {
 	}}
 	assert.Equal(t, expected, requests)
 }
+
+func TestGetTargetAllocatorRequestsFromLabel(t *testing.T) {
+	testCollector := &v1beta1.OpenTelemetryCollector{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test",
+			Namespace: "default",
+			Labels: map[string]string{
+				collectorTargetAllocatorLabelName: "label-ta",
+			},
+		},
+	}
+	requests := getTargetAllocatorRequestsFromLabel(context.Background(), testCollector)
+	expected := []reconcile.Request{{
+		NamespacedName: types.NamespacedName{
+			Name:      "label-ta",
+			Namespace: "default",
+		},
+	}}
+	assert.Equal(t, expected, requests)
+}

main.go

@@ -440,7 +440,7 @@ func main() {
 
 		bv := func(collector otelv1beta1.OpenTelemetryCollector) admission.Warnings {
 			var warnings admission.Warnings
-			params, newErr := collectorReconciler.GetParams(collector)
+			params, newErr := collectorReconciler.GetParams(context.Background(), collector)
 			if err != nil {
 				warnings = append(warnings, newErr.Error())
 				return warnings

tests/e2e-targetallocator-cr/targetallocator-label/00-assert.yaml

@@ -0,0 +1,40 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  labels:
+    app.kubernetes.io/name: ta-collector
+data:
+  collector.yaml: |
+    receivers:
+      prometheus:
+        config:
+          scrape_configs:
+            - job_name: otel-collector
+              scrape_interval: 10s
+              static_configs:
+                - targets:
+                    - 0.0.0.0:8888
+    exporters:
+      debug: {}
+    service:
+      telemetry:
+        metrics:
+          address: 0.0.0.0:8888
+      pipelines:
+        metrics:
+          exporters:
+            - debug
+          receivers:
+            - prometheus
+
+---
+apiVersion: v1
+data:
+  targetallocator.yaml: |
+    allocation_strategy: consistent-hashing
+    collector_selector: null
+    filter_strategy: ""
+kind: ConfigMap
+metadata:
+  name: ta-targetallocator

tests/e2e-targetallocator-cr/targetallocator-label/00-install.yaml

@@ -0,0 +1,30 @@
+---
+apiVersion: opentelemetry.io/v1alpha1
+kind: TargetAllocator
+metadata:
+  name: ta
+spec:
+---
+apiVersion: opentelemetry.io/v1beta1
+kind: OpenTelemetryCollector
+metadata:
+  name: ta
+spec:
+  mode: statefulset
+  config:
+    receivers:
+      prometheus:
+        config:
+          scrape_configs:
+            - job_name: 'otel-collector'
+              scrape_interval: 10s
+              static_configs:
+                - targets: [ '0.0.0.0:8888' ]
+    exporters:
+      debug: {}
+    service:
+      pipelines:
+        metrics:
+          receivers: [prometheus]
+          exporters: [debug]
+

tests/e2e-targetallocator-cr/targetallocator-label/01-add-ta-label.yaml

@@ -0,0 +1,26 @@
+---
+apiVersion: opentelemetry.io/v1beta1
+kind: OpenTelemetryCollector
+metadata:
+  name: ta
+  labels:
+    opentelemetry.io/target-allocator: ta
+spec:
+  mode: statefulset
+  config:
+    receivers:
+      prometheus:
+        config:
+          scrape_configs:
+            - job_name: 'otel-collector'
+              scrape_interval: 10s
+              static_configs:
+                - targets: [ '0.0.0.0:8888' ]
+    exporters:
+      debug: {}
+    service:
+      pipelines:
+        metrics:
+          receivers: [prometheus]
+          exporters: [debug]
+

tests/e2e-targetallocator-cr/targetallocator-label/01-assert.yaml

@@ -0,0 +1,39 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  labels:
+    app.kubernetes.io/name: ta-collector
+data:
+  collector.yaml: |
+    exporters:
+        debug: {}
+    receivers:
+        prometheus:
+            config: {}
+            target_allocator:
+                collector_id: ${POD_NAME}
+                endpoint: http://ta-targetallocator:80
+                interval: 30s
+    service:
+        pipelines:
+            metrics:
+                exporters:
+                    - debug
+                receivers:
+                    - prometheus
+        telemetry:
+            metrics:
+                address: 0.0.0.0:8888
+---
+apiVersion: v1
+data:
+  targetallocator.yaml:
+    ( contains(@, join(':', ['app.kubernetes.io/component', ' opentelemetry-collector'])) ): true
+    ( contains(@, join('', ['app.kubernetes.io/instance:', ' ', $namespace, '.ta'])) ): true
+    ( contains(@, join(':', ['app.kubernetes.io/managed-by', ' opentelemetry-operator'])) ): true
+    ( contains(@, join(':', ['app.kubernetes.io/part-of', ' opentelemetry'])) ): true
+    ( contains(@, join(':', ['job_name', ' otel-collector'])) ): true
+kind: ConfigMap
+metadata:
+  name: ta-targetallocator