diff --git a/cmd/windows_exporter/0_service.go b/cmd/windows_exporter/0_service.go index 4a76561f..cd931ba9 100644 --- a/cmd/windows_exporter/0_service.go +++ b/cmd/windows_exporter/0_service.go @@ -24,6 +24,7 @@ import ( "log/slog" "os" "strings" + "sync" "unsafe" "golang.org/x/sys/windows" @@ -40,8 +41,9 @@ var ( // the exit code is sent to the service manager as well. exitCodeCh = make(chan int, 1) - // stopCh is a channel to send a signal to the service manager that the service is stopping. - stopCh = make(chan struct{}) + // serviceStop is closed when the service manager asks the service to stop. + // main cancels the root context then, so startup and the running exporter both observe it. + serviceStop = newBroadcast() // serviceManagerFinishedCh is a channel to send a signal to the main function that the service manager has stopped the service. serviceManagerFinishedCh = make(chan struct{}, 1) @@ -76,7 +78,11 @@ var IsService = func() bool { defer func() { go func() { - err := svc.Run(serviceName, &windowsExporterService{}) + err := svc.Run(serviceName, &windowsExporterService{ + stop: serviceStop, + exitCodeCh: exitCodeCh, + logEvent: logToEventToLog, + }) if err != nil { // https://github.com/open-telemetry/opentelemetry-collector/pull/9042 if !errors.Is(err, windows.ERROR_FAILED_SERVICE_CONTROLLER_CONNECT) { @@ -99,7 +105,37 @@ var IsService = func() bool { return true }() -type windowsExporterService struct{} +// errServiceStop is the cause of the root context's cancellation when the service manager stops the service. +var errServiceStop = errors.New("service stop requested") + +// broadcast is a channel that is closed once to signal all receivers. +type broadcast struct { + once sync.Once + ch chan struct{} +} + +func newBroadcast() *broadcast { + return &broadcast{ch: make(chan struct{})} +} + +// Close closes the channel. Later calls do nothing. +func (b *broadcast) Close() { + b.once.Do(func() { close(b.ch) }) +} + +// Done returns the channel that Close closes. +func (b *broadcast) Done() <-chan struct{} { + return b.ch +} + +type windowsExporterService struct { + // stop is closed when a stop or shutdown request is received. + stop *broadcast + // exitCodeCh receives the exit code of the main function. + exitCodeCh <-chan int + // logEvent writes to the event log. + logEvent func(eType uint16, msg string) error +} // Execute is the entry point for the Windows service manager. func (s *windowsExporterService) Execute(_ []string, r <-chan svc.ChangeRequest, changes chan<- svc.Status) (bool, uint32) { @@ -109,11 +145,11 @@ func (s *windowsExporterService) Execute(_ []string, r <-chan svc.ChangeRequest, for { select { - case exitCodeCh := <-exitCodeCh: + case exitCode := <-s.exitCodeCh: // Stop the service if an exit code from the main function is received. changes <- svc.Status{State: svc.StopPending} - return true, uint32(exitCodeCh) + return serviceExitCode(exitCode) case c := <-r: // Handle the service control request. switch c.Cmd { @@ -121,22 +157,29 @@ func (s *windowsExporterService) Execute(_ []string, r <-chan svc.ChangeRequest, changes <- c.CurrentStatus case svc.Stop, svc.Shutdown: // Stop the service if a stop or shutdown request is received. - _ = logToEventToLog(windows.EVENTLOG_INFORMATION_TYPE, "service stop received") + _ = s.logEvent(windows.EVENTLOG_INFORMATION_TYPE, "service stop received") changes <- svc.Status{State: svc.StopPending} - // Send a signal to the main function to stop the service. - stopCh <- struct{}{} + // Signal the main function to stop. This never blocks, + // even if the main function is still starting or has already returned. + s.stop.Close() // Wait for the main function to stop the service. - return false, uint32(<-exitCodeCh) + return serviceExitCode(<-s.exitCodeCh) default: - _ = logToEventToLog(windows.EVENTLOG_ERROR_TYPE, fmt.Sprintf("unexpected control request #%d", c)) + _ = s.logEvent(windows.EVENTLOG_ERROR_TYPE, fmt.Sprintf("unexpected control request #%d", c)) } } } } +// serviceExitCode converts an exit code of the main function into the return values of [svc.Handler.Execute]. +// A non-zero exit code is reported as a service-specific exit code. +func serviceExitCode(exitCode int) (bool, uint32) { + return exitCode != 0, uint32(exitCode) +} + // logToEventToLog logs a message to the Windows event log. func logToEventToLog(eType uint16, msg string) error { eventLog, err := eventlog.Open(serviceName) diff --git a/cmd/windows_exporter/main.go b/cmd/windows_exporter/main.go index 1b3220e4..bc9fbf33 100644 --- a/cmd/windows_exporter/main.go +++ b/cmd/windows_exporter/main.go @@ -48,11 +48,27 @@ import ( "golang.org/x/sys/windows" ) +// collectionCloseTimeout bounds how long the exporter waits for the collectors to close on shutdown. +const collectionCloseTimeout = 10 * time.Second + func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill) + // A stop request from the service manager cancels the root context, + // so that it aborts the startup as well as the running exporter. + ctx, cancel := context.WithCancelCause(ctx) + + go func() { + select { + case <-serviceStop.Done(): + cancel(errServiceStop) + case <-ctx.Done(): + } + }() + exitCode := run(ctx, os.Args[1:]) + cancel(nil) stop() // If we are running as a service, we need to signal the service control manager that we are done. @@ -178,12 +194,25 @@ func run(ctx context.Context, args []string) int { return 1 } + var disabledCollectorList []string if *disabledCollectors != "" { - collectors.Disable(slices.Compact(strings.Split(*disabledCollectors, ","))) + disabledCollectorList = slices.Compact(strings.Split(*disabledCollectors, ",")) + collectors.Disable(disabledCollectorList) } + // Close the collectors on every return from here on. On a regular shutdown, this runs after the HTTP server has shut down. + defer closeCollection(ctx, logger, collectors) + // Initialize collectors before loading if err = collectors.Build(ctx, logger); err != nil { + if ctx.Err() != nil { + logger.LogAttrs(ctx, slog.LevelInfo, "windows_exporter startup aborted", + slog.Any("reason", context.Cause(ctx)), + ) + + return 0 + } + for _, err := range utils.SplitError(err) { logger.LogAttrs(ctx, slog.LevelError, "couldn't initialize collector", slog.Any("err", err), @@ -193,7 +222,7 @@ func run(ctx context.Context, args []string) int { return 1 } - logger.InfoContext(ctx, "Enabled collectors: "+strings.Join(enabledCollectorList, ", ")) + logger.InfoContext(ctx, "Enabled collectors: "+strings.Join(effectiveCollectors(enabledCollectorList, disabledCollectorList), ", ")) mux := http.NewServeMux() mux.Handle("GET /health", httphandler.NewHealthHandler()) @@ -233,9 +262,11 @@ func run(ctx context.Context, args []string) int { select { case <-ctx.Done(): - logger.LogAttrs(ctx, slog.LevelInfo, "Shutting down windows_exporter via kill signal") - case <-stopCh: - logger.LogAttrs(ctx, slog.LevelInfo, "Shutting down windows_exporter via service control") + if errors.Is(context.Cause(ctx), errServiceStop) { + logger.LogAttrs(ctx, slog.LevelInfo, "Shutting down windows_exporter via service control") + } else { + logger.LogAttrs(ctx, slog.LevelInfo, "Shutting down windows_exporter via kill signal") + } case err := <-errCh: if err != nil { logger.LogAttrs(ctx, slog.LevelError, "Failed to start windows_exporter", @@ -321,6 +352,42 @@ func setPriorityWindows(ctx context.Context, logger *slog.Logger, pid int, prior return nil } +// closeCollection closes the collectors. It waits at most collectionCloseTimeout, +// so that a collector that doesn't return can't block the shutdown. +func closeCollection(ctx context.Context, logger *slog.Logger, collection *collector.Collection) { + errCh := make(chan error, 1) + + //nolint:contextcheck // Close has no context parameter. The select below bounds the wait instead. + go func() { + errCh <- collection.Close() + }() + + timer := time.NewTimer(collectionCloseTimeout) + defer timer.Stop() + + select { + case err := <-errCh: + if err == nil { + return + } + + for _, err := range utils.SplitError(err) { + logger.LogAttrs(ctx, slog.LevelWarn, "couldn't close collector", + slog.Any("err", err), + ) + } + case <-timer.C: + logger.LogAttrs(ctx, slog.LevelWarn, fmt.Sprintf("collectors didn't close within %s", collectionCloseTimeout)) + } +} + +// effectiveCollectors returns the sorted list of enabled collectors without the disabled ones. +func effectiveCollectors(enabled, disabled []string) []string { + return slices.DeleteFunc(slices.Compact(slices.Sorted(slices.Values(enabled))), func(name string) bool { + return slices.Contains(disabled, name) + }) +} + func expandEnabledCollectors(enabled string) []string { expanded := strings.ReplaceAll(enabled, "[defaults]", collector.DefaultCollectors) diff --git a/cmd/windows_exporter/service_test.go b/cmd/windows_exporter/service_test.go new file mode 100644 index 00000000..4a6122df --- /dev/null +++ b/cmd/windows_exporter/service_test.go @@ -0,0 +1,141 @@ +// 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 main + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/sys/windows/svc" +) + +func TestServiceExecuteStop(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + exitCode int + wantServiceSpecific bool + wantExitCode uint32 + }{ + {name: "clean stop", exitCode: 0}, + // The main function fails, for example because the stop aborted its startup. + {name: "failed stop", exitCode: 1, wantServiceSpecific: true, wantExitCode: 1}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + exitCodeCh := make(chan int, 1) + service := &windowsExporterService{ + stop: newBroadcast(), + exitCodeCh: exitCodeCh, + logEvent: func(uint16, string) error { return nil }, + } + + requests := make(chan svc.ChangeRequest) + changes := make(chan svc.Status, 10) + + type result struct { + serviceSpecific bool + exitCode uint32 + } + + done := make(chan result, 1) + + go func() { + serviceSpecific, exitCode := service.Execute(nil, requests, changes) + done <- result{serviceSpecific, exitCode} + }() + + requests <- svc.ChangeRequest{Cmd: svc.Stop} + + // Execute signals the stop without waiting for the main function to receive it. + select { + case <-service.stop.Done(): + case <-time.After(10 * time.Second): + t.Fatal("Execute didn't signal the stop") + } + + exitCodeCh <- tc.exitCode + + select { + case got := <-done: + require.Equal(t, result{tc.wantServiceSpecific, tc.wantExitCode}, got) + case <-time.After(10 * time.Second): + t.Fatal("Execute didn't return the exit code") + } + + require.Equal(t, svc.StartPending, (<-changes).State) + require.Equal(t, svc.Running, (<-changes).State) + require.Equal(t, svc.StopPending, (<-changes).State) + }) + } +} + +func TestBroadcast(t *testing.T) { + t.Parallel() + + b := newBroadcast() + + select { + case <-b.Done(): + t.Fatal("broadcast is done before Close") + default: + } + + b.Close() + b.Close() + + <-b.Done() +} + +func TestRunStartupCanceled(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancelCause(t.Context()) + cancel(errServiceStop) + + exitCodeCh := make(chan int, 1) + + go func() { + exitCodeCh <- run(ctx, []string{"--web.listen-address=127.0.0.1:0"}) + }() + + select { + case exitCode := <-exitCodeCh: + // A stop during startup is a clean stop. + require.Equal(t, 0, exitCode) + case <-time.After(30 * time.Second): + t.Fatal("run didn't return after its context was canceled") + } +} + +func TestEffectiveCollectors(t *testing.T) { + t.Parallel() + + require.Equal(t, + []string{"cpu", "memory", "os"}, + effectiveCollectors([]string{"os", "cpu", "net", "memory", "cpu"}, []string{"net", "unknown"}), + ) + require.Equal(t, + []string{"cpu", "net"}, + effectiveCollectors([]string{"net", "cpu"}, nil), + ) +} diff --git a/internal/httphandler/httphandler.go b/internal/httphandler/httphandler.go index d0865deb..da1d3936 100644 --- a/internal/httphandler/httphandler.go +++ b/internal/httphandler/httphandler.go @@ -18,8 +18,10 @@ package httphandler import ( + "context" "fmt" "log/slog" + "math" "net/http" "strconv" "time" @@ -34,7 +36,15 @@ import ( // Interface guard. var _ http.Handler = (*MetricsHTTPHandler)(nil) -const defaultScrapeTimeout = 10.0 +const ( + // defaultScrapeTimeout applies if the request has no valid X-Prometheus-Scrape-Timeout-Seconds header. + defaultScrapeTimeout = 10 * time.Second + // minScrapeTimeout is the smallest time given to the collectors. + minScrapeTimeout = 10 * time.Millisecond + // maxScrapeTimeout is the largest time given to the collectors. + // The exporter's HTTP server ends responses after 5 minutes anyway. + maxScrapeTimeout = 5 * time.Minute +) type MetricsHTTPHandler struct { metricCollectors *collector.Collection @@ -42,8 +52,8 @@ type MetricsHTTPHandler struct { // the exporter itself. exporterMetricsRegistry *prometheus.Registry - logger *slog.Logger - options Options + logger *slog.Logger + timeoutMargin time.Duration } type Options struct { @@ -62,7 +72,7 @@ func New(logger *slog.Logger, metricCollectors *collector.Collection, options *O handler := &MetricsHTTPHandler{ metricCollectors: metricCollectors, logger: logger, - options: *options, + timeoutMargin: toDuration(options.TimeoutMargin), } if !options.DisableExporterMetrics { @@ -82,9 +92,10 @@ func (c *MetricsHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { slog.String("remote", r.RemoteAddr), ) + // The scrape's deadline starts now, so time spent waiting for an earlier scrape counts against it. scrapeTimeout := c.getScrapeTimeout(logger, r) - handler, err := c.handlerFactory(logger, scrapeTimeout, r.URL.Query()["collect[]"]) + handler, err := c.handlerFactory(r.Context(), logger, scrapeTimeout, r.URL.Query()["collect[]"]) if err != nil { logger.WarnContext(r.Context(), "Couldn't create filtered metrics handler", slog.Any("err", err), @@ -99,32 +110,51 @@ func (c *MetricsHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { handler.ServeHTTP(w, r) } +// getScrapeTimeout returns the time the collectors get for this request. +// It is the client's X-Prometheus-Scrape-Timeout-Seconds minus the timeout margin, +// but the margin takes at most half of the client's timeout. +// The result is between minScrapeTimeout and maxScrapeTimeout. func (c *MetricsHTTPHandler) getScrapeTimeout(logger *slog.Logger, r *http.Request) time.Duration { - var timeoutSeconds float64 + timeout := defaultScrapeTimeout if v := r.Header.Get("X-Prometheus-Scrape-Timeout-Seconds"); v != "" { - var err error + seconds, err := strconv.ParseFloat(v, 64) - timeoutSeconds, err = strconv.ParseFloat(v, 64) - if err != nil { - logger.WarnContext(r.Context(), fmt.Sprintf("Couldn't parse X-Prometheus-Scrape-Timeout-Seconds: %q. Defaulting timeout to %f", v, defaultScrapeTimeout)) + switch { + case math.IsInf(seconds, 1): + // Includes values too large for a float64, which ParseFloat reports as an error. + timeout = maxScrapeTimeout + case err != nil, math.IsNaN(seconds), seconds <= 0: + logger.WarnContext(r.Context(), fmt.Sprintf("Invalid X-Prometheus-Scrape-Timeout-Seconds: %q. Defaulting timeout to %s", v, defaultScrapeTimeout)) + default: + timeout = toDuration(seconds) } } - if timeoutSeconds == 0 { - timeoutSeconds = defaultScrapeTimeout - } + timeout = max(timeout-c.timeoutMargin, timeout/2) - timeoutSeconds -= c.options.TimeoutMargin - - return time.Duration(timeoutSeconds*1e9) * time.Nanosecond + return min(max(timeout, minScrapeTimeout), maxScrapeTimeout) } -func (c *MetricsHTTPHandler) handlerFactory(logger *slog.Logger, scrapeTimeout time.Duration, requestedCollectors []string) (http.Handler, error) { +// toDuration converts seconds to a duration between 0 and maxScrapeTimeout. +// NaN and negative values become 0. +func toDuration(seconds float64) time.Duration { + if math.IsNaN(seconds) || seconds <= 0 { + return 0 + } + + if seconds >= maxScrapeTimeout.Seconds() { + return maxScrapeTimeout + } + + return time.Duration(seconds * float64(time.Second)) +} + +func (c *MetricsHTTPHandler) handlerFactory(ctx context.Context, logger *slog.Logger, scrapeTimeout time.Duration, requestedCollectors []string) (http.Handler, error) { reg := prometheus.NewRegistry() reg.MustRegister(version.NewCollector("windows_exporter")) - collectionHandler, err := c.metricCollectors.NewHandler(scrapeTimeout, c.logger, requestedCollectors) + collectionHandler, err := c.metricCollectors.NewHandlerWithContext(ctx, scrapeTimeout, logger, requestedCollectors) if err != nil { return nil, fmt.Errorf("couldn't create collector handler: %w", err) } @@ -133,17 +163,17 @@ func (c *MetricsHTTPHandler) handlerFactory(logger *slog.Logger, scrapeTimeout t return nil, fmt.Errorf("couldn't register Prometheus collector: %w", err) } + // Scrapes are serialized by the collection, so the handler doesn't limit requests in flight. var regHandler http.Handler if c.exporterMetricsRegistry != nil { regHandler = promhttp.HandlerFor( prometheus.Gatherers{c.exporterMetricsRegistry, reg}, promhttp.HandlerOpts{ - ErrorLog: slog.NewLogLogger(logger.Handler(), slog.LevelError), - ErrorHandling: promhttp.ContinueOnError, - MaxRequestsInFlight: 1, - Registry: c.exporterMetricsRegistry, - EnableOpenMetrics: true, - ProcessStartTime: c.metricCollectors.GetStartTime(), + ErrorLog: slog.NewLogLogger(logger.Handler(), slog.LevelError), + ErrorHandling: promhttp.ContinueOnError, + Registry: c.exporterMetricsRegistry, + EnableOpenMetrics: true, + ProcessStartTime: c.metricCollectors.GetStartTime(), }, ) @@ -156,11 +186,10 @@ func (c *MetricsHTTPHandler) handlerFactory(logger *slog.Logger, scrapeTimeout t regHandler = promhttp.HandlerFor( reg, promhttp.HandlerOpts{ - ErrorLog: slog.NewLogLogger(logger.Handler(), slog.LevelError), - ErrorHandling: promhttp.ContinueOnError, - MaxRequestsInFlight: 1, - EnableOpenMetrics: true, - ProcessStartTime: c.metricCollectors.GetStartTime(), + ErrorLog: slog.NewLogLogger(logger.Handler(), slog.LevelError), + ErrorHandling: promhttp.ContinueOnError, + EnableOpenMetrics: true, + ProcessStartTime: c.metricCollectors.GetStartTime(), }, ) } diff --git a/internal/httphandler/httphandler_test.go b/internal/httphandler/httphandler_test.go index 3f1be0bf..b3d64140 100644 --- a/internal/httphandler/httphandler_test.go +++ b/internal/httphandler/httphandler_test.go @@ -18,8 +18,10 @@ package httphandler_test import ( + "context" "errors" "log/slog" + "math" "net/http" "net/http/httptest" "testing" @@ -154,6 +156,87 @@ func TestMetricsHTTPHandlerScrapeTimeout(t *testing.T) { } } +func TestMetricsHTTPHandlerScrapeBudget(t *testing.T) { + const maxScrapeTimeout = 5 * time.Minute + + for _, tc := range []struct { + name string + header string + margin float64 + want time.Duration + }{ + {name: "missing header", margin: 0.5, want: 9500 * time.Millisecond}, + {name: "invalid header", header: "invalid", margin: 0.5, want: 9500 * time.Millisecond}, + {name: "zero", header: "0", margin: 0.5, want: 9500 * time.Millisecond}, + {name: "negative", header: "-1", margin: 0.5, want: 9500 * time.Millisecond}, + {name: "negative infinity", header: "-Inf", margin: 0.5, want: 9500 * time.Millisecond}, + {name: "NaN", header: "NaN", margin: 0.5, want: 9500 * time.Millisecond}, + {name: "infinity", header: "+Inf", margin: 0.5, want: maxScrapeTimeout - 500*time.Millisecond}, + {name: "out of float range", header: "1e400", margin: 0.5, want: maxScrapeTimeout - 500*time.Millisecond}, + {name: "huge", header: "1e12", margin: 0.5, want: maxScrapeTimeout - 500*time.Millisecond}, + {name: "long", header: "120", margin: 0.5, want: 119500 * time.Millisecond}, + {name: "regular", header: "2", margin: 0.5, want: 1500 * time.Millisecond}, + {name: "margin larger than timeout", header: "0.3", margin: 0.5, want: 150 * time.Millisecond}, + {name: "margin takes at most half", header: "0.8", margin: 0.5, want: 400 * time.Millisecond}, + {name: "tiny", header: "0.000001", margin: 0.5, want: 10 * time.Millisecond}, + {name: "no margin", header: "2", want: 2 * time.Second}, + {name: "negative margin", header: "2", margin: -1, want: 2 * time.Second}, + {name: "NaN margin", header: "2", margin: math.NaN(), want: 2 * time.Second}, + } { + t.Run(tc.name, func(t *testing.T) { + // Synthetic time doesn't pass between the request and the collector's call. + synctest.Test(t, func(t *testing.T) { + test := &budgetCollector{} + collection := collector.New(collector.Map{"test": test}) + handler := httphandler.New(slog.New(slog.DiscardHandler), collection, &httphandler.Options{DisableExporterMetrics: true, TimeoutMargin: tc.margin}) + request := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/metrics", nil) + + if tc.header != "" { + request.Header.Set("X-Prometheus-Scrape-Timeout-Seconds", tc.header) + } + + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + + require.Equal(t, http.StatusOK, response.Code) + require.Contains(t, response.Body.String(), `windows_exporter_collector_success{collector="test"} 1`) + require.Equal(t, tc.want, test.budget) + }) + }) + } +} + +func TestMetricsHTTPHandlerCanceledRequest(t *testing.T) { + test := &budgetCollector{} + collection := collector.New(collector.Map{"test": test}) + handler := httphandler.New(slog.New(slog.DiscardHandler), collection, &httphandler.Options{DisableExporterMetrics: true}) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + response := httptest.NewRecorder() + handler.ServeHTTP(response, httptest.NewRequestWithContext(ctx, http.MethodGet, "/metrics", nil)) + + // The client is gone, so the collectors don't run. + require.False(t, test.called) +} + +// budgetCollector records the time it is given. +type budgetCollector struct { + called bool + budget time.Duration +} + +func (c *budgetCollector) GetName() string { return "test" } +func (c *budgetCollector) Build(_ *slog.Logger, _ *mi.Session) error { return nil } +func (c *budgetCollector) Close() error { return nil } +func (c *budgetCollector) Collect(_ chan<- prometheus.Metric, maxScrapeDuration time.Duration) error { + c.called = true + c.budget = maxScrapeDuration + + return nil +} + type testCollector struct { name string err error diff --git a/pkg/collector/collect.go b/pkg/collector/collect.go index 17272047..3ff38953 100644 --- a/pkg/collector/collect.go +++ b/pkg/collector/collect.go @@ -24,7 +24,6 @@ import ( "log/slog" "runtime/debug" "sync" - "sync/atomic" "time" "github.com/prometheus-community/windows_exporter/internal/pdh" @@ -41,17 +40,42 @@ type collectorStatus struct { type collectorStatusCode int const ( + // pending means the collector timed out, or was skipped because an earlier call is still running. pending collectorStatusCode = iota success failed + // unavailable means the collector was skipped because its Build failed or the collection is closed. + unavailable ) -func (c *Collection) collectAll(ch chan<- prometheus.Metric, logger *slog.Logger, maxScrapeDuration time.Duration) { +// collectAll runs all collectors once. It returns when ctx is done, even if collectors are still running. +// ctx must have a deadline. +func (c *Collection) collectAll(ctx context.Context, ch chan<- prometheus.Metric, logger *slog.Logger) { collectorStartTime := time.Now() + if err := acquireScrapeSlot(ctx); err != nil { + level := slog.LevelWarn + if errors.Is(err, context.Canceled) { + level = slog.LevelDebug + } + + logger.LogAttrs(ctx, level, "scrape not started, because an earlier scrape was still running", + slog.Any("err", err), + ) + + for name := range c.collectors { + c.sendCollectorStatus(ch, collectorStatus{name: name, statusCode: pending}) + } + + c.sendScrapeDuration(ch, collectorStartTime) + + return + } + + defer releaseScrapeSlot() + // WaitGroup to wait for all collectors to finish wg := sync.WaitGroup{} - wg.Add(len(c.collectors)) // Using a channel to collect the status of each collector // A channel is safe to use concurrently while a map is not @@ -60,14 +84,12 @@ func (c *Collection) collectAll(ch chan<- prometheus.Metric, logger *slog.Logger // Execute all collectors concurrently // timeout handling is done in the execute function for name, metricsCollector := range c.collectors { - go func(name string, metricsCollector Collector) { - defer wg.Done() - + wg.Go(func() { collectorStatusCh <- collectorStatus{ name: name, - statusCode: c.collectCollector(ch, logger, name, metricsCollector, maxScrapeDuration), + statusCode: c.collectCollector(ctx, ch, logger, name, metricsCollector), } - }(name, metricsCollector) + }) } // Wait for all collectors to finish @@ -77,30 +99,38 @@ func (c *Collection) collectAll(ch chan<- prometheus.Metric, logger *slog.Logger close(collectorStatusCh) for status := range collectorStatusCh { - var successValue, timeoutValue float64 - if status.statusCode == pending { - timeoutValue = 1.0 - } - - if status.statusCode == success { - successValue = 1.0 - } - - ch <- prometheus.MustNewConstMetric( - c.collectorScrapeSuccessDesc, - prometheus.GaugeValue, - successValue, - status.name, - ) - - ch <- prometheus.MustNewConstMetric( - c.collectorScrapeTimeoutDesc, - prometheus.GaugeValue, - timeoutValue, - status.name, - ) + c.sendCollectorStatus(ch, status) } + c.sendScrapeDuration(ch, collectorStartTime) +} + +func (c *Collection) sendCollectorStatus(ch chan<- prometheus.Metric, status collectorStatus) { + var successValue, timeoutValue float64 + if status.statusCode == pending { + timeoutValue = 1.0 + } + + if status.statusCode == success { + successValue = 1.0 + } + + ch <- prometheus.MustNewConstMetric( + c.collectorScrapeSuccessDesc, + prometheus.GaugeValue, + successValue, + status.name, + ) + + ch <- prometheus.MustNewConstMetric( + c.collectorScrapeTimeoutDesc, + prometheus.GaugeValue, + timeoutValue, + status.name, + ) +} + +func (c *Collection) sendScrapeDuration(ch chan<- prometheus.Metric, collectorStartTime time.Time) { ch <- prometheus.MustNewConstMetric( c.scrapeDurationDesc, prometheus.GaugeValue, @@ -108,105 +138,118 @@ func (c *Collection) collectAll(ch chan<- prometheus.Metric, logger *slog.Logger ) } -func (c *Collection) collectCollector(ch chan<- prometheus.Metric, logger *slog.Logger, name string, collector Collector, maxScrapeDuration time.Duration) collectorStatusCode { - var ( - err error - numMetrics int - duration time.Duration - timeout atomic.Bool - ) +// collectCollector runs one collector and forwards its metrics to ch until ctx is done. +// +// Only one Collect call runs per collector instance. A call that outlives ctx keeps running in the +// background; its late metrics are dropped, and later scrapes skip the collector until the call returns. +func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus.Metric, logger *slog.Logger, name string, collector Collector) collectorStatusCode { + deadline, _ := ctx.Deadline() + + maxScrapeDuration := time.Until(deadline) + if ctx.Err() != nil || maxScrapeDuration <= 0 { + return pending + } + + state := c.state.collector(name) + + switch state.tryCollect() { + case started: + case notAvailable: + logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf("collector %s skipped, because it isn't initialized", name)) + + return unavailable + case busy: + level := slog.LevelDebug + if state.logBusy() { + level = slog.LevelWarn + } + + logger.LogAttrs(ctx, level, fmt.Sprintf("collector %s skipped, because its call from an earlier scrape is still running", name)) + + return pending + } // bufCh is a buffer channel to store the metrics // This is needed because once timeout is reached, the prometheus registry channel is closed. bufCh := make(chan prometheus.Metric, 1000) errCh := make(chan error, 1) - ctx, cancel := context.WithTimeout(context.Background(), maxScrapeDuration) - defer cancel() + t := time.Now() - // execute the collector + // execute the collector. The goroutine owns the collector's slot until Collect returns. go func() { - defer func() { - if r := recover(); r != nil { - errCh <- fmt.Errorf("panic in collector %s: %v. stack: %s", name, r, - string(debug.Stack()), + // Release the slot before closing bufCh, so the next scrape never sees this call as running. + defer close(bufCh) + defer state.release() + + errCh <- collect(name, collector, bufCh, maxScrapeDuration) + }() + + numMetrics := 0 + + // Pass metrics to the prometheus registry until the collector finishes or the scrape ends. + // This is the only goroutine that writes to ch, and it stops before collectCollector returns. + for { + select { + case m, ok := <-bufCh: + if ok { + ch <- m + + numMetrics++ + + continue + } + + duration := time.Since(t) + c.sendCollectorDuration(ch, name, duration) + + return logCollectorResult(ctx, logger, name, <-errCh, duration, numMetrics) + case <-ctx.Done(): + // Drain late metrics, so that Collect doesn't block on a full buffer. + go func() { + for range bufCh { + } + }() + + duration := time.Since(t) + c.sendCollectorDuration(ch, name, duration) + + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + logger.LogAttrs(ctx, slog.LevelWarn, fmt.Sprintf("collector %s timeouted after %s, resulting in %d metrics", name, duration, numMetrics)) + } else { + logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf("collector %s canceled after %s, resulting in %d metrics", name, duration, numMetrics), + slog.Any("err", ctx.Err()), ) } - close(bufCh) - }() + return pending + } + } +} - errCh <- collector.Collect(bufCh, maxScrapeDuration) - }() - - wg := sync.WaitGroup{} - wg.Add(1) - - go func() { - defer func() { - // This prevents a panic from race-condition when closing the ch channel too early. - _ = recover() - - wg.Done() - }() - - // Pass metrics to the prometheus registry - // If timeout is reached, the channel is closed. - // This will cause a panic if we try to write to it. - for { - select { - case <-ctx.Done(): - return - case m, ok := <-bufCh: - if !ok { - return - } - - if !timeout.Load() { - ch <- m - - numMetrics++ - } - } +// collect calls Collector.Collect and turns a panic into an error. +func collect(name string, collector Collector, ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("panic in collector %s: %v. stack: %s", name, r, + string(debug.Stack()), + ) } }() - t := time.Now() + return collector.Collect(ch, maxScrapeDuration) +} - // Wait for the collector to finish or timeout - select { - case err = <-errCh: - wg.Wait() // Wait for the buffer channel to be closed and empty - - duration = time.Since(t) - ch <- prometheus.MustNewConstMetric( - c.collectorScrapeDurationDesc, - prometheus.GaugeValue, - duration.Seconds(), - name, - ) - case <-ctx.Done(): - timeout.Store(true) - - duration = time.Since(t) - ch <- prometheus.MustNewConstMetric( - c.collectorScrapeDurationDesc, - prometheus.GaugeValue, - duration.Seconds(), - name, - ) - - logger.LogAttrs(ctx, slog.LevelWarn, fmt.Sprintf("collector %s timeouted after %s, resulting in %d metrics", name, maxScrapeDuration, numMetrics)) - - go func() { - // Drain channel in case of premature return to not leak a goroutine. - for range bufCh { - } - }() - - return pending - } +func (c *Collection) sendCollectorDuration(ch chan<- prometheus.Metric, name string, duration time.Duration) { + ch <- prometheus.MustNewConstMetric( + c.collectorScrapeDurationDesc, + prometheus.GaugeValue, + duration.Seconds(), + name, + ) +} +func logCollectorResult(ctx context.Context, logger *slog.Logger, name string, err error, duration time.Duration, numMetrics int) collectorStatusCode { slogAttrs := make([]slog.Attr, 0) result := "succeeded" diff --git a/pkg/collector/collect_test.go b/pkg/collector/collect_test.go new file mode 100644 index 00000000..2ace88a0 --- /dev/null +++ b/pkg/collector/collect_test.go @@ -0,0 +1,574 @@ +// 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 collector_test + +import ( + "context" + "errors" + "log/slog" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/pkg/collector" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/common/expfmt" + "github.com/stretchr/testify/require" +) + +// The tests in this file share the package-level scrape slot, so they don't run in parallel. + +func TestCollectTimedOutCollectorDoesNotOverlap(t *testing.T) { + // The collector's delay and the scrape deadlines use synthetic time. + synctest.Test(t, func(t *testing.T) { + logs := &recordHandler{} + logger := slog.New(logs) + + slow := &fakeCollector{name: "slow", delay: 2 * time.Second} + fast := &fakeCollector{name: "fast"} + collection := collector.New(collector.Map{"slow": slow, "fast": fast}) + + all, err := collection.NewHandler(300*time.Millisecond, logger, nil) + require.NoError(t, err) + + // A filtered view must share the running state with the full collection. + filtered, err := collection.NewHandler(300*time.Millisecond, logger, []string{"slow"}) + require.NoError(t, err) + + start := time.Now() + + for i := range 5 { + handler := all + if i%2 == 1 { + handler = filtered + } + + out := scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_timeout{collector="slow"} 1`) + require.Contains(t, out, `windows_exporter_collector_success{collector="slow"} 0`) + require.NotContains(t, out, "windows_test_slow") + + if handler == all { + require.Contains(t, out, `windows_exporter_collector_success{collector="fast"} 1`) + require.Contains(t, out, "windows_test_fast 42") + } + } + + // Only the first scrape waited for its deadline, because the others skipped the running collector. + require.Equal(t, 300*time.Millisecond, time.Since(start)) + + // The first call is still running, and no second call was started. + require.Equal(t, int32(1), slow.calls.Load()) + require.Equal(t, int32(1), slow.maxActive.Load()) + require.Equal(t, 1, logs.warnings("timeouted")) + require.Equal(t, 1, logs.warnings("is still running")) + + // After the first call returns, the next scrape starts a new one. + time.Sleep(2 * time.Second) + synctest.Wait() + + scrape(t, all) + require.Equal(t, int32(2), slow.calls.Load()) + require.Equal(t, int32(1), slow.maxActive.Load()) + + // Let the second call finish before leaving the bubble. + time.Sleep(2 * time.Second) + synctest.Wait() + }) +} + +func TestCollectAdmission(t *testing.T) { + t.Run("canceled while queued", func(t *testing.T) { + running := newBlockingCollector() + collection := collector.New(collector.Map{"test": running}) + first := scrapeInBackground(t.Context(), collection, time.Minute) + + <-running.started + + ctx, cancel := context.WithCancel(t.Context()) + second := scrapeInBackground(ctx, collection, time.Minute) + + // Give the second scrape time to queue behind the first one. The result is the same if it hasn't yet. + time.Sleep(50 * time.Millisecond) + cancel() + + select { + case out := <-second: + require.Contains(t, out, `windows_exporter_collector_timeout{collector="test"} 1`) + case <-time.After(10 * time.Second): + t.Fatal("canceled scrape didn't return") + } + + close(running.block) + require.Contains(t, <-first, `windows_exporter_collector_success{collector="test"} 1`) + require.Equal(t, int32(1), running.calls.Load()) + }) + + t.Run("deadline passes while queued", func(t *testing.T) { + running := newBlockingCollector() + collection := collector.New(collector.Map{"test": running}) + first := scrapeInBackground(t.Context(), collection, time.Minute) + + <-running.started + + start := time.Now() + second := scrapeInBackground(t.Context(), collection, 100*time.Millisecond) + + select { + case out := <-second: + require.Contains(t, out, `windows_exporter_collector_timeout{collector="test"} 1`) + require.Less(t, time.Since(start), 10*time.Second) + case <-time.After(10 * time.Second): + t.Fatal("queued scrape didn't return at its deadline") + } + + close(running.block) + <-first + require.Equal(t, int32(1), running.calls.Load()) + }) + + t.Run("canceled before admission", func(t *testing.T) { + test := &fakeCollector{name: "test"} + collection := collector.New(collector.Map{"test": test}) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + handler, err := collection.NewHandlerWithContext(ctx, time.Minute, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + require.Contains(t, scrape(t, handler), `windows_exporter_collector_timeout{collector="test"} 1`) + require.Equal(t, int32(0), test.calls.Load()) + }) + + t.Run("deadline starts when the handler is created", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := &fakeCollector{name: "test"} + collection := collector.New(collector.Map{"test": test}) + + handler, err := collection.NewHandlerWithContext(t.Context(), 10*time.Second, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + time.Sleep(4 * time.Second) + + require.Contains(t, scrape(t, handler), `windows_exporter_collector_success{collector="test"} 1`) + require.Equal(t, 6*time.Second, time.Duration(test.budget.Load())) + }) + }) +} + +func TestCollectFailedBuild(t *testing.T) { + for _, tc := range []struct { + name string + buildErr error + panics bool + wantErr string + }{ + {name: "tolerated error", buildErr: pdh.ErrNoData}, + {name: "error", buildErr: errors.New("build failed"), wantErr: "build failed"}, + {name: "panic", panics: true, wantErr: "panic: build panic"}, + } { + t.Run(tc.name, func(t *testing.T) { + logs := &recordHandler{} + logger := slog.New(logs) + + failing := &fakeCollector{name: "failing", buildErr: tc.buildErr, buildPanic: tc.panics} + working := &fakeCollector{name: "working"} + collection := collector.New(collector.Map{"failing": failing, "working": working}) + + err := collection.Build(t.Context(), logger) + if tc.wantErr == "" { + require.NoError(t, err) + require.Equal(t, 1, logs.warnings("couldn't initialize collector")) + } else { + require.ErrorContains(t, err, tc.wantErr) + } + + if tc.panics { + // The error carries the stack of the panic. + require.ErrorContains(t, err, "runtime/debug.Stack") + } + + handler, err := collection.NewHandler(time.Minute, logger, nil) + require.NoError(t, err) + + logs.reset() + + for range 3 { + out := scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_success{collector="failing"} 0`) + require.Contains(t, out, `windows_exporter_collector_timeout{collector="failing"} 0`) + require.Contains(t, out, `windows_exporter_collector_success{collector="working"} 1`) + require.NotContains(t, out, "windows_test_failing") + } + + require.Equal(t, int32(0), failing.calls.Load()) + require.Equal(t, int32(3), working.calls.Load()) + require.Zero(t, logs.warnings(""), "a collector that failed to build must not log on every scrape") + + // The collector may hold resources from its partial Build, so it is closed. + require.NoError(t, collection.Close()) + require.Equal(t, int32(1), failing.closes.Load()) + }) + } +} + +func TestBuildCanceled(t *testing.T) { + building := &fakeCollector{name: "building", buildBlock: make(chan struct{}), buildStarted: make(chan struct{}, 1)} + collection := collector.New(collector.Map{"building": building}) + + ctx, cancel := context.WithCancel(t.Context()) + + errCh := make(chan error, 1) + + go func() { + errCh <- collection.Build(ctx, slog.New(slog.DiscardHandler)) + }() + + <-building.buildStarted + cancel() + + select { + case err := <-errCh: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(10 * time.Second): + t.Fatal("Build didn't return after its context was canceled") + } + + // Close waits for the abandoned Build to return before it closes the collector. + closeErr := make(chan error, 1) + + go func() { + closeErr <- collection.Close() + }() + + close(building.buildBlock) + require.NoError(t, <-closeErr) + require.Equal(t, int32(1), building.closes.Load()) + + require.ErrorIs(t, collection.Build(canceledContext(), slog.New(slog.DiscardHandler)), context.Canceled) +} + +func TestCollectionClose(t *testing.T) { + t.Run("without Build", func(t *testing.T) { + test := &fakeCollector{name: "test"} + collection := collector.New(collector.Map{"test": test}) + + require.NoError(t, collection.Close()) + require.NoError(t, collection.Close()) + require.Equal(t, int32(0), test.closes.Load()) + }) + + t.Run("idempotent", func(t *testing.T) { + test := &fakeCollector{name: "test"} + collection := collector.New(collector.Map{"test": test}) + require.NoError(t, collection.Build(t.Context(), slog.New(slog.DiscardHandler))) + + require.NoError(t, collection.Close()) + require.NoError(t, collection.Close()) + require.Equal(t, int32(1), test.closes.Load()) + + // A closed collector is not collected anymore. + handler, err := collection.NewHandler(time.Minute, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + require.Contains(t, scrape(t, handler), `windows_exporter_collector_success{collector="test"} 0`) + require.Equal(t, int32(0), test.calls.Load()) + }) + + t.Run("waits for a running Collect", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + slow := &fakeCollector{name: "slow", delay: 2 * time.Second} + collection := collector.New(collector.Map{"slow": slow}) + require.NoError(t, collection.Build(t.Context(), slog.New(slog.DiscardHandler))) + + handler, err := collection.NewHandler(100*time.Millisecond, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + require.Contains(t, scrape(t, handler), `windows_exporter_collector_timeout{collector="slow"} 1`) + + require.NoError(t, collection.Close()) + require.Equal(t, int32(1), slow.closes.Load()) + require.Equal(t, int32(0), slow.closedWhileActive.Load()) + }) + }) + + t.Run("leaves a hung collector open", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + hung := &fakeCollector{name: "hung", delay: time.Minute} + collection := collector.New(collector.Map{"hung": hung}) + require.NoError(t, collection.Build(t.Context(), slog.New(slog.DiscardHandler))) + + handler, err := collection.NewHandler(100*time.Millisecond, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + scrape(t, handler) + + require.ErrorContains(t, collection.Close(), "still running") + require.Equal(t, int32(0), hung.closes.Load()) + + // Once the call returns, a later Close closes the collector. + time.Sleep(time.Minute) + synctest.Wait() + + require.NoError(t, collection.Close()) + require.Equal(t, int32(1), hung.closes.Load()) + require.Equal(t, int32(0), hung.closedWhileActive.Load()) + }) + }) +} + +func TestBuildCloseCyclesDoNotLeakGoroutines(t *testing.T) { + // The bubble also fails the test if any goroutine of it is still blocked at the end. + synctest.Test(t, func(t *testing.T) { + logger := slog.New(slog.DiscardHandler) + + cycle := func() { + slow := &fakeCollector{name: "slow", delay: time.Second} + fast := &fakeCollector{name: "fast"} + collection := collector.New(collector.Map{"slow": slow, "fast": fast}) + require.NoError(t, collection.Build(t.Context(), logger)) + + handler, err := collection.NewHandler(100*time.Millisecond, logger, nil) + require.NoError(t, err) + + out := scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_timeout{collector="slow"} 1`) + require.Contains(t, out, `windows_exporter_collector_success{collector="fast"} 1`) + + require.NoError(t, collection.Close()) + require.Equal(t, int32(1), slow.closes.Load()) + require.Equal(t, int32(1), fast.closes.Load()) + } + + cycle() + synctest.Wait() + + baseline := runtime.NumGoroutine() + + for range 20 { + cycle() + } + + synctest.Wait() + + // One leaked goroutine per cycle would add 20. + require.LessOrEqual(t, runtime.NumGoroutine(), baseline+2) + }) +} + +// fakeCollector is a collector whose Build and Collect behave as configured. +type fakeCollector struct { + name string + buildErr error + buildPanic bool + buildStarted chan struct{} + buildBlock chan struct{} + delay time.Duration + + calls atomic.Int32 + active atomic.Int32 + maxActive atomic.Int32 + closes atomic.Int32 + closedWhileActive atomic.Int32 + budget atomic.Int64 +} + +func (c *fakeCollector) GetName() string { return c.name } + +func (c *fakeCollector) Build(_ *slog.Logger, _ *mi.Session) error { + if c.buildStarted != nil { + c.buildStarted <- struct{}{} + } + + if c.buildBlock != nil { + <-c.buildBlock + } + + if c.buildPanic { + panic("build panic") + } + + return c.buildErr +} + +func (c *fakeCollector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error { + c.calls.Add(1) + c.budget.Store(int64(maxScrapeDuration)) + + active := c.active.Add(1) + defer c.active.Add(-1) + + for { + maxActive := c.maxActive.Load() + if active <= maxActive || c.maxActive.CompareAndSwap(maxActive, active) { + break + } + } + + time.Sleep(c.delay) + + ch <- prometheus.MustNewConstMetric(prometheus.NewDesc("windows_test_"+c.name, "Test metric", nil, nil), prometheus.GaugeValue, 42) + + return nil +} + +func (c *fakeCollector) Close() error { + c.closes.Add(1) + + if c.active.Load() != 0 { + c.closedWhileActive.Add(1) + } + + return nil +} + +// blockingCollector is a collector whose Collect signals started and waits until block is closed. +type blockingCollector struct { + *fakeCollector + + started chan struct{} + block chan struct{} +} + +func newBlockingCollector() *blockingCollector { + c := &blockingCollector{ + started: make(chan struct{}, 1), + block: make(chan struct{}), + } + c.fakeCollector = &fakeCollector{name: "test"} + + return c +} + +func (c *blockingCollector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error { + c.started <- struct{}{} + + <-c.block + + return c.fakeCollector.Collect(ch, maxScrapeDuration) +} + +// scrapeInBackground starts a scrape with a handler bound to ctx and returns its output, or the error text. +func scrapeInBackground(ctx context.Context, collection *collector.Collection, timeout time.Duration) <-chan string { + out := make(chan string, 1) + + handler, err := collection.NewHandlerWithContext(ctx, timeout, slog.New(slog.DiscardHandler), nil) + if err != nil { + out <- err.Error() + + return out + } + + go func() { + text, err := gather(handler) + if err != nil { + text = err.Error() + } + + out <- text + }() + + return out +} + +func scrape(t *testing.T, handler *collector.Handler) string { + t.Helper() + + out, err := gather(handler) + require.NoError(t, err) + + return out +} + +// gather collects handler once and returns the metrics in the text format. +func gather(handler *collector.Handler) (string, error) { + registry := prometheus.NewRegistry() + if err := registry.Register(handler); err != nil { + return "", err + } + + families, err := registry.Gather() + if err != nil { + return "", err + } + + var out strings.Builder + + for _, family := range families { + if _, err := expfmt.MetricFamilyToText(&out, family); err != nil { + return "", err + } + } + + return out.String(), nil +} + +func canceledContext() context.Context { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + return ctx +} + +// recordHandler is a [slog.Handler] that records all messages. +type recordHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (h *recordHandler) Enabled(context.Context, slog.Level) bool { return true } + +func (h *recordHandler) Handle(_ context.Context, record slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + + h.records = append(h.records, record) + + return nil +} + +func (h *recordHandler) WithAttrs([]slog.Attr) slog.Handler { return h } + +func (h *recordHandler) WithGroup(string) slog.Handler { return h } + +// warnings returns the number of warnings whose message contains substr. +func (h *recordHandler) warnings(substr string) int { + h.mu.Lock() + defer h.mu.Unlock() + + n := 0 + + for _, record := range h.records { + if record.Level == slog.LevelWarn && strings.Contains(record.Message, substr) { + n++ + } + } + + return n +} + +func (h *recordHandler) reset() { + h.mu.Lock() + defer h.mu.Unlock() + + h.records = nil +} diff --git a/pkg/collector/collection.go b/pkg/collector/collection.go index d2f2d553..270961e5 100644 --- a/pkg/collector/collection.go +++ b/pkg/collector/collection.go @@ -24,7 +24,6 @@ import ( "log/slog" "maps" "slices" - "sync" gotime "time" "github.com/alecthomas/kingpin/v2" @@ -158,8 +157,8 @@ func NewWithConfig(config Config) *Collection { // New To be called by the external libraries for collector initialization. func New(collectors Map) *Collection { return &Collection{ - collectors: collectors, - concurrencyCh: make(chan struct{}, 1), + collectors: collectors, + state: newCollectionState(), scrapeDurationDesc: prometheus.NewDesc( prometheus.BuildFQName(types.Namespace, "exporter", "scrape_duration_seconds"), "windows_exporter: Total scrape duration.", @@ -213,39 +212,68 @@ func (c *Collection) Disable(disabledCollectors []string) { } } +// closeTimeout bounds how long Close waits for running Build and Collect calls to return. +const closeTimeout = 5 * gotime.Second + // Build To be called by the exporter for collector initialization. // Instead, fail fast, it will try to build all collectors and return all errors. // errors are joined with errors.Join. +// +// A collector whose Build fails is not collected. Scrapes report it with +// windows_exporter_collector_success 0. A panic in a collector's Build is returned as an error. +// +// If ctx is done before all collectors are built, Build returns without waiting for them. +// Collectors that are still building finish in the background, and Close waits for them. func (c *Collection) Build(ctx context.Context, logger *slog.Logger) error { c.startTime = gotime.Now() - err := c.initMI() + if err := ctx.Err(); err != nil { + return fmt.Errorf("collector initialization aborted: %w", context.Cause(ctx)) + } + + miSession, err := c.initMI() if err != nil { return fmt.Errorf("error from initialize MI: %w", err) } - wg := sync.WaitGroup{} - wg.Add(len(c.collectors)) + type buildResult struct { + name string + err error + } - errCh := make(chan error, len(c.collectors)) + // The channel is buffered, so that builds abandoned by a done ctx don't block. + resultCh := make(chan buildResult, len(c.collectors)) + + for name, collector := range c.collectors { + state := c.state.collector(name) - for _, collector := range c.collectors { go func() { - defer wg.Done() - - if err := collector.Build(logger, c.miSession); err != nil { - errCh <- fmt.Errorf("error build collector %s: %w", collector.GetName(), err) + resultCh <- buildResult{ + name: collector.GetName(), + err: state.build(ctx, logger, miSession, collector), } }() } - wg.Wait() - - close(errCh) - errs := make([]error, 0, len(c.collectors)) - for err := range errCh { + for range len(c.collectors) { + var result buildResult + + select { + case result = <-resultCh: + case <-ctx.Done(): + errs = append(errs, fmt.Errorf("collector initialization aborted: %w", context.Cause(ctx))) + + return errors.Join(errs...) + } + + if result.err == nil { + continue + } + + err := fmt.Errorf("error build collector %s: %w", result.name, result.err) + if errors.Is(err, pdh.ErrNoData) || errors.Is(err, winregistry.ErrNotExist) || errors.Is(err, pdh.NewPdhError(pdh.CstatusNoObject)) || @@ -253,7 +281,9 @@ func (c *Collection) Build(ctx context.Context, logger *slog.Logger) error { errors.Is(err, mi.MI_RESULT_INVALID_OPERATION_TIMEOUT) || errors.Is(err, mi.MI_RESULT_INVALID_NAMESPACE) || errors.Is(err, mi.MI_RESULT_INVALID_CLASS) { - logger.LogAttrs(ctx, slog.LevelWarn, "couldn't initialize collector", slog.Any("err", err)) + logger.LogAttrs(ctx, slog.LevelWarn, "couldn't initialize collector", + slog.Any("err", err), + ) continue } @@ -265,65 +295,98 @@ func (c *Collection) Build(ctx context.Context, logger *slog.Logger) error { } // Close To be called by the exporter for collector cleanup. +// +// Close waits up to 5 seconds for running Build and Collect calls to return. +// A collector that is still running after that is left open and reported in the returned error, +// because closing it would release resources that are still in use. +// The MI session is then left open as well. +// +// Close is safe to call after a failed or partial Build and more than once. +// Collectors that were never built are not closed. A closed collector is never built or collected again. func (c *Collection) Close() error { + c.state.closeMu.Lock() + defer c.state.closeMu.Unlock() + + ctx, cancel := context.WithTimeout(context.Background(), closeTimeout) + defer cancel() + errs := make([]error, 0, len(c.collectors)) + collectorBusy := false + + for name, collector := range c.collectors { + if err := c.state.collector(name).close(ctx, collector); err != nil { + collectorBusy = collectorBusy || errors.Is(err, errCollectorBusy) - for _, collector := range c.collectors { - if err := collector.Close(); err != nil { errs = append(errs, fmt.Errorf("error from close collector %s: %w", collector.GetName(), err)) } } - app, err := c.miSession.GetApplication() - if err != nil && !errors.Is(err, mi.ErrNotInitialized) { - errs = append(errs, fmt.Errorf("error from get MI application: %w", err)) - } - - if err := c.miSession.Close(); err != nil && !errors.Is(err, mi.ErrNotInitialized) { - errs = append(errs, fmt.Errorf("error from close MI session: %w", err)) - } - - if err := app.Close(); err != nil && !errors.Is(err, mi.ErrNotInitialized) { - errs = append(errs, fmt.Errorf("error from close MI application: %w", err)) + if collectorBusy { + // Closing the MI session would block on, or break, the operations of running collectors. + errs = append(errs, errors.New("MI session left open, because collectors are still running")) + } else { + errs = append(errs, c.state.closeMI()...) } return errors.Join(errs...) } // initMI To be called by the exporter for collector initialization. -func (c *Collection) initMI() error { +// It returns the session of an earlier call, if there is one. +func (c *Collection) initMI() (*mi.Session, error) { + if session := c.state.session(); session != nil { + return session, nil + } + app, err := mi.ApplicationInitialize() if err != nil { - return fmt.Errorf("error from initialize MI application: %w", err) + return nil, fmt.Errorf("error from initialize MI application: %w", err) } destinationOptions, err := app.NewDestinationOptions() if err != nil { - return fmt.Errorf("error from create NewDestinationOptions: %w", err) + _ = app.Close() + + return nil, fmt.Errorf("error from create NewDestinationOptions: %w", err) } - if err = destinationOptions.SetLocale(mi.LocaleEnglish); err != nil { - return fmt.Errorf("error from set locale: %w", err) - } - - if err = destinationOptions.SetTimeout(gotime.Second); err != nil { - return fmt.Errorf("error from set timeout: %w", err) - } - - c.miSession, err = app.NewSession(destinationOptions) + session, err := newMISession(app, destinationOptions) if err != nil { - return fmt.Errorf("error from create NewSession: %w", err) + _ = destinationOptions.Delete() + _ = app.Close() + + return nil, err } - return nil + c.state.setMI(app, session) + + return session, nil +} + +func newMISession(app *mi.Application, destinationOptions *mi.DestinationOptions) (*mi.Session, error) { + if err := destinationOptions.SetLocale(mi.LocaleEnglish); err != nil { + return nil, fmt.Errorf("error from set locale: %w", err) + } + + if err := destinationOptions.SetTimeout(gotime.Second); err != nil { + return nil, fmt.Errorf("error from set timeout: %w", err) + } + + session, err := app.NewSession(destinationOptions) + if err != nil { + return nil, fmt.Errorf("error from create NewSession: %w", err) + } + + return session, nil } // WithCollectors To be called by the exporter for collector initialization. +// The returned Collection shares the collector instances and their state with c, +// so a collector never runs twice at the same time, whichever Collection scrapes it. func (c *Collection) WithCollectors(collectors []string) (*Collection, error) { metricCollectors := &Collection{ - miSession: c.miSession, startTime: c.startTime, - concurrencyCh: c.concurrencyCh, + state: c.state, scrapeDurationDesc: c.scrapeDurationDesc, collectorScrapeDurationDesc: c.collectorScrapeDurationDesc, collectorScrapeSuccessDesc: c.collectorScrapeSuccessDesc, diff --git a/pkg/collector/handler.go b/pkg/collector/handler.go index efad9936..4abcc37f 100644 --- a/pkg/collector/handler.go +++ b/pkg/collector/handler.go @@ -18,9 +18,9 @@ package collector import ( + "context" "fmt" "log/slog" - "sync" "time" "github.com/prometheus/client_golang/prometheus" @@ -29,20 +29,27 @@ import ( // Interface guard. var _ prometheus.Collector = (*Handler)(nil) +// scrapeSlot admits one scrape at a time, across all collections. // We are expose metrics directly from the memory region of the Win32 API. // We should not allow more than one request at a time. // //nolint:gochecknoglobals -var concurrencyMu sync.Mutex +var scrapeSlot = make(chan struct{}, 1) // Handler implements [prometheus.Collector] for a set of Windows Collection. type Handler struct { maxScrapeDuration time.Duration logger *slog.Logger collection *Collection + + // ctx and deadline are set by [Collection.NewHandlerWithContext] for a handler that serves one scrape. + // [prometheus.Collector.Collect] has no context parameter, so the handler carries it. + ctx context.Context //nolint:containedctx + deadline time.Time } // NewHandler returns a new Handler that implements a [prometheus.Collector] for the given metrics Collection. +// Each call of Collect may take up to maxScrapeDuration, including the time it waits for an earlier scrape to finish. func (c *Collection) NewHandler(maxScrapeDuration time.Duration, logger *slog.Logger, collectors []string) (*Handler, error) { collection := c @@ -62,12 +69,60 @@ func (c *Collection) NewHandler(maxScrapeDuration time.Duration, logger *slog.Lo }, nil } +// NewHandlerWithContext returns a Handler for a single scrape, such as one HTTP request. +// The scrape's deadline is maxScrapeDuration from now, so the time Collect waits for an earlier scrape +// to finish counts against it. Collect returns early once ctx is done, without starting collectors +// that have not started yet. +func (c *Collection) NewHandlerWithContext(ctx context.Context, maxScrapeDuration time.Duration, logger *slog.Logger, collectors []string) (*Handler, error) { + handler, err := c.NewHandler(maxScrapeDuration, logger, collectors) + if err != nil { + return nil, err + } + + handler.ctx = ctx + handler.deadline = time.Now().Add(maxScrapeDuration) + + return handler, nil +} + func (p *Handler) Describe(_ chan<- *prometheus.Desc) {} // Collect sends the collected metrics from each of the Collection to // prometheus. func (p *Handler) Collect(ch chan<- prometheus.Metric) { - concurrencyMu.Lock() - p.collection.collectAll(ch, p.logger, p.maxScrapeDuration) - concurrencyMu.Unlock() + ctx, deadline := p.ctx, p.deadline + if ctx == nil { + ctx, deadline = context.Background(), time.Now().Add(p.maxScrapeDuration) + } + + ctx, cancel := context.WithDeadline(ctx, deadline) + defer cancel() + + p.collection.collectAll(ctx, ch, p.logger) +} + +// acquireScrapeSlot waits for the scrape slot until ctx is done. +func acquireScrapeSlot(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + + select { + case scrapeSlot <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + + // Both cases may have been ready. Don't start a scrape that nobody waits for anymore. + if err := ctx.Err(); err != nil { + <-scrapeSlot + + return err + } + + return nil +} + +func releaseScrapeSlot() { + <-scrapeSlot } diff --git a/pkg/collector/state.go b/pkg/collector/state.go new file mode 100644 index 00000000..6aaafd53 --- /dev/null +++ b/pkg/collector/state.go @@ -0,0 +1,242 @@ +// 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 collector + +import ( + "context" + "errors" + "fmt" + "log/slog" + "runtime/debug" + "sync" + + "github.com/prometheus-community/windows_exporter/internal/mi" +) + +var ( + errCollectorBusy = errors.New("collector is still running") + errCollectionClosed = errors.New("collection is closed") +) + +// collectionState is the state of a Collection that its filtered views share. +type collectionState struct { + mu sync.Mutex + collectors map[string]*collectorState + miApp *mi.Application + miSession *mi.Session + + // closeMu serializes Close calls. + closeMu sync.Mutex +} + +func newCollectionState() *collectionState { + return &collectionState{collectors: make(map[string]*collectorState)} +} + +// collector returns the state of the named collector, creating it on first use. +func (s *collectionState) collector(name string) *collectorState { + s.mu.Lock() + defer s.mu.Unlock() + + state, ok := s.collectors[name] + if !ok { + state = &collectorState{slot: make(chan struct{}, 1)} + s.collectors[name] = state + } + + return state +} + +func (s *collectionState) session() *mi.Session { + s.mu.Lock() + defer s.mu.Unlock() + + return s.miSession +} + +func (s *collectionState) setMI(app *mi.Application, session *mi.Session) { + s.mu.Lock() + defer s.mu.Unlock() + + s.miApp = app + s.miSession = session +} + +// closeMI closes the MI session and application. Later calls do nothing. +func (s *collectionState) closeMI() []error { + s.mu.Lock() + app, session := s.miApp, s.miSession + s.miApp, s.miSession = nil, nil + s.mu.Unlock() + + var errs []error + + if session != nil { + if err := session.Close(); err != nil && !errors.Is(err, mi.ErrNotInitialized) { + errs = append(errs, fmt.Errorf("error from close MI session: %w", err)) + } + } + + if app != nil { + if err := app.Close(); err != nil && !errors.Is(err, mi.ErrNotInitialized) { + errs = append(errs, fmt.Errorf("error from close MI application: %w", err)) + } + } + + return errs +} + +type startResult int + +const ( + started startResult = iota + busy + notAvailable +) + +// collectorState tracks the lifecycle of one collector instance. +type collectorState struct { + // slot is held while Build, Collect or Close of the collector runs. + // A Collect call that outlives its scrape keeps the slot until it returns, + // so no scrape starts a second call on the same instance. + slot chan struct{} + + mu sync.Mutex + // buildErr is the error of the last Build. A collector that failed to build is not collected. + buildErr error + // built is true once Build was called, so Close only closes collectors that may hold resources. + built bool + // closing is true once Close was called. No Build or Collect starts afterwards. + closing bool + // closed is true once Collector.Close was called. + closed bool + // busyLogged is true once a scrape logged that it skipped the current run. + busyLogged bool +} + +// tryCollect acquires the slot for a Collect call without waiting. +func (s *collectorState) tryCollect() startResult { + s.mu.Lock() + defer s.mu.Unlock() + + if s.closing || s.buildErr != nil { + return notAvailable + } + + select { + case s.slot <- struct{}{}: + return started + default: + return busy + } +} + +// release releases the slot acquired by tryCollect. +func (s *collectorState) release() { + s.mu.Lock() + s.busyLogged = false + s.mu.Unlock() + + <-s.slot +} + +// logBusy reports whether a skipped scrape is the first one of the current run. +func (s *collectorState) logBusy() bool { + s.mu.Lock() + defer s.mu.Unlock() + + first := !s.busyLogged + s.busyLogged = true + + return first +} + +// build runs Collector.Build while it holds the slot and records the result. +// A panic in Collector.Build is returned as an error. +func (s *collectorState) build(ctx context.Context, logger *slog.Logger, miSession *mi.Session, collector Collector) (err error) { + select { + case s.slot <- struct{}{}: + case <-ctx.Done(): + err = context.Cause(ctx) + + s.mu.Lock() + s.buildErr = err + s.mu.Unlock() + + return err + } + + defer func() { + s.mu.Lock() + s.buildErr = err + s.mu.Unlock() + + <-s.slot + }() + + s.mu.Lock() + closing := s.closing + s.built = !closing + s.mu.Unlock() + + if closing { + return errCollectionClosed + } + + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("panic: %v. stack: %s", r, string(debug.Stack())) + } + }() + + return collector.Build(logger, miSession) +} + +// close calls Collector.Close once the collector is idle. It waits until ctx is done for a running +// Build or Collect call to return. If the call does not return in time, the collector is left open, +// because closing it would release resources that are still in use. +// After close, the collector is never built or collected again. +func (s *collectorState) close(ctx context.Context, collector Collector) (err error) { + s.mu.Lock() + s.closing = true + done := s.closed || !s.built + s.mu.Unlock() + + if done { + return nil + } + + select { + case s.slot <- struct{}{}: + // The slot is never released, so no call starts after Close. + case <-ctx.Done(): + return errCollectorBusy + } + + s.mu.Lock() + s.closed = true + s.mu.Unlock() + + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("panic: %v. stack: %s", r, string(debug.Stack())) + } + }() + + return collector.Close() +} diff --git a/pkg/collector/types.go b/pkg/collector/types.go index e191ea7c..a22e5d26 100644 --- a/pkg/collector/types.go +++ b/pkg/collector/types.go @@ -29,10 +29,11 @@ import ( const DefaultCollectors = "cpu,memory,logical_disk,physical_disk,net,os,service,system" type Collection struct { - collectors Map - miSession *mi.Session - startTime time.Time - concurrencyCh chan struct{} + collectors Map + startTime time.Time + + // state is shared with every Collection derived from this one by WithCollectors. + state *collectionState scrapeDurationDesc *prometheus.Desc collectorScrapeDurationDesc *prometheus.Desc