From 7db68764acdaf0d2b6975ca5e159161dff1251fb Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Tue, 14 Mar 2023 13:57:25 +0000 Subject: [PATCH] Structure debug logs and retry workflow registration up to 4 times --- apps/wss/src/server.ts | 68 +++++++++++++------- packages/internal-bridge/src/logger.ts | 28 ++++++-- packages/internal-platform/src/api/client.ts | 23 ++++++- packages/internal-platform/src/logger.ts | 24 ++++++- 4 files changed, 111 insertions(+), 32 deletions(-) diff --git a/apps/wss/src/server.ts b/apps/wss/src/server.ts index 04fd64e9c..bdd77e96d 100644 --- a/apps/wss/src/server.ts +++ b/apps/wss/src/server.ts @@ -8,6 +8,8 @@ import { import { CommandCatalog, InternalApiClient, + InternalResponseError, + RegisterWorkflowResponse, TriggerCatalog, triggerCatalog, ZodPublisher, @@ -22,6 +24,9 @@ import { env } from "./env"; import { pulsarClient } from "./pulsarClient"; import { WorkflowRunController } from "./runController"; +const MAX_RETRY_ATTEMPTS = 4; +const RETRY_DELAY_MS = 888; + export class TriggerServer { #connection?: TriggerServerConnection; #serverRPC?: ZodRPC; @@ -423,17 +428,7 @@ export class TriggerServer { try { // register the workflow with the platform - const response = await this.#apiClient.registerWorkflow({ - id: data.workflowId, - name: data.workflowName, - trigger: data.trigger, - package: { - name: data.packageName, - version: data.packageVersion, - }, - triggerTTL: data.triggerTTL, - metadata: data.metadata ? JSON.stringify(data.metadata) : undefined, - }); + const response = await this.#registerWorkflow(data); this.#workflowId = response.workflow.id; @@ -448,20 +443,12 @@ export class TriggerServer { subscriptionType: "Shared", subscriptionInitialPosition: "Latest", }, + filter: { + "x-workflow-id": this.#workflowId, + "x-api-key": this.#apiKey, + }, handlers: { TRIGGER_WORKFLOW: async (id, data, properties, messageAttributes) => { - // If the API keys don't match, then we should ignore it - // This ensures the workflow is triggered for the correct environment - if (properties["x-api-key"] !== this.#apiKey) { - return true; - } - - // If the workflow id is not the same as the workflow id - // that we are listening for, then we should ignore it - if (properties["x-workflow-id"] !== this.#workflowId) { - return true; - } - if (!this.#serverRPC) { throw new Error( "Cannot trigger workflow without an RPC connection" @@ -582,6 +569,41 @@ export class TriggerServer { } } + async #registerWorkflow( + data: z.infer<(typeof ServerRPCSchema)["INITIALIZE_HOST_V2"]["request"]>, + attempt: number = 0 + ): Promise { + try { + return await this.#apiClient.registerWorkflow({ + id: data.workflowId, + name: data.workflowName, + trigger: data.trigger, + package: { + name: data.packageName, + version: data.packageVersion, + }, + triggerTTL: data.triggerTTL, + metadata: data.metadata ? JSON.stringify(data.metadata) : undefined, + }); + } catch (error) { + if (error instanceof InternalResponseError && error.retryable) { + // If error.status is a retryable status code, and we haven't exceeded the max number of attempts, retry + if (attempt < MAX_RETRY_ATTEMPTS) { + this.#logger.debug( + `Failed to register workflow, retrying in ${RETRY_DELAY_MS}ms...`, + error + ); + + await sleep(RETRY_DELAY_MS); + + return this.#registerWorkflow(data, attempt + 1); + } + } + + throw error; + } + } + async #closePubSub() { this.#logger.debug( "Closing run controllers...", diff --git a/packages/internal-bridge/src/logger.ts b/packages/internal-bridge/src/logger.ts index bbea4ff3d..6eaec0479 100644 --- a/packages/internal-bridge/src/logger.ts +++ b/packages/internal-bridge/src/logger.ts @@ -71,13 +71,18 @@ export class Logger { ); } - debug(...args: any[]) { + debug(message: string, ...args: any[]) { if (this.#level < 5) return; - console.debug( - `[${formattedDateTime()}] ${this.#formatName()} `, - ...[...args, ...this.#formatTags()] - ); + const structuredLog = { + timestamp: formattedDateTime(), + name: this.#name, + tags: this.#tags, + message, + args: structureArgs(args), + }; + + console.debug(JSON.stringify(structuredLog)); } #formatName() { @@ -114,3 +119,16 @@ function formattedDateTime() { return `${formattedHours}:${formattedMinutes}:${formattedSeconds}.${formattedMilliseconds}`; } + +// If args is has a single item that is an object, return that object +function structureArgs(args: any[]) { + if (args.length === 0) { + return; + } + + if (args.length === 1 && typeof args[0] === "object") { + return args[0]; + } + + return args; +} diff --git a/packages/internal-platform/src/api/client.ts b/packages/internal-platform/src/api/client.ts index 9b45cf2e1..cce5d1506 100644 --- a/packages/internal-platform/src/api/client.ts +++ b/packages/internal-platform/src/api/client.ts @@ -3,6 +3,24 @@ import { UpdateWorkflowRun, RegisteredWorkflow } from "../schemas"; import fetch from "node-fetch"; import { Logger } from "../logger"; +const RETRYABLE_STATUS_CODES = [500, 502, 503, 504]; + +// Response error with status code +export class InternalResponseError extends Error { + retryable: boolean; + + constructor(retryable: boolean, message: string) { + super(message); + + this.retryable = retryable; + } +} + +// export type of InternalApiClient.registerWorkflow +export type RegisterWorkflowResponse = Awaited< + ReturnType["registerWorkflow"]> +>; + export class InternalApiClient { #apiKey: string; #baseUrl: string; @@ -104,8 +122,9 @@ export class InternalApiClient { throw new Error(body.error); } - throw new Error( - `[${response.status}] Something went wrong: ${response.statusText}` + throw new InternalResponseError( + RETRYABLE_STATUS_CODES.includes(response.status), + response.statusText ); } diff --git a/packages/internal-platform/src/logger.ts b/packages/internal-platform/src/logger.ts index b3ccd59e1..00522a2e9 100644 --- a/packages/internal-platform/src/logger.ts +++ b/packages/internal-platform/src/logger.ts @@ -39,10 +39,17 @@ export class Logger { console.info(`[${formattedDateTime()}] [${this.#name}] `, ...args); } - debug(...args: any[]) { + debug(message: string, ...args: any[]) { if (this.#level < 4) return; - console.debug(`[${formattedDateTime()}] [${this.#name}] `, ...args); + const structuredLog = { + timestamp: formattedDateTime(), + name: this.#name, + message, + args: structureArgs(args), + }; + + console.debug(JSON.stringify(structuredLog)); } } @@ -67,3 +74,16 @@ function formattedDateTime() { return `${formattedHours}:${formattedMinutes}:${formattedSeconds}.${formattedMilliseconds}`; } + +// If args is has a single item that is an object, return that object +function structureArgs(args: any[]) { + if (args.length === 0) { + return; + } + + if (args.length === 1 && typeof args[0] === "object") { + return args[0]; + } + + return args; +}