e2e: add TestOpAMPWithEDOTCollector and refactor shared OpAMP setup

- Extract shared download/extract helpers into agent_download.go with
  caching (sha512 comparison), FileReplacer, ExtractFilter, and correct
  chmod after extraction
- Extract startFleetServerForOpAMP and writeOpAMPCollectorConfig helpers
  shared by both OpAMP tests
- Rename TestOpAMP → TestOpAMPWithUpstreamCollector
- Add TestOpAMPWithEDOTCollector: downloads elastic-agent package,
  runs elastic-agent otel subcommand, verifies EDOT Collector enrolls
  in Fleet Server over OpAMP

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ycombinator

testing/e2e/const.go

@@ -7,4 +7,3 @@ package e2e
 const binaryName = "fleet-server"                //nolint:unused // work around to get platform specific binary name for tests
 const agentName = "elastic-agent"                //nolint:unused // work around to get platform specific binary name for tests
 const agentDevName = "elastic-development-agent" //nolint:unused // work around to get platform specific binary name for tests
-const edotCollectorName = "elastic-agent-otelcol" //nolint:unused // EDOT collector binary name bundled inside elastic-agent package

testing/e2e/const_windows.go

@@ -7,4 +7,3 @@ package e2e
 const binaryName = "fleet-server.exe"
 const agentName = "elastic-agent.exe"
 const agentDevName = "elastic-development-agent.exe"
-const edotCollectorName = "elastic-agent-otelcol.exe"

testing/e2e/stand_alone_test.go

@@ -581,60 +581,82 @@ func (suite *StandAloneSuite) TestAPMInstrumentation() {
 	cmd.Wait()
 }
 
