Translate beat receiver status to component state (#8273)

* Add otel status translation

* Modify monitoring comparison integration test

# Conflicts:
#	testing/integration/beat_receivers_test.go

* Add coordinator test

* Add more status unit tests

* Make messages more compatible with beats processes

* Check agent status in monitoring integration test

* Improve translation error handling

Author: swiatekm

internal/pkg/agent/application/coordinator/coordinator.go

@@ -13,7 +13,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/elastic/elastic-agent/internal/pkg/otel/configtranslate"
+	"github.com/elastic/elastic-agent/internal/pkg/otel/translate"
 
 	"go.opentelemetry.io/collector/component/componentstatus"
 
@@ -349,7 +349,7 @@ type managerChans struct {
 	varsManagerUpdate <-chan []*transpiler.Vars
 	varsManagerError  <-chan error
 
-	otelManagerUpdate <-chan *status.AggregateStatus
+	otelManagerUpdate chan *status.AggregateStatus
 	otelManagerError  <-chan error
 
 	upgradeMarkerUpdate <-chan upgrade.UpdateMarker
@@ -463,7 +463,9 @@ func New(logger *logger.Logger, cfg *configuration.Configuration, logLevel logp.
 		c.managerChans.varsManagerError = varsMgr.Errors()
 	}
 	if otelMgr != nil {
-		c.managerChans.otelManagerUpdate = otelMgr.Watch()
+		// The otel manager sends updates to the watchRuntimeComponents function, which extracts component status
+		// and forwards the rest to this channel.
+		c.managerChans.otelManagerUpdate = make(chan *status.AggregateStatus)
 		c.managerChans.otelManagerError = otelMgr.Errors()
 	}
 	if upgradeMgr != nil && upgradeMgr.MarkerWatcher() != nil {
@@ -656,77 +658,148 @@ func (c *Coordinator) watchRuntimeComponents(ctx context.Context) {
 	state := make(map[string]runtime.ComponentState)
 
 	var subChan <-chan runtime.ComponentComponentState
+	var otelChan <-chan *status.AggregateStatus
 	// A real Coordinator will always have a runtime manager, but unit tests
 	// may not initialize all managers -- in that case we leave subChan nil,
 	// and just idle until Coordinator shuts down.
 	if c.runtimeMgr != nil {
 		subChan = c.runtimeMgr.SubscribeAll(ctx).Ch()
 	}
+	if c.otelMgr != nil {
+		otelChan = c.otelMgr.Watch()
+	}
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case s := <-subChan:
-			oldState, ok := state[s.Component.ID]
-			if !ok {
-				componentLog := coordinatorComponentLog{
-					ID:    s.Component.ID,
-					State: s.State.State.String(),
-				}
-				logBasedOnState(c.logger, s.State.State, fmt.Sprintf("Spawned new component %s: %s", s.Component.ID, s.State.Message), "component", componentLog)
-				for ui, us := range s.State.Units {
-					unitLog := coordinatorUnitLog{
-						ID:    ui.UnitID,
-						Type:  ui.UnitType.String(),
-						State: us.State.String(),
-					}
-					logBasedOnState(c.logger, us.State, fmt.Sprintf("Spawned new unit %s: %s", ui.UnitID, us.Message), "component", componentLog, "unit", unitLog)
-				}
-			} else {
-				componentLog := coordinatorComponentLog{
-					ID:    s.Component.ID,
-					State: s.State.State.String(),
-				}
-				if oldState.State != s.State.State {
-					cl := coordinatorComponentLog{
-						ID:       s.Component.ID,
-						State:    s.State.State.String(),
-						OldState: oldState.State.String(),
-					}
-					logBasedOnState(c.logger, s.State.State, fmt.Sprintf("Component state changed %s (%s->%s): %s", s.Component.ID, oldState.State.String(), s.State.State.String(), s.State.Message), "component", cl)
-				}
-				for ui, us := range s.State.Units {
-					oldUS, ok := oldState.Units[ui]
-					if !ok {
-						unitLog := coordinatorUnitLog{
-							ID:    ui.UnitID,
-							Type:  ui.UnitType.String(),
-							State: us.State.String(),
-						}
-						logBasedOnState(c.logger, us.State, fmt.Sprintf("Spawned new unit %s: %s", ui.UnitID, us.Message), "component", componentLog, "unit", unitLog)
-					} else if oldUS.State != us.State {
-						unitLog := coordinatorUnitLog{
-							ID:       ui.UnitID,
-							Type:     ui.UnitType.String(),
-							State:    us.State.String(),
-							OldState: oldUS.State.String(),
-						}
-						logBasedOnState(c.logger, us.State, fmt.Sprintf("Unit state changed %s (%s->%s): %s", ui.UnitID, oldUS.State.String(), us.State.String(), us.Message), "component", componentLog, "unit", unitLog)
-					}
-				}
-			}
-			state[s.Component.ID] = s.State
-			if s.State.State == client.UnitStateStopped {
-				delete(state, s.Component.ID)
-			}
+		case componentState := <-subChan:
+			logComponentStateChange(c.logger, state, &componentState)
 			// Forward the final changes back to Coordinator, unless our context
 			// has ended.
 			select {
-			case c.managerChans.runtimeManagerUpdate <- s:
+			case c.managerChans.runtimeManagerUpdate <- componentState:
 			case <-ctx.Done():
 				return
 			}
+		case otelStatus := <-otelChan:
+			// We don't break on errors here, because we want to forward the status
+			// even if there was an error, and the rest of the code gracefully handles componentStates being nil
+			componentStates, err := translate.GetAllComponentStates(otelStatus, c.componentModel)
+			if err != nil {
+				c.setOTelError(err)
+			}
+			err = translate.DropComponentStateFromOtelStatus(otelStatus)
+			if err != nil {
+				c.setOTelError(err)
+			}
+
+			// forward the remaining otel status
+			// TODO: Implement subscriptions for otel manager status to avoid the need for this
+			select {
+			case c.managerChans.otelManagerUpdate <- otelStatus:
+			case <-ctx.Done():
+				return
+			}
+
+			// drop component states which don't exist in the configuration anymore
+			// we need to do this because we aren't guaranteed to receive a STOPPED state when the component is removed
+			componentIds := make(map[string]bool)
+			for _, componentState := range componentStates {
+				componentIds[componentState.Component.ID] = true
+			}
+			for id := range state {
+				if _, ok := componentIds[id]; !ok {
+					// this component is not in the configuration anymore, emit a fake STOPPED state
+					componentStates = append(componentStates, runtime.ComponentComponentState{
+						Component: component.Component{
+							ID: id,
+						},
+						State: runtime.ComponentState{
+							State: client.UnitStateStopped,
+						},
+					})
+				}
+			}
+			// now handle the component states
+			for _, componentState := range componentStates {
+				logComponentStateChange(c.logger, state, &componentState)
+				// Forward the final changes back to Coordinator, unless our context
+				// has ended.
+				select {
+				case c.managerChans.runtimeManagerUpdate <- componentState:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}
+}
+
+// logComponentStateChange emits a log message based on the new component state.
+func logComponentStateChange(
+	logger *logger.Logger,
+	coordinatorState map[string]runtime.ComponentState,
+	componentState *runtime.ComponentComponentState) {
+	oldState, ok := coordinatorState[componentState.Component.ID]
+	if !ok {
+		componentLog := coordinatorComponentLog{
+			ID:    componentState.Component.ID,
+			State: componentState.State.State.String(),
 		}
+		logMessage := fmt.Sprintf("Spawned new component %s: %s",
+			componentState.Component.ID,
+			componentState.State.Message)
+		logBasedOnState(logger, componentState.State.State, logMessage, "component", componentLog)
+		for ui, us := range componentState.State.Units {
+			unitLog := coordinatorUnitLog{
+				ID:    ui.UnitID,
+				Type:  ui.UnitType.String(),
+				State: us.State.String(),
+			}
+			unitLogMessage := fmt.Sprintf("Spawned new unit %s: %s", ui.UnitID, us.Message)
+			logBasedOnState(logger, us.State, unitLogMessage, "component", componentLog, "unit", unitLog)
+		}
+	} else {
+		componentLog := coordinatorComponentLog{
+			ID:    componentState.Component.ID,
+			State: componentState.State.State.String(),
+		}
+		if oldState.State != componentState.State.State {
+			cl := coordinatorComponentLog{
+				ID:       componentState.Component.ID,
+				State:    componentState.State.State.String(),
+				OldState: oldState.State.String(),
+			}
+			logBasedOnState(logger, componentState.State.State,
+				fmt.Sprintf("Component state changed %s (%s->%s): %s",
+					componentState.Component.ID, oldState.State.String(),
+					componentState.State.State.String(),
+					componentState.State.Message),
+				"component", cl)
+		}
+		for ui, us := range componentState.State.Units {
+			oldUS, ok := oldState.Units[ui]
+			if !ok {
+				unitLog := coordinatorUnitLog{
+					ID:    ui.UnitID,
+					Type:  ui.UnitType.String(),
+					State: us.State.String(),
+				}
+				logBasedOnState(logger, us.State, fmt.Sprintf("Spawned new unit %s: %s", ui.UnitID, us.Message), "component", componentLog, "unit", unitLog)
+			} else if oldUS.State != us.State {
+				unitLog := coordinatorUnitLog{
+					ID:       ui.UnitID,
+					Type:     ui.UnitType.String(),
+					State:    us.State.String(),
+					OldState: oldUS.State.String(),
+				}
+				logBasedOnState(logger, us.State, fmt.Sprintf("Unit state changed %s (%s->%s): %s", ui.UnitID, oldUS.State.String(), us.State.String(), us.Message), "component", componentLog, "unit", unitLog)
+			}
+		}
+	}
+	coordinatorState[componentState.Component.ID] = componentState.State
+	if componentState.State.State == client.UnitStateStopped {
+		delete(coordinatorState, componentState.Component.ID)
 	}
 }
 
@@ -1231,8 +1304,8 @@ func (c *Coordinator) runLoopIteration(ctx context.Context) {
 			c.processVars(ctx, vars)
 		}
 
-	case collector := <-c.managerChans.otelManagerUpdate:
-		c.state.Collector = collector
+	case collectorStatus := <-c.managerChans.otelManagerUpdate:
+		c.state.Collector = collectorStatus
 		c.stateNeedsRefresh = true
 
 	case ll := <-c.logLevelCh:
@@ -1469,7 +1542,7 @@ func (c *Coordinator) updateOtelManagerConfig(model *component.Model) error {
 	if len(model.Components) > 0 {
 		var err error
 		c.logger.With("components", model.Components).Debug("Updating otel manager model")
-		componentOtelCfg, err = configtranslate.GetOtelConfig(model, c.agentInfo, c.monitorMgr.ComponentMonitoringConfig)
+		componentOtelCfg, err = translate.GetOtelConfig(model, c.agentInfo, c.monitorMgr.ComponentMonitoringConfig)
 		if err != nil {
 			c.logger.Errorf("failed to generate otel config: %v", err)
 		}

internal/pkg/agent/application/coordinator/coordinator_test.go

@@ -1342,6 +1342,7 @@ type fakeOTelManager struct {
 	updateCallback func(*confmap.Conf) error
 	result         error
 	errChan        chan error
+	statusChan     chan *status.AggregateStatus
 }
 
 func (f *fakeOTelManager) Run(ctx context.Context) error {
@@ -1365,7 +1366,7 @@ func (f *fakeOTelManager) Update(cfg *confmap.Conf) {
 }
 
 func (f *fakeOTelManager) Watch() <-chan *status.AggregateStatus {
-	return nil
+	return f.statusChan
 }
 
 // An implementation of the RuntimeManager interface for use in testing.