[beatreceiver] fix status reporting (#47936) (#48098)

* [beatreceiver] fix status reporting

Author: mergify[bot]

x-pack/libbeat/cmd/instance/receiver.go

@@ -16,8 +16,9 @@ import (
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
 	"github.com/elastic/beats/v7/libbeat/cmd/instance"
 	"github.com/elastic/beats/v7/libbeat/common/backoff"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/beats/v7/x-pack/libbeat/common/otelbeat/otelmanager"
-	"github.com/elastic/beats/v7/x-pack/libbeat/common/otelbeat/status"
+	otelstatus "github.com/elastic/beats/v7/x-pack/libbeat/common/otelbeat/status"
 	_ "github.com/elastic/beats/v7/x-pack/libbeat/include"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
@@ -96,9 +97,10 @@ func NewBeatReceiver(ctx context.Context, b *instance.Beat, creator beat.Creator
 
 // BeatReceiver.Start() starts the beat receiver.
 func (br *BeatReceiver) Start(host component.Host) error {
+	var groupReporter otelstatus.RunnerReporter
 	if w, ok := br.beater.(cfgfile.WithOtelFactoryWrapper); ok {
-		groupReporter := status.NewGroupStatusReporter(host)
-		w.WithOtelFactoryWrapper(status.StatusReporterFactory(groupReporter))
+		groupReporter = otelstatus.NewGroupStatusReporter(host)
+		w.WithOtelFactoryWrapper(otelstatus.StatusReporterFactory(groupReporter))
 	}
 
 	// We go through all extensions to find any that implement the DiagnosticExtension interface.
@@ -126,6 +128,8 @@ func (br *BeatReceiver) Start(host component.Host) error {
 	}
 
 	if err := br.beater.Run(&br.beat.Beat); err != nil {
+		// set beatreceiver status
+		groupReporter.UpdateStatus(status.Failed, err.Error())
 		return fmt.Errorf("beat receiver run error: %w", err)
 	}
 

x-pack/libbeat/common/otelbeat/status/reporter.go

@@ -36,6 +36,9 @@ func toPdata(r *runnerState) pcommon.Map {
 // This is used for grouping and managing statuses of multiple runners
 type RunnerReporter interface {
 	GetReporterForRunner(id string) status.StatusReporter
+
+	// UpdateStatus updates the group status of a runnerReporter
+	UpdateStatus(status status.Status, msg string)
 }
 
 type reporter struct {
@@ -81,22 +84,40 @@ func (r *reporter) updateStatusForRunner(id string, state status.Status, msg str
 		}
 	}
 
-	// report status to parent reporter
-	r.UpdateStatus()
+	// report aggregated status for all sub-components
+	evt := r.calculateOtelStatus()
+	r.emitDummyStatus(evt)
+	componentstatus.ReportStatus(r.host, evt)
 }
 
-func (r *reporter) UpdateStatus() {
-	evt := r.calculateOtelStatus()
+// UpdateStatus reports the overall status of the group.
+// This is useful to report any failures encountered before a runner is initialized.
+// Note: This will override all sub-reporter statuses if any
+func (r *reporter) UpdateStatus(status status.Status, msg string) {
+	otelStatus := beatStatusToOtelStatus(status)
+	if otelStatus == componentstatus.StatusNone {
+		return
+	}
+	var eventBuilderOpts []componentstatus.EventBuilderOption
+	if componentstatus.StatusIsError(otelStatus) {
+		eventBuilderOpts = append(eventBuilderOpts, componentstatus.WithError(errors.New(msg)))
+	}
+	evt := componentstatus.NewEvent(otelStatus, eventBuilderOpts...)
+	r.emitDummyStatus(evt)
+	componentstatus.ReportStatus(r.host, evt)
+}
+
+func (r *reporter) emitDummyStatus(evt *componentstatus.Event) {
 	oppositeStatus := getOppositeStatus(evt.Status())
 	if oppositeStatus != componentstatus.StatusNone {
 		// emit a dummy event first to ensure the otel core framework acknowledges the change
 		// workaround for https://github.com/open-telemetry/opentelemetry-collector/issues/14282
 		dummyEvt := componentstatus.NewEvent(oppositeStatus)
 		componentstatus.ReportStatus(r.host, dummyEvt)
 	}
-	componentstatus.ReportStatus(r.host, evt)
 }
 
+// calculateOtelStatus aggregates the statuses of all runners
 func (r *reporter) calculateOtelStatus() *componentstatus.Event {
 	var evt *componentstatus.Event
 	s, msg := r.calculateAggregateState()

x-pack/libbeat/common/otelbeat/status/reporter_test.go

@@ -40,6 +40,13 @@ func TestGroupStatus(t *testing.T) {
 	require.Equalf(t, m.Evt.Status(), componentstatus.StatusPermanentError, "expected StatusPermanentError, got %v", m.Evt.Status())
 	require.NotNil(t, m.Evt.Err(), "expected non-nil error, got nil")
 	require.Equalf(t, m.Evt.Err().Error(), "Failed Runner2", "expected 'Failed Runner1', got %v", m.Evt.Err())
+
+	// group reporter is updated directly
+	reporter.UpdateStatus(status.Failed, "beatreceiver failed to start")
+
+	require.Equalf(t, m.Evt.Status(), componentstatus.StatusPermanentError, "expected StatusPermanentError, got %v", m.Evt.Status())
+	require.NotNil(t, m.Evt.Err(), "expected non-nil error, got nil")
+	require.Equalf(t, m.Evt.Err().Error(), "beatreceiver failed to start", "expected 'beatreceiver failed to start', got %v", m.Evt.Err())
 }
 
 func TestToPdata(t *testing.T) {