Don't start components until Run is called

internal/component/otelcol/auth/auth.go

@@ -198,7 +198,7 @@ func (a *Auth) Update(args component.Arguments) error {
 	})
 
 	// Schedule the components to run once our component is running.
-	a.sched.Schedule(a.ctx, host, components...)
+	a.sched.Schedule(a.ctx, func() {}, host, components...)
 	return nil
 }
 

internal/component/otelcol/connector/connector.go

@@ -214,11 +214,12 @@ func (p *Connector) Update(args component.Arguments) error {
 		return errors.New("unsupported connector type")
 	}
 
+	updateConsumersFunc := func() {
+		p.consumer.SetConsumers(tracesConnector, metricsConnector, logsConnector)
+	}
+
 	// Schedule the components to run once our component is running.
-	p.consumer.Pause()
-	p.consumer.SetConsumers(tracesConnector, metricsConnector, logsConnector)
-	p.sched.Schedule(p.ctx, host, components...)
-	p.consumer.Resume()
+	p.sched.Schedule(p.ctx, updateConsumersFunc, host, components...)
 	return nil
 }
 

internal/component/otelcol/exporter/exporter.go

@@ -242,11 +242,12 @@ func (e *Exporter) Update(args component.Arguments) error {
 		}
 	}
 
+	updateConsumersFunc := func() {
+		e.consumer.SetConsumers(tracesExporter, metricsExporter, logsExporter)
+	}
+
 	// Schedule the components to run once our component is running.
-	e.consumer.Pause()
-	e.consumer.SetConsumers(tracesExporter, metricsExporter, logsExporter)
-	e.sched.Schedule(e.ctx, host, components...)
-	e.consumer.Resume()
+	e.sched.Schedule(e.ctx, updateConsumersFunc, host, components...)
 	return nil
 }
 

internal/component/otelcol/extension/extension.go

@@ -162,7 +162,7 @@ func (e *Extension) Update(args component.Arguments) error {
 	}
 
 	// Schedule the components to run once our component is running.
-	e.sched.Schedule(e.ctx, host, components...)
+	e.sched.Schedule(e.ctx, func() {}, host, components...)
 	return nil
 }
 

internal/component/otelcol/internal/scheduler/scheduler.go

@@ -37,13 +37,33 @@ type Scheduler struct {
 	schedMut        sync.Mutex
 	schedComponents []otelcomponent.Component // Most recently created components
 	host            otelcomponent.Host
+	running         bool
+
+	// onPause is called when scheduler is making changes to running components.
+	onPause func()
+	// onResume is called when scheduler is done making changes to running components.
+	onResume func()
 }
 
+// TODO: Delete this function? I don't think it's used anywhere.
 // New creates a new unstarted Scheduler. Call Run to start it, and call
 // Schedule to schedule components to run.
 func New(l log.Logger) *Scheduler {
 	return &Scheduler{
-		log: l,
+		log:      l,
+		onPause:  func() {},
+		onResume: func() {},
+	}
+}
+
+// TODO: Rename to "New"?
+// TODO: Write a new comment to explain what this method does.
+func NewWithPauseCallbacks(l log.Logger, onPause func(), onResume func()) *Scheduler {
+	//TODO: Instead of assuming that the scheduler is paused, just call onPause() here.
+	return &Scheduler{
+		log:      l,
+		onPause:  onPause,
+		onResume: onResume,
 	}
 }
 
