diff --git a/test/integration/apimachinery/watchlist_test.go b/test/integration/apimachinery/watchlist_test.go index 2f73d6b4d7b..442fd73e7b1 100644 --- a/test/integration/apimachinery/watchlist_test.go +++ b/test/integration/apimachinery/watchlist_test.go @@ -35,15 +35,16 @@ import ( clientfeaturestesting "k8s.io/client-go/features/testing" "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" ) 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() diff --git a/test/integration/apiserver/apply/apply_crd_test.go b/test/integration/apiserver/apply/apply_crd_test.go index 4ac048e7974..e363cfd7d47 100644 --- 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 { diff --git a/test/integration/apiserver/watchcache_test.go b/test/integration/apiserver/watchcache_test.go index 3c280119fb2..f93a85d4cd4 100644 --- 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) } diff --git a/test/integration/framework/etcd.go b/test/integration/framework/etcd.go index f381292782c..f7261be3d21 100644 --- 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) } diff --git a/test/integration/scheduler_perf/scheduler_perf.go b/test/integration/scheduler_perf/scheduler_perf.go index 45edd7ced63..0db82c85393 100644 --- a/test/integration/scheduler_perf/scheduler_perf.go +++ b/test/integration/scheduler_perf/scheduler_perf.go @@ -1021,7 +1021,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 { @@ -1086,7 +1086,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) // We need to set emulation version for QueueingHints feature gate, which is locked at 1.34. featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.33")) @@ -1193,7 +1198,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. @@ -1305,7 +1310,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)