feat(server): allow tests to be stopped in all test pipeline steps (#…

…3504)
kubeshop · Jan 9, 2024 · db11183 · db11183
1 parent 3a75722
commit db11183
Show file tree

Hide file tree

Showing 7 changed files with 42 additions and 0 deletions.
diff --git a/server/executor/assertion_runner.go b/server/executor/assertion_runner.go
@@ -45,6 +45,12 @@ func (e *defaultAssertionRunner) SetOutputQueue(pipeline.Enqueuer[Job]) {
 }
 
 func (e *defaultAssertionRunner) ProcessItem(ctx context.Context, job Job) {
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	run, err := e.runAssertionsAndUpdateResult(ctx, job)
 
 	log.Printf("[AssertionRunner] Test %s Run %d: update channel start\n", job.Test.ID, job.Run.ID)

diff --git a/server/executor/linter_runner.go b/server/executor/linter_runner.go
@@ -45,6 +45,12 @@ func (e *defaultLinterRunner) SetOutputQueue(queue pipeline.Enqueuer[Job]) {
 }
 
 func (e *defaultLinterRunner) ProcessItem(ctx context.Context, job Job) {
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	lintResource := e.analyzerGetter.GetDefault(ctx)
 
 	shouldSkip := lintResource.ShouldSkip()

diff --git a/server/executor/tracepollerworker/evaluator_worker.go b/server/executor/tracepollerworker/evaluator_worker.go
@@ -61,6 +61,12 @@ func (w *tracePollerEvaluatorWorker) SetOutputQueue(queue pipeline.Enqueuer[exec
 }
 
 func (w *tracePollerEvaluatorWorker) ProcessItem(ctx context.Context, job executor.Job) {
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	ctx, span := w.state.tracer.Start(ctx, "Evaluating trace")
 	defer span.End()
 

diff --git a/server/executor/tracepollerworker/fetcher_worker.go b/server/executor/tracepollerworker/fetcher_worker.go
@@ -53,6 +53,12 @@ func (w *traceFetcherWorker) ProcessItem(ctx context.Context, job executor.Job)
 		return
 	}
 
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	ctx, span := w.state.tracer.Start(ctx, "Fetching trace")
 	defer span.End()
 

diff --git a/server/executor/trigger_executer_worker.go b/server/executor/trigger_executer_worker.go
@@ -62,6 +62,12 @@ func (r triggerExecuterWorker) ProcessItem(ctx context.Context, job Job) {
 		return
 	}
 
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	err := r.eventEmitter.Emit(ctx, events.TriggerExecutionStart(job.Run.TestID, job.Run.ID))
 	if err != nil {
 		r.handleError(job.Run, err)

diff --git a/server/executor/trigger_resolver_worker.go b/server/executor/trigger_resolver_worker.go
@@ -77,6 +77,12 @@ func (r triggerResolverWorker) traceDB(ctx context.Context) (tracedb.TraceDB, er
 }
 
 func (r triggerResolverWorker) ProcessItem(ctx context.Context, job Job) {
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	ctx, pollingSpan := r.tracer.Start(ctx, "Resolve trigger")
 	defer pollingSpan.End()
 

diff --git a/server/executor/trigger_result_processor_worker.go b/server/executor/trigger_result_processor_worker.go
@@ -58,6 +58,12 @@ func (r triggerResultProcessorWorker) handleError(run test.Run, err error) {
 }
 
 func (r triggerResultProcessorWorker) ProcessItem(ctx context.Context, job Job) {
+	select {
+	default:
+	case <-ctx.Done():
+		return
+	}
+
 	ctx, pollingSpan := r.tracer.Start(ctx, "Start processing trigger response")
 	defer pollingSpan.End()