Add nonblock logsink for non-blocking log writes to stderr/stdout/file

client/pkg/logutil/nonblock/nonblock.go

@@ -0,0 +1,157 @@
+package nonblock
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+	"os"
+	"strconv"
+	"sync"
+	"syscall"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+const Scheme = "nonblock"
+const DefaultMode = os.FileMode(0666)
+
+var DefaultTimeout = time.Millisecond
+
+func init() {
+	if err := zap.RegisterSink(Scheme, NewSink); err != nil {
+		panic("Failed to register zap sink for scheme " + Scheme + ": " + err.Error())
+	}
+}
+
+type nonblock struct {
+	sync.Mutex
+
+	file     *os.File
+	timeout  time.Duration
+	errCount uint64
+	warn     bool
+	must     bool
+}
+
+func NewSink(u *url.URL) (zap.Sink, error) {
+	if u == nil || u.Scheme != Scheme {
+		return nil, errors.New("invalid url or scheme")
+	}
+
+	n := &nonblock{timeout: DefaultTimeout}
+
+	if timeout := u.Query().Get("timeout"); timeout != "" {
+		dur, err := time.ParseDuration(timeout)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse timeout parameter: %w", err)
+		}
+		if dur < 1 {
+			return nil, errors.New("timeout must be greater than zero")
+		}
+		n.timeout = dur
+	}
+
+	if warn := u.Query().Get("warn"); warn != "" {
+		b, err := strconv.ParseBool(warn)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse warn parameter: %w", err)
+		}
+		n.warn = b
+	}
+
+	if must := u.Query().Get("must"); must != "" {
+		b, err := strconv.ParseBool(must)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse must parameter: %w", err)
+		}
+		n.must = b
+	}
+
+	fileMode := DefaultMode
+	if mode := u.Query().Get("mode"); mode != "" {
+		u, err := strconv.ParseUint(mode, 0, 32)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse mode parameter: %w", err)
+		}
+		fileMode = os.FileMode(u)
+	}
+
+	var err error
+	switch u.Opaque {
+	case "stdout":
+		n.file, err = dup(os.Stdout)
+	case "stderr":
+		n.file, err = dup(os.Stderr)
+	default:
+		n.file, err = os.OpenFile(u.Opaque, os.O_WRONLY|os.O_APPEND|os.O_CREATE|syscall.O_NONBLOCK, fileMode)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return n, nil
+}
+
+func (n *nonblock) Write(p []byte) (int, error) {
+	n.Lock()
+	defer n.Unlock()
+
+	if err := n.file.SetWriteDeadline(time.Now().Add(n.timeout)); err != nil && n.must {
+		n.errCount++
+		return len(p), nil
+	}
+	i, err := n.file.Write(p)
+	if errors.Is(err, os.ErrDeadlineExceeded) {
+		n.errCount++
+		return len(p), nil
+	}
+	if err == nil && n.errCount != 0 {
+		if n.warn {
+			fmt.Fprintf(n.file, `\n{"level":"error","msg":"Sink dropped %d writes prior to the last entry"}\n`, n.errCount)
+		}
+		n.errCount = 0
+	}
+	return i, err
+}
+
+func (n *nonblock) Sync() error {
+	n.Lock()
+	defer n.Unlock()
+
+	if n.errCount == 0 {
+		return n.file.Sync()
+	}
+	return nil
+}
+
+func (n *nonblock) Close() error {
+	n.Lock()
+	defer n.Unlock()
+
+	return n.file.Close()
+}
+
+func dup(f *os.File) (*os.File, error) {
+	rc, err := f.SyscallConn()
+	if err != nil {
+		return nil, err
+	}
+	var duperr error
+	var newfd int
+	if err := rc.Control(func(oldfd uintptr) { newfd, err = syscall.Dup(int(oldfd)) }); err != nil {
+		return nil, err
+	}
+	if duperr != nil {
+		return nil, err
+	}
+	if err := syscall.SetNonblock(newfd, true); err != nil {
+		return nil, err
+	}
+	f = os.NewFile(uintptr(newfd), f.Name())
+	if f == nil {
+		return nil, errors.New("failed to reopen file")
+	}
+	return f, nil
+}

server/embed/config_logging.go

@@ -31,6 +31,7 @@ import (
 	"gopkg.in/natefinch/lumberjack.v2"
 
 	"go.etcd.io/etcd/client/pkg/v3/logutil"
+	_ "go.etcd.io/etcd/client/pkg/v3/logutil/nonblock"
 )
 
 // GetLogger returns the logger.