Add beat serverless tests & related support

magefile.go

@@ -1540,6 +1540,31 @@ func (Integration) Single(ctx context.Context, testName string) error {
 	return integRunner(ctx, false, testName)
 }
 
+// Run beat serverless tests
+func (Integration) TestBeatServerless(ctx context.Context, beatname string) error {
+	beatBuildPath := filepath.Join("..", "beats", "x-pack", beatname, "build", "distributions")
+	err := os.Setenv("AGENT_BUILD_DIR", beatBuildPath)
+	if err != nil {
+		return fmt.Errorf("error setting build dir: %s", err)
+	}
+
+	// a bit of bypass logic; run as serverless by default
+	if os.Getenv("STACK_PROVISIONER") == "" {
+		err = os.Setenv("STACK_PROVISIONER", "serverless")
+		if err != nil {
+			return fmt.Errorf("error setting serverless stack var: %w", err)
+		}
+	} else if os.Getenv("STACK_PROVISIONER") == "ess" {
+		fmt.Printf(">>> Warning: running TestBeatServerless as stateful\n")
+	}
+
+	err = os.Setenv("TEST_BINARY_NAME", beatname)
+	if err != nil {
+		return fmt.Errorf("error setting binary name: %w", err)
+	}
+	return integRunner(ctx, false, "TestMetricbeatSeverless")
+}
+
 // PrepareOnRemote shouldn't be called locally (called on remote host to prepare it for testing)
 func (Integration) PrepareOnRemote() {
 	mg.Deps(mage.InstallGoTestTools)
@@ -1686,6 +1711,7 @@ func createTestRunner(matrix bool, singleTest string, goTestFlags string, batche
 	if err != nil {
 		return nil, err
 	}
+
 	agentStackVersion := os.Getenv("AGENT_STACK_VERSION")
 	agentVersion := os.Getenv("AGENT_VERSION")
 	if agentVersion == "" {
@@ -1703,6 +1729,7 @@ func createTestRunner(matrix bool, singleTest string, goTestFlags string, batche
 			agentVersion = fmt.Sprintf("%s-SNAPSHOT", agentVersion)
 		}
 	}
+
 	if agentStackVersion == "" {
 		agentStackVersion = agentVersion
 	}
@@ -1762,24 +1789,35 @@ func createTestRunner(matrix bool, singleTest string, goTestFlags string, batche
 		extraEnv["AGENT_KEEP_INSTALLED"] = os.Getenv("AGENT_KEEP_INSTALLED")
 	}
 
+	binaryName := os.Getenv("TEST_BINARY_NAME")
+	if binaryName == "" {
+		binaryName = "elastic-agent"
+	}
+
+	repoDir := os.Getenv("TEST_INTEG_REPO_PATH")
+	if repoDir == "" {
+		repoDir = "."
+	}
+
 	diagDir := filepath.Join("build", "diagnostics")
 	_ = os.MkdirAll(diagDir, 0755)
 
 	cfg := runner.Config{
-		AgentVersion:      agentVersion,
-		AgentStackVersion: agentStackVersion,
-		BuildDir:          agentBuildDir,
-		GOVersion:         goVersion,
-		RepoDir:           ".",
-		StateDir:          ".integration-cache",
-		DiagnosticsDir:    diagDir,
-		Platforms:         testPlatforms(),
-		Matrix:            matrix,
-		SingleTest:        singleTest,
-		VerboseMode:       mg.Verbose(),
-		Timestamp:         timestamp,
-		TestFlags:         goTestFlags,
-		ExtraEnv:          extraEnv,
+		AgentVersion:   agentVersion,
+		StackVersion:   agentStackVersion,
+		BuildDir:       agentBuildDir,
+		GOVersion:      goVersion,
+		RepoDir:        repoDir,
+		DiagnosticsDir: diagDir,
+		StateDir:       ".integration-cache",
+		Platforms:      testPlatforms(),
+		Matrix:         matrix,
+		SingleTest:     singleTest,
+		VerboseMode:    mg.Verbose(),
+		Timestamp:      timestamp,
+		TestFlags:      goTestFlags,
+		ExtraEnv:       extraEnv,
+		BinaryName:     binaryName,
 	}
 	ogcCfg := ogc.Config{
 		ServiceTokenPath: serviceTokenPath,

pkg/testing/define/define.go

@@ -78,19 +78,29 @@ func NewFixture(t *testing.T, version string, opts ...atesting.FixtureOpt) (*ate
 		buildsDir = filepath.Join(projectDir, "build", "distributions")
 	}
 
+	return NewFixtureWithBinary(t, version, "elastic-agent", buildsDir)
+
+}
+
+// NewFixture returns a new Elastic Agent testing fixture with a LocalFetcher and
+// the agent logging to the test logger.
+func NewFixtureWithBinary(t *testing.T, version string, binary string, buildsDir string, opts ...atesting.FixtureOpt) (*atesting.Fixture, error) {
 	ver, err := semver.ParseVersion(version)
 	if err != nil {
 		return nil, fmt.Errorf("%q is an invalid agent version: %w", version, err)
 	}
 
-	var f atesting.Fetcher
+	var binFetcher atesting.Fetcher
 	if ver.IsSnapshot() {
-		f = atesting.LocalFetcher(buildsDir, atesting.WithLocalSnapshotOnly())
+		binFetcher = atesting.LocalFetcher(buildsDir, atesting.WithLocalSnapshotOnly(), atesting.WithCustomBinaryName(binary))
 	} else {
-		f = atesting.LocalFetcher(buildsDir)
+		binFetcher = atesting.LocalFetcher(buildsDir, atesting.WithCustomBinaryName(binary))
 	}
 
-	opts = append(opts, atesting.WithFetcher(f), atesting.WithLogOutput())
+	opts = append(opts, atesting.WithFetcher(binFetcher), atesting.WithLogOutput())
+	if binary != "elastic-agent" {
+		opts = append(opts, atesting.WithBinaryName(binary))
+	}
 	return atesting.NewFixture(t, version, opts...)
 }
 

pkg/testing/ess/serverless.go

@@ -54,6 +54,13 @@ type Project struct {
 	} `json:"endpoints"`
 }
 
+// CredResetResponse contains the new auth details for a
+// stack credential reset
+type CredResetResponse struct {
+	Password string `json:"password"`
+	Username string `json:"username"`
+}
+
 // NewServerlessClient creates a new instance of the serverless client
 func NewServerlessClient(region, projectType, api string, logger runner.Logger) *ServerlessClient {
 	return &ServerlessClient{
@@ -97,6 +104,16 @@ func (srv *ServerlessClient) DeployStack(ctx context.Context, req ServerlessRequ
 		return Project{}, fmt.Errorf("error decoding JSON response: %w", err)
 	}
 	srv.proj = serverlessHandle
+
+	// as of 8/8-ish, the serverless ESS cloud no longer provides credentials on the first POST request, we must send an additional POST
+	// to reset the credentials
+	updated, err := srv.ResetCredentials(ctx)
+	if err != nil {
+		return serverlessHandle, fmt.Errorf("error resetting credentials: %w", err)
+	}
+	srv.proj.Credentials.Username = updated.Username
+	srv.proj.Credentials.Password = updated.Password
+
 	return serverlessHandle, nil
 }
 
@@ -181,27 +198,15 @@ func (srv *ServerlessClient) WaitForEndpoints(ctx context.Context) error {
 
 // WaitForElasticsearch waits until the ES endpoint is healthy
 func (srv *ServerlessClient) WaitForElasticsearch(ctx context.Context) error {
-	endpoint := fmt.Sprintf("%s/_cluster/health", srv.proj.Endpoints.Elasticsearch)
-	req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil)
+	req, err := http.NewRequestWithContext(ctx, "GET", srv.proj.Endpoints.Elasticsearch, nil)
 	if err != nil {
 		return fmt.Errorf("error creating HTTP request: %w", err)
 	}
 	req.SetBasicAuth(srv.proj.Credentials.Username, srv.proj.Credentials.Password)
 
+	// _cluster/health no longer works on serverless, just check response code
 	readyFunc := func(resp *http.Response) bool {
-		var health struct {
-			Status string `json:"status"`
-		}
-		err = json.NewDecoder(resp.Body).Decode(&health)
-		resp.Body.Close()
-		if err != nil {
-			srv.log.Logf("response decoding error: %v", err)
-			return false
-		}
-		if health.Status == "green" {
-			return true
-		}
-		return false
+		return resp.StatusCode == 200
 	}
 
 	err = srv.waitForRemoteState(ctx, req, time.Second*5, readyFunc)
@@ -243,6 +248,38 @@ func (srv *ServerlessClient) WaitForKibana(ctx context.Context) error {
 	return nil
 }
 
+// ResetCredentials resets the credentials for the given ESS instance
+func (srv *ServerlessClient) ResetCredentials(ctx context.Context) (CredResetResponse, error) {
+	resetURL := fmt.Sprintf("%s/api/v1/serverless/projects/%s/%s/_reset-credentials", serverlessURL, srv.projectType, srv.proj.ID)
+
+	resetHandler, err := http.NewRequestWithContext(ctx, "POST", resetURL, nil)
+	if err != nil {
+		return CredResetResponse{}, fmt.Errorf("error creating new httpRequest: %w", err)
+	}
+
+	resetHandler.Header.Set("Content-Type", "application/json")
+	resetHandler.Header.Set("Authorization", fmt.Sprintf("ApiKey %s", srv.api))
+
+	resp, err := http.DefaultClient.Do(resetHandler)
+	if err != nil {
+		return CredResetResponse{}, fmt.Errorf("error performing HTTP request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		p, _ := io.ReadAll(resp.Body)
+		return CredResetResponse{}, fmt.Errorf("Non-201 status code returned by server: %d, body: %s", resp.StatusCode, string(p))
+	}
+
+	updated := CredResetResponse{}
+	err = json.NewDecoder(resp.Body).Decode(&updated)
+	if err != nil {
+		return CredResetResponse{}, fmt.Errorf("error decoding JSON response: %w", err)
+	}
+
+	return updated, nil
+}
+
 func (srv *ServerlessClient) waitForRemoteState(ctx context.Context, httpHandler *http.Request, tick time.Duration, isReady func(*http.Response) bool) error {
 	timer := time.NewTimer(time.Millisecond)
 	// in cases where we get a timeout, also return the last error returned via HTTP

pkg/testing/ess/serverless_provision.go

@@ -80,7 +80,7 @@ func (prov *ServerlessProvision) Provision(ctx context.Context, requests []runne
 	for _, req := range requests {
 		client := NewServerlessClient(prov.cfg.Region, "observability", prov.cfg.APIKey, prov.log)
 		srvReq := ServerlessRequest{Name: req.ID, RegionID: prov.cfg.Region}
-		_, err := client.DeployStack(ctx, srvReq)
+		proj, err := client.DeployStack(ctx, srvReq)
 		if err != nil {
 			return nil, fmt.Errorf("error deploying stack for request %s: %w", req.ID, err)
 		}
@@ -95,6 +95,7 @@ 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{}{"ID": proj.ID},
 		}
 		stacks = append(stacks, newStack)
 		prov.stacksMut.Lock()

pkg/testing/fetcher_local.go

@@ -17,6 +17,7 @@ import (
 type localFetcher struct {
 	dir          string
 	snapshotOnly bool
+	binaryName   string
 }
 
 type localFetcherOpt func(f *localFetcher)
@@ -28,10 +29,18 @@ func WithLocalSnapshotOnly() localFetcherOpt {
 	}
 }
 
+// WithCustomBinaryName sets the binary to a custom name, the default is `elastic-agent`
+func WithCustomBinaryName(name string) localFetcherOpt {
+	return func(f *localFetcher) {
+		f.binaryName = name
+	}
+}
+
 // LocalFetcher returns a fetcher that pulls the binary of the Elastic Agent from a local location.
 func LocalFetcher(dir string, opts ...localFetcherOpt) Fetcher {
 	f := &localFetcher{
-		dir: dir,
+		dir:        dir,
+		binaryName: "elastic-agent",
 	}
 	for _, o := range opts {
 		o(f)
@@ -56,7 +65,7 @@ func (f *localFetcher) Fetch(_ context.Context, operatingSystem string, architec
 		return nil, fmt.Errorf("invalid version: %q: %w", ver, err)
 	}
 
-	mainBuildfmt := "elastic-agent-%s-%s"
+	mainBuildfmt := "%s-%s-%s"
 	if f.snapshotOnly && !ver.IsSnapshot() {
 		if ver.Prerelease() == "" {
 			ver = semver.NewParsedSemVer(ver.Major(), ver.Minor(), ver.Patch(), "SNAPSHOT", ver.BuildMetadata())
@@ -66,7 +75,7 @@ func (f *localFetcher) Fetch(_ context.Context, operatingSystem string, architec
 
 	}
 
-	mainBuild := fmt.Sprintf(mainBuildfmt, ver, suffix)
+	mainBuild := fmt.Sprintf(mainBuildfmt, f.binaryName, ver, suffix)
 	mainBuildPath := filepath.Join(f.dir, mainBuild)
 	build := mainBuild
 	buildPath := mainBuildPath