Gracefully terminate watcher process on windows

Author: pchila

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

@@ -8,7 +8,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
 	"os/exec"
 	"path/filepath"
 	"time"
@@ -20,8 +19,6 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
 	"github.com/elastic/elastic-agent/pkg/control/v2/client"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
-	"github.com/elastic/elastic-agent/pkg/core/process"
-	"github.com/elastic/elastic-agent/pkg/utils"
 )
 
 const (
@@ -286,17 +283,14 @@ func (a AgentWatcherHelper) WaitForWatcher(ctx context.Context, log *logger.Logg
 }
 
 func (a AgentWatcherHelper) TakeOverWatcher(ctx context.Context, log *logger.Logger, topDir string) (*filelock.AppLocker, error) {
-	return takeOverWatcher(ctx, log, topDir, utils.GetWatcherPIDs, 30*time.Second, 500*time.Millisecond, 100*time.Millisecond)
+	return takeOverWatcher(ctx, log, topDir, 30*time.Second, 500*time.Millisecond, 100*time.Millisecond)
 }
 
-// watcherPIDsFetcher defines the type of function responsible for fetching watcher PIDs.
-// This will allow for easier testing of takeOverWatcher using fake binaries
-type watcherPIDsFetcher func() ([]int, error)
-
 // Private functions providing implementation of AgentWatcherHelper
-func takeOverWatcher(ctx context.Context, log *logger.Logger, topDir string, pidFetchFunc watcherPIDsFetcher, timeout time.Duration, watcherSweepInterval time.Duration, takeOverInterval time.Duration) (*filelock.AppLocker, error) {
+func takeOverWatcher(ctx context.Context, log *logger.Logger, topDir string, timeout time.Duration, watcherSweepInterval time.Duration, takeOverInterval time.Duration) (*filelock.AppLocker, error) {
 	takeoverCtx, takeoverCancel := context.WithTimeout(ctx, timeout)
 	defer takeoverCancel()
+
 	go func() {
 		sweepTicker := time.NewTicker(watcherSweepInterval)
 		defer sweepTicker.Stop()
@@ -305,27 +299,15 @@ func takeOverWatcher(ctx context.Context, log *logger.Logger, topDir string, pid
 			case <-takeoverCtx.Done():
 				return
 			case <-sweepTicker.C:
-				pids, err := pidFetchFunc()
+				cmd := createTakeDownWatcherCommand(takeoverCtx)
+				log.Debugf("launching takedown with %v", cmd.Args)
+				output, err := cmd.CombinedOutput()
+				log.Debugf("takedown output: %s", string(output))
 				if err != nil {
-					log.Errorf("error listing watcher processes: %s", err)
+					log.Errorf("error taking down watcher: %s", err)
 					continue
 				}
 
-				// this should be run continuously and concurrently attempting to get the app locker
-				for _, pid := range pids {
-					log.Debugf("attempting to kill watcher process with PID: %d", pid)
-					watcherProcess, findProcErr := os.FindProcess(pid)
-					if findProcErr != nil {
-						log.Errorf("error finding process with PID: %d: %s", pid, findProcErr)
-						continue
-					}
-					killProcErr := process.Terminate(watcherProcess)
-					if killProcErr != nil {
-						log.Errorf("error killing process with PID: %d: %s", pid, killProcErr)
-						continue
-					}
-					log.Debugf("killed watcher process with PID: %d", pid)
-				}
 			}
 		}
 	}()

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

@@ -0,0 +1,27 @@
+// 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.
+
+//go:build !windows
+
+package upgrade
+
+import (
+	"context"
+	"os"
+	"os/exec"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
+)
+
+func createTakeDownWatcherCommand(ctx context.Context) *exec.Cmd {
+	executable, _ := os.Executable()
+
+	// #nosec G204 -- user cannot inject any parameters to this command
+	cmd := exec.CommandContext(ctx, executable, watcherSubcommand,
+		"--path.config", paths.Config(),
+		"--path.home", paths.Top(),
+		"--takedown",
+	)
+	return cmd
+}

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

@@ -0,0 +1,34 @@
+// 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.
+
+//go:build windows
+
+package upgrade
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"syscall"
+
+	"golang.org/x/sys/windows"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
+)
+
+func createTakeDownWatcherCommand(ctx context.Context) *exec.Cmd {
+	executable, _ := os.Executable()
+
+	// #nosec G204 -- user cannot inject any parameters to this command
+	cmd := exec.CommandContext(ctx, executable, watcherSubcommand,
+		"--path.config", paths.Config(),
+		"--path.home", paths.Top(),
+		"--takedown",
+	)
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		// https://learn.microsoft.com/en-us/windows/win32/procthread/process-creation-flags
+		CreationFlags: windows.DETACHED_PROCESS,
+	}
+	return cmd
+}

