collector: serialize collector runs and harden the scrape lifecycle (#2541)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>
2026-10-18 18:49:05 +02:00 · 2026-10-08 14:25:17 +02:00
parent 16aa107c9f
commit 96a998772b
11 changed files with 1554 additions and 213 deletions
@@ -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)
@@ -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)

@@ -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),
+	)
+}