fix: scheduled upgrade details state (#9562)

* fix: persisting and reporting of upgrade details

* ci: align and extend dispatcher unit-tests

* ci: update coordinator and application new signatures in unit-tests

* ci: add integration tests for scheduled upgrade details

* doc: add changelog fragment

* doc: reword existing and add more comments in code

* feat: change queuedUpgradeActions inside dispatchCancelActions to have values of struct{}

* fix: remove redundant continue

* fix: dedupe upgrade actions from fleetgateway actions, handle correctly the expiration of retried stored actions, and update upgrade details on retries

(cherry picked from commit ff80471)

# Conflicts:
#	internal/pkg/agent/application/application.go
#	internal/pkg/agent/application/coordinator/coordinator.go
#	internal/pkg/agent/cmd/run.go

Author: pkoutsovasilis

changelog/fragments/1756218044-fix-upgrade-details-state.yaml

@@ -0,0 +1,5 @@
+kind: bug-fix
+summary: fix reporting of scheduled upgrade details across restarts and cancels
+component: elastic-agent
+pr: https://github.com/elastic/elastic-agent/pull/9562
+issue: https://github.com/elastic/elastic-agent/issues/8778

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

@@ -115,7 +115,7 @@ func TestUpgradeHandler(t *testing.T) {
 				return nil, nil
 			},
 		},
-		nil, nil, nil, nil, nil, false, nil, nil)
+		nil, nil, nil, nil, nil, false, nil, nil, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 
@@ -174,7 +174,7 @@ func TestUpgradeHandlerSameVersion(t *testing.T) {
 				return nil, err
 			},
 		},
-		nil, nil, nil, nil, nil, false, nil, nil)
+		nil, nil, nil, nil, nil, false, nil, nil, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 
@@ -233,7 +233,7 @@ func TestDuplicateActionsHandled(t *testing.T) {
 				return nil, nil
 			},
 		},
-		nil, nil, nil, nil, nil, false, nil, acker)
+		nil, nil, nil, nil, nil, false, nil, acker, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 
@@ -327,7 +327,7 @@ func TestUpgradeHandlerNewVersion(t *testing.T) {
 				return nil, nil
 			},
 		},
-		nil, nil, nil, nil, nil, false, nil, nil)
+		nil, nil, nil, nil, nil, false, nil, nil, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 

internal/pkg/agent/application/application.go

