[otelmanager] emit starting state in the beginning (#11234)

* feat: emit starting state

* test

* rename method

* emit starting state

* fix test

* comments

* fix case

* ut

Author: VihasMakwana

internal/pkg/otel/manager/execution_subprocess.go

@@ -171,6 +171,9 @@ func (r *subprocessExecution) startCollector(ctx context.Context, baseLogger *lo
 					// after the collector exits, we need to report a nil status
 					r.reportSubprocessCollectorStatus(ctx, statusCh, nil)
 					return
+				default:
+					// if we face any other error (most likely, connection refused), log the error.
+					logger.Debugf("Received an unexpected error while fetching component status: %v", err)
 				}
 			} else {
 				maxFailuresTimer.Reset(maxFailuresDuration)

internal/pkg/otel/manager/manager_test.go

@@ -1649,6 +1649,90 @@ func TestManagerAlwaysEmitsStoppedStatesForComponents(t *testing.T) {
 	}, time.Millisecond, time.Second*5)
 }
 
+func TestManagerEmitsStartingStatesWhenHealthcheckIsUnavailable(t *testing.T) {
+	testLogger, _ := loggertest.New("test")
+	agentInfo := &info.AgentInfo{}
+	beatMonitoringConfigGetter := mockBeatMonitoringConfigGetter
+	collectorStarted := make(chan struct{})
+
+	execution := &mockExecution{
+		collectorStarted: collectorStarted,
+	}
+
+	// Create manager with test dependencies
+	mgr, err := NewOTelManager(
+		testLogger,
+		logp.DebugLevel,
+		testLogger,
+		config.SubprocessExecutionMode, // irrelevant, we'll override it
+		agentInfo,
+		nil,
+		beatMonitoringConfigGetter,
+		time.Second,
+	)
+	require.NoError(t, err)
+	mgr.recoveryTimer = newRestarterNoop()
+	mgr.execution = execution
+
+	// Start manager in a goroutine
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
+	defer cancel()
+
+	go func() {
+		err := mgr.Run(ctx)
+		assert.ErrorIs(t, err, context.Canceled)
+	}()
+
+	testComp := testComponent("test")
+	components := []component.Component{testComp}
+	otelStatus := &status.AggregateStatus{
+		Event: componentstatus.NewEvent(componentstatus.StatusStarting),
+	}
+	// start the collector by giving it a mock config
+	mgr.Update(nil, components)
+	select {
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for collector status update")
+	case <-execution.collectorStarted:
+	}
+
+	// send the status from the execution
+	select {
+	case <-ctx.Done():
+		t.Fatal("timeout waiting for collector status update")
+	case execution.statusCh <- otelStatus:
+	}
+
+	// verify we get the component Starting state from the manager
+	componentStates, err := getFromChannelOrErrorWithContext(t, ctx, mgr.WatchComponents(), mgr.Errors())
+	require.NoError(t, err)
+	require.NotNil(t, componentStates)
+	require.Len(t, componentStates, 1)
+	componentState := componentStates[0]
+	assert.Equal(t, componentState.State.State, client.UnitStateStarting)
+	assert.Equal(t, componentState.State.Message, "STARTING")
+
+	// stop the component by sending a nil config
+	mgr.Update(nil, nil)
+
+	// then send a nil status, indicating the collector is not running the component anymore
+	// do this a few times to see if the STOPPED state isn't lost along the way
+	for range 3 {
+		reportCollectorStatus(ctx, execution.statusCh, nil)
+		time.Sleep(time.Millisecond * 100) //  TODO: Replace this with synctest after we upgrade to Go 1.25
+	}
+
+	// verify that we get a STOPPED state for the component
+	assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+		componentStates, err := getFromChannelOrErrorWithContext(t, ctx, mgr.WatchComponents(), mgr.Errors())
+		require.NoError(collect, err)
+		require.NotNil(collect, componentStates)
+		require.Len(collect, componentStates, 1)
+		componentState := componentStates[0]
+		assert.Equal(collect, componentState.State.State, client.UnitStateStopped)
+	}, time.Millisecond, time.Second*5)
+}
+
 func getFromChannelOrErrorWithContext[T any](t *testing.T, ctx context.Context, ch <-chan T, errCh <-chan error) (T, error) {
 	t.Helper()
 	var result T

internal/pkg/otel/translate/status.go

@@ -39,6 +39,8 @@ func GetAllComponentStates(otelStatus *status.AggregateStatus, components []comp
 				if compState, statusErr = getComponentState(pipelineStatus, comp); statusErr != nil {
 					return nil, statusErr
 				}
+			} else if otelStatus != nil && otelStatus.Event != nil && otelStatus.Status() == componentstatus.StatusStarting {
+				compState = getComponentStartingState(comp)
 			} else {
 				// If the component is not found in the OTel status, we return a stopped state.
 				compState = runtime.ComponentComponentState{
@@ -290,6 +292,38 @@ func getComponentState(pipelineStatus *status.AggregateStatus, comp component.Co
 	return compStatus, nil
 }
 
+// getComponentStartingState returns a ComponentComponentState with all units in the starting state,
+// including version info and initial status for each unit.
+func getComponentStartingState(comp component.Component) runtime.ComponentComponentState {
+	compState := runtime.ComponentState{
+		State:   client.UnitStateStarting,
+		Message: client.UnitStateStarting.String(),
+		Units:   make(map[runtime.ComponentUnitKey]runtime.ComponentUnitState),
+		VersionInfo: runtime.ComponentVersionInfo{
+			Name: OtelComponentName,
+			Meta: map[string]string{ // mimic what beats return over the control protocol
+				"build_time": version.BuildTime().String(),
+				"commit":     version.Commit(),
+			},
+			BuildHash: version.Commit(),
+		},
+	}
+	for _, unit := range comp.Units {
+		unitKey := runtime.ComponentUnitKey{
+			UnitID:   unit.ID,
+			UnitType: unit.Type,
+		}
+		compState.Units[unitKey] = getComponentUnitState(&status.AggregateStatus{
+			Event:              componentstatus.NewEvent(componentstatus.StatusStarting),
+			ComponentStatusMap: map[string]*status.AggregateStatus{},
+		}, unit)
+	}
+	return runtime.ComponentComponentState{
+		Component: comp,
+		State:     compState,
+	}
+}
+
 // getUnitOtelStatuses extracts the receiver and exporter status from otel pipeline status.
 func getUnitOtelStatuses(pipelineStatus *status.AggregateStatus, comp component.Component) (
 	receiverStatuses map[otelcomponent.ID]*status.AggregateStatus,

internal/pkg/otel/translate/status_test.go

@@ -167,6 +167,53 @@ func TestGetAllComponentState(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:       "component state starting",
+			components: []component.Component{fileStreamOtelComponent},
+			otelStatus: &status.AggregateStatus{
+				Event:              componentstatus.NewEvent(componentstatus.StatusStarting),
+				ComponentStatusMap: map[string]*status.AggregateStatus{},
+			},
+			expected: []runtime.ComponentComponentState{
+				{
+					Component: fileStreamOtelComponent,
+					State: runtime.ComponentState{
+						State:   client.UnitStateStarting,
+						Message: "STARTING",
+						Units: map[runtime.ComponentUnitKey]runtime.ComponentUnitState{
+							runtime.ComponentUnitKey{UnitID: "filestream-unit", UnitType: client.UnitTypeInput}: {
+								State:   client.UnitStateStarting,
+								Message: "STARTING",
+								Payload: map[string]any{
+									"streams": map[string]map[string]string{
+										"test-1": {
+											"error":  "",
+											"status": client.UnitStateStarting.String(),
+										},
+										"test-2": {
+											"error":  "",
+											"status": client.UnitStateStarting.String(),
+										},
+									},
+								},
+							},
+							runtime.ComponentUnitKey{UnitID: "filestream-default", UnitType: client.UnitTypeOutput}: {
+								State:   client.UnitStateStarting,
+								Message: "STARTING",
+							},
+						},
+						VersionInfo: runtime.ComponentVersionInfo{
+							Name:      OtelComponentName,
+							BuildHash: version.Commit(),
+							Meta: map[string]string{
+								"build_time": version.BuildTime().String(),
+								"commit":     version.Commit(),
+							},
+						},
+					},
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {