Give every test a bus of its own, at the release the mesh runs (hq ADR 0237)

The live tests reached one shared bus and assert, read and remove the mesh's own objects by
their fixed names, so packages run in parallel deleted what each other read and the suite
passed only one package at a time; a red suite read as noise. internal/testbus starts a server
per test, linked in at the nats-server release go.mod pins, and a test holds that pin to the
catalogue's bus image and to the facts snapshot's bus when there is one, so the tests never run
a bus the mesh does not. The waiter test read a timing (the most connections held at one look)
and now reads the state it means (the fewest held across the wait). make check runs the packages
in parallel under the race detector, with a timeout.
This commit is contained in:
jochen
2026-10-06 21:17:02 +02:00
parent 0090bf6af7
commit be92762969
627 changed files with 437174 additions and 166 deletions
+405
View File
@@ -0,0 +1,405 @@
// Copyright 2012-2025 The NATS 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 logger provides logging facilities for the NATS server
package logger
import (
"fmt"
"log"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"time"
)
// Default file permissions for log files.
const defaultLogPerms = os.FileMode(0640)
// Logger is the server logger
type Logger struct {
sync.Mutex
logger *log.Logger
debug bool
trace bool
infoLabel string
warnLabel string
errorLabel string
fatalLabel string
debugLabel string
traceLabel string
fl *fileLogger
}
type LogOption interface {
isLoggerOption()
}
// LogUTC controls whether timestamps in the log output should be UTC or local time.
type LogUTC bool
func (l LogUTC) isLoggerOption() {}
func logFlags(time bool, opts ...LogOption) int {
flags := 0
if time {
flags = log.LstdFlags | log.Lmicroseconds
}
for _, opt := range opts {
switch v := opt.(type) {
case LogUTC:
if time && bool(v) {
flags |= log.LUTC
}
}
}
return flags
}
// NewStdLogger creates a logger with output directed to Stderr
func NewStdLogger(time, debug, trace, colors, pid bool, opts ...LogOption) *Logger {
flags := logFlags(time, opts...)
pre := ""
if pid {
pre = pidPrefix()
}
l := &Logger{
logger: log.New(os.Stderr, pre, flags),
debug: debug,
trace: trace,
}
if colors {
setColoredLabelFormats(l)
} else {
setPlainLabelFormats(l)
}
return l
}
// NewFileLogger creates a logger with output directed to a file
func NewFileLogger(filename string, time, debug, trace, pid bool, opts ...LogOption) *Logger {
flags := logFlags(time, opts...)
pre := ""
if pid {
pre = pidPrefix()
}
fl, err := newFileLogger(filename, pre, time)
if err != nil {
log.Fatalf("error opening file: %v", err)
return nil
}
l := &Logger{
logger: log.New(fl, pre, flags),
debug: debug,
trace: trace,
fl: fl,
}
fl.Lock()
fl.l = l
fl.Unlock()
setPlainLabelFormats(l)
return l
}
type writerAndCloser interface {
Write(b []byte) (int, error)
Close() error
Name() string
}
type fileLogger struct {
out int64
canRotate int32
sync.Mutex
l *Logger
f writerAndCloser
limit int64
olimit int64
pid string
time bool
closed bool
maxNumFiles int
}
func newFileLogger(filename, pidPrefix string, time bool) (*fileLogger, error) {
fileflags := os.O_WRONLY | os.O_APPEND | os.O_CREATE
f, err := os.OpenFile(filename, fileflags, defaultLogPerms)
if err != nil {
return nil, err
}
stats, err := f.Stat()
if err != nil {
f.Close()
return nil, err
}
fl := &fileLogger{
canRotate: 0,
f: f,
out: stats.Size(),
pid: pidPrefix,
time: time,
}
return fl, nil
}
func (l *fileLogger) setLimit(limit int64) {
l.Lock()
l.olimit, l.limit = limit, limit
atomic.StoreInt32(&l.canRotate, 1)
rotateNow := l.out > l.limit
l.Unlock()
if rotateNow {
l.l.Noticef("Rotating logfile...")
}
}
func (l *fileLogger) setMaxNumFiles(max int) {
l.Lock()
l.maxNumFiles = max
l.Unlock()
}
func (l *fileLogger) logDirect(label, format string, v ...any) int {
var entrya = [256]byte{}
var entry = entrya[:0]
if l.pid != "" {
entry = append(entry, l.pid...)
}
if l.time {
now := time.Now()
year, month, day := now.Date()
hour, min, sec := now.Clock()
microsec := now.Nanosecond() / 1000
entry = append(entry, fmt.Sprintf("%04d/%02d/%02d %02d:%02d:%02d.%06d ",
year, month, day, hour, min, sec, microsec)...)
}
entry = append(entry, label...)
entry = append(entry, fmt.Sprintf(format, v...)...)
entry = append(entry, '\r', '\n')
l.f.Write(entry)
return len(entry)
}
func (l *fileLogger) logPurge(fname string) {
var backups []string
lDir := filepath.Dir(fname)
lBase := filepath.Base(fname)
entries, err := os.ReadDir(lDir)
if err != nil {
l.logDirect(l.l.errorLabel, "Unable to read directory %q for log purge (%v), will attempt next rotation", lDir, err)
return
}
for _, entry := range entries {
if entry.IsDir() || entry.Name() == lBase || !strings.HasPrefix(entry.Name(), lBase) {
continue
}
if stamp, found := strings.CutPrefix(entry.Name(), fmt.Sprintf("%s%s", lBase, ".")); found {
_, err := time.Parse("2006:01:02:15:04:05.999999999", strings.Replace(stamp, ".", ":", 5))
if err == nil {
backups = append(backups, entry.Name())
}
}
}
currBackups := len(backups)
maxBackups := l.maxNumFiles - 1
if currBackups > maxBackups {
// backups sorted oldest to latest based on timestamped lexical filename (ReadDir)
for i := 0; i < currBackups-maxBackups; i++ {
if err := os.Remove(filepath.Join(lDir, string(os.PathSeparator), backups[i])); err != nil {
l.logDirect(l.l.errorLabel, "Unable to remove backup log file %q (%v), will attempt next rotation", backups[i], err)
// Bail fast, we'll try again next rotation
return
}
l.logDirect(l.l.infoLabel, "Purged log file %q", backups[i])
}
}
}
func (l *fileLogger) Write(b []byte) (int, error) {
if atomic.LoadInt32(&l.canRotate) == 0 {
n, err := l.f.Write(b)
if err == nil {
atomic.AddInt64(&l.out, int64(n))
}
return n, err
}
l.Lock()
n, err := l.f.Write(b)
if err == nil {
l.out += int64(n)
if l.out > l.limit {
if err := l.f.Close(); err != nil {
l.limit *= 2
l.logDirect(l.l.errorLabel, "Unable to close logfile for rotation (%v), will attempt next rotation at size %v", err, l.limit)
l.Unlock()
return n, err
}
fname := l.f.Name()
now := time.Now()
bak := fmt.Sprintf("%s.%04d.%02d.%02d.%02d.%02d.%02d.%09d", fname,
now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(),
now.Second(), now.Nanosecond())
os.Rename(fname, bak)
fileflags := os.O_WRONLY | os.O_APPEND | os.O_CREATE
f, err := os.OpenFile(fname, fileflags, defaultLogPerms)
if err != nil {
l.Unlock()
panic(fmt.Sprintf("Unable to re-open the logfile %q after rotation: %v", fname, err))
}
l.f = f
n := l.logDirect(l.l.infoLabel, "Rotated log, backup saved as %q", bak)
l.out = int64(n)
l.limit = l.olimit
if l.maxNumFiles > 0 {
l.logPurge(fname)
}
}
}
l.Unlock()
return n, err
}
func (l *fileLogger) close() error {
l.Lock()
if l.closed {
l.Unlock()
return nil
}
l.closed = true
l.Unlock()
return l.f.Close()
}
// SetSizeLimit sets the size of a logfile after which a backup
// is created with the file name + "year.month.day.hour.min.sec.nanosec"
// and the current log is truncated.
func (l *Logger) SetSizeLimit(limit int64) error {
l.Lock()
if l.fl == nil {
l.Unlock()
return fmt.Errorf("can set log size limit only for file logger")
}
fl := l.fl
l.Unlock()
fl.setLimit(limit)
return nil
}
// SetMaxNumFiles sets the number of archived log files that will be retained
func (l *Logger) SetMaxNumFiles(max int) error {
l.Lock()
if l.fl == nil {
l.Unlock()
return fmt.Errorf("can set log max number of files only for file logger")
}
fl := l.fl
l.Unlock()
fl.setMaxNumFiles(max)
return nil
}
// NewTestLogger creates a logger with output directed to Stderr with a prefix.
// Useful for tracing in tests when multiple servers are in the same pid
func NewTestLogger(prefix string, time bool) *Logger {
flags := 0
if time {
flags = log.LstdFlags | log.Lmicroseconds
}
l := &Logger{
logger: log.New(os.Stderr, prefix, flags),
debug: true,
trace: true,
}
setColoredLabelFormats(l)
return l
}
// Close implements the io.Closer interface to clean up
// resources in the server's logger implementation.
// Caller must ensure threadsafety.
func (l *Logger) Close() error {
if l.fl != nil {
return l.fl.close()
}
return nil
}
// Generate the pid prefix string
func pidPrefix() string {
return fmt.Sprintf("[%d] ", os.Getpid())
}
func setPlainLabelFormats(l *Logger) {
l.infoLabel = "[INF] "
l.debugLabel = "[DBG] "
l.warnLabel = "[WRN] "
l.errorLabel = "[ERR] "
l.fatalLabel = "[FTL] "
l.traceLabel = "[TRC] "
}
func setColoredLabelFormats(l *Logger) {
colorFormat := "[\x1b[%sm%s\x1b[0m] "
l.infoLabel = fmt.Sprintf(colorFormat, "32", "INF")
l.debugLabel = fmt.Sprintf(colorFormat, "36", "DBG")
l.warnLabel = fmt.Sprintf(colorFormat, "0;93", "WRN")
l.errorLabel = fmt.Sprintf(colorFormat, "31", "ERR")
l.fatalLabel = fmt.Sprintf(colorFormat, "31", "FTL")
l.traceLabel = fmt.Sprintf(colorFormat, "33", "TRC")
}
// Noticef logs a notice statement
func (l *Logger) Noticef(format string, v ...any) {
l.logger.Printf(l.infoLabel+format, v...)
}
// Warnf logs a notice statement
func (l *Logger) Warnf(format string, v ...any) {
l.logger.Printf(l.warnLabel+format, v...)
}
// Errorf logs an error statement
func (l *Logger) Errorf(format string, v ...any) {
l.logger.Printf(l.errorLabel+format, v...)
}
// Fatalf logs a fatal error
func (l *Logger) Fatalf(format string, v ...any) {
l.logger.Fatalf(l.fatalLabel+format, v...)
}
// Debugf logs a debug statement
func (l *Logger) Debugf(format string, v ...any) {
if l.debug {
l.logger.Printf(l.debugLabel+format, v...)
}
}
// Tracef logs a trace statement
func (l *Logger) Tracef(format string, v ...any) {
if l.trace {
l.logger.Printf(l.traceLabel+format, v...)
}
}
+132
View File
@@ -0,0 +1,132 @@
// Copyright 2012-2025 The NATS 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 logger
import (
"fmt"
"log"
"log/syslog"
"net/url"
"os"
"strings"
)
// SysLogger provides a system logger facility
type SysLogger struct {
writer *syslog.Writer
debug bool
trace bool
}
// SetSyslogName sets the name to use for the syslog.
// Currently used only on Windows.
func SetSyslogName(name string) {}
// GetSysLoggerTag generates the tag name for use in syslog statements. If
// the executable is linked, the name of the link will be used as the tag,
// otherwise, the name of the executable is used. "nats-server" is the default
// for the NATS server.
func GetSysLoggerTag() string {
procName := os.Args[0]
if strings.ContainsRune(procName, os.PathSeparator) {
parts := strings.FieldsFunc(procName, func(c rune) bool {
return c == os.PathSeparator
})
procName = parts[len(parts)-1]
}
return procName
}
// NewSysLogger creates a new system logger
func NewSysLogger(debug, trace bool) *SysLogger {
w, err := syslog.New(syslog.LOG_DAEMON|syslog.LOG_NOTICE, GetSysLoggerTag())
if err != nil {
log.Fatalf("error connecting to syslog: %q", err.Error())
}
return &SysLogger{
writer: w,
debug: debug,
trace: trace,
}
}
// NewRemoteSysLogger creates a new remote system logger
func NewRemoteSysLogger(fqn string, debug, trace bool) *SysLogger {
network, addr := getNetworkAndAddr(fqn)
w, err := syslog.Dial(network, addr, syslog.LOG_DEBUG, GetSysLoggerTag())
if err != nil {
log.Fatalf("error connecting to syslog: %q", err.Error())
}
return &SysLogger{
writer: w,
debug: debug,
trace: trace,
}
}
func getNetworkAndAddr(fqn string) (network, addr string) {
u, err := url.Parse(fqn)
if err != nil {
log.Fatal(err)
}
network = u.Scheme
if network == "udp" || network == "tcp" {
addr = u.Host
} else if network == "unix" {
addr = u.Path
} else {
log.Fatalf("error invalid network type: %q", u.Scheme)
}
return
}
// Noticef logs a notice statement
func (l *SysLogger) Noticef(format string, v ...any) {
l.writer.Notice(fmt.Sprintf(format, v...))
}
// Warnf logs a warning statement
func (l *SysLogger) Warnf(format string, v ...any) {
l.writer.Warning(fmt.Sprintf(format, v...))
}
// Fatalf logs a fatal error
func (l *SysLogger) Fatalf(format string, v ...any) {
l.writer.Crit(fmt.Sprintf(format, v...))
}
// Errorf logs an error statement
func (l *SysLogger) Errorf(format string, v ...any) {
l.writer.Err(fmt.Sprintf(format, v...))
}
// Debugf logs a debug statement
func (l *SysLogger) Debugf(format string, v ...any) {
if l.debug {
l.writer.Debug(fmt.Sprintf(format, v...))
}
}
// Tracef logs a trace statement
func (l *SysLogger) Tracef(format string, v ...any) {
if l.trace {
l.writer.Notice(fmt.Sprintf(format, v...))
}
}
+112
View File
@@ -0,0 +1,112 @@
// Copyright 2012-2025 The NATS 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 logger logs to the windows event log
package logger
import (
"fmt"
"os"
"strings"
"golang.org/x/sys/windows/svc/eventlog"
)
var natsEventSource = "NATS-Server"
// SetSyslogName sets the name to use for the system log event source
func SetSyslogName(name string) {
natsEventSource = name
}
// SysLogger logs to the windows event logger
type SysLogger struct {
writer *eventlog.Log
debug bool
trace bool
}
// NewSysLogger creates a log using the windows event logger
func NewSysLogger(debug, trace bool) *SysLogger {
if err := eventlog.InstallAsEventCreate(natsEventSource, eventlog.Info|eventlog.Error|eventlog.Warning); err != nil {
if !strings.Contains(err.Error(), "registry key already exists") {
panic(fmt.Sprintf("could not access event log: %v", err))
}
}
w, err := eventlog.Open(natsEventSource)
if err != nil {
panic(fmt.Sprintf("could not open event log: %v", err))
}
return &SysLogger{
writer: w,
debug: debug,
trace: trace,
}
}
// NewRemoteSysLogger creates a remote event logger
func NewRemoteSysLogger(fqn string, debug, trace bool) *SysLogger {
w, err := eventlog.OpenRemote(fqn, natsEventSource)
if err != nil {
panic(fmt.Sprintf("could not open event log: %v", err))
}
return &SysLogger{
writer: w,
debug: debug,
trace: trace,
}
}
func formatMsg(tag, format string, v ...any) string {
orig := fmt.Sprintf(format, v...)
return fmt.Sprintf("pid[%d][%s]: %s", os.Getpid(), tag, orig)
}
// Noticef logs a notice statement
func (l *SysLogger) Noticef(format string, v ...any) {
l.writer.Info(1, formatMsg("NOTICE", format, v...))
}
// Warnf logs a warning statement
func (l *SysLogger) Warnf(format string, v ...any) {
l.writer.Info(1, formatMsg("WARN", format, v...))
}
// Fatalf logs a fatal error
func (l *SysLogger) Fatalf(format string, v ...any) {
msg := formatMsg("FATAL", format, v...)
l.writer.Error(5, msg)
panic(msg)
}
// Errorf logs an error statement
func (l *SysLogger) Errorf(format string, v ...any) {
l.writer.Error(2, formatMsg("ERROR", format, v...))
}
// Debugf logs a debug statement
func (l *SysLogger) Debugf(format string, v ...any) {
if l.debug {
l.writer.Info(3, formatMsg("DEBUG", format, v...))
}
}
// Tracef logs a trace statement
func (l *SysLogger) Tracef(format string, v ...any) {
if l.trace {
l.writer.Info(4, formatMsg("TRACE", format, v...))
}
}