otelcol: synchronize Run and Shutdown lifecycle (open-telemetry#14989)

Right now `Shutdown()` returns before `Run()` is done cleaning up, which
means callers that treat "Shutdown returned" as "everything is freed"
can run into goroutine leaks and use-after-close problems. This is the
root of open-telemetry#4947.

The fix adds a `done` channel that `Run()` closes via `defer` when it
finishes, and an `atomic.Bool` so `Shutdown()` knows whether `Run()` was
called. If it was, `Shutdown()` blocks on `<-col.done` until `Run()` is
fully done. This sidesteps the WaitGroup idea from open-telemetry#8811 that deadlocks
when `Shutdown` happens before `Run`.

I also added a `CompareAndSwap` guard on the `Run()` entry so calling it
twice returns an error instead of panicking on the double-close of the
done channel, and the early-return path (when Shutdown was called before
Run) now properly shuts down the config provider so we don't leak
confmap resources.

Tests cover shutdown-before-run, shutdown-during-run, double-run, and
the blocking guarantee.

Fixes open-telemetry#4947

---------

Signed-off-by: Rajneesh Chaudhary <rajneeshrehsaan48@gmail.com>

Author: Rajneesh180

.chloggen/sync-run-shutdown-lifecycle.yaml

@@ -0,0 +1,29 @@
+# cspell:ignore lifecycles
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/otlp)
+component: pkg/otelcol
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Synchronize Collector Run and Shutdown lifecycles so that Shutdown blocks until Run completes all cleanup.
+
+# One or more tracking issues or pull requests related to the change
+issues: [4947]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Shutdown now blocks until Run finishes cleanup, matching http.Server semantics.
+  If Shutdown is called before Run, the next Run call returns nil after cleaning up
+  the config provider.
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user, api]

otelcol/collector.go

