wip - isolating trace poller start

server/app/test_pipeline.go

@@ -61,7 +61,11 @@ func buildTestPipeline(
 		eventEmitter,
 	)
 
-	tracePoller := executor.NewTracePoller(
+	tracePollerStartWorker := executor.NewTracePollerStartWorker(
+		eventEmitter,
+	)
+
+	tracePollerWorker := executor.NewTracePoller(
 		pollerExecutor,
 		execTestUpdater,
 		subscriptionManager,
@@ -108,12 +112,13 @@ func buildTestPipeline(
 		executor.PipelineStep{Processor: triggerResolverWorker, Driver: pgQueue.Channel("trigger_resolve")},
 		executor.PipelineStep{Processor: triggerExecuterWorker, Driver: pgQueue.Channel("trigger_execute")},
 		executor.PipelineStep{Processor: triggerResultProcessorWorker, Driver: pgQueue.Channel("trigger_result")},
-		executor.PipelineStep{Processor: tracePoller, Driver: pgQueue.Channel("tracePoller")},
+		executor.PipelineStep{Processor: tracePollerStartWorker, Driver: pgQueue.Channel("tracePoller_start")},
+		executor.PipelineStep{Processor: tracePollerWorker, Driver: pgQueue.Channel("tracePoller")},
 		executor.PipelineStep{Processor: linterRunner, Driver: pgQueue.Channel("linterRunner")},
 		executor.PipelineStep{Processor: assertionRunner, Driver: pgQueue.Channel("assertionRunner")},
 	)
 
-	const assertionRunnerStepIndex = 3
+	const assertionRunnerStepIndex = 6
 
 	return executor.NewTestPipeline(
 		pipeline,

server/executor/trace_poller.go

@@ -99,10 +99,6 @@ func (tp tracePoller) enqueueJob(ctx context.Context, job Job) {
 	tp.inputQueue.Enqueue(ctx, job)
 }
 
-func (tp tracePoller) isFirstRequest(job Job) bool {
-	return job.EnqueueCount() == 0
-}
-
 func (tp *tracePoller) SetOutputQueue(queue Enqueuer) {
 	tp.outputQueue = queue
 }
@@ -118,15 +114,6 @@ func (tp *tracePoller) ProcessItem(ctx context.Context, job Job) {
 		return
 	}
 
-	if tp.isFirstRequest(job) {
-		err := tp.eventEmitter.Emit(ctx, events.TraceFetchingStart(job.Test.ID, job.Run.ID))
-		if err != nil {
-			log.Printf("[TracePoller] Test %s Run %d: fail to emit TracePollingStart event: %s", job.Test.ID, job.Run.ID, err.Error())
-		}
-	}
-
-	log.Println("TracePoller] processJob", job.EnqueueCount())
-
 	result, err := tp.pollerExecutor.ExecuteRequest(ctx, &job)
 	run := result.run
 	if err != nil {
@@ -226,7 +213,3 @@ func (tp tracePoller) requeue(ctx context.Context, job Job) {
 		tp.enqueueJob(ctx, job)
 	}()
 }
-
-func isFirstRequest(job *Job) bool {
-	return !job.Headers.GetBool("requeued")
-}

server/executor/trace_poller_start_worker.go

@@ -0,0 +1,44 @@
+package executor
+
+import (
+	"context"
+	"log"
+
+	"github.com/kubeshop/tracetest/server/model/events"
+)
+
+type tracePollerStartWorker struct {
+	eventEmitter EventEmitter
+	outputQueue  Enqueuer
+}
+
+func NewTracePollerStartWorker(
+	eventEmitter EventEmitter,
+) *tracePollerStartWorker {
+	return &tracePollerStartWorker{
+		eventEmitter: eventEmitter,
+	}
+}
+
+func (w *tracePollerStartWorker) SetOutputQueue(queue Enqueuer) {
+	w.outputQueue = queue
+}
+
+func (w *tracePollerStartWorker) ProcessItem(ctx context.Context, job Job) {
+	if w.isFirstRequest(job) {
+		err := w.eventEmitter.Emit(ctx, events.TraceFetchingStart(job.Test.ID, job.Run.ID))
+		if err != nil {
+			log.Printf("[TracePoller] Test %s Run %d: fail to emit TracePollingStart event: %s", job.Test.ID, job.Run.ID, err.Error())
+		}
+	}
+
+	log.Println("[TracePoller] processJob", job.EnqueueCount())
+
+	// TODO: check if there is more "pre-processing" things to do here
+
+	w.outputQueue.Enqueue(ctx, job)
+}
+
+func (w *tracePollerStartWorker) isFirstRequest(job Job) bool {
+	return job.EnqueueCount() == 0
+}