Target Allocator implementation (Part 1 - OTEL Operator Enhancements)

api/v1alpha1/opentelemetrycollector_types.go

@@ -41,6 +41,11 @@ type OpenTelemetryCollectorSpec struct {
 	// +operator-sdk:gen-csv:customresourcedefinitions.specDescriptors=true
 	Image string `json:"image,omitempty"`
 
+	// TargetAllocator indicates a value which determines whether to spawn a target allocation resource or not.
+	// +optional
+	// +operator-sdk:gen-csv:customresourcedefinitions.specDescriptors=true
+	TargetAllocator OpenTelemetryTargetAllocator `json:"targetAllocator,omitempty"`
+
 	// Mode represents how the collector should be deployed (deployment, daemonset, statefulset or sidecar)
 	// +optional
 	// +operator-sdk:gen-csv:customresourcedefinitions.specDescriptors=true
@@ -116,6 +121,17 @@ type OpenTelemetryCollectorStatus struct {
 	Messages []string `json:"messages,omitempty"`
 }
 
+// OpenTelemetryTargetAllocator defines the configurations for the Prometheus target allocator.
+type OpenTelemetryTargetAllocator struct {
+	// Enabled indicates whether to use a target allocation mechanism for Prometheus targets or not.
+	// +optional
+	Enabled bool `json:"enabled,omitempty"`
+
+	// Image indicates the container image to use for the OpenTelemetry TargetAllocator.
+	// +optional
+	Image string `json:"image,omitempty"`
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:resource:shortName=otelcol;otelcols
 // +kubebuilder:subresource:status

api/v1alpha1/opentelemetrycollector_webhook.go

@@ -91,5 +91,10 @@ func (r *OpenTelemetryCollector) validateCRDSpec() error {
 		return fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which does not support the attribute 'tolerations'", r.Spec.Mode)
 	}
 
+	// validate target allocation
+	if r.Spec.TargetAllocator.Enabled && r.Spec.Mode != ModeStatefulSet {
+		return fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which does not support the target allocation deployment", r.Spec.Mode)
+	}
+
 	return nil
 }

bundle/manifests/opentelemetry.io_opentelemetrycollectors.yaml

@@ -442,6 +442,19 @@ spec:
                 description: ServiceAccount indicates the name of an existing service
                   account to use with this instance.
                 type: string
+              targetAllocator:
+                description: TargetAllocator indicates a value which determines whether
+                  to spawn a target allocation resource or not.
+                properties:
+                  enabled:
+                    description: Enabled indicates whether to use a target allocation
+                      mechanism for Prometheus targets or not.
+                    type: boolean
+                  image:
+                    description: Image indicates the container image to use for the
+                      OpenTelemetry TargetAllocator.
+                    type: string
+                type: object
               tolerations:
                 description: Toleration to schedule OpenTelemetry Collector pods.
                   This is only relevant to daemonsets, statefulsets and deployments

config/crd/bases/opentelemetry.io_opentelemetrycollectors.yaml

@@ -430,6 +430,19 @@ spec:
                 description: ServiceAccount indicates the name of an existing service
                   account to use with this instance.
                 type: string
+              targetAllocator:
+                description: TargetAllocator indicates a value which determines whether
+                  to spawn a target allocation resource or not.
+                properties:
+                  enabled:
+                    description: Enabled indicates whether to use a target allocation
+                      mechanism for Prometheus targets or not.
+                    type: boolean
+                  image:
+                    description: Image indicates the container image to use for the
+                      OpenTelemetry TargetAllocator.
+                    type: string
+                type: object
               tolerations:
                 description: Toleration to schedule OpenTelemetry Collector pods.
                   This is only relevant to daemonsets, statefulsets and deployments

internal/config/main.go

@@ -29,8 +29,9 @@ import (
 )
 
 const (
-	defaultAutoDetectFrequency     = 5 * time.Second
-	defaultCollectorConfigMapEntry = "collector.yaml"
+	defaultAutoDetectFrequency           = 5 * time.Second
+	defaultCollectorConfigMapEntry       = "collector.yaml"
+	defaultTargetAllocatorConfigMapEntry = "targetallocator.yaml"
 )
 
 // Config holds the static configuration for this operator.
@@ -44,21 +45,24 @@ type Config struct {
 	onChange            []func() error
 
 	// config state
-	collectorImage          string
-	collectorConfigMapEntry string
-	platform                platform.Platform
-	version                 version.Version
+	collectorImage                string
+	collectorConfigMapEntry       string
+	targetAllocatorImage          string
+	targetAllocatorConfigMapEntry string
+	platform                      platform.Platform
+	version                       version.Version
 }
 
 // New constructs a new configuration based on the given options.
 func New(opts ...Option) Config {
 	// initialize with the default values
 	o := options{
-		autoDetectFrequency:     defaultAutoDetectFrequency,
-		collectorConfigMapEntry: defaultCollectorConfigMapEntry,
-		logger:                  logf.Log.WithName("config"),
-		platform:                platform.Unknown,
-		version:                 version.Get(),
+		autoDetectFrequency:           defaultAutoDetectFrequency,
+		collectorConfigMapEntry:       defaultCollectorConfigMapEntry,
+		targetAllocatorConfigMapEntry: defaultTargetAllocatorConfigMapEntry,
+		logger:                        logf.Log.WithName("config"),
+		platform:                      platform.Unknown,
+		version:                       version.Get(),
 	}
 	for _, opt := range opts {
 		opt(&o)
@@ -70,15 +74,22 @@ func New(opts ...Option) Config {
 		o.collectorImage = fmt.Sprintf("otel/opentelemetry-collector:%s", o.version.OpenTelemetryCollector)
 	}
 
+	if len(o.targetAllocatorImage) == 0 {
+		// will be replcaed with target allocator image once approved and available
+		o.targetAllocatorImage = "raul9595/otel-loadbalancer:0.0.1"
+	}
+
 	return Config{
-		autoDetect:              o.autoDetect,
-		autoDetectFrequency:     o.autoDetectFrequency,
-		collectorImage:          o.collectorImage,
-		collectorConfigMapEntry: o.collectorConfigMapEntry,
-		logger:                  o.logger,
-		onChange:                o.onChange,
-		platform:                o.platform,
-		version:                 o.version,
+		autoDetect:                    o.autoDetect,
+		autoDetectFrequency:           o.autoDetectFrequency,
+		collectorImage:                o.collectorImage,
+		collectorConfigMapEntry:       o.collectorConfigMapEntry,
+		targetAllocatorImage:          o.targetAllocatorImage,
+		targetAllocatorConfigMapEntry: o.targetAllocatorConfigMapEntry,
+		logger:                        o.logger,
+		onChange:                      o.onChange,
+		platform:                      o.platform,
+		version:                       o.version,
 	}
 }
 
@@ -155,6 +166,16 @@ func (c *Config) CollectorConfigMapEntry() string {
 	return c.collectorConfigMapEntry
 }
 
+// TargetAllocatorImage represents the flag to override the OpenTelemetry TargetAllocator container image.
+func (c *Config) TargetAllocatorImage() string {
+	return c.targetAllocatorImage
+}
+
+// TargetAllocatorConfigMapEntry represents the configuration file name for the TargetAllocator. Immutable.
+func (c *Config) TargetAllocatorConfigMapEntry() string {
+	return c.targetAllocatorConfigMapEntry
+}
+
 // Platform represents the type of the platform this operator is running.
 func (c *Config) Platform() platform.Platform {
 	return c.platform

internal/config/options.go

@@ -28,14 +28,16 @@ import (
 type Option func(c *options)
 
 type options struct {
-	autoDetect              autodetect.AutoDetect
-	autoDetectFrequency     time.Duration
-	collectorImage          string
-	collectorConfigMapEntry string
-	logger                  logr.Logger
-	onChange                []func() error
-	platform                platform.Platform
-	version                 version.Version
+	autoDetect                    autodetect.AutoDetect
+	autoDetectFrequency           time.Duration
+	targetAllocatorImage          string
+	collectorImage                string
+	collectorConfigMapEntry       string
+	targetAllocatorConfigMapEntry string
+	logger                        logr.Logger
+	onChange                      []func() error
+	platform                      platform.Platform
+	version                       version.Version
 }
 
 func WithAutoDetect(a autodetect.AutoDetect) Option {
@@ -48,6 +50,12 @@ func WithAutoDetectFrequency(t time.Duration) Option {
 		o.autoDetectFrequency = t
 	}
 }
+func WithTargetAllocatorImage(s string) Option {
+	return func(o *options) {
+		o.targetAllocatorImage = s
+	}
+}
+
 func WithCollectorImage(s string) Option {
 	return func(o *options) {
 		o.collectorImage = s
@@ -58,6 +66,11 @@ func WithCollectorConfigMapEntry(s string) Option {
 		o.collectorConfigMapEntry = s
 	}
 }
+func WithTargetAllocatorConfigMapEntry(s string) Option {
+	return func(o *options) {
+		o.targetAllocatorConfigMapEntry = s
+	}
+}
 func WithLogger(logger logr.Logger) Option {
 	return func(o *options) {
 		o.logger = logger

pkg/collector/container_test.go

@@ -17,12 +17,11 @@ package collector_test
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/open-telemetry/opentelemetry-operator/api/v1alpha1"
 	"github.com/open-telemetry/opentelemetry-operator/internal/config"
 	. "github.com/open-telemetry/opentelemetry-operator/pkg/collector"

pkg/collector/deployment_test.go

@@ -18,7 +18,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 

pkg/collector/reconcile/configmap.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"reflect"
 
+	"gopkg.in/yaml.v2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -28,6 +29,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-operator/pkg/collector"
 	"github.com/open-telemetry/opentelemetry-operator/pkg/naming"
+	"github.com/open-telemetry/opentelemetry-operator/pkg/targetallocator"
 )
 
 // +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch;create;update;patch;delete
@@ -38,6 +40,14 @@ func ConfigMaps(ctx context.Context, params Params) error {
 		desiredConfigMap(ctx, params),
 	}
 
+	if params.Instance.Spec.TargetAllocator.Enabled {
+		cm, err := desiredTAConfigMap(ctx, params)
+		if err != nil {
+			return fmt.Errorf("failed to parse config: %v", err)
+		}
+		desired = append(desired, cm)
+	}
+
 	// first, handle the create/update parts
 	if err := expectedConfigMaps(ctx, params, desired, true); err != nil {
 		return fmt.Errorf("failed to reconcile the expected configmaps: %v", err)
@@ -69,6 +79,41 @@ func desiredConfigMap(_ context.Context, params Params) corev1.ConfigMap {
 	}
 }
 
+func desiredTAConfigMap(_ context.Context, params Params) (corev1.ConfigMap, error) {
+	name := naming.TAConfigMap(params.Instance)
+	labels := targetallocator.Labels(params.Instance)
+	labels["app.kubernetes.io/name"] = name
+
+	promConfig, err := GetPromConfig(params)
+	if err != nil {
+		return corev1.ConfigMap{}, err
+	}
+
+	taConfig := make(map[interface{}]interface{})
+	taConfig["label_selector"] = map[string]string{
+		"app.kubernetes.io/instance":   fmt.Sprintf("%s.%s", params.Instance.Namespace, params.Instance.Name),
+		"app.kubernetes.io/managed-by": "opentelemetry-operator",
+		"app.kubernetes.io/component":  "opentelemetry-collector",
+	}
+	taConfig["config"] = promConfig
+	taConfigYAML, err := yaml.Marshal(taConfig)
+	if err != nil {
+		return corev1.ConfigMap{}, err
+	}
+
+	return corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        name,
+			Namespace:   params.Instance.Namespace,
+			Labels:      labels,
+			Annotations: params.Instance.Annotations,
+		},
+		Data: map[string]string{
+			"targetallocator.yaml": string(taConfigYAML),
+		},
+	}, nil
+}
+
 func expectedConfigMaps(ctx context.Context, params Params, expected []corev1.ConfigMap, retry bool) error {
 	for _, obj := range expected {
 		desired := obj