Elastic Agent upgrade: lock-free manual rollback

.mockery.yaml

@@ -33,10 +33,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_test.go

@@ -56,15 +56,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/application.go

@@ -124,7 +124,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
@@ -700,6 +700,7 @@ type upgradeOpts struct {
 	skipDefaultPgp     bool
 	pgpBytes           []string
 	preUpgradeCallback func(ctx context.Context, log *logger.Logger, action *fleetapi.ActionUpgrade) error
+	rollback           bool
 }
 
 type UpgradeOpt func(*upgradeOpts)
@@ -728,6 +729,12 @@ func WithPreUpgradeCallback(preUpgradeCallback func(ctx context.Context, log *lo
 	}
 }
 
+func WithRollback(rollback bool) UpgradeOpt {
+	return func(opts *upgradeOpts) {
+		opts.rollback = rollback
+	}
+}
+
 // Upgrade runs the upgrade process.
 // Called from external goroutines.
 func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, opts ...UpgradeOpt) error {
@@ -741,7 +748,8 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 	var err error
 	for i := 0; i < 5; i++ {
 		s := c.State()
-		if s.State != agentclient.Upgrading {
+		// if we are not already upgrading or if the incoming is a rollback request while the watcher is running, we can continue processing
+		if s.State != agentclient.Upgrading || (uOpts.rollback && s.UpgradeDetails != nil && s.UpgradeDetails.State == details.StateWatching) {
 			err = nil
 			break
 		}
@@ -785,7 +793,7 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 		}
 	}
 
-	cb, err := c.upgradeMgr.Upgrade(ctx, version, sourceURI, action, det, uOpts.skipVerifyOverride, uOpts.skipDefaultPgp, uOpts.pgpBytes...)
+	cb, err := c.upgradeMgr.Upgrade(ctx, version, uOpts.rollback, sourceURI, action, det, uOpts.skipVerifyOverride, uOpts.skipDefaultPgp, uOpts.pgpBytes...)
 	if err != nil {
 		c.ClearOverrideState()
 		if errors.Is(err, upgrade.ErrUpgradeSameVersion) {

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

@@ -1204,7 +1204,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

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"
+	ReasonManualRollbackPattern = "manual rollback requested to version %s"
 )

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": {