Make otel runtime component updates synchronous (#10675) (#10692)

* Make otel runtime component updates synchronous

* Make the regression test runnable on main

* Add comments on final status in collector executions

* Add some more comments

(cherry picked from commit f950f1a)

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

Author: mergify[bot]

internal/pkg/otel/manager/execution.go

@@ -15,6 +15,11 @@ import (
 )
 
 type collectorExecution interface {
+	// startCollector starts the otel collector with the given arguments, returning a handle allowing it to be stopped.
+	// Cancelling the context will stop all goroutines involved in the execution.
+	// The collector will report status events in the statusCh channel and errors on errCh in a non-blocking fashion,
+	// draining the channel before writing to it.
+	// After the collector exits, it will emit an error describing the exit status (nil if successful) and a nil status.
 	startCollector(ctx context.Context, logger *logger.Logger, cfg *confmap.Conf, errCh chan error, statusCh chan *status.AggregateStatus) (collectorHandle, error)
 }
 

internal/pkg/otel/manager/execution_embedded.go

@@ -87,7 +87,9 @@ func (r *embeddedExecution) startCollector(ctx context.Context, logger *logger.L
 		}
 		runErr := svc.Run(collectorCtx)
 		close(ctl.collectorDoneCh)
+		// after the collector exits, we need to report the error and a nil status
 		reportErr(ctx, errCh, runErr)
+		reportCollectorStatus(ctx, statusCh, nil)
 	}()
 	return ctl, nil
 }

internal/pkg/otel/manager/execution_subprocess.go

@@ -168,7 +168,8 @@ func (r *subprocessExecution) startCollector(ctx context.Context, logger *logger
 			if err != nil {
 				switch {
 				case errors.Is(err, context.Canceled):
-					r.reportSubprocessCollectorStatus(ctx, statusCh, aggregateStatus(componentstatus.StatusStopped, nil))
+					// after the collector exits, we need to report a nil status
+					r.reportSubprocessCollectorStatus(ctx, statusCh, nil)
 					return
 				}
 			} else {
@@ -182,7 +183,8 @@ func (r *subprocessExecution) startCollector(ctx context.Context, logger *logger
 
 			select {
 			case <-procCtx.Done():
-				r.reportSubprocessCollectorStatus(ctx, statusCh, aggregateStatus(componentstatus.StatusStopped, nil))
+				// after the collector exits, we need to report a nil status
+				r.reportSubprocessCollectorStatus(ctx, statusCh, nil)
 				return
 			case <-healthCheckPollTimer.C:
 				healthCheckPollTimer.Reset(healthCheckPollDuration)
@@ -225,6 +227,10 @@ func (r *subprocessExecution) startCollector(ctx context.Context, logger *logger
 
 // cloneCollectorStatus creates a deep copy of the provided AggregateStatus.
 func cloneCollectorStatus(aStatus *status.AggregateStatus) *status.AggregateStatus {
+	if aStatus == nil {
+		return nil
+	}
+
 	st := &status.AggregateStatus{
 		Event: aStatus.Event,
 	}

internal/pkg/otel/manager/manager.go

@@ -181,7 +181,7 @@ func NewOTelManager(
 		beatMonitoringConfigGetter: beatMonitoringConfigGetter,
 		errCh:                      make(chan error, 1), // holds at most one error
 		collectorStatusCh:          make(chan *status.AggregateStatus, 1),
-		componentStateCh:           make(chan []runtime.ComponentComponentState, 1),
+		componentStateCh:           make(chan []runtime.ComponentComponentState),
 		updateCh:                   make(chan configUpdate, 1),
 		doneChan:                   make(chan struct{}),
 		execution:                  exec,
@@ -241,10 +241,6 @@ func (m *OTelManager) Run(ctx context.Context) error {
 				if m.proc != nil {
 					m.proc.Stop(m.stopTimeout)
 					m.proc = nil
-					updateErr := m.reportOtelStatusUpdate(ctx, nil)
-					if updateErr != nil {
-						reportErr(ctx, m.errCh, updateErr)
-					}
 				}
 
 				if m.mergedCollectorCfg == nil {
@@ -284,12 +280,6 @@ func (m *OTelManager) Run(ctx context.Context) error {
 				if m.proc != nil {
 					m.proc.Stop(m.stopTimeout)
 					m.proc = nil
-					// don't wait here for <-collectorRunErr, already occurred
-					// clear status, no longer running
-					updateErr := m.reportOtelStatusUpdate(ctx, nil)
-					if updateErr != nil {
-						err = errors.Join(err, updateErr)
-					}
 				}
 				// pass the error to the errCh so the coordinator, unless it's a cancel error
 				if !errors.Is(err, context.Canceled) {
@@ -429,30 +419,19 @@ func (m *OTelManager) applyMergedConfig(ctx context.Context, collectorStatusCh c
 	if m.proc != nil {
 		m.proc.Stop(m.stopTimeout)
 		m.proc = nil
+		// We wait here for the collector to exit before possibly starting a new one. The execution indicates this
+		// by sending an error over the appropriate channel. It will also send a nil status that we'll either process
+		// after exiting from this function and going back to the main loop, or it will be overridden by the status
+		// from the newly started collector.
+		// This is the only blocking wait inside the main loop involving channels, so we need to be extra careful not to
+		// deadlock.
+		// TODO: Verify if we need to wait for the error at all. Stop() is already blocking.
 		select {
 		case <-collectorRunErr:
 		case <-ctx.Done():
 			// our caller ctx is Done
 			return ctx.Err()
 		}
-		// drain the internal status update channel
-		// this status handling is normally done in the main loop, but in this case we want to ensure that we emit a
-		// nil status after the collector has stopped
-		select {
-		case statusCh := <-collectorStatusCh:
-			updateErr := m.reportOtelStatusUpdate(ctx, statusCh)
-			if updateErr != nil {
-				m.logger.Error("failed to update otel status", zap.Error(updateErr))
-			}
-		case <-ctx.Done():
-			// our caller ctx is Done
-			return ctx.Err()
-		default:
-		}
-		err := m.reportOtelStatusUpdate(ctx, nil)
-		if err != nil {
-			return err
-		}
 	}
 
 	if m.mergedCollectorCfg == nil {
@@ -625,18 +604,10 @@ func (m *OTelManager) maybeUpdateMergedConfig(mergedCfg *confmap.Conf) (updated
 	return !bytes.Equal(mergedCfgHash, previousConfigHash) || err != nil, err
 }
 
-// reportComponentStateUpdates sends component state updates to the component watch channel. It first drains
-// the channel to ensure that only the most recent status is kept, as intermediate statuses can be safely discarded.
-// This ensures the receiver always observes the latest reported status.
+// reportComponentStateUpdates sends component state updates to the component watch channel. It is synchronous and
+// blocking - the update must be received before this function returns. We are not allowed to drop older updates
+// in favor of newer ones here, as the coordinator expected incremental updates.
 func (m *OTelManager) reportComponentStateUpdates(ctx context.Context, componentUpdates []runtime.ComponentComponentState) {
-	select {
-	case <-ctx.Done():
-		// context is already done
-		return
-	case <-m.componentStateCh:
-	// drain the channel first
-	default:
-	}
 	select {
 	case m.componentStateCh <- componentUpdates:
 	case <-ctx.Done():