Change FF to explicitly disable acks

Author: michel-laterman

changelog/fragments/1757710842-Add-agent_policy_id-and-policy_revision_idx-to-checkin-requests.yaml

@@ -19,7 +19,7 @@ summary: Add agent_policy_id and policy_revision_idx to checkin requests
 description: |
   Add agent_policy_id and policy_revision_idx attributes to checkin requests.
   These attributes are used to inform fleet-server of the policy id and revision that the agent is currently running.
-  Agents that use these policies no longer need to send acks for POLICY_CHANGE actions.
+  Add a feature flag to disable sending acks for POLICY_CHANGE actions on a future release.
 
 # Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
 component: elastic-agent
@@ -28,7 +28,7 @@ component: elastic-agent
 # If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
 # NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
 # Please provide it if you are adding a fragment for a different PR.
-#pr: https://github.com/owner/repo/1234
+pr: https://github.com/elastic/elastic-agent/pull/9931
 
 # Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
 # If not present is automatically filled by the tooling with the issue linked to the PR number.

internal/pkg/agent/application/actions/handlers/handler_action_policy_change.go

@@ -42,7 +42,7 @@ type PolicyChangeHandler struct {
 	setters              []actions.ClientSetter
 	policyLogLevelSetter logLevelSetter
 	coordinator          *coordinator.Coordinator
-	forceAckFn           func() bool
+	disableAckFn         func() bool
 	// Disabled for 8.8.0 release in order to limit the surface
 	// https://github.com/elastic/security-team/issues/6501
 	// // Last known valid signature validation key
@@ -69,7 +69,7 @@ func NewPolicyChangeHandler(
 		setters:              setters,
 		coordinator:          coordinator,
 		policyLogLevelSetter: policyLogLevelSetter,
-		forceAckFn:           features.ForcePolicyChangeAcks,
+		disableAckFn:         features.DisablePolicyChangeAcks,
 	}
 }
 
@@ -114,7 +114,7 @@ func (h *PolicyChangeHandler) Handle(ctx context.Context, a fleetapi.Action, ack
 		return err
 	}
 
-	h.ch <- newPolicyChange(ctx, c, a, acker, false, h.forceAckFn())
+	h.ch <- newPolicyChange(ctx, c, a, acker, false, h.disableAckFn())
 	return nil
 }
 
