feat(orchestrator): implement task processor #2221

TBonnin · 2024-05-29T17:41:34Z

Describe your changes

This PR is implementing the orchestrator processor which, for now, is being run by jobs.
The processor is behind a flag flag:orchestrator:dryrun:process:global. It is still in dryrun mode since no real processing is happening, tasks are either immediately succeeded or failed. It is the last step before actually processing webhook/actions

Notes

I have added orchestrator endpoints to support dequeueing, set task state, heartbeat and their respective functions in the orchestrator client
A processor dequeues tasks in a infinite loop and process the task via the process functions passed as argument.
A processor also check if pending tasks have been completed (ie: they might have expired or be cancelled) and send abort signal.
In jobs, 2 processor worker_thread are being started. One for action and one for webhook to reproduce what we have right now with webhooks being in a separate temporal queue

Tested in staging

Checklist before requesting a review (skip if just adding/editing APIs & templates)

I added tests, otherwise the reason is:
I added observability, otherwise the reason is:
I added analytics, otherwise the reason is:

khaliqgant · 2024-05-29T19:08:12Z

packages/orchestrator/lib/clients/client.ts

+                payload: { taskId }
+            });
+        } else {
+            return Ok(undefined);


should this be null?

null is not assignable to type void. Only undefined is. 🤷

TBonnin · 2024-05-29T17:42:56Z

packages/jobs/lib/app.ts

+            processor.stop();
+        }
+    }, 1000);
+


using setInterval because of the flag so it can be enabled/disabled without restarting jobs. Once stable it will be a simple processor.start()

TBonnin · 2024-05-29T17:43:30Z

packages/jobs/lib/processor/process.ts

+    // TODO: Implement action processing
+    // Returning an error for now
+    return Err(`Not implemented: ${JSON.stringify({ taskId: task.id })}`);
+}


dry-run: not doing real processing yet, just marking action as successful and webhook as failed for testing purpse

TBonnin · 2024-05-29T17:45:59Z

packages/orchestrator/lib/clients/client.ts

