Elastic Agent upgrade: lock-free manual rollback

.mockery.yaml

@@ -30,10 +30,24 @@ packages:
     interfaces:
       Agent:
   github.com/elastic/elastic-agent/internal/pkg/agent/cmd:
+    config:
+      inpackage: True
+      with-expecter: True
+      dir: "{{.InterfaceDirRelative}}"
+      mockname: "{{.Mock}}{{.InterfaceName | firstUpper}}"
+      outpkg: "{{.PackageName}}"
+      filename: "{{.Mock | lower}}_{{.InterfaceName | lower}}_test.go"
     interfaces:
       agentWatcher:
-        config:
-          mockname: "AgentWatcher"
       installationModifier:
-        config:
-          mockname: "InstallationModifier"
+  github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade:
+    config:
+      inpackage: True
+      with-expecter: True
+      dir: "{{.InterfaceDirRelative}}"
+      mockname: "{{.Mock}}{{.InterfaceName | firstUpper}}"
+      outpkg: "{{.PackageName}}"
+      filename: "{{.Mock | lower}}_{{.InterfaceName | lower}}_test.go"
+    interfaces:
+      WatcherHelper:
+      watcherGrappler:

_meta/config/common.reference.p2.yml.tmpl

@@ -123,7 +123,7 @@ inputs:
 #   # rollback settings
 #   rollback:
 #       # duration in which an upgraded Agent may be manually rolled back.
-#       window: 168h
+#       window: 0
 
 # agent.process:
 #   # timeout for creating new processes. when process is not successfully created by this timeout

control_v2.proto

@@ -116,6 +116,9 @@ message UpgradeRequest {
   //
   // If provided Elastic Agent package embedded PGP key is not checked for signature during upgrade.
   bool skipDefaultPgp = 5;
+
+  // If true it indicates that we wish to rollback the current/last upgrade
+  bool rollback = 6;
 }
 
 // A upgrade response message.

elastic-agent.reference.yml

@@ -129,7 +129,7 @@ inputs:
 #   # rollback settings
 #   rollback:
 #       # duration in which an upgraded Agent may be manually rolled back.
-#       window: 168h
+#       window: 0
 
 # agent.process:
 #   # timeout for creating new processes. when process is not successfully created by this timeout

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