-// TestOpAMP ensures that the OpAMP endpoint in Fleet Server works as expected by installing
-// an OTel Collector, configuring it with the OpAMP extension, and having it connect to Fleet
-// Server using OpAMP, and verifying that Fleet Server responds to this request with an HTTP
-// 200 OK status response.
-func (suite *StandAloneSuite) TestOpAMP() {
-	// Create a config file from a template in the test temp dir
-	dir := suite.T().TempDir()
+// startFleetServerForOpAMP creates the fleet-server config from stand-alone-opamp.tpl,
+// starts the fleet-server binary, waits for it to be healthy, fetches an enrollment token
+// for "dummy-policy", and enrolls a dummy agent (to ensure .fleet-agents exists before any
+// OpAMP collector connects). It returns the enrollment API key. Fleet-server is stopped when
+// ctx is cancelled; the caller owns the context lifetime.
+func (suite *StandAloneSuite) startFleetServerForOpAMP(ctx context.Context, dir, staticTokenKey string) string {
+	suite.T().Helper()
 	tpl, err := template.ParseFiles(filepath.Join("testdata", "stand-alone-opamp.tpl"))
 	suite.Require().NoError(err)
 	f, err := os.Create(filepath.Join(dir, "config.yml"))
 	suite.Require().NoError(err)
 	err = tpl.Execute(f, map[string]interface{}{
 		"Hosts":          suite.ESHosts,
 		"ServiceToken":   suite.ServiceToken,
-		"StaticTokenKey": "opamp-e2e-test-key",
+		"StaticTokenKey": staticTokenKey,
 	})
 	f.Close()
 	suite.Require().NoError(err)
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
-	defer cancel()
-
-	// Run the fleet-server binary
 	cmd := exec.CommandContext(ctx, suite.binaryPath, "-c", filepath.Join(dir, "config.yml"))
-	cmd.Cancel = func() error {
-		return cmd.Process.Signal(syscall.SIGTERM)
-	}
+	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
 	cmd.Env = []string{"GOCOVERDIR=" + suite.CoverPath}
 	err = cmd.Start()
 	suite.Require().NoError(err)
-	defer cmd.Wait()
+	suite.T().Cleanup(func() { cmd.Wait() })
 
 	suite.FleetServerStatusOK(ctx, "http://localhost:8220")
 
 	apiKey := suite.GetEnrollmentTokenForPolicyID(ctx, "dummy-policy")
+	tester := api_version.NewClientAPITesterCurrent(suite.Scaffold, "http://localhost:8220", apiKey)
+	tester.Enroll(ctx, apiKey)
+	return apiKey
+}
 
-	// Make sure the OpAMP endpoint works.
+// writeOpAMPCollectorConfig renders otelcol-opamp.tpl into dir/configFile and returns
+// the full path to the written file.
+func (suite *StandAloneSuite) writeOpAMPCollectorConfig(dir, configFile, instanceUID, apiKey string) string {
+	suite.T().Helper()
+	tpl, err := template.ParseFiles(filepath.Join("testdata", "otelcol-opamp.tpl"))
+	suite.Require().NoError(err)
+	path := filepath.Join(dir, configFile)
+	f, err := os.Create(path)
+	suite.Require().NoError(err)
+	err = tpl.Execute(f, map[string]interface{}{
+		"OpAMP": map[string]string{
+			"InstanceUID": instanceUID,
+			"APIKey":      apiKey,
+		},
+	})
+	f.Close()
+	suite.Require().NoError(err)
+	return path
+}
+
+// TestOpAMP ensures that the OpAMP endpoint in Fleet Server works as expected by installing
+// an OTel Collector, configuring it with the OpAMP extension, and having it connect to Fleet
+// Server using OpAMP, and verifying that Fleet Server responds to this request with an HTTP
+// 200 OK status response.
+func (suite *StandAloneSuite) TestOpAMPWithUpstreamCollector() {
+	dir := suite.T().TempDir()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	apiKey := suite.startFleetServerForOpAMP(ctx, dir, "opamp-e2e-test-key")
+
+	// Make sure the OpAMP endpoint works before proceeding to build the collector.
 	req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:8220/v1/opamp", nil)
 	suite.Require().NoError(err)
 	req.Header.Set("Authorization", "ApiKey "+apiKey)
 	req.Header.Set("Content-Type", "application/x-protobuf")
-
 	resp, err := suite.Client.Do(req)
 	suite.Require().NoError(err)
 	resp.Body.Close()
 	suite.Require().Equal(http.StatusOK, resp.StatusCode)
 
-	// Enroll a dummy agent to initialize the .fleet-agents index before the OTel Collector connects.
-	// Without this, findEnrolledAgent fails with index_not_found_exception when the OTel Collector
-	// sends its first AgentToServer message, because .fleet-agents doesn't exist yet in a fresh
-	// standalone fleet-server environment (unlike agent-managed fleet-server which self-enrolls).
-	tester := api_version.NewClientAPITesterCurrent(suite.Scaffold, "http://localhost:8220", apiKey)
-	tester.Enroll(ctx, apiKey)
-
 	// Clone OTel Collector contrib repository (shallow clone of main branch)
 	cloneDir := filepath.Join(dir, "opentelemetry-collector-contrib")
 	suite.T().Logf("Cloning opentelemetry-collector-contrib (main) to %s", cloneDir)
@@ -670,22 +692,11 @@ func (suite *StandAloneSuite) TestOpAMP() {
 	// Configure it with the OpAMP extension
 	instanceUID := "019b8d7a-2da8-7657-b52d-492a9de33319"
 	suite.T().Logf("Configuring OTel Collector with OpAMP extension (instanceUID=%s)", instanceUID)
-	tpl, err = template.ParseFiles(filepath.Join("testdata", "otelcol-opamp.tpl"))
-	suite.Require().NoError(err)
-	f, err = os.Create(filepath.Join(dir, "otelcol.yml"))
-	suite.Require().NoError(err)
-	err = tpl.Execute(f, map[string]interface{}{
-		"OpAMP": map[string]string{
-			"InstanceUID": instanceUID,
-			"APIKey":      apiKey,
-		},
-	})
-	f.Close()
-	suite.Require().NoError(err)
+	collectorConfig := suite.writeOpAMPCollectorConfig(dir, "otelcol.yml", instanceUID, apiKey)
 
 	// Start OTel Collector
 	suite.T().Log("Starting OTel Collector")
-	otelCmd := exec.CommandContext(ctx, otelBinaryPath, "--config", filepath.Join(dir, "otelcol.yml"))
+	otelCmd := exec.CommandContext(ctx, otelBinaryPath, "--config", collectorConfig)
 	otelCmd.Cancel = func() error {
 		return otelCmd.Process.Signal(syscall.SIGTERM)
 	}
@@ -711,105 +722,87 @@ func (suite *StandAloneSuite) TestOpAMP() {
 	suite.Contains(agentDoc.Tags, "otelcontribcol", "expected tags to contain otelcontribcol")
 }
 
-// TestEDOTOpAMP ensures that the EDOT (Elastic Distribution of OpenTelemetry) Collector,
-// bundled inside the Elastic Agent package, can connect to Fleet Server over OpAMP and
-// enroll as an agent in the .fleet-agents index.
-func (suite *StandAloneSuite) TestEDOTOpAMP() {
-	// Create a config file from a template in the test temp dir
+// TestOpAMPWithEDOTCollector ensures that the EDOT Collector can connect to Fleet Server
+// over OpAMP and enroll as an agent in the .fleet-agents index.
+func (suite *StandAloneSuite) TestOpAMPWithEDOTCollector() {
 	dir := suite.T().TempDir()
-	tpl, err := template.ParseFiles(filepath.Join("testdata", "stand-alone-opamp.tpl"))
-	suite.Require().NoError(err)
-	f, err := os.Create(filepath.Join(dir, "config.yml"))
-	suite.Require().NoError(err)
-	err = tpl.Execute(f, map[string]interface{}{
-		"Hosts":          suite.ESHosts,
-		"ServiceToken":   suite.ServiceToken,
-		"StaticTokenKey": "edot-opamp-e2e-test-key",
-	})
-	f.Close()
-	suite.Require().NoError(err)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
-	defer cancel()
 
-	// Run the fleet-server binary
-	cmd := exec.CommandContext(ctx, suite.binaryPath, "-c", filepath.Join(dir, "config.yml"))
-	cmd.Cancel = func() error {
-		return cmd.Process.Signal(syscall.SIGTERM)
-	}
-	cmd.Env = []string{"GOCOVERDIR=" + suite.CoverPath}
-	err = cmd.Start()
-	suite.Require().NoError(err)
-	defer cmd.Wait()
-
-	suite.FleetServerStatusOK(ctx, "http://localhost:8220")
-
-	apiKey := suite.GetEnrollmentTokenForPolicyID(ctx, "dummy-policy")
-
-	// Enroll a dummy agent to initialize the .fleet-agents index before the EDOT Collector connects.
-	// Without this, WaitForAgentDoc fails with index_not_found_exception when the EDOT Collector
-	// sends its first AgentToServer message, because .fleet-agents doesn't exist yet in a fresh
-	// standalone fleet-server environment.
-	tester := api_version.NewClientAPITesterCurrent(suite.Scaffold, "http://localhost:8220", apiKey)
-	tester.Enroll(ctx, apiKey)
-
-	// Download and extract the Elastic Agent package to obtain the bundled EDOT Collector binary.
-	suite.T().Log("Downloading Elastic Agent package to extract EDOT Collector binary")
-	downloadCtx, downloadCancel := context.WithTimeout(ctx, 5*time.Minute)
+	// Download and extract the full Elastic Agent package before starting the timed
+	// portion of the test. The archive is cached on disk after the first run so this
+	// is fast on subsequent runs; extracting everything ensures all components
+	// (e.g. elastic-otel-collector) needed by elastic-agent otel are present.
+	suite.T().Log("Downloading Elastic Agent package")
+	agentExtractDir := filepath.Join(dir, "elastic-agent-package")
+	suite.Require().NoError(os.MkdirAll(agentExtractDir, 0755))
+	downloadCtx, downloadCancel := context.WithTimeout(context.Background(), 10*time.Minute)
 	defer downloadCancel()
 	rc := downloadElasticAgent(downloadCtx, suite.T(), suite.Client)
-	defer rc.Close()
-
-	agentExtractDir := filepath.Join(dir, "elastic-agent-package")
-	err = os.MkdirAll(agentExtractDir, 0755)
-	suite.Require().NoError(err)
-	paths := extractAgentArchive(suite.T(), rc, agentExtractDir, nil)
+	paths := extractAgentArchive(suite.T(), rc, agentExtractDir, nil, nil)
 	rc.Close()
 
-	edotBinaryPath, ok := paths[edotCollectorName]
-	suite.Require().Truef(ok, "EDOT Collector binary %q not found in elastic-agent package", edotCollectorName)
-	suite.T().Logf("Found EDOT Collector binary at %s", edotBinaryPath)
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+
+	apiKey := suite.startFleetServerForOpAMP(ctx, dir, "edot-opamp-e2e-test-key")
+
+	agentBinaryPath, ok := paths[agentName]
+	suite.Require().Truef(ok, "elastic-agent binary %q not found in package", agentName)
+	suite.T().Logf("Found elastic-agent binary at %s", agentBinaryPath)
 
-	// Configure the EDOT Collector with the OpAMP extension
 	instanceUID := "029c9e8b-3eb9-8768-c63e-593b0ef44430"
 	suite.T().Logf("Configuring EDOT Collector with OpAMP extension (instanceUID=%s)", instanceUID)
-	tpl, err = template.ParseFiles(filepath.Join("testdata", "otelcol-opamp.tpl"))
-	suite.Require().NoError(err)
-	f, err = os.Create(filepath.Join(dir, "edot-otelcol.yml"))
-	suite.Require().NoError(err)
-	err = tpl.Execute(f, map[string]interface{}{
-		"OpAMP": map[string]string{
-			"InstanceUID": instanceUID,
-			"APIKey":      apiKey,
-		},
-	})
-	f.Close()
-	suite.Require().NoError(err)
+	collectorConfig := suite.writeOpAMPCollectorConfig(dir, "edot-otelcol.yml", instanceUID, apiKey)
 
-	// Start the EDOT Collector
-	suite.T().Log("Starting EDOT Collector")
-	edotCmd := exec.CommandContext(ctx, edotBinaryPath, "--config", filepath.Join(dir, "edot-otelcol.yml"))
+	// Start the EDOT Collector via `elastic-agent otel`
+	suite.T().Log("Starting EDOT Collector via elastic-agent otel")
+	edotOutputFile, err := os.CreateTemp(dir, "edot-output-*.log")
+	suite.Require().NoError(err)
+	edotCmd := exec.CommandContext(ctx, agentBinaryPath, "otel", "--config", collectorConfig)
 	edotCmd.Cancel = func() error {
 		return edotCmd.Process.Signal(syscall.SIGTERM)
 	}
-	edotCmd.Stdout = os.Stdout
-	edotCmd.Stderr = os.Stderr
+	edotCmd.Stdout = edotOutputFile
+	edotCmd.Stderr = edotOutputFile
 	err = edotCmd.Start()
 	suite.Require().NoError(err)
-	defer edotCmd.Wait()
+	// processExited owns the single Wait() call on edotCmd.
+	processExited := make(chan error, 1)
+	go func() { processExited <- edotCmd.Wait() }()
+	suite.T().Cleanup(func() {
+		// Wait for the process to exit (context cancellation will have killed it)
+		// before closing the output file and reading it. The 30s fallback handles
+		// the case where the early-exit path already consumed processExited.
+		select {
+		case <-processExited:
+		case <-time.After(30 * time.Second):
+		}
+		edotOutputFile.Close()
+		if out, readErr := os.ReadFile(edotOutputFile.Name()); readErr == nil {
+			suite.T().Logf("EDOT Collector output:\n%s", string(out))
+		}
+	})
+	// Detect immediate exit — if the process dies within 5s it's a startup failure.
+	select {
+	case exitErr := <-processExited:
+		edotOutputFile.Close()
+		if out, readErr := os.ReadFile(edotOutputFile.Name()); readErr == nil {
+			suite.T().Logf("EDOT Collector output:\n%s", string(out))
+		}
+		suite.Require().NoError(exitErr, "EDOT Collector exited prematurely")
+		return
+	case <-time.After(5 * time.Second):
+		// Process is still running after 5s — proceed
+	}
 
 	// Verify that the EDOT Collector was enrolled in Fleet by fetching its document from
 	// .fleet-agents and asserting on its contents.
 	suite.T().Logf("Waiting for EDOT agent %s to appear in .fleet-agents", instanceUID)
 	agentDoc := suite.WaitForAgentDoc(ctx, instanceUID)
 
 	suite.Equal(instanceUID, agentDoc.Agent.ID, "expected agent.id to match instanceUID")
-	versionOut, err := exec.Command(edotBinaryPath, "--version").Output()
-	suite.Require().NoError(err)
-	edotVersion := strings.TrimPrefix(strings.TrimSpace(string(versionOut)), "elastic-agent-otelcol version ")
 	suite.Equal("OPAMP", agentDoc.Type, "expected type to be OPAMP")
-	suite.Equal("elastic-agent-otelcol", agentDoc.Agent.Type, "expected agent.type to be elastic-agent-otelcol")
-	suite.Equal(edotVersion, agentDoc.Agent.Version, "expected agent.version to match EDOT Collector binary version")
+	suite.Equal("elastic-otel-collector", agentDoc.Agent.Type, "expected agent.type to be elastic-otel-collector")
+	suite.NotEmpty(agentDoc.Agent.Version, "expected agent.version to be set")
+	suite.Contains(agentDoc.Tags, "elastic-otel-collector", "expected tags to contain elastic-otel-collector")
 	suite.Equal(1, agentDoc.Revision, "expected policy_revision_idx to be 1")
-	suite.Contains(agentDoc.Tags, "elastic-agent-otelcol", "expected tags to contain elastic-agent-otelcol")
 }