feat(server): refactor runner channels into abstract queues #2971

schoren · 2023-07-20T20:38:56Z

This PR introduces the concept of test runner pipelines, which decouples the steps of the pipeline from each other. The pipeline consists of several steps, including trigger, poll trace, analyze result, and assert. Instead of each step needing to know what step comes next, it only needs to have a reference to an "outputQueue". This outputQueue is the "input" queue for the next step, which allows for greater flexibility and modularity in the codebase.

The main goal of this PR is to remove all the manual queues that were implemented using Go channels. Previously, each step had a copy/pasted worker code that relied on channels, which made it hard to replace the queue mechanism. The new pipeline queues rely on an abstract queue driver, which allows for different types of queues to be created independently. Currently, we only implement an in-memory driver using Go channels, so we're not changing behavior, just architecture.

With everything decoupled, we can independently create different types of queues, such as a PostgreSQL listen/notify queue. This also adds flexibility to modify the pipeline, such as adding steps or changing the order, without needing to modify any existing steps.

Explanation of the Queue:
https://www.loom.com/share/8209a55d500c4766a0f137f0c723bf22

Explanation of the Pipeline:
https://www.loom.com/share/94aca45c8e8f4f17ab06403e1f65833a

How those fit in the application:
https://www.loom.com/share/43f74fccecb042f69b89f5f9098e56d6

Changes

Introduced the concept of test runner pipelines
Decoupled the steps of the pipeline using output queues
Removed manual queues implemented using Go channels
Added an abstract queue driver for greater flexibility
Added an in-memory driver using Go channels
Updated the pipeline steps to use the new queue mechanism

Checklist

tested locally
added new dependencies
updated the docs
added a test

schoren · 2023-07-27T14:56:09Z

server/pkg/id/generator.go

