Add agent_policy_id and policy_revision_idx to checkin requests

Add the agent_policy_id and policy_revision_idx attributes to checkin
requests.

Author: michel-laterman

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

@@ -0,0 +1,35 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add agent_policy_id and policy_revision_idx to checkin requests
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+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.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# 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
+
+# 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.
+issue: https://github.com/elastic/elastic-agent/issues/6446

internal/pkg/agent/application/gateway/fleet/fleet_gateway.go

@@ -71,6 +71,7 @@ type stateStore interface {
 	AckToken() string
 	SetAckToken(ackToken string)
 	Save() error
+	Action() fleetapi.Action
 }
 
 type FleetGateway struct {
@@ -356,15 +357,20 @@ func (f *FleetGateway) execute(ctx context.Context) (*fleetapi.CheckinResponse,
 	// Fix loglevel with the current log level used by coordinator
 	ecsMeta.Elastic.Agent.LogLevel = state.LogLevel.String()
 
+	agentPolicyID := getPolicyID(f.stateStore.Action())
+	policyRevisionIDX := getPolicyRevisionIDX(f.stateStore.Action())
+
 	// checkin
 	cmd := fleetapi.NewCheckinCmd(f.agentInfo, f.client)
 	req := &fleetapi.CheckinRequest{
-		AckToken:       ackToken,
-		Metadata:       ecsMeta,
-		Status:         agentStateToString(state.State),
-		Message:        state.Message,
-		Components:     components,
-		UpgradeDetails: state.UpgradeDetails,
+		AckToken:          ackToken,
+		Metadata:          ecsMeta,
+		Status:            agentStateToString(state.State),
+		Message:           state.Message,
+		Components:        components,
+		UpgradeDetails:    state.UpgradeDetails,
+		AgentPolicyID:     agentPolicyID,
+		PolicyRevisionIDX: policyRevisionIDX,
 	}
 
 	resp, took, err := cmd.Execute(ctx, req)
@@ -447,3 +453,43 @@ func RequestBackoff(done <-chan struct{}) backoff.Backoff {
 		defaultFleetBackoffSettings.Max,
 	)
 }
+
+// getPolicyID will check that the passed action is a POLICY_CHANGE action and return the policy_id attribute of the policy as a string.
+func getPolicyID(action fleetapi.Action) string {
+	policyChange, ok := action.(*fleetapi.ActionPolicyChange)
+	if !ok {
+		return ""
+	}
+	v, ok := policyChange.Data.Policy["policy_id"]
+	if !ok {
+		return ""
+	}
+	vv, ok := v.(string)
+	if !ok {
+		return ""
+	}
+	return vv
+}
+
+// getPolicyRevisionIDX will check that the passed action is a POLICY_CHANGE action and return the policy_revision_idx attribute of the policy as an int64.
+// The function will attempt to convert the attribute to int64 if int or float64 is used in order to prevent issues from serialization.
+func getPolicyRevisionIDX(action fleetapi.Action) int64 {
+	policyChange, ok := action.(*fleetapi.ActionPolicyChange)
+	if !ok {
+		return 0
+	}
+	v, ok := policyChange.Data.Policy["policy_revision_idx"]
+	if !ok {
+		return 0
+	}
+	switch vv := v.(type) {
+	case int64:
+		return vv
+	case int:
+		return int64(vv)
+	case float64:
+		return int64(vv)
+	default:
+		return 0
+	}
+}

internal/pkg/agent/application/gateway/fleet/fleet_gateway_test.go

@@ -377,6 +377,74 @@ func TestFleetGateway(t *testing.T) {
 		default:
 		}
 	})
+
+	t.Run("sends agent_policy_id and policy_revision_idx", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+
+		scheduler := scheduler.NewStepper()
+		client := newTestingClient()
+
+		log, _ := loggertest.New("fleet_gateway")
+
+		stateStore := newStateStore(t, log)
+		stateStore.SetAction(&fleetapi.ActionPolicyChange{
+			ActionID:   "test-action-id",
+			ActionType: fleetapi.ActionTypePolicyChange,
+			Data: fleetapi.ActionPolicyChangeData{
+				Policy: map[string]interface{}{
+					"policy_id":           "test-policy-id",
+					"policy_revision_idx": 1,
+				},
+			},
+		})
+		err := stateStore.Save()
+		require.NoError(t, err)
+
+		gateway, err := newFleetGatewayWithScheduler(
+			log,
+			settings,
+			agentInfo,
+			client,
+			scheduler,
+			noop.New(),
+			emptyStateFetcher,
+			stateStore,
+		)
+		require.NoError(t, err)
+
+		waitFn := ackSeq(
+			client.Answer(func(headers http.Header, body io.Reader) (*http.Response, error) {
+				data, err := io.ReadAll(body)
+				require.NoError(t, err)
+
+				var checkinRequest fleetapi.CheckinRequest
+				err = json.Unmarshal(data, &checkinRequest)
+				require.NoError(t, err)
+
+				require.Equal(t, "test-policy-id", checkinRequest.AgentPolicyID)
+				require.Equal(t, int64(1), checkinRequest.PolicyRevisionIDX)
+
+				resp := wrapStrToResp(http.StatusOK, `{ "actions": [] }`)
+				return resp, nil
+			}),
+		)
+
+		errCh := runFleetGateway(ctx, gateway)
+
+		// Synchronize scheduler and acking of calls from the worker go routine.
+		scheduler.Next()
+		waitFn()
+
+		cancel()
+		err = <-errCh
+		require.NoError(t, err)
+		select {
+		case actions := <-gateway.Actions():
+			t.Errorf("Expected no actions, got %v", actions)
+		default:
+		}
+	})
 }
 
 func TestRetriesOnFailures(t *testing.T) {

internal/pkg/fleetapi/checkin_cmd.go

@@ -41,12 +41,14 @@ type CheckinComponent struct {
 
 // CheckinRequest consists of multiple events reported to fleet ui.
 type CheckinRequest struct {
-	Status         string             `json:"status"`
-	AckToken       string             `json:"ack_token,omitempty"`
-	Metadata       *info.ECSMeta      `json:"local_metadata,omitempty"`
-	Message        string             `json:"message"`    // V2 Agent message
-	Components     []CheckinComponent `json:"components"` // V2 Agent components
-	UpgradeDetails *details.Details   `json:"upgrade_details,omitempty"`
+	Status            string             `json:"status"`
+	AckToken          string             `json:"ack_token,omitempty"`
+	Metadata          *info.ECSMeta      `json:"local_metadata,omitempty"`
+	Message           string             `json:"message"`    // V2 Agent message
+	Components        []CheckinComponent `json:"components"` // V2 Agent components
+	UpgradeDetails    *details.Details   `json:"upgrade_details,omitempty"`
+	AgentPolicyID     string             `json:"agent_policy_id,omitempty"`
+	PolicyRevisionIDX int64              `json:"policy_revision_idx,omitempty"`
 }
 
 // SerializableEvent is a representation of the event to be send to the Fleet Server API via the checkin