From b3659f304a5b18021ae3b8d93958cfa967c104e4 Mon Sep 17 00:00:00 2001 From: Piotr <17101802+thampiotr@users.noreply.github.com> Date: Mon, 4 Aug 2025 14:08:33 +0100 Subject: [PATCH 1/2] Fix worker pool stuck on shutdown in certain cases --- .../runtime/internal/controller/loader.go | 6 +- .../runtime/internal/worker/worker_pool.go | 38 +++++++---- .../internal/worker/worker_pool_test.go | 66 ++++++++++++++++--- internal/runtime/module_test.go | 11 ++-- 4 files changed, 95 insertions(+), 26 deletions(-) diff --git a/internal/runtime/internal/controller/loader.go b/internal/runtime/internal/controller/loader.go index 4c62e98d105..105ea52095f 100644 --- a/internal/runtime/internal/controller/loader.go +++ b/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 diff --git a/internal/runtime/internal/worker/worker_pool.go b/internal/runtime/internal/worker/worker_pool.go index ee9c27c0c34..be1a0b700c1 100644 --- a/internal/runtime/internal/worker/worker_pool.go +++ b/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 } diff --git a/internal/runtime/internal/worker/worker_pool_test.go b/internal/runtime/internal/worker/worker_pool_test.go index 7b9b1b6a1e9..69aedba307a 100644 --- a/internal/runtime/internal/worker/worker_pool_test.go +++ b/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) + }) }) } diff --git a/internal/runtime/module_test.go b/internal/runtime/module_test.go index c226e5c1d41..2dfd2df2058 100644 --- a/internal/runtime/module_test.go +++ b/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) From 6323cacdb6152d2940ef131ce7a4bc1ee2c497da Mon Sep 17 00:00:00 2001 From: Piotr <17101802+thampiotr@users.noreply.github.com> Date: Mon, 4 Aug 2025 14:13:36 +0100 Subject: [PATCH 2/2] Changelog --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 726612edd98..24a234a55b1 100644 --- a/CHANGELOG.md +++ b/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 -----------------