Add ForcePolicyChangeAcks feature flag

Author: michel-laterman

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

@@ -29,6 +29,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/client"
 	"github.com/elastic/elastic-agent/internal/pkg/remote"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
+	"github.com/elastic/elastic-agent/pkg/features"
 )
 
 // PolicyChangeHandler is a handler for POLICY_CHANGE action.
@@ -41,6 +42,7 @@ type PolicyChangeHandler struct {
 	setters              []actions.ClientSetter
 	policyLogLevelSetter logLevelSetter
 	coordinator          *coordinator.Coordinator
+	forceAckFn           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
@@ -67,6 +69,7 @@ func NewPolicyChangeHandler(
 		setters:              setters,
 		coordinator:          coordinator,
 		policyLogLevelSetter: policyLogLevelSetter,
+		forceAckFn:           features.ForcePolicyChangeAcks,
 	}
 }
 
@@ -111,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.ch <- newPolicyChange(ctx, c, a, acker, false, h.forceAckFn())
 	return nil
 }
 
@@ -474,14 +477,16 @@ type policyChange struct {
 	action     fleetapi.Action
 	acker      acker.Acker
 	ackWatcher chan struct{}
+	forceAck   bool
 }
 
 func newPolicyChange(
 	ctx context.Context,
 	config *config.Config,
 	action fleetapi.Action,
 	acker acker.Acker,
-	makeCh bool) *policyChange {
+	makeCh bool,
+	forceAck bool) *policyChange {
 	var ackWatcher chan struct{}
 	if makeCh {
 		// we don't need it otherwise
@@ -493,6 +498,7 @@ func newPolicyChange(
 		action:     action,
 		acker:      acker,
 		ackWatcher: ackWatcher,
+		forceAck:   forceAck,
 	}
 }
 
@@ -503,15 +509,15 @@ 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.
 func (l *policyChange) Ack() error {
-	if l.action == nil || l.ackWatcher == nil {
+	if !l.forceAck || l.action == nil {
 		return nil
 	}
 	err := l.acker.Ack(l.ctx, l.action)
 	if err != nil {
 		return err
 	}
 	err = l.acker.Commit(l.ctx)
-	if err == nil {
+	if err == nil && l.ackWatcher != nil {
 		close(l.ackWatcher)
 	}
 	return err

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

@@ -131,6 +131,34 @@ func TestPolicyAcked(t *testing.T) {
 		actions := tacker.Items()
 		assert.Empty(t, actions)
 	})
+	t.Run("Config change acks when forced", 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.forceAckFn = 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.Len(t, actions, 1)
+		assert.Equal(t, actionID, actions[0])
+	})
 }
 
 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)
+	unenrollPolicy := newPolicyChange(ctx, config.New(), a, acker, true, true)
 	h.ch <- unenrollPolicy
 
 	// backup action for future start to avoid starting fleet gateway loop

pkg/features/features.go

@@ -36,7 +36,8 @@ type Flags struct {
 	fqdn          bool
 	fqdnCallbacks map[string]BoolValueOnChangeCallback
 
-	tamperProtection bool
+	tamperProtection      bool
+	forcePolicyChangeAcks bool
 }
 
 type cfg struct {
@@ -48,6 +49,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 {
+				Enabled bool `json:"enabled" yaml:"enabled" config:"enabled"`
+			} `json:"force_policy_change_acks" toml: "force_policy_change_acks" config:"force_policy_change_acks"`
 		} `json:"features" yaml:"features" config:"features"`
 	} `json:"agent" yaml:"agent" config:"agent"`
 }
@@ -66,6 +70,13 @@ func (f *Flags) TamperProtection() bool {
 	return f.tamperProtection
 }
 
+func (f *Flags) ForcePolicyChangeAcks() bool {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+
+	return f.forcePolicyChangeAcks
+}
+
 func (f *Flags) AsProto() *proto.Features {
 	return &proto.Features{
 		Fqdn: &proto.FQDNFeature{
@@ -121,6 +132,13 @@ func (f *Flags) setTamperProtection(newValue bool) {
 	f.tamperProtection = newValue
 }
 
+func (f *Flags) setForcePolicyChangeAcks(newValue bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	f.forcePolicyChangeAcks = newValue
+}
+
 // setSource sets the source from he given cfg.
 func (f *Flags) setSource(c cfg) error {
 	// Use JSON marshalling-unmarshalling to convert cfg to mapstr
@@ -208,6 +226,7 @@ func Apply(c *config.Config) error {
 
 	current.setFQDN(parsed.FQDN())
 	current.setTamperProtection(parsed.TamperProtection())
+	current.setForcePolicyChangeAcks(parsed.ForcePolicyChangeAcks())
 	return err
 }
 
@@ -220,3 +239,8 @@ func FQDN() bool {
 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()
+}