Merge branch 'main' into improve-TestKubernetesJournaldInput

belimawr · web-flow · commit 895d629ee06e · 2025-11-19T15:06:21.000-05:00
diff --git a/NOTICE-fips.txt b/NOTICE-fips.txt
@@ -787,11 +787,11 @@ Contents of probable licence file $GOMODCACHE/github.com/dolmen-go/contextio@v0.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/beats/v7
-Version: v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d
+Version: v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2
 Licence type (autodetected): Elastic
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/v7@v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d/LICENSE.txt:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/v7@v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2/LICENSE.txt:
 
 Source code in this repository is variously licensed under the Apache License
 Version 2.0, an Apache compatible license, or the Elastic License. Outside of
diff --git a/NOTICE.txt b/NOTICE.txt
@@ -787,11 +787,11 @@ Contents of probable licence file $GOMODCACHE/github.com/dolmen-go/contextio@v0.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/beats/v7
-Version: v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d
+Version: v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2
 Licence type (autodetected): Elastic
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/v7@v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d/LICENSE.txt:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/v7@v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2/LICENSE.txt:
 
 Source code in this repository is variously licensed under the Apache License
 Version 2.0, an Apache compatible license, or the Elastic License. Outside of
diff --git a/go.mod b/go.mod
@@ -14,7 +14,7 @@ require (
 	github.com/docker/docker v28.4.0+incompatible
 	github.com/docker/go-units v0.5.0
 	github.com/dolmen-go/contextio v0.0.0-20200217195037-68fc5150bcd5
-	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d
+	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2
 	github.com/elastic/cloud-on-k8s/v2 v2.0.0-20250327073047-b624240832ae
 	github.com/elastic/elastic-agent-autodiscover v0.10.0
 	github.com/elastic/elastic-agent-client/v7 v7.17.2
diff --git a/go.sum b/go.sum
@@ -511,8 +511,8 @@ github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumpti
 github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumption v1.1.0-elastic/go.mod h1:0vCBR1wgGwZeGmloJ+eCWIZF2S47grTXRzj2mftg2Nk=
 github.com/elastic/bayeux v1.0.5 h1:UceFq01ipmT3S8DzFK+uVAkbCdiPR0Bqei8qIGmUeY0=
 github.com/elastic/bayeux v1.0.5/go.mod h1:CSI4iP7qeo5MMlkznGvYKftp8M7qqP/3nzmVZoXHY68=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d h1:WiLFsGBR7P1ErPCUngBbOWU3poKF7WAdycwWcp5HeIA=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d/go.mod h1:c8O8MuFlYOkEbT1qdC/iaiTHD2u2mCrCJZtkl4+shTg=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2 h1:0VrGnx4wwU2K16wDOVAKxSI+TkJQQvYmQ+SJQQw/0rM=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2/go.mod h1:c8O8MuFlYOkEbT1qdC/iaiTHD2u2mCrCJZtkl4+shTg=
 github.com/elastic/cloud-on-k8s/v2 v2.0.0-20250327073047-b624240832ae h1:OiShmbWAyGU0MS0ADJWr1/QgeLIZliMk9xsrFicR3/s=
 github.com/elastic/cloud-on-k8s/v2 v2.0.0-20250327073047-b624240832ae/go.mod h1:D2IckZVXARugvikE4fv1glvaJMohKSZRzrPsxCjo9O0=
 github.com/elastic/elastic-agent-autodiscover v0.10.0 h1:WJ4zl9uSfk1kHmn2B/0byQBLIL607Zt4LNfOgV7+XN0=
diff --git a/internal/edot/go.mod b/internal/edot/go.mod
@@ -5,7 +5,7 @@ go 1.24.10
 replace github.com/elastic/elastic-agent => ../../
 
 require (
-	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d
+	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2
 	github.com/elastic/elastic-agent v0.0.0-00010101000000-000000000000
 	github.com/elastic/elastic-agent-libs v0.26.1
 	github.com/elastic/opentelemetry-collector-components/connector/elasticapmconnector v0.8.0
diff --git a/internal/edot/go.sum b/internal/edot/go.sum
@@ -440,8 +440,8 @@ github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumpti
 github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumption v1.1.0-elastic/go.mod h1:0vCBR1wgGwZeGmloJ+eCWIZF2S47grTXRzj2mftg2Nk=
 github.com/elastic/bayeux v1.0.5 h1:UceFq01ipmT3S8DzFK+uVAkbCdiPR0Bqei8qIGmUeY0=
 github.com/elastic/bayeux v1.0.5/go.mod h1:CSI4iP7qeo5MMlkznGvYKftp8M7qqP/3nzmVZoXHY68=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d h1:WiLFsGBR7P1ErPCUngBbOWU3poKF7WAdycwWcp5HeIA=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251117222225-3d17cad1ab6d/go.mod h1:c8O8MuFlYOkEbT1qdC/iaiTHD2u2mCrCJZtkl4+shTg=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2 h1:0VrGnx4wwU2K16wDOVAKxSI+TkJQQvYmQ+SJQQw/0rM=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251118234110-a4e336c8b5b2/go.mod h1:c8O8MuFlYOkEbT1qdC/iaiTHD2u2mCrCJZtkl4+shTg=
 github.com/elastic/elastic-agent-autodiscover v0.10.0 h1:WJ4zl9uSfk1kHmn2B/0byQBLIL607Zt4LNfOgV7+XN0=
 github.com/elastic/elastic-agent-autodiscover v0.10.0/go.mod h1:Nf3zh9FcJ9nTTswTwDTUAqXmvQllOrNliM6xmORSxwE=
 github.com/elastic/elastic-agent-client/v7 v7.17.2 h1:Cl2TeABqWZgW40t5fchGWT/sRk4MDDLWA0d8iHHOxLA=
diff --git a/internal/pkg/agent/application/upgrade/watcher.go b/internal/pkg/agent/application/upgrade/watcher.go
@@ -86,6 +86,7 @@ func (ch *AgentWatcher) Run(ctx context.Context) {
 
 	ch.connectCounter = 0
 	ch.lostCounter = 0
+	ch.lastPid = -1
 
 	// tracking of an error runs in a separate goroutine, because
 	// the call to `watch.Recv` blocks and a timer is needed
@@ -143,7 +144,6 @@ func (ch *AgentWatcher) Run(ctx context.Context) {
 
 LOOP:
 	for {
-		ch.lastPid = -1
 		connectTimer := time.NewTimer(ch.checkInterval)
 		select {
 		case <-ctx.Done():
diff --git a/pkg/testing/fixture.go b/pkg/testing/fixture.go
@@ -55,6 +55,7 @@ type Fixture struct {
 	srcPackage string
 	workDir    string
 	extractDir string
+	socketPath string
 
 	installed   bool
 	installOpts *InstallOpts
@@ -186,12 +187,21 @@ func NewFixture(t *testing.T, version string, opts ...FixtureOpt) (*Fixture, err
 }
 
 // Client returns the Elastic Agent communication client.
+// This client is shared across multiple calls to Client()
 func (f *Fixture) Client() client.Client {
 	f.cMx.RLock()
 	defer f.cMx.RUnlock()
 	return f.c
 }
 
+// NewClient returns a new Elastic Agent communication client.
+// The client is NOT shared across multiple calls but a new instance is allocated every time
+func (f *Fixture) NewClient() client.Client {
+	f.cMx.Lock()
+	defer f.cMx.Unlock()
+	return client.New(client.WithAddress(f.socketPath))
+}
+
 // Version returns the Elastic Agent version.
 func (f *Fixture) Version() string {
 	return f.version
@@ -1136,6 +1146,12 @@ func (f *Fixture) prepareComponents(workDir string, components ...UsableComponen
 	return nil
 }
 
+func (f *Fixture) setSocketPath(socketPath string) {
+	f.cMx.Lock()
+	defer f.cMx.Unlock()
+	f.socketPath = socketPath
+}
+
 func (f *Fixture) setClient(c client.Client) {
 	f.cMx.Lock()
 	defer f.cMx.Unlock()
@@ -1605,6 +1621,7 @@ type AgentBinaryVersion struct {
 	Commit    string `yaml:"commit"`
 	BuildTime string `yaml:"build_time"`
 	Snapshot  bool   `yaml:"snapshot"`
+	Fips      bool   `yaml:"fips"`
 }
 
 // String returns the version string.
diff --git a/pkg/testing/fixture_install.go b/pkg/testing/fixture_install.go
@@ -278,6 +278,7 @@ func (f *Fixture) installNoPkgManager(ctx context.Context, installOpts *InstallO
 		socketPath = paths.ControlSocketFromPath(runtime.GOOS, f.workDir)
 	}
 	c := client.New(client.WithAddress(socketPath))
+	f.setSocketPath(socketPath)
 	f.setClient(c)
 
 	f.t.Cleanup(func() {
diff --git a/testing/integration/ess/monitoring_probe_preserve_text_cfg_test.go b/testing/integration/ess/monitoring_probe_preserve_text_cfg_test.go
@@ -33,6 +33,8 @@ outputs:
     api_key: "example-key"
     preset: balanced
     allow_older_versions: true
+    status_reporting:
+      enabled: false
 
 inputs:
   - type: system/metrics
diff --git a/testing/integration/ess/upgrade_rollback_test.go b/testing/integration/ess/upgrade_rollback_test.go
@@ -97,6 +97,8 @@ outputs:
   default:
     type: elasticsearch
     hosts: [127.0.0.1:9200]
+    status_reporting:
+      enabled: false
 
 inputs:
   - condition: '${agent.version.version} == "%s"'
@@ -529,9 +531,13 @@ func managedRollbackRestartTest(ctx context.Context, t *testing.T, info *define.
 	// we expect ErrSkipGrace at this point, meaning that we finished installing but didn't wait for agent to become healthy
 	require.ErrorIs(t, err, ErrSkipGrace, "managed upgrade failed with unexpected error")
 
-	// A few seconds after the upgrade, deliberately restart upgraded Agent a
-	// couple of times to simulate Agent crashing.
-	restartAgentNTimes(t, 3, 10*time.Second)
+	installedAgentClient := from.NewClient()
+	targetVersion, err := to.ExecVersion(ctx)
+	require.NoError(t, err, "failed to get target version")
+	restartContext, cancel := context.WithTimeout(t.Context(), 1*time.Minute)
+	defer cancel()
+	// restart the agent only if it matches the (upgraded) target version
+	restartAgentVersion(restartContext, t, installedAgentClient, targetVersion.Binary, 10*time.Second)
 
 	// wait for the agent to be healthy and correct version
 	err = upgradetest.WaitHealthyAndVersion(ctx, from, startVersionInfo.Binary, 2*time.Minute, 10*time.Second, t)
@@ -657,42 +663,95 @@ func restartAgentNTimes(t *testing.T, noOfRestarts int, sleepBetweenIterations t
 
 	for restartIdx := 0; restartIdx < noOfRestarts; restartIdx++ {
 		time.Sleep(sleepBetweenIterations)
+		restartAgent(t, topPath, 5*time.Minute)
+	}
+}
 
-		t.Logf("Stopping agent via service to simulate crashing")
-		err := install.StopService(topPath, install.DefaultStopTimeout, install.DefaultStopInterval)
-		if err != nil && runtime.GOOS == define.Windows && strings.Contains(err.Error(), "The service has not been started.") {
-			// Due to the quick restarts every 10 seconds its possible that this is faster than Windows
-			// can handle. Decrementing restartIdx means that the loop will occur again.
-			t.Logf("Got an allowed error on Windows: %s", err)
-			err = nil
+func restartAgent(t *testing.T, topPath string, operationTimeout time.Duration) {
+	t.Logf("Stopping agent via service to simulate crashing")
+	stopRequested := time.Now()
+	err := install.StopService(topPath, install.DefaultStopTimeout, install.DefaultStopInterval)
+	if err != nil && runtime.GOOS == define.Windows && strings.Contains(err.Error(), "The service has not been started.") {
+		// Due to the quick restarts every sleepBetweenIterations its possible that this is faster than Windows
+		// can handle. Decrementing restartIdx means that the loop will occur again.
+		t.Logf("Got an allowed error on Windows: %s", err)
+		err = nil
+	}
+	require.NoError(t, err)
+
+	// ensure that it's stopped before starting it again
+	var status service.Status
+	var statusErr error
+	require.Eventuallyf(t, func() bool {
+		status, statusErr = install.StatusService(topPath)
+		if statusErr != nil {
+			return false
 		}
-		require.NoError(t, err)
+		return status != service.StatusRunning
+	}, operationTimeout, 500*time.Millisecond, "service never fully stopped (status: %v): %s", status, statusErr)
+	t.Logf("Stopped agent via service. Took roughly %s", time.Since(stopRequested))
+
+	// start it again
+	t.Logf("Starting agent via service to simulate crashing")
+	startRequested := time.Now()
+	err = install.StartService(topPath)
+	require.NoError(t, err)
 
-		// ensure that it's stopped before starting it again
-		var status service.Status
-		var statusErr error
-		require.Eventuallyf(t, func() bool {
-			status, statusErr = install.StatusService(topPath)
-			if statusErr != nil {
-				return false
-			}
-			return status != service.StatusRunning
-		}, 5*time.Minute, 1*time.Second, "service never fully stopped (status: %v): %s", status, statusErr)
-		t.Logf("Stopped agent via service to simulate crashing")
+	// ensure that it's started before next loop
+	require.Eventuallyf(t, func() bool {
+		status, statusErr = install.StatusService(topPath)
+		if statusErr != nil {
+			return false
+		}
+		return status == service.StatusRunning
+	}, operationTimeout, 500*time.Millisecond, "service never fully started (status: %v): %s", status, statusErr)
+	t.Logf("Started agent after stopping to simulate crashing. Took roughly %s", time.Since(startRequested))
+}
 
-		// start it again
-		t.Logf("Starting agent via service to simulate crashing")
-		err = install.StartService(topPath)
-		require.NoError(t, err)
+func restartAgentVersion(ctx context.Context, t *testing.T, client client.Client, targetVersion atesting.AgentBinaryVersion, restartInterval time.Duration) {
+	topPath := paths.Top()
+
+	ticker := time.NewTicker(restartInterval)
+	defer ticker.Stop()
 
-		// ensure that it's started before next loop
-		require.Eventuallyf(t, func() bool {
-			status, statusErr = install.StatusService(topPath)
-			if statusErr != nil {
-				return false
+	for {
+		select {
+		case <-ctx.Done():
+			t.Log("restart context is done, returning")
+			return
+
+		case <-ticker.C:
+			if !versionMatch(ctx, t, client, targetVersion) {
+				// version of running agent does not match the target, continue to the next iteration
+				continue
 			}
-			return status == service.StatusRunning
-		}, 5*time.Minute, 1*time.Second, "service never fully started (status: %v): %s", status, statusErr)
-		t.Logf("Started agent via service to simulate crashing")
+
+			restartAgent(t, topPath, restartInterval)
+		}
+
+	}
+}
+
+func versionMatch(ctx context.Context, t *testing.T, c client.Client, targetVersion atesting.AgentBinaryVersion) bool {
+	err := c.Connect(ctx)
+	if err != nil {
+		t.Logf("failed to connect to agent: %v", err)
+		return false
+	}
+	defer c.Disconnect()
+
+	actualVersion, err := c.Version(ctx)
+	if err != nil {
+		t.Logf("failed to detect agent version: %v", err)
+		return false
+	}
+
+	if actualVersion.Version != targetVersion.Version ||
+		actualVersion.Snapshot != targetVersion.Snapshot ||
+		actualVersion.Commit != targetVersion.Commit ||
+		actualVersion.Fips != targetVersion.Fips {
+		t.Logf("actual agent version %+v does not match target agent version %+v, skipping restart", actualVersion, targetVersion)
+		return false
 	}
+	return true
 }

Original file line number	Diff line number	Diff line change
`@@ -278,6 +278,7 @@ func (f Fixture) installNoPkgManager(ctx context.Context, installOpts InstallO`
`278`	`278`	`socketPath = paths.ControlSocketFromPath(runtime.GOOS, f.workDir)`
`279`	`279`	`}`
`280`	`280`	`c := client.New(client.WithAddress(socketPath))`
	`281`	`+ f.setSocketPath(socketPath)`
`281`	`282`	`f.setClient(c)`
`282`	`283`
`283`	`284`	`f.t.Cleanup(func() {`