Send upgrade details to Fleet Server in check-in API requests (#3528)

* Adding FSM for upgrades

* Implementing TODO

* WIP

* WIP

* Reorganizing imports

* Running go mod tidy

* Fix type

* Handle failures in one place

* Remove Fleet changes

* Send upgrade details in check-in requests to Fleet

* Add unit test

* Adding CHANGELOG entry

* Compare pointer values so we're not copying locks

* Add missing assertion

* Update Fleet-managed upgrade E2E test

* Make scheduled_at optional so empty time is not serialized

* Normalize comparison of hostnames

* Convert comment to log

* Fix assertion logic

* Removing replace directive

* Address linter error

NOTICE.txt

@@ -1166,11 +1166,11 @@ SOFTWARE
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-libs
-Version: v0.6.2
+Version: v0.7.0
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-libs@v0.6.2/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-libs@v0.7.0/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004

changelog/fragments/1699016628-send-upgrade-details-to-fleet.yaml

@@ -0,0 +1,32 @@
+# 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: Send upgrade details to Fleet
+
+# 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:
+
+# 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/elastic/elastic-agent/pull/3528
+
+# 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/3119

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/elastic/e2e-testing v1.1.0
 	github.com/elastic/elastic-agent-autodiscover v0.6.5
 	github.com/elastic/elastic-agent-client/v7 v7.5.0
-	github.com/elastic/elastic-agent-libs v0.6.2
+	github.com/elastic/elastic-agent-libs v0.7.0
 	github.com/elastic/elastic-agent-system-metrics v0.8.0
 	github.com/elastic/elastic-transport-go/v8 v8.3.0
 	github.com/elastic/go-elasticsearch/v8 v8.10.1

go.sum

@@ -781,8 +781,8 @@ github.com/elastic/elastic-agent-autodiscover v0.6.5 h1:5DeMpuNc8c/tN6HN0A4A2uOF
 github.com/elastic/elastic-agent-autodiscover v0.6.5/go.mod h1:chulyCAyZb/njMHgzkhC/yWnt8v/Y6eCRUhmFVnsA5o=
 github.com/elastic/elastic-agent-client/v7 v7.5.0 h1:niI3WQ+01Lnp2r5LxK8SyNhrPJe13vBiOkqrDRK2oTA=
 github.com/elastic/elastic-agent-client/v7 v7.5.0/go.mod h1:DYoX95xjC4BW/p2avyu724Qr2+hoUIz9eCU9CVS1d+0=
-github.com/elastic/elastic-agent-libs v0.6.2 h1:tE5pFK4y7xm1FtXm+r+63G7STjJAaWh3+oKIQDzdPDo=
-github.com/elastic/elastic-agent-libs v0.6.2/go.mod h1:o+EySawBZGeYu49shJxerg2wRCimS1dhrD4As0MS700=
+github.com/elastic/elastic-agent-libs v0.7.0 h1:g/+Gzpn4ayXPFbfZsn5lGjbPR1TGqlVpshJVVUNJGlQ=
+github.com/elastic/elastic-agent-libs v0.7.0/go.mod h1:o+EySawBZGeYu49shJxerg2wRCimS1dhrD4As0MS700=
 github.com/elastic/elastic-agent-system-metrics v0.8.0 h1:EsWbtd83JvnaqnL57bKS1E6GhOdemTRbxdFDcenR8zQ=
 github.com/elastic/elastic-agent-system-metrics v0.8.0/go.mod h1:9C1UEfj0P687HAzZepHszN6zXA+2tN2Lx3Osvq1zby8=
 github.com/elastic/elastic-integration-corpus-generator-tool v0.5.0/go.mod h1:uf9N86y+UACGybdEhZLpwZ93XHWVhsYZAA4c2T2v6YM=

internal/pkg/agent/application/coordinator/diagnostics_test.go

@@ -441,7 +441,7 @@ func TestDiagnosticState(t *testing.T) {
 			ActionID:      "foobar",
 			Metadata: details.Metadata{
 				DownloadPercent: 0.17469,
-				ScheduledAt:     now,
+				ScheduledAt:     &now,
 				DownloadRate:    123.56,
 			},
 		},

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

@@ -332,11 +332,12 @@ func (f *FleetGateway) execute(ctx context.Context) (*fleetapi.CheckinResponse,
 	// checkin
 	cmd := fleetapi.NewCheckinCmd(f.agentInfo, f.client)
 	req := &fleetapi.CheckinRequest{
-		AckToken:   ackToken,
-		Metadata:   ecsMeta,
-		Status:     agentStateToString(state.State),
-		Message:    state.Message,
-		Components: components,
+		AckToken:       ackToken,
+		Metadata:       ecsMeta,
+		Status:         agentStateToString(state.State),
+		Message:        state.Message,
+		Components:     components,
+		UpgradeDetails: state.UpgradeDetails,
 	}
 
 	resp, took, err := cmd.Execute(ctx, req)

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

@@ -7,6 +7,7 @@ package fleet
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"io"
@@ -23,9 +24,11 @@ import (
 	"gotest.tools/assert"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage/store"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker/noop"
 	"github.com/elastic/elastic-agent/internal/pkg/scheduler"
 	agentclient "github.com/elastic/elastic-agent/pkg/control/v2/client"
@@ -308,6 +311,73 @@ func TestFleetGateway(t *testing.T) {
 		require.NoError(t, err)
 	})
 
+	t.Run("Sends upgrade details", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		scheduler := scheduler.NewStepper()
+		client := newTestingClient()
+
+		log, _ := logger.NewTesting("fleet_gateway")
+
+		stateStore := newStateStore(t, log)
+
+		upgradeDetails := &details.Details{
+			TargetVersion: "8.12.0",
+			State:         "UPG_WATCHING",
+			ActionID:      "foobarbaz",
+		}
+		stateFetcher := func() coordinator.State {
+			return coordinator.State{
+				UpgradeDetails: upgradeDetails,
+			}
+		}
+
+		gateway, err := newFleetGatewayWithScheduler(
+			log,
+			settings,
+			agentInfo,
+			client,
+			scheduler,
+			noop.New(),
+			stateFetcher,
+			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.NotNil(t, checkinRequest.UpgradeDetails)
+				require.Equal(t, upgradeDetails, checkinRequest.UpgradeDetails)
+
+				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/agent/application/upgrade/details/details.go

@@ -35,7 +35,7 @@ type Details struct {
 
 // Metadata consists of metadata relating to a specific upgrade state
 type Metadata struct {
-	ScheduledAt time.Time `json:"scheduled_at,omitempty" yaml:"scheduled_at,omitempty"`
+	ScheduledAt *time.Time `json:"scheduled_at,omitempty" yaml:"scheduled_at,omitempty"`
 
 	// DownloadPercent is the percentage of the artifact that has been
 	// downloaded. Minimum value is 0 and maximum value is 1.

internal/pkg/fleetapi/checkin_cmd.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/client"
 )
@@ -47,11 +48,12 @@ 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
+	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"`
 }
 
 // SerializableEvent is a representation of the event to be send to the Fleet Server API via the checkin

pkg/control/v2/server/server.go

@@ -370,12 +370,17 @@ func stateToProto(state *coordinator.State, agentInfo *info.AgentInfo) (*cproto.
 			State:         string(state.UpgradeDetails.State),
 			ActionId:      state.UpgradeDetails.ActionID,
 			Metadata: &cproto.UpgradeDetailsMetadata{
-				ScheduledAt:     timestamppb.New(state.UpgradeDetails.Metadata.ScheduledAt),
 				DownloadPercent: float32(state.UpgradeDetails.Metadata.DownloadPercent),
 				FailedState:     string(state.UpgradeDetails.Metadata.FailedState),
 				ErrorMsg:        state.UpgradeDetails.Metadata.ErrorMsg,
 			},
 		}
+
+		if state.UpgradeDetails.Metadata.ScheduledAt != nil &&
+			!state.UpgradeDetails.Metadata.ScheduledAt.IsZero() {
+			upgradeDetails.Metadata.ScheduledAt = timestamppb.New(*state.UpgradeDetails.Metadata.ScheduledAt)
+
+		}
 	}
 
 	return &cproto.StateResponse{

pkg/control/v2/server/server_test.go

@@ -164,11 +164,16 @@ func TestStateMapping(t *testing.T) {
 
 			if tc.upgradeDetails != nil {
 				expectedMetadata := &cproto.UpgradeDetailsMetadata{
-					ScheduledAt:     timestamppb.New(tc.upgradeDetails.Metadata.ScheduledAt),
 					DownloadPercent: float32(tc.upgradeDetails.Metadata.DownloadPercent),
 					FailedState:     string(tc.upgradeDetails.Metadata.FailedState),
 					ErrorMsg:        tc.upgradeDetails.Metadata.ErrorMsg,
 				}
+
+				if tc.upgradeDetails.Metadata.ScheduledAt != nil &&
+					!tc.upgradeDetails.Metadata.ScheduledAt.IsZero() {
+					expectedMetadata.ScheduledAt = timestamppb.New(*tc.upgradeDetails.Metadata.ScheduledAt)
+				}
+
 				assert.Equal(t, string(tc.upgradeDetails.State), stateResponse.UpgradeDetails.State)
 				assert.Equal(t, tc.upgradeDetails.TargetVersion, stateResponse.UpgradeDetails.TargetVersion)
 				assert.Equal(t, tc.upgradeDetails.ActionID, stateResponse.UpgradeDetails.ActionId)

pkg/testing/tools/fleettools/fleet.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/elastic/elastic-agent-libs/kibana"
 )
@@ -25,7 +26,7 @@ func GetAgentByPolicyIDAndHostnameFromList(client *kibana.Client, policyID, host
 		agentHostname := item.LocalMetadata.Host.Hostname
 		agentPolicyID := item.PolicyID
 
-		if agentHostname == hostname && agentPolicyID == policyID {
+		if strings.EqualFold(agentHostname, hostname) && agentPolicyID == policyID {
 			hostnameAgents = append(hostnameAgents, &listAgentsResp.Items[i])
 		}
 	}

testing/integration/upgrade_fleet_test.go

@@ -8,6 +8,7 @@ package integration
 
 import (
 	"context"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -140,6 +141,15 @@ func testUpgradeFleetManagedElasticAgent(ctx context.Context, t *testing.T, info
 	err = fleettools.UpgradeAgent(kibClient, policy.ID, endVersionInfo.Binary.String(), true)
 	require.NoError(t, err)
 
+	t.Log("Waiting from upgrade details to show up in Fleet")
+	hostname, err := os.Hostname()
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		agent, err := fleettools.GetAgentByPolicyIDAndHostnameFromList(kibClient, policy.ID, hostname)
+		return err == nil && agent.UpgradeDetails != nil
+
+	}, 5*time.Minute, time.Second)
+
 	// wait for the watcher to show up
 	t.Logf("Waiting for upgrade watcher to start...")
 	err = upgradetest.WaitForWatcher(ctx, 5*time.Minute, 10*time.Second)