+import { validateTask } from './validate.js';
+import type { JsonValue } from 'type-fest';
+
+export class OrchestratorClient {


everything until dequeue was already there. I have moved the file in a clients subdir

TBonnin · 2024-05-29T17:48:59Z

packages/orchestrator/lib/routes/health.ts

@@ -10,12 +10,9 @@ type Health = Endpoint<{
 const path = '/health';
 const method = 'GET';

-export const handler: RouteHandler<Health> = {
+export const getHandler: RouteHandler<Health> = {


Since each route file corresponds to a path it is possible to have multiple Endpoint/handler per file, one for each HTTP method. So I renamed the Endpoint and handlers to make it clearer

I tried to keep one per file by prefixing the filename, i.e: getHealth.ts would make sense to me and keep files small

sensible. I will do the same then

khaliqgant · 2024-05-29T19:18:56Z

packages/orchestrator/lib/routes/v1/search.ts

+import type { ApiError, Endpoint } from '@nangohq/types';
+import type { EndpointRequest, EndpointResponse, RouteHandler, Route } from '@nangohq/utils';
+import { validateRequest } from '@nangohq/utils';
+


What is the purpose of search in the context of the processor?

It is the search task endpoint. The processor must be able to check if pending tasks have been cancelled so it need to be able to search for specific tasks.
This endpoint could also be used in internal tooling to display list of tasks per queue/state/etc...

bodinsamuel · 2024-05-30T10:23:02Z

I'm late in my reviews, checking this in a couple of hours 👍🏻

bodinsamuel

Very nice progress! That's a lot of code and complexity to review, hopefully we'll get used to it quickly 💯

packages/orchestrator/lib/clients/processor.ts

bodinsamuel · 2024-05-30T13:12:58Z

packages/orchestrator/lib/clients/validate.ts

+    payload: webhookArgsSchema
+});
+
+export function validateTask(task: Task): Result<TaskAction | TaskWebhook> {


I feel like Task{ type: string } would have been more scalable, because we'll need to add syncs and then maybe more stuff along the way

they don't have the same fields though. Having some type guard functionality is nice though even though I agree that the definitions are more complex

I mean you can still validate the payload but it's just a waste of time to not use a discriminator 👍🏻

bodinsamuel · 2024-05-30T13:14:11Z

packages/orchestrator/lib/routes/health.ts

@@ -10,12 +10,9 @@ type Health = Endpoint<{
 const path = '/health';
 const method = 'GET';

-export const handler: RouteHandler<Health> = {
+export const getHandler: RouteHandler<Health> = {


I tried to keep one per file by prefixing the filename, i.e: getHealth.ts would make sense to me and keep files small

packages/orchestrator/lib/routes/v1/search.ts

bodinsamuel · 2024-05-30T14:03:46Z

packages/orchestrator/lib/routes/v1/dequeue.ts

+            cleanupAndRespond((res) => res.status(200).send([]));
+        }, waitForCompletionTimeoutMs);
+
+        eventEmitter.once(eventId, onTaskStarted);


not sure I get the wait for completion in that case, because if don't receive an answer you'll just wait 60s for nothing

you are waiting for a task to be created/dequeuable, so you don't have to keep making request. Trying hard not to have a push based system that would requires handling states but in the scenario of a processor it could be an acceptable solution.

okay took me a while to understand the executeAction -> execute -> schedule -> immediate -> event
that works 👍🏻

packages/orchestrator/lib/clients/processor.ts

packages/jobs/lib/app.ts

packages/jobs/lib/env.ts

bodinsamuel · 2024-05-30T16:36:02Z

packages/orchestrator/lib/clients/processor.ts

+            if (this.queue.available() > 0) {
+                const tasks = await this.orchestratorClient.dequeue({ groupKey: this.groupKey, limit: this.queue.available() * 2, waitForCompletion: true }); // fetch more than available to keep the queue full
+                if (tasks.isErr()) {
+                    logger.error(`failed to dequeue tasks: ${stringifyError(tasks.error)}`);


just a heads up, happened to me once when the API stop responding all fetch are failing fast ultimately using the whole cpu. Might be an edge case but fyi

good point. introducing a little delay in case of error

bodinsamuel

🛫

TBonnin requested review from bodinsamuel and khaliqgant May 29, 2024 17:41

TBonnin force-pushed the tbonnin/orchestrator-processor branch from 853da7a to 8a1e376 Compare May 29, 2024 18:47

khaliqgant reviewed May 29, 2024

View reviewed changes

TBonnin force-pushed the tbonnin/orchestrator-processor branch from 8a1e376 to d4d3bd4 Compare May 29, 2024 19:11

khaliqgant self-requested a review May 29, 2024 19:12

TBonnin commented May 29, 2024

View reviewed changes

khaliqgant approved these changes May 29, 2024

View reviewed changes

khaliqgant reviewed May 29, 2024

View reviewed changes

TBonnin force-pushed the tbonnin/orchestrator-processor branch 2 times, most recently from d3591ab to e57716e Compare May 29, 2024 19:47

bodinsamuel reviewed May 30, 2024

View reviewed changes

TBonnin changed the title ~~chore(orchestrator): implement task processor~~ feat(orchestrator): implement task processor May 30, 2024

bodinsamuel reviewed May 30, 2024

View reviewed changes

bodinsamuel approved these changes May 30, 2024

View reviewed changes

TBonnin force-pushed the tbonnin/orchestrator-processor branch from 0a8e96d to 8ac4bb9 Compare May 30, 2024 17:50

TBonnin added 3 commits May 30, 2024 14:53

chore(orchestrator): implement task processor

c89a18e

address PR review comments

8c85ed6

chore(processor): add delay when dequeuing fails

3503468

TBonnin force-pushed the tbonnin/orchestrator-processor branch from 8ac4bb9 to 3503468 Compare May 30, 2024 18:54

TBonnin enabled auto-merge (squash) May 30, 2024 18:54

TBonnin merged commit d1ac7a3 into master May 30, 2024
21 checks passed

TBonnin deleted the tbonnin/orchestrator-processor branch May 30, 2024 18:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(orchestrator): implement task processor #2221

feat(orchestrator): implement task processor #2221

TBonnin commented May 29, 2024 •

edited

Loading

khaliqgant May 29, 2024

TBonnin May 29, 2024

TBonnin May 29, 2024

TBonnin May 29, 2024

TBonnin May 29, 2024

TBonnin May 29, 2024

bodinsamuel May 30, 2024

TBonnin May 30, 2024

khaliqgant May 29, 2024

TBonnin May 29, 2024

bodinsamuel commented May 30, 2024

bodinsamuel left a comment

bodinsamuel May 30, 2024

TBonnin May 30, 2024

bodinsamuel May 30, 2024

bodinsamuel May 30, 2024

bodinsamuel May 30, 2024

TBonnin May 30, 2024 •

edited

Loading

bodinsamuel May 30, 2024

bodinsamuel May 30, 2024

TBonnin May 30, 2024 •

edited

Loading

bodinsamuel left a comment

feat(orchestrator): implement task processor #2221

feat(orchestrator): implement task processor #2221

Conversation

TBonnin commented May 29, 2024 • edited Loading

Describe your changes

Notes

Checklist before requesting a review (skip if just adding/editing APIs & templates)

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

bodinsamuel commented May 30, 2024

bodinsamuel 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

TBonnin May 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TBonnin May 30, 2024 • edited Loading

Choose a reason for hiding this comment

bodinsamuel left a comment

Choose a reason for hiding this comment

TBonnin commented May 29, 2024 •

edited

Loading

TBonnin May 30, 2024 •

edited

Loading

TBonnin May 30, 2024 •

edited

Loading