Update serverless support in integration tests, expand test tooling, add support for beats testing

.buildkite/pipeline.yml

@@ -33,7 +33,7 @@ steps:
 
   - label: "Serverless integration test"
     key: "serverless-integration-tests"
-    command: ".buildkite/scripts/steps/integration_tests.sh serverless integration:single TestMonitoringLogsShipped"
+    command: ".buildkite/scripts/steps/integration_tests.sh serverless integration:single TestMonitoringLogsShipped" #right now, run a single test in serverless mode as a sort of smoke test, instead of re-running the entire suite
     artifact_paths:
       - "build/TEST-**"
       - "build/diagnostics/*"

pkg/testing/ess/provisioner.go

@@ -82,8 +82,10 @@ func (p *provisioner) Provision(ctx context.Context, requests []runner.StackRequ
 		results[r] = resp
 	}
 
-	// wait 15 minutes for all stacks to be ready
-	readyCtx, readyCancel := context.WithTimeout(ctx, 15*time.Minute)
+	// set a long timeout
+	// this context travels up to the magefile, clients that want a shorter timeout can set
+	// it via mage's -t flag
+	readyCtx, readyCancel := context.WithTimeout(ctx, 25*time.Minute)
 	defer readyCancel()
 
 	g, gCtx := errgroup.WithContext(readyCtx)

pkg/testing/ess/serverless_provision.go

@@ -95,7 +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},
+			Internal:      map[string]interface{}{"deployment_id": proj.ID},
 		}
 		stacks = append(stacks, newStack)
 		prov.stacksMut.Lock()

pkg/testing/fixture.go

@@ -275,20 +275,94 @@ func ExtractArtifact(l Logger, artifactFile, outputDir string) error {
 	return nil
 }
 
+// RunBeat runs the given given beat
+// the beat will run until an error, or the given timeout is reached
+func (f *Fixture) RunBeat(ctx context.Context) error {
+	if f.binaryName == "elastic-agent" {
+		return errors.New("RunBeat() can't be run against elastic-agent")
+	}
+
+	var err error
+	err = f.EnsurePrepared(ctx)
+	if err != nil {
+		return fmt.Errorf("error preparing beat: %w", err)
+	}
+
+	var logProxy Logger
+	if f.logOutput {
+		logProxy = f.t
+	}
+	stdOut := newLogWatcher(logProxy)
+	stdErr := newLogWatcher(logProxy)
+	args := []string{"run", "-e", "-c", filepath.Join(f.workDir, fmt.Sprintf("%s.yml", f.binaryName))}
+
+	args = append(args, f.additionalArgs...)
+
+	proc, err := process.Start(
+		f.binaryPath(),
+		process.WithContext(ctx),
+		process.WithArgs(args),
+		process.WithCmdOptions(attachOutErr(stdOut, stdErr)))
+
+	if err != nil {
+		return fmt.Errorf("failed to spawn %s: %w", f.binaryName, err)
+	}
+
+	killProc := func() {
+		_ = proc.Kill()
+		<-proc.Wait()
+	}
+
+	var doneChan <-chan time.Time
+	if f.runLength != 0 {
+		doneChan = time.After(f.runLength)
+	}
+
+	stopping := false
+	for {
+		select {
+		case <-ctx.Done():
+			killProc()
+			return ctx.Err()
+		case ps := <-proc.Wait():
+			if stopping {
+				return nil
+			}
+			return fmt.Errorf("elastic-agent exited unexpectedly with exit code: %d", ps.ExitCode())
+		case err := <-stdOut.Watch():
+			if !f.allowErrs {
+				// no errors allowed
+				killProc()
+				return fmt.Errorf("elastic-agent logged an unexpected error: %w", err)
+			}
+		case err := <-stdErr.Watch():
+			if !f.allowErrs {
+				// no errors allowed
+				killProc()
+				return fmt.Errorf("elastic-agent logged an unexpected error: %w", err)
+			}
+		case <-doneChan:
+			if !stopping {
+				// trigger the stop
+				stopping = true
+				_ = proc.Stop()
+			}
+		}
+	}
+}
+
 // Run runs the provided binary.
 //
