Pass down remote config id to kafka actions (#43356)

### What does this PR do?

Follow up to #43325

### Motivation

Events generated by the kafka_actions can easily be correlated with the action taken in the UI

### Describe how you validated your changes

QA locally

### Additional Notes


Co-authored-by: piotr.wolski <[email protected]>

Author: piochelepiotr

comp/core/autodiscovery/providers/datastreams/kafka_actions.go

@@ -152,6 +152,7 @@ func (c *actionsController) update(updates map[string]state.RawConfig, applyStat
 		}
 
 		actionsMap["run_once"] = true
+		actionsMap["remote_config_id"] = parsed.remoteConfigID
 		for k, v := range auth {
 			actionsMap[k] = v
 		}
@@ -182,6 +183,7 @@ type parsedActionsConfig struct {
 	path             string
 	bootstrapServers string
 	actionsJSON      json.RawMessage
+	remoteConfigID   string
 }
 
 func parseActionsConfig(updates map[string]state.RawConfig, applyStateCallback func(string, state.ApplyStatus)) []parsedActionsConfig {
@@ -205,6 +207,7 @@ func parseActionsConfig(updates map[string]state.RawConfig, applyStateCallback f
 			path:             path,
 			bootstrapServers: bootstrapServers,
 			actionsJSON:      cfg.Actions,
+			remoteConfigID:   rawConfig.Metadata.ID,
 		})
 	}
 	return configs

comp/core/autodiscovery/providers/datastreams/kafka_actions_test.go

@@ -90,7 +90,7 @@ func TestActionsController(t *testing.T) {
 
 	rcUpdate := map[string]state.RawConfig{
 		"config_1": {Config: []byte("invalid")},
-		"config_2": {Config: serializedConfig},
+		"config_2": {Config: serializedConfig, Metadata: state.Metadata{ID: "test_remote_config_id"}},
 	}
 
 	updateStatus := make(map[string]state.ApplyStatus)
@@ -133,6 +133,9 @@ func TestActionsController(t *testing.T) {
 	// Check that run_once was injected
 	assert.Equal(t, true, instance["run_once"])
 
+	// Check that remote_config_id was injected
+	assert.Equal(t, "test_remote_config_id", instance["remote_config_id"])
+
 	// Check that actions are present
 	assert.NotNil(t, instance["read_messages"])
 }
@@ -167,7 +170,7 @@ func TestActionsControllerNoBootstrapServers(t *testing.T) {
 	require.NoError(t, err)
 
 	rcUpdate := map[string]state.RawConfig{
-		"config_1": {Config: serializedConfig},
+		"config_1": {Config: serializedConfig, Metadata: state.Metadata{ID: "no_bootstrap_test_id"}},
 	}
 
 	updateStatus := make(map[string]state.ApplyStatus)
@@ -204,7 +207,7 @@ func TestActionsControllerNoMatchingKafkaConsumer(t *testing.T) {
 	require.NoError(t, err)
 
 	rcUpdate := map[string]state.RawConfig{
-		"config_1": {Config: serializedConfig},
+		"config_1": {Config: serializedConfig, Metadata: state.Metadata{ID: "no_match_test_id"}},
 	}
 
 	updateStatus := make(map[string]state.ApplyStatus)

releasenotes/notes/kafka-actions-tagging-848281611b7fd5bb.yaml

@@ -0,0 +1,2 @@
+enhancements:
+  - Add remote config ID tagging to events generated by kafka_action integration for easy UI filtering