From cebd8efd2e704ff186aea47c52227eeea2dc062c Mon Sep 17 00:00:00 2001 From: Junyi Date: Wed, 3 Jan 2024 09:59:33 +0800 Subject: [PATCH] chore(plugin-workflow): add metric example (#3305) * chore(plugin-workflow): add metric example * refactor(plugin-workflow): refactor some code * refactor(plugin-workflow): remove type for less dependencies --- .../plugin-workflow/src/server/Plugin.ts | 29 ++++++++++++------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts index d13c914e2..2b1bd099b 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts @@ -39,8 +39,10 @@ export default class WorkflowPlugin extends Plugin { private executing: Promise | null = null; private pending: Pending[] = []; private events: CachedEvent[] = []; + private eventsCount = 0; private loggerCache: LRUCache; + private meter = null; getLogger(workflowId: ID): Logger { const now = new Date(); @@ -165,6 +167,12 @@ export default class WorkflowPlugin extends Plugin { }, }); + this.meter = this.app.telemetry.metric.getMeter(); + const counter = this.meter.createObservableGauge('workflow.events.counter'); + counter.addCallback((result) => { + result.observe(this.eventsCount); + }); + this.app.acl.registerSnippet({ name: `pm.${this.name}.workflows`, actions: [ @@ -275,6 +283,7 @@ export default class WorkflowPlugin extends Plugin { } this.events.push([workflow, context, options]); + this.eventsCount = this.events.length; logger.info(`new event triggered, now events: ${this.events.length}`); logger.debug(`event data:`, { @@ -324,7 +333,7 @@ export default class WorkflowPlugin extends Plugin { } } - const execution = await this.db.sequelize.transaction(async (transaction) => { + return this.db.sequelize.transaction(async (transaction) => { const execution = await workflow.createExecution( { context, @@ -334,6 +343,8 @@ export default class WorkflowPlugin extends Plugin { { transaction }, ); + this.getLogger(workflow.id).info(`execution of workflow ${workflow.id} created as ${execution.id}`); + await workflow.increment(['executed', 'allExecuted'], { transaction }); // NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment if (this.db.options.dialect !== 'postgres') { @@ -356,19 +367,11 @@ export default class WorkflowPlugin extends Plugin { return execution; }); - - this.getLogger(workflow.id).info(`execution of workflow ${workflow.id} created as ${execution.id}`); - - // NOTE: cache first execution for most cases - if (!this.executing && !this.pending.length) { - this.pending.push([execution]); - } - - return execution; } private prepare = async () => { const event = this.events.shift(); + this.eventsCount = this.events.length; if (!event) { this.getLogger('dispatcher').warn(`events queue is empty, no need to prepare`); return; @@ -378,7 +381,11 @@ export default class WorkflowPlugin extends Plugin { logger.info(`preparing execution for event`); try { - await this.createExecution(event); + const execution = await this.createExecution(event); + // NOTE: cache first execution for most cases + if (!this.executing && !this.pending.length) { + this.pending.push([execution]); + } } catch (err) { logger.error(`failed to create execution: ${err.message}`, err); // this.events.push(event); // NOTE: retry will cause infinite loop