diff --git a/filebeat/beater/filebeat.go b/filebeat/beater/filebeat.go index c3187fe60745..e09e24e66dfb 100644 --- a/filebeat/beater/filebeat.go +++ b/filebeat/beater/filebeat.go @@ -287,6 +287,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error { // 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 { + fb.runReady.Close() return err } diff --git a/filebeat/beater/stop_test.go b/filebeat/beater/stop_test.go new file mode 100644 index 000000000000..72b3eb61bd82 --- /dev/null +++ b/filebeat/beater/stop_test.go @@ -0,0 +1,110 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package beater + +import ( + "errors" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/elastic/beats/v7/filebeat/fileset" + "github.com/elastic/beats/v7/libbeat/beat" + "github.com/elastic/beats/v7/libbeat/beatmonitoring" + "github.com/elastic/beats/v7/libbeat/management" + "github.com/elastic/elastic-agent-libs/logp" +) + +// TestRunClosesRunReadyOnPreInitFailure guards against the regression in +// commit 00068f7 where the defer fb.runReady.Close() was placed after the +// PreInit call, leaving runReady unclosed when PreInit fails. An unclosed +// runReady causes StopWithContext to wait the full five-second timeout. +func TestRunClosesRunReadyOnPreInitFailure(t *testing.T) { + preInitErr := errors.New("preinit failed") + fb := &Filebeat{ + done: make(chan struct{}), + runReady: &closeOnce{ch: make(chan struct{})}, + logger: logp.NewNopLogger(), + moduleRegistry: new(fileset.ModuleRegistry), + } + b := &beat.Beat{ + Info: beat.Info{Logger: logp.NewNopLogger()}, + Manager: &preinitFailManager{err: preInitErr}, + Monitoring: beatmonitoring.NewMonitoring(), + } + + err := fb.Run(b) + require.ErrorIs(t, err, preInitErr) + + select { + case <-fb.runReady.ch: + // runReady was closed — StopWithContext will not block + default: + t.Fatal("runReady was not closed after PreInit failure; StopWithContext would wait five seconds") + } +} + +// preinitFailManager is a management.Manager stub whose PreInit returns an +// error. Only the methods called by Run before PreInit are implemented; any +// other call panics, catching unexpected code paths during the test. +type preinitFailManager struct { + management.Manager // zero-valued; panics on any unimplemented method + err error +} + +func (m *preinitFailManager) Enabled() bool { return true } +func (m *preinitFailManager) PreInit() error { return m.err } +func (m *preinitFailManager) RegisterDiagnosticHook(_, _, _, _ string, _ management.DiagnosticHook) { +} + +// TestStopWaitsForRunReady proves that Stop does not close the done channel +// until Run has closed runReady (i.e., reached waitFinished.Wait). +// This guards against a race where the OTel collector calls Shutdown before +// the beat's Run goroutine has initialised its shutdown-signal machinery. +func TestStopWaitsForRunReady(t *testing.T) { + fb := &Filebeat{ + done: make(chan struct{}), + runReady: &closeOnce{ch: make(chan struct{})}, + logger: logp.NewNopLogger(), + } + + stopDone := make(chan struct{}) + go func() { + defer close(stopDone) + fb.Stop() + }() + + // done must still be open: Stop is waiting for runReady to be closed. + select { + case <-fb.done: + t.Fatal("Stop closed done before Run closed runReady") + default: + } + + // Simulate Run() reaching the waitFinished.Wait() call. + fb.runReady.Close() + + select { + case <-fb.done: + case <-time.After(500 * time.Millisecond): + t.Fatal("Stop did not close done after runReady was closed") + } + + <-stopDone +}