diff --git a/internal/collector/process/process.go b/internal/collector/process/process.go index 255d7b09..73f3f009 100644 --- a/internal/collector/process/process.go +++ b/internal/collector/process/process.go @@ -70,6 +70,7 @@ type Collector struct { perfDataCollector pdhtypes.Collector[perfDataCounterValues] perfDataObject []perfDataCounterValues workerCh chan processWorkerRequest + workerWG sync.WaitGroup lookupCache sync.Map @@ -179,6 +180,7 @@ func (c *Collector) Close() error { if c.workerCh != nil { close(c.workerCh) c.workerCh = nil + c.workerWG.Wait() } return nil @@ -216,7 +218,8 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { c.workerCh = make(chan processWorkerRequest, 32) for range 4 { - go c.collectWorker() + c.workerWG.Add(1) + go c.collectWorker(c.workerCh) } c.mu = sync.RWMutex{} diff --git a/internal/collector/process/process_worker.go b/internal/collector/process/process_worker.go index 338dd7fd..2d0ac6cb 100644 --- a/internal/collector/process/process_worker.go +++ b/internal/collector/process/process_worker.go @@ -93,22 +93,20 @@ func (c *Collector) collect(ch chan<- prometheus.Metric, maxScrapeDuration time. return err } -func (c *Collector) collectWorker() { - defer func() { - if r := recover(); r != nil { - c.logger.Error("Worker panic", - slog.Any("panic", r), - slog.String("stack", string(debug.Stack())), - ) +func (c *Collector) collectWorker(requests <-chan processWorkerRequest) { + defer c.workerWG.Done() - // Restart the collectWorker - go c.collectWorker() - } - }() - - for req := range c.workerCh { + for req := range requests { (func() { defer req.waitGroup.Done() + defer func() { + if r := recover(); r != nil { + c.logger.Error("Worker panic", + slog.Any("panic", r), + slog.String("stack", string(debug.Stack())), + ) + } + }() ch := req.ch name := req.name diff --git a/internal/collector/process/process_worker_test.go b/internal/collector/process/process_worker_test.go new file mode 100644 index 00000000..58f78f4a --- /dev/null +++ b/internal/collector/process/process_worker_test.go @@ -0,0 +1,89 @@ +// SPDX-License-Identifier: Apache-2.0 +// +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build windows + +package process + +import ( + "log/slog" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestCloseBeforeWorkerStarts(t *testing.T) { + t.Parallel() + + c := New(nil) + requests := make(chan processWorkerRequest) + c.workerCh = requests + c.workerWG.Add(1) + + closed := make(chan error, 1) + go func() { closed <- c.Close() }() + + select { + case _, ok := <-requests: + require.False(t, ok) + case <-time.After(time.Second): + t.Fatal("Close did not close the worker channel") + } + + // The worker starts after Close has set c.workerCh to nil. + go c.collectWorker(requests) + + select { + case err := <-closed: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Close did not wait for the delayed worker to exit") + } + + require.NoError(t, c.Close()) +} + +func TestWorkerContinuesAfterRequestPanic(t *testing.T) { + t.Parallel() + + c := New(nil) + c.logger = slog.New(slog.DiscardHandler) + c.workerCh = make(chan processWorkerRequest, 2) + + c.workerWG.Add(1) + go c.collectWorker(c.workerCh) + + var requests sync.WaitGroup + requests.Add(2) + + for range 2 { + // Missing descriptors make metric construction panic in both requests. + c.workerCh <- processWorkerRequest{waitGroup: &requests} + } + + processed := make(chan struct{}) + + go func() { requests.Wait(); close(processed) }() + + select { + case <-processed: + case <-time.After(time.Second): + t.Fatal("worker did not recover and process the next request") + } + + require.NoError(t, c.Close()) +}