etcd: better logging

Previously, etcd wrote to stderr in JSON format: {"level":"warn","ts":"2025-04-11T03:32:06.676527Z","caller":"embed/config.go:689","msg":"Running http and grpc server on single port. This is not recommended for production."} {"level":"warn","ts":"2025-04-11T03:32:06.676707Z","caller":"embed/config.go:689","msg":"Running http and grpc server on single port. This is not recommended for production."} {"level":"warn","ts":"2025-04-11T03:32:06.677056Z","caller":"etcdmain/etcd.go:146","msg":"failed to start etcd","error":"listen tcp 127.0.0.1:37803: bind: address already in use"} {"level":"fatal","ts":"2025-04-11T03:32:06.677104Z","caller":"etcdmain/etcd.go:204","msg":"discovery failed","error":"listen tcp 127.0.0.1:37803: bind: address already in use","stacktrace":"go.etcd.io/etcd/server/v3/etcdmain.startEtcdOrProxyV2\n\tgo.etcd.io/etcd/server/v3/etcdmain/etcd.go:204\ngo.etcd.io/etcd/server/v3/etcdmain.Main\n\tgo.etcd.io/etcd/server/v3/etcdmain/main.go:40\nmain.main\n\tgo.etcd.io/etcd/server/v3/main.go:31\nruntime.main\n\truntime/proc.go:272"} This has several drawbacks: - Not very readable. - When used in tests which start etcd themselves (for example, scheduler_perf), the output is not associated with the current test. - Contains warnings that are confusing for developers who don't know that they are harmless. Intercepting output, parsing it and reformating makes the output nicer. Instead of a mixture of JSON messages (see above) and normal test output, we now get the etcd output embedded inside the test output. We can also filter out some known harmless messages. Cleaning up more output or avoiding it in the first place might be a good next step. With `go test -v ./test/integration/scheduler_perf/dra`: === RUN TestSchedulerPerf === RUN TestSchedulerPerf/SchedulingWithResourceClaimTemplate === RUN TestSchedulerPerf/SchedulingWithResourceClaimTemplate/fast I0411 13:21:03.353458 65212 feature_gate.go:385] feature gates: {map[SchedulerQueueingHints:false]} ... I0411 13:21:10.552975 65212 cidrallocator.go:210] stopping ServiceCIDR Allocator Controller I0411 13:21:10.567327 65212 etcd.go:210] "etcd output" logger="TestSchedulerPerf/SchedulingWithResourceClaimTemplate/fast" error="accept tcp 127.0.0.1:42245: use of closed network connection" level="warn" ts="2025-04-11T13:21:10.567045+0200" caller="embed/serve.go:160" msg="stopping insecure grpc server due to error" I0411 13:21:10.567398 65212 etcd.go:210] "etcd output" logger="TestSchedulerPerf/SchedulingWithResourceClaimTemplate/fast" ts="2025-04-11T13:21:10.567198+0200" caller="embed/serve.go:162" msg="stopped insecure grpc server due to error" error="accept tcp 127.0.0.1:42245: use of closed network connection" level="warn" I0411 13:21:15.567917 65212 etcd.go:227] "etcd didn't exit in 5 seconds, killing it" logger="TestSchedulerPerf/SchedulingWithResourceClaimTemplate/fast" I0411 13:21:15.567964 65212 etcd.go:234] "etcd exited" logger="TestSchedulerPerf/SchedulingWithResourceClaimTemplate/fast" err="signal: terminated" With per-test output `go test -v ./test/integration/scheduler_perf/dra -args -use-testing-log`: === RUN TestSchedulerPerf === RUN TestSchedulerPerf/SchedulingWithResourceClaimTemplate === RUN TestSchedulerPerf/SchedulingWithResourceClaimTemplate/fast I0411 13:19:03.540497 28645 feature_gate.go:385] feature gates: {map[DynamicResourceAllocation:true]} ... I0411 13:19:10.519994 28645 cidrallocator.go:210] stopping ServiceCIDR Allocator Controller etcd.go:210: I0411 13:19:10.533131] etcd output msg="stopping insecure grpc server due to error" error="accept tcp 127.0.0.1:46637: use of closed network connection" level="warn" ts="2025-04-11T13:19:10.532900+0200" caller="embed/serve.go:160" etcd.go:210: I0411 13:19:10.533274] etcd output msg="stopped insecure grpc server due to error" error="accept tcp 127.0.0.1:46637: use of closed network connection" level="warn" ts="2025-04-11T13:19:10.533022+0200" caller="embed/serve.go:162" etcd.go:227: I0411 13:19:15.533715] etcd didn't exit in 5 seconds, killing it etcd.go:234: I0411 13:19:15.533803] etcd exited err="signal: terminated"
2026-07-18 04:54:54 +00:00 · 2025-04-11 13:22:44 +02:00
parent 5ec7b65318
commit f74424bd7c
5 changed files with 100 additions and 55 deletions
--- a/test/integration/apimachinery/watchlist_test.go
+++ b/test/integration/apimachinery/watchlist_test.go
@@ -33,16 +33,17 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2/ktesting"
 	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
 	"k8s.io/kubernetes/test/integration/framework"
 	"k8s.io/utils/ptr"
 )

 func TestReflectorWatchListFallback(t *testing.T) {
-	ctx := context.TODO()
+	logger, ctx := ktesting.NewTestContext(t)

 	t.Log("Starting etcd that will be used by two different instances of kube-apiserver")
-	etcdURL, etcdTearDownFn, err := framework.RunCustomEtcd("etcd_watchlist", []string{"--experimental-watch-progress-notify-interval", "1s"}, nil)
+	etcdURL, etcdTearDownFn, err := framework.RunCustomEtcd(logger, "etcd_watchlist", []string{"--experimental-watch-progress-notify-interval", "1s"})
 	require.NoError(t, err)
 	defer etcdTearDownFn()
 	etcdOptions := framework.DefaultEtcdOptions()
--- a/test/integration/apiserver/apply/apply_crd_test.go
+++ b/test/integration/apiserver/apply/apply_crd_test.go
@@ -31,10 +31,6 @@ import (

 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"

-	"go.etcd.io/etcd/client/pkg/v3/transport"
-	clientv3 "go.etcd.io/etcd/client/v3"
-	"google.golang.org/grpc"
-
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
@@ -416,35 +412,13 @@ func findCRDCondition(crd *apiextensionsv1.CustomResourceDefinition, conditionTy
 // apply falls back to non-schema behavior
 func TestApplyCRDUnhandledSchema(t *testing.T) {
 	storageConfig := framework.SharedEtcd()
-	tlsInfo := transport.TLSInfo{
-		CertFile:      storageConfig.Transport.CertFile,
-		KeyFile:       storageConfig.Transport.KeyFile,
-		TrustedCAFile: storageConfig.Transport.TrustedCAFile,
-	}
-	tlsConfig, err := tlsInfo.ClientConfig()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcdConfig := clientv3.Config{
-		Endpoints:   storageConfig.Transport.ServerList,
-		DialTimeout: 20 * time.Second,
-		DialOptions: []grpc.DialOption{
-			grpc.WithBlock(), // block until the underlying connection is up
-		},
-		TLS: tlsConfig,
-	}
-	etcdclient, err := clientv3.New(etcdConfig)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer etcdclient.Close()
-
 	server, err := apiservertesting.StartTestServer(t, apiservertesting.NewDefaultTestServerOptions(), nil, storageConfig)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer server.TearDownFn()
 	config := server.ClientConfig
+	etcdclient := server.EtcdClient

 	apiExtensionClient, err := clientset.NewForConfig(config)
 	if err != nil {
--- a/test/integration/apiserver/watchcache_test.go
+++ b/test/integration/apiserver/watchcache_test.go
@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	"k8s.io/kubernetes/pkg/controlplane"
 	"k8s.io/kubernetes/pkg/controlplane/reconcilers"
@@ -38,12 +39,12 @@ import (
 // with one of them containing events and the other all other objects.
 func multiEtcdSetup(ctx context.Context, t *testing.T) (clientset.Interface, framework.TearDownFunc) {
 	etcdArgs := []string{"--experimental-watch-progress-notify-interval", "1s"}
-	etcd0URL, stopEtcd0, err := framework.RunCustomEtcd("etcd_watchcache0", etcdArgs, nil)
+	etcd0URL, stopEtcd0, err := framework.RunCustomEtcd(klog.FromContext(ctx).WithName("etcd0"), "etcd_watchcache0", etcdArgs)
 	if err != nil {
 		t.Fatalf("Couldn't start etcd: %v", err)
 	}

-	etcd1URL, stopEtcd1, err := framework.RunCustomEtcd("etcd_watchcache1", etcdArgs, nil)
+	etcd1URL, stopEtcd1, err := framework.RunCustomEtcd(klog.FromContext(ctx).WithName("etcd1"), "etcd_watchcache1", etcdArgs)
 	if err != nil {
 		t.Fatalf("Couldn't start etcd: %v", err)
 	}
--- a/test/integration/framework/etcd.go
+++ b/test/integration/framework/etcd.go
@@ -17,7 +17,10 @@ limitations under the License.
 package framework

 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -35,6 +38,7 @@ import (
 	"go.uber.org/goleak"
 	"google.golang.org/grpc/grpclog"

+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/util/env"
 )
@@ -54,7 +58,7 @@ func getEtcdPath() (string, error) {

 // startEtcd executes an etcd instance. The returned function will signal the
 // etcd process and wait for it to exit.
-func startEtcd(output io.Writer, forceCreate bool) (func(), error) {
+func startEtcd(logger klog.Logger, forceCreate bool) (func(), error) {
 	if !forceCreate {
 		etcdURL := env.GetEnvAsStringOrFallback("KUBE_INTEGRATION_ETCD_URL", "http://127.0.0.1:2379")
 		conn, err := net.Dial("tcp", strings.TrimPrefix(etcdURL, "http://"))
@@ -63,10 +67,10 @@ func startEtcd(output io.Writer, forceCreate bool) (func(), error) {
 			_ = conn.Close()
 			return func() {}, nil
 		}
-		klog.V(1).Infof("could not connect to etcd: %v", err)
+		logger.V(1).Info("could not connect to etcd", "err", err)
 	}

-	currentURL, stop, err := RunCustomEtcd("integration_test_etcd_data", nil, output)
+	currentURL, stop, err := RunCustomEtcd(logger, "integration_test_etcd_data", nil)
 	if err != nil {
 		return nil, err
 	}
@@ -86,8 +90,15 @@ func init() {
 	grpclog.SetLoggerV2(grpclog.NewLoggerV2(io.Discard, io.Discard, os.Stderr))
 }

+// hiddenEtcdMessages contains all messages that occur due to the way we run
+// etcd and don't need to be logged in all tests.
+var hiddenEtcdMessages = sets.New(
+	"simple token is not cryptographically signed",
+	"Running http and grpc server on single port. This is not recommended for production.",
+)
+
 // RunCustomEtcd starts a custom etcd instance for test purposes.
-func RunCustomEtcd(dataDir string, customFlags []string, output io.Writer) (url string, stopFn func(), err error) {
+func RunCustomEtcd(logger klog.Logger, dataDir string, customFlags []string) (url string, stopFn func(), err error) {
 	// TODO: Check for valid etcd version.
 	etcdPath, err := getEtcdPath()
 	if err != nil {
@@ -101,7 +112,7 @@ func RunCustomEtcd(dataDir string, customFlags []string, output io.Writer) (url
 	etcdSocketPath := path.Join(etcdDataDir, "etcd.sock")
 	customURL := "unix://" + etcdSocketPath

-	klog.V(2).InfoS("starting etcd", "url", customURL, "dataDir", etcdDataDir)
+	logger.V(2).Info("starting etcd", "url", customURL, "dataDir", etcdDataDir)
 	ctx, cancel := context.WithCancel(context.Background())
 	args := []string{
 		"--data-dir",
@@ -127,37 +138,89 @@ func RunCustomEtcd(dataDir string, customFlags []string, output io.Writer) (url
 	}
 	args = append(args, customFlags...)
 	cmd := exec.CommandContext(ctx, etcdPath, args...)
-	if output == nil {
-		cmd.Stdout = os.Stdout
-		cmd.Stderr = os.Stderr
-	} else {
-		cmd.Stdout = output
-		cmd.Stderr = output
+
+	// Always filter the etcd output. This allows us to get rid of known harmless messages
+	// and includes the etcd log messages in the normal test log.
+	reader, err := cmd.StderrPipe()
+	if err != nil {
+		cancel()
+		return "", nil, fmt.Errorf("prepared etcd command's stderr pipe: %w", err)
 	}
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		buffer := make([]byte, 100*1024)
+		for {
+			n, err := reader.Read(buffer)
+			// Unfortunately in practice we get an untyped errors.errorString wrapped in an os.Path error,
+			// so we have to fall back to text matching.
+			if errors.Is(err, io.EOF) || err != nil && err.Error() == "read |0: file already closed" {
+				return
+			}
+			if err != nil {
+				logger.Error(err, "read etcd output")
+				return
+			}
+			if n == 0 {
+				continue
+			}
+			dec := json.NewDecoder(bytes.NewBuffer(buffer[0:n]))
+			for {
+				// Try to parse as JSON object. If we get anything that isn't JSON or an object, we just dump the remaining output.
+				var msg map[string]any
+				err := dec.Decode(&msg)
+				if err == io.EOF {
+					// all done
+					break
+				}
+				if err != nil {
+					offset := int(dec.InputOffset())
+					if offset < n {
+						logger.Info("etcd output", "msg", string(buffer[0:n]))
+					}
+					continue
+				}
+
+				// Skip harmless messages.
+				msgStr, _ := msg["msg"].(string)
+				if hiddenEtcdMessages.Has(msgStr) {
+					continue
+				}
+
+				// Log it using structured logging.
+				kvs := make([]any, 0, 2*len(msg))
+				for key, value := range msg {
+					kvs = append(kvs, key, value)
+				}
+				logger.Info("etcd output", kvs...)
+			}
+		}
+	}()
+
 	stop := func() {
 		// try to exit etcd gracefully
 		defer cancel()
 		cmd.Process.Signal(syscall.SIGTERM)
-		var wg sync.WaitGroup
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
 			select {
 			case <-ctx.Done():
-				klog.V(6).InfoS("etcd exited gracefully, context cancelled")
+				logger.V(6).Info("etcd exited gracefully, context cancelled")
 			case <-time.After(5 * time.Second):
-				klog.Infof("etcd didn't exit in 5 seconds, killing it")
+				logger.Info("etcd didn't exit in 5 seconds, killing it")
 				cancel()
 			}
 		}()
 		err := cmd.Wait()
-		klog.V(2).InfoS("etcd exited", "err", err)
+		logger.V(2).Info("etcd exited", "err", err)
 		// Tell goroutine that we are done.
 		cancel()
 		wg.Wait()
 		err = os.RemoveAll(etcdDataDir)
 		if err != nil {
-			klog.Warningf("error during etcd cleanup: %v", err)
+			logger.Info("Warning: error during etcd cleanup", "err", err)
 		}
 	}

@@ -217,7 +280,7 @@ func EtcdMain(tests func() int) {
 		goleak.IgnoreTopFunction("github.com/moby/spdystream.(*Connection).shutdown"),
 	)

-	stop, err := startEtcd(nil, false)
+	stop, err := startEtcd(klog.Background(), false)
 	if err != nil {
 		klog.Fatalf("cannot run integration tests: unable to start etcd: %v", err)
 	}
@@ -240,15 +303,16 @@ func GetEtcdURL() string {

 // StartEtcd starts an etcd instance inside a test. It will abort the test if
 // startup fails and clean up after the test automatically. Stdout and stderr
-// of the etcd binary go to the provided writer.
+// of the etcd binary and all log output of the etcd helper code go to the
+// provided logger.
 //
 // In contrast to EtcdMain, StartEtcd will not do automatic leak checking.
 // Tests can decide if and where they want to do that.
 //
 // Starting etcd multiple times per test run instead of once with EtcdMain
 // provides better separation between different tests.
-func StartEtcd(tb testing.TB, etcdOutput io.Writer, forceCreate bool) {
-	stop, err := startEtcd(etcdOutput, forceCreate)
+func StartEtcd(logger klog.Logger, tb testing.TB, forceCreate bool) {
+	stop, err := startEtcd(logger, forceCreate)
 	if err != nil {
 		tb.Fatalf("unable to start etcd: %v", err)
 	}
--- a/test/integration/scheduler_perf/scheduler_perf.go
+++ b/test/integration/scheduler_perf/scheduler_perf.go
@@ -1020,7 +1020,7 @@ func initTestOutput(tb testing.TB) io.Writer {

 var specialFilenameChars = regexp.MustCompile(`[^a-zA-Z0-9-_]`)

-func setupTestCase(t testing.TB, tc *testCase, featureGates map[featuregate.Feature]bool, output io.Writer, outOfTreePluginRegistry frameworkruntime.Registry) (informers.SharedInformerFactory, ktesting.TContext) {
+func setupTestCase(t testing.TB, tc *testCase, featureGates map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (informers.SharedInformerFactory, ktesting.TContext) {
 	tCtx := ktesting.Init(t, initoption.PerTestOutput(UseTestingLog))
 	artifacts, doArtifacts := os.LookupEnv("ARTIFACTS")
 	if !UseTestingLog && doArtifacts {
@@ -1085,7 +1085,12 @@ func setupTestCase(t testing.TB, tc *testCase, featureGates map[featuregate.Feat

 	// Now that we are ready to run, start
 	// a brand new etcd.
-	framework.StartEtcd(t, output, true)
+	logger := tCtx.Logger()
+	if !UseTestingLog {
+		// Associate output going to the global log with the current test.
+		logger = logger.WithName(tCtx.Name())
+	}
+	framework.StartEtcd(logger, t, true)

 	for feature, flag := range featureGates {
 		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, feature, flag)
@@ -1190,7 +1195,7 @@ func RunBenchmarkPerfScheduling(b *testing.B, configFile string, topicName strin
 					fixJSONOutput(b)

 					featureGates := featureGatesMerge(tc.FeatureGates, w.FeatureGates)
-					informerFactory, tCtx := setupTestCase(b, tc, featureGates, output, outOfTreePluginRegistry)
+					informerFactory, tCtx := setupTestCase(b, tc, featureGates, outOfTreePluginRegistry)

 					// TODO(#93795): make sure each workload within a test case has a unique
 					// name? The name is used to identify the stats in benchmark reports.
@@ -1302,7 +1307,7 @@ func RunIntegrationPerfScheduling(t *testing.T, configFile string) {
 						t.Skipf("disabled by label filter %q", TestSchedulingLabelFilter)
 					}
 					featureGates := featureGatesMerge(tc.FeatureGates, w.FeatureGates)
-					informerFactory, tCtx := setupTestCase(t, tc, featureGates, nil, nil)
+					informerFactory, tCtx := setupTestCase(t, tc, featureGates, nil)
 					err := w.isValid(tc.MetricsCollectorConfig)
 					if err != nil {
 						t.Fatalf("workload %s is not valid: %v", w.Name, err)