opamp-bridge: add standalone mode for ConfigMaps

Introduces a new standalone mode that allows managing collector config from a remote server without requiring Operator CRDs.
Adds the standalone client, a plain collector instance type, Kubernetes RBAC and deployment manifests, and a --mode flag to select between operator (default) and standalone at startup.

Author: owais

.chloggen/standalone-mode.yaml

@@ -0,0 +1,18 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. collector, target allocator, auto-instrumentation, opamp, github action)
+component: opamp-bridge
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: OpAMP Bridge standalone mode 
+
+# One or more tracking issues related to the change
+issues: [4913]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Standalone mode for OpAMP Bridge allows users to manage collector configuration from a remote
+  OpAMP server without the need to deploy full Otel Operator.

Makefile

@@ -333,6 +333,26 @@ deploy: install-gateway-api-crds set-image-controller
 undeploy: set-image-controller
 	$(KUSTOMIZE) build config/default | kubectl delete --ignore-not-found=$(ignore-not-found) -f -
 
+##@ Standalone OpAMP Bridge (no operator / CRDs required)
+
+# Deploy the standalone OpAMP bridge into the current Kubernetes context.
+# Does not require the operator, CRDs, or cert-manager.
+.PHONY: deploy-standalone-bridge
+deploy-standalone-bridge: kustomize
+	cd config/standalone-bridge && $(KUSTOMIZE) edit set image operator-opamp-bridge=${OPERATOROPAMPBRIDGE_IMG}
+	$(KUSTOMIZE) build config/standalone-bridge | kubectl apply -f -
+	kubectl rollout status deployment/otel-opamp-bridge-standalone -n opentelemetry-opamp-bridge --timeout=120s
+
+# Undeploy the standalone OpAMP bridge from the current Kubernetes context.
+.PHONY: undeploy-standalone-bridge
+undeploy-standalone-bridge: kustomize
+	$(KUSTOMIZE) build config/standalone-bridge | kubectl delete --ignore-not-found=true -f -
+
+# Build, load, and deploy the standalone bridge to a kind cluster.
+# Assumes a kind cluster is already running (use start-kind first).
+.PHONY: deploy-standalone-bridge-kind
+deploy-standalone-bridge-kind: load-image-operator-opamp-bridge deploy-standalone-bridge
+
 # Generates the released manifests
 .PHONY: release-artifacts
 release-artifacts: set-image-controller

cmd/operator-opamp-bridge/internal/agent/agent.go