@@ -53,19 +73,45 @@ func New(l log.Logger) *Scheduler {
 // Schedule completely overrides the set of previously running components;
 // components which have been removed since the last call to Schedule will be
 // stopped.
-func (cs *Scheduler) Schedule(ctx context.Context, h otelcomponent.Host, cc ...otelcomponent.Component) {
+func (cs *Scheduler) Schedule(ctx context.Context, updateConsumers func(), h otelcomponent.Host, cc ...otelcomponent.Component) {
 	cs.schedMut.Lock()
 	defer cs.schedMut.Unlock()
 
+	// If the scheduler isn't running yet, just update the state.
+	// That way the Run function is ready to go.
+	if !cs.running {
+		cs.schedComponents = cc
+		cs.host = h
+		updateConsumers()
+		return
+	}
+
+	cs.onPause()
+
 	// Stop the old components before running new scheduled ones.
 	cs.stopComponents(ctx, cs.schedComponents...)
 
-	level.Debug(cs.log).Log("msg", "scheduling components", "count", len(cc))
+	updateConsumers()
+
+	level.Debug(cs.log).Log("msg", "scheduling components", "count", len(cs.schedComponents))
 	cs.schedComponents = cs.startComponents(ctx, h, cc...)
+	cs.host = h
+	//TODO: Check if there were errors? What if the trace component failed but the metrics one didn't? Should we resume all consumers?
+
+	cs.onResume()
 }
 
 // Run starts the Scheduler and stops the components when the context is cancelled.
 func (cs *Scheduler) Run(ctx context.Context) error {
+	cs.schedMut.Lock()
+	cs.running = true
+
+	cs.onPause()
+	cs.startComponents(ctx, cs.host, cs.schedComponents...)
+	cs.onResume()
+
+	cs.schedMut.Unlock()
+
 	// Make sure we terminate all of our running components on shutdown.
 	defer func() {
 		cs.schedMut.Lock()

internal/component/otelcol/internal/scheduler/scheduler_test.go

@@ -30,7 +30,7 @@ func TestScheduler(t *testing.T) {
 		// Schedule our component, which should notify the started trigger once it is
 		// running.
 		component, started, _ := newTriggerComponent()
-		cs.Schedule(context.Background(), h, component)
+		cs.Schedule(context.Background(), func() {}, h, component)
 		require.NoError(t, started.Wait(5*time.Second), "component did not start")
 	})
 
@@ -50,12 +50,12 @@ func TestScheduler(t *testing.T) {
 		// Schedule our component, which should notify the started and stopped
 		// trigger once it starts and stops respectively.
 		component, started, stopped := newTriggerComponent()
-		cs.Schedule(context.Background(), h, component)
+		cs.Schedule(context.Background(), func() {}, h, component)
 
 		// Wait for the component to start, and then unschedule all components, which
 		// should cause our running component to terminate.
 		require.NoError(t, started.Wait(5*time.Second), "component did not start")
-		cs.Schedule(context.Background(), h)
+		cs.Schedule(context.Background(), func() {}, h)
 		require.NoError(t, stopped.Wait(5*time.Second), "component did not shutdown")
 	})
 
@@ -78,7 +78,7 @@ func TestScheduler(t *testing.T) {
 		// Schedule our component which will notify our trigger when Shutdown gets
 		// called.
 		component, started, stopped := newTriggerComponent()
-		cs.Schedule(ctx, h, component)
+		cs.Schedule(ctx, func() {}, h, component)
 
 		// Wait for the component to start, and then stop our scheduler, which
 		// should cause our running component to terminate.

internal/component/otelcol/processor/processor.go

@@ -117,7 +117,7 @@ func New(opts component.Options, f otelprocessor.Factory, args Arguments) (*Proc
 		factory:  f,
 		consumer: consumer,
 
-		sched:     scheduler.New(opts.Logger),
+		sched:     scheduler.NewWithPauseCallbacks(opts.Logger, consumer.Pause, consumer.Resume),
 		collector: collector,
 
 		liveDebuggingConsumer: livedebuggingconsumer.New(debugDataPublisher.(livedebugging.DebugDataPublisher), opts.ID),
@@ -139,6 +139,8 @@ func (p *Processor) Run(ctx context.Context) error {
 // configuration for OpenTelemetry Collector processor configuration and manage
 // the underlying OpenTelemetry Collector processor.
 func (p *Processor) Update(args component.Arguments) error {
+	//TODO: Lock a mutex? There could be a race condition with multiple calls to Update
+
 	p.args = args.(Arguments)
 
 	host := scheduler.NewHost(
@@ -237,11 +239,12 @@ func (p *Processor) Update(args component.Arguments) error {
 		}
 	}
 
+	updateConsumersFunc := func() {
+		p.consumer.SetConsumers(tracesProcessor, metricsProcessor, logsProcessor)
+	}
+
 	// Schedule the components to run once our component is running.
-	p.consumer.Pause()
-	p.consumer.SetConsumers(tracesProcessor, metricsProcessor, logsProcessor)
-	p.sched.Schedule(p.ctx, host, components...)
-	p.consumer.Resume()
+	p.sched.Schedule(p.ctx, updateConsumersFunc, host, components...)
 
 	return nil
 }

internal/component/otelcol/receiver/receiver.go

@@ -233,7 +233,7 @@ func (r *Receiver) Update(args component.Arguments) error {
 	}
 
 	// Schedule the components to run once our component is running.
-	r.sched.Schedule(r.ctx, host, components...)
+	r.sched.Schedule(r.ctx, func() {}, host, components...)
 	return nil
 }