wip

Author: owais

cmd/operator-opamp-bridge/internal/config/flags.go

@@ -26,6 +26,7 @@ const (
 	nameFlagName              = "name"
 	modeFlagName              = "mode"
 	defaultHeartbeatInterval  = 30 * time.Second
+	defaultMode               = "operator"
 )
 
 var defaultKubeConfigPath = filepath.Join(homedir.HomeDir(), ".kube", "config")
@@ -40,7 +41,7 @@ func GetFlagSet(errorHandling pflag.ErrorHandling) *pflag.FlagSet {
 	flagSet.String(kubeConfigPathFlagName, defaultKubeConfigPath, "absolute path to the KubeconfigPath file.")
 	flagSet.Duration(heartbeatIntervalFlagName, defaultHeartbeatInterval, "The interval to use for sending a heartbeat. Setting it to 0 disables the heartbeat.")
 	flagSet.String(nameFlagName, opampBridgeName, "The name of the bridge to use for querying managed collectors.")
-	flagSet.String(modeFlagName, "", `Operating mode: "operator" (default, uses CRDs) or "standalone" (manages ConfigMaps for Deployments/DaemonSets).`)
+	flagSet.String(modeFlagName, defaultMode, `Operating mode: "operator" (default, uses CRDs) or "standalone" (manages ConfigMaps for Deployments/DaemonSets).`)
 	zapFlagSet := flag.NewFlagSet("", flag.ErrorHandling(errorHandling))
 	zapCmdLineOpts.BindFlags(zapFlagSet)
 	flagSet.AddGoFlagSet(zapFlagSet)

cmd/operator-opamp-bridge/internal/workload/client.go

@@ -14,7 +14,10 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/yaml"
 
 	"github.com/open-telemetry/opentelemetry-operator/cmd/operator-opamp-bridge/internal/operator"
 )
@@ -24,23 +27,16 @@ const (
 	managedByValue    = "opamp-bridge"
 	restartAnnotation = "kubectl.kubernetes.io/restartedAt"
 
-	// ManagedLabelKey is the label that opts a Deployment or DaemonSet into standalone OpAMP ConfigMap management.
-	// Workloads must carry this label with value "true" to be discovered by the bridge.
+	// ManagedLabelKey opts a Deployment or DaemonSet into standalone OpAMP ConfigMap management.
+	// The label value must be the name of the ConfigMap to manage (e.g. "otel-agent-config").
 	ManagedLabelKey = "opentelemetry.io/opamp-standalone-configmap"
-
-	// AnnotationConfigMapName is a required annotation specifying the ConfigMap the bridge writes collector config into.
-	AnnotationConfigMapName = "opentelemetry.io/opamp-standalone-configmap-name"
-
-	// AnnotationConfigMapKey is an optional annotation specifying the key within the ConfigMap.
-	// Defaults to the workload name if not set.
-	AnnotationConfigMapKey = "opentelemetry.io/opamp-standalone-configmap-key"
 )
 
 var _ operator.ConfigApplier = &Client{}
 
 // Client implements operator.ConfigApplier for standalone Deployment/DaemonSet
-// workloads. It discovers targets dynamically via the ManagedLabelKey label and
-// reads ConfigMap details from annotations on the workload.
+// workloads. It discovers targets via the ManagedLabelKey label whose value is
+// the name of the ConfigMap to manage.
 type Client struct {
 	log       logr.Logger
 	k8sClient client.Client
@@ -55,39 +51,28 @@ func NewClient(name string, log logr.Logger, c client.Client) *Client {
 	}
 }
 
