log: open the event log lazily, bound close, keep the service running, structured messages (#2734)

Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>
This commit is contained in:
Jan-Otto Kröpke
2026-10-16 02:31:55 +02:00
committed by GitHub
parent a1ac462daf
commit af2a7c3f6c
19 changed files with 610 additions and 81 deletions
+5 -4
View File
@@ -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
}
+3 -1
View File
@@ -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,
+12 -3
View File
@@ -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 == "" {
+2 -1
View File
@@ -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),
)
@@ -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(
+3 -1
View File
@@ -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()
+3 -1
View File
@@ -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,
+4 -1
View File
@@ -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)
}
+40 -2
View File
@@ -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()
}
@@ -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)
}
+21 -12
View File
@@ -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 {
+36
View File
@@ -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()
+95 -11
View File
@@ -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
}
+73
View File
@@ -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()