From d0141640fe261dc5370e1d631c7bfefe0e80e79a Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 6 Nov 2024 15:15:24 +0100 Subject: [PATCH 1/3] ktesting: print info and progress to /dev/tty The "received interrupt signal" is useful also when running with "go test" without -v because it shows that the shutdown has started. But more important is that a progress report gets shown because that feature is useful in particular when "go test" produces no output while it runs. --- test/utils/ktesting/signals.go | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/test/utils/ktesting/signals.go b/test/utils/ktesting/signals.go index f498c46259c..605045a1fe9 100644 --- a/test/utils/ktesting/signals.go +++ b/test/utils/ktesting/signals.go @@ -29,11 +29,7 @@ import ( var ( // defaultProgressReporter is inactive until init is called. - defaultProgressReporter = &progressReporter{ - // os.Stderr gets redirected by "go test". "go test -v" has to be - // used to see the output while a test runs. - out: os.Stderr, - } + defaultProgressReporter = &progressReporter{} ) const ginkgoSpecContextKey = "GINKGO_SPEC_CONTEXT" @@ -57,6 +53,7 @@ type progressReporter struct { reporterCounter int64 reporters map[int64]func() string out io.Writer + closeOut func() error } var _ ginkgoReporter = &progressReporter{} @@ -88,6 +85,24 @@ func (p *progressReporter) init(tb TB) context.Context { return p.interruptCtx } + // Might have been set for testing purposes. + if p.out == nil { + // os.Stderr gets redirected by "go test". "go test -v" has to be + // used to see that output while a test runs. + // + // Opening /dev/tty during init avoids the redirection. + // May fail, depending on the OS, in which case + // os.Stderr is used. + if console, err := os.OpenFile("/dev/tty", os.O_RDWR|os.O_APPEND, 0); err == nil { + p.out = console + p.closeOut = console.Close + + } else { + p.out = os.Stdout + p.closeOut = nil + } + } + p.signalCtx, p.signalCancel = signal.NotifyContext(context.Background(), os.Interrupt) cancelCtx, cancel := context.WithCancelCause(context.Background()) p.wg.Go(func() { @@ -126,6 +141,12 @@ func (p *progressReporter) finalize() { p.signalCancel() p.wg.Wait() + + // Now that all goroutines are stopped, we can clean up some more. + if p.closeOut != nil { + _ = p.closeOut() + p.out = nil + } } // AttachProgressReporter implements Gomega's contextWithAttachProgressReporter. From 8946e86e3a9ee33c3cff30e74291305dac745fcf Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Thu, 29 Jan 2026 08:31:54 +0100 Subject: [PATCH 2/3] ktesting: properly clean up When cleaning up the progress channel properly (stop signal delivery, closing the channel), the loop dumping progress reports no longer needs to check for the separate shutdown context. Instead, it can distinguish between "signal received" and "channel closed". The signal context was getting cleanup by canceling it, but a channel is better because it avoids the slightly misleading "received interrupt signal" cancellation when the test was only shutting down. --- test/utils/ktesting/signals.go | 34 +++++++++++++++------------------- 1 file changed, 15 insertions(+), 19 deletions(-) diff --git a/test/utils/ktesting/signals.go b/test/utils/ktesting/signals.go index 605045a1fe9..8f3fe3cca73 100644 --- a/test/utils/ktesting/signals.go +++ b/test/utils/ktesting/signals.go @@ -45,7 +45,7 @@ type progressReporter struct { usageCount int64 wg sync.WaitGroup signalCtx, interruptCtx context.Context - signalCancel func() + signalChannel chan os.Signal progressChannel chan os.Signal // reportMutex protects report creation and settings. @@ -103,11 +103,13 @@ func (p *progressReporter) init(tb TB) context.Context { } } - p.signalCtx, p.signalCancel = signal.NotifyContext(context.Background(), os.Interrupt) - cancelCtx, cancel := context.WithCancelCause(context.Background()) + p.signalChannel = make(chan os.Signal) + signal.Notify(p.signalChannel, os.Interrupt) p.wg.Go(func() { - <-p.signalCtx.Done() - cancel(errors.New("received interrupt signal")) + _, ok := <-p.signalChannel + if ok { + interrupted(errors.New("received interrupt signal")) + } }) // This reimplements the contract between Ginkgo and Gomega for progress reporting. @@ -139,7 +141,10 @@ func (p *progressReporter) finalize() { return } - p.signalCancel() + signal.Stop(p.signalChannel) + close(p.signalChannel) + signal.Stop(p.progressChannel) + close(p.progressChannel) p.wg.Wait() // Now that all goroutines are stopped, we can clean up some more. @@ -175,21 +180,12 @@ func (p *progressReporter) detachProgressReporter(id int64) { func (p *progressReporter) run() { for { - select { - case <-p.interruptCtx.Done(): - // Maybe do one last progress report? - // - // This is primarily for unit testing of ktesting itself, - // in a normal test we don't care anymore. - select { - case <-p.progressChannel: - p.dumpProgress() - default: - } + _, ok := <-p.progressChannel + if !ok { + // Shut down. return - case <-p.progressChannel: - p.dumpProgress() } + p.dumpProgress() } } From 36bcd43fca2fb24f110ad1b26f3ac1cd6e20f003 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 6 Nov 2024 11:04:45 +0100 Subject: [PATCH 3/3] ktesting: abort entire test suite on SIGINT When aborting an integration test with CTRL-C while it runs, the current test fails and etcd exits. But additional tests were still being started and the failed slowly because they couldn't connect to etcd. It's better to fail additional tests in ktesting.Init when the test run has already been interrupted. While at it, also make it a bit more obvious that testing was interrupted by logging it and update one comment about this and clean up the naming of contexts in the code. Example: $ go test -v ./test/integration/quota ... I1106 11:42:48.857162 147325 etcd.go:416] "Not using watch cache" resource="events.events.k8s.io" I1106 11:42:48.857204 147325 handler.go:286] Adding GroupVersion events.k8s.io v1 to ResourceManager W1106 11:42:48.857209 147325 genericapiserver.go:765] Skipping API events.k8s.io/v1beta1 because it has no resources. ^C INFO: canceling test context: received interrupt signal {"level":"warn","ts":"2024-11-06T11:42:48.984676+0100","caller":"embed/serve.go:160","msg":"stopping insecure grpc server due to error","error":"accept tcp 127.0.0.1:44177: use of closed network connection"} ... I1106 11:42:50.042430 147325 handler.go:142] kube-apiserver: GET "/apis/rbac.authorization.k8s.io/v1/clusterroles" satisfied by gorestful with webservice /apis/rbac.authorization.k8s.io/v1 test_server.go:241: timed out waiting for the condition --- FAIL: TestQuota (11.45s) === RUN TestQuotaLimitedResourceDenial quota_test.go:292: testing has been interrupted: received interrupt signal --- FAIL: TestQuotaLimitedResourceDenial (0.00s) === RUN TestQuotaLimitService quota_test.go:418: testing has been interrupted: received interrupt signal --- FAIL: TestQuotaLimitService (0.00s) FAIL --- test/utils/ktesting/signals.go | 35 +++++++++++++++++++------ test/utils/ktesting/stepcontext_test.go | 32 ++++++++++++++++------ 2 files changed, 51 insertions(+), 16 deletions(-) diff --git a/test/utils/ktesting/signals.go b/test/utils/ktesting/signals.go index 8f3fe3cca73..4edef18bd96 100644 --- a/test/utils/ktesting/signals.go +++ b/test/utils/ktesting/signals.go @@ -19,6 +19,7 @@ package ktesting import ( "context" "errors" + "fmt" "io" "os" "os/signal" @@ -30,6 +31,13 @@ import ( var ( // defaultProgressReporter is inactive until init is called. defaultProgressReporter = &progressReporter{} + + // interruptCtx tracks whether the process got interrupted via SIGINT. + // In that case, interrupted gets called to cancel interruptCtx with + // a suitable message. + // + // This gets set up once per process and never gets reset. + interruptCtx, interrupted = context.WithCancelCause(context.Background()) ) const ginkgoSpecContextKey = "GINKGO_SPEC_CONTEXT" @@ -42,11 +50,11 @@ type progressReporter struct { // initMutex protects initialization and finalization of the reporter. initMutex sync.Mutex - usageCount int64 - wg sync.WaitGroup - signalCtx, interruptCtx context.Context - signalChannel chan os.Signal - progressChannel chan os.Signal + usageCount int64 + wg sync.WaitGroup + testCtx context.Context + signalChannel chan os.Signal + progressChannel chan os.Signal // reportMutex protects report creation and settings. reportMutex sync.Mutex @@ -75,6 +83,16 @@ func (p *progressReporter) init(tb TB) context.Context { return context.Background() } + tb.Helper() + + // If already interrupted, then don't start the new test. + // This is necessary because normally CTRL-C would exit + // the entire process immediately. Now we keep running + // to clean up. + if interruptCtx.Err() != nil { + tb.Fatalf("testing has been interrupted: %v", context.Cause(interruptCtx)) + } + p.initMutex.Lock() defer p.initMutex.Unlock() @@ -82,7 +100,7 @@ func (p *progressReporter) init(tb TB) context.Context { tb.Cleanup(p.finalize) if p.usageCount > 1 { // Was already initialized. - return p.interruptCtx + return p.testCtx } // Might have been set for testing purposes. @@ -108,6 +126,7 @@ func (p *progressReporter) init(tb TB) context.Context { p.wg.Go(func() { _, ok := <-p.signalChannel if ok { + _, _ = fmt.Fprint(p.out, "\n\nINFO: canceling test context: received interrupt signal\n\n") interrupted(errors.New("received interrupt signal")) } }) @@ -118,7 +137,7 @@ func (p *progressReporter) init(tb TB) context.Context { // nolint:staticcheck // It complains about using a plain string. This can only be fixed // by Ginkgo and Gomega formalizing this interface and define a type (somewhere... // probably cannot be in either Ginkgo or Gomega). - p.interruptCtx = context.WithValue(cancelCtx, ginkgoSpecContextKey, defaultProgressReporter) + p.testCtx = context.WithValue(interruptCtx, ginkgoSpecContextKey, defaultProgressReporter) p.progressChannel = make(chan os.Signal, 1) // progressSignals will be empty on Windows. @@ -128,7 +147,7 @@ func (p *progressReporter) init(tb TB) context.Context { p.wg.Go(p.run) - return p.interruptCtx + return p.testCtx } func (p *progressReporter) finalize() { diff --git a/test/utils/ktesting/stepcontext_test.go b/test/utils/ktesting/stepcontext_test.go index c8f3c143011..314fa485daf 100644 --- a/test/utils/ktesting/stepcontext_test.go +++ b/test/utils/ktesting/stepcontext_test.go @@ -17,9 +17,11 @@ limitations under the License. package ktesting import ( + "context" "io" "os" "testing" + "time" "github.com/onsi/gomega" "go.uber.org/goleak" @@ -73,15 +75,20 @@ func TestStepContext(t *testing.T) { } func TestProgressReport(t *testing.T) { + oldOut := defaultProgressReporter.out + out := newOutputStream() + defaultProgressReporter.out = out t.Cleanup(func() { goleak.VerifyNone(t) - }) - - oldOut := defaultProgressReporter.out - reportStream := newOutputStream() - defaultProgressReporter.out = reportStream - t.Cleanup(func() { defaultProgressReporter.out = oldOut + + // If we get here, the defaultProgressReporter is not active anymore, + // but the interrupt context should still be canceled. + gomega.NewGomegaWithT(t).Expect(defaultProgressReporter.usageCount).To(gomega.Equal(int64(0)), "usage count") + gomega.NewGomegaWithT(t).Expect(context.Cause(interruptCtx)).To(gomega.MatchError(gomega.Equal("received interrupt signal")), "interrupted persistently") + + // Reset for next test. + interruptCtx, interrupted = context.WithCancelCause(context.Background()) }) // This must use a real testing.T, otherwise Init doesn't initialize signal handling. @@ -93,11 +100,21 @@ func TestProgressReport(t *testing.T) { // Trigger report and wait for it. defaultProgressReporter.progressChannel <- os.Interrupt - report := <-reportStream.stream + report := <-out.stream tCtx.Expect(report).To(gomega.Equal(`You requested a progress report. step: hello world `), "report") + + gomega.NewGomegaWithT(t).Expect(context.Cause(interruptCtx)).To(gomega.Succeed(), "not interrupted yet") + defaultProgressReporter.signalChannel <- os.Interrupt + message := <-out.stream + tCtx.Expect(message).To(gomega.Equal(` + +INFO: canceling test context: received interrupt signal + +`)) + gomega.NewGomegaWithT(t).Eventually(func() error { return context.Cause(tCtx) }).WithTimeout(30*time.Second).To(gomega.MatchError(gomega.Equal("received interrupt signal")), "interrupted") } // outputStream forwards exactly one Write call to a stream. @@ -116,6 +133,5 @@ func newOutputStream() *outputStream { func (s *outputStream) Write(buf []byte) (int, error) { s.stream <- string(buf) - close(s.stream) return len(buf), nil }