Merge pull request #37 from deads2k/thread-context

add context down through testing

Author: openshift-merge-bot[bot]

pkg/cmd/cmdrun/runsuite.go

@@ -1,8 +1,12 @@
 package cmdrun
 
 import (
+	"context"
 	"fmt"
 	"os"
+	"os/signal"
+	"syscall"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
@@ -31,6 +35,32 @@ func NewRunSuiteCommand(registry *extension.Registry) *cobra.Command {
 			"development use. Orchestration parameters, scheduling, isolation, etc are not obeyed, and Ginkgo tests are executed serially.",
 		SilenceUsage: true,
 		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx, cancelCause := context.WithCancelCause(context.Background())
+			defer cancelCause(errors.New("exiting"))
+
+			abortCh := make(chan os.Signal, 2)
+			go func() {
+				<-abortCh
+				fmt.Fprintf(os.Stderr, "Interrupted, terminating tests")
+				cancelCause(errors.New("interrupt received"))
+
+				select {
+				case sig := <-abortCh:
+					fmt.Fprintf(os.Stderr, "Interrupted twice, exiting (%s)", sig)
+					switch sig {
+					case syscall.SIGINT:
+						os.Exit(130)
+					default:
+						os.Exit(130) // if we were interrupted, never return zero.
+					}
+
+				case <-time.After(30 * time.Minute): // allow time for cleanup.  If we finish before this, we'll exit
+					fmt.Fprintf(os.Stderr, "Timed out during cleanup, exiting")
+					os.Exit(130) // if we were interrupted, never return zero.
+				}
+			}()
+			signal.Notify(abortCh, syscall.SIGINT, syscall.SIGTERM)
+
 			ext := registry.Get(opts.componentFlags.Component)
 			if ext == nil {
 				return fmt.Errorf("component not found: %s", opts.componentFlags.Component)
@@ -72,7 +102,7 @@ func NewRunSuiteCommand(registry *extension.Registry) *cobra.Command {
 				return errors.Wrap(err, "couldn't filter specs")
 			}
 
-			return specs.Run(compositeWriter, opts.concurrencyFlags.MaxConcurency)
+			return specs.Run(ctx, compositeWriter, opts.concurrencyFlags.MaxConcurency)
 		},
 	}
 	opts.componentFlags.BindFlags(cmd.Flags())

pkg/cmd/cmdrun/runtest.go

@@ -2,8 +2,13 @@ package cmdrun
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"os"
+	"errors"
+	"os/signal"
+	"syscall"
+	"time"
 
 	"github.com/spf13/cobra"
 
@@ -30,6 +35,32 @@ func NewRunTestCommand(registry *extension.Registry) *cobra.Command {
 		Short:        "Runs tests by name",
 		SilenceUsage: true,
 		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx, cancelCause := context.WithCancelCause(context.Background())
+			defer cancelCause(errors.New("exiting"))
+
+			abortCh := make(chan os.Signal, 2)
+			go func() {
+				<-abortCh
+				fmt.Fprintf(os.Stderr, "Interrupted, terminating tests")
+				cancelCause(errors.New("interrupt received"))
+
+				select {
+				case sig := <-abortCh:
+					fmt.Fprintf(os.Stderr, "Interrupted twice, exiting (%s)", sig)
+					switch sig {
+					case syscall.SIGINT:
+						os.Exit(130)
+					default:
+						os.Exit(130) // if we were interrupted, never return zero.
+					}
+
+				case <-time.After(30 * time.Minute): // allow time for cleanup.  If we finish before this, we'll exit
+					fmt.Fprintf(os.Stderr, "Timed out during cleanup, exiting")
+					os.Exit(130) // if we were interrupted, never return zero.
+				}
+			}()
+			signal.Notify(abortCh, syscall.SIGINT, syscall.SIGTERM)
+
 			ext := registry.Get(opts.componentFlags.Component)
 			if ext == nil {
 				return fmt.Errorf("component not found: %s", opts.componentFlags.Component)
@@ -69,7 +100,7 @@ func NewRunTestCommand(registry *extension.Registry) *cobra.Command {
 			}
 			defer w.Flush()
 
-			return specs.Run(w, opts.concurrencyFlags.MaxConcurency)
+			return specs.Run(ctx, w, opts.concurrencyFlags.MaxConcurency)
 		},
 	}
 	opts.componentFlags.BindFlags(cmd.Flags())

pkg/cypress/util.go

@@ -1,6 +1,7 @@
 package cypress
 
 import (
+	"context"
 	"encoding/json"
 	"encoding/xml"
 	"fmt"
@@ -54,10 +55,10 @@ func BuildExtensionTestSpecsFromCypressMetadata(metadata []byte) (ext.ExtensionT
 			Name:          tc.Name,
 			Labels:        sets.New(tc.Tags...),
 			CodeLocations: []string{tc.FilePath},
-			Run: func() *ext.ExtensionTestResult {
+			Run: func(ctx context.Context) *ext.ExtensionTestResult {
 				return runCypressTest(tc.ID, tc.Name, tc.FilePath)
 			},
-			RunParallel: func() *ext.ExtensionTestResult {
+			RunParallel: func(ctx context.Context) *ext.ExtensionTestResult {
 				// this is equivalent to before, but potentially could be improved.
 				return runCypressTest(tc.ID, tc.Name, tc.FilePath)
 			},

pkg/extension/extensiontests/spec.go

@@ -1,6 +1,7 @@
 package extensiontests
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"sync"
@@ -178,7 +179,7 @@ func (specs ExtensionTestSpecs) Names() []string {
 // are written to the given ResultWriter after each spec has completed execution.  BeforeEach,
 // BeforeAll, AfterEach, AfterAll hooks are executed when specified. "Each" hooks must be thread
 // safe. Returns an error if any test spec failed, indicating the quantity of failures.
-func (specs ExtensionTestSpecs) Run(w ResultWriter, maxConcurrent int) error {
+func (specs ExtensionTestSpecs) Run(ctx context.Context, w ResultWriter, maxConcurrent int) error {
 	queue := make(chan *ExtensionTestSpec)
 	failures := atomic.Int64{}
 
@@ -214,7 +215,7 @@ func (specs ExtensionTestSpecs) Run(w ResultWriter, maxConcurrent int) error {
 					beforeEachTask.Run(*spec)
 				}
 
-				res := runSpec(spec, runSingleSpec)
+				res := runSpec(ctx, spec, runSingleSpec)
 				if res.Result == ResultFailed {
 					failures.Add(1)
 				}
@@ -546,13 +547,13 @@ func (spec *ExtensionTestSpec) Exclude(excludeCEL string) *ExtensionTestSpec {
 	return spec
 }
 
-func runSpec(spec *ExtensionTestSpec, runSingleSpec bool) *ExtensionTestResult {
+func runSpec(ctx context.Context, spec *ExtensionTestSpec, runSingleSpec bool) *ExtensionTestResult {
 	startTime := time.Now().UTC()
 	var res *ExtensionTestResult
 	if runSingleSpec || spec.RunParallel == nil {
-		res = spec.Run()
+		res = spec.Run(ctx)
 	} else {
-		res = spec.RunParallel()
+		res = spec.RunParallel(ctx)
 	}
 	duration := time.Since(startTime)
 	endTime := startTime.Add(duration).UTC()

pkg/extension/extensiontests/spec_test.go

@@ -1,6 +1,7 @@
 package extensiontests
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"sync/atomic"
@@ -209,7 +210,7 @@ func TestExtensionTestSpecs_HookExecution(t *testing.T) {
 			for i := 0; i < tc.numSpecs; i++ {
 				specs = append(specs, &ExtensionTestSpec{
 					Name: fmt.Sprintf("test spec %d", i+1),
-					Run: func() *ExtensionTestResult {
+					Run: func(ctx context.Context) *ExtensionTestResult {
 						return produceTestResult(fmt.Sprintf("test result %d", i+1), 20*time.Second)
 					},
 				})
@@ -241,7 +242,7 @@ func TestExtensionTestSpecs_HookExecution(t *testing.T) {
 			}
 
 			// Run the test specs
-			err := specs.Run(NullResultWriter{}, 10)
+			err := specs.Run(context.TODO(), NullResultWriter{}, 10)
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}

pkg/extension/extensiontests/types.go

@@ -1,6 +1,8 @@
 package extensiontests
 
 import (
+	"context"
+
 	"github.com/openshift-eng/openshift-tests-extension/pkg/dbtime"
 	"github.com/openshift-eng/openshift-tests-extension/pkg/util/sets"
 )
@@ -45,11 +47,11 @@ type ExtensionTestSpec struct {
 
 	// Run invokes a test in-process.  It must not call back into `ote-binary run-test` because that will usually
 	// cause an infinite recursion.
-	Run func() *ExtensionTestResult `json:"-"`
+	Run func(ctx context.Context) *ExtensionTestResult `json:"-"`
 
 	// RunParallel invokes a test in parallel with other tests.  This is usually done by exec-ing out
 	// to the `ote-binary run-test "test name"` commmand and interpretting the result.
-	RunParallel func() *ExtensionTestResult `json:"-"`
+	RunParallel func(ctx context.Context) *ExtensionTestResult `json:"-"`
 
 	// Hook functions
 	afterAll   []*OneTimeTask

pkg/ginkgo/util.go

@@ -70,7 +70,7 @@ func BuildExtensionTestSpecsFromOpenShiftGinkgoSuite(selectFns ...ext.SelectFunc
 			Labels:        sets.New[string](spec.Labels()...),
 			CodeLocations: codeLocations,
 			Lifecycle:     GetLifecycle(spec.Labels()),
-			Run: func() *ext.ExtensionTestResult {
+			Run: func(ctx context.Context) *ext.ExtensionTestResult {
 				enforceSerialExecutionForGinkgo.Lock()
 				defer enforceSerialExecutionForGinkgo.Unlock()
 
@@ -131,9 +131,9 @@ func BuildExtensionTestSpecsFromOpenShiftGinkgoSuite(selectFns ...ext.SelectFunc
 
 				return result
 			},
-			RunParallel: func() *ext.ExtensionTestResult {
+			RunParallel: func(ctx context.Context) *ext.ExtensionTestResult {
 				// TODO pass through timeout and determine Lifecycle
-				return SpawnProcessToRunTest(context.TODO(), name, 90*time.Minute)
+				return SpawnProcessToRunTest(ctx, name, 90*time.Minute)
 			},
 		}
 		specs = append(specs, testCase)