internal/pkg/agent/cmd/watch.go

@@ -16,6 +16,7 @@ import (
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/configure"
 	"github.com/elastic/elastic-agent/pkg/control/v2/client"
+	"github.com/elastic/elastic-agent/pkg/utils"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/filelock"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
@@ -42,7 +43,7 @@ func newWatchCommandWithArgs(_ []string, streams *cli.IOStreams) *cobra.Command
 		Use:   "watch",
 		Short: "Watch the Elastic Agent for failures and initiate rollback",
 		Long:  `This command watches Elastic Agent for failures and initiates rollback if necessary.`,
-		Run: func(_ *cobra.Command, _ []string) {
+		Run: func(c *cobra.Command, _ []string) {
 			cfg := getConfig(streams)
 			log, err := configuredLogger(cfg, watcherName)
 			if err != nil {
@@ -53,14 +54,25 @@ func newWatchCommandWithArgs(_ []string, streams *cli.IOStreams) *cobra.Command
 			// Make sure to flush any buffered logs before we're done.
 			defer log.Sync() //nolint:errcheck // flushing buffered logs is best effort.
 
+			takedown, _ := c.Flags().GetBool("takedown")
+			if takedown {
+				err = takedownWatcher(log, utils.GetWatcherPIDs)
+				if err != nil {
+					log.Errorf("error taking down watcher: %v", err)
+					os.Exit(5)
+				}
+				return
+			}
+
 			if err := watchCmd(log, paths.Top(), cfg.Settings.Upgrade.Watcher, new(upgradeAgentWatcher), new(upgradeInstallationModifier)); err != nil {
 				log.Errorw("Watch command failed", "error.message", err)
 				fmt.Fprintf(streams.Err, "Watch command failed: %v\n%s\n", err, troubleshootMessage())
 				os.Exit(4)
 			}
 		},
 	}
-
+	cmd.Flags().BoolP("takedown", "t", false, "Take down the running watcher")
+	cmd.Flags().MarkHidden("takedown") //nolint:errcheck // not required
 	return cmd
 }
 
@@ -104,7 +116,7 @@ func watchCmd(log *logp.Logger, topDir string, cfg *configuration.UpgradeWatcher
 
 	if marker.DesiredOutcome == upgrade.OUTCOME_ROLLBACK {
 		// TODO: there should be some sanity check in rollback functions like the installation we are going back to should exist and work
-		log.Info("rolling back because of DesiredOutcome=%s", marker.DesiredOutcome.String())
+		log.Infof("rolling back because of DesiredOutcome=%s", marker.DesiredOutcome.String())
 		err = installModifier.Rollback(context.Background(), log, client.New(), paths.Top(), marker.PrevVersionedHome, marker.PrevHash)
 		if err != nil {
 			return fmt.Errorf("rolling back: %w", err)
@@ -118,7 +130,7 @@ func watchCmd(log *logp.Logger, topDir string, cfg *configuration.UpgradeWatcher
 			marker.Details = details.NewDetails(marker.Version, details.StateRollback, actionID)
 		}
 		marker.Details.SetStateWithReason(details.StateRollback, details.ReasonManualRollback)
-		err := upgrade.SaveMarker(dataDir, marker, true)
+		err = upgrade.SaveMarker(dataDir, marker, true)
 		if err != nil {
 			return fmt.Errorf("saving marker after rolling back: %w", err)
 		}

internal/pkg/agent/cmd/watch_impl.go

@@ -48,7 +48,7 @@ func watch(ctx context.Context, tilGrace time.Duration, errorCheckInterval time.
 	go agtWatcher.Run(ctx)
 
 	signals := make(chan os.Signal, 1)
-	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGHUP)
 
 	graceTimer := time.NewTimer(tilGrace)
 	defer graceTimer.Stop()

internal/pkg/agent/cmd/watch_impl_notwindows.go

@@ -0,0 +1,43 @@
+// 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.
+
+//go:build !windows
+
+package cmd
+
+import (
+	"os"
+	"syscall"
+)
+
+func takedownWatcher(log *logger.Logger, pidFetchFunc watcherPIDsFetcher) error {
+	pids, err := pidFetchFunc()
+	if err != nil {
+		return fmt.Errorf("error listing watcher processes: %s", err)
+	}
+
+	ownPID := os.Getpid()
+
+	for _, pid := range pids {
+
+		if pid == ownPID {
+			continue
+		}
+
+		log.Debugf("attempting to terminate watcher process with PID: %d", pid)
+
+		process, err := os.FindProcess(pid)
+		if err != nil {
+			log.Errorf("error finding watcher process with PID: %d: %s", pid, err)
+			continue
+		}
+
+		err = process.Signal(syscall.SIGTERM)
+		if err != nil {
+			log.Errorf("error killing watcher process with PID: %d: %s", pid, err)
+			continue
+		}
+
+	}
+}

internal/pkg/agent/cmd/watch_impl_windows.go

@@ -0,0 +1,109 @@
+// 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.
+
+//go:build windows
+
+package cmd
+
+import (
+	"fmt"
+	"os"
+	"unsafe"
+
+	gowindows "golang.org/x/sys/windows"
+
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+)
+
+// watcherPIDsFetcher defines the type of function responsible for fetching watcher PIDs.
+// This will allow for easier testing of takeOverWatcher using fake binaries
+type watcherPIDsFetcher func() ([]int, error)
+
+var (
+	kernel32API = gowindows.NewLazySystemDLL("kernel32.dll")
+
+	freeConsoleProc           = kernel32API.NewProc("FreeConsole")
+	attachConsoleProc         = kernel32API.NewProc("AttachConsole")
+	procGetConsoleProcessList = kernel32API.NewProc("GetConsoleProcessList")
+)
+
+func takedownWatcher(log *logger.Logger, pidFetchFunc watcherPIDsFetcher) error {
+	pids, err := pidFetchFunc()
+	if err != nil {
+		return fmt.Errorf("error listing watcher processes: %s", err)
+	}
+
+	ownPID := os.Getpid()
+
+	for _, pid := range pids {
+
+		if pid == ownPID {
+			continue
+		}
+
+		log.Debugf("attempting to terminate watcher process with PID: %d", pid)
+		// define an anonymous function in order to leverage the defer() for freeing console and other housekeeping
+		func() {
+
+			r1, _, consoleErr := freeConsoleProc.Call()
+			if r1 == 0 {
+				log.Errorf("error preemptively detaching from console: %s", consoleErr)
+			}
+
+			r1, _, consoleErr = attachConsoleProc.Call(uintptr(pid))
+			if r1 == 0 {
+				log.Errorf("error attaching console to watcher process with PID: %d -> %s", pid, consoleErr)
+				return
+			}
+			log.Infof("successfully attached console with PID: %d", pid)
+
+			defer func() {
+				r1, _, consoleErr = freeConsoleProc.Call()
+				if r1 == 0 {
+					log.Errorf("error detaching from console: %s", consoleErr)
+				} else {
+					log.Infof("successfully detached from console of PID: %d", pid)
+				}
+			}()
+
+			list, consoleErr := GetConsoleProcessList()
+			if consoleErr != nil {
+				log.Errorf("error listing console processes: %s", consoleErr)
+			}
+
+			log.Infof("Own PID: %d, Watcher pid %d, Process list on console: %v", os.Getpid(), pid, list)
+
+			// Normally we would want to send the Ctrl+Break event only to the watcher process but due to the fact that
+			// the parent process of the watcher has already terminated, we have to hug it tightly and take it down with us
+			// by specifying processGroupID=0
+			killProcErr := gowindows.GenerateConsoleCtrlEvent(gowindows.CTRL_BREAK_EVENT, 0)
+
+			if killProcErr != nil {
+				log.Errorf("error terminating process with PID: %d: %s", pid, killProcErr)
+				return
+			}
+		}()
+
+	}
+	return nil
+}
+
+// GetConsoleProcessList retrieves the list of process IDs attached to the current console
+func GetConsoleProcessList() ([]uint32, error) {
+	// Allocate a buffer for PIDs
+	const maxProcs = 64
+	pids := make([]uint32, maxProcs)
+
+	r1, _, err := procGetConsoleProcessList.Call(
+		uintptr(unsafe.Pointer(&pids[0])),
+		uintptr(maxProcs),
+	)
+
+	count := uint32(r1)
+	if count == 0 {
+		return nil, err
+	}
+
+	return pids[:count], nil
+}