Fix worker pool stuck on shutdown in certain cases

CHANGELOG.md

@@ -67,6 +67,8 @@ Main (unreleased)
 
 - Fix issue with `faro.receiver` cors not allowing X-Scope-OrgID and traceparent headers. (@mar4uk)
 
+- Fixed an issue where certain `otelcol.*` components could prevent Alloy from shutting down when provided invalid configuration. (@thampiotr)
+
 v1.10.0
 -----------------
 

internal/runtime/internal/controller/loader.go

@@ -291,7 +291,11 @@ func (l *Loader) Apply(options ApplyOptions) diag.Diagnostics {
 // Cleanup unregisters any existing metrics and optionally stops the worker pool.
 func (l *Loader) Cleanup(stopWorkerPool bool) {
 	if stopWorkerPool {
-		l.workerPool.Stop()
+		// Wait at most 5 seconds for currently evaluating components to finish.
+		err := l.workerPool.Stop(time.Second * 5)
+		if err != nil {
+			level.Warn(l.log).Log("msg", "timed out stopping worker pool", "err", err)
+		}
 	}
 	if l.globals.Registerer == nil {
 		return

internal/runtime/internal/worker/worker_pool.go

@@ -4,16 +4,19 @@ import (
 	"fmt"
 	"runtime"
 	"sync"
+	"time"
 )
 
 type Pool interface {
 	// Stop stops the worker pool. It does not wait to drain any internal queues, but it does wait for the currently
-	// running tasks to complete. It must only be called once.
-	Stop()
+	// running tasks to complete up to the specified timeout. It must only be called once.
+	// Returns an error if the timeout is exceeded before all workers have stopped, indicating tasks that
+	// take longer than timeout to complete.
+	Stop(timeout time.Duration) error
 	// SubmitWithKey submits a function to be executed by the worker pool, ensuring that:
-	//   * Only one job with given key can be waiting to be executed at the time. This is desired if we don't want to
-	//     run the same task multiple times, e.g. if it's a component update that we only need to run once.
-	//   * Only one job with given key can be running at the time. This is desired when we don't want to duplicate work,
+	//   * Only one job with a given key can be waiting to be executed at the time. This is desired if we don't want to
+	//     run the same task multiple times, e.g., if it's a component update that we only need to run once.
+	//   * Only one job with a given key can be running at the time. This is desired when we don't want to duplicate work,
 	//     and we want to protect the pool from a slow task hogging all the workers.
 	//
 	// Note that it is possible to have two tasks with the same key in the pool at the same time: one waiting to be
@@ -43,7 +46,7 @@ func NewDefaultWorkerPool() Pool {
 
 // NewFixedWorkerPool creates a new Pool with the given number of workers and given max queue size.
 // The max queue size is the maximum number of tasks that can be queued OR running at the same time.
-// The tasks can run on a random worker, but workQueue ensures only one task with given key is running at a time.
+// The tasks can run on a random worker, but workQueue ensures only one task with a given key is running at a time.
 // The pool is automatically started and ready to accept work. To prevent resource leak, Stop() must be called when the
 // pool is no longer needed.
 func NewFixedWorkerPool(workersCount int, maxQueueSize int) Pool {
@@ -69,9 +72,20 @@ func (w *fixedWorkerPool) QueueSize() int {
 	return w.workQueue.queueSize()
 }
 
-func (w *fixedWorkerPool) Stop() {
+func (w *fixedWorkerPool) Stop(timeout time.Duration) error {
 	close(w.quit)
-	w.allStopped.Wait()
+	done := make(chan struct{})
+	go func() {
+		w.allStopped.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		return nil
+	case <-time.After(timeout):
+		return fmt.Errorf("worker pool did not stop within %v timeout", timeout)
+	}
 }
 
 func (w *fixedWorkerPool) start() {
@@ -114,7 +128,7 @@ func (w *workQueue) tryEnqueue(key string, f func()) (bool, error) {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 
-	// Don't enqueue if same task already waiting
+	// Don't enqueue if same the task already waiting
 	if _, exists := w.waiting[key]; exists {
 		return false, nil
 	}
@@ -144,7 +158,7 @@ func (w *workQueue) taskDone(key string) {
 }
 
 // emitNextTask emits the next eligible task to be run if there is one. It must be called whenever the queue state
-// changes (e.g. a task is added or a task finishes). The lock must be held when calling this function.
+// changes (e.g., a task is added or a task finishes). The lock must be held when calling this function.
 func (w *workQueue) emitNextTask() {
 	var (
 		task  func()
@@ -168,7 +182,7 @@ func (w *workQueue) emitNextTask() {
 
 	// Remove the task from waiting and add it to running set.
 	// NOTE: Even though we remove an element from the middle of a collection, we use a slice instead of a linked list.
-	// This code is NOT identified as a performance hot spot and given that in large Alloy instances we observe max number of
+	// This code is NOT identified as a performance hot spot and given that in large Alloy instances we observe the max number of
 	// tasks queued to be ~10, the slice is actually faster because it does not allocate memory. See BenchmarkQueue.
 	w.waitingOrder = append(w.waitingOrder[:index], w.waitingOrder[index+1:]...)
 	task = w.waiting[key]
@@ -181,7 +195,7 @@ func (w *workQueue) emitNextTask() {
 		task()
 	}
 
-	// Emit the task to be run. There will always be space in this buffered channel, because we limit queue size.
+	// Emit the task to be run. There will always be space in this buffered channel because we limit queue size.
 	w.tasksToRun <- wrapped
 }
 

internal/runtime/internal/worker/worker_pool_test.go

@@ -11,13 +11,15 @@ import (
 	"go.uber.org/goleak"
 )
 
+const testTimeout = 3 * time.Second
+
 func TestWorkerPool(t *testing.T) {
 	t.Run("worker pool", func(t *testing.T) {
 		t.Run("should start and stop cleanly", func(t *testing.T) {
 			defer goleak.VerifyNone(t)
 			pool := NewFixedWorkerPool(4, 1)
 			require.Equal(t, 0, pool.QueueSize())
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 		})
 
 		t.Run("should reject invalid worker count", func(t *testing.T) {
@@ -43,7 +45,7 @@ func TestWorkerPool(t *testing.T) {
 			defer goleak.VerifyNone(t)
 			done := make(chan struct{})
 			pool := NewFixedWorkerPool(4, 1)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 
 			err := pool.SubmitWithKey("123", func() {
 				done <- struct{}{}
@@ -62,7 +64,7 @@ func TestWorkerPool(t *testing.T) {
 			defer goleak.VerifyNone(t)
 			done := make(chan struct{})
 			pool := NewFixedWorkerPool(4, 1)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 
 			err := pool.SubmitWithKey("testKey", func() {
 				done <- struct{}{}
@@ -80,7 +82,7 @@ func TestWorkerPool(t *testing.T) {
 		t.Run("should not queue duplicated keys", func(t *testing.T) {
 			defer goleak.VerifyNone(t)
 			pool := NewFixedWorkerPool(4, 10)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 			tasksDone := atomic.Int32{}
 
 			// First task will block the worker
@@ -129,7 +131,7 @@ func TestWorkerPool(t *testing.T) {
 		t.Run("should concurrently process for different keys", func(t *testing.T) {
 			defer goleak.VerifyNone(t)
 			pool := NewFixedWorkerPool(4, 10)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 			tasksDone := atomic.Int32{}
 
 			// First task will block the worker
@@ -169,7 +171,7 @@ func TestWorkerPool(t *testing.T) {
 			defer goleak.VerifyNone(t)
 			// Pool with one worker and queue size of 1 - all work goes to one queue
 			pool := NewFixedWorkerPool(1, 2)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 			tasksDone := atomic.Int32{}
 
 			// First task will block the worker
@@ -204,7 +206,7 @@ func TestWorkerPool(t *testing.T) {
 
 			// Queue size is sufficient to queue all tasks
 			pool := NewFixedWorkerPool(3, tasksCount+1)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 			tasksDone := atomic.Int32{}
 
 			// First task will block
@@ -240,7 +242,7 @@ func TestWorkerPool(t *testing.T) {
 
 			// Queue size is sufficient to queue all tasks
 			pool := NewFixedWorkerPool(10, 10)
-			defer pool.Stop()
+			defer pool.Stop(testTimeout)
 			tasksDone := atomic.Int32{}
 
 			// First task will block
@@ -278,6 +280,54 @@ func TestWorkerPool(t *testing.T) {
 				return tasksDone.Load() == 2
 			}, 3*time.Second, 1*time.Millisecond)
 		})
+
+		t.Run("should timeout when stopping with stuck workers", func(t *testing.T) {
+			defer goleak.VerifyNone(t)
+			pool := NewFixedWorkerPool(2, 5)
+
+			// Submit a task that will block indefinitely
+			blockTask := make(chan struct{})
+			taskRunning := make(chan struct{})
+			taskDone := make(chan struct{})
+			err := pool.SubmitWithKey("blocking-task", func() {
+				taskRunning <- struct{}{}
+				<-blockTask // This will block forever
+				close(taskDone)
+			})
+			require.NoError(t, err)
+
+			// Wait for the task to start running
+			<-taskRunning
+
+			// Try to stop with a short timeout - should time out
+			shortTimeout := 100 * time.Millisecond
+			err = pool.Stop(shortTimeout)
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "worker pool did not stop within 100ms timeout")
+
+			// Clean up the blocking task
+			close(blockTask)
+			<-taskDone
+		})
+
+		t.Run("should stop successfully when no tasks are running", func(t *testing.T) {
+			defer goleak.VerifyNone(t, goleak.IgnoreAnyFunction("github.com/grafana/alloy/internal/runtime/internal/worker.(*fixedWorkerPool).Stop.func1"))
+			pool := NewFixedWorkerPool(2, 5)
+
+			// Submit a quick task that completes immediately
+			done := make(chan struct{}, 1)
+			err := pool.SubmitWithKey("quick-task", func() {
+				done <- struct{}{}
+			})
+			require.NoError(t, err)
+
+			// Wait for the task to complete
+			<-done
+
+			// Stop should succeed immediately
+			err = pool.Stop(testTimeout)
+			require.NoError(t, err)
+		})
 	})
 }
 

internal/runtime/module_test.go

@@ -6,14 +6,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/featuregate"
 	"github.com/grafana/alloy/internal/runtime/internal/controller"
 	"github.com/grafana/alloy/internal/runtime/internal/worker"
 	"github.com/grafana/alloy/internal/runtime/logging"
 	"github.com/grafana/alloy/internal/service"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/stretchr/testify/require"
 )
 
 const loggingConfig = `
@@ -127,7 +128,7 @@ func TestModule(t *testing.T) {
 			defer verifyNoGoroutineLeaks(t)
 			mc := newModuleController(testModuleControllerOptions(t)).(*moduleController)
 			// modules do not clean up their own worker pool as we normally use a shared one from the root controller
-			defer mc.o.WorkerPool.Stop()
+			defer mc.o.WorkerPool.Stop(5 * time.Second)
 
 			tm := &testModule{
 				content: tc.argumentModuleContent + tc.exportModuleContent,
@@ -192,7 +193,7 @@ func TestExportsWhenNotUsed(t *testing.T) {
 func TestIDList(t *testing.T) {
 	defer verifyNoGoroutineLeaks(t)
 	o := testModuleControllerOptions(t)
-	defer o.WorkerPool.Stop()
+	defer o.WorkerPool.Stop(5 * time.Second)
 	nc := newModuleController(o)
 	require.Len(t, nc.ModuleIDs(), 0)
 
@@ -226,7 +227,7 @@ func TestIDList(t *testing.T) {
 func TestDuplicateIDList(t *testing.T) {
 	defer verifyNoGoroutineLeaks(t)
 	o := testModuleControllerOptions(t)
-	defer o.WorkerPool.Stop()
+	defer o.WorkerPool.Stop(5 * time.Second)
 	nc := newModuleController(o)
 	require.Len(t, nc.ModuleIDs(), 0)