-// resolveTarget looks up a Deployment or DaemonSet by name/namespace and returns
-// its ConfigMap name and key from annotations.
-func (c *Client) resolveTarget(ctx context.Context, name, namespace string) (configMapName, configMapKey string, workloadKind string, workloadName string, err error) {
-	// Try Deployment first
+// resolveTarget looks up a Deployment or DaemonSet by name/namespace and reads
+// the target ConfigMap name from the ManagedLabelKey label value.
+func (c *Client) resolveTarget(ctx context.Context, name, namespace string) (configMapName, workloadKind, workloadName string, err error) {
 	deploy := &appsv1.Deployment{}
 	if getErr := c.k8sClient.Get(ctx, client.ObjectKey{Name: name, Namespace: namespace}, deploy); getErr == nil {
-		if deploy.Labels[ManagedLabelKey] != "true" {
-			return "", "", "", "", fmt.Errorf("workload %s/%s does not have label %s=true", namespace, name, ManagedLabelKey)
+		cmName := deploy.Labels[ManagedLabelKey]
+		if cmName == "" {
+			return "", "", "", fmt.Errorf("workload %s/%s has label %s but its value (ConfigMap name) is empty", namespace, name, ManagedLabelKey)
 		}
-		cmName, cmKey := configMapDetailsFromAnnotations(deploy.Annotations, name)
-		return cmName, cmKey, "Deployment", name, nil
+		return cmName, "Deployment", name, nil
 	}
 
-	// Try DaemonSet
 	ds := &appsv1.DaemonSet{}
 	if getErr := c.k8sClient.Get(ctx, client.ObjectKey{Name: name, Namespace: namespace}, ds); getErr == nil {
-		if ds.Labels[ManagedLabelKey] != "true" {
-			return "", "", "", "", fmt.Errorf("workload %s/%s does not have label %s=true", namespace, name, ManagedLabelKey)
+		cmName := ds.Labels[ManagedLabelKey]
+		if cmName == "" {
+			return "", "", "", fmt.Errorf("workload %s/%s has label %s but its value (ConfigMap name) is empty", namespace, name, ManagedLabelKey)
 		}
-		cmName, cmKey := configMapDetailsFromAnnotations(ds.Annotations, name)
-		return cmName, cmKey, "DaemonSet", name, nil
+		return cmName, "DaemonSet", name, nil
 	}
 
-	return "", "", "", "", fmt.Errorf("no managed Deployment or DaemonSet named %s found in namespace %s", name, namespace)
-}
-
-func configMapDetailsFromAnnotations(annotations map[string]string, defaultKey string) (cmName, cmKey string) {
-	cmName = annotations[AnnotationConfigMapName]
-	cmKey = annotations[AnnotationConfigMapKey]
-	if cmKey == "" {
-		cmKey = defaultKey
-	}
-	return
+	return "", "", "", fmt.Errorf("no managed Deployment or DaemonSet named %s found in namespace %s", name, namespace)
 }
 
 func (c *Client) Apply(name, namespace string, configFile *protobufs.AgentConfigFile) error {
@@ -97,12 +82,14 @@ func (c *Client) Apply(name, namespace string, configFile *protobufs.AgentConfig
 
 	ctx := context.Background()
 
-	configMapName, configMapKey, workloadKind, workloadName, err := c.resolveTarget(ctx, name, namespace)
+	configMapName, workloadKind, workloadName, err := c.resolveTarget(ctx, name, namespace)
 	if err != nil {
 		return err
 	}
-	if configMapName == "" {
-		return fmt.Errorf("workload %s/%s is missing annotation %s", namespace, name, AnnotationConfigMapName)
+
+	var newData map[string]string
+	if err := yaml.Unmarshal(configFile.Body, &newData); err != nil {
+		return fmt.Errorf("failed to unmarshal config body into ConfigMap data: %w", err)
 	}
 
 	c.log.Info("Applying config", "name", name, "namespace", namespace, "configMap", configMapName)
@@ -122,20 +109,15 @@ func (c *Client) Apply(name, namespace string, configFile *protobufs.AgentConfig
 					managedByLabel: managedByValue,
 				},
 			},
-			Data: map[string]string{
-				configMapKey: string(configFile.Body),
-			},
+			Data: newData,
 		}
 		if createErr := c.k8sClient.Create(ctx, cm); createErr != nil {
 			return fmt.Errorf("failed to create ConfigMap %s/%s: %w", namespace, configMapName, createErr)
 		}
 	} else if err != nil {
 		return fmt.Errorf("failed to get ConfigMap %s/%s: %w", namespace, configMapName, err)
 	} else {
-		if cm.Data == nil {
-			cm.Data = map[string]string{}
-		}
-		cm.Data[configMapKey] = string(configFile.Body)
+		cm.Data = newData
 		if updateErr := c.k8sClient.Update(ctx, cm); updateErr != nil {
 			return fmt.Errorf("failed to update ConfigMap %s/%s: %w", namespace, configMapName, updateErr)
 		}
@@ -147,14 +129,11 @@ func (c *Client) Apply(name, namespace string, configFile *protobufs.AgentConfig
 func (c *Client) Delete(name, namespace string) error {
 	ctx := context.Background()
 
-	configMapName, _, _, _, err := c.resolveTarget(ctx, name, namespace)
+	configMapName, _, _, err := c.resolveTarget(ctx, name, namespace)
 	if err != nil {
 		// Workload no longer exists or isn't managed; nothing to delete.
 		return nil
 	}
-	if configMapName == "" {
-		return nil
-	}
 
 	cm := &v1.ConfigMap{}
 	err = c.k8sClient.Get(ctx, client.ObjectKey{
@@ -174,10 +153,15 @@ func (c *Client) ListInstances() ([]operator.CollectorInstance, error) {
 	ctx := context.Background()
 	var result []operator.CollectorInstance
 
-	managedLabel := client.MatchingLabels{ManagedLabelKey: "true"}
+	// Match any workload that has the label set to a non-empty value.
+	labelReq, err := labels.NewRequirement(ManagedLabelKey, selection.Exists, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build label requirement: %w", err)
+	}
+	managedSelector := client.MatchingLabelsSelector{Selector: labels.NewSelector().Add(*labelReq)}
 
 	deployList := &appsv1.DeploymentList{}
-	if err := c.k8sClient.List(ctx, deployList, managedLabel); err != nil {
+	if err := c.k8sClient.List(ctx, deployList, managedSelector); err != nil {
 		return nil, fmt.Errorf("failed to list managed Deployments: %w", err)
 	}
 	for i := range deployList.Items {
@@ -190,7 +174,7 @@ func (c *Client) ListInstances() ([]operator.CollectorInstance, error) {
 	}
 
 	dsList := &appsv1.DaemonSetList{}
-	if err := c.k8sClient.List(ctx, dsList, managedLabel); err != nil {
+	if err := c.k8sClient.List(ctx, dsList, managedSelector); err != nil {
 		return nil, fmt.Errorf("failed to list managed DaemonSets: %w", err)
 	}
 	for i := range dsList.Items {
@@ -213,21 +197,17 @@ func (c *Client) GetCollectorPods(selectorLabels map[string]string, namespace st
 }
 
 func (c *Client) getDeploymentInstance(ctx context.Context, deploy *appsv1.Deployment) (*standaloneCollectorInstance, error) {
-	cmName, cmKey := configMapDetailsFromAnnotations(deploy.Annotations, deploy.Name)
+	cmName := deploy.Labels[ManagedLabelKey]
 	if cmName == "" {
-		return nil, fmt.Errorf("missing annotation %s on Deployment %s/%s", AnnotationConfigMapName, deploy.Namespace, deploy.Name)
+		return nil, fmt.Errorf("Deployment %s/%s has label %s but its value (ConfigMap name) is empty", deploy.Namespace, deploy.Name, ManagedLabelKey)
 	}
 
 	var desired int32
 	if deploy.Spec.Replicas != nil {
 		desired = *deploy.Spec.Replicas
 	}
 
-	var configBody []byte
-	cm := &v1.ConfigMap{}
-	if err := c.k8sClient.Get(ctx, client.ObjectKey{Name: cmName, Namespace: deploy.Namespace}, cm); err == nil && cm.Data != nil {
-		configBody = []byte(cm.Data[cmKey])
-	}
+	configBody := c.readConfigMapData(ctx, cmName, deploy.Namespace)
 
 	return &standaloneCollectorInstance{
 		name:           deploy.Name,
@@ -240,16 +220,12 @@ func (c *Client) getDeploymentInstance(ctx context.Context, deploy *appsv1.Deplo
 }
 
 func (c *Client) getDaemonSetInstance(ctx context.Context, ds *appsv1.DaemonSet) (*standaloneCollectorInstance, error) {
-	cmName, cmKey := configMapDetailsFromAnnotations(ds.Annotations, ds.Name)
+	cmName := ds.Labels[ManagedLabelKey]
 	if cmName == "" {
-		return nil, fmt.Errorf("missing annotation %s on DaemonSet %s/%s", AnnotationConfigMapName, ds.Namespace, ds.Name)
+		return nil, fmt.Errorf("DaemonSet %s/%s has label %s but its value (ConfigMap name) is empty", ds.Namespace, ds.Name, ManagedLabelKey)
 	}
 
-	var configBody []byte
-	cm := &v1.ConfigMap{}
-	if err := c.k8sClient.Get(ctx, client.ObjectKey{Name: cmName, Namespace: ds.Namespace}, cm); err == nil && cm.Data != nil {
-		configBody = []byte(cm.Data[cmKey])
-	}
+	configBody := c.readConfigMapData(ctx, cmName, ds.Namespace)
 
 	return &standaloneCollectorInstance{
 		name:           ds.Name,
@@ -261,6 +237,22 @@ func (c *Client) getDaemonSetInstance(ctx context.Context, ds *appsv1.DaemonSet)
 	}, nil
 }
 
+// readConfigMapData fetches a ConfigMap and serializes all its data keys into
+// a single YAML document (map[string]string). Returns nil if the ConfigMap
+// does not exist or has no data.
+func (c *Client) readConfigMapData(ctx context.Context, cmName, namespace string) []byte {
+	cm := &v1.ConfigMap{}
+	if err := c.k8sClient.Get(ctx, client.ObjectKey{Name: cmName, Namespace: namespace}, cm); err != nil || len(cm.Data) == 0 {
+		return nil
+	}
+	b, err := yaml.Marshal(cm.Data)
+	if err != nil {
+		c.log.Error(err, "Failed to marshal ConfigMap data", "configMap", cmName, "namespace", namespace)
+		return nil
+	}
+	return b
+}
+
 func (c *Client) triggerRollout(ctx context.Context, kind, name, namespace string) error {
 	restartVal := time.Now().Format(time.RFC3339)