Feature/gen 2030 group sources to destination

Makefile

@@ -206,6 +206,11 @@ cli-uninstall:
 	@echo "Uninstalling odigos from source. version: $(ODIGOS_CLI_VERSION)"
 	cd ./cli ; go run -tags=embed_manifests . uninstall
 
+.PHONY: cli-uninstall
+cli-uninstall:
+	@echo "Installing odigos from source. version: $(ODIGOS_CLI_VERSION)"
+	cd ./cli ; go run -tags=embed_manifests . uninstall
+
 .PHONY: cli-upgrade
 cli-upgrade:
 	@echo "Upgrading odigos from source. version: $(ODIGOS_CLI_VERSION)"

api/config/crd/bases/odigos.io_destinations.yaml

@@ -72,6 +72,34 @@ spec:
                   - METRICS
                   type: string
                 type: array
+              sourceSelector:
+                description: |-
+                  SourceSelector defines which sources can send data to this destination.
+                  If not specified, defaults to "all".
+                properties:
+                  groups:
+                    description: Groups specifies the groups for "groups" mode.
+                    items:
+                      type: string
+                    type: array
+                  modes:
+                    description: |-
+                      Mode can contain a combination of "all", "namespaces", or "groups".
+                      Determines how sources are selected for this destination.
+                    enum:
+                    - all
+                    - namespaces
+                    - groups
+                    items:
+                      type: string
+                    type: array
+                  namespaces:
+                    description: Namespaces specifies the namespaces for "namespaces"
+                      mode.
+                    items:
+                      type: string
+                    type: array
+                type: object
               type:
                 type: string
             required:

api/odigos/v1alpha1/destination_types.go

@@ -30,6 +30,27 @@ type DestinationSpec struct {
 	Data            map[string]string            `json:"data"`
 	SecretRef       *v1.LocalObjectReference     `json:"secretRef,omitempty"`
 	Signals         []common.ObservabilitySignal `json:"signals"`
+
+	// SourceSelector defines which sources can send data to this destination.
+	// If not specified, defaults to "all".
+	// +optional
+	SourceSelector *SourceSelector `json:"sourceSelector,omitempty"`
+}
+
+// SourceSelector defines the criteria for selecting sources.
+type SourceSelector struct {
+	// Mode can contain a combination of "all", "namespaces", or "groups".
+	// Determines how sources are selected for this destination.
+	// +kubebuilder:validation:Enum=all;namespaces;groups
+	Modes []string `json:"modes,omitempty"`
+
+	// Namespaces specifies the namespaces for "namespaces" mode.
+	// +optional
+	Namespaces []string `json:"namespaces,omitempty"`
+
+	// Groups specifies the groups for "groups" mode.
+	// +optional
+	Groups []string `json:"groups,omitempty"`
 }
 
 // DestinationStatus defines the observed state of Destination

autoscaler/controllers/common/processors.go

@@ -1,13 +1,20 @@
 package common
 
 import (
+	"context"
 	"encoding/json"
+	"fmt"
 	"sort"
+	"strings"
 
+	"github.com/go-logr/logr"
 	odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
 	"github.com/odigos-io/odigos/common"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 func FilterAndSortProcessorsByOrderHint(processors *odigosv1.ProcessorList, collectorRole odigosv1.CollectorsGroupRole) []*odigosv1.Processor {
@@ -70,3 +77,115 @@ func GetGenericBatchProcessor() odigosv1.Processor {
 		},
 	}
 }
