From b9a08b72ebdf0da9a193bfb3cd020881a23670d9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:31:16 +0000 Subject: [PATCH] fix: avoid blocking exporter on console logging Co-authored-by: jkroepke <1560587+jkroepke@users.noreply.github.com> --- internal/log/logger.go | 15 +++- internal/log/nonblocking_writer.go | 81 +++++++++++++++++ internal/log/nonblocking_writer_test.go | 114 ++++++++++++++++++++++++ 3 files changed, 208 insertions(+), 2 deletions(-) create mode 100644 internal/log/nonblocking_writer.go create mode 100644 internal/log/nonblocking_writer_test.go diff --git a/internal/log/logger.go b/internal/log/logger.go index fb6d4872..bbd8492f 100644 --- a/internal/log/logger.go +++ b/internal/log/logger.go @@ -26,6 +26,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/log/eventlog" "github.com/prometheus/common/promslog" + "golang.org/x/sys/windows" wineventlog "golang.org/x/sys/windows/svc/eventlog" ) @@ -49,9 +50,9 @@ func (f *AllowedFile) Set(s string) error { switch s { case "stdout": - f.w = os.Stdout + f.w = maybeNonBlockingConsoleWriter(os.Stdout) case "stderr": - f.w = os.Stderr + f.w = maybeNonBlockingConsoleWriter(os.Stderr) case "eventlog": eventLog, err := wineventlog.Open("windows_exporter") if err != nil { @@ -71,6 +72,16 @@ func (f *AllowedFile) Set(s string) error { return nil } +func maybeNonBlockingConsoleWriter(file *os.File) io.Writer { + var mode uint32 + + if err := windows.GetConsoleMode(windows.Handle(file.Fd()), &mode); err == nil { + return newNonBlockingWriter(file, defaultNonBlockingWriterBufferSize) + } + + return file +} + // Config is a struct containing configurable settings for the logger. type Config struct { *promslog.Config diff --git a/internal/log/nonblocking_writer.go b/internal/log/nonblocking_writer.go new file mode 100644 index 00000000..2da7ffd8 --- /dev/null +++ b/internal/log/nonblocking_writer.go @@ -0,0 +1,81 @@ +// 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. + +package log + +import ( + "bytes" + "io" + "sync" + "sync/atomic" +) + +const defaultNonBlockingWriterBufferSize = 1024 + +type nonBlockingWriter struct { + writer io.Writer + queue chan []byte + done chan struct{} + closed atomic.Bool + once sync.Once +} + +func newNonBlockingWriter(writer io.Writer, queueSize int) *nonBlockingWriter { + if queueSize <= 0 { + queueSize = defaultNonBlockingWriterBufferSize + } + + w := &nonBlockingWriter{ + writer: writer, + queue: make(chan []byte, queueSize), + done: make(chan struct{}), + } + + go func() { + defer close(w.done) + + for p := range w.queue { + _, _ = w.writer.Write(p) + } + }() + + return w +} + +func (w *nonBlockingWriter) Write(p []byte) (int, error) { + if w.closed.Load() { + return 0, io.ErrClosedPipe + } + + msg := bytes.Clone(p) + + select { + case w.queue <- msg: + default: + } + + return len(p), nil +} + +func (w *nonBlockingWriter) Close() error { + w.once.Do(func() { + w.closed.Store(true) + close(w.queue) + }) + + <-w.done + + return nil +} diff --git a/internal/log/nonblocking_writer_test.go b/internal/log/nonblocking_writer_test.go new file mode 100644 index 00000000..a78edca5 --- /dev/null +++ b/internal/log/nonblocking_writer_test.go @@ -0,0 +1,114 @@ +// 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. + +package log + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type blockingWriter struct { + started chan struct{} + release chan struct{} + + mu sync.Mutex + writes [][]byte +} + +func (w *blockingWriter) Write(p []byte) (int, error) { + select { + case <-w.started: + default: + close(w.started) + } + + <-w.release + + w.mu.Lock() + defer w.mu.Unlock() + + w.writes = append(w.writes, append([]byte(nil), p...)) + + return len(p), nil +} + +func (w *blockingWriter) Writes() [][]byte { + w.mu.Lock() + defer w.mu.Unlock() + + result := make([][]byte, len(w.writes)) + copy(result, w.writes) + + return result +} + +func TestNonBlockingWriterDropsInsteadOfBlocking(t *testing.T) { + t.Parallel() + + writer := &blockingWriter{ + started: make(chan struct{}), + release: make(chan struct{}), + } + var cleanupOnce sync.Once + + nonBlockingWriter := newNonBlockingWriter(writer, 1) + t.Cleanup(func() { + cleanupOnce.Do(func() { + close(writer.release) + require.NoError(t, nonBlockingWriter.Close()) + }) + }) + + _, err := nonBlockingWriter.Write([]byte("first")) + require.NoError(t, err) + + select { + case <-writer.started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for background write to start") + } + + _, err = nonBlockingWriter.Write([]byte("second")) + require.NoError(t, err) + + done := make(chan struct{}) + + go func() { + defer close(done) + + _, err := nonBlockingWriter.Write([]byte("third")) + require.NoError(t, err) + }() + + select { + case <-done: + case <-time.After(100 * time.Millisecond): + t.Fatal("write blocked while output writer was stalled") + } + + cleanupOnce.Do(func() { + close(writer.release) + require.NoError(t, nonBlockingWriter.Close()) + }) + + writes := writer.Writes() + require.Len(t, writes, 2) + require.Equal(t, []byte("first"), writes[0]) + require.Equal(t, []byte("second"), writes[1]) +}