-// If `states` are provided and the binary is Elastic Agent, agent runs until each state has been reached. Once reached the
+// If `states` are provided, agent runs until each state has been reached. Once reached the
 // Elastic Agent is stopped. If at any time the Elastic Agent logs an error log and the Fixture is not started
 // with `WithAllowErrors()` then `Run` will exit early and return the logged error.
 //
 // If no `states` are provided then the Elastic Agent runs until the context is or the timeout specified with WithRunLength is reached.
 func (f *Fixture) Run(ctx context.Context, states ...State) error {
-	isBeat := false
 	if f.binaryName != "elastic-agent" {
-		isBeat = true
+		return errors.New("Run() can only be used with elastic-agent, use RunBeat()")
 	}
-
-	if !isBeat && f.installed {
+	if f.installed {
 		return errors.New("fixture is installed; cannot be run")
 	}
 
@@ -302,9 +376,6 @@ func (f *Fixture) Run(ctx context.Context, states ...State) error {
 	defer cancel()
 
 	var smInstance *stateMachine
-	if states != nil && isBeat {
-		return errors.New("states not supported when running against beats")
-	}
 	if states != nil {
 		smInstance, err = newStateMachine(states)
 		if err != nil {
@@ -316,16 +387,15 @@ func (f *Fixture) Run(ctx context.Context, states ...State) error {
 	var agentClient client.Client
 	var stateCh chan *client.AgentState
 	var stateErrCh chan error
-	if !isBeat {
-		cAddr, err := control.AddressFromPath(f.operatingSystem, f.workDir)
-		if err != nil {
-			return fmt.Errorf("failed to get control protcol address: %w", err)
-		}
-		agentClient = client.New(client.WithAddress(cAddr))
-		f.setClient(agentClient)
-		defer f.setClient(nil)
-		stateCh, stateErrCh = watchState(ctx, agentClient, f.connectTimout)
+
+	cAddr, err := control.AddressFromPath(f.operatingSystem, f.workDir)
+	if err != nil {
+		return fmt.Errorf("failed to get control protcol address: %w", err)
 	}
+	agentClient = client.New(client.WithAddress(cAddr))
+	f.setClient(agentClient)
+	defer f.setClient(nil)
+	stateCh, stateErrCh = watchState(ctx, agentClient, f.connectTimout)
 
 	var logProxy Logger
 	if f.logOutput {
@@ -335,9 +405,7 @@ func (f *Fixture) Run(ctx context.Context, states ...State) error {
 	stdErr := newLogWatcher(logProxy)
 
 	args := []string{"run", "-e", "--disable-encrypted-store", "--testing-mode"}
-	if isBeat {
-		args = []string{"run", "-e", "-c", filepath.Join(f.workDir, fmt.Sprintf("%s.yml", f.binaryName))}
-	}
+
 	args = append(args, f.additionalArgs...)
 
 	proc, err := process.Start(

testing/integration/monitoring_logs_test.go

@@ -30,7 +30,6 @@ import (
 
 func TestMonitoringLogsShipped(t *testing.T) {
 	info := define.Require(t, define.Requirements{
-		OS:    []define.OS{{Type: define.Linux}},
 		Stack: &define.Stack{},
 		Local: false,
 		Sudo:  true,
@@ -86,8 +85,8 @@ func TestMonitoringLogsShipped(t *testing.T) {
 	docs := findESDocs(t, func() (estools.Documents, error) {
 		return estools.GetLogsForDatastream(info.ESClient, "elastic_agent.metricbeat")
 	})
-	require.NotZero(t, len(docs.Hits.Hits))
 	t.Logf("metricbeat: Got %d documents", len(docs.Hits.Hits))
+	require.NotZero(t, len(docs.Hits.Hits))
 
 	// Stage 4: make sure all components are healthy
 	t.Log("Making sure all components are healthy")
@@ -181,7 +180,7 @@ func findESDocs(t *testing.T, findFn func() (estools.Documents, error)) estools.
 			docs, err = findFn()
 			return err == nil
 		},
-		3*time.Minute,
+		10*time.Minute,
 		15*time.Second,
 	)