Introduce WatcherHelper

Author: pchila

.mockery.yaml

@@ -37,3 +37,6 @@ packages:
       installationModifier:
         config:
           mockname: "InstallationModifier"
+  github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade:
+    interfaces:
+      WatcherHelper:

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, cfg.Settings.Upgrade, 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_unit_test.go

@@ -461,7 +461,7 @@ func TestCoordinatorReportsInvalidPolicy(t *testing.T) {
 		}
 	}()
 
-	upgradeMgr, err := upgrade.NewUpgrader(log, &artifact.Config{}, nil, &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/upgrade/step_download_test.go

@@ -91,7 +91,7 @@ func TestDownloadWithRetries(t *testing.T) {
 			return &mockDownloader{expectedDownloadPath, nil}, nil
 		}
 
-		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{})
+		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{}, new(AgentWatcherHelper))
 		require.NoError(t, err)
 
 		parsedVersion, err := agtversion.ParseVersion("8.9.0")
@@ -141,7 +141,7 @@ func TestDownloadWithRetries(t *testing.T) {
 			return nil, nil
 		}
 
-		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{})
+		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{}, new(AgentWatcherHelper))
 		require.NoError(t, err)
 
 		parsedVersion, err := agtversion.ParseVersion("8.9.0")
@@ -196,7 +196,7 @@ func TestDownloadWithRetries(t *testing.T) {
 			return nil, nil
 		}
 
-		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{})
+		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{}, new(AgentWatcherHelper))
 		require.NoError(t, err)
 
 		parsedVersion, err := agtversion.ParseVersion("8.9.0")
@@ -241,7 +241,7 @@ func TestDownloadWithRetries(t *testing.T) {
 			return &mockDownloader{"", errors.New("download failed")}, nil
 		}
 
-		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{})
+		u, err := NewUpgrader(testLogger, &settings, nil, &info.AgentInfo{}, new(AgentWatcherHelper))
 		require.NoError(t, err)
 
 		parsedVersion, err := agtversion.ParseVersion("8.9.0")

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

@@ -36,8 +36,6 @@ import (
 	"github.com/elastic/elastic-agent/pkg/control/v2/client"
 	"github.com/elastic/elastic-agent/pkg/control/v2/cproto"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
-	"github.com/elastic/elastic-agent/pkg/core/process"
-	"github.com/elastic/elastic-agent/pkg/utils"
 	agtversion "github.com/elastic/elastic-agent/pkg/version"
 	currentagtversion "github.com/elastic/elastic-agent/version"
 )
@@ -71,12 +69,21 @@ func init() {
 	}
 }
 
-// TODO substitute all the references to watcher with calls to the interface
+// WatcherHelper is an abstraction of operations that Upgrader will trigger on elastic-agent watcher.
+// This is defined to help with Upgrader testing and verify interactions with elastic-agent watcher
 type WatcherHelper interface {
+	// InvokeWatcher invokes an elastic-agent watcher using the agentExecutable passed as argument
 	InvokeWatcher(log *logger.Logger, agentExecutable string) (*exec.Cmd, error)
-	SelectWatcherExecutable(topDir string, agentInstalls ...agentInstall) string
+	// SelectWatcherExecutable will return the path to the newer elastic-agent executable that will be used to invoke the
+	// more recent watcher between the previous (the agent that started the upgrade) and current (the agent that will run after restart)
+	// agent installation
+	SelectWatcherExecutable(topDir string, previous agentInstall, current agentInstall) string
+	// WaitForWatcher will listen for changes to the update marker, waiting for the elastic-agent watcher to set UPG_WATCHING state
+	// in the upgrade details' metadata
 	WaitForWatcher(ctx context.Context, log *logger.Logger, markerFilePath string, waitTime time.Duration) error
-	TakeOverWatcher(ctx context.Context, topDir string) (*filelock.AppLocker, error)
+	// TakeOverWatcher will look for watcher processes and terminate them while at the same time trying to acquire the watcher AppLocker.
+	// It will return once it managed to get the AppLocker or with an error if the lock could not be acquired.
+	TakeOverWatcher(ctx context.Context, log *logger.Logger, topDir string) (*filelock.AppLocker, error)
 }
 
 // Upgrader performs an upgrade
