feat: enable agent handling of stop requests #3539
Conversation
| // TODO: get context from request | ||
| err = c.stopListener(context.Background(), &resp) | ||
| if err != nil { | ||
| fmt.Println(err.Error()) |
There was a problem hiding this comment.
nit: Can we print this message with a logger?
There was a problem hiding this comment.
all the workers need to implement the logger so I'll create a new PR for this
| func (o *verboseObserver) StartStopRequest(request *proto.StopRequest) { | ||
| o.ui.Infof("%s Stopping test run %d, test %s ...", consoleUI.Emoji_Sparkles, request.RunID, request.TestID) | ||
| } |
server/executor/queue.go
Outdated
| uReq := UserRequest{ | ||
| TenantID: job.TenantID, | ||
| TestID: job.Test.ID, | ||
| RunID: job.Run.ID, | ||
| } | ||
|
|
||
| q.subscriptor.Subscribe(uReq.ResourceID(UserRequestTypeStop), sfn) | ||
| q.subscriptor.Subscribe(uReq.ResourceID(UserRequestSkipTraceCollection), spfn) |
There was a problem hiding this comment.
nit: Can we improve the naming of the uReq, sfn and spfn variables? Looking at them at the Subscribe function they are a little bit confusing to understand.
Maybe we can rename to something like:
uReq->userRequestsfn->stopFunctionCallbackspfn->skipTraceCollectionCallback
| go func() { | ||
| err = w.trigger(subcontext, triggerRequest) | ||
| done <- true | ||
| }() |
There was a problem hiding this comment.
I suggest a short comment explaining that we moved this execution to a go routine because the user can cancel the test, and in that case, we will ignore the outcome of this go routine.
I'm thinking about this because I'm sure that we will look at this code in 6 months and think: "why did we do that?" 😂
There was a problem hiding this comment.
I'll make another PR implementing the same feature for the trace poller, so I'll try to make a nicer implementation there. If I cannot make it and end up copy/pasting, I'll add a comment on both!
This PR enables the agent to receive users "stop test" request so it can stop what it is doing and correctly flag the test run.
Architecture
The challenge for this feature is that the process that gets the users requests needs to stop a separated, independent process (for example, the trigger being executed). To solve this, this PR implements a map of context Cancel functions. Whenever a new process starts, it registers a new cancelFunc with a unique key based on the testID/runID. The stop listener notifies that channel when it gets a stop request, and the processor func can handle the cancelation.
When the control plane sends the agent TriggerResponse back to the controller, it checks if that response was cancelled by the user, if so it handles the case accordingly, emiting events, updating the run state, etc.
Workflow
sequenceDiagram participant U as User participant C as Controller(Core) participant TP as TestPipeline(Core) participant P as Publisher(NATS) participant CA as Cloud API participant A as Agent rect rgb(200, 200, 200) Note over A: Agent Startup Registration A->>+CA: Register Trigger/Poller/Etc stream A->>+CA: Register StopRequest handler stream end U->>C: Initiate StopRun C->>TP: Forward StopRun request TP->>P: Publish StopRun message P->>CA: Notify StopRun event CA->>A: Notify Agent via StopRequest stream A->>A: Stop running process (Internal) A->>CA: Send TriggerResponse with UserCancelled error CA->>C: Forward TriggerResponse to Controller C->>TP: If UserCancelled error, call UpdateStoppedTest TP->>TP: Set Run state, emit events (Internal)Changes
Fixes
Checklist
Loom video
Working demo: https://www.loom.com/share/99f4bf306e284474bb5c37bf59feeda1?sid=18fea06a-de95-46fc-b3ab-bb605178997f
PR walkthrough: https://www.loom.com/share/30d16279d9c0491ebff17d252d5d2192?sid=b2a28755-3f21-4cef-803c-f1a281764c70