@@ -14,10 +14,12 @@ import (
 	"github.com/elastic/elastic-agent-libs/logp"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/dispatcher"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/configuration"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage"
@@ -32,6 +34,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker/retrier"
 	fleetclient "github.com/elastic/elastic-agent/internal/pkg/fleetapi/client"
 	otelmanager "github.com/elastic/elastic-agent/internal/pkg/otel/manager"
+	"github.com/elastic/elastic-agent/internal/pkg/queue"
 	"github.com/elastic/elastic-agent/internal/pkg/release"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/component/runtime"
@@ -57,6 +60,7 @@ func New(
 	fleetInitTimeout time.Duration,
 	disableMonitoring bool,
 	override CfgOverrider,
+	initialUpgradeDetails *details.Details,
 	modifiers ...component.PlatformModifier,
 ) (*coordinator.Coordinator, coordinator.ConfigManager, composable.Controller, error) {
 
@@ -141,7 +145,6 @@ func New(
 	var compModifiers = []coordinator.ComponentsModifier{InjectAPMConfig}
 	var composableManaged bool
 	var isManaged bool
-
 	var actionAcker acker.Acker
 	if testingMode {
 		log.Info("Elastic Agent has been started in testing mode and is managed through the control protocol")
@@ -210,8 +213,19 @@ func New(
 			batchedAcker := lazy.NewAcker(fleetAcker, log, lazy.WithRetrier(retrier))
 			actionAcker = stateStore.NewStateStoreActionAcker(batchedAcker, stateStorage)
 
+			actionQueue, err := queue.NewActionQueue(stateStorage.Queue(), stateStorage)
+			if err != nil {
+				return nil, nil, nil, fmt.Errorf("unable to initialize action queue: %w", err)
+			}
+
+			if initialUpgradeDetails == nil {
+				// initial upgrade details  are nil (normally the caller supplies the ones from the marker file at this point),
+				// hence, extract any scheduled upgrade details from the action queue.
+				initialUpgradeDetails = dispatcher.GetScheduledUpgradeDetails(log, actionQueue.Actions(), time.Now())
+			}
+
 			// TODO: stop using global state
-			managed, err = newManagedConfigManager(ctx, log, agentInfo, cfg, store, runtime, fleetInitTimeout, paths.Top(), client, fleetAcker, actionAcker, retrier, stateStorage, upgrader)
+			managed, err = newManagedConfigManager(ctx, log, agentInfo, cfg, store, runtime, fleetInitTimeout, paths.Top(), client, fleetAcker, actionAcker, retrier, stateStorage, actionQueue, upgrader)
 			if err != nil {
 				return nil, nil, nil, err
 			}
@@ -224,8 +238,16 @@ func New(
 		return nil, nil, nil, errors.New(err, "failed to initialize composable controller")
 	}
 
+<<<<<<< HEAD
 	otelManager := otelmanager.NewOTelManager(log.Named("otel_manager"))
 	coord := coordinator.New(log, cfg, logLevel, agentInfo, specs, reexec, upgrader, runtime, configMgr, varsManager, caps, monitor, isManaged, otelManager, actionAcker, compModifiers...)
+=======
+	otelManager, err := otelmanager.NewOTelManager(log.Named("otel_manager"), logLevel, baseLogger, otelmanager.EmbeddedExecutionMode, agentInfo, monitor.ComponentMonitoringConfig)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("failed to create otel manager: %w", err)
+	}
+	coord := coordinator.New(log, cfg, logLevel, agentInfo, specs, reexec, upgrader, runtime, configMgr, varsManager, caps, monitor, isManaged, otelManager, actionAcker, initialUpgradeDetails, compModifiers...)
+>>>>>>> ff8047180 (fix: scheduled upgrade details state (#9562))
 	if managed != nil {
 		// the coordinator requires the config manager as well as in managed-mode the config manager requires the
 		// coordinator, so it must be set here once the coordinator is created

internal/pkg/agent/application/application_test.go

@@ -64,6 +64,7 @@ func TestLimitsLog(t *testing.T) {
 		time.Millisecond,  // fleetInitTimeout
 		true,              // disable monitoring
 		nil,               // no configuration overrides
+		nil,
 	)
 	require.NoError(t, err)
 

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

@@ -378,7 +378,29 @@ type UpdateComponentChange struct {
 }
 
 // New creates a new coordinator.
+<<<<<<< HEAD
 func New(logger *logger.Logger, cfg *configuration.Configuration, logLevel logp.Level, agentInfo info.Agent, specs component.RuntimeSpecs, reexecMgr ReExecManager, upgradeMgr UpgradeManager, runtimeMgr RuntimeManager, configMgr ConfigManager, varsMgr VarsManager, caps capabilities.Capabilities, monitorMgr MonitorManager, isManaged bool, otelMgr OTelManager, fleetAcker acker.Acker, modifiers ...ComponentsModifier) *Coordinator {
+=======
+func New(
+	logger *logger.Logger,
+	cfg *configuration.Configuration,
+	logLevel logp.Level,
+	agentInfo info.Agent,
+	specs component.RuntimeSpecs,
+	reexecMgr ReExecManager,
+	upgradeMgr UpgradeManager,
+	runtimeMgr RuntimeManager,
+	configMgr ConfigManager,
+	varsMgr VarsManager,
+	caps capabilities.Capabilities,
+	monitorMgr MonitorManager,
+	isManaged bool,
+	otelMgr OTelManager,
+	fleetAcker acker.Acker,
+	initialUpgradeDetails *details.Details,
+	modifiers ...ComponentsModifier,
+) *Coordinator {
+>>>>>>> ff8047180 (fix: scheduled upgrade details state (#9562))
 	var fleetState cproto.State
 	var fleetMessage string
 	if !isManaged {
@@ -387,11 +409,12 @@ func New(logger *logger.Logger, cfg *configuration.Configuration, logLevel logp.
 		fleetMessage = "Not enrolled into Fleet"
 	}
 	state := State{
-		State:        agentclient.Starting,
-		Message:      "Starting",
-		FleetState:   fleetState,
-		FleetMessage: fleetMessage,
-		LogLevel:     logLevel,
+		State:          agentclient.Starting,
+		Message:        "Starting",
+		FleetState:     fleetState,
+		FleetMessage:   fleetMessage,
+		LogLevel:       logLevel,
+		UpgradeDetails: initialUpgradeDetails,
 	}
 	c := &Coordinator{
 		logger:     logger,

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

@@ -1096,7 +1096,7 @@ func createCoordinator(t testing.TB, ctx context.Context, opts ...CoordinatorOpt
 		acker = &fakeActionAcker{}
 	}
 
-	coord := New(l, nil, logp.DebugLevel, ai, specs, &fakeReExecManager{}, upgradeManager, rm, cfgMgr, varsMgr, caps, monitoringMgr, o.managed, otelMgr, acker)
+	coord := New(l, nil, logp.DebugLevel, ai, specs, &fakeReExecManager{}, upgradeManager, rm, cfgMgr, varsMgr, caps, monitoringMgr, o.managed, otelMgr, acker, nil)
 	return coord, cfgMgr, varsMgr
 }