Merge branch 'main' into fix-install-prevention-on-linux

Author: intxgo

.buildkite/bk.integration.pipeline.yml

@@ -44,6 +44,7 @@ steps:
           - fleet-privileged
           - standalone-upgrade
           - upgrade
+          - upgrade-flavor
           - install-uninstall
 
       - label: "Win2022:non-sudo:{{matrix}}"
@@ -95,6 +96,7 @@ steps:
         matrix:
           - default
           - upgrade
+          - upgrade-flavor
           - standalone-upgrade
           - fleet
           - fleet-endpoint-security
@@ -122,6 +124,7 @@ steps:
         matrix:
           - default
           - upgrade
+          - upgrade-flavor
           - standalone-upgrade
           - fleet
           - fleet-endpoint-security

changelog/fragments/1739375044-windows-change-how-process-handles-are-obtained-when-assigning-sub-processes-to-Job-objects..yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Change how Windows process handles are obtained when assigning sub-processes to Job objects.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: "elastic-agent"
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/owner/repo/6825
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

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

@@ -219,19 +219,26 @@ func getPackageMetadataFromZip(archivePath string) (packageMetadata, error) {
 }
 
 func skipFnFromZip(log *logger.Logger, r *zip.ReadCloser, detectedFlavor string, fileNamePrefix string, versionedHome string, registry map[string][]string) (install.SkipFn, error) {
+	acceptAllFn := func(relPath string) bool { return false }
 	if detectedFlavor == "" {
 		// no flavor don't skip anything
-		return func(relPath string) bool { return false }, nil
+		return acceptAllFn, nil
 	}
 
 	flavor, err := install.Flavor(detectedFlavor, "", registry)
 	if err != nil {
 		if errors.Is(err, install.ErrUnknownFlavor) {
 			// unknown flavor fallback to copy all
-			return func(relPath string) bool { return false }, nil
+			return acceptAllFn, nil
 		}
 		return nil, err
 	}
+
+	// no flavor loaded
+	if flavor.Name == "" {
+		return acceptAllFn, nil
+	}
+
 	specsInFlavor := install.SpecsForFlavor(flavor) // ignoring error flavor exists, it was loaded before
 
 	// fix versionedHome
@@ -455,20 +462,13 @@ func untar(log *logger.Logger, archivePath, dataDir string, flavor string) (Unpa
 }
 
 func skipFnFromTar(log *logger.Logger, archivePath string, flavor string, registry map[string][]string) (install.SkipFn, error) {
+	acceptAllFn := func(relPath string) bool { return false }
 	if flavor == "" {
 		// no flavor don't skip anything
-		return func(relPath string) bool { return false }, nil
+		return acceptAllFn, nil
 	}
 
 	fileNamePrefix := getFileNamePrefix(archivePath)
-	loadFlavor := func(flavor string) install.FlavorDefinition {
-		components, found := registry[flavor]
-		if !found {
-			return install.FlavorDefinition{}
-		}
-
-		return install.FlavorDefinition{Name: flavor, Components: components}
-	}
 
 	// scan tar archive for spec file and extract allowed paths
 	r, err := os.Open(archivePath)
@@ -485,7 +485,20 @@ func skipFnFromTar(log *logger.Logger, archivePath string, flavor string, regist
 	tr := tar.NewReader(zr)
 
 	var allowedPaths []string
-	flavorDefinition := loadFlavor(flavor)
+	flavorDefinition, err := install.Flavor(flavor, "", registry)
+	if err != nil {
+		if errors.Is(err, install.ErrUnknownFlavor) {
+			// unknown flavor fallback to copy all
+			return acceptAllFn, nil
+		}
+		return nil, err
+	}
+
+	// no flavor loaded
+	if flavorDefinition.Name == "" {
+		return acceptAllFn, nil
+	}
+
 	specs, err := specRegistry(flavorDefinition)
 	if err != nil {
 		return nil, err

internal/pkg/agent/cmd/install.go

@@ -11,8 +11,10 @@ import (
 	"os/exec"
 	"path/filepath"
 	"runtime"
+	"strings"
 
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/filelock"
@@ -84,6 +86,11 @@ would like the Agent to operate.
 func installCmd(streams *cli.IOStreams, cmd *cobra.Command) error {
 	var err error
 
+	if installServers, _ := cmd.Flags().GetBool(flagInstallServers); isFleetServerFlagProvided(cmd) && !installServers {
+		_ = cmd.Flags().Lookup(flagInstallServers).Value.Set("true") // this can fail only when parsing bool
+		fmt.Fprintf(streams.Out, "fleet-server installation detected, using --%s flag\n", flagInstallServers)
+	}
+
 	err = validateEnrollFlags(cmd)
 	if err != nil {
 		return fmt.Errorf("could not validate flags: %w", err)
@@ -356,6 +363,27 @@ func installCmd(streams *cli.IOStreams, cmd *cobra.Command) error {
 	return nil
 }
 
+func isFleetServerFlagProvided(cmd *cobra.Command) bool {
+	var fleetServerFlagPresent bool
+	cmd.Flags().VisitAll(func(f *pflag.Flag) {
+		if fleetServerFlagPresent {
+			return
+		}
+
+		if !strings.HasPrefix(f.Name, "fleet-server-") {
+			return
+		}
+
+		flag := cmd.Flags().Lookup(f.Name)
+		if flag != nil && flag.Changed {
+			fleetServerFlagPresent = true
+		}
+
+	})
+
+	return fleetServerFlagPresent
+}
+
 // execUninstall execs "elastic-agent uninstall --force" from the elastic agent installed on the system (found in PATH)
 func execUninstall(streams *cli.IOStreams, topPath string, binName string) error {
 	args := []string{

internal/pkg/agent/install/flavors.go

@@ -45,7 +45,7 @@ func UsedFlavor(topPath, defaultFlavor string) (string, error) {
 		}
 
 		// failed reading flavor, do not break behavior and apply none as widest
-		return "", err
+		return "", fmt.Errorf("failed reading flavor marker file: %w", err)
 	}
 
 	return string(content), nil
@@ -55,11 +55,15 @@ func Flavor(detectedFlavor string, registryPath string, flavorsRegistry map[stri
 	if flavorsRegistry == nil {
 		f, err := os.Open(registryPath)
 		if err != nil {
-			return FlavorDefinition{}, err
+			if os.IsNotExist(err) {
+				return FlavorDefinition{}, ErrUnknownFlavor
+			}
+
+			return FlavorDefinition{}, fmt.Errorf("failed opening flavor registry: %w", err)
 		}
 		manifest, err := v1.ParseManifest(f)
 		if err != nil {
-			return FlavorDefinition{}, err
+			return FlavorDefinition{}, fmt.Errorf("failed parsing flavor registry: %w", err)
 		}
 		defer f.Close()
 		flavorsRegistry = manifest.Package.Flavors
@@ -110,12 +114,12 @@ func ApplyFlavor(versionedHome string, flavor FlavorDefinition) error {
 		return nil
 	})
 	if err != nil {
-		return err
+		return fmt.Errorf("failed traversing components directory: %w", err)
 	}
 
 	for _, ftr := range filesToRemove {
 		if removeErr := os.RemoveAll(ftr); !os.IsNotExist(removeErr) {
-			err = removeErr
+			err = fmt.Errorf("failed cleaning components: %w", removeErr)
 		}
 	}
 
@@ -222,7 +226,7 @@ func subpathsForComponent(componentName, sourceComponentsDir string) ([]string,
 		if os.IsNotExist(err) {
 			return nil, nil
 		}
-		return nil, err
+		return nil, fmt.Errorf("failed reading spec file for component %q: %w", componentName, err)
 	}
 
 	return component.ParseComponentFiles(content, specFilename, true)

pkg/core/process/external_windows.go

@@ -10,6 +10,8 @@ import (
 	"os"
 	"syscall"
 	"time"
+
+	"golang.org/x/sys/windows"
 )
 
 const (
@@ -34,16 +36,17 @@ func externalProcess(proc *os.Process) {
 
 func isWindowsProcessExited(pid int) bool {
 	const desiredAccess = syscall.STANDARD_RIGHTS_READ | syscall.PROCESS_QUERY_INFORMATION | syscall.SYNCHRONIZE
-	h, err := syscall.OpenProcess(desiredAccess, false, uint32(pid))
+	h, err := windows.OpenProcess(desiredAccess, false, uint32(pid)) //nolint:gosec // G115 Conversion from int to uint32 is safe here.
 	if err != nil {
 		// failed to open handle, report exited
 		return true
 	}
+	defer windows.CloseHandle(h) //nolint:errcheck // No way to handle errors returned here so safe to ignore.
 
 	// get exit code, this returns immediately in case it is still running
 	// it returns exitCodeStillActive
 	var ec uint32
-	if err := syscall.GetExitCodeProcess(h, &ec); err != nil {
+	if err := windows.GetExitCodeProcess(h, &ec); err != nil {
 		// failed to contact, report exited
 		return true
 	}

pkg/core/process/job_windows.go

@@ -7,6 +7,7 @@
 package process
 
 import (
+	"fmt"
 	"os"
 	"unsafe"
 
@@ -76,12 +77,23 @@ func (job Job) Assign(p *os.Process) error {
 	if job == 0 || p == nil {
 		return nil
 	}
-	return windows.AssignProcessToJobObject(
-		windows.Handle(job),
-		windows.Handle((*process)(unsafe.Pointer(p)).Handle))
-}
 
-type process struct {
-	Pid    int
-	Handle uintptr
+	// To assign a process to a job, you need a handle to the process. Since os.Process provides no
+	// way to obtain it's underlying handle safely, get one with OpenProcess().
+	//   https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-openprocess
+	// This requires at least the PROCESS_SET_QUOTA and PROCESS_TERMINATE access rights.
+	//   https://learn.microsoft.com/en-us/windows/win32/api/jobapi2/nf-jobapi2-assignprocesstojobobject
+	desiredAccess := uint32(windows.PROCESS_SET_QUOTA | windows.PROCESS_TERMINATE)
+	processHandle, err := windows.OpenProcess(desiredAccess, false, uint32(p.Pid)) //nolint:gosec // G115 Conversion from int to uint32 is safe here.
+	if err != nil {
+		return fmt.Errorf("opening process handle: %w", err)
+	}
+	defer windows.CloseHandle(processHandle) //nolint:errcheck // No way to handle errors returned here so safe to ignore.
+
+	err = windows.AssignProcessToJobObject(windows.Handle(job), processHandle)
+	if err != nil {
+		return fmt.Errorf("assigning to job object: %w", err)
+	}
+
+	return nil
 }

pkg/testing/fixture.go

@@ -726,7 +726,7 @@ func (f *Fixture) ExecStatus(ctx context.Context, opts ...process.CmdOption) (Ag
 	status := AgentStatusOutput{}
 	if uerr := json.Unmarshal(out, &status); uerr != nil {
 		return AgentStatusOutput{},
-			fmt.Errorf("could not unmarshal agent status output: %w", errors.Join(uerr, err))
+			fmt.Errorf("could not unmarshal agent status output: %w:\n%s", errors.Join(uerr, err), out)
 	} else if status.IsZero() {
 		return status, fmt.Errorf("agent status output is empty: %w", err)
 	}

pkg/testing/fixture_install.go

@@ -200,7 +200,7 @@ func (f *Fixture) installFunc(ctx context.Context, installOpts *InstallOpts, sho
 
 	// check for running agents before installing, but only if not installed into a namespace whose point is allowing two agents at once.
 	if installOpts != nil && !installOpts.Develop && installOpts.Namespace == "" {
-		assert.Empty(f.t, getElasticAgentProcesses(f.t), "there should be no running agent at beginning of Install()")
+		assert.Empty(f.t, getElasticAgentAndAgentbeatProcesses(f.t), "there should be no running agent at beginning of Install()")
 	}
 
 	switch f.packageFormat {
@@ -287,19 +287,22 @@ func (f *Fixture) installNoPkgManager(ctx context.Context, installOpts *InstallO
 
 	f.t.Cleanup(func() {
 		// check for running agents after uninstall had a chance to run
-		processes := getElasticAgentProcesses(f.t)
-
 		// there can be a single agent left when using --develop mode
 		if f.installOpts != nil && f.installOpts.Namespace != "" {
-			assert.LessOrEqualf(f.t, len(processes), 1, "More than one agent left running at the end of the test when second agent in namespace %s was used: %v", f.installOpts.Namespace, processes)
+			// Only consider the main agent process and not sub-processes so that we can detect when
+			// multiple agents are running without needing to know the number of input sub-processes to expect.
+			agentProcesses := getElasticAgentProcesses(f.t)
+			assert.LessOrEqualf(f.t, len(agentProcesses), 1, "More than one agent left running at the end of the test when second agent in namespace %s was used: %v", f.installOpts.Namespace, agentProcesses)
 			// The agent left running has to be the non-development agent. The development agent should be uninstalled first as a convention.
-			if len(processes) > 0 {
-				assert.NotContainsf(f.t, processes[0].Cmdline, paths.InstallDirNameForNamespace(f.installOpts.Namespace),
-					"The agent installed into namespace %s was left running at the end of the test or was not uninstalled first: %v", f.installOpts.Namespace, processes)
+			if len(agentProcesses) > 0 {
+				assert.NotContainsf(f.t, agentProcesses[0].Cmdline, paths.InstallDirNameForNamespace(f.installOpts.Namespace),
+					"The agent installed into namespace %s was left running at the end of the test or was not uninstalled first: %v", f.installOpts.Namespace, agentProcesses)
 			}
 			return
 		}
 
+		// If not using an installation namespace, there should be no elastic-agent or agentbeat processes left running.
+		processes := getElasticAgentAndAgentbeatProcesses(f.t)
 		assert.Empty(f.t, processes, "there should be no running agent at the end of the test")
 	})
 
@@ -409,6 +412,13 @@ func getElasticAgentProcesses(t *gotesting.T) []runningProcess {
 	return getProcesses(t, `.*elastic\-agent.*`)
 }
 
+// Includes both the main elastic-agent process and the agentbeat sub-processes for ensuring
+// that no sub-processes are orphaned from their parent process and left running. This
+// primarily tests that Windows Job Object assignment works.
+func getElasticAgentAndAgentbeatProcesses(t *gotesting.T) []runningProcess {
+	return getProcesses(t, `.*(elastic\-agent|agentbeat).*`)
+}
+
 func getProcesses(t *gotesting.T, regex string) []runningProcess {
 	procStats := agentsystemprocess.Stats{
 		Procs: []string{regex},

pkg/testing/tools/check/check.go

@@ -47,6 +47,7 @@ func ConnectedToFleet(ctx context.Context, t *testing.T, fixture *integrationtes
 // for use with assert.Eventually or require.Eventually.
 func FleetAgentStatus(ctx context.Context,
 	t *testing.T,
+	fixture *integrationtest.Fixture,
 	client *kibana.Client,
 	policyID,
 	expectedStatus string) func() bool {
@@ -61,7 +62,13 @@ func FleetAgentStatus(ctx context.Context,
 			return true
 		}
 
-		t.Logf("Agent fleet status: %s", currentStatus)
+		agentStatus, err := fixture.ExecStatus(ctx)
+		if err != nil {
+			t.Logf("Agent fleet status: %s Error getting local status: %s", currentStatus, err)
+			return false
+		}
+
+		t.Logf("Agent fleet status: %s Local status: %v", currentStatus, agentStatus)
 		return false
 	}
 }