[receiver/receivercreator] Ensure unique receiver names (#12670)

Adding a feature - Currently receiver creator-generated receiver instances will use the same ComponentID by their endpoint (observer.Endpoint.Target), which can cause collisions when multiple observer.Endpoint instances have the same target value. These changes provide EndpointID suffixes to their names to prevent this.

receiver/receivercreator/config.go

@@ -39,7 +39,8 @@ type receiverConfig struct {
 	id config.ComponentID
 	// config is the map configured by the user in the config file. It is the contents of the map from
 	// the "config" section. The keys and values are arbitrarily configured by the user.
-	config userConfigMap
+	config     userConfigMap
+	endpointID observer.EndpointID
 }
 
 // userConfigMap is an arbitrary map of string keys to arbitrary values as specified by the user
@@ -71,8 +72,9 @@ func newReceiverTemplate(name string, cfg userConfigMap) (receiverTemplate, erro
 
 	return receiverTemplate{
 		receiverConfig: receiverConfig{
-			id:     id,
-			config: cfg,
+			id:         id,
+			config:     cfg,
+			endpointID: observer.EndpointID("endpoint.id"),
 		},
 	}, nil
 }

receiver/receivercreator/observerhandler.go

@@ -142,8 +142,9 @@ func (obs *observerHandler) OnAdd(added []observer.Endpoint) {
 
 			rcvr, err := obs.runner.start(
 				receiverConfig{
-					id:     template.id,
-					config: resolvedConfig,
+					id:         template.id,
+					config:     resolvedConfig,
+					endpointID: e.ID,
 				},
 				resolvedDiscoveredConfig,
 				resourceEnhancer,

receiver/receivercreator/observerhandler_test.go

@@ -49,7 +49,8 @@ var _ runner = (*mockRunner)(nil)
 
 func TestOnAdd(t *testing.T) {
 	runner := &mockRunner{}
-	rcvrCfg := receiverConfig{id: config.NewComponentIDWithName("name", "1"), config: userConfigMap{"foo": "bar"}}
+
+	rcvrCfg := receiverConfig{id: config.NewComponentIDWithName("name", "1"), config: userConfigMap{"foo": "bar"}, endpointID: portEndpoint.ID}
 	cfg := createDefaultConfig().(*Config)
 	cfg.receiverTemplates = map[string]receiverTemplate{
 		"name/1": {rcvrCfg, "", map[string]interface{}{}, newRuleOrPanic(`type == "port"`)},
@@ -99,7 +100,7 @@ func TestOnRemove(t *testing.T) {
 
 func TestOnChange(t *testing.T) {
 	runner := &mockRunner{}
-	rcvrCfg := receiverConfig{id: config.NewComponentIDWithName("name", "1"), config: userConfigMap{"foo": "bar"}}
+	rcvrCfg := receiverConfig{id: config.NewComponentIDWithName("name", "1"), config: userConfigMap{"foo": "bar"}, endpointID: portEndpoint.ID}
 	oldRcvr := &nopWithEndpointReceiver{}
 	newRcvr := &nopWithEndpointReceiver{}
 	cfg := createDefaultConfig().(*Config)
@@ -135,7 +136,7 @@ func TestDynamicConfig(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	cfg.receiverTemplates = map[string]receiverTemplate{
 		"name/1": {
-			receiverConfig: receiverConfig{id: config.NewComponentIDWithName("name", "1"), config: userConfigMap{"endpoint": "`endpoint`:6379"}},
+			receiverConfig: receiverConfig{id: config.NewComponentIDWithName("name", "1"), config: userConfigMap{"endpoint": "`endpoint`:6379"}, endpointID: podEndpoint.ID},
 			Rule:           `type == "pod"`,
 			rule:           newRuleOrPanic("type == \"pod\""),
 		},
@@ -149,8 +150,9 @@ func TestDynamicConfig(t *testing.T) {
 	runner.On(
 		"start",
 		receiverConfig{
-			id:     config.NewComponentIDWithName("name", "1"),
-			config: userConfigMap{endpointConfigKey: "localhost:6379"},
+			id:         config.NewComponentIDWithName("name", "1"),
+			config:     userConfigMap{endpointConfigKey: "localhost:6379"},
+			endpointID: podEndpoint.ID,
 		},
 		userConfigMap{},
 		mock.IsType(&resourceEnhancer{}),

receiver/receivercreator/receiver.go

@@ -75,7 +75,8 @@ func (rc *receiverCreator) Start(_ context.Context, host component.Host) error {
 			params:      rc.params,
 			idNamespace: rc.cfg.ID(),
 			host:        &loggingHost{host, rc.params.Logger},
-		}}
+		},
+	}
 
 	observers := map[config.Type]observer.Observable{}
 

receiver/receivercreator/runner.go

@@ -99,9 +99,8 @@ func (run *receiverRunner) loadRuntimeReceiverConfig(
 	if err := config.UnmarshalReceiver(mergedConfig, receiverCfg); err != nil {
 		return nil, fmt.Errorf("failed to load template config: %w", err)
 	}
-	// Sets dynamically created receiver to something like receiver_creator/1/redis{endpoint="localhost:6380"}.
-	// TODO: Need to make sure this is unique (just endpoint is probably not totally sufficient).
-	receiverCfg.SetIDName(fmt.Sprintf("%s/%s{endpoint=%q}", receiver.id.Name(), run.idNamespace, cast.ToString(mergedConfig.Get(endpointConfigKey))))
+	// Sets dynamically created receiver to something like receiver_creator/1/redis{endpoint="localhost:6380"}/<EndpointID>.
+	receiverCfg.SetIDName(fmt.Sprintf("%s/%s{endpoint=%q}/%s", receiver.id.Name(), run.idNamespace, cast.ToString(mergedConfig.Get(endpointConfigKey)), receiver.endpointID))
 	return receiverCfg, nil
 }
 

receiver/receivercreator/runner_test.go

@@ -43,7 +43,7 @@ func Test_loadAndCreateRuntimeReceiver(t *testing.T) {
 	nopConfig := loadedConfig.(*nopWithEndpointConfig)
 	// Verify that the overridden endpoint is used instead of the one in the config file.
 	assert.Equal(t, "localhost:12345", nopConfig.Endpoint)
-	expectedID := `nop/1/receiver_creator/1{endpoint="localhost:12345"}`
+	expectedID := `nop/1/receiver_creator/1{endpoint="localhost:12345"}/endpoint.id`
 	assert.Equal(t, expectedID, nopConfig.ID().String())
 
 	// Test that metric receiver can be created from loaded config and it logs its id for the "name" field.

unreleased/receivercreatoruniquereceivernames.yaml

@@ -0,0 +1,4 @@
+change_type: enhancement
+component: receivercreator
+note: adds the unique EndpointID to generated receiver componentID to prevent collisions
+issues: [12670]