From af2a7c3f6c65e9c71e46f36827495128233af79f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Fri, 16 Oct 2026 02:31:55 +0200 Subject: [PATCH] log: open the event log lazily, bound close, keep the service running, structured messages (#2734) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Jan-Otto Kröpke --- cmd/windows_exporter/0_service.go | 77 ++++++++++- cmd/windows_exporter/main.go | 20 ++- cmd/windows_exporter/service_test.go | 58 +++++++++ internal/collector/hyperv/hyperv.go | 9 +- internal/collector/iis/iis.go | 4 +- internal/collector/mssql/mssql.go | 15 ++- internal/collector/mssql/mssql_instance.go | 3 +- .../performancecounter/performancecounter.go | 9 +- internal/collector/textfile/textfile.go | 4 +- internal/collector/update/update.go | 4 +- internal/httphandler/httphandler.go | 5 +- internal/log/eventlog/eventlog.go | 42 +++++- .../log/eventlog/eventlog_internal_test.go | 123 ++++++++++++++++++ internal/log/logger.go | 33 +++-- internal/log/logger_test.go | 36 +++++ internal/log/rotate.go | 106 +++++++++++++-- internal/log/rotate_test.go | 73 +++++++++++ pkg/collector/collect.go | 68 +++++----- pkg/collector/collect_test.go | 2 +- 19 files changed, 610 insertions(+), 81 deletions(-) create mode 100644 internal/log/eventlog/eventlog_internal_test.go diff --git a/cmd/windows_exporter/0_service.go b/cmd/windows_exporter/0_service.go index cd931ba9..1ce6ed79 100644 --- a/cmd/windows_exporter/0_service.go +++ b/cmd/windows_exporter/0_service.go @@ -21,10 +21,13 @@ import ( "context" "errors" "fmt" + "io" "log/slog" "os" + "path/filepath" "strings" "sync" + "time" "unsafe" "golang.org/x/sys/windows" @@ -96,10 +99,11 @@ var IsService = func() bool { }() }() + // This event is informational. If it cannot be written, the exporter still + // starts: stopping the service here would report it as stopped while run + // keeps serving. if err := logToEventToLog(windows.EVENTLOG_INFORMATION_TYPE, "attempting to start exporter service"); err != nil { logToFile(fmt.Sprintf("failed sent log to event log: %v", err)) - - exitCodeCh <- 2 } return true @@ -222,11 +226,74 @@ func logStartupError(ctx context.Context, msg string, err error) { ) } +const ( + // serviceErrorLogName is the file in the temporary directory that logToFile appends to. + serviceErrorLogName = "windows_exporter.service.error.log" + // serviceErrorLogMaxSize is the size above which logToFile starts the file over. + serviceErrorLogMaxSize = 1 << 20 +) + +// logToFile is the last resort when the event log cannot be written. It appends +// msg to one file in the temporary directory, so that repeated failures don't +// leave a new file behind each time. For a service running as LocalSystem, Go +// returns C:\Windows\SystemTemp, which only administrators can write to. func logToFile(msg string) { - if file, err := os.CreateTemp("", "windows_exporter.service.error.log"); err == nil { - _, _ = file.WriteString(msg) - _ = file.Close() + _ = appendToFile(filepath.Join(os.TempDir(), serviceErrorLogName), msg, time.Now(), serviceErrorLogMaxSize) +} + +// appendToFile appends a timestamped line to path. It starts the file over once +// it is larger than maxSize. It refuses to write through a symbolic link, +// junction or other reparse point, or to a file with more than one hard link, +// so that a file planted at path cannot redirect the write to another file. +func appendToFile(path, msg string, now time.Time, maxSize int64) error { + pathPtr, err := windows.UTF16PtrFromString(path) + if err != nil { + return err } + + handle, err := windows.CreateFile( + pathPtr, + windows.GENERIC_READ|windows.GENERIC_WRITE, + windows.FILE_SHARE_READ, + nil, + windows.OPEN_ALWAYS, + windows.FILE_ATTRIBUTE_NORMAL|windows.FILE_FLAG_OPEN_REPARSE_POINT, + 0, + ) + if err != nil { + return fmt.Errorf("open %s: %w", path, err) + } + + file := os.NewFile(uintptr(handle), path) + + defer func() { + _ = file.Close() + }() + + var info windows.ByHandleFileInformation + if err = windows.GetFileInformationByHandle(handle, &info); err != nil { + return fmt.Errorf("stat %s: %w", path, err) + } + + if info.FileAttributes&windows.FILE_ATTRIBUTE_REPARSE_POINT != 0 || info.NumberOfLinks > 1 { + return fmt.Errorf("refusing to write to %s: it is a reparse point or has more than one link", path) + } + + if size := int64(info.FileSizeHigh)<<32 | int64(info.FileSizeLow); size > maxSize { + if err = file.Truncate(0); err != nil { + return fmt.Errorf("truncate %s: %w", path, err) + } + } + + if _, err = file.Seek(0, io.SeekEnd); err != nil { + return fmt.Errorf("seek %s: %w", path, err) + } + + if _, err = file.WriteString(now.Format(time.RFC3339) + " " + strings.TrimSpace(msg) + "\n"); err != nil { + return fmt.Errorf("write %s: %w", path, err) + } + + return nil } // isWindowsService is a clone of "golang.org/x/sys/windows/svc:IsWindowsService", but with a fix diff --git a/cmd/windows_exporter/main.go b/cmd/windows_exporter/main.go index 8ade988d..8a4ac637 100644 --- a/cmd/windows_exporter/main.go +++ b/cmd/windows_exporter/main.go @@ -200,7 +200,9 @@ func run(ctx context.Context, args []string) int { logger.LogAttrs(ctx, slog.LevelDebug, "logging has Started") if configFile != nil && *configFile != "" { - logger.LogAttrs(ctx, slog.LevelInfo, "using configuration file: "+*configFile) + logger.LogAttrs(ctx, slog.LevelInfo, "using configuration file", + slog.String("file", *configFile), + ) } // Log the build and the account before the collectors are built, @@ -261,7 +263,9 @@ func run(ctx context.Context, args []string) int { return 1 } - logger.InfoContext(ctx, "Enabled collectors: "+strings.Join(effectiveCollectors(enabledCollectorList, disabledCollectorList), ", ")) + logger.LogAttrs(ctx, slog.LevelInfo, "enabled collectors", + slog.String("collectors", strings.Join(effectiveCollectors(enabledCollectorList, disabledCollectorList), ",")), + ) handlerOpts := &httphandler.Options{ DisableExporterMetrics: *disableExporterMetrics, @@ -284,7 +288,9 @@ func run(ctx context.Context, args []string) int { mux.HandleFunc("GET /debug/pprof/trace", pprof.Trace) } - logger.LogAttrs(ctx, slog.LevelInfo, fmt.Sprintf("started windows_exporter in %s", time.Since(startTime))) + logger.LogAttrs(ctx, slog.LevelInfo, "started windows_exporter", + slog.Duration("duration", time.Since(startTime)), + ) // Requests derive their context from serveCtx, so that canceling it cancels in-flight scrapes. serveCtx, cancelServe := context.WithCancel(ctx) @@ -384,7 +390,9 @@ func logCurrentUser(ctx context.Context, logger *slog.Logger) { return } - logger.LogAttrs(ctx, slog.LevelInfo, "Running as "+u.Username) + logger.LogAttrs(ctx, slog.LevelInfo, "running as user", + slog.String("user", u.Username), + ) if strings.Contains(u.Username, "ContainerAdministrator") || strings.Contains(u.Username, "ContainerUser") { logger.LogAttrs(ctx, slog.LevelWarn, "Running as a preconfigured Windows Container user. This may mean you do not have Windows HostProcess containers configured correctly and some functionality will not work as expected.") @@ -460,7 +468,9 @@ func closeCollection(ctx context.Context, logger *slog.Logger, collection *colle ) } case <-timer.C: - logger.LogAttrs(ctx, slog.LevelWarn, fmt.Sprintf("collectors didn't close within %s", collectionCloseTimeout)) + logger.LogAttrs(ctx, slog.LevelWarn, "collectors didn't close in time", + slog.Duration("timeout", collectionCloseTimeout), + ) } } diff --git a/cmd/windows_exporter/service_test.go b/cmd/windows_exporter/service_test.go index 4a6122df..c64e0c30 100644 --- a/cmd/windows_exporter/service_test.go +++ b/cmd/windows_exporter/service_test.go @@ -19,6 +19,8 @@ package main import ( "context" + "os" + "path/filepath" "testing" "time" @@ -26,6 +28,62 @@ import ( "golang.org/x/sys/windows/svc" ) +func TestAppendToFile(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + path := filepath.Join(dir, serviceErrorLogName) + now := time.Date(2026, 10, 15, 14, 3, 5, 0, time.UTC) + + require.NoError(t, appendToFile(path, "first failure", now, serviceErrorLogMaxSize)) + require.NoError(t, appendToFile(path, "second failure\n", now.Add(time.Second), serviceErrorLogMaxSize)) + + entries, err := os.ReadDir(dir) + require.NoError(t, err) + require.Len(t, entries, 1, "repeated failures must not create a new file each time") + + data, err := os.ReadFile(path) + require.NoError(t, err) + require.Equal(t, "2026-10-15T14:03:05Z first failure\n2026-10-15T14:03:06Z second failure\n", string(data)) +} + +func TestAppendToFileStartsOverAboveMaxSize(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), serviceErrorLogName) + now := time.Date(2026, 10, 15, 14, 3, 5, 0, time.UTC) + + require.NoError(t, appendToFile(path, "first failure", now, 10)) + require.NoError(t, appendToFile(path, "second failure", now, 10)) + + data, err := os.ReadFile(path) + require.NoError(t, err) + require.Equal(t, "2026-10-15T14:03:05Z second failure\n", string(data)) +} + +func TestAppendToFileRefusesLinks(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + target := filepath.Join(dir, "target.txt") + require.NoError(t, os.WriteFile(target, []byte("keep"), 0o600)) + + hardLink := filepath.Join(dir, "hardlink.log") + require.NoError(t, os.Link(target, hardLink)) + require.ErrorContains(t, appendToFile(hardLink, "failure", time.Now(), serviceErrorLogMaxSize), "refusing to write") + + symlink := filepath.Join(dir, "symlink.log") + if err := os.Symlink(target, symlink); err == nil { + require.ErrorContains(t, appendToFile(symlink, "failure", time.Now(), serviceErrorLogMaxSize), "refusing to write") + } else { + t.Logf("skipping the symbolic link case: %v", err) + } + + data, err := os.ReadFile(target) + require.NoError(t, err) + require.Equal(t, "keep", string(data)) +} + func TestServiceExecuteStop(t *testing.T) { t.Parallel() diff --git a/internal/collector/hyperv/hyperv.go b/internal/collector/hyperv/hyperv.go index 3b068cdc..9e1e6435 100644 --- a/internal/collector/hyperv/hyperv.go +++ b/internal/collector/hyperv/hyperv.go @@ -338,10 +338,11 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { for _, name := range collectorsEnabled { if buildNumber < subCollectors[name].minBuildNumber { - c.logger.Warn(fmt.Sprintf( - "collector %s requires windows build version %d. Current build version: %d", - name, subCollectors[name].minBuildNumber, buildNumber, - )) + c.logger.Warn("sub-collector requires a newer Windows build", + slog.String("sub_collector", name), + slog.Int("required_build", int(subCollectors[name].minBuildNumber)), + slog.Int("build", int(buildNumber)), + ) continue } diff --git a/internal/collector/iis/iis.go b/internal/collector/iis/iis.go index 04f6c0c0..69285206 100644 --- a/internal/collector/iis/iis.go +++ b/internal/collector/iis/iis.go @@ -236,7 +236,9 @@ func (c *Collector) getIISVersion() simpleVersion { return simpleVersion{} } - c.logger.Debug(fmt.Sprintf("Detected IIS %d.%d\n", major, minor)) + c.logger.Debug("detected IIS version", + slog.String("version", fmt.Sprintf("%d.%d", major, minor)), + ) return simpleVersion{ major: major, diff --git a/internal/collector/mssql/mssql.go b/internal/collector/mssql/mssql.go index 44dc6e93..ee5600d0 100644 --- a/internal/collector/mssql/mssql.go +++ b/internal/collector/mssql/mssql.go @@ -337,7 +337,9 @@ func (c *Collector) getMSSQLInstances() ([]mssqlInstance, error) { sqlInstances = append(sqlInstances, instance) } - c.logger.Debug(fmt.Sprintf("detected MSSQL Instances: %#v\n", sqlInstances)) + c.logger.Debug("detected MSSQL instances", + slog.Any("instances", sqlInstances), + ) return sqlInstances, nil } @@ -382,11 +384,18 @@ func (c *Collector) collect[T any]( errs = append(errs, err) success = 0.0 - c.logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf("mssql class collector %s for instance %s failed after %s", collector, sqlInstance.name, duration), + c.logger.LogAttrs(ctx, slog.LevelDebug, "mssql class collector failed", + slog.String("class", collector), + slog.String("instance", sqlInstance.name), + slog.Duration("duration", duration), slog.Any("err", err), ) } else { - c.logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf("mssql class collector %s for instance %s succeeded after %s", collector, sqlInstance.name, duration)) + c.logger.LogAttrs(ctx, slog.LevelDebug, "mssql class collector succeeded", + slog.String("class", collector), + slog.String("instance", sqlInstance.name), + slog.Duration("duration", duration), + ) } if collector == "" { diff --git a/internal/collector/mssql/mssql_instance.go b/internal/collector/mssql/mssql_instance.go index f524d921..3dd410e6 100644 --- a/internal/collector/mssql/mssql_instance.go +++ b/internal/collector/mssql/mssql_instance.go @@ -47,7 +47,8 @@ func (c *Collector) collectInstance(ch chan<- prometheus.Metric) error { regKey, err := registry.OpenKey(registry.LOCAL_MACHINE, regKeyName, registry.QUERY_VALUE) if err != nil { - c.logger.Debug(fmt.Sprintf("couldn't open registry %s:", regKeyName), + c.logger.Debug("couldn't open registry key", + slog.String("key", regKeyName), slog.Any("err", err), ) diff --git a/internal/collector/performancecounter/performancecounter.go b/internal/collector/performancecounter/performancecounter.go index 6acbf31f..48e0d61b 100644 --- a/internal/collector/performancecounter/performancecounter.go +++ b/internal/collector/performancecounter/performancecounter.go @@ -246,11 +246,16 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error errs = append(errs, fmt.Errorf("failed to collect object %s: %w", perfDataObject.Name, err)) success = 0.0 - c.logger.Debug(fmt.Sprintf("performancecounter collector %s failed after %s", perfDataObject.Name, duration), + c.logger.Debug("performancecounter object collection failed", + slog.String("object", perfDataObject.Name), + slog.Duration("duration", duration), slog.Any("err", err), ) } else { - c.logger.Debug(fmt.Sprintf("performancecounter collector %s succeeded after %s", perfDataObject.Name, duration)) + c.logger.Debug("performancecounter object collection succeeded", + slog.String("object", perfDataObject.Name), + slog.Duration("duration", duration), + ) } ch <- prometheus.MustNewConstMetric( diff --git a/internal/collector/textfile/textfile.go b/internal/collector/textfile/textfile.go index 618cab0c..bab2a4e5 100644 --- a/internal/collector/textfile/textfile.go +++ b/internal/collector/textfile/textfile.go @@ -187,7 +187,9 @@ func (c *Collector) convertMetricFamily(logger *slog.Logger, metricFamily *dto.M for _, metric := range metricFamily.GetMetric() { if metric.TimestampMs != nil { - logger.Warn(fmt.Sprintf("Ignoring unsupported custom timestamp on textfile metric %v", metric)) + logger.Warn("ignoring unsupported custom timestamp on textfile metric", + slog.String("metric", metric.String()), + ) } labels := metric.GetLabel() diff --git a/internal/collector/update/update.go b/internal/collector/update/update.go index 1dee53ee..5098ba2d 100644 --- a/internal/collector/update/update.go +++ b/internal/collector/update/update.go @@ -303,7 +303,9 @@ func (c *Collector) fetchUpdates(logger *slog.Logger, searcher *wuapi.UpdateSear defer result.Release() - logger.Debug(fmt.Sprintf("search for updates took %s", time.Since(timeStart))) + logger.Debug("search for updates finished", + slog.Duration("duration", time.Since(timeStart)), + ) metricsBuf = append(metricsBuf, prometheus.MustNewConstMetric( c.queryDurationSeconds, diff --git a/internal/httphandler/httphandler.go b/internal/httphandler/httphandler.go index 993b6e04..9eb27e6d 100644 --- a/internal/httphandler/httphandler.go +++ b/internal/httphandler/httphandler.go @@ -135,7 +135,10 @@ func (c *MetricsHTTPHandler) getScrapeTimeout(logger *slog.Logger, r *http.Reque // 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)) + logger.LogAttrs(r.Context(), slog.LevelWarn, "invalid X-Prometheus-Scrape-Timeout-Seconds header, using the default timeout", + slog.String("value", v), + slog.Duration("timeout", defaultScrapeTimeout), + ) default: timeout = toDuration(seconds) } diff --git a/internal/log/eventlog/eventlog.go b/internal/log/eventlog/eventlog.go index 3de637ef..bb90fc2a 100644 --- a/internal/log/eventlog/eventlog.go +++ b/internal/log/eventlog/eventlog.go @@ -20,19 +20,36 @@ package eventlog import ( "io" + "os" "regexp" "strings" + "sync" "golang.org/x/sys/windows/svc/eventlog" ) // Interface guard. -var _ io.Writer = (*Writer)(nil) +var _ io.WriteCloser = (*Writer)(nil) var reStripTimeAndLevel = regexp.MustCompile(`^time=\S+ level=\S+ `) +// Writer writes log lines to the Windows event log. It is safe for concurrent +// use. After Close, Write returns os.ErrClosed. type Writer struct { - handle *eventlog.Log + handle eventLog + + // mu is held for reading by Write and for writing by Close, so that + // writes run in parallel but never use a closed handle. + mu sync.RWMutex + closed bool +} + +// eventLog is the part of *eventlog.Log that Writer uses. +type eventLog interface { + Info(eid uint32, msg string) error + Warning(eid uint32, msg string) error + Error(eid uint32, msg string) error + Close() error } // NewEventLogWriter returns a new Writer, which writes to Windows EventLog. @@ -41,6 +58,13 @@ func NewEventLogWriter(handle *eventlog.Log) *Writer { } func (w *Writer) Write(p []byte) (int, error) { + w.mu.RLock() + defer w.mu.RUnlock() + + if w.closed { + return 0, os.ErrClosed + } + var err error msg := strings.TrimSpace(string(p)) @@ -57,3 +81,17 @@ func (w *Writer) Write(p []byte) (int, error) { return len(p), err } + +// Close releases the event log handle. Later calls do nothing. +func (w *Writer) Close() error { + w.mu.Lock() + defer w.mu.Unlock() + + if w.closed { + return nil + } + + w.closed = true + + return w.handle.Close() +} diff --git a/internal/log/eventlog/eventlog_internal_test.go b/internal/log/eventlog/eventlog_internal_test.go new file mode 100644 index 00000000..4fde782b --- /dev/null +++ b/internal/log/eventlog/eventlog_internal_test.go @@ -0,0 +1,123 @@ +// 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 eventlog + +import ( + "os" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +// fakeEventLog records the events it receives and fails every use after Close, +// like a closed event log handle. +type fakeEventLog struct { + mu sync.Mutex + events []string + closes int +} + +func (f *fakeEventLog) record(level, msg string) error { + f.mu.Lock() + defer f.mu.Unlock() + + if f.closes > 0 { + return os.ErrInvalid + } + + f.events = append(f.events, level+": "+msg) + + return nil +} + +func (f *fakeEventLog) Info(_ uint32, msg string) error { return f.record("info", msg) } +func (f *fakeEventLog) Warning(_ uint32, msg string) error { return f.record("warning", msg) } +func (f *fakeEventLog) Error(_ uint32, msg string) error { return f.record("error", msg) } + +func (f *fakeEventLog) Close() error { + f.mu.Lock() + defer f.mu.Unlock() + + f.closes++ + + return nil +} + +func TestWriterLevels(t *testing.T) { + t.Parallel() + + handle := &fakeEventLog{} + writer := &Writer{handle: handle} + + for _, line := range []string{ + "time=2026-10-15T14:03:05Z level=INFO msg=started\n", + "time=2026-10-15T14:03:05Z level=WARN msg=slow\n", + "time=2026-10-15T14:03:05Z level=ERROR msg=failed\n", + } { + n, err := writer.Write([]byte(line)) + require.NoError(t, err) + require.Equal(t, len(line), n) + } + + require.Equal(t, []string{"info: msg=started", "warning: msg=slow", "error: msg=failed"}, handle.events) +} + +func TestWriterWriteAfterClose(t *testing.T) { + t.Parallel() + + handle := &fakeEventLog{} + writer := &Writer{handle: handle} + + require.NoError(t, writer.Close()) + require.NoError(t, writer.Close(), "a second Close is a no-op") + require.Equal(t, 1, handle.closes, "the handle must be closed once") + + n, err := writer.Write([]byte("level=INFO msg=late\n")) + require.ErrorIs(t, err, os.ErrClosed) + require.Zero(t, n) + require.Empty(t, handle.events) +} + +func TestWriterConcurrentClose(t *testing.T) { + t.Parallel() + + handle := &fakeEventLog{} + writer := &Writer{handle: handle} + + var wg sync.WaitGroup + + for range 8 { + wg.Go(func() { + for range 100 { + // A write either reaches the open handle or fails with os.ErrClosed. + if _, err := writer.Write([]byte("level=INFO msg=x\n")); err != nil { + require.ErrorIs(t, err, os.ErrClosed) + } + } + }) + } + + wg.Go(func() { + require.NoError(t, writer.Close()) + }) + + wg.Wait() + + require.Equal(t, 1, handle.closes) +} diff --git a/internal/log/logger.go b/internal/log/logger.go index 0bcd8cb2..0a181a8a 100644 --- a/internal/log/logger.go +++ b/internal/log/logger.go @@ -31,9 +31,15 @@ import ( wineventlog "golang.org/x/sys/windows/svc/eventlog" ) +// eventLogFile is the AllowedFile value that selects the Windows event log. +const eventLogFile = "eventlog" + +// eventLogSource is the event source the event log output writes as. +const eventLogSource = "windows_exporter" + // AllowedFile is a settable identifier for the output file that the logger can have. -// A path is only stored by Set; New opens it, since the rotation settings are -// parsed after the file flag. +// Set only stores the value, because a later flag or the configuration file can +// replace it; New opens the file or the event log that was selected last. type AllowedFile struct { s string w io.Writer @@ -56,13 +62,6 @@ func (f *AllowedFile) Set(s string) error { f.w = os.Stdout case "stderr": f.w = os.Stderr - case "eventlog": - eventLog, err := wineventlog.Open("windows_exporter") - if err != nil { - return fmt.Errorf("failed to open event log: %w", err) - } - - f.w = eventlog.NewEventLogWriter(eventLog) default: f.w = nil } @@ -72,7 +71,7 @@ func (f *AllowedFile) Set(s string) error { // isPath reports whether the output is a log file path rather than a stream. func (f *AllowedFile) isPath() bool { - return f.s != "" && f.w == nil + return f.s != "" && f.s != eventLogFile && f.w == nil } // Rotation configures the size-based rotation of a log file. Rotation is @@ -112,8 +111,8 @@ type Config struct { Rotation Rotation } -// New creates the logger. The returned io.Closer releases the log file; it is -// nil for stdout, stderr and the event log. +// New creates the logger. The returned io.Closer releases the log file or the +// event log handle; it is nil for stdout and stderr. func New(config *Config) (*slog.Logger, io.Closer, error) { if config.File == nil { return nil, nil, errors.New("log file undefined") @@ -132,6 +131,16 @@ func New(config *Config) (*slog.Logger, io.Closer, error) { if config.Rotation.configured() { return nil, nil, fmt.Errorf("log rotation requires log.file to be a file path, not %s", config.File.s) } + + if config.File.s == eventLogFile { + eventLog, err := wineventlog.Open(eventLogSource) + if err != nil { + return nil, nil, fmt.Errorf("failed to open event log: %w", err) + } + + eventLogWriter := eventlog.NewEventLogWriter(eventLog) + writer, closer = eventLogWriter, eventLogWriter + } case config.Rotation.MaxSize > 0: file, err := newRotatingFile(config.File.s, int64(config.Rotation.MaxSize)<<20, config.Rotation.MaxBackups, config.Rotation.Compress, time.Now) if err != nil { diff --git a/internal/log/logger_test.go b/internal/log/logger_test.go index 87cf0a92..b920f1d9 100644 --- a/internal/log/logger_test.go +++ b/internal/log/logger_test.go @@ -147,6 +147,42 @@ func TestNewRotationValidation(t *testing.T) { } } +func TestAllowedFileSetEventLogIsLazy(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "windows_exporter.log") + + // The service default is replaced by --log.file: Set must not open the + // event log, or the handle would never be closed. + file := &AllowedFile{} + require.NoError(t, file.Set(eventLogFile)) + require.Nil(t, file.w) + require.False(t, file.isPath()) + + require.NoError(t, file.Set(path)) + require.True(t, file.isPath()) + + logger, closer, err := New(&Config{Config: &promslog.Config{}, File: file}) + require.NoError(t, err) + + logger.Info("hello") + require.NoError(t, closer.Close()) + require.Contains(t, readFile(t, path), "hello") +} + +func TestNewEventLog(t *testing.T) { + t.Parallel() + + logger, closer, err := New(newConfig(t, eventLogFile, Rotation{})) + require.NoError(t, err) + require.NotNil(t, logger) + require.NotNil(t, closer, "the event log handle must be released") + require.NoError(t, closer.Close()) + + _, _, err = New(newConfig(t, eventLogFile, Rotation{MaxSize: 1})) + require.ErrorContains(t, err, "requires log.file to be a file path") +} + func TestNewStreamHasNoCloser(t *testing.T) { t.Parallel() diff --git a/internal/log/rotate.go b/internal/log/rotate.go index 1a324ef5..dc9bcd3a 100644 --- a/internal/log/rotate.go +++ b/internal/log/rotate.go @@ -34,10 +34,22 @@ const ( // appended to before the rotation is tried again. rotationRetryInterval = time.Minute + // cleanupCloseTimeout bounds how long Close waits for the background + // compression and pruning of rotated files. A service has limited time + // to stop, and a backup that is left uncompressed is still a valid log. + cleanupCloseTimeout = 2 * time.Second + // cleanupAbortTimeout is how long Close waits for an aborted compression + // to remove its temporary file. + cleanupAbortTimeout = time.Second + backupTimeFormat = "2006-01-02T15-04-05.000" gzipSuffix = ".gz" + tempSuffix = ".tmp" ) +// errCompressionAborted is returned by compressFile when the log file is closed. +var errCompressionAborted = errors.New("compression aborted, the log file is closed") + // rotatingFile appends to a log file and rotates it before a write would make // it larger than maxSize. A rotation that fails, for example because another // process holds the file open without delete sharing, never drops a line: the @@ -61,9 +73,13 @@ type rotatingFile struct { checkAfter time.Time // cleanup compresses and prunes rotated files in the background; Close - // waits for it. cleanupMu runs one cleanup at a time. - cleanup sync.WaitGroup - cleanupMu sync.Mutex + // waits for it up to closeTimeout. cleanupMu runs one cleanup at a time. + // After closeTimeout, Close closes stop, which aborts a running compression. + cleanup sync.WaitGroup + cleanupMu sync.Mutex + stop chan struct{} + stopOnce sync.Once + closeTimeout time.Duration } func newRotatingFile(path string, maxSize int64, maxBackups int, compress bool, now func() time.Time) (*rotatingFile, error) { @@ -86,6 +102,8 @@ func newRotatingFileWithRename( now: now, rename: rename, retryInterval: rotationRetryInterval, + stop: make(chan struct{}), + closeTimeout: cleanupCloseTimeout, } file, size, err := openLogFileWithSize(path) @@ -96,6 +114,11 @@ func newRotatingFileWithRename( r.file, r.size = file, size r.checkAfter = now().Add(r.retryInterval) + // A compression that was interrupted, for example because the process + // was killed, leaves its temporary file behind. Its backup still exists + // uncompressed. + r.removeCompressionTempFiles() + // A file that is already too large is rotated right away. If that fails, // the logger still starts and the rotation is retried on a later write. if r.size >= r.maxSize { @@ -153,12 +176,16 @@ func (r *rotatingFile) Write(p []byte) (int, error) { return n, err } -// Close closes the log file and waits for compression and pruning of rotated files. +// Close closes the log file and waits up to closeTimeout for the background +// cleanup of rotated files. If the cleanup is still running then, Close aborts +// the compression, which keeps the rotated file uncompressed. func (r *rotatingFile) Close() error { r.mu.Lock() var err error + // Write returns before rotating once file is nil, so no cleanup starts + // after this. if r.file != nil { err = r.file.Close() r.file = nil @@ -166,9 +193,32 @@ func (r *rotatingFile) Close() error { r.mu.Unlock() - r.cleanup.Wait() + done := make(chan struct{}) - return err + go func() { + r.cleanup.Wait() + close(done) + }() + + timer := time.NewTimer(r.closeTimeout) + defer timer.Stop() + + select { + case <-done: + return err + case <-timer.C: + } + + r.stopOnce.Do(func() { close(r.stop) }) + + timer.Reset(cleanupAbortTimeout) + + select { + case <-done: + case <-timer.C: + } + + return errors.Join(err, fmt.Errorf("aborted the cleanup of rotated log files after %s", r.closeTimeout)) } // rotate renames the log file to a backup and opens a new one. r.mu must be held. @@ -256,6 +306,24 @@ func (r *rotatingFile) nameParts() (string, string, string) { return filepath.Dir(r.path), strings.TrimSuffix(base, ext) + "-", ext } +// removeCompressionTempFiles removes the temporary files of compressions of +// this log file's backups. Errors only leave files behind, so they are ignored. +func (r *rotatingFile) removeCompressionTempFiles() { + dir, prefix, ext := r.nameParts() + + entries, err := os.ReadDir(dir) + if err != nil { + return + } + + for _, entry := range entries { + name := entry.Name() + if entry.Type().IsRegular() && strings.HasPrefix(name, prefix) && strings.HasSuffix(name, ext+gzipSuffix+tempSuffix) { + _ = os.Remove(filepath.Join(dir, name)) + } + } +} + func exists(path string) bool { _, err := os.Lstat(path) @@ -266,7 +334,7 @@ func exists(path string) bool { // beyond maxBackups. Errors only leave files behind, so they are ignored. func (r *rotatingFile) cleanupBackups(backup string) { if r.compress { - _ = compressFile(backup) + _ = compressFile(backup, r.stop) } if r.maxBackups <= 0 { @@ -346,8 +414,24 @@ func (r *rotatingFile) backups() []backupFile { return backups } -// compressFile replaces path with path.gz. -func compressFile(path string) (err error) { +// stopReader reads from r until stop is closed. +type stopReader struct { + r io.Reader + stop <-chan struct{} +} + +func (s stopReader) Read(p []byte) (int, error) { + select { + case <-s.stop: + return 0, errCompressionAborted + default: + return s.r.Read(p) + } +} + +// compressFile replaces path with path.gz. When stop is closed, it stops and +// keeps path uncompressed. +func compressFile(path string, stop <-chan struct{}) (err error) { src, err := os.Open(path) if err != nil { return err @@ -355,7 +439,7 @@ func compressFile(path string) (err error) { defer func() { _ = src.Close() }() - tmp := path + gzipSuffix + ".tmp" + tmp := path + gzipSuffix + tempSuffix dst, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o640) if err != nil { @@ -371,7 +455,7 @@ func compressFile(path string) (err error) { writer := gzip.NewWriter(dst) - if _, err = io.Copy(writer, src); err != nil { + if _, err = io.Copy(writer, stopReader{r: src, stop: stop}); err != nil { return err } diff --git a/internal/log/rotate_test.go b/internal/log/rotate_test.go index d4909f6f..4e04b215 100644 --- a/internal/log/rotate_test.go +++ b/internal/log/rotate_test.go @@ -222,6 +222,79 @@ func TestRotatingFileCompress(t *testing.T) { require.Equal(t, "abcd", string(data)) } +func TestRotatingFileCloseBoundsCleanup(t *testing.T) { + t.Parallel() + + file, _, _ := testRotatingFile(t, 10, 0, false) + file.closeTimeout = 50 * time.Millisecond + + // A cleanup that only returns once Close aborts it, like a compression + // that is still reading. + file.cleanup.Go(func() { + <-file.stop + }) + + start := time.Now() + err := file.Close() + elapsed := time.Since(start) + + require.ErrorContains(t, err, "aborted the cleanup of rotated log files") + require.GreaterOrEqual(t, elapsed, file.closeTimeout) + require.Less(t, elapsed, file.closeTimeout+cleanupAbortTimeout, "Close did not return when the aborted cleanup did") + + // A cleanup that ignores the abort delays Close by at most cleanupAbortTimeout. + block := make(chan struct{}) + defer close(block) + + file.cleanup.Go(func() { + <-block + }) + + start = time.Now() + + require.Error(t, file.Close()) + require.Less(t, time.Since(start), file.closeTimeout+cleanupAbortTimeout+time.Second) +} + +func TestRotatingFileRemovesCompressionTempFiles(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + path := filepath.Join(dir, "windows_exporter.log") + + for _, name := range []string{ + "windows_exporter-2026-10-15T14-03-05.000.log", + "windows_exporter-2026-10-15T14-03-05.000.log.gz.tmp", + "other-2026-10-15T14-03-05.000.log.gz.tmp", + } { + require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte("x"), 0o600)) + } + + file, err := newRotatingFile(path, 100, 0, true, time.Now) + require.NoError(t, err) + require.NoError(t, file.Close()) + + require.Equal(t, []string{ + "other-2026-10-15T14-03-05.000.log.gz.tmp", + "windows_exporter-2026-10-15T14-03-05.000.log", + "windows_exporter.log", + }, dirEntries(t, dir)) +} + +func TestCompressFileAborts(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "windows_exporter-2026-10-15T14-03-05.000.log") + require.NoError(t, os.WriteFile(path, []byte("abcd"), 0o600)) + + stop := make(chan struct{}) + close(stop) + + require.ErrorIs(t, compressFile(path, stop), errCompressionAborted) + require.Equal(t, []string{filepath.Base(path)}, dirEntries(t, filepath.Dir(path)), "the rotated file must stay uncompressed, without a temporary file") + require.Equal(t, "abcd", readFile(t, path)) +} + func TestRotatingFileBackupNamesAreUnique(t *testing.T) { t.Parallel() diff --git a/pkg/collector/collect.go b/pkg/collector/collect.go index aee87e74..a1d53443 100644 --- a/pkg/collector/collect.go +++ b/pkg/collector/collect.go @@ -179,7 +179,9 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus. switch state.tryCollect() { case started: case notAvailable: - logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf("collector %s skipped, because it isn't initialized", name)) + logger.LogAttrs(ctx, slog.LevelDebug, "collector skipped, because it isn't initialized", + slog.String("collector", name), + ) return unavailable case busy: @@ -188,7 +190,9 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus. level = slog.LevelWarn } - logger.LogAttrs(ctx, level, fmt.Sprintf("collector %s skipped, because its call from an earlier scrape is still running", name)) + logger.LogAttrs(ctx, level, "collector skipped, because its call from an earlier scrape is still running", + slog.String("collector", name), + ) return pending } @@ -239,9 +243,16 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus. 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)) + logger.LogAttrs(ctx, slog.LevelWarn, "collector timed out", + slog.String("collector", name), + slog.Duration("duration", duration), + slog.Int("metrics", numMetrics), + ) } else { - logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf("collector %s canceled after %s, resulting in %d metrics", name, duration, numMetrics), + logger.LogAttrs(ctx, slog.LevelDebug, "collector canceled", + slog.String("collector", name), + slog.Duration("duration", duration), + slog.Int("metrics", numMetrics), slog.Any("err", ctx.Err()), ) } @@ -291,36 +302,31 @@ func (c *Collection) sendCollectorDuration(ch chan<- prometheus.Metric, name str } 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" - - if err != nil { - if !expectedCollectionError(err) { - if errors.Is(err, pdh.ErrPerformanceCounterNotInitialized) { - err = fmt.Errorf("%w. Check application logs from initialization phase for more information", err) - } - - logger.LogAttrs(ctx, slog.LevelWarn, - fmt.Sprintf("collector %s failed after %s, resulting in %d metrics", name, duration, numMetrics), - slog.Any("err", err), - ) - - return failed - } - - slogAttrs = append(slogAttrs, slog.Any("err", err)) - - result = "succeeded with warnings" + attrs := []slog.Attr{ + slog.String("collector", name), + slog.Duration("duration", duration), + slog.Int("metrics", numMetrics), } - logger.LogAttrs(ctx, slog.LevelDebug, fmt.Sprintf( - "collector %s %s after %s, resulting in %d metrics", name, result, duration, numMetrics, - ), - slogAttrs..., - ) + if err == nil { + logger.LogAttrs(ctx, slog.LevelDebug, "collector succeeded", attrs...) - return success + return success + } + + if expectedCollectionError(err) { + logger.LogAttrs(ctx, slog.LevelDebug, "collector succeeded with warnings", append(attrs, slog.Any("err", err))...) + + return success + } + + if errors.Is(err, pdh.ErrPerformanceCounterNotInitialized) { + err = fmt.Errorf("%w. Check application logs from initialization phase for more information", err) + } + + logger.LogAttrs(ctx, slog.LevelWarn, "collector failed", append(attrs, slog.Any("err", err))...) + + return failed } // expectedCollectionError checks every cause so an expected error cannot hide an independent failure. diff --git a/pkg/collector/collect_test.go b/pkg/collector/collect_test.go index 192de785..43a9772e 100644 --- a/pkg/collector/collect_test.go +++ b/pkg/collector/collect_test.go @@ -175,7 +175,7 @@ func TestCollectTimedOutCollectorDoesNotOverlap(t *testing.T) { // 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("timed out")) require.Equal(t, 1, logs.warnings("is still running")) // After the first call returns, the next scrape starts a new one.