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,15 @@ 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(fmt.Errorf("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,
+			Body:        body,
 			ContentType: "yaml",
 		}
 	}
@@ -390,11 +365,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[name/namespace] -> 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) {

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 {

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

@@ -29,17 +29,14 @@ const (
 )
 
 type ConfigApplier interface {
-	// Apply receives a name and namespace to apply an OpenTelemetryCollector CRD that is contained in the configmap.
+	// Apply receives a name and namespace to apply a collector configuration.
 	Apply(name, namespace string, configmap *protobufs.AgentConfigFile) error
 
-	// Delete attempts to delete an OpenTelemetryCollector object given a name and namespace.
+	// Delete attempts to delete a collector resource given a name and namespace.
 	Delete(name, namespace string) error
 
-	// ListInstances retrieves all OpenTelemetryCollector CRDs created by the operator-opamp-bridge agent.
-	ListInstances() ([]v1beta1.OpenTelemetryCollector, error)
-
-	// GetInstance retrieves an OpenTelemetryCollector CRD given a name and namespace.
-	GetInstance(name, namespace string) (*v1beta1.OpenTelemetryCollector, error)
+	// ListInstances retrieves all collector instances managed by the bridge.
+	ListInstances() ([]CollectorInstance, error)
 
 	// GetCollectorPods retrieves all pods that match the given collector's selector labels and namespace.
 	GetCollectorPods(selectorLabels map[string]string, namespace string) (*v1.PodList, error)
@@ -207,7 +204,19 @@ func (c Client) Delete(name, namespace string) error {
 	return c.k8sClient.Delete(ctx, &result)
 }
 
-func (c Client) ListInstances() ([]v1beta1.OpenTelemetryCollector, error) {
+func (c Client) ListInstances() ([]CollectorInstance, error) {
+	rawInstances, err := c.listRawInstances()
+	if err != nil {
+		return nil, err
+	}
+	result := make([]CollectorInstance, len(rawInstances))
+	for i := range rawInstances {
+		result[i] = CRDInstance{Col: rawInstances[i]}
+	}
+	return result, nil
+}
+
+func (c Client) listRawInstances() ([]v1beta1.OpenTelemetryCollector, error) {
 	ctx := context.Background()
 
 	var instances []v1beta1.OpenTelemetryCollector

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

@@ -246,8 +246,12 @@ func TestClient_ApplyUpdate(t *testing.T) {
 	allInstances, err = c.ListInstances()
 	require.NoError(t, err, "Should be able to list all collectors")
 	assert.Len(t, allInstances, 2)
-	assert.Contains(t, allInstances, reportingCol)
-	assert.Contains(t, allInstances, *updatedInstance)
+	instanceNames := make([]string, len(allInstances))
+	for i, inst := range allInstances {
+		instanceNames[i] = inst.GetNamespace() + "/" + inst.GetName()
+	}
+	assert.Contains(t, instanceNames, reportingCol.GetNamespace()+"/"+reportingCol.GetName())
+	assert.Contains(t, instanceNames, updatedInstance.GetNamespace()+"/"+updatedInstance.GetName())
 }
 
 func TestClient_Delete(t *testing.T) {

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

@@ -0,0 +1,65 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package operator
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"sigs.k8s.io/yaml"
+
+	"github.com/open-telemetry/opentelemetry-operator/apis/v1beta1"
+)
+
+var _ CollectorInstance = CRDInstance{}
+
+// CRDInstance wraps an OpenTelemetryCollector CRD to implement CollectorInstance.
+type CRDInstance struct {
+	Col v1beta1.OpenTelemetryCollector
+}
+
+func (c CRDInstance) GetName() string {
+	return c.Col.GetName()
+}
+
+func (c CRDInstance) GetNamespace() string {
+	return c.Col.GetNamespace()
+}
+
+func (c CRDInstance) GetCreationTimestamp() time.Time {
+	return c.Col.GetCreationTimestamp().Time
+}
+
+func (c CRDInstance) GetSelectorLabels() map[string]string {
+	if c.Col.Status.Scale.Selector != "" {
+		selMap := map[string]string{}
+		for kvPair := range strings.SplitSeq(c.Col.Status.Scale.Selector, ",") {
+			kv := strings.Split(kvPair, "=")
+			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", c.Col.GetNamespace(), c.Col.GetName()),
+		"app.kubernetes.io/part-of":    "opentelemetry",
+		"app.kubernetes.io/component":  "opentelemetry-collector",
+	}
+}
+
+func (c CRDInstance) GetStatusReplicas() string {
+	return c.Col.Status.Scale.StatusReplicas
+}
+
+func (c CRDInstance) GetEffectiveConfig() []byte {
+	marshaled, err := yaml.Marshal(&c.Col)
+	if err != nil {
+		return nil
+	}
+	return marshaled
+}

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

@@ -0,0 +1,17 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package operator
+
+import "time"
+
+// CollectorInstance represents a collector managed by the bridge, abstracting
+// the underlying Kubernetes resource (CRD or Deployment/DaemonSet).
+type CollectorInstance interface {
+	GetName() string
+	GetNamespace() string
+	GetCreationTimestamp() time.Time
+	GetSelectorLabels() map[string]string
+	GetStatusReplicas() string
+	GetEffectiveConfig() []byte
+}