@@ -56,12 +52,12 @@ func (g randGenerator) ID() ID {

 func (g randGenerator) TraceID() trace.TraceID {
 	tid := trace.TraceID{}
-	g.rand.Read(tid[:])
+	rand.New(rand.NewSource(time.Now().UnixNano())).Read(tid[:])


this fixes a race condition

The question is: how many days of life this line owes you? 😂

schoren · 2023-07-27T14:56:22Z

server/test/run_repository.go

@@ -602,7 +603,7 @@ func (r *runRepository) GetTransactionRunSteps(ctx context.Context, id id.ID, ru
 WHERE transaction_run_steps.transaction_run_id = $1 AND transaction_run_steps.transaction_run_transaction_id = $2
 ORDER BY test_runs.completed_at ASC
 `
-	query, params := sqlutil.Tenant(ctx, query, runID, id)
+	query, params := sqlutil.Tenant(ctx, query, strconv.Itoa(runID), id)


this is required for the new pgx db driver

Question: what is the idea of using pgx instead of pq directly?

pgx has support for LISTEN / NOTIFY. Originally this PR was going to do everything but it got larger than expected so I'll implement that in a separated PR, but we already did the driver change at the beginning. The idea was to check that everything worked with the new driver

schoren · 2023-07-27T15:01:44Z

server/testdb/test_run_event.go

@@ -136,10 +135,6 @@ func readTestRunEventFromRows(rows *sql.Rows) (model.TestRunEvent, error) {
 	)

 	if err != nil {
-		if errors.Is(err, sql.ErrNoRows) {


the controller compares with errors.Is(err, sql.ErrNoRows) now. this func checks the wrapped errors, so it will correctly assert that the error is sql.ErrNoRows without the need to return a custom error

danielbdias

The overall PR is excellent! I liked the way of handling drivers and queues here.
I've added some comments on it, but no blocker.

danielbdias · 2023-07-27T17:14:20Z

server/app/test_pipeline.go

+	return executor.NewTestPipeline(
+		pipeline,
+		subscriptionManager,
+		pipeline.GetQueueForStep(3), // assertion runner step
+		runRepo,
+		trRepo,
+		ppRepo,
+		dsRepo,
+	)


Can we move this magic index 3 (for the assertion runner step) to a variable (or even a constant)?

Suggested change

return executor.NewTestPipeline(

pipeline,

subscriptionManager,

pipeline.GetQueueForStep(3), // assertion runner step

runRepo,

trRepo,

ppRepo,

dsRepo,

)

assertionRunnerStepIndex := 3

return executor.NewTestPipeline(

pipeline,

subscriptionManager,

pipeline.GetQueueForStep(assertionRunnerStepIndex), // assertion runner step

runRepo,

trRepo,

ppRepo,

dsRepo,

)

danielbdias · 2023-07-27T17:16:21Z

server/app/test_pipeline.go

+	pipeline := executor.NewPipeline(queueBuilder,
+		executor.PipelineStep{Processor: runner, Driver: executor.NewInMemoryQueueDriver("runner")},
+		executor.PipelineStep{Processor: tracePoller, Driver: executor.NewInMemoryQueueDriver("tracePoller")},
+		executor.PipelineStep{Processor: linterRunner, Driver: executor.NewInMemoryQueueDriver("linterRunner")},


One thing to consider in the future: since we are considering renaming the internals of environment and transaction soon, we can also add the linter to the packet and change it to analyzer explicitly. Does it make sense?

I agree, but I saw that there's a distinction between linter and analyzer. I didn't dig too much into it, but that looks like a good time to normalize this kinds of things

danielbdias · 2023-07-27T17:19:53Z

server/config/server.go

@@ -80,23 +82,18 @@ func (c *AppConfig) PostgresConnString() string {
 	defer c.mu.Unlock()

 	if postgresConnString := c.vp.GetString("postgresConnString"); postgresConnString != "" {
-		return postgresConnString
+		fmt.Println("ERROR: postgresConnString was discontinued. Migrate to the new postgres format")


Something to think about in the future: we should move these server logs from fmt to something else, like zap or logrus, that we can centralize and configure later.
Does it make sense to create a ticket for that? (I can do that if needed!)

+1, do you mind creating the ticket @danielbdias ?

done! Issue: #2996

danielbdias · 2023-07-27T17:27:00Z

server/executor/pipeline.go

+}
+
+func (p *Pipeline) Begin(ctx context.Context, job Job) {
+	p.queues[0].Enqueue(ctx, job)


nit: Even knowing that we will always have a queue, does it make sense to add a guard clause here or on NewPipeline t avoid having 0 queues?

makes sense. I'll add it

danielbdias · 2023-07-27T17:29:14Z

server/executor/poller_executor_test.go


-		finished, finishReason, anotherRun, err := pollerExecutor.ExecuteRequest(request)
-		run = anotherRun // should store a run to use in another iteration
+	job := executor.NewJob()


On every call to NewJob, we set Test, Run, and PollingProfile. Does it make sense to add them as parameters on NewJob?

This is not the case for transactions, where it only passes Transaction and TransactionRun.

danielbdias · 2023-07-27T17:45:12Z

server/pkg/id/generator.go

@@ -56,12 +52,12 @@ func (g randGenerator) ID() ID {

 func (g randGenerator) TraceID() trace.TraceID {
 	tid := trace.TraceID{}
-	g.rand.Read(tid[:])
+	rand.New(rand.NewSource(time.Now().UnixNano())).Read(tid[:])


The question is: how many days of life this line owes you? 😂

server/pkg/id/generator.go

danielbdias · 2023-07-27T17:49:32Z

server/pkg/id/generator.go

 	return tid
 }

 func (g randGenerator) SpanID() trace.SpanID {
 	sid := trace.SpanID{}
-	g.rand.Read(sid[:])
+	rand.New(rand.NewSource(time.Now().UnixNano())).Read(sid[:])


Same here:

Suggested change

rand.New(rand.NewSource(time.Now().UnixNano())).Read(sid[:])

rndSeed := rand.NewSource(time.Now().UnixNano())

rand.New(rndSeed).Read(sid[:])

danielbdias · 2023-07-27T17:50:27Z

server/test/run_repository.go

@@ -602,7 +603,7 @@ func (r *runRepository) GetTransactionRunSteps(ctx context.Context, id id.ID, ru
 WHERE transaction_run_steps.transaction_run_id = $1 AND transaction_run_steps.transaction_run_transaction_id = $2
 ORDER BY test_runs.completed_at ASC
 `
-	query, params := sqlutil.Tenant(ctx, query, runID, id)
+	query, params := sqlutil.Tenant(ctx, query, strconv.Itoa(runID), id)


Question: what is the idea of using pgx instead of pq directly?

danielbdias · 2023-07-27T17:53:32Z

testing/server-tracetesting/features/http_test/08_rerun_http_test.yml

-  - selector: span[name = "exec UPDATE"]
-    assertions:
-    - attr:tracetest.selected_spans.count = 1


What is the rationale for removing this assertion? Don't we need to update the test anymore?

Good question. This was an RunUpdate made after creating the new run when rerunning a test, to update the state. I removed it incorrectly. Readding it

Co-authored-by: Daniel Baptista Dias <danielbdias@users.noreply.github.com>

schoren changed the title ~~Pg pubsub~~ feat(server): refactor runner channels into abstract queues Jul 20, 2023

schoren force-pushed the pg-pubsub branch from f629c50 to 56905b8 Compare July 25, 2023 16:07

schoren and others added 28 commits July 26, 2023 16:18

replace pg driver

5e12879

fix test

e435f93

--wip-- [skip ci]

74cc661

WIP: add pipeline structure

63b152f

update assertion runner

98ec189

implement linter runner

cca4bdf

WIP: convert other workers

ed4d1cb

wip

d571b57

implement runner

86440e8

fix rerun

d4baa2a

cleanup

5893a9e

fix test

a88c4b2

fix config test

e3cea98

clenaup test

0b5fa81

--wip-- [skip ci]

2d80e77

fix transactions

c012f93

fix poller test

b55841d

fixes

10ec3ab

fix queries

ab31f99

setup context propagation

e52736f

fix tests

b0b17fc

fix: fixes header propagation

93bfd02

fix

8cbbf64

--wip-- [skip ci]

782e0f2

--wip-- [skip ci]

25d62c7

remove race condiion

945b5f9

fix: weird polling bug

6d93f17

--wip-- [skip ci]

15d7823

schoren added 3 commits July 26, 2023 18:09

implement stop

ceda8eb

remove debug logs

f8cc155

remove debug logs

c9caa5b

schoren marked this pull request as ready for review July 26, 2023 21:15

schoren requested review from xoscar, mathnogueira and danielbdias and removed request for xoscar and mathnogueira July 26, 2023 21:15

rollback change

84430ea

schoren requested review from mathnogueira and xoscar July 27, 2023 14:31

schoren added 2 commits July 27, 2023 11:32

revert change

0dd4ecd

organize code

a3b7acd

schoren commented Jul 27, 2023

View reviewed changes

schoren added 2 commits July 27, 2023 11:57

server/openapi

5dc4e28

fix

65ad8f8

schoren commented Jul 27, 2023

View reviewed changes

fix pp

7269452

danielbdias approved these changes Jul 27, 2023

View reviewed changes

schoren and others added 6 commits July 27, 2023 15:10

Update server/pkg/id/generator.go

a81e761

Co-authored-by: Daniel Baptista Dias <danielbdias@users.noreply.github.com>

name magic constant

66add09

handle empty queues

796aeec

name magic constant

8a20d1d

fix rerun state

f36fbbf

fix

d910771

schoren merged commit f9e4bc2 into main Jul 27, 2023
30 checks passed

schoren deleted the pg-pubsub branch July 27, 2023 18:56

danielbdias pushed a commit that referenced this pull request Jul 27, 2023

feat(server): refactor runner channels into abstract queues (#2971)

5c67b03

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(server): refactor runner channels into abstract queues #2971

feat(server): refactor runner channels into abstract queues #2971

schoren commented Jul 20, 2023 •

edited

schoren Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

schoren Jul 27, 2023

danielbdias left a comment

danielbdias Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

danielbdias Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

danielbdias Jul 27, 2023

danielbdias Jul 27, 2023

danielbdias Jul 27, 2023

danielbdias Jul 27, 2023

schoren Jul 27, 2023

	rand.New(rand.NewSource(time.Now().UnixNano())).Read(sid[:])
	rndSeed := rand.NewSource(time.Now().UnixNano())
	rand.New(rndSeed).Read(sid[:])

feat(server): refactor runner channels into abstract queues #2971

feat(server): refactor runner channels into abstract queues #2971

Conversation

schoren commented Jul 20, 2023 • edited

Changes

Checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

danielbdias left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

schoren commented Jul 20, 2023 •

edited