Fix non-otel component status being incorrectly deleted (#8435) (#8582)

* Fix non-otel component status being incorrectly deleted

Otel component status processing needs to emit fake STOPPED events for
components which don't exist anymore. This was incorrectly taking place
for components running as beats processes, resulting in them incorrectly
reporting a STOPPED state.

* Fix incorrect timeout for coordinator test

* Add a more detailed comment about the test logic

(cherry picked from commit 70f2450)

Co-authored-by: Mikołaj Świątek <[email protected]>

Author: mergify[bot]

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

@@ -654,49 +654,47 @@ func (c *Coordinator) SetLogLevel(ctx context.Context, lvl *logp.Level) error {
 // watchRuntimeComponents listens for state updates from the runtime
 // manager, logs them, and forwards them to CoordinatorState.
 // Runs in its own goroutine created in Coordinator.Run.
-func (c *Coordinator) watchRuntimeComponents(ctx context.Context) {
-	state := make(map[string]runtime.ComponentState)
+func (c *Coordinator) watchRuntimeComponents(
+	ctx context.Context,
+	runtimeComponentStates <-chan runtime.ComponentComponentState,
+	otelStatuses <-chan *status.AggregateStatus,
+) {
+	// We need to track otel component state separately because otel components may not always get a STOPPED status
+	// If we receive an otel status without the state of a component we're tracking, we need to emit a fake STOPPED
+	// status for it. Process component states should not be affected by this logic.
+	processState := make(map[string]runtime.ComponentState)
+	otelState := 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 componentState := <-subChan:
-			logComponentStateChange(c.logger, state, &componentState)
+		case componentState := <-runtimeComponentStates:
+			logComponentStateChange(c.logger, processState, &componentState)
 			// Forward the final changes back to Coordinator, unless our context
 			// has ended.
 			select {
 			case c.managerChans.runtimeManagerUpdate <- componentState:
 			case <-ctx.Done():
 				return
 			}
-		case otelStatus := <-otelChan:
+		case otelStatus := <-otelStatuses:
 			// 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)
+			finalOtelStatus, err := translate.DropComponentStateFromOtelStatus(otelStatus)
 			if err != nil {
 				c.setOTelError(err)
+				finalOtelStatus = otelStatus
 			}
 
 			// 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 c.managerChans.otelManagerUpdate <- finalOtelStatus:
 			case <-ctx.Done():
 				return
 			}
@@ -707,7 +705,7 @@ func (c *Coordinator) watchRuntimeComponents(ctx context.Context) {
 			for _, componentState := range componentStates {
 				componentIds[componentState.Component.ID] = true
 			}
-			for id := range state {
+			for id := range otelState {
 				if _, ok := componentIds[id]; !ok {
 					// this component is not in the configuration anymore, emit a fake STOPPED state
 					componentStates = append(componentStates, runtime.ComponentComponentState{
@@ -722,7 +720,7 @@ func (c *Coordinator) watchRuntimeComponents(ctx context.Context) {
 			}
 			// now handle the component states
 			for _, componentState := range componentStates {
-				logComponentStateChange(c.logger, state, &componentState)
+				logComponentStateChange(c.logger, otelState, &componentState)
 				// Forward the final changes back to Coordinator, unless our context
 				// has ended.
 				select {
@@ -813,7 +811,19 @@ func (c *Coordinator) Run(ctx context.Context) error {
 	// log all changes in the state of the runtime and update the coordinator state
 	watchCtx, watchCanceller := context.WithCancel(ctx)
 	defer watchCanceller()
-	go c.watchRuntimeComponents(watchCtx)
+
+	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()
+	}
+	go c.watchRuntimeComponents(watchCtx, subChan, otelChan)
 
 	// Close the state broadcaster on finish, but leave it running in the
 	// background until all subscribers have read the final values or their

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

@@ -1341,42 +1341,43 @@ func TestCoordinatorTranslatesOtelStatusToComponentState(t *testing.T) {
 	logger := logp.NewLogger("testing")
 
 	statusChan := make(chan *status.AggregateStatus)
-	otelManager := &fakeOTelManager{
-		statusChan: statusChan,
-	}
-
-	componentModel := []component.Component{
-		{
-			ID:             "filestream-default",
-			InputType:      "filestream",
-			OutputType:     "elasticsearch",
-			RuntimeManager: component.OtelRuntimeManager,
-			InputSpec: &component.InputRuntimeSpec{
-				BinaryName: "agentbeat",
-				Spec: component.InputSpec{
-					Command: &component.CommandSpec{
-						Args: []string{"filebeat"},
-					},
+
+	runtimeStateChan := make(chan runtime.ComponentComponentState)
+
+	otelComponent := component.Component{
+		ID:             "filestream-default",
+		InputType:      "filestream",
+		OutputType:     "elasticsearch",
+		RuntimeManager: component.OtelRuntimeManager,
+		InputSpec: &component.InputRuntimeSpec{
+			BinaryName: "agentbeat",
+			Spec: component.InputSpec{
+				Command: &component.CommandSpec{
+					Args: []string{"filebeat"},
 				},
 			},
-			Units: []component.Unit{
-				{
-					ID:   "filestream-unit",
-					Type: client.UnitTypeInput,
-					Config: &proto.UnitExpectedConfig{
-						Streams: []*proto.Stream{
-							{Id: "test-1"},
-							{Id: "test-2"},
-						},
+		},
+		Units: []component.Unit{
+			{
+				ID:   "filestream-unit",
+				Type: client.UnitTypeInput,
+				Config: &proto.UnitExpectedConfig{
+					Streams: []*proto.Stream{
+						{Id: "test-1"},
+						{Id: "test-2"},
 					},
 				},
-				{
-					ID:   "filestream-default",
-					Type: client.UnitTypeOutput,
-				},
+			},
+			{
+				ID:   "filestream-default",
+				Type: client.UnitTypeOutput,
 			},
 		},
 	}
+	processComponent := otelComponent
+	processComponent.RuntimeManager = component.ProcessRuntimeManager
+	processComponent.ID = "filestream-process"
+
 	otelStatus := &status.AggregateStatus{
 		Event: componentstatus.NewEvent(componentstatus.StatusOK),
 		ComponentStatusMap: map[string]*status.AggregateStatus{
@@ -1411,17 +1412,17 @@ func TestCoordinatorTranslatesOtelStatusToComponentState(t *testing.T) {
 			otelManagerUpdate:    make(chan *status.AggregateStatus),
 			runtimeManagerUpdate: make(chan runtime.ComponentComponentState),
 		},
-		otelMgr:        otelManager,
-		state:          State{},
-		componentModel: componentModel,
+		state: State{},
 	}
 
 	// start runtime status watching
-	go coord.watchRuntimeComponents(ctx)
+	go coord.watchRuntimeComponents(ctx, runtimeStateChan, statusChan)
 
 	// no component status
 	assert.Empty(t, coord.state.Components)
 
+	coord.componentModel = []component.Component{otelComponent}
+
 	// push the status into the coordinator
 	select {
 	case statusChan <- otelStatus:
@@ -1446,8 +1447,57 @@ func TestCoordinatorTranslatesOtelStatusToComponentState(t *testing.T) {
 
 	assert.Len(t, coord.state.Components, 1)
 
+	// Add both a process component and an otel component, in that order. Both should appear in the state.
+	coord.componentModel = []component.Component{otelComponent, processComponent}
+
+	// push the process component state into the coordinator
+	select {
+	case runtimeStateChan <- runtime.ComponentComponentState{
+		Component: processComponent,
+		State: runtime.ComponentState{
+			State: client.UnitStateHealthy,
+		},
+	}:
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for coordinator to receive status")
+	}
+
+	select {
+	case componentState := <-coord.managerChans.runtimeManagerUpdate:
+		coord.applyComponentState(componentState)
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for coordinator to receive status")
+	}
+
+	// push the otel status into the coordinator
+	select {
+	case statusChan <- otelStatus:
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for coordinator to receive status")
+	}
+
+	select {
+	case finalOtelStatus := <-coord.managerChans.otelManagerUpdate:
+		// we shouldn't have any status remaining for the otel collector, as the status we've pushed earlier only
+		// contains beats receiver status for the "filestream-default" component
+		// this status is removed from the otel collector status, because it's reported as component state instead
+		assert.Empty(t, finalOtelStatus.ComponentStatusMap)
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for coordinator to receive status")
+	}
+
+	select {
+	case componentState := <-coord.managerChans.runtimeManagerUpdate:
+		coord.applyComponentState(componentState)
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for coordinator to receive status")
+	}
+
+	assert.Len(t, coord.state.Components, 2)
+
 	// Now, we remove the component and resend the same status. The component state should be deleted.
 	coord.componentModel = []component.Component{}
+	coord.state = State{}
 	select {
 	case statusChan <- otelStatus:
 	case <-ctx.Done():
@@ -1471,7 +1521,7 @@ func TestCoordinatorTranslatesOtelStatusToComponentState(t *testing.T) {
 
 	assert.Empty(t, coord.state.Components)
 
-	// Push an invalid status, there should be no component state, but there should be an otel status
+	// Push an invalid status, there should be no otel component state, but there should be an otel status
 	select {
 	case statusChan <- invalidOtelStatus:
 	case <-ctx.Done():

internal/pkg/otel/translate/status.go

@@ -56,26 +56,28 @@ func GetAllComponentStates(otelStatus *status.AggregateStatus, components []comp
 
 // DropComponentStateFromOtelStatus removes the statuses of otel pipelines representing runtime components from the
 // given status.
-func DropComponentStateFromOtelStatus(otelStatus *status.AggregateStatus) error {
+func DropComponentStateFromOtelStatus(otelStatus *status.AggregateStatus) (*status.AggregateStatus, error) {
 	if otelStatus == nil {
-		return nil
+		return nil, nil
 	}
-	for pipelineStatusId := range otelStatus.ComponentStatusMap {
+
+	newStatus := deepCopyStatus(otelStatus)
+	for pipelineStatusId := range newStatus.ComponentStatusMap {
 		pipelineId := &pipeline.ID{}
 		componentKind, pipelineIdStr := parseEntityStatusId(pipelineStatusId)
 		if componentKind != "pipeline" {
-			return fmt.Errorf("pipeline status id %s is not a pipeline", pipelineStatusId)
+			return nil, fmt.Errorf("pipeline status id %s is not a pipeline", pipelineStatusId)
 		}
 		err := pipelineId.UnmarshalText([]byte(pipelineIdStr)) // there's no ergonomic way to do this conversion
 		if err != nil {
-			return err
+			return nil, err
 		}
 		if strings.HasPrefix(pipelineId.Name(), OtelNamePrefix) {
-			delete(otelStatus.ComponentStatusMap, pipelineStatusId)
+			delete(newStatus.ComponentStatusMap, pipelineStatusId)
 		}
 	}
 
-	return nil
+	return newStatus, nil
 }
 
 // getOtelRuntimePipelineStatuses finds otel pipeline statuses belonging to runtime components and returns them as a map
@@ -273,3 +275,24 @@ func parseEntityStatusId(id string) (kind string, entityId string) {
 	}
 	return parts[0], parts[1]
 }
+
+// deepCopyStatus makes a deep copy of the status.
+func deepCopyStatus(otelStatus *status.AggregateStatus) *status.AggregateStatus {
+	if otelStatus == nil {
+		return nil
+	}
+
+	newStatus := &status.AggregateStatus{
+		Event: otelStatus.Event,
+	}
+	if otelStatus.ComponentStatusMap == nil {
+		return newStatus
+	}
+
+	newStatus.ComponentStatusMap = make(map[string]*status.AggregateStatus, len(otelStatus.ComponentStatusMap))
+	for k, v := range otelStatus.ComponentStatusMap {
+		newStatus.ComponentStatusMap[k] = deepCopyStatus(v)
+	}
+
+	return newStatus
+}

internal/pkg/otel/translate/status_test.go

@@ -180,8 +180,9 @@ func TestGetAllComponentState(t *testing.T) {
 
 func TestDropComponentStateFromOtelStatus(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		err := DropComponentStateFromOtelStatus(nil)
+		s, err := DropComponentStateFromOtelStatus(nil)
 		require.NoError(t, err)
+		require.Nil(t, s)
 	})
 
 	t.Run("drop non otel", func(t *testing.T) {
@@ -195,10 +196,10 @@ func TestDropComponentStateFromOtelStatus(t *testing.T) {
 				},
 			},
 		}
-		err := DropComponentStateFromOtelStatus(otelStatus)
+		s, err := DropComponentStateFromOtelStatus(otelStatus)
 		require.NoError(t, err)
-		assert.Len(t, otelStatus.ComponentStatusMap, 1)
-		assert.Contains(t, otelStatus.ComponentStatusMap, "pipeline:logs")
+		assert.Len(t, s.ComponentStatusMap, 1)
+		assert.Contains(t, s.ComponentStatusMap, "pipeline:logs")
 	})
 
 	t.Run("invalid status", func(t *testing.T) {
@@ -209,8 +210,9 @@ func TestDropComponentStateFromOtelStatus(t *testing.T) {
 				},
 			},
 		}
-		err := DropComponentStateFromOtelStatus(otelStatus)
+		s, err := DropComponentStateFromOtelStatus(otelStatus)
 		require.Error(t, err)
+		require.Nil(t, s)
 		assert.Equal(t, "pipeline status id logs is not a pipeline", err.Error())
 	})
 }