+
+type MatchCondition struct {
+	Name      string `mapstructure:"name"`
+	Namespace string `mapstructure:"namespace"`
+	Kind      string `mapstructure:"kind"`
+}
+
+func AddFilterProcessors(ctx context.Context, kubeClient client.Client, allProcessors *odigosv1.ProcessorList, dests *odigosv1.DestinationList) {
+	for _, dest := range dests.Items {
+		logger := log.Log.WithValues("destination", dest.Name)
+		logger.Info("Processing destination for filter processor")
+
+		if dest.Spec.SourceSelector == nil || contains(dest.Spec.SourceSelector.Modes, "all") {
+			logger.Info("Skipping destination as SourceSelector is nil or set to 'all'")
+			continue
+		}
+
+		var matchedSources []odigosv1.Source
+		if contains(dest.Spec.SourceSelector.Modes, "namespaces") {
+			matchedSources = append(matchedSources, fetchSourcesByNamespaces(ctx, kubeClient, dest.Spec.SourceSelector.Namespaces, logger)...)
+		}
+		if contains(dest.Spec.SourceSelector.Modes, "groups") {
+			matchedSources = append(matchedSources, fetchSourcesByGroups(ctx, kubeClient, dest.Spec.SourceSelector.Groups, logger)...)
+		}
+
+		logger.Info("Matched sources for destination", "matchedSources", matchedSources)
+
+		matchConditions := make(map[string]bool)
+		for _, source := range matchedSources {
+			logger.Info("Adding match condition for source", "sourceName", source.Spec.Workload.Name, "namespace", source.Spec.Workload.Namespace, "kind", source.Spec.Workload.Kind)
+
+			key := fmt.Sprintf("%s/%s/%s", source.Spec.Workload.Namespace, source.Spec.Workload.Name, source.Spec.Workload.Kind)
+			matchConditions[key] = true
+		}
+
+		if len(matchConditions) == 0 {
+			logger.Info("No matched sources for destination. Skipping processor creation.")
+			continue
+		}
+
+		filterConfig := map[string]interface{}{
+			"match_conditions": matchConditions,
+		}
+
+		allProcessors.Items = append(allProcessors.Items, odigosv1.Processor{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("odigosroutingfilterprocessor-%s", dest.Name),
+			},
+			Spec: odigosv1.ProcessorSpec{
+				Type:            "odigosroutingfilterprocessor",
+				ProcessorConfig: runtime.RawExtension{Raw: marshalConfig(filterConfig)},
+				CollectorRoles: []odigosv1.CollectorsGroupRole{
+					odigosv1.CollectorsGroupRoleClusterGateway,
+				},
+				OrderHint: len(allProcessors.Items) + 1,
+				Signals:   dest.Spec.Signals,
+			},
+		})
+
+	}
+}
+
+func fetchSourcesByNamespaces(ctx context.Context, kubeClient client.Client, namespaces []string, logger logr.Logger) []odigosv1.Source {
+	var sources []odigosv1.Source
+	for _, ns := range namespaces {
+		sourceList := &odigosv1.SourceList{}
+		err := kubeClient.List(ctx, sourceList, &client.ListOptions{Namespace: ns})
+		if err != nil {
+			logger.Error(err, "Failed to fetch sources by namespace", "namespace", ns)
+			continue
+		}
+		sources = append(sources, sourceList.Items...)
+	}
+	return sources
+}
+
+func fetchSourcesByGroups(ctx context.Context, kubeClient client.Client, groups []string, logger logr.Logger) []odigosv1.Source {
+	selectors := make([]string, len(groups))
+	for i, group := range groups {
+		selectors[i] = fmt.Sprintf("odigos.io/group-%s=true", group)
+	}
+	labelSelector := labels.SelectorFromSet(labels.Set{
+		strings.Join(selectors, ","): "",
+	})
+
+	sourceList := &odigosv1.SourceList{}
+	err := kubeClient.List(ctx, sourceList, &client.ListOptions{
+		LabelSelector: labelSelector,
+	})
+	if err != nil {
+		logger.Error(err, "Failed to fetch sources by groups", "groups", groups)
+		return nil
+	}
+	return sourceList.Items
+}
+
+func marshalConfig(config map[string]interface{}) []byte {
+	data, err := json.Marshal(config)
+	if err != nil {
+		log.Log.Error(err, "Failed to marshal processor config")
+	}
+	return data
+}
+
+func contains(arr []string, val string) bool {
+	for _, item := range arr {
+		if item == val {
+			return true
+		}
+	}
+	return false
+}

autoscaler/controllers/destination_controller.go

@@ -41,6 +41,7 @@ type DestinationReconciler struct {
 func (r *DestinationReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	logger := log.FromContext(ctx)
 	logger.V(0).Info("Reconciling Destination")
+
 	err := gateway.Sync(ctx, r.Client, r.Scheme, r.ImagePullSecrets, r.OdigosVersion, r.Config)
 	if err != nil {
 		return ctrl.Result{}, err

autoscaler/controllers/gateway/configmap.go

@@ -114,7 +114,7 @@ func addSelfTelemetryPipeline(c *config.Config, ownTelemetryPort int32) error {
 func syncConfigMap(dests *odigosv1.DestinationList, allProcessors *odigosv1.ProcessorList, gateway *odigosv1.CollectorsGroup, ctx context.Context, c client.Client, scheme *runtime.Scheme) ([]odigoscommon.ObservabilitySignal, error) {
 	logger := log.FromContext(ctx)
 	memoryLimiterConfiguration := common.GetMemoryLimiterConfig(gateway.Spec.ResourcesSettings)
-
+	common.AddFilterProcessors(ctx, c, allProcessors, dests)
 	processors := common.FilterAndSortProcessorsByOrderHint(allProcessors, odigosv1.CollectorsGroupRoleClusterGateway)
 
 	desiredData, err, status, signals := config.Calculate(