feat(plugin-workflow): add workflow specific logger (#1677)

* feat(plugin-workflow): add workflow specific logger

* fix(plugin-workflow): fix packages

* refactor(logger): adjust logger path env

.env.example

@@ -19,6 +19,10 @@ API_BASE_URL=
 
 PROXY_TARGET_URL=
 
+LOGGER_TRANSPORT=
+LOGGER_LEVEL=
+LOGGER_BASE_PATH=storage/logs
+
 ################# DATABASE #################
 
 DB_DIALECT=sqlite

docs/en-US/api/env.md

@@ -159,9 +159,9 @@ Log transport, default is `console,dailyRotateFile`, options include
 - `console`
 - `dailyRotateFile`
 
-### DAILY_ROTATE_FILE_DIRNAME
+### LOGGER_BASE_PATH
 
-Path to save `dailyRotateFile` logs, default is `storage/logs`
+Base path to save file based logs, default is `storage/logs`
 
 ## Temporary Environment Variables
 

docs/en-US/welcome/release/logger.md

@@ -18,7 +18,7 @@ app.use(middleware);
 logger 相关环境变量有：
 
 - [LOGGER_TRANSPORT](/api/env#logger_transport)
-- [DAILY_ROTATE_FILE_DIRNAME](/api/env#daily_rotate_file_dirname)
+- [LOGGER_BASE_PATH](/api/env#logger_base_path)
 
 ## Application 的 logger 配置
 

docs/zh-CN/api/env.md

@@ -154,14 +154,14 @@ DB_LOGGING=on
 
 ### LOGGER_TRANSPORT
 
-日志 transport，默认值 `console,dailyRotateFile`，可选项
+日志 transport，默认值 `console,dailyRotateFile`，可选项：
 
 - `console`
 - `dailyRotateFile`
 
-### DAILY_ROTATE_FILE_DIRNAME
+### LOGGER_BASE_PATH
 
-`dailyRotateFile` 日志的存储路径，默认为 `storage/logs`
+基于文件的日志存储路径，默认为 `storage/logs`。
 
 ## 临时环境变量
 

docs/zh-CN/welcome/release/logger.md

@@ -18,7 +18,7 @@ app.use(middleware);
 logger 相关环境变量有：
 
 - [LOGGER_TRANSPORT](/api/env#logger_transport)
-- [DAILY_ROTATE_FILE_DIRNAME](/api/env#daily_rotate_file_dirname)
+- [LOGGER_BASE_PATH](/api/env#logger_base_path)
 
 ## Application 的 logger 配置
 

packages/core/logger/src/create-logger.ts

@@ -4,10 +4,14 @@ import 'winston-daily-rotate-file';
 
 const { combine, timestamp, colorize, simple } = format;
 
-function loggingLevel() {
+export function getLoggerLevel(): string {
   return process.env.LOGGER_LEVEL || 'info';
 }
 
+export function getLoggerFilePath(...paths: string[]): string {
+  return path.resolve(process.env.LOGGER_BASE_PATH || path.resolve(process.cwd(), 'storage', 'logs'), ...paths);
+}
+
 const Transports = {
   console(options) {
     return new winston.transports.Console({
@@ -16,13 +20,13 @@ const Transports = {
     });
   },
   dailyRotateFile(options: any) {
-    let dirname = process.env.DAILY_ROTATE_FILE_DIRNAME || path.resolve(process.cwd(), './storage/logs');
+    let dirname = getLoggerFilePath();
     if (!path.isAbsolute(dirname)) {
       dirname = path.resolve(process.cwd(), dirname);
     }
     return new winston.transports.DailyRotateFile({
       dirname,
-      level: loggingLevel(),
+      level: getLoggerLevel(),
       filename: 'nocobase-%DATE%.log',
       datePattern: 'YYYY-MM-DD-HH',
       maxFiles: '14d',
@@ -54,7 +58,7 @@ function createLogger(options: LoggerOptions = {}) {
     })
     .filter((t) => t);
   const logger = winston.createLogger({
-    level: loggingLevel(),
+    level: getLoggerLevel(),
     levels: winston.config.cli.levels,
     format: combine(timestamp(), format.errors({ stack: true }), format.json(), colorize()),
     ...options,

packages/plugins/workflow/package.json

@@ -10,6 +10,7 @@
     "@nocobase/client": "0.9.1-alpha.2",
     "@nocobase/database": "0.9.1-alpha.2",
     "@nocobase/evaluators": "0.9.1-alpha.2",
+    "@nocobase/logger": "0.9.1-alpha.2",
     "@nocobase/resourcer": "0.9.1-alpha.2",
     "@nocobase/server": "0.9.1-alpha.2",
     "@nocobase/utils": "0.9.1-alpha.2",
@@ -18,6 +19,7 @@
     "classnames": "^2.3.1",
     "cron-parser": "4.4.0",
     "json-templates": "^4.2.0",
+    "lru-cache": "8.0.5",
     "moment": "^2.29.2",
     "react-js-cron": "^3.1.0"
   },

packages/plugins/workflow/src/server/Plugin.ts

@@ -1,5 +1,8 @@
 import path from 'path';
 
+import winston from 'winston';
+import LRUCache from 'lru-cache';
+
 import { Op } from '@nocobase/database';
 import { Plugin } from '@nocobase/server';
 import { Registry } from '@nocobase/utils';
@@ -14,15 +17,43 @@ import WorkflowModel from './models/Workflow';
 import Processor from './Processor';
 import initTriggers, { Trigger } from './triggers';
 import initFunctions from './functions';
+import { createLogger, Logger, LoggerOptions, getLoggerLevel, getLoggerFilePath } from '@nocobase/logger';
 
 type Pending = [ExecutionModel, JobModel?];
+
+type ID = number | string;
 export default class WorkflowPlugin extends Plugin {
   instructions: Registry<Instruction> = new Registry();
   triggers: Registry<Trigger> = new Registry();
   functions: Registry<Function> = new Registry();
-  executing: ExecutionModel | null = null;
-  pending: Pending[] = [];
-  events: [WorkflowModel, any, { context?: any }][] = [];
+  private executing: ExecutionModel | null = null;
+  private pending: Pending[] = [];
+  private events: [WorkflowModel, any, { context?: any }][] = [];
+
+  private loggerCache: LRUCache<string, Logger>;
+
+  getLogger(workflowId: ID): Logger {
+    const now = new Date();
+    const date = `${now.getFullYear()}-${`0${now.getMonth() + 1}`.slice(-2)}-${`0${now.getDate()}`.slice(-2)}`;
+    const key = `${date}-${workflowId}}`;
+    if (this.loggerCache.has(key)) {
+      return this.loggerCache.get(key);
+    }
+
+    const logger = createLogger({
+      transports: [
+        'console',
+        new winston.transports.File({
+          filename: getLoggerFilePath('workflows', date, `${workflowId}.log`),
+          level: getLoggerLevel(),
+        })
+      ],
+    } as LoggerOptions);
+
+    this.loggerCache.set(key, logger);
+
+    return logger;
+  }
 
   onBeforeSave = async (instance: WorkflowModel, options) => {
     const Model = <typeof WorkflowModel>instance.constructor;
@@ -79,6 +110,14 @@ export default class WorkflowPlugin extends Plugin {
     initInstructions(this, options.instructions);
     initFunctions(this, options.functions);
 
+    this.loggerCache = new LRUCache({
+      max: 20,
+      updateAgeOnGet: true,
+      dispose(logger) {
+        (<Logger>logger).end();
+      }
+    });
+
     this.app.acl.registerSnippet({
       name: `pm.${this.name}.workflows`,
       actions: [
@@ -162,7 +201,7 @@ export default class WorkflowPlugin extends Plugin {
 
     this.events.push([workflow, context, options]);
 
-    this.app.logger.debug(`[Workflow] new event triggered, now events: ${this.events.length}`);
+    this.getLogger(workflow.id).debug(`new event triggered, now events: ${this.events.length}`, { data: workflow.config });
 
     if (this.events.length > 1) {
       return;
@@ -189,9 +228,8 @@ export default class WorkflowPlugin extends Plugin {
       });
 
       if (existed) {
-        this.app.logger.warn(
-          `[Workflow] workflow ${workflow.id} has already been triggered in same execution (${options.context.executionId}), and newly triggering will be skipped.`,
-        );
+        this.getLogger(workflow.id).warn(`workflow ${workflow.id} has already been triggered in same execution (${options.context.executionId}), and newly triggering will be skipped.`);
+
         valid = false;
       }
     }
@@ -234,7 +272,7 @@ export default class WorkflowPlugin extends Plugin {
         return execution;
       });
 
-      this.app.logger.debug(`[Workflow] execution of workflow ${workflow.id} created as ${execution.id}`);
+      this.getLogger(workflow.id).debug(`execution of workflow ${workflow.id} created as ${execution.id}`, { data: execution.context });
 
       // NOTE: cache first execution for most cases
       if (!this.executing && !this.pending.length) {
@@ -292,12 +330,13 @@ export default class WorkflowPlugin extends Plugin {
 
     const processor = this.createProcessor(execution);
 
-    this.app.logger.info(`[Workflow] execution ${execution.id} ${job ? 'resuming' : 'starting'}...`);
+    this.getLogger(execution.workflowId).info(`execution (${execution.id}) ${job ? 'resuming' : 'starting'}...`);
 
     try {
       await (job ? processor.resume(job) : processor.start());
+      this.getLogger(execution.workflowId).info(`execution (${execution.id}) finished with status: ${execution.status}`);
     } catch (err) {
-      this.app.logger.error(`[Workflow] ${err.message}`, err);
+      this.getLogger(execution.workflowId).error(`execution (${execution.id}) error: ${err.message}`, err);
     }
 
     this.executing = null;

packages/plugins/workflow/src/server/Processor.ts

@@ -9,6 +9,7 @@ import ExecutionModel from './models/Execution';
 import JobModel from './models/Job';
 import FlowNodeModel from './models/FlowNode';
 import { EXECUTION_STATUS, JOB_STATUS } from './constants';
+import { Logger } from '@nocobase/logger';
 
 
 
@@ -29,6 +30,8 @@ export default class Processor {
     [JOB_STATUS.REJECTED]: EXECUTION_STATUS.REJECTED,
   };
 
+  logger: Logger;
+
   transaction?: Transaction;
 
   nodes: FlowNodeModel[] = [];
@@ -37,6 +40,7 @@ export default class Processor {
   jobsMapByNodeId: { [key: number]: any } = {};
 
   constructor(public execution: ExecutionModel, public options: ProcessorOptions) {
+    this.logger = options.plugin.getLogger(execution.workflowId);
   }
 
   // make dual linked nodes list then cache
@@ -137,12 +141,15 @@ export default class Processor {
     let job;
     try {
       // call instruction to get result and status
+      this.logger.info(`execution (${this.execution.id}) run instruction [${node.type}] for node (${node.id})`);
+      this.logger.debug(`config of node`, { data: node.config });
       job = await instruction(node, prevJob, this);
       if (!job) {
         return null;
       }
     } catch (err) {
       // for uncaught error, set to error
+      this.logger.error(`execution (${this.execution.id}) run instruction [${node.type}] for node (${node.id}) failed: `, { error: err });
       job = {
         result: err instanceof Error
           ? { message: err.message, stack: process.env.NODE_ENV === 'production' ? [] : err.stack }
@@ -162,8 +169,12 @@ export default class Processor {
     }
     const savedJob = await this.saveJob(job);
 
+    this.logger.info(`execution (${this.execution.id}) run instruction [${node.type}] for node (${node.id}) finished as status: ${savedJob.status}`);
+    this.logger.debug(`result of node`, { data: savedJob.result });
+
     if (savedJob.status === JOB_STATUS.RESOLVED && node.downstream) {
       // run next node
+      this.logger.debug(`run next node (${node.id})`);
       return this.run(node.downstream, savedJob);
     }
 
@@ -183,9 +194,11 @@ export default class Processor {
 
   // parent node should take over the control
   public async end(node, job) {
+    this.logger.debug(`branch ended at node (${node.id})})`);
     const parentNode = this.findBranchParentNode(node);
     // no parent, means on main flow
     if (parentNode) {
+      this.logger.debug(`not on main, recall to parent entry node (${node.id})})`);
       await this.recall(parentNode, job);
       return job;
     }
@@ -207,6 +220,7 @@ export default class Processor {
 
   async exit(job: JobModel | null) {
     const status = job ? (<typeof Processor>this.constructor).StatusMap[job.status] ?? Math.sign(job.status) : EXECUTION_STATUS.RESOLVED;
+    this.logger.info(`execution (${this.execution.id}) all nodes finished, finishing execution...`);
     await this.execution.update({ status }, { transaction: this.transaction });
     return null;
   }

packages/plugins/workflow/src/server/collections/workflows.ts

@@ -38,7 +38,7 @@ export default function () {
       {
         type: 'boolean',
         name: 'useTransaction',
-        defaultValue: true,
+        // defaultValue: true,
       },
       {
         type: 'hasMany',