@@ -17,9 +17,7 @@ import (
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
 	"k8s.io/utils/clock"
-	"sigs.k8s.io/yaml"
 
-	"github.com/open-telemetry/opentelemetry-operator/apis/v1beta1"
 	"github.com/open-telemetry/opentelemetry-operator/cmd/operator-opamp-bridge/internal/config"
 	"github.com/open-telemetry/opentelemetry-operator/cmd/operator-opamp-bridge/internal/metrics"
 	"github.com/open-telemetry/opentelemetry-operator/cmd/operator-opamp-bridge/internal/operator"
@@ -118,21 +116,20 @@ func (agent *Agent) generateCollectorPoolHealth() (map[string]*protobufs.Compone
 	proxiesUsed := make(map[uuid.UUID]struct{}, len(agentsByHostName))
 	for _, col := range cols {
 		key := newKubeResourceKey(col.GetNamespace(), col.GetName())
-		podMap, err := agent.generateCollectorHealth(agent.getCollectorSelector(col), col.GetNamespace())
+		podMap, err := agent.generateCollectorHealth(col.GetSelectorLabels(), col.GetNamespace())
 		if err != nil {
 			return nil, err
 		}
 		isPoolHealthy := true
 		for podName, pod := range podMap {
-			// we need to remove the prefix here as we don't have the namespace from the hostname.
 			podNameWithoutNamespace := strings.TrimPrefix(podName, col.GetNamespace()+"/")
 			isPoolHealthy = isPoolHealthy && pod.Healthy
 			if uid, ok := agentsByHostName[podNameWithoutNamespace]; ok {
 				podMap[podName].ComponentHealthMap[uid.String()] = proxyHealth[uid]
 				proxiesUsed[uid] = struct{}{}
 			}
 		}
-		podStartTime, err := timeToUnixNanoUnsigned(col.ObjectMeta.GetCreationTimestamp().Time)
+		podStartTime, err := timeToUnixNanoUnsigned(col.GetCreationTimestamp())
 		if err != nil {
 			return nil, err
 		}
@@ -143,7 +140,7 @@ func (agent *Agent) generateCollectorPoolHealth() (map[string]*protobufs.Compone
 		healthMap[key.String()] = &protobufs.ComponentHealth{
 			StartTimeUnixNano:  podStartTime,
 			StatusTimeUnixNano: statusTime,
-			Status:             col.Status.Scale.StatusReplicas,
+			Status:             col.GetStatusReplicas(),
 			ComponentHealthMap: podMap,
 			Healthy:            isPoolHealthy,
 		}
@@ -157,28 +154,6 @@ func (agent *Agent) generateCollectorPoolHealth() (map[string]*protobufs.Compone
 	return healthMap, nil
 }
 
-// getCollectorSelector destructures the collectors scale selector if present, it uses the labelmap from the operator.
-func (*Agent) getCollectorSelector(col v1beta1.OpenTelemetryCollector) map[string]string {
-	if col.Status.Scale.Selector != "" {
-		selMap := map[string]string{}
-		for kvPair := range strings.SplitSeq(col.Status.Scale.Selector, ",") {
-			kv := strings.Split(kvPair, "=")
-			// skip malformed pairs
-			if len(kv) != 2 {
-				continue
-			}
-			selMap[kv[0]] = kv[1]
-		}
-		return selMap
-	}
-	return map[string]string{
-		"app.kubernetes.io/managed-by": "opentelemetry-operator",
-		"app.kubernetes.io/instance":   fmt.Sprintf("%s.%s", col.GetNamespace(), col.GetName()),
-		"app.kubernetes.io/part-of":    "opentelemetry",
-		"app.kubernetes.io/component":  "opentelemetry-collector",
-	}
-}
-
 func (agent *Agent) generateCollectorHealth(selectorLabels map[string]string, namespace string) (map[string]*protobufs.ComponentHealth, error) {
 	statusTime, err := agent.getCurrentTimeUnixNano()
 	if err != nil {
@@ -348,15 +323,14 @@ func (agent *Agent) getEffectiveConfig(context.Context) (*protobufs.EffectiveCon
 	}
 	instanceMap := map[string]*protobufs.AgentConfigFile{}
 	for _, instance := range instances {
-		col := instance
-		marshaled, err := yaml.Marshal(&col)
-		if err != nil {
-			agent.logger.Error(err, "failed to marhsal config")
-			return nil, err
+		body := instance.GetEffectiveConfig()
+		if body == nil {
+			agent.logger.Error(errors.New("nil effective config"), "failed to get effective config",
+				"name", instance.GetName(), "namespace", instance.GetNamespace())
+			continue
 		}
-		mapKey := newKubeResourceKey(instance.GetNamespace(), instance.GetName())
-		instanceMap[mapKey.String()] = &protobufs.AgentConfigFile{
-			Body:        marshaled,
+		instanceMap[instance.GetConfigMapKey()] = &protobufs.AgentConfigFile{
+			Body:        body,
 			ContentType: "yaml",
 		}
 	}
@@ -390,11 +364,11 @@ func (agent *Agent) initMeter(settings *protobufs.TelemetryConnectionSettings) {
 
 // applyRemoteConfig receives a remote configuration from a remote server of the following form:
 //
-//	map[name/namespace] -> collector CRD spec
+//	map[resource key] -> AgentConfigFile body
 //
-// For every key in the received remote configuration, the agent attempts to apply it to the connected
-// Kubernetes cluster. If an agent fails to apply a collector CRD, it will continue to the next entry. The agent will
-// store the received configuration hash regardless of application status as per the OpAMP spec.
+// For every key in the received remote configuration, the agent attempts to apply it via the configured
+// applier. If an entry fails to apply, the agent continues to the next entry. The agent stores the
+// received configuration hash regardless of application status, as per the OpAMP spec.
 //
 // INVARIANT: The caller must verify that config isn't nil _and_ the configuration has changed between calls.
 func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (*protobufs.RemoteConfigStatus, error) {
@@ -409,6 +383,10 @@ func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (*pro
 			errs = append(errs, err)
 			continue
 		}
+		if err = agent.validateConfigMapKey(colKey); err != nil {
+			errs = append(errs, err)
+			continue
+		}
 		err = agent.applier.Apply(colKey.name, colKey.namespace, file)
 		if err != nil {
 			errs = append(errs, err)
@@ -417,9 +395,9 @@ func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (*pro
 		agent.appliedKeys[colKey] = true
 	}
 	// Check if anything was deleted
-	for collectorKey := range agent.appliedKeys {
-		if _, ok := config.Config.GetConfigMap()[collectorKey.String()]; !ok {
-			err := agent.applier.Delete(collectorKey.name, collectorKey.namespace)
+	for key := range agent.appliedKeys {
+		if _, ok := config.Config.GetConfigMap()[key.String()]; !ok {
+			err := agent.applier.Delete(key.name, key.namespace)
 			if err != nil {
 				errs = append(errs, err)
 			}
@@ -440,6 +418,19 @@ func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (*pro
 	}, nil
 }
 
+func (agent *Agent) validateConfigMapKey(key kubeResourceKey) error {
+	if agent.config.IsStandaloneMode() {
+		if key.kind != "configmap" {
+			return errors.New("standalone config key must use configmap kind")
+		}
+		return nil
+	}
+	if key.kind != "" {
+		return errors.New("operator config key must not include kind")
+	}
+	return nil
+}
+
 // Shutdown will stop the OpAMP client gracefully.
 func (agent *Agent) Shutdown() {
 	agent.logger.V(3).Info("Agent shutting down...")

cmd/operator-opamp-bridge/internal/agent/kube_resource_key.go

@@ -12,21 +12,33 @@ import (
 type kubeResourceKey struct {
 	name      string
 	namespace string
+	kind      string
 }
 
 func newKubeResourceKey(namespace, name string) kubeResourceKey {
 	return kubeResourceKey{name: name, namespace: namespace}
 }
 
+func newKubeResourceKeyWithKind(namespace, name, kind string) kubeResourceKey {
+	return kubeResourceKey{name: name, namespace: namespace, kind: kind}
+}
+
 func kubeResourceFromKey(key string) (kubeResourceKey, error) {
 	s := strings.Split(key, "/")
-	// We expect map keys to be of the form name/namespace
-	if len(s) != 2 {
+	// We expect map keys to be of the form namespace/name or kind/namespace/name.
+	switch len(s) {
+	case 2:
+		return newKubeResourceKey(s[0], s[1]), nil
+	case 3:
+		return newKubeResourceKeyWithKind(s[1], s[2], s[0]), nil
+	default:
 		return kubeResourceKey{}, errors.New("invalid key")
 	}
-	return newKubeResourceKey(s[0], s[1]), nil
 }
 
 func (k kubeResourceKey) String() string {
+	if k.kind != "" {
+		return fmt.Sprintf("%s/%s/%s", k.kind, k.namespace, k.name)
+	}
 	return fmt.Sprintf("%s/%s", k.namespace, k.name)
 }

cmd/operator-opamp-bridge/internal/agent/kube_resource_key_test.go

@@ -31,6 +31,18 @@ func Test_collectorKeyFromKey(t *testing.T) {
 			},
 			wantErr: assert.NoError,
 		},
+		{
+			name: "with kind",
+			args: args{
+				key: "configmap/namespace/good",
+			},
+			want: kubeResourceKey{
+				name:      "good",
+				namespace: "namespace",
+				kind:      "configmap",
+			},
+			wantErr: assert.NoError,
+		},
 		{
 			name: "unable to get key",
 			args: args{
@@ -42,7 +54,7 @@ func Test_collectorKeyFromKey(t *testing.T) {
 		{
 			name: "too many slashes",
 			args: args{
-				key: "too/many/slashes",
+				key: "too/many/slashes/here",
 			},
 			want:    kubeResourceKey{},
 			wantErr: assert.Error,
@@ -63,6 +75,7 @@ func Test_collectorKey_String(t *testing.T) {
 	type fields struct {
 		name      string
 		namespace string
+		kind      string
 	}
 	tests := []struct {
 		name   string
@@ -77,10 +90,22 @@ func Test_collectorKey_String(t *testing.T) {
 			},
 			want: "namespace/good",
 		},
+		{
+			name: "can make a key with kind",
+			fields: fields{
+				name:      "good",
+				namespace: "namespace",
+				kind:      "configmap",
+			},
+			want: "configmap/namespace/good",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			k := newKubeResourceKey(tt.fields.namespace, tt.fields.name)
+			if tt.fields.kind != "" {
+				k = newKubeResourceKeyWithKind(tt.fields.namespace, tt.fields.name, tt.fields.kind)
+			}
 			assert.Equalf(t, tt.want, k.String(), "String()")
 		})
 	}

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

@@ -91,6 +91,10 @@ type Config struct {
 	HeartbeatInterval time.Duration       `yaml:"heartbeatInterval,omitempty"`
 	Name              string              `yaml:"name,omitempty"`
 	AgentDescription  AgentDescription    `yaml:"description,omitempty"`
+
+	// Mode selects the operating mode: "operator" (default) uses OpenTelemetryCollector CRDs,
+	// "standalone" discovers Deployments/DaemonSets by label and manages their ConfigMaps.
+	Mode string `yaml:"mode,omitempty"`
 }
 
 // AgentDescription is copied from the OpAMP Extension in the collector.
@@ -225,15 +229,25 @@ func (c *Config) RemoteConfigEnabled() bool {
 }
 
 func (c *Config) GetKubernetesClient() (client.Client, error) {
-	err := schemeBuilder.AddToScheme(scheme.Scheme)
-	if err != nil {
-		return nil, err
+	if c.Mode != "standalone" {
+		err := schemeBuilder.AddToScheme(scheme.Scheme)
+		if err != nil {
+			return nil, err
+		}
 	}
 	return client.New(c.ClusterConfig, client.Options{
 		Scheme: scheme.Scheme,
 	})
 }
 
+func (c *Config) IsStandaloneMode() bool {
+	return c.Mode == "standalone"
+}
+
+func (c *Config) GetRestConfig() *rest.Config {
+	return c.ClusterConfig
+}
+
 func Load(logger logr.Logger, args []string) (*Config, error) {
 	flagSet := GetFlagSet(pflag.ExitOnError)
 	err := flagSet.Parse(args)
@@ -300,6 +314,11 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error {
 	} else if changed {
 		target.Name = name
 	}
+	if mode, changed, err := getMode(flagSet); err != nil {
+		return err
+	} else if changed {
+		target.Mode = mode
+	}
 	return nil
 }
 

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

@@ -24,7 +24,9 @@ const (
 	kubeConfigPathFlagName    = "kubeconfig-path"
 	heartbeatIntervalFlagName = "heartbeat-interval"
 	nameFlagName              = "name"
+	modeFlagName              = "mode"
 	defaultHeartbeatInterval  = 30 * time.Second
+	defaultMode               = "operator"
 )
 
 var defaultKubeConfigPath = filepath.Join(homedir.HomeDir(), ".kube", "config")
@@ -39,6 +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, 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)
@@ -65,6 +68,10 @@ func getListenAddr(flagSet *pflag.FlagSet) (value string, changed bool, err erro
 	return getFlagValueAndChanged[string](flagSet, listenAddrFlagName)
 }
 
+func getMode(flagSet *pflag.FlagSet) (value string, changed bool, err error) {
+	return getFlagValueAndChanged[string](flagSet, modeFlagName)
+}
+
 func getFlagValueAndChanged[T any](flagSet *pflag.FlagSet, flagName string) (value T, changed bool, err error) {
 	var zero T
 	if changed = flagSet.Changed(flagName); !changed {