fix: redirect worker logs to main process (#988)

* add workerMode to pino logger

* redirect worker stdout

* style: resolve style guide violations

* use Set

* expose isValidLevel

* adjust format

* Rename instance

---------

Co-authored-by: sebastijankuzner <sebastijan.kuzner@outlook.com>

Author: oXtxNt9U

packages/contracts/source/contracts/kernel/log.ts

@@ -15,6 +15,8 @@ export interface Logger {
 
 	debug(message: string): void;
 
+	isValidLevel(level: string): boolean;
+
 	suppressConsoleOutput(suppress: boolean): void;
 
 	dispose(): Promise<void>;

packages/crypto-worker/source/service-provider.ts

@@ -25,8 +25,11 @@ export class ServiceProvider extends Providers.ServiceProvider {
 		this.app
 			.bind<() => Ipc.Subprocess<any>>(Identifiers.CryptoWorker.WorkerSubprocess.Factory)
 			.toFactory(() => () => {
-				const subprocess = new Worker(`${new URL(".", import.meta.url).pathname}/worker-script.js`, {});
-				return new Ipc.Subprocess(subprocess);
+				const subprocess = new Worker(`${new URL(".", import.meta.url).pathname}/worker-script.js`, {
+					stderr: true,
+					stdout: true,
+				});
+				return new Ipc.Subprocess(this.app, subprocess);
 			});
 	}
 

packages/evm-api-worker/source/service-provider.ts

@@ -13,8 +13,11 @@ export class ServiceProvider extends Providers.ServiceProvider {
 
 	public async register(): Promise<void> {
 		this.app.bind<() => Ipc.Subprocess<any>>(Identifiers.Evm.WorkerSubprocess.Factory).toFactory(() => () => {
-			const subprocess = new Worker(`${new URL(".", import.meta.url).pathname}/worker-script.js`, {});
-			return new Ipc.Subprocess(subprocess);
+			const subprocess = new Worker(`${new URL(".", import.meta.url).pathname}/worker-script.js`, {
+				stderr: true,
+				stdout: true,
+			});
+			return new Ipc.Subprocess(this.app, subprocess);
 		});
 
 		this.app.bind(Identifiers.Evm.Worker).toConstantValue(this.app.resolve(WorkerInstance));

packages/evm-service/source/instances/evm.ts

@@ -21,7 +21,7 @@ export class EvmInstance implements Contracts.Evm.Instance, Contracts.Evm.Storag
 
 	@postConstruct()
 	public initialize() {
-		const logPrefix = `${this.constructor.name}`;
+		const logPrefix = `evm`;
 
 		this.#evm = new Evm({
 			historySize: 256n,

packages/kernel/package.json

@@ -42,13 +42,15 @@
 		"log-process-errors": "12.0.0",
 		"nanomatch": "1.2.13",
 		"nsfw": "2.2.5",
-		"semver": "7.7.1"
+		"semver": "7.7.1",
+		"split2": "4.2.0"
 	},
 	"devDependencies": {
 		"@types/capture-console": "1.0.5",
 		"@types/fs-extra": "11.0.4",
 		"@types/log-process-errors": "6.3.1",
 		"@types/semver": "7.5.8",
+		"@types/split2": "4.2.3",
 		"@types/tmp": "0.2.6",
 		"capture-console": "1.0.2",
 		"moment-timezone": "0.5.47",

packages/kernel/source/ipc/sub-process.ts

@@ -1,4 +1,5 @@
-import { Contracts } from "@mainsail/contracts";
+import { Contracts, Identifiers } from "@mainsail/contracts";
+import split from "split2";
 import { Worker } from "worker_threads";
 
 export class Subprocess<T extends Record<string, any>> implements Contracts.Kernel.IPC.Subprocess<T> {
@@ -7,10 +8,34 @@ export class Subprocess<T extends Record<string, any>> implements Contracts.Kern
 	private readonly callbacks = new Map<number, Contracts.Kernel.IPC.RequestCallbacks<T>>();
 	private readonly eventHandlers = new Map<string, Contracts.Kernel.IPC.EventCallback<any>>();
 
-	public constructor(subprocess: Worker) {
+	public constructor(app: Contracts.Kernel.Application, subprocess: Worker) {
 		this.subprocess = subprocess;
 		this.subprocess.on("message", this.onSubprocessMessage.bind(this));
 		this.subprocess.on("message", this.onEmit.bind(this));
+
+		const logger = app.get<Contracts.Kernel.Logger>(Identifiers.Services.Log.Service);
+
+		this.subprocess.stdout.pipe(split()).on("data", (line) => {
+			// [LEVEL] MESSAGE
+			const match = line.match(/^\[(\w+)]\s+(.*)$/);
+			if (!match) {
+				// Fallback to normal console.log if output doesn't match expected format.
+				// For example, this is the case when worker uses `console.log` directly instead of logger service.
+				console.log(line);
+				return;
+			}
+
+			const [, level, message] = match;
+			if (logger.isValidLevel(level)) {
+				logger[level](message);
+			} else {
+				logger.warning(`[unknown:${level}] ${message}`);
+			}
+		});
+
+		this.subprocess.stderr.pipe(split()).on("data", (line) => {
+			logger.error(line);
+		});
 	}
 
 	public async kill(): Promise<number> {

packages/kernel/source/services/log/drivers/memory.ts

@@ -58,6 +58,10 @@ export class MemoryLogger implements Contracts.Kernel.Logger {
 		this.log("debug", message);
 	}
 
+	public isValidLevel(level: string): boolean {
+		return !!this.levelStyles[level];
+	}
+
 	public suppressConsoleOutput(suppress: boolean): void {
 		this.silentConsole = suppress;
 	}

packages/kernel/source/services/log/drivers/null.ts

@@ -39,6 +39,10 @@ export class NullLogger implements Contracts.Kernel.Logger {
 		//
 	}
 
+	public isValidLevel(level: string): boolean {
+		return true;
+	}
+
 	public suppressConsoleOutput(suppress: boolean): void {
 		//
 	}

packages/logger-pino/source/driver.ts

@@ -15,6 +15,8 @@ import { inspect } from "util";
 
 @injectable()
 export class PinoLogger implements Contracts.Kernel.Logger {
+	static LOG_LEVELS = new Set(["emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"]);
+
 	@inject(Identifiers.Application.Instance)
 	private readonly app!: Contracts.Kernel.Application;
 
@@ -39,6 +41,22 @@ export class PinoLogger implements Contracts.Kernel.Logger {
 
 	public async make(options?: any): Promise<Contracts.Kernel.Logger> {
 		this.#stream = new PassThrough();
+
+		if (options.workerMode) {
+			this.#logger = Object.fromEntries(
+				[...PinoLogger.LOG_LEVELS].map((level) => [
+					level,
+					(message: string) => {
+						process.stdout.write(`[${level}] (${this.app.thread()}) ${message}\n`);
+					},
+				]),
+			) as unknown as pino.Logger<
+				"alert" | "critical" | "debug" | "emergency" | "error" | "info" | "notice" | "warning"
+			>;
+
+			return this;
+		}
+
 		this.#logger = pino.default(
 			{
 				base: null,
@@ -64,7 +82,7 @@ export class PinoLogger implements Contracts.Kernel.Logger {
 			this.#stream,
 		);
 
-		if (this.#isValidLevel(options.levels.console)) {
+		if (this.isValidLevel(options.levels.console)) {
 			pump(
 				this.#stream,
 				split(),
@@ -77,7 +95,7 @@ export class PinoLogger implements Contracts.Kernel.Logger {
 			);
 		}
 
-		if (this.#isValidLevel(options.levels.file)) {
+		if (this.isValidLevel(options.levels.file)) {
 			this.#combinedFileStream = new pumpify(
 				split(),
 				this.#createPrettyTransport(options.levels.file, { colorize: false }),
@@ -164,14 +182,8 @@ export class PinoLogger implements Contracts.Kernel.Logger {
 	}
 
 	#createPrettyTransport(level: string, prettyOptions?: PrettyOptions): Transform {
-		const thread = this.app.thread();
-		const ignore = this.app.isWorker() ? `` : `pid`;
-
 		const pinoPretty = prettyFactory({
-			customPrettifiers: {
-				pid: () => thread,
-			},
-			ignore,
+			ignore: "pid",
 			levelFirst: false,
 			translateTime: "yyyy-mm-dd HH:MM:ss.l",
 			...prettyOptions,
@@ -229,7 +241,7 @@ export class PinoLogger implements Contracts.Kernel.Logger {
 		);
 	}
 
-	#isValidLevel(level: string): boolean {
-		return ["emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"].includes(level);
+	public isValidLevel(level: string): boolean {
+		return PinoLogger.LOG_LEVELS.has(level);
 	}
 }

packages/logger-pino/source/service-provider.ts

@@ -12,7 +12,16 @@ export class ServiceProvider extends Providers.ServiceProvider {
 			Identifiers.Services.Log.Manager,
 		);
 
-		await logManager.extend("pino", async () => this.app.resolve<PinoLogger>(PinoLogger).make(this.config().all()));
+		await logManager.extend("pino", async () => {
+			const logger = this.app.resolve<PinoLogger>(PinoLogger);
+
+			if (this.app.thread() === "main") {
+				return logger.make(this.config().all());
+			}
+
+			// Log output from workers is piped to main logger via Ipc.Subprocess.
+			return logger.make({ workerMode: true });
+		});
 
 		logManager.setDefaultDriver("pino");
 	}