@@ -88,6 +95,7 @@ type Upgrader struct {
 	upgradeable     bool
 	fleetServerURI  string
 	markerWatcher   MarkerWatcher
+	watcherHelper   WatcherHelper
 }
 
 // IsUpgradeable when agent is installed and running as a service or flag was provided.
@@ -98,14 +106,15 @@ func IsUpgradeable() bool {
 }
 
 // NewUpgrader creates an upgrader which is capable of performing upgrade operation
-func NewUpgrader(log *logger.Logger, settings *artifact.Config, upgradeConfig *configuration.UpgradeConfig, agentInfo info.Agent) (*Upgrader, error) {
+func NewUpgrader(log *logger.Logger, settings *artifact.Config, upgradeConfig *configuration.UpgradeConfig, agentInfo info.Agent, watcherHelper WatcherHelper) (*Upgrader, error) {
 	return &Upgrader{
 		log:             log,
 		settings:        settings,
 		upgradeSettings: upgradeConfig,
 		agentInfo:       agentInfo,
 		upgradeable:     IsUpgradeable(),
 		markerWatcher:   newMarkerFileWatcher(markerFilePath(paths.Data()), log),
+		watcherHelper:   watcherHelper,
 	}, nil
 }
 
@@ -368,16 +377,16 @@ func (u *Upgrader) Upgrade(ctx context.Context, version string, rollback bool, s
 		return nil, goerrors.Join(err, rollbackErr)
 	}
 
-	watcherExecutable := selectWatcherExecutable(paths.Top(), previous, current)
+	watcherExecutable := u.watcherHelper.SelectWatcherExecutable(paths.Top(), previous, current)
 
 	var watcherCmd *exec.Cmd
-	if watcherCmd, err = InvokeWatcher(u.log, watcherExecutable); err != nil {
+	if watcherCmd, err = u.watcherHelper.InvokeWatcher(u.log, watcherExecutable); err != nil {
 		u.log.Errorw("Rolling back: starting watcher failed", "error.message", err)
 		rollbackErr := rollbackInstall(ctx, u.log, paths.Top(), hashedDir, currentVersionedHome)
 		return nil, goerrors.Join(err, rollbackErr)
 	}
 
-	watcherWaitErr := waitForWatcher(ctx, u.log, markerFilePath(paths.Data()), watcherMaxWaitTime)
+	watcherWaitErr := u.watcherHelper.WaitForWatcher(ctx, u.log, markerFilePath(paths.Data()), watcherMaxWaitTime)
 	if watcherWaitErr != nil {
 		killWatcherErr := watcherCmd.Process.Kill()
 		rollbackErr := rollbackInstall(ctx, u.log, paths.Top(), hashedDir, currentVersionedHome)
@@ -408,7 +417,7 @@ func (u *Upgrader) forceRollbackToPreviousVersion(ctx context.Context, topDir st
 	}
 
 	// Invoke watcher again
-	_, err = InvokeWatcher(u.log, paths.BinaryPath(paths.VersionedHome(topDir), agentName))
+	_, err = u.watcherHelper.InvokeWatcher(u.log, paths.BinaryPath(paths.VersionedHome(topDir), agentName))
 	if err != nil {
 		return nil, fmt.Errorf("invoking watcher: %w", err)
 	}
@@ -418,7 +427,7 @@ func (u *Upgrader) forceRollbackToPreviousVersion(ctx context.Context, topDir st
 }
 
 func (u *Upgrader) PersistManualRollback(ctx context.Context, topDir string) error {
-	watcherApplock, err := u.takeOverWatcher(ctx, topDir)
+	watcherApplock, err := u.watcherHelper.TakeOverWatcher(ctx, u.log, topDir)
 	if err != nil {
 		return fmt.Errorf("taking over watcher processes: %w", err)
 	}
@@ -443,107 +452,6 @@ func (u *Upgrader) PersistManualRollback(ctx context.Context, topDir string) err
 	return nil
 }
 
-func (u *Upgrader) takeOverWatcher(ctx context.Context, topDir string) (*filelock.AppLocker, error) {
-
-	takeoverCtx, takeoverCancel := context.WithTimeout(ctx, 30*time.Second)
-	defer takeoverCancel()
-	go func() {
-		killingTicker := time.NewTicker(500 * time.Millisecond)
-		defer killingTicker.Stop()
-		for {
-			select {
-			case <-takeoverCtx.Done():
-				return
-			case <-killingTicker.C:
-				pids, err := utils.GetWatcherPIDs()
-				if err != nil {
-					u.log.Errorf("error listing watcher processes: %s", err)
-					continue
-				}
-
-				// this should be run continuously and concurrently attempting to get the app locker
-				for _, pid := range pids {
-					u.log.Debugf("attempting to kill watcher process with PID: %d", pid)
-					watcherProcess, findProcErr := os.FindProcess(pid)
-					if findProcErr != nil {
-						u.log.Errorf("error finding process with PID: %d: %s", pid, findProcErr)
-						continue
-					}
-					killProcErr := process.Terminate(watcherProcess)
-					if killProcErr != nil {
-						u.log.Errorf("error killing process with PID: %d: %s", pid, killProcErr)
-					}
-					u.log.Debugf("killed watcher process with PID: %d", pid)
-				}
-			}
-		}
-	}()
-
-	// we should retry to take over the AppLocker for 30s, but AppLocker interface is limited
-	takeOverTicker := time.NewTicker(100 * time.Millisecond)
-	defer takeOverTicker.Stop()
-	for {
-		select {
-		case <-takeoverCtx.Done():
-			return nil, fmt.Errorf("timed out taking over watcher applocker")
-		case <-takeOverTicker.C:
-			locker := filelock.NewAppLocker(topDir, "watcher.lock")
-			err := locker.TryLock()
-			if err != nil {
-				u.log.Errorf("error locking watcher applocker: %s", err)
-				continue
-			}
-			return locker, nil
-		}
-	}
-}
-
-func selectWatcherExecutable(topDir string, previous agentInstall, current agentInstall) string {
-	// check if the upgraded version is less than the previous (currently installed) version
-	if current.parsedVersion.Less(*previous.parsedVersion) {
-		// use the current agent executable for watch, if downgrading the old agent doesn't understand the current agent's path structure.
-		return paths.BinaryPath(filepath.Join(topDir, previous.versionedHome), agentName)
-	} else {
-		// use the new agent executable as it should be able to parse the new update marker
-		return paths.BinaryPath(filepath.Join(topDir, current.versionedHome), agentName)
-	}
-}
-
-func waitForWatcher(ctx context.Context, log *logger.Logger, markerFilePath string, waitTime time.Duration) error {
-	return waitForWatcherWithTimeoutCreationFunc(ctx, log, markerFilePath, waitTime, context.WithTimeout)
-}
-
-type createContextWithTimeout func(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc)
-
-func waitForWatcherWithTimeoutCreationFunc(ctx context.Context, log *logger.Logger, markerFilePath string, waitTime time.Duration, createTimeoutContext createContextWithTimeout) error {
-	// Wait for the watcher to be up and running
-	watcherContext, cancel := createTimeoutContext(ctx, waitTime)
-	defer cancel()
-
-	markerWatcher := newMarkerFileWatcher(markerFilePath, log)
-	err := markerWatcher.Run(watcherContext)
-	if err != nil {
-		return fmt.Errorf("error starting update marker watcher: %w", err)
-	}
-
-	log.Infof("waiting up to %s for upgrade watcher to set %s state in upgrade marker", waitTime, details.StateWatching)
-
-	for {
-		select {
-		case updMarker := <-markerWatcher.Watch():
-			if updMarker.Details != nil && updMarker.Details.State == details.StateWatching {
-				// watcher started and it is watching, all good
-				log.Infof("upgrade watcher set %s state in upgrade marker: exiting wait loop", details.StateWatching)
-				return nil
-			}
-
-		case <-watcherContext.Done():
-			log.Errorf("upgrade watcher did not start watching within %s or context has expired", waitTime)
-			return goerrors.Join(ErrWatcherNotStarted, watcherContext.Err())
-		}
-	}
-}
-
 // Ack acks last upgrade action
 func (u *Upgrader) Ack(ctx context.Context, acker acker.Acker) error {
 	// get upgrade action