@@ -75,7 +75,7 @@ func (h *Upgrade) Handle(ctx context.Context, a fleetapi.Action, ack acker.Acker
 
 	go func() {
 		h.log.Infof("starting upgrade to version %s in background", action.Data.Version)
-		if err := h.coord.Upgrade(asyncCtx, action.Data.Version, action.Data.SourceURI, action, false, false); err != nil {
+		if err := h.coord.Upgrade(asyncCtx, action.Data.Version, false, action.Data.SourceURI, action, false, false); err != nil {
 			h.log.Errorf("upgrade to version %s failed: %v", action.Data.Version, err)
 			// If context is cancelled in getAsyncContext, the actions are acked there
 			if !errors.Is(asyncCtx.Err(), context.Canceled) {

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

@@ -48,15 +48,7 @@ func (u *mockUpgradeManager) Reload(rawConfig *config.Config) error {
 	return nil
 }
 
-func (u *mockUpgradeManager) Upgrade(
-	ctx context.Context,
-	version string,
-	sourceURI string,
-	action *fleetapi.ActionUpgrade,
-	details *details.Details,
-	skipVerifyOverride bool,
-	skipDefaultPgp bool,
-	pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+func (u *mockUpgradeManager) Upgrade(ctx context.Context, version string, rollback bool, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
 
 	return u.UpgradeFn(
 		ctx,

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

@@ -28,7 +28,7 @@ type actionCoordinator interface {
 
 type upgradeCoordinator interface {
 	actionCoordinator
-	Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) error
+	Upgrade(ctx context.Context, version string, rollback bool, sourceURI string, action *fleetapi.ActionUpgrade, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) error
 }
 
 type performActionFunc func(context.Context, component.Component, component.Unit, string, map[string]interface{}) (map[string]interface{}, error)

internal/pkg/agent/application/application.go

@@ -120,7 +120,7 @@ func New(
 
 	// monitoring is not supported in bootstrap mode https://github.com/elastic/elastic-agent/issues/1761
 	isMonitoringSupported := !disableMonitoring && cfg.Settings.V1MonitoringEnabled
-	upgrader, err := upgrade.NewUpgrader(log, cfg.Settings.DownloadConfig, agentInfo)
+	upgrader, err := upgrade.NewUpgrader(log, cfg.Settings.DownloadConfig, cfg.Settings.Upgrade, agentInfo, new(upgrade.AgentWatcherHelper))
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to create upgrader: %w", err)
 	}

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

@@ -85,7 +85,7 @@ type UpgradeManager interface {
 	Reload(rawConfig *config.Config) error
 
 	// Upgrade upgrades running agent.
-	Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (_ reexec.ShutdownCallbackFn, err error)
+	Upgrade(ctx context.Context, version string, rollback bool, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (_ reexec.ShutdownCallbackFn, err error)
 
 	// Ack is used on startup to check if the agent has upgraded and needs to send an ack for the action
 	Ack(ctx context.Context, acker acker.Acker) error
@@ -695,7 +695,7 @@ func (c *Coordinator) Migrate(ctx context.Context, action *fleetapi.ActionMigrat
 
 // Upgrade runs the upgrade process.
 // Called from external goroutines.
-func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) error {
+func (c *Coordinator) Upgrade(ctx context.Context, version string, rollback bool, sourceURI string, action *fleetapi.ActionUpgrade, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) error {
 	// early check outside of upgrader before overriding the state
 	if !c.upgradeMgr.Upgradeable() {
 		return ErrNotUpgradable
@@ -735,7 +735,7 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 	det := details.NewDetails(version, details.StateRequested, actionID)
 	det.RegisterObserver(c.SetUpgradeDetails)
 
-	cb, err := c.upgradeMgr.Upgrade(ctx, version, sourceURI, action, det, skipVerifyOverride, skipDefaultPgp, pgpBytes...)
+	cb, err := c.upgradeMgr.Upgrade(ctx, version, rollback, sourceURI, action, det, skipVerifyOverride, skipDefaultPgp, pgpBytes...)
 	if err != nil {
 		c.ClearOverrideState()
 		if errors.Is(err, upgrade.ErrUpgradeSameVersion) {

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

@@ -535,7 +535,7 @@ func TestUpgradeSameErrorAcked(t *testing.T) {
 
 	acker.On("Ack", mock.Anything, actionUpgrade).Return(nil)
 
-	require.NoError(t, coord.Upgrade(t.Context(), "9.0", "http://localhost", actionUpgrade, true, true))
+	require.NoError(t, coord.Upgrade(t.Context(), "9.0", false, "http://localhost", actionUpgrade, true, true))
 
 	acker.AssertCalled(t, "Ack", mock.Anything, actionUpgrade)
 }
@@ -917,7 +917,7 @@ func TestCoordinator_Upgrade(t *testing.T) {
 	require.NoError(t, err)
 	cfgMgr.Config(ctx, cfg)
 
-	err = coord.Upgrade(ctx, "9.0.0", "", nil, true, false)
+	err = coord.Upgrade(ctx, "9.0.0", false, "", nil, true, false)
 	require.ErrorIs(t, err, ErrNotUpgradable)
 	cancel()
 
@@ -954,7 +954,7 @@ func TestCoordinator_UpgradeDetails(t *testing.T) {
 	require.NoError(t, err)
 	cfgMgr.Config(ctx, cfg)
 
-	err = coord.Upgrade(ctx, "9.0.0", "", nil, true, false)
+	err = coord.Upgrade(ctx, "9.0.0", false, "", nil, true, false)
 	require.ErrorIs(t, expectedErr, err)
 	cancel()
 
@@ -1159,7 +1159,7 @@ func (f *fakeUpgradeManager) Reload(cfg *config.Config) error {
 	return nil
 }
 
-func (f *fakeUpgradeManager) Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (_ reexec.ShutdownCallbackFn, err error) {
+func (f *fakeUpgradeManager) Upgrade(ctx context.Context, version string, rollback bool, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (_ reexec.ShutdownCallbackFn, err error) {
 	f.upgradeCalled = true
 	if f.upgradeErr != nil {
 		return nil, f.upgradeErr

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

@@ -462,11 +462,7 @@ func TestCoordinatorReportsInvalidPolicy(t *testing.T) {
 		}
 	}()
 
-	upgradeMgr, err := upgrade.NewUpgrader(
-		log,
-		&artifact.Config{},
-		&info.AgentInfo{},
-	)
+	upgradeMgr, err := upgrade.NewUpgrader(log, &artifact.Config{}, nil, &info.AgentInfo{}, new(upgrade.AgentWatcherHelper))
 	require.NoError(t, err, "errored when creating a new upgrader")
 
 	// Channels have buffer length 1, so we don't have to run on multiple
@@ -1526,7 +1522,7 @@ func TestCoordinatorInitiatesUpgrade(t *testing.T) {
 	}
 
 	// Call upgrade and make sure the upgrade manager receives an Upgrade call
-	err := coord.Upgrade(ctx, "1.2.3", "", nil, false, false)
+	err := coord.Upgrade(ctx, "1.2.3", false, "", nil, false, false)
 	assert.True(t, upgradeMgr.upgradeCalled, "Coordinator Upgrade should call upgrade manager Upgrade")
 	assert.Equal(t, upgradeMgr.upgradeErr, err, "Upgrade should report upgrade manager error")
 

internal/pkg/agent/application/filelock/locker.go

@@ -39,6 +39,7 @@ func (a *AppLocker) TryLock() error {
 	if !locked {
 		return ErrAppAlreadyRunning
 	}
+
 	return nil
 }
 

internal/pkg/agent/application/filelock/testlocker/.gitignore

@@ -0,0 +1,2 @@
+# Ignore test binary
+testlocker

internal/pkg/agent/application/filelock/testlocker/main.go

@@ -0,0 +1,64 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+// This is a simple program that will lock an applocker using a file passed using the -lockfile option, used for testing file lock works properly.
+// os.Interrupt or signal.SIGTERM will make the program release the lock and exit
+package main
+
+import (
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/filelock"
+)
+
+const AcquiredLockLogFmt = "Acquired lock on file %s\n"
+
+const lockFileFlagName = "lockfile"
+const ignoreSignalFlagName = "ignoresignals"
+
+var lockFile = flag.String(lockFileFlagName, "", "path to lock file")
+var ignoreSignals = flag.Bool(ignoreSignalFlagName, false, "ignore signals")
+
+func main() {
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	flag.Parse()
+	if *lockFile == "" {
+		log.Fatalf("No lockfile specified. Please run %s -%s <path to lockfile>", os.Args[0], lockFileFlagName)
+	}
+
+	appLocker := filelock.NewAppLocker(filepath.Dir(*lockFile), filepath.Base(*lockFile))
+
+	err := appLocker.TryLock()
+	if err != nil {
+		log.Fatalf("Error locking %s: %s", *lockFile, err.Error())
+	}
+
+	defer func(aLocker *filelock.AppLocker) {
+
+		if unlockErr := aLocker.Unlock(); unlockErr != nil {
+			log.Printf("Error unlocking %s: %s", *lockFile, unlockErr.Error())
+		}
+	}(appLocker)
+
+	log.Printf(AcquiredLockLogFmt, *lockFile)
+
+	for {
+
+		s := <-signalChan
+		if *ignoreSignals {
+			log.Printf("Received signal %v , ignoring it...", s)
+			continue
+		}
+
+		log.Printf("Received signal %v , exiting...", s)
+		break
+	}
+}

internal/pkg/agent/application/upgrade/details/details.go

@@ -231,7 +231,8 @@ func (m Metadata) Equals(otherM Metadata) bool {
 		m.DownloadPercent == otherM.DownloadPercent &&
 		m.DownloadRate == otherM.DownloadRate &&
 		equalTimePointers(m.RetryUntil, otherM.RetryUntil) &&
-		m.RetryErrorMsg == otherM.RetryErrorMsg
+		m.RetryErrorMsg == otherM.RetryErrorMsg &&
+		m.Reason == otherM.Reason
 }
 
 func equalTimePointers(t, otherT *time.Time) bool {

internal/pkg/agent/application/upgrade/details/state.go

@@ -21,5 +21,6 @@ const (
 	StateFailed      State = "UPG_FAILED"
 
 	// List of well-known reasons for state transitions
-	ReasonWatchFailed = "watch failed"
+	ReasonWatchFailed    = "watch failed"
+	ReasonManualRollback = "manual rollback requested"
 )

internal/pkg/agent/application/upgrade/marker_watcher_test.go

@@ -126,6 +126,9 @@ details:
 			expectedDetails: &details.Details{
 				TargetVersion: "8.9.2",
 				State:         details.StateRollback,
+				Metadata: details.Metadata{
+					Reason: details.ReasonWatchFailed,
+				},
 			},
 		},
 		"same_version_with_details_some_state": {