Merge pull request #13864 from serathius/logs

Fix inconsistent log format

CHANGELOG/CHANGELOG-3.5.md

@@ -9,6 +9,7 @@ Previous change logs can be found at [CHANGELOG-3.4](https://github.com/etcd-io/
 
 ### etcd server
 - Fix [Provide a better liveness probe for when etcd runs as a Kubernetes pod](https://github.com/etcd-io/etcd/pull/13706)
+- Fix [inconsistent log format](https://github.com/etcd-io/etcd/pull/13864)
 
 ### package `client/pkg/v3`
 

CHANGELOG/CHANGELOG-3.6.md

@@ -63,6 +63,7 @@ See [code changes](https://github.com/etcd-io/etcd/compare/v3.5.0...v3.6.0).
 - Fix [A client can cause a nil dereference in etcd by passing an invalid SortTarget](https://github.com/etcd-io/etcd/pull/13555)
 - Fix [Grant lease with negative ID can possibly cause db out of sync](https://github.com/etcd-io/etcd/pull/13676)
 - Fix [segmentation violation(SIGSEGV) error due to premature unlocking of watchableStore](https://github.com/etcd-io/etcd/pull/13505)
+- Fix [inconsistent log format](https://github.com/etcd-io/etcd/pull/13864)
 
 ### tools/benchmark
 

client/pkg/logutil/zap.go

@@ -21,6 +21,17 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
+// CreateDefaultZapLogger creates a logger with default zap configuration
+func CreateDefaultZapLogger(level zapcore.Level) (*zap.Logger, error) {
+	lcfg := DefaultZapLoggerConfig
+	lcfg.Level = zap.NewAtomicLevelAt(level)
+	c, err := lcfg.Build()
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
 // DefaultZapLoggerConfig defines default zap logger configuration.
 var DefaultZapLoggerConfig = zap.Config{
 	Level: zap.NewAtomicLevelAt(ConvertToZapLevel(DefaultLogLevel)),

client/v3/client.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+	"go.etcd.io/etcd/client/pkg/v3/logutil"
 	"go.etcd.io/etcd/client/v3/credentials"
 	"go.etcd.io/etcd/client/v3/internal/endpoint"
 	"go.etcd.io/etcd/client/v3/internal/resolver"
@@ -370,7 +371,10 @@ func newClient(cfg *Config) (*Client, error) {
 	} else if cfg.LogConfig != nil {
 		client.lg, err = cfg.LogConfig.Build()
 	} else {
-		client.lg, err = CreateDefaultZapLogger()
+		client.lg, err = logutil.CreateDefaultZapLogger(etcdClientDebugLevel())
+		if client.lg != nil {
+			client.lg = client.lg.Named("etcd-client")
+		}
 	}
 	if err != nil {
 		return nil, err

client/v3/logger.go

@@ -19,7 +19,6 @@ import (
 	"os"
 
 	"go.etcd.io/etcd/client/pkg/v3/logutil"
-	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zapgrpc"
 	"google.golang.org/grpc/grpclog"
@@ -29,10 +28,11 @@ func init() {
 	// We override grpc logger only when the environment variable is set
 	// in order to not interfere by default with user's code or other libraries.
 	if os.Getenv("ETCD_CLIENT_DEBUG") != "" {
-		lg, err := CreateDefaultZapLogger()
+		lg, err := logutil.CreateDefaultZapLogger(etcdClientDebugLevel())
 		if err != nil {
 			panic(err)
 		}
+		lg = lg.Named("etcd-client")
 		grpclog.SetLoggerV2(zapgrpc.NewLogger(lg))
 	}
 }
@@ -57,21 +57,3 @@ func etcdClientDebugLevel() zapcore.Level {
 	}
 	return l
 }
-
-// CreateDefaultZapLoggerConfig creates a logger config that is configurable using env variable:
-// ETCD_CLIENT_DEBUG= debug|info|warn|error|dpanic|panic|fatal|true (true=info)
-func CreateDefaultZapLoggerConfig() zap.Config {
-	lcfg := logutil.DefaultZapLoggerConfig
-	lcfg.Level = zap.NewAtomicLevelAt(etcdClientDebugLevel())
-	return lcfg
-}
-
-// CreateDefaultZapLogger creates a logger that is configurable using env variable:
-// ETCD_CLIENT_DEBUG= debug|info|warn|error|dpanic|panic|fatal|true (true=info)
-func CreateDefaultZapLogger() (*zap.Logger, error) {
-	c, err := CreateDefaultZapLoggerConfig().Build()
-	if err != nil {
-		return nil, err
-	}
-	return c.Named("etcd-client"), nil
-}

client/v3/snapshot/v3_snapshot.go

@@ -45,9 +45,6 @@ func hasChecksum(n int64) bool {
 // the selected node.
 // Etcd <v3.6 will return "" as version.
 func SaveWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config, dbPath string) (version string, err error) {
-	if lg == nil {
-		lg = zap.NewExample()
-	}
 	cfg.Logger = lg.Named("client")
 	if len(cfg.Endpoints) != 1 {
 		return "", fmt.Errorf("snapshot must be requested to one selected node, not multiple %v", cfg.Endpoints)

etcdctl/ctlv3/command/ep_command.go

@@ -22,6 +22,7 @@ import (
 
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+	"go.etcd.io/etcd/client/pkg/v3/logutil"
 	"go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/pkg/v3/cobrautl"
 	"go.etcd.io/etcd/pkg/v3/flags"
@@ -88,7 +89,7 @@ type epHealth struct {
 
 // epHealthCommandFunc executes the "endpoint-health" command.
 func epHealthCommandFunc(cmd *cobra.Command, args []string) {
-	lg, err := zap.NewProduction()
+	lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	if err != nil {
 		cobrautl.ExitWithError(cobrautl.ExitError, err)
 	}
@@ -260,7 +261,7 @@ func endpointsFromCluster(cmd *cobra.Command) []string {
 		cobrautl.ExitWithError(cobrautl.ExitError, err)
 	}
 	// exclude auth for not asking needless password (MemberList() doesn't need authentication)
-	lg, _ := zap.NewProduction()
+	lg, _ := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	cfg, err := clientv3.NewClientConfig(&clientv3.ConfigSpec{
 		Endpoints:        eps,
 		DialTimeout:      dt,

etcdctl/ctlv3/command/global.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/bgentry/speakeasy"
+	"go.etcd.io/etcd/client/pkg/v3/logutil"
 	"go.etcd.io/etcd/client/pkg/v3/srv"
 	"go.etcd.io/etcd/client/pkg/v3/transport"
 	"go.etcd.io/etcd/client/v3"
@@ -88,7 +89,7 @@ func (*discardValue) Set(string) error { return nil }
 func (*discardValue) Type() string     { return "" }
 
 func clientConfigFromCmd(cmd *cobra.Command) *clientv3.ConfigSpec {
-	lg, err := zap.NewProduction()
+	lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	if err != nil {
 		cobrautl.ExitWithError(cobrautl.ExitError, err)
 	}
@@ -137,7 +138,7 @@ func clientConfigFromCmd(cmd *cobra.Command) *clientv3.ConfigSpec {
 
 func mustClientCfgFromCmd(cmd *cobra.Command) *clientv3.Config {
 	cc := clientConfigFromCmd(cmd)
-	lg, _ := zap.NewProduction()
+	lg, _ := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	cfg, err := clientv3.NewClientConfig(cc, lg)
 	if err != nil {
 		cobrautl.ExitWithError(cobrautl.ExitBadArgs, err)
@@ -151,7 +152,7 @@ func mustClientFromCmd(cmd *cobra.Command) *clientv3.Client {
 }
 
 func mustClient(cc *clientv3.ConfigSpec) *clientv3.Client {
-	lg, _ := zap.NewProduction()
+	lg, _ := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	cfg, err := clientv3.NewClientConfig(cc, lg)
 	if err != nil {
 		cobrautl.ExitWithError(cobrautl.ExitBadArgs, err)

etcdctl/ctlv3/command/snapshot_command.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"github.com/spf13/cobra"
+	"go.etcd.io/etcd/client/pkg/v3/logutil"
 	snapshot "go.etcd.io/etcd/client/v3/snapshot"
 	"go.etcd.io/etcd/pkg/v3/cobrautl"
 	"go.uber.org/zap"
@@ -48,7 +49,7 @@ func snapshotSaveCommandFunc(cmd *cobra.Command, args []string) {
 		cobrautl.ExitWithError(cobrautl.ExitBadArgs, err)
 	}
 
-	lg, err := zap.NewProduction()
+	lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	if err != nil {
 		cobrautl.ExitWithError(cobrautl.ExitError, err)
 	}

etcdutl/etcdutl/common.go

@@ -15,13 +15,14 @@
 package etcdutl
 
 import (
+	"go.etcd.io/etcd/client/pkg/v3/logutil"
 	"go.etcd.io/etcd/pkg/v3/cobrautl"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
 
 func GetLogger() *zap.Logger {
-	config := zap.NewProductionConfig()
+	config := logutil.DefaultZapLoggerConfig
 	config.Encoding = "console"
 	config.EncoderConfig.EncodeTime = zapcore.RFC3339TimeEncoder
 	lg, err := config.Build()

etcdutl/snapshot/v3_snapshot.go

@@ -70,9 +70,6 @@ type Manager interface {
 
 // NewV3 returns a new snapshot Manager for v3.x snapshot.
 func NewV3(lg *zap.Logger) Manager {
-	if lg == nil {
-		lg = zap.NewExample()
-	}
 	return &v3Manager{lg: lg}
 }
 

pkg/expect/expect.go

@@ -25,6 +25,7 @@ import (
 	"strings"
 	"sync"
 	"syscall"
+	"time"
 
 	"github.com/creack/pty"
 )
@@ -36,7 +37,6 @@ type ExpectProcess struct {
 	fpty *os.File
 	wg   sync.WaitGroup
 
-	cond  *sync.Cond // for broadcasting updates are available
 	mu    sync.Mutex // protects lines and err
 	lines []string
 	count int // increment whenever new line gets added
@@ -60,7 +60,6 @@ func NewExpectWithEnv(name string, args []string, env []string) (ep *ExpectProce
 		cmd:        cmd,
 		StopSignal: syscall.SIGKILL,
 	}
-	ep.cond = sync.NewCond(&ep.mu)
 	ep.cmd.Stderr = ep.cmd.Stdout
 	ep.cmd.Stdin = nil
 
@@ -77,52 +76,56 @@ func (ep *ExpectProcess) read() {
 	defer ep.wg.Done()
 	printDebugLines := os.Getenv("EXPECT_DEBUG") != ""
 	r := bufio.NewReader(ep.fpty)
-	for ep.err == nil {
-		l, rerr := r.ReadString('\n')
+	for {
+		l, err := r.ReadString('\n')
 		ep.mu.Lock()
-		ep.err = rerr
 		if l != "" {
 			if printDebugLines {
 				fmt.Printf("%s-%d: %s", ep.cmd.Path, ep.cmd.Process.Pid, l)
 			}
 			ep.lines = append(ep.lines, l)
 			ep.count++
-			if len(ep.lines) == 1 {
-				ep.cond.Signal()
-			}
+		}
+		if err != nil {
+			ep.err = err
+			ep.mu.Unlock()
+			break
 		}
 		ep.mu.Unlock()
 	}
-	ep.cond.Signal()
 }
 
 // ExpectFunc returns the first line satisfying the function f.
 func (ep *ExpectProcess) ExpectFunc(f func(string) bool) (string, error) {
-	lastLinesBuffer := make([]string, 0)
+	i := 0
 
-	ep.mu.Lock()
 	for {
-		for len(ep.lines) == 0 && ep.err == nil {
-			ep.cond.Wait()
-		}
-		if len(ep.lines) == 0 {
-			break
-		}
-		l := ep.lines[0]
-		ep.lines = ep.lines[1:]
-		lastLinesBuffer = append(lastLinesBuffer, l)
-		if l := len(lastLinesBuffer); l > DEBUG_LINES_TAIL {
-			lastLinesBuffer = lastLinesBuffer[l-DEBUG_LINES_TAIL : l-1]
+		ep.mu.Lock()
+		for i < len(ep.lines) {
+			line := ep.lines[i]
+			i++
+			if f(line) {
+				ep.mu.Unlock()
+				return line, nil
+			}
 		}
-		if f(l) {
+		if ep.err != nil {
 			ep.mu.Unlock()
-			return l, nil
+			break
 		}
+		ep.mu.Unlock()
+		time.Sleep(time.Millisecond * 100)
+	}
+	ep.mu.Lock()
+	lastLinesIndex := len(ep.lines) - DEBUG_LINES_TAIL
+	if lastLinesIndex < 0 {
+		lastLinesIndex = 0
 	}
+	lastLines := strings.Join(ep.lines[lastLinesIndex:], "")
 	ep.mu.Unlock()
 	return "", fmt.Errorf("match not found."+
 		" Set EXPECT_DEBUG for more info Err: %v, last lines:\n%s",
-		ep.err, strings.Join(lastLinesBuffer, ""))
+		ep.err, lastLines)
 }
 
 // Expect returns the first line containing the given string.
@@ -189,3 +192,9 @@ func (ep *ExpectProcess) ProcessError() error {
 	}
 	return ep.err
 }
+
+func (ep *ExpectProcess) Lines() []string {
+	ep.mu.Lock()
+	defer ep.mu.Unlock()
+	return ep.lines
+}

pkg/netutil/netutil.go

@@ -193,12 +193,6 @@ func URLStringsEqual(ctx context.Context, lg *zap.Logger, a []string, b []string
 	if err != nil {
 		return false, err
 	}
-	if lg == nil {
-		lg, _ = zap.NewProduction()
-		if lg == nil {
-			lg = zap.NewExample()
-		}
-	}
 	return urlsEqual(ctx, lg, urlsA, urlsB)
 }
 

pkg/proxy/server.go

@@ -37,17 +37,8 @@ var (
 	defaultDialTimeout   = 3 * time.Second
 	defaultBufferSize    = 48 * 1024
 	defaultRetryInterval = 10 * time.Millisecond
-	defaultLogger        *zap.Logger
 )
 
-func init() {
-	var err error
-	defaultLogger, err = zap.NewProduction()
-	if err != nil {
-		panic(err)
-	}
-}
-
 // Server defines proxy server layer that simulates common network faults:
 // latency spikes and packet drop or corruption. The proxy overhead is very
 // small overhead (<500μs per request). Please run tests to compute actual
@@ -240,9 +231,6 @@ func NewServer(cfg ServerConfig) Server {
 	if s.retryInterval == 0 {
 		s.retryInterval = defaultRetryInterval
 	}
-	if s.lg == nil {
-		s.lg = defaultLogger
-	}
 
 	close(s.pauseAcceptc)
 	close(s.pauseTxc)

server/etcdmain/config.go

@@ -377,11 +377,10 @@ func (cfg *config) parse(arguments []string) error {
 
 func (cfg *config) configFromCmdLine() error {
 	// user-specified logger is not setup yet, use this logger during flag parsing
-	lg, err := zap.NewProduction()
+	lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
 	if err != nil {
 		return err
 	}
-
 	verKey := "ETCD_VERSION"
 	if verVal := os.Getenv(verKey); verVal != "" {
 		// unset to avoid any possible side-effect.