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

@@ -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() {

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
 }