Miscellaneous fixes for agent testing and build processes (#3656)

* misc fixes for different agent testing issues

* remove creds from log line

* fix serverless deployment cleanup

* fix spelling

magefile.go

@@ -20,6 +20,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/elastic/e2e-testing/pkg/downloads"
@@ -493,7 +494,7 @@ func FixDRADockerArtifacts() error {
 }
 
 func getPackageName(beat, version, pkg string) (string, string) {
-	if _, ok := os.LookupEnv(snapshotEnv); ok {
+	if hasSnapshotEnv() {
 		version += "-SNAPSHOT"
 	}
 	return version, fmt.Sprintf("%s-%s-%s", beat, version, pkg)
@@ -958,20 +959,24 @@ func packageAgent(platforms []string, packagingFn func()) {
 			logrus.SetLevel(logrus.FatalLevel)
 
 			errGroup, ctx := errgroup.WithContext(context.Background())
+			completedDownloads := &atomic.Int32{}
 			for binary, project := range externalBinaries {
 				for _, platform := range platforms {
 					reqPackage := platformPackages[platform]
 					targetPath := filepath.Join(archivePath, reqPackage)
 					os.MkdirAll(targetPath, 0755)
 					newVersion, packageName := getPackageName(binary, packageVersion, reqPackage)
-					errGroup.Go(downloadBinary(ctx, project, packageName, binary, platform, newVersion, targetPath))
+					errGroup.Go(downloadBinary(ctx, project, packageName, binary, platform, newVersion, targetPath, completedDownloads))
 				}
 			}
 
 			err := errGroup.Wait()
 			if err != nil {
 				panic(err)
 			}
+			if completedDownloads.Load() == 0 {
+				panic(fmt.Sprintf("No packages were successfully downloaded. You may be building against an invalid or unreleased version. version=%s. If this is an unreleased version, try SNAPSHOT=true or EXTERNAL=false", packageVersion))
+			}
 		} else {
 			packedBeats := []string{"filebeat", "heartbeat", "metricbeat", "osquerybeat"}
 			// build from local repo, will assume beats repo is located on the same root level
@@ -1143,15 +1148,17 @@ func packageAgent(platforms []string, packagingFn func()) {
 
 // Helper that wraps the fetchBinaryFromArtifactsApi in a way that is compatible with the errgroup.Go() function.
 // Ensures the arguments are captured by value before starting the goroutine.
-func downloadBinary(ctx context.Context, project string, packageName string, binary string, platform string, version string, targetPath string) func() error {
+func downloadBinary(ctx context.Context, project string, packageName string, binary string, platform string, version string, targetPath string, compl *atomic.Int32) func() error {
 	return func() error {
 		_, err := downloads.FetchProjectBinary(ctx, project, packageName, binary, version, 3, false, targetPath, true)
 		if err != nil {
 			if strings.Contains(err.Error(), "not found") {
-				fmt.Printf("Done downloading %s: unsupported on %s, skipping\n", binary, platform)
+				fmt.Printf("Could not download %s: %s\n", binary, err)
 			} else {
 				return fmt.Errorf("FetchProjectBinary failed for %s on %s: %v", binary, platform, err)
 			}
+		} else {
+			compl.Add(1)
 		}
 
 		fmt.Printf("Done downloading %s\n", packageName)
@@ -2171,6 +2178,7 @@ func hasSnapshotEnv() bool {
 		return false
 	}
 	b, _ := strconv.ParseBool(snapshot)
+
 	return b
 }
 

pkg/testing/ess/deployment.go

@@ -10,6 +10,7 @@ import (
 	_ "embed"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"strings"
@@ -195,7 +196,8 @@ func (c *Client) ShutdownDeployment(ctx context.Context, deploymentID string) er
 	defer res.Body.Close()
 
 	if res.StatusCode != 200 {
-		return fmt.Errorf("got unexpected response code [%d] from deployment shutdown API", res.StatusCode)
+		resBytes, _ := io.ReadAll(res.Body)
+		return fmt.Errorf("got unexpected response code [%d] from deployment shutdown API: %s", res.StatusCode, string(resBytes))
 	}
 
 	return nil

pkg/testing/ess/serverless_provision.go

@@ -95,7 +95,10 @@ func (prov *ServerlessProvision) Provision(ctx context.Context, requests []runne
 			Kibana:        client.proj.Endpoints.Kibana,
 			Username:      client.proj.Credentials.Username,
 			Password:      client.proj.Credentials.Password,
-			Internal:      map[string]interface{}{"deployment_id": proj.ID},
+			Internal: map[string]interface{}{
+				"deployment_id":   proj.ID,
+				"deployment_type": proj.Type,
+			},
 		}
 		stacks = append(stacks, newStack)
 		prov.stacksMut.Lock()
@@ -136,15 +139,45 @@ func (prov *ServerlessProvision) Provision(ctx context.Context, requests []runne
 func (prov *ServerlessProvision) Clean(ctx context.Context, stacks []runner.Stack) error {
 	for _, stack := range stacks {
 		prov.stacksMut.RLock()
+		// because of the way the provisioner initializes,
+		// we can't guarantee that we have a valid client/stack setup, as we might have just re-initialized from a file.
+		// If that's the case, create a new client
 		stackRef, ok := prov.stacks[stack.ID]
 		prov.stacksMut.RUnlock()
+		// we can't reference the client, it won't be created when we just run mage:clean
+		// instead, grab the project ID from `stacks`, create a new client
 		if ok {
 			err := stackRef.client.DeleteDeployment()
 			if err != nil {
 				prov.log.Logf("error removing deployment: %w", err)
 			}
 		} else {
-			prov.log.Logf("error: could not find deployment for ID %s", stack.ID)
+			// create a new client
+			client := NewServerlessClient(prov.cfg.Region, "observability", prov.cfg.APIKey, prov.log)
+			dep_id, ok := stack.Internal["deployment_id"]
+			if !ok {
+				return fmt.Errorf("could not find deployment_id for serverless")
+			}
+			dep_id_str, ok := dep_id.(string)
+			if !ok {
+				return fmt.Errorf("deployment_id is not a string: %v", dep_id)
+			}
+			client.proj.ID = dep_id_str
+
+			dep_type, ok := stack.Internal["deployment_type"]
+			if !ok {
+				return fmt.Errorf("could not find deployment_type in stack for serverless")
+			}
+			dep_type_str, ok := dep_type.(string)
+			if !ok {
+				return fmt.Errorf("deployment_type is not a string: %v", dep_id_str)
+			}
+			client.proj.Type = dep_type_str
+			err := client.DeleteDeployment()
+			if err != nil {
+				return fmt.Errorf("error removing deployment after re-creating client: %w", err)
+			}
+
 		}
 	}
 	return nil

pkg/testing/runner/runner.go

@@ -381,7 +381,7 @@ func (r *Runner) runInstance(ctx context.Context, sshAuth ssh.AuthMethod, logger
 		env["KIBANA_HOST"] = stack.Kibana
 		env["KIBANA_USERNAME"] = stack.Username
 		env["KIBANA_PASSWORD"] = stack.Password
-		logger.Logf("Using Stack with Kibana host %s, %s/%s", stack.Kibana, stack.Username, stack.Password)
+		logger.Logf("Using Stack with Kibana host %s, credentials available under .integration-cache", stack.Kibana)
 	}
 
 	// set the go test flags

pkg/testing/tools/tools.go

@@ -10,7 +10,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/elastic-agent-libs/kibana"
 	atesting "github.com/elastic/elastic-agent/pkg/testing"
@@ -123,15 +123,8 @@ func InstallAgentForPolicy(ctx context.Context, t *testing.T,
 		timeout = time.Until(deadline)
 	}
 
-	assert.Eventually(
-		t,
-		check.FleetAgentStatus(t, kibClient, policyID, "online"),
-		timeout,
-		10*time.Second,
-		"Elastic Agent status is not online",
-	)
 	// Wait for Agent to be healthy
-	assert.Eventually(
+	require.Eventually(
 		t,
 		check.FleetAgentStatus(t, kibClient, policyID, "online"),
 		timeout,

testing/integration/endpoint_security_test.go

@@ -248,7 +248,8 @@ func testInstallAndUnenrollWithEndpointSecurity(t *testing.T, info *define.Info,
 	require.NoError(t, err)
 
 	t.Log("Installing Elastic Defend")
-	installElasticDefendPackage(t, info, policy.ID)
+	_, err = installElasticDefendPackage(t, info, policy.ID)
+	require.NoError(t, err)
 
 	t.Log("Polling for endpoint-security to become Healthy")
 	ctx, cancel := context.WithTimeout(context.Background(), endpointHealthPollingTimeout)
@@ -599,7 +600,7 @@ func agentAndEndpointAreHealthy(t *testing.T, ctx context.Context, agentClient c
 
 	// Ensure both the endpoint input and output units were found and healthy.
 	if !foundEndpointInputUnit || !foundEndpointOutputUnit {
-		t.Logf("State did not contain endpoint units. state: %+v", state)
+		t.Logf("State did not contain endpoint units (input: %v/output: %v) state: %+v. ", foundEndpointInputUnit, foundEndpointOutputUnit, state)
 		return false
 	}