[libbeat] Stop publisher properly

[marc-gr]

Proposed commit message

We avoided closing the publisher on Stop to avoid races with the ES output since we were not aborting the active connections. This adds:

• ES output now aborts active requests on Close()
• Beater closes publisher on stop.
• Agent manager also closes telemetry and the publisher on the Close callback.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

Disruptive User Impact

• Now ES connections will be aborted on stop and some events might need to be re-sent. This is in any case a change that puts the ES output on par with how the rest of the outputs work, since was the only one leaving active conns unattended from what I observed.

Related issues

• Closes Windows service for Beat does not stop when output is unreachable #40518
• Closes Windows service for Beat does not stop gracefully #38666 (needs to test)

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[AndersonQ]

looks good, but I'll wait the tests to pass to approve it

[AndersonQ]

[Blocker] cont.

you could use it instead

Suggested change

	r, err := http.Get("http://localhost:5066") //nolint:noctx // fine for tests
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, r.StatusCode, "incorrect status code")
	buff := &bytes.Buffer{}
	require.Eventually(t, func() bool {
	buff.Reset()
	r, err := http.Get("http://localhost:5066")
	if err != nil {
	_, _ = fmt.Fprintf(buff, "stats endpoint not available: %v", err)
	return false
	}
	if r.StatusCode != http.StatusOK {
	_, _ = fmt.Fprintf(buff, "stats endpoint: bad HTTPnstatus: %s",
	r.Status)
	return false
	}
	return true
	}, time.Second, 100*time.Millisecond,
	"stats endpoint never become available: %s", buff)

if you want, you could even remove the WaitForLogs. Just don't mix WaitForLogs and Eventually because WaitForLogs uses eventually underneath.

[AndersonQ]

[Blocker]
I see why just WaitForLogs("Starting stats endpoint" isn't enough:

beats/libbeat/api/server.go

Lines 69 to 72 in 9249566

	s.log.Info("Starting stats endpoint")
	go func(l net.Listener) {
	s.log.Infof("Metrics endpoint listening on: %s (configured: %s)", l.Addr().String(), s.config.Host)
	err := http.Serve(l, s.mux)

but we don't use time.Sleep anymore, unless there is no other way. However here there is a better alternative. See below

[AndersonQ]

to avoid blocking the PR and it not making to 8.15.1, I'll approve and later open another PR to properly fix those tests. The issue is on libbeat, the mock beat does not stops properly, if it did, your PR would not have impacted those tests