@@ -111,6 +111,9 @@ type Collector struct {
 	shutdownChan chan struct{}
 	shutdownOnce sync.Once
 
+	// wg is used by Shutdown to wait for Run to complete all cleanup.
+	wg sync.WaitGroup
+
 	// signalsChannel is used to receive termination signals from the OS.
 	signalsChannel chan os.Signal
 	// asyncErrorChannel is used to signal a fatal error from any component.
@@ -156,10 +159,12 @@ func (col *Collector) GetState() State {
 }
 
 // Shutdown shuts down the collector server.
+// If Run has been called, Shutdown blocks until Run completes all cleanup.
 func (col *Collector) Shutdown() {
 	col.shutdownOnce.Do(func() {
 		close(col.shutdownChan)
 	})
+	col.wg.Wait()
 }
 
 func buildModuleInfo(m map[component.Type]string) map[component.Type]service.ModuleInfo {
@@ -324,7 +329,22 @@ func newFallbackLogger(options []zap.Option) (*zap.Logger, error) {
 // Run starts the collector according to the given configuration, and waits for it to complete.
 // Consecutive calls to Run are not allowed, Run shouldn't be called once a collector is shut down.
 // Sets up the control logic for config reloading and shutdown.
+// If Shutdown was called before Run, Run returns nil after cleaning up resources.
 func (col *Collector) Run(ctx context.Context) error {
+	col.wg.Add(1)
+	defer col.wg.Done()
+
+	// If Shutdown was already called, return immediately without starting the service.
+	select {
+	case <-col.shutdownChan:
+		col.setCollectorState(StateClosed)
+		if err := col.configProvider.Shutdown(ctx); err != nil {
+			return fmt.Errorf("failed to shutdown config provider: %w", err)
+		}
+		return nil
+	default:
+	}
+
 	// setupConfigurationComponents is the "main" function responsible for startup
 	if err := col.setupConfigurationComponents(ctx); err != nil {
 		col.setCollectorState(StateClosed)

otelcol/collector_test.go

@@ -145,7 +145,7 @@ func TestCollectorStateAfterConfigChange(t *testing.T) {
 	watcher(&confmap.ChangeEvent{})
 	unblock = <-shutdownRequests
 	assert.Equal(t, StateClosing, col.GetState())
-	col.Shutdown()
+	go col.Shutdown() // Shutdown now blocks until Run completes, so signal asynchronously.
 	close(unblock)
 
 	// After the config reload, the final shutdown should occur.
@@ -410,6 +410,10 @@ func TestCollectorRun(t *testing.T) {
 
 			wg := startCollector(context.Background(), t, col)
 
+			assert.Eventually(t, func() bool {
+				return StateRunning == col.GetState()
+			}, 2*time.Second, 200*time.Millisecond)
+
 			col.Shutdown()
 			wg.Wait()
 			assert.Equal(t, StateClosed, col.GetState())
@@ -429,10 +433,75 @@ func TestCollectorRun_AfterShutdown(t *testing.T) {
 	// Calling shutdown before collector is running should cause it to return quickly
 	require.NotPanics(t, func() { col.Shutdown() })
 
+	// Run after Shutdown should return nil without starting the service.
+	err = col.Run(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, StateClosed, col.GetState())
+}
+
+func TestCollectorRun_AfterShutdown_ConfigProviderShutdownError(t *testing.T) {
+	set := CollectorSettings{
+		BuildInfo:              component.NewDefaultBuildInfo(),
+		Factories:              nopFactories,
+		ConfigProviderSettings: newDefaultConfigProviderSettings(t, []string{filepath.Join("testdata", "otelcol-nop.yaml")}),
+	}
+	col, err := NewCollector(set)
+	require.NoError(t, err)
+
+	col.Shutdown()
+
+	wantErr := errors.New("provider shutdown failed")
+	resolver, resolverErr := confmap.NewResolver(confmap.ResolverSettings{
+		URIs: []string{"err:config"},
+		ProviderFactories: []confmap.ProviderFactory{
+			confmap.NewProviderFactory(func(_ confmap.ProviderSettings) confmap.Provider {
+				return &errShutdownProvider{err: wantErr}
+			}),
+		},
+	})
+	require.NoError(t, resolverErr)
+	col.configProvider = &ConfigProvider{mapResolver: resolver}
+
+	runErr := col.Run(context.Background())
+	require.ErrorContains(t, runErr, "failed to shutdown config provider")
+	require.ErrorIs(t, runErr, wantErr)
+	assert.Equal(t, StateClosed, col.GetState())
+}
+
+func TestShutdownBlocksUntilRunCompletes(t *testing.T) {
+	set := CollectorSettings{
+		BuildInfo:              component.NewDefaultBuildInfo(),
+		Factories:              nopFactories,
+		ConfigProviderSettings: newDefaultConfigProviderSettings(t, []string{filepath.Join("testdata", "otelcol-nop.yaml")}),
+	}
+	col, err := NewCollector(set)
+	require.NoError(t, err)
+
+	// Start the collector in a goroutine.
 	wg := startCollector(context.Background(), t, col)
 
+	assert.Eventually(t, func() bool {
+		return StateRunning == col.GetState()
+	}, 2*time.Second, 200*time.Millisecond)
+
+	// Record whether Run has finished by the time Shutdown returns.
+	runFinished := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(runFinished)
+	}()
+
+	// Shutdown should block until Run completes.
 	col.Shutdown()
-	wg.Wait()
+
+	// After Shutdown returns, Run must have finished.
+	select {
+	case <-runFinished:
+		// expected: Run completed before or at the same time Shutdown returned.
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run did not complete after Shutdown returned")
+	}
+
 	assert.Equal(t, StateClosed, col.GetState())
 }
 
@@ -603,6 +672,22 @@ func (*failureProvider) Shutdown(context.Context) error {
 	return nil
 }
 
+type errShutdownProvider struct {
+	err error
+}
+
+func (p *errShutdownProvider) Retrieve(context.Context, string, confmap.WatcherFunc) (*confmap.Retrieved, error) {
+	return confmap.NewRetrieved(nil)
+}
+
+func (p *errShutdownProvider) Scheme() string {
+	return "err"
+}
+
+func (p *errShutdownProvider) Shutdown(context.Context) error {
+	return p.err
+}
+
 type fakeProvider struct {
 	scheme string
 	ret    func(ctx context.Context, uri string, watcher confmap.WatcherFunc) (*confmap.Retrieved, error)
@@ -708,6 +793,11 @@ func TestProviderAndConverterModules(t *testing.T) {
 	require.NoError(t, err)
 	wg := startCollector(context.Background(), t, col)
 	require.NoError(t, err)
+
+	assert.Eventually(t, func() bool {
+		return StateRunning == col.GetState()
+	}, 2*time.Second, 200*time.Millisecond)
+
 	providerModules := map[string]string{
 		"nop": "go.opentelemetry.io/collector/confmap/provider/testprovider v1.2.3",
 	}