Add release note

Author: piochelepiotr

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

@@ -86,7 +86,7 @@ func (c *actionsController) manageSubscriptionToRC() {
 		}
 		c.closeMutex.RUnlock()
 		if isConnectedToKafka(c.ac) {
-			c.rcclient.Subscribe(data.ProductDebug, c.update)
+			c.rcclient.Subscribe(data.ProductDataStreamsKafkaActions, c.update)
 			return
 		}
 	}

pkg/collector/scheduler/scheduler_test.go

@@ -207,19 +207,31 @@ func TestStopOneTimeSchedule(t *testing.T) {
 // RunOnceTestCheck is a test check that can be configured as run-once or regular
 type RunOnceTestCheck struct {
 	stub.StubCheck
+	mu         sync.RWMutex
 	id         string
 	runOnce    bool
 	intl       time.Duration
 	runCounter int
 }
 
-func (c *RunOnceTestCheck) ID() checkid.ID          { return checkid.ID(c.id) }
-func (c *RunOnceTestCheck) RunOnce() bool           { return c.runOnce }
+func (c *RunOnceTestCheck) ID() checkid.ID { return checkid.ID(c.id) }
+func (c *RunOnceTestCheck) RunOnce() bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.runOnce
+}
 func (c *RunOnceTestCheck) Interval() time.Duration { return c.intl }
 func (c *RunOnceTestCheck) Run() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	c.runCounter++
 	return nil
 }
+func (c *RunOnceTestCheck) GetRunCounter() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.runCounter
+}
 
 func TestRunOnceCheckDescheduling(t *testing.T) {
 	checkChan := make(chan check.Check, 100)
@@ -269,30 +281,29 @@ func TestRunOnceCheckDescheduling(t *testing.T) {
 	runOnceCheck := &RunOnceTestCheck{
 		id:      "run-once-check",
 		runOnce: true,
+		intl:    1 * time.Second,
 	}
-	runOnceCheck.intl = 1 * time.Second
 
 	regularCheck := &RunOnceTestCheck{
 		id:      "regular-check",
 		runOnce: false,
+		intl:    1 * time.Second,
 	}
-	regularCheck.intl = 1 * time.Second
+
+	// Manually create a job queue for 1 second interval and replace its ticker
+	// This must be done BEFORE calling Enter to avoid race conditions
+	jq := newJobQueue(1 * time.Second)
+	tickerChan := make(chan time.Time, 10)
+	jq.bucketTicker.Stop()
+	jq.bucketTicker = &time.Ticker{C: tickerChan}
+	s.jobQueues[1*time.Second] = jq
 
 	// Schedule both checks
 	err := s.Enter(runOnceCheck)
 	require.NoError(t, err)
 	err = s.Enter(regularCheck)
 	require.NoError(t, err)
 
-	// Verify both checks are in the queue
-	jq := s.jobQueues[1*time.Second]
-	require.NotNil(t, jq)
-
-	// Replace the ticker's channel with one we control
-	jq.bucketTicker.Stop()
-	tickerChan := make(chan time.Time, 10)
-	jq.bucketTicker = &time.Ticker{C: tickerChan}
-
 	// Start the scheduler
 	s.Run()
 

pkg/config/remote/data/product.go

@@ -38,8 +38,8 @@ const (
 	ProductContainerAutoscalingValues = "CONTAINER_AUTOSCALING_VALUES"
 	// ProductDataStreamsLiveMessages is to capture messages from Kafka
 	ProductDataStreamsLiveMessages = "DSM_LIVE_MESSAGES"
-	// ProductDebug is for debugging and testing remote configuration
-	ProductDebug = "DEBUG"
+	// ProductDataStreamsKafkaActions is to execute Kafka actions remotely
+	ProductDataStreamsKafkaActions = "DSM_KAFKA_ACTIONS"
 )
 
 // ProductListToString converts a product list to string list

pkg/remoteconfig/state/products.go

@@ -37,13 +37,13 @@ var validProducts = map[string]struct{}{
 	ProductNDMDeviceProfilesCustom:      {},
 	ProductMetricControl:                {},
 	ProductDataStreamsLiveMessages:      {},
+	ProductDataStreamsKafkaActions:      {},
 	ProductLiveDebuggingSymbolDB:        {},
 	ProductGradualRollout:               {},
 	ProductApmPolicies:                  {},
 	ProductSyntheticsTest:               {},
 	ProductBTFDD:                        {},
 	ProductFFEFlags:                     {},
-	ProductDebug:                        {},
 }
 
 const (
@@ -114,6 +114,8 @@ const (
 	ProductMetricControl = "METRIC_CONTROL"
 	// ProductDataStreamsLiveMessages is used for capturing messages from Kafka
 	ProductDataStreamsLiveMessages = "DSM_LIVE_MESSAGES"
+	// ProductDataStreamsKafkaActions is used for executing Kafka actions remotely
+	ProductDataStreamsKafkaActions = "DSM_KAFKA_ACTIONS"
 	// ProductGradualRollout tracks the latest stable release versions for K8s gradual rollout.
 	ProductGradualRollout = "K8S_INJECTION_DD"
 	// ProductBTFDD accesses a BTF catalog used when the kernel is newer than the system-probe has bundled support for
@@ -122,6 +124,4 @@ const (
 	ProductApmPolicies = "APM_POLICIES"
 	// ProductFFEFlags is used for feature flagging experiments remote updates
 	ProductFFEFlags = "FFE_FLAGS"
-	// ProductDebug is for debugging and testing remote configuration
-	ProductDebug = "DEBUG"
 )

releasenotes/notes/data-streams-actions-39bfd4c97d878c29.yaml

@@ -0,0 +1,4 @@
+features:
+  - Checks can be scheduled only once with run_once configuration
+  - Data Streams Kafka actions perform actions on Kafka clusters
+