[target-allocator] Introduce "per node" allocation strategy to target allocator

apis/v1alpha1/collector_webhook.go

@@ -228,40 +228,14 @@ func (c CollectorWebhook) validate(ctx context.Context, r *OpenTelemetryCollecto
 		return warnings, fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which does not support the attribute 'AdditionalContainers'", r.Spec.Mode)
 	}
 
-	// validate target allocation
-	if r.Spec.TargetAllocator.Enabled && (r.Spec.Mode != ModeStatefulSet && r.Spec.Mode != ModeDaemonSet) {
-		return warnings, fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which does not support the target allocation deployment", r.Spec.Mode)
-	}
-
-	if r.Spec.TargetAllocator.Enabled && (r.Spec.Mode == ModeDaemonSet && r.Spec.TargetAllocator.AllocationStrategy != OpenTelemetryTargetAllocatorAllocationStrategyPerNode) {
-		return warnings, fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which must be used with target allocation strategy %s ", r.Spec.Mode, OpenTelemetryTargetAllocatorAllocationStrategyPerNode)
-	}
-
-	if r.Spec.TargetAllocator.AllocationStrategy == OpenTelemetryTargetAllocatorAllocationStrategyPerNode && r.Spec.Mode != ModeDaemonSet {
-		return warnings, fmt.Errorf("target allocation strategy %s is only supported in OpenTelemetry Collector mode %s", OpenTelemetryTargetAllocatorAllocationStrategyPerNode, ModeDaemonSet)
-	}
-
-	// validate Prometheus config for target allocation
+	// validate target allocator configs
 	if r.Spec.TargetAllocator.Enabled {
-		promCfg, err := ta.ConfigToPromConfig(r.Spec.Config)
-		if err != nil {
-			return warnings, fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
-		}
-		err = ta.ValidatePromConfig(promCfg, r.Spec.TargetAllocator.Enabled, featuregate.EnableTargetAllocatorRewrite.IsEnabled())
+		taWarnings, err := c.validateTargetAllocatorConfig(ctx, r)
 		if err != nil {
-			return warnings, fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
+			return warnings, err
 		}
-		err = ta.ValidateTargetAllocatorConfig(r.Spec.TargetAllocator.PrometheusCR.Enabled, promCfg)
-		if err != nil {
-			return warnings, fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
-		}
-		// if the prometheusCR is enabled, it needs a suite of permissions to function
-		if r.Spec.TargetAllocator.PrometheusCR.Enabled {
-			if subjectAccessReviews, err := c.reviewer.CheckPolicyRules(ctx, r.GetNamespace(), r.Spec.TargetAllocator.ServiceAccount, targetAllocatorCRPolicyRules...); err != nil {
-				return warnings, fmt.Errorf("unable to check rbac rules %w", err)
-			} else if allowed, deniedReviews := rbac.AllSubjectAccessReviewsAllowed(subjectAccessReviews); !allowed {
-				warnings = append(warnings, warningsGroupedByResource(deniedReviews)...)
-			}
+		if taWarnings != nil {
+			warnings = append(warnings, taWarnings...)
 		}
 	}
 
@@ -368,6 +342,44 @@ func (c CollectorWebhook) validate(ctx context.Context, r *OpenTelemetryCollecto
 	return warnings, nil
 }
 
+func (c CollectorWebhook) validateTargetAllocatorConfig(ctx context.Context, r *OpenTelemetryCollector) (admission.Warnings, error) {
+	if r.Spec.Mode != ModeStatefulSet && r.Spec.Mode != ModeDaemonSet {
+		return nil, fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which does not support the target allocation deployment", r.Spec.Mode)
+	}
+
+	if r.Spec.Mode == ModeDaemonSet && r.Spec.TargetAllocator.AllocationStrategy != OpenTelemetryTargetAllocatorAllocationStrategyPerNode {
+		return nil, fmt.Errorf("the OpenTelemetry Collector mode is set to %s, which must be used with target allocation strategy %s ", r.Spec.Mode, OpenTelemetryTargetAllocatorAllocationStrategyPerNode)
+	}
+
+	if r.Spec.TargetAllocator.AllocationStrategy == OpenTelemetryTargetAllocatorAllocationStrategyPerNode && r.Spec.Mode != ModeDaemonSet {
+		return nil, fmt.Errorf("target allocation strategy %s is only supported in OpenTelemetry Collector mode %s", OpenTelemetryTargetAllocatorAllocationStrategyPerNode, ModeDaemonSet)
+	}
+
+	// validate Prometheus config for target allocation
+	promCfg, err := ta.ConfigToPromConfig(r.Spec.Config)
+	if err != nil {
+		return nil, fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
+	}
+	err = ta.ValidatePromConfig(promCfg, r.Spec.TargetAllocator.Enabled, featuregate.EnableTargetAllocatorRewrite.IsEnabled())
+	if err != nil {
+		return nil, fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
+	}
+	err = ta.ValidateTargetAllocatorConfig(r.Spec.TargetAllocator.PrometheusCR.Enabled, promCfg)
+	if err != nil {
+		return nil, fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
+	}
+	// if the prometheusCR is enabled, it needs a suite of permissions to function
+	if r.Spec.TargetAllocator.PrometheusCR.Enabled {
+		if subjectAccessReviews, err := c.reviewer.CheckPolicyRules(ctx, r.GetNamespace(), r.Spec.TargetAllocator.ServiceAccount, targetAllocatorCRPolicyRules...); err != nil {
+			return nil, fmt.Errorf("unable to check rbac rules %w", err)
+		} else if allowed, deniedReviews := rbac.AllSubjectAccessReviewsAllowed(subjectAccessReviews); !allowed {
+			return warningsGroupedByResource(deniedReviews), nil
+		}
+	}
+
+	return nil, nil
+}
+
 func checkAutoscalerSpec(autoscaler *AutoscalerSpec) error {
 	if autoscaler.Behavior != nil {
 		if autoscaler.Behavior.ScaleDown != nil && autoscaler.Behavior.ScaleDown.StabilizationWindowSeconds != nil &&

cmd/otel-allocator/allocation/per_node.go

@@ -190,12 +190,8 @@ func (allocator *perNodeAllocator) handleTargets(diff diff.Changes[*target.Item]
 
 	// Check for unassigned targets
 	if len(unassignedTargetsForJobs) > 0 {
-		jobs := make([]string, 0, len(unassignedTargetsForJobs))
-		for j := range unassignedTargetsForJobs {
-			jobs = append(jobs, j)
-		}
-
-		allocator.log.Info("Could not assign targets for the following jobs due to missing node labels", "jobs", jobs)
+		allocator.log.Info("Could not assign targets for some jobs due to missing node labels", "jobs", unassignedTargetsForJobs)
+		RecordJobsWithUnassignedTargets(unassignedTargetsForJobs)
 	}
 }
 

cmd/otel-allocator/allocation/strategy.go

@@ -49,6 +49,10 @@ var (
 		Name: "opentelemetry_allocator_targets_remaining",
 		Help: "Number of targets kept after filtering.",
 	})
+	jobsUnassigned = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "opentelemetry_allocator_jobs_with_unassigned_targets",
+		Help: "Number of jobs with targets that could not be assigned due to missing node label.",
+	})
 )
 
 type AllocationOption func(Allocator)
@@ -67,6 +71,10 @@ func RecordTargetsKept(targets map[string]*target.Item) {
 	targetsRemaining.Add(float64(len(targets)))
 }
 
+func RecordJobsWithUnassignedTargets(jobs map[string]struct{}) {
+	jobsUnassigned.Add(float64(len(jobs)))
+}
+
 func New(name string, log logr.Logger, opts ...AllocationOption) (Allocator, error) {
 	if p, ok := registry[name]; ok {
 		return p(log.WithValues("allocator", name), opts...), nil