[9.2](backport #49796) Split BeatV2Manager Start into two methods, so Beats can reply to check-in in parallel to its initialisation (#49849)

The Start method from BeatV2Manager is split into two methods:
 - PreInit: responsible for starting the Elastic Agent client and
   start replying to check-ins.
 - PostInit: responsible for setting the Beats status to 'Running' and
   start executing Unit changes.

A new method, WaitForStop is also added. It stops the BeatV2Manager
and waits until all goroutines have returned. Currently it is only
used in tests that use `testing.T` as the logger output to ensure no
panics happen because the logger was used after the test ended.

Multiple lint warnings are fixed

GenAI-Assisted: Yes
Human-Reviewed: Yes
Tool: Cursor-CLI, Model: GPT-5.3 Codex Extra High Fast
(cherry picked from commit 034546f)

# Conflicts:
#	x-pack/libbeat/management/managerV2.go
#	x-pack/osquerybeat/beater/osquerybeat_status_test.go

* Fix race that can block managerV2 shutdown (#49414)

As described in #49388, `BeatV2Manager` can miss the shutdown signal because its `Stop` method notifies the manager by sending to its signal channel `stopChan` rather than closing it, but there are two goroutines that both listen on that channel.

This PR changes `Stop` to close the channel rather than just sending. It also removes the second `stopChan` listener in `watchErrChan`, since the main goroutine already calls the context canceler for that helper when `stopChan` unblocks (this isn't strictly necessary but it will keep error states visible for a little longer during shutdown, and is what was previously happening in the "good" path where the main worker received the stop signal first).

(cherry picked from commit d39cb49)

---------

Co-authored-by: Tiago Queiroz <tiago.queiroz@elastic.co>
Co-authored-by: Fae Charlton <fae.charlton@elastic.co>

Author: 3 people

changelog/fragments/1773263871-beats-manager-shutdown.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Fix an issue that could delay reporting shutdown of Agent components
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

changelog/fragments/1774984035-move-manager-start.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Fix Filebeat crash loop when running under Elastic Agent and taking too long to initialise
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

filebeat/beater/filebeat.go

@@ -285,6 +285,21 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	}
 	finishedLogger := newFinishedLogger(wgEvents)
 
+	// Start the check-in loop, so Filebeat can respond to Elastic Agent,
+	// but it won't start any inputs/output
+	if err := b.Manager.PreInit(); err != nil {
+		return err
+	}
+
+	// Ensure that we only call b.Manager.Stop out of order
+	// if Run has failed early/before b.Manager.PostInit() was called.
+	managerEarlyStop := b.Manager.Stop
+	defer func() {
+		if managerEarlyStop != nil {
+			managerEarlyStop()
+		}
+	}()
+
 	registryMigrator := registrar.NewMigrator(config.Registry, fb.logger, b.Paths)
 	if err := registryMigrator.Run(); err != nil {
 		fb.logger.Errorf("Failed to migrate registry file: %+v", err)
@@ -484,10 +499,8 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	}
 	adiscover.Start()
 
-	// We start the manager when all the subsystem are initialized and ready to received events.
-	if err := b.Manager.Start(); err != nil {
-		return err
-	}
+	b.Manager.PostInit()
+	managerEarlyStop = nil
 
 	// Add done channel to wait for shutdown signal
 	waitFinished.AddChan(fb.done)

libbeat/beat/beat_test.go

@@ -21,6 +21,7 @@ import (
 	"regexp"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -38,6 +39,8 @@ type testManager struct {
 func (tm testManager) UpdateStatus(_ status.Status, _ string) {}
 func (tm testManager) Enabled() bool                          { return tm.isEnabled }
 func (tm testManager) Start() error                           { return nil }
+func (tm testManager) PreInit() error                         { return nil }
+func (tm testManager) PostInit()                              {}
 func (tm testManager) Stop()                                  {}
 func (tm testManager) AgentInfo() management.AgentInfo {
 	return management.AgentInfo{Unprivileged: tm.isUnpriv, ManagedMode: tm.mgmtMode}
@@ -49,6 +52,10 @@ func (tm testManager) UnregisterAction(_ management.Action) {}
 func (tm testManager) SetPayload(_ map[string]interface{})  {}
 func (tm testManager) RegisterDiagnosticHook(_ string, _ string, _ string, _ string, _ management.DiagnosticHook) {
 }
+func (tm testManager) WaitForStop(_ time.Duration) bool {
+	tm.Stop()
+	return true
+}
 
 func TestUserAgentString(t *testing.T) {
 	tests := []struct {

libbeat/cmd/instance/beat_test.go

@@ -26,6 +26,7 @@ import (
 	"path/filepath"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
@@ -507,10 +508,16 @@ func (m mockManager) RegisterDiagnosticHook(name, description, filename, content
 func (m mockManager) SetPayload(payload map[string]any)             {}
 func (m mockManager) SetStopCallback(f func())                      {}
 func (m mockManager) Start() error                                  { return nil }
+func (m mockManager) PreInit() error                                { return nil }
+func (m mockManager) PostInit()                                     {}
 func (m mockManager) Status() status.Status                         { return status.Status(-42) }
 func (m mockManager) Stop()                                         {}
 func (m mockManager) UnregisterAction(action management.Action)     {}
 func (m mockManager) UpdateStatus(status status.Status, msg string) {}
+func (m mockManager) WaitForStop(_ time.Duration) bool {
+	m.Stop()
+	return true
+}
 
 func TestManager(t *testing.T) {
 	// set the mockManger factory.

libbeat/management/management.go

@@ -19,6 +19,7 @@ package management
 
 import (
 	"sync"
+	"time"
 
 	"github.com/elastic/beats/v7/libbeat/common/reload"
 	"github.com/elastic/beats/v7/libbeat/management/status"
@@ -37,12 +38,25 @@ type Manager interface {
 	// Enabled returns true if manager is enabled.
 	Enabled() bool
 
-	// Start needs to invoked when the system is ready to receive an external configuration and
+	// Starts the unitListen loop, so the manager can already
+	// check-in with Elastic Agent, but no input/output will be
+	// started yet. Call [PostInit] to enable starting/stopping
+	// inputs/output.
+	PreInit() error
+
+	// PostInit needs to be invoked when the system is ready to receive an external configuration and
 	// also ready to start ingesting new events. The manager expects that all the reloadable and
 	// reloadable list are fixed for the whole lifetime of the manager.
 	//
 	// Notes: Adding dynamically new reloadable hooks at runtime can lead to inconsistency in the
 	// execution.
+	PostInit()
+
+	// Start starts the manager.
+	//
+	// Deprecated: Use [PreInit] and [PostInit] instead
+	//
+	// For backwards compatibility, [Start] calls [PreInit] then [PostInit].
 	Start() error
 
 	// Stop when this method is called, the manager will stop receiving new actions, no more action
@@ -54,6 +68,11 @@ type Manager interface {
 	// Note: Stop will not call 'UnregisterAction()' automatically.
 	Stop()
 
+	// WaitForStop blocks until the manager has fully stopped, or timeout elapses.
+	// It returns true if the manager stopped before timeout, false otherwise.
+	// A non-positive timeout means wait indefinitely.
+	WaitForStop(timeout time.Duration) bool
+
 	// AgentInfo returns the information of the agent to which the manager is connected.
 	AgentInfo() AgentInfo
 
@@ -161,9 +180,15 @@ func (n *FallbackManager) Stop() {
 // the nilManager is still used for shutdown on some cases,
 // but that does not mean the Beat is being managed externally,
 // hence it will always return false.
-func (n *FallbackManager) Enabled() bool                      { return false }
-func (n *FallbackManager) AgentInfo() AgentInfo               { return AgentInfo{} }
-func (n *FallbackManager) Start() error                       { return nil }
+func (n *FallbackManager) Enabled() bool        { return false }
+func (n *FallbackManager) AgentInfo() AgentInfo { return AgentInfo{} }
+func (n *FallbackManager) PreInit() error       { return nil }
+func (n *FallbackManager) PostInit()            {}
+func (n *FallbackManager) Start() error         { return nil }
+func (n *FallbackManager) WaitForStop(_ time.Duration) bool {
+	n.Stop()
+	return true
+}
 func (n *FallbackManager) CheckRawConfig(cfg *config.C) error { return nil }
 func (n *FallbackManager) RegisterAction(action Action)       {}
 func (n *FallbackManager) UnregisterAction(action Action)     {}

x-pack/filebeat/tests/integration/managerV2_test.go

@@ -283,7 +283,7 @@ func TestFailedOutputReportsUnhealthy(t *testing.T) {
 			Id:             "input-unit",
 			Type:           proto.UnitType_INPUT,
 			ConfigStateIdx: 1,
-			State:          proto.State_STARTING,
+			State:          proto.State_HEALTHY,
 			LogLevel:       proto.UnitLogLevel_DEBUG,
 			Config: &proto.UnitExpectedConfig{
 				Id:   "log-input",
@@ -331,7 +331,7 @@ func TestFailedOutputReportsUnhealthy(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		return finalStateReached.Load()
-	}, 30*time.Second, 100*time.Millisecond, "Output unit did not report unhealthy")
+	}, 30*time.Second, 100*time.Millisecond, "Output unit did not report unhealthy or input did not report health")
 
 	t.Cleanup(server.Stop)
 }

x-pack/filebeat/tests/integration/status_reporter_test.go

@@ -103,7 +103,7 @@ func TestLogStatusReporter(t *testing.T) {
 		},
 		{
 			proto.State_DEGRADED,
-			&inputStream,
+			&inputStreamIrregular,
 		},
 		{
 			proto.State_HEALTHY,
@@ -159,7 +159,7 @@ func getInputStream(id string, path string, stateIdx int) proto.UnitExpected {
 	return proto.UnitExpected{
 		Id:             id,
 		Type:           proto.UnitType_INPUT,
-		ConfigStateIdx: uint64(stateIdx),
+		ConfigStateIdx: uint64(stateIdx), //nolint:gosec // stateIdx is always positive
 		State:          proto.State_HEALTHY,
 		Config: &proto.UnitExpectedConfig{
 			Streams: []*proto.Stream{{