Prevent races when writing to stdout/stderr (#1031)

Author: Andrew-Lees11

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Change log
 
+## 9.4.3.1-r2 (2025-09)
+
+* Logs now consistently routed to stdout/stderr; logs to stdout will conform to the selected log format (MQ_LOGGING_CONSOLE_FORMAT), logs to stderr are unformatted errors
+
+## 9.4.3.1-r1 (2025-08)
+
+* Updated to MQ version 9.4.3.1
+
 ## 9.4.3.0 (2025-06)
 
 * Updated to MQ version 9.4.3.0

cmd/runmqdevserver/main.go

@@ -65,17 +65,17 @@ func configureLogger() error {
 	}
 	switch f {
 	case "json":
-		log, err = logger.NewLogger(os.Stderr, d, true, n)
+		log, err = logger.NewLogger(os.Stdout, d, true, n)
 		if err != nil {
 			return err
 		}
 	case "basic":
-		log, err = logger.NewLogger(os.Stderr, d, false, n)
+		log, err = logger.NewLogger(os.Stdout, d, false, n)
 		if err != nil {
 			return err
 		}
 	default:
-		log, err = logger.NewLogger(os.Stdout, d, false, n)
+		log, err = logger.NewLogger(os.Stderr, d, false, n)
 		return fmt.Errorf("invalid value for LOG_FORMAT: %v", f)
 	}
 	return nil

cmd/runmqserver/logging.go

@@ -34,6 +34,7 @@ import (
 	"github.com/ibm-messaging/mq-container/internal/pathutils"
 	"github.com/ibm-messaging/mq-container/pkg/logger"
 	"github.com/ibm-messaging/mq-container/pkg/mqini"
+	"github.com/ibm-messaging/mq-container/pkg/syncwriter"
 )
 
 // var debug = false
@@ -288,10 +289,11 @@ func configureLogger(name string) (mirrorFunc, error) {
 	d := getDebug()
 	switch f {
 	case "json":
-		log, err = logger.NewLogger(os.Stderr, d, true, name)
+		log, err = logger.NewLogger(os.Stdout, d, true, name)
 		if err != nil {
 			return nil, err
 		}
+		mirrorLogWriter := syncwriter.For(os.Stdout)
 		return func(msg string, isQMLog bool) bool {
 			arrLoggingConsoleExcludeIds := strings.Split(strings.ToUpper(os.Getenv("MQ_LOGGING_CONSOLE_EXCLUDE_ID")), ",")
 			if isExcludedMsgIdPresent(msg, arrLoggingConsoleExcludeIds) {
@@ -307,24 +309,25 @@ func configureLogger(name string) (mirrorFunc, error) {
 				if err != nil {
 					log.Printf("Failed to unmarshall JSON in log message - %v", msg)
 				} else {
-					fmt.Println(msg)
+					mirrorLogWriter.Println(msg)
 				}
 			} else {
 				// The log being mirrored isn't JSON. This can happen only in case of 'mqsc' logs
 				// Also if the logging source is from autocfgmqsc.LOG, then we have to construct the json string as per below logic
 				if checkLogSourceForMirroring("mqsc") && canMQSCLogBeMirroredToConsole(msg) {
 					logLevel := determineMQSCLogLevel(strings.TrimSpace(msg))
-					fmt.Printf("{\"ibm_datetime\":\"%s\",\"type\":\"mqsc_log\",\"loglevel\":\"%s\",\"message\":\"%s\"}\n",
+					mirrorLogWriter.Printf("{\"ibm_datetime\":\"%s\",\"type\":\"mqsc_log\",\"loglevel\":\"%s\",\"message\":\"%s\"}\n",
 						getTimeStamp(), logLevel, strings.TrimSpace(msg))
 				}
 			}
 			return true
 		}, nil
 	case "basic":
-		log, err = logger.NewLogger(os.Stderr, d, false, name)
+		log, err = logger.NewLogger(os.Stdout, d, false, name)
 		if err != nil {
 			return nil, err
 		}
+		mirrorLogWriter := syncwriter.For(os.Stdout)
 		return func(msg string, isQMLog bool) bool {
 			arrLoggingConsoleExcludeIds := strings.Split(strings.ToUpper(os.Getenv("MQ_LOGGING_CONSOLE_EXCLUDE_ID")), ",")
 			if isExcludedMsgIdPresent(msg, arrLoggingConsoleExcludeIds) {
@@ -341,18 +344,18 @@ func configureLogger(name string) (mirrorFunc, error) {
 				if err != nil {
 					log.Printf("Failed to unmarshall JSON in log message - %v", err)
 				} else {
-					fmt.Print(formatBasic(obj))
+					mirrorLogWriter.Print(formatBasic(obj))
 				}
 			} else {
 				// The log being mirrored isn't JSON, so just print it. This can happen only in case of mqsc logs
 				if checkLogSourceForMirroring("mqsc") && canMQSCLogBeMirroredToConsole(msg) {
-					log.Printf("%s", strings.TrimSpace(msg))
+					log.Print(strings.TrimSpace(msg))
 				}
 			}
 			return true
 		}, nil
 	default:
-		log, err = logger.NewLogger(os.Stdout, d, false, name)
+		log, err = logger.NewLogger(os.Stderr, d, false, name)
 		if err != nil {
 			return nil, err
 		}

internal/fips/fips_test.go

@@ -18,7 +18,6 @@ limitations under the License.
 package fips
 
 import (
-	"fmt"
 	"os"
 	"testing"
 )
@@ -35,7 +34,7 @@ func TestEnableFIPSAuto(t *testing.T) {
 func TestEnableFIPSTrue(t *testing.T) {
 	// Test MQ_ENABLE_FIPS=true
 	os.Setenv("MQ_ENABLE_FIPS", "true")
-	fmt.Println(os.Getenv("MQ_ENABLE_FIPS"))
+	t.Log(os.Getenv("MQ_ENABLE_FIPS"))
 	ProcessFIPSType(nil)
 	fipsType := IsFIPSEnabled()
 	if !fipsType {

internal/metrics/audit.go

@@ -22,6 +22,8 @@ import (
 	"net/http"
 	"os"
 	"time"
+
+	"github.com/ibm-messaging/mq-container/pkg/syncwriter"
 )
 
 type auditEvent struct {
@@ -63,7 +65,7 @@ func newAuditingHandlerFuncWrapper(qmName string, logger logHandler) handlerFunc
 			eventBytes, err := json.Marshal(event)
 			if err != nil {
 				logger.Append(fmt.Sprintf("Error writing audit log; next event may contain incomplete data: %s", err.Error()), false)
-				fmt.Printf("Error constructing audit log event: %s\n", err.Error())
+				syncwriter.For(os.Stderr).Printf("Error constructing audit log event: %s\n", err.Error())
 			}
 			logger.Append(string(eventBytes), false)
 		}

pkg/logger/logger.go

@@ -24,8 +24,9 @@ import (
 	"os"
 	"os/user"
 	"strconv"
-	"sync"
 	"time"
+
+	"github.com/ibm-messaging/mq-container/pkg/syncwriter"
 )
 
 // timestampFormat matches the format used by MQ messages (includes milliseconds)
@@ -36,8 +37,7 @@ const errorLevel string = "ERROR"
 
 // A Logger is used to log messages to stdout
 type Logger struct {
-	mutex       sync.Mutex
-	writer      io.Writer
+	writer      *syncwriter.SyncWriter
 	debug       bool
 	json        bool
 	processName string
@@ -62,8 +62,7 @@ func NewLogger(writer io.Writer, debug bool, json bool, serverName string) (*Log
 		userName = user.Username
 	}
 	return &Logger{
-		mutex:       sync.Mutex{},
-		writer:      writer,
+		writer:      syncwriter.For(writer),
 		debug:       debug,
 		json:        json,
 		processName: os.Args[0],
@@ -101,17 +100,14 @@ func (l *Logger) log(level string, msg string) {
 		"type":            "mq_containerlog",
 	}
 	s, err := l.format(entry)
-	l.mutex.Lock()
 	if err != nil {
-		// TODO: Fix this
-		fmt.Println(err)
+		syncwriter.For(os.Stderr).Println(err)
 	}
 	if l.json {
-		fmt.Fprintln(l.writer, s)
+		l.writer.Println(s)
 	} else {
-		fmt.Fprint(l.writer, s)
+		l.writer.Print(s)
 	}
-	l.mutex.Unlock()
 }
 
 // Debug logs a line as debug

pkg/logger/logger_test.go

@@ -32,7 +32,7 @@ func TestJSONLogger(t *testing.T) {
 	s := "Hello world"
 	l.Print(s)
 	var e map[string]interface{}
-	err = json.Unmarshal([]byte(buf.String()), &e)
+	err = json.Unmarshal(buf.Bytes(), &e)
 	if err != nil {
 		t.Error(err)
 	}

pkg/logrotation/logrotation.go

@@ -25,6 +25,8 @@ import (
 	"slices"
 	"strings"
 	"sync"
+
+	"github.com/ibm-messaging/mq-container/pkg/syncwriter"
 )
 
 type RotatingLogger struct {
@@ -86,24 +88,26 @@ func (r *RotatingLogger) Append(messageLine string, deduplicateLine bool) {
 	// we will always log in the first instance of the log files
 	logFilePath := r.instanceFileName(1)
 
+	errOutput := syncwriter.For(os.Stderr)
+
 	// open the log file in append mode
 	// for the gosec rule Id: G302 - Expect file permissions to be 0600 or less
 	logFile, err := os.OpenFile(logFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
 	if err != nil {
-		fmt.Printf("Failed to open log file: %v\n", err)
+		errOutput.Printf("Failed to open log file: %v\n", err)
 		return
 	}
 
 	defer func(f *os.File) {
 		if err := logFile.Close(); err != nil {
-			fmt.Printf("Error: %v, Failed to close log file: %s\n", err, logFilePath)
+			errOutput.Printf("Error: %v, Failed to close log file: %s\n", err, logFilePath)
 		}
 	}(logFile)
 
 	// check if the message should be appended to the file
 	shouldBeAppended, err := r.checkIfMessageShouldBeAppended(logFilePath, messageLine, deduplicateLine)
 	if err != nil {
-		fmt.Printf("Failed to validate the currentLog and the lastLog line %v\n", err)
+		errOutput.Printf("Failed to validate the currentLog and the lastLog line %v\n", err)
 	}
 
 	if !shouldBeAppended {
@@ -113,7 +117,7 @@ func (r *RotatingLogger) Append(messageLine string, deduplicateLine bool) {
 	// check if the logFileSize has exceeded the maxFileSize then perform the logrotation
 	logFileSizeExceeded, err := r.checkIfLogFileSizeExceeded(len(messageLine), logFile)
 	if err != nil {
-		fmt.Printf("Failed to validate log file size: %v\n", err)
+		errOutput.Printf("Failed to validate log file size: %v\n", err)
 		return
 	}
 
@@ -122,33 +126,33 @@ func (r *RotatingLogger) Append(messageLine string, deduplicateLine bool) {
 		// close the current log file
 		err = logFile.Close()
 		if err != nil {
-			fmt.Printf("Error: %v, Failed to close log file: %v\n", err, logFile.Name())
+			errOutput.Printf("Error: %v, Failed to close log file: %v\n", err, logFile.Name())
 		}
 
 		// perform log rotation
 		err = r.performLogRotation()
 		if err != nil {
-			fmt.Printf("Failed to perform log-rotation: %v\n", err)
+			errOutput.Printf("Failed to perform log-rotation: %v\n", err)
 		}
 
 		// open the newly created logFile
 		logFile, err = os.OpenFile(logFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
 		if err != nil {
-			fmt.Printf("Failed to open log file: %v\n", err)
+			errOutput.Printf("Failed to open log file: %v\n", err)
 			return
 		}
 
 		defer func(f *os.File) {
 			if err := logFile.Close(); err != nil {
-				fmt.Printf("Error: %v, Failed to close log file: %s\n", err, logFilePath)
+				errOutput.Printf("Error: %v, Failed to close log file: %s\n", err, logFilePath)
 			}
 		}(logFile)
 	}
 
 	// append the message to the file
 	_, err = logFile.WriteString(messageLine)
 	if err != nil {
-		fmt.Printf("Failed to write to log file: %v\n", err)
+		errOutput.Printf("Failed to write to log file: %v\n", err)
 	}
 
 }
@@ -225,7 +229,7 @@ func (r *RotatingLogger) getLogLastLine(logFilePath string) (string, error) {
 
 	defer func() {
 		if err := logFile.Close(); err != nil {
-			fmt.Printf("error closing logfile: %s", logFilePath)
+			syncwriter.For(os.Stderr).Printf("error closing logfile: %s", logFilePath)
 		}
 	}()
 

pkg/syncwriter/syncwriter.go

@@ -0,0 +1,66 @@
+package syncwriter
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"sync"
+)
+
+var (
+	syncWriters = map[io.Writer]*SyncWriter{}
+
+	loggerMutex            = sync.Mutex{}
+	sharedStdoutStderrLock = sync.Mutex{}
+)
+
+type SyncWriter struct {
+	writeLock *sync.Mutex
+	output    io.Writer
+}
+
+func (s *SyncWriter) Write(p []byte) (n int, err error) {
+	s.writeLock.Lock()
+	defer s.writeLock.Unlock()
+
+	bytesWritten := 0
+	for len(p) > 0 {
+		n, err := s.output.Write(p)
+		bytesWritten += n
+		if err != nil {
+			return bytesWritten, err
+		}
+		p = p[n:]
+	}
+	return bytesWritten, nil
+}
+
+func (s *SyncWriter) Print(a ...any)                 { fmt.Fprint(s, a...) }
+func (s *SyncWriter) Println(a ...any)               { fmt.Fprintln(s, a...) }
+func (s *SyncWriter) Printf(format string, a ...any) { fmt.Fprintf(s, format, a...) }
+
+// For returns a SyncWriter for the given underlying writer.
+//
+// A separate SyncWriter will be created for each underlying writer but multiple calls supplying the same writer will return the same SyncWriter.
+//
+// Note: as a special case, stdout and stderr share a write lock to prevent race conditions where these streams are converged in container logs
+func For(w io.Writer) *SyncWriter {
+	loggerMutex.Lock()
+	defer loggerMutex.Unlock()
+
+	if writer, found := syncWriters[w]; found {
+		return writer
+	}
+	var lock *sync.Mutex
+	if w == os.Stdout || w == os.Stderr {
+		lock = &sharedStdoutStderrLock
+	} else {
+		lock = &sync.Mutex{}
+	}
+	writer := &SyncWriter{
+		output:    w,
+		writeLock: lock,
+	}
+	syncWriters[w] = writer
+	return writer
+}