fix: avoid blocking exporter on console logging

Co-authored-by: jkroepke <1560587+jkroepke@users.noreply.github.com>

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

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
+}

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])
+}