-
Notifications
You must be signed in to change notification settings - Fork 806
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Create a goroutine worker pool to send data from distributors to inge…
…sters. (#6406) * Creating a worker pool to be used on distributors Signed-off-by: alanprot <[email protected]> * metric + test Signed-off-by: alanprot <[email protected]> * Changelog Signed-off-by: alanprot <[email protected]> --------- Signed-off-by: alanprot <[email protected]>
- Loading branch information
Showing
9 changed files
with
234 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package util | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promauto" | ||
| ) | ||
|
|
||
| // This code was based on: https://github.com/grpc/grpc-go/blob/66ba4b264d26808cb7af3c86eee66e843472915e/server.go | ||
|
|
||
| // serverWorkerResetThreshold defines how often the stack must be reset. Every | ||
| // N requests, by spawning a new goroutine in its place, a worker can reset its | ||
| // stack so that large stacks don't live in memory forever. 2^16 should allow | ||
| // each goroutine stack to live for at least a few seconds in a typical | ||
| // workload (assuming a QPS of a few thousand requests/sec). | ||
| const serverWorkerResetThreshold = 1 << 16 | ||
|
|
||
| type AsyncExecutor interface { | ||
| Submit(f func()) | ||
| Stop() | ||
| } | ||
|
|
||
| type noOpExecutor struct{} | ||
|
|
||
| func (n noOpExecutor) Stop() {} | ||
|
|
||
| func NewNoOpExecutor() AsyncExecutor { | ||
| return &noOpExecutor{} | ||
| } | ||
|
|
||
| func (n noOpExecutor) Submit(f func()) { | ||
| go f() | ||
| } | ||
|
|
||
| type workerPoolExecutor struct { | ||
| serverWorkerChannel chan func() | ||
| closeOnce sync.Once | ||
|
|
||
| fallbackTotal prometheus.Counter | ||
| } | ||
|
|
||
| func NewWorkerPool(name string, numWorkers int, reg prometheus.Registerer) AsyncExecutor { | ||
| wp := &workerPoolExecutor{ | ||
| serverWorkerChannel: make(chan func()), | ||
| fallbackTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{ | ||
| Namespace: "cortex", | ||
| Name: "worker_pool_fallback_total", | ||
| Help: "The total number additional go routines that needed to be created to run jobs.", | ||
| ConstLabels: prometheus.Labels{"name": name}, | ||
| }), | ||
| } | ||
|
|
||
| for i := 0; i < numWorkers; i++ { | ||
| go wp.run() | ||
| } | ||
|
|
||
| return wp | ||
| } | ||
|
|
||
| func (s *workerPoolExecutor) Stop() { | ||
| s.closeOnce.Do(func() { | ||
| close(s.serverWorkerChannel) | ||
| }) | ||
| } | ||
|
|
||
| func (s *workerPoolExecutor) Submit(f func()) { | ||
| select { | ||
| case s.serverWorkerChannel <- f: | ||
| default: | ||
| s.fallbackTotal.Inc() | ||
| go f() | ||
| } | ||
| } | ||
|
|
||
| func (s *workerPoolExecutor) run() { | ||
| for completed := 0; completed < serverWorkerResetThreshold; completed++ { | ||
| f, ok := <-s.serverWorkerChannel | ||
| if !ok { | ||
| return | ||
| } | ||
| f() | ||
| } | ||
| go s.run() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| package util | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "sync" | ||
| "testing" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/testutil" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestNewWorkerPool_CreateMultiplesPoolsWithSameRegistry(t *testing.T) { | ||
| reg := prometheus.NewPedanticRegistry() | ||
| wp1 := NewWorkerPool("test1", 100, reg) | ||
| defer wp1.Stop() | ||
| wp2 := NewWorkerPool("test2", 100, reg) | ||
| defer wp2.Stop() | ||
| } | ||
|
|
||
| func TestWorkerPool_TestMetric(t *testing.T) { | ||
| reg := prometheus.NewPedanticRegistry() | ||
| workerPool := NewWorkerPool("test1", 1, reg) | ||
| defer workerPool.Stop() | ||
|
|
||
| require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(` | ||
| # HELP cortex_worker_pool_fallback_total The total number additional go routines that needed to be created to run jobs. | ||
| # TYPE cortex_worker_pool_fallback_total counter | ||
| cortex_worker_pool_fallback_total{name="test1"} 0 | ||
| `), "cortex_worker_pool_fallback_total")) | ||
|
|
||
| wg := &sync.WaitGroup{} | ||
| wg.Add(1) | ||
|
|
||
| // Block the first job | ||
| workerPool.Submit(func() { | ||
| wg.Wait() | ||
| }) | ||
|
|
||
| // create an extra job to increment the metric | ||
| workerPool.Submit(func() {}) | ||
| require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(` | ||
| # HELP cortex_worker_pool_fallback_total The total number additional go routines that needed to be created to run jobs. | ||
| # TYPE cortex_worker_pool_fallback_total counter | ||
| cortex_worker_pool_fallback_total{name="test1"} 1 | ||
| `), "cortex_worker_pool_fallback_total")) | ||
|
|
||
| wg.Done() | ||
| } | ||
|
|
||
| func TestWorkerPool_ShouldFallbackWhenAllWorkersAreBusy(t *testing.T) { | ||
| reg := prometheus.NewPedanticRegistry() | ||
| numberOfWorkers := 10 | ||
| workerPool := NewWorkerPool("test1", numberOfWorkers, reg) | ||
| defer workerPool.Stop() | ||
|
|
||
| m := sync.Mutex{} | ||
| blockerWg := sync.WaitGroup{} | ||
| blockerWg.Add(numberOfWorkers) | ||
|
|
||
| // Lets lock all submited jobs | ||
| m.Lock() | ||
|
|
||
| for i := 0; i < numberOfWorkers; i++ { | ||
| workerPool.Submit(func() { | ||
| defer blockerWg.Done() | ||
| m.Lock() | ||
| m.Unlock() //nolint:staticcheck | ||
| }) | ||
| } | ||
|
|
||
| // At this point all workers should be busy. lets try to create a new job | ||
| wg := sync.WaitGroup{} | ||
| wg.Add(1) | ||
| workerPool.Submit(func() { | ||
| defer wg.Done() | ||
| }) | ||
|
|
||
| // Make sure the last job ran to the end | ||
| wg.Wait() | ||
|
|
||
| // Lets release the jobs | ||
| m.Unlock() | ||
|
|
||
| blockerWg.Wait() | ||
|
|
||
| } |