@@ -477,7 +477,7 @@ type policyChange struct {
 	action     fleetapi.Action
 	acker      acker.Acker
 	ackWatcher chan struct{}
-	forceAck   bool
+	disableAck bool
 }
 
 func newPolicyChange(
@@ -486,7 +486,7 @@ func newPolicyChange(
 	action fleetapi.Action,
 	acker acker.Acker,
 	makeCh bool,
-	forceAck bool) *policyChange {
+	disableAck bool) *policyChange {
 	var ackWatcher chan struct{}
 	if makeCh {
 		// we don't need it otherwise
@@ -498,7 +498,7 @@ func newPolicyChange(
 		action:     action,
 		acker:      acker,
 		ackWatcher: ackWatcher,
-		forceAck:   forceAck,
+		disableAck: disableAck,
 	}
 }
 
@@ -507,9 +507,9 @@ func (l *policyChange) Config() *config.Config {
 }
 
 // Ack sends an ack for the associated action if the results are expected.
-// An ack will not be sent for a POLICY_CHANGE action, but will be when this method is used by UNENROLL actions.
+// An ack will be sent for UNENROLL actions, or by POLICY_CHANGE actions if it has not been explicitly disabled.
 func (l *policyChange) Ack() error {
-	if !l.forceAck || l.action == nil {
+	if l.disableAck || l.action == nil {
 		return nil
 	}
 	err := l.acker.Ack(l.ctx, l.action)

internal/pkg/agent/application/actions/handlers/handler_action_policy_change_test.go

@@ -105,7 +105,7 @@ func TestPolicyAcked(t *testing.T) {
 	agentInfo := &info.AgentInfo{}
 	nullStore := &storage.NullStore{}
 
-	t.Run("Config change shouldn't ACK", func(t *testing.T) {
+	t.Run("Default: Config changes are ACKed", func(t *testing.T) {
 		ch := make(chan coordinator.ConfigChange, 1)
 		tacker := &testAcker{}
 
@@ -119,6 +119,7 @@ func TestPolicyAcked(t *testing.T) {
 			},
 		}
 
+		// Test default FF value
 		cfg := configuration.DefaultConfiguration()
 		handler := NewPolicyChangeHandler(log, agentInfo, cfg, nullStore, ch, nilLogLevelSet(t), &coordinator.Coordinator{})
 
@@ -129,7 +130,8 @@ func TestPolicyAcked(t *testing.T) {
 		require.NoError(t, change.Ack())
 
 		actions := tacker.Items()
-		assert.Empty(t, actions)
+		assert.Len(t, actions, 1)
+		assert.Equal(t, actionID, actions[0])
 	})
 	t.Run("Config change acks when forced", func(t *testing.T) {
 		ch := make(chan coordinator.ConfigChange, 1)
@@ -147,7 +149,7 @@ func TestPolicyAcked(t *testing.T) {
 
 		cfg := configuration.DefaultConfiguration()
 		handler := NewPolicyChangeHandler(log, agentInfo, cfg, nullStore, ch, nilLogLevelSet(t), &coordinator.Coordinator{})
-		handler.forceAckFn = func() bool { return true }
+		handler.disableAckFn = func() bool { return false }
 
 		err := handler.Handle(context.Background(), action, tacker)
 		require.NoError(t, err)
@@ -159,6 +161,33 @@ func TestPolicyAcked(t *testing.T) {
 		assert.Len(t, actions, 1)
 		assert.Equal(t, actionID, actions[0])
 	})
+	t.Run("Config change do not ack when disabled", func(t *testing.T) {
+		ch := make(chan coordinator.ConfigChange, 1)
+		tacker := &testAcker{}
+
+		config := map[string]interface{}{"hello": "world"}
+		actionID := "abc123"
+		action := &fleetapi.ActionPolicyChange{
+			ActionID:   actionID,
+			ActionType: "POLICY_CHANGE",
+			Data: fleetapi.ActionPolicyChangeData{
+				Policy: config,
+			},
+		}
+
+		cfg := configuration.DefaultConfiguration()
+		handler := NewPolicyChangeHandler(log, agentInfo, cfg, nullStore, ch, nilLogLevelSet(t), &coordinator.Coordinator{})
+		handler.disableAckFn = func() bool { return true }
+
+		err := handler.Handle(context.Background(), action, tacker)
+		require.NoError(t, err)
+
+		change := <-ch
+		require.NoError(t, change.Ack())
+
+		actions := tacker.Items()
+		assert.Empty(t, actions)
+	})
 }
 
 func TestPolicyChangeHandler_handlePolicyChange_FleetClientSettings(t *testing.T) {

internal/pkg/agent/application/actions/handlers/handler_action_unenroll.go

@@ -93,7 +93,7 @@ func (h *Unenroll) handle(ctx context.Context, a fleetapi.Action, acker acker.Ac
 	}
 
 	// Generate empty policy change, this removing all the running components
-	unenrollPolicy := newPolicyChange(ctx, config.New(), a, acker, true, true)
+	unenrollPolicy := newPolicyChange(ctx, config.New(), a, acker, true, false)
 	h.ch <- unenrollPolicy
 
 	// backup action for future start to avoid starting fleet gateway loop

pkg/features/features.go

@@ -21,6 +21,10 @@ import (
 // 8.11+ - default is enabled
 const defaultTamperProtection = true
 
+// The default value of the disable policy change acks flag if the flag is missing.
+// 9.2 - disabled (acks are sent)
+const defaultDisablePolicyChangeAcks = false
+
 var (
 	current = Flags{
 		tamperProtection: defaultTamperProtection,
@@ -36,8 +40,8 @@ type Flags struct {
 	fqdn          bool
 	fqdnCallbacks map[string]BoolValueOnChangeCallback
 
-	tamperProtection      bool
-	forcePolicyChangeAcks bool
+	tamperProtection        bool
+	disablePolicyChangeAcks bool
 }
 
 type cfg struct {
@@ -49,9 +53,9 @@ type cfg struct {
 			TamperProtection *struct {
 				Enabled bool `json:"enabled" yaml:"enabled" config:"enabled"`
 			} `json:"tamper_protection,omitempty" yaml:"tamper_protection,omitempty" config:"tamper_protection,omitempty"`
-			ForcePolicyChangeAcks struct {
+			DisablePolicyChangeAcks struct {
 				Enabled bool `json:"enabled" yaml:"enabled" config:"enabled"`
-			} `json:"force_policy_change_acks" toml: "force_policy_change_acks" config:"force_policy_change_acks"`
+			} `json:"disable_policy_change_acks" toml: "disable_policy_change_acks" config:"disable_policy_change_acks"`
 		} `json:"features" yaml:"features" config:"features"`
 	} `json:"agent" yaml:"agent" config:"agent"`
 }
@@ -70,11 +74,11 @@ func (f *Flags) TamperProtection() bool {
 	return f.tamperProtection
 }
 
-func (f *Flags) ForcePolicyChangeAcks() bool {
+func (f *Flags) DisablePolicyChangeAcks() bool {
 	f.mu.RLock()
 	defer f.mu.RUnlock()
 
-	return f.forcePolicyChangeAcks
+	return f.disablePolicyChangeAcks
 }
 
 func (f *Flags) AsProto() *proto.Features {
@@ -132,11 +136,11 @@ func (f *Flags) setTamperProtection(newValue bool) {
 	f.tamperProtection = newValue
 }
 
-func (f *Flags) setForcePolicyChangeAcks(newValue bool) {
+func (f *Flags) setDisablePolicyChangeAcks(newValue bool) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
-	f.forcePolicyChangeAcks = newValue
+	f.disablePolicyChangeAcks = newValue
 }
 
 // setSource sets the source from he given cfg.
@@ -204,6 +208,12 @@ func Parse(policy any) (*Flags, error) {
 		flags.setTamperProtection(defaultTamperProtection)
 	}
 
+	if parsedFlags.Agent.Features.DisablePolicyChangeAcks != nil {
+		flags.setDisablePolicyChangeAcks(parsedFlags.Agent.Features.DisablePolicyChangeAcks.Enabled)
+	} else {
+		flags.setDisablePolicyChangeAcks(defaultDisablePolicyChangeAcks)
+	}
+
 	if err := flags.setSource(parsedFlags); err != nil {
 		return nil, fmt.Errorf("error creating feature flags source: %w", err)
 	}
@@ -226,7 +236,7 @@ func Apply(c *config.Config) error {
 
 	current.setFQDN(parsed.FQDN())
 	current.setTamperProtection(parsed.TamperProtection())
-	current.setForcePolicyChangeAcks(parsed.ForcePolicyChangeAcks())
+	current.setDisablePolicyChangeAcks(parsed.DisablePolicyChangeAcks())
 	return err
 }
 
@@ -240,7 +250,7 @@ func TamperProtection() bool {
 	return current.TamperProtection()
 }
 
-// ForcePolicyChangeAcks reports if the agent should force sending an ACK for POLICY_CHANGE actions.
-func ForcePolicyChangeAcks() bool {
-	return current.ForcePolicyChangeAcks()
+// DisablePolicyChangeAcks reports if the agent will stop using ACKs for POLICY_CHANGE actions.
+func DisablePolicyChangeAcks() bool {
+	return current.DisablePolicyChangeAcks()
 }