Merge branch '8.19' into mergify/bp/8.19/pr-8592

Author: rogercoll

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())
 	})
 }