[Integration Test Framework] fix createTempDir and flaky tests (#5409)

createTempDir register a test cleanup function to remove the folder it
created, however, on Windows, this folder sometimes fails to be
removed because there are still open file handlers for the files
within the folder.

We fix this problem calling install.RemovePath that will retry removing
the folder for about 2s. This is a very similar approach to
what Go's t.TempDir does.

Fix the flakiness from TestUpgradeHandler* tests by re-working the
mockUpgradeManager, now it accepts a function for its Upgrade method
and their implementation is goroutine safe

(cherry picked from commit 1242e71)

# Conflicts:
#	internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go
#	pkg/testing/fixture.go
#	testing/integration/event_logging_test.go

Author: belimawr

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

@@ -6,6 +6,8 @@ package handlers
 
 import (
 	"context"
+	"errors"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -26,7 +28,19 @@ import (
 )
 
 type mockUpgradeManager struct {
+<<<<<<< HEAD
 	msgChan chan string
+=======
+	UpgradeFn func(
+		ctx context.Context,
+		version string,
+		sourceURI string,
+		action *fleetapi.ActionUpgrade,
+		details *details.Details,
+		skipVerifyOverride bool,
+		skipDefaultPgp bool,
+		pgpBytes ...string) (reexec.ShutdownCallbackFn, error)
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }
 
 func (u *mockUpgradeManager) Upgradeable() bool {
@@ -37,6 +51,7 @@ func (u *mockUpgradeManager) Reload(rawConfig *config.Config) error {
 	return nil
 }
 
+<<<<<<< HEAD
 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, err error) {
 	select {
 	case <-time.After(2 * time.Second):
@@ -46,6 +61,27 @@ func (u *mockUpgradeManager) Upgrade(ctx context.Context, version string, source
 		u.msgChan <- "canceled " + version
 		return nil, ctx.Err()
 	}
+=======
+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) {
+
+	return u.UpgradeFn(
+		ctx,
+		version,
+		sourceURI,
+		action,
+		details,
+		skipVerifyOverride,
+		skipDefaultPgp,
+		pgpBytes...)
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }
 
 func (u *mockUpgradeManager) Ack(ctx context.Context, acker acker.Acker) error {
@@ -65,7 +101,11 @@ func TestUpgradeHandler(t *testing.T) {
 	log, _ := logger.New("", false)
 
 	agentInfo := &info.AgentInfo{}
+<<<<<<< HEAD
 	msgChan := make(chan string)
+=======
+	upgradeCalledChan := make(chan struct{})
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 
 	// Create and start the coordinator
 	c := coordinator.New(
@@ -75,7 +115,25 @@ func TestUpgradeHandler(t *testing.T) {
 		agentInfo,
 		component.RuntimeSpecs{},
 		nil,
+<<<<<<< HEAD
 		&mockUpgradeManager{msgChan: msgChan},
+=======
+		&mockUpgradeManager{
+			UpgradeFn: func(
+				ctx context.Context,
+				version string,
+				sourceURI string,
+				action *fleetapi.ActionUpgrade,
+				details *details.Details,
+				skipVerifyOverride bool,
+				skipDefaultPgp bool,
+				pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+				upgradeCalledChan <- struct{}{}
+				return nil, nil
+			},
+		},
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 		nil, nil, nil, nil, nil, false)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
@@ -86,8 +144,13 @@ func TestUpgradeHandler(t *testing.T) {
 	ack := noopacker.New()
 	err := u.Handle(ctx, &a, ack)
 	require.NoError(t, err)
-	msg := <-msgChan
-	require.Equal(t, "completed 8.3.0", msg)
+
+	// Make sure this test does not dead lock or wait for too long
+	select {
+	case <-time.Tick(50 * time.Millisecond):
+		t.Fatal("mockUpgradeManager.Upgrade was not called")
+	case <-upgradeCalledChan:
+	}
 }
 
 func TestUpgradeHandlerSameVersion(t *testing.T) {
@@ -99,17 +162,45 @@ func TestUpgradeHandlerSameVersion(t *testing.T) {
 	log, _ := logger.New("", false)
 
 	agentInfo := &info.AgentInfo{}
+<<<<<<< HEAD
 	msgChan := make(chan string)
+=======
+	upgradeCalledChan := make(chan struct{})
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 
 	// Create and start the Coordinator
+	upgradeCalled := atomic.Bool{}
 	c := coordinator.New(
 		log,
 		configuration.DefaultConfiguration(),
 		logger.DefaultLogLevel,
 		agentInfo,
 		component.RuntimeSpecs{},
 		nil,
+<<<<<<< HEAD
 		&mockUpgradeManager{msgChan: msgChan},
+=======
+		&mockUpgradeManager{
+			UpgradeFn: func(
+				ctx context.Context,
+				version string,
+				sourceURI string,
+				action *fleetapi.ActionUpgrade,
+				details *details.Details,
+				skipVerifyOverride bool,
+				skipDefaultPgp bool,
+				pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+				if upgradeCalled.CompareAndSwap(false, true) {
+					upgradeCalledChan <- struct{}{}
+					return nil, nil
+				}
+				err := errors.New("mockUpgradeManager.Upgrade called more than once")
+				t.Error(err.Error())
+				return nil, err
+			},
+		},
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 		nil, nil, nil, nil, nil, false)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
@@ -122,8 +213,18 @@ func TestUpgradeHandlerSameVersion(t *testing.T) {
 	err2 := u.Handle(ctx, &a, ack)
 	require.NoError(t, err1)
 	require.NoError(t, err2)
+<<<<<<< HEAD
 	msg := <-msgChan
 	require.Equal(t, "completed 8.3.0", msg)
+=======
+
+	// Make sure this test does not dead lock or wait for too long
+	select {
+	case <-time.Tick(50 * time.Millisecond):
+		t.Fatal("mockUpgradeManager.Upgrade was not called")
+	case <-upgradeCalledChan:
+	}
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }
 
 func TestUpgradeHandlerNewVersion(t *testing.T) {
@@ -133,9 +234,13 @@ func TestUpgradeHandlerNewVersion(t *testing.T) {
 	defer cancel()
 
 	log, _ := logger.New("", false)
+	upgradeCalledChan := make(chan string)
 
 	agentInfo := &info.AgentInfo{}
+<<<<<<< HEAD
 	msgChan := make(chan string)
+=======
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 
 	// Create and start the Coordinator
 	c := coordinator.New(
@@ -145,7 +250,31 @@ func TestUpgradeHandlerNewVersion(t *testing.T) {
 		agentInfo,
 		component.RuntimeSpecs{},
 		nil,
+<<<<<<< HEAD
 		&mockUpgradeManager{msgChan: msgChan},
+=======
+		&mockUpgradeManager{
+			UpgradeFn: func(
+				ctx context.Context,
+				version string,
+				sourceURI string,
+				action *fleetapi.ActionUpgrade,
+				details *details.Details,
+				skipVerifyOverride bool,
+				skipDefaultPgp bool,
+				pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+				defer func() {
+					upgradeCalledChan <- version
+				}()
+				if version == "8.2.0" {
+					return nil, errors.New("upgrade to 8.2.0 will always fail")
+				}
+
+				return nil, nil
+			},
+		},
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 		nil, nil, nil, nil, nil, false)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
@@ -156,13 +285,35 @@ func TestUpgradeHandlerNewVersion(t *testing.T) {
 	a2 := fleetapi.ActionUpgrade{Data: fleetapi.ActionUpgradeData{
 		Version: "8.5.0", SourceURI: "http://localhost"}}
 	ack := noopacker.New()
+
+	checkMsg := func(c <-chan string, expected, errMsg string) {
+		t.Helper()
+		// Make sure this test does not dead lock or wait for too long
+		// For some reason < 1s sometimes makes the test fail.
+		select {
+		case <-time.Tick(1300 * time.Millisecond):
+			t.Fatal("timed out waiting for Upgrade to return")
+		case msg := <-c:
+			require.Equal(t, expected, msg, errMsg)
+		}
+	}
+
+	// Send both upgrade actions, a1 will error before a2 succeeds
 	err1 := u.Handle(ctx, &a1, ack)
 	require.NoError(t, err1)
+<<<<<<< HEAD
 	time.Sleep(1 * time.Second)
 	err2 := u.Handle(ctx, &a2, ack)
 	require.NoError(t, err2)
 	msg1 := <-msgChan
 	require.Equal(t, "canceled 8.2.0", msg1)
 	msg2 := <-msgChan
 	require.Equal(t, "completed 8.5.0", msg2)
+=======
+	checkMsg(upgradeCalledChan, "8.2.0", "first call must be with version 8.2.0")
+
+	err2 := u.Handle(ctx, &a2, ack)
+	require.NoError(t, err2)
+	checkMsg(upgradeCalledChan, "8.5.0", "second call to Upgrade must be with version 8.5.0")
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }

pkg/testing/fixture.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/install"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/control"
 	"github.com/elastic/elastic-agent/pkg/control/v2/client"
@@ -1196,6 +1197,34 @@ func performConfigure(ctx context.Context, c client.Client, cfg string, timeout
 	return nil
 }
 
+<<<<<<< HEAD
+=======
+// createTempDir creates a temporary directory that will be
+// removed after the tests passes. If the test fails, the
+// directory is kept for further investigation.
+//
+// If the test is run with -v and fails the temporary directory is logged
+func createTempDir(t *testing.T) string {
+	tempDir, err := os.MkdirTemp("", strings.ReplaceAll(t.Name(), "/", "-"))
+	if err != nil {
+		t.Fatalf("failed to make temp directory: %s", err)
+	}
+
+	cleanup := func() {
+		if !t.Failed() {
+			if err := install.RemovePath(tempDir); err != nil {
+				t.Errorf("could not remove temp dir '%s': %s", tempDir, err)
+			}
+		} else {
+			t.Logf("Temporary directory %q preserved for investigation/debugging", tempDir)
+		}
+	}
+	t.Cleanup(cleanup)
+
+	return tempDir
+}
+
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 type AgentStatusOutput struct {
 	Info struct {
 		ID           string `json:"id"`

testing/integration/event_logging_test.go

@@ -75,7 +75,10 @@ func TestEventLogFile(t *testing.T) {
 		Local: true,
 		Sudo:  false,
 	})
+<<<<<<< HEAD
 
+=======
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 	ctx, cancel := testcontext.WithDeadline(
 		t,
 		context.Background(),