diff --git a/packages/cluster/package.json b/packages/cluster/package.json index 6899ec7b89..90d44b67fc 100644 --- a/packages/cluster/package.json +++ b/packages/cluster/package.json @@ -54,6 +54,7 @@ "egg-logger": "catalog:", "get-ready": "catalog:", "graceful-process": "catalog:", + "onelogger": "^1.0.1", "sendmessage": "catalog:", "terminal-link": "catalog:", "utility": "catalog:" diff --git a/packages/cluster/src/app_worker.ts b/packages/cluster/src/app_worker.ts index 9d1cffdf09..185b455c96 100644 --- a/packages/cluster/src/app_worker.ts +++ b/packages/cluster/src/app_worker.ts @@ -8,6 +8,7 @@ import { debuglog } from 'node:util'; import { importModule } from '@eggjs/utils'; import { EggConsoleLogger as ConsoleLogger } from 'egg-logger'; +import { ipcLogger, formatIpcMessage, internalIpcLogEnabled } from './utils/ipc_logger.ts'; import { BaseAppWorker } from './utils/mode/base/app.ts'; import { AppProcessWorker } from './utils/mode/impl/process/app.ts'; import { AppThreadWorker } from './utils/mode/impl/worker_threads/app.ts'; @@ -48,6 +49,25 @@ async function main() { AppWorker = AppThreadWorker as any; } else { AppWorker = AppProcessWorker as any; + // D. master -> app (recv): log every IPC message delivered to this worker via the cluster channel. + // Handle is present when master forwards a net.Socket (sticky-session mode). + // Read-only listener; other `process.on('message')` listeners (framework, sticky handler, etc.) + // are unaffected. + process.on('message', (msg: any, handle: any) => { + const body = typeof msg === 'string' ? { action: msg } : msg; + ipcLogger.info(formatIpcMessage(`app#${process.pid}<-master`, body, handle)); + }); + + // F. master -> app internal NODE_CLUSTER messages (newconn with fd, disconnect, suicide, ...). + // `internalMessage` is an undocumented but stable Node.js event. + // Opt-in via EGG_CLUSTER_IPC_LOG because `newconn` fires once per HTTP request. + if (internalIpcLogEnabled) { + process.on('internalMessage', (msg: { cmd?: string; act?: string; ack?: number }, handle: unknown) => { + if (!msg || msg.cmd !== 'NODE_CLUSTER') return; + const label = msg.act ? `cluster:${msg.act}` : `cluster:ack#${msg.ack ?? '?'}`; + ipcLogger.info(formatIpcMessage(`app#${process.pid}<-master`, { action: label, data: msg }, handle)); + }); + } } const consoleLogger = new ConsoleLogger({ diff --git a/packages/cluster/src/master.ts b/packages/cluster/src/master.ts index abcd45ff9b..c5bef8ac86 100644 --- a/packages/cluster/src/master.ts +++ b/packages/cluster/src/master.ts @@ -15,6 +15,7 @@ import terminalLink from 'terminal-link'; import { readJSONSync } from 'utility'; import { ClusterWorkerExceptionError } from './error/ClusterWorkerExceptionError.ts'; +import { ipcLogger, formatIpcMessage } from './utils/ipc_logger.ts'; import { Messenger } from './utils/messenger.ts'; import { AgentProcessWorker, AgentProcessUtils as ProcessAgentWorker } from './utils/mode/impl/process/agent.ts'; import { AppProcessWorker, AppProcessUtils as ProcessAppWorker } from './utils/mode/impl/process/app.ts'; @@ -291,6 +292,11 @@ export class Master extends ReadyEventEmitter { connection.destroy(); } else { const worker = this.stickyWorker(connection.remoteAddress) as AppProcessWorker; + // A'. master -> app sticky-session direct send (bypasses AppProcessWorker#send), + // carries a net.Socket handle — safe-printed by formatIpcMessage's replacer. + ipcLogger.info( + formatIpcMessage(`master->app#${worker.workerId}`, { action: 'sticky-session:connection' }, connection), + ); worker.instance.send('sticky-session:connection', connection); } }, diff --git a/packages/cluster/src/utils/ipc_logger.ts b/packages/cluster/src/utils/ipc_logger.ts new file mode 100644 index 0000000000..b8352dec9d --- /dev/null +++ b/packages/cluster/src/utils/ipc_logger.ts @@ -0,0 +1,86 @@ +import net from 'node:net'; + +import { getLogger } from 'onelogger'; + +import type { MessageBody } from './messenger.ts'; + +/** + * onelogger instance for Node.js cluster module IPC traffic between master and app workers. + * Users can override the underlying sink via `onelogger.setLogger()` / `setCustomLogger()`. + */ +export const ipcLogger = getLogger('egg/cluster/ipc'); + +/** + * Whether internal-level cluster IPC logs (NODE_CLUSTER newconn/accepted/listening/...) are enabled. + * These are very verbose (one `cluster:newconn` per HTTP request), so they are opt-in + * via the `EGG_CLUSTER_IPC_LOG` environment variable (any truthy value enables). + */ +export const internalIpcLogEnabled = !!process.env.EGG_CLUSTER_IPC_LOG; + +const MAX_STRING_LEN = 200; +const MAX_TOTAL_LEN = 1024; + +function describeHandle(handle: unknown): string { + if (handle == null) return ''; + if (handle instanceof net.Socket) { + const fd = (handle as unknown as { _handle?: { fd?: number } })._handle?.fd; + return fd != null ? `` : ''; + } + if (handle instanceof net.Server) { + return ''; + } + const ctor = (handle as { constructor?: { name?: string } })?.constructor?.name; + return ctor ? `<${ctor}>` : ''; +} + +function makeReplacer() { + const seen = new WeakSet(); + return function replacer(_key: string, value: unknown) { + if (value && typeof value === 'object') { + if (seen.has(value as object)) return ''; + seen.add(value as object); + if (value instanceof net.Socket) return describeHandle(value); + if (value instanceof net.Server) return ''; + if (Buffer.isBuffer(value)) return ``; + } + if (typeof value === 'string' && value.length > MAX_STRING_LEN) { + return `${value.slice(0, MAX_STRING_LEN)}...(truncated)`; + } + return value; + }; +} + +function stringifyData(data: unknown): string { + let out: string; + try { + out = JSON.stringify(data, makeReplacer()); + } catch (err) { + return ``; + } + if (out && out.length > MAX_TOTAL_LEN) { + out = `${out.slice(0, MAX_TOTAL_LEN)}...(truncated)`; + } + return out; +} + +/** + * Format a single IPC message into a one-line log string. + * @param direction e.g. 'master->app#12345' / 'app#12345<-master' + * @param msg the message body (supports cluster internal msgs via `action: 'cluster:'`) + * @param handle optional handle (net.Socket / net.Server / TCP) attached to the IPC message + */ +export function formatIpcMessage( + direction: string, + msg: Partial & { action?: string; data?: unknown }, + handle?: unknown, +): string { + const action = msg?.action ?? ''; + let out = `[${direction}] action=${action}`; + if (msg && msg.data !== undefined) { + out += ` data=${stringifyData(msg.data)}`; + } + if (handle) { + out += ` +handle=${describeHandle(handle)}`; + } + return out; +} diff --git a/packages/cluster/src/utils/mode/impl/process/app.ts b/packages/cluster/src/utils/mode/impl/process/app.ts index 0c9895a94c..d45a034d42 100644 --- a/packages/cluster/src/utils/mode/impl/process/app.ts +++ b/packages/cluster/src/utils/mode/impl/process/app.ts @@ -5,6 +5,7 @@ import { cfork } from 'cfork'; import { graceful as gracefulExit, type Options as gracefulExitOptions } from 'graceful-process'; import { sendmessage } from 'sendmessage'; +import { ipcLogger, formatIpcMessage, internalIpcLogEnabled } from '../../../ipc_logger.ts'; import type { MessageBody } from '../../../messenger.ts'; import { terminate } from '../../../terminate.ts'; import { BaseAppWorker, BaseAppUtils } from '../../base/app.ts'; @@ -29,6 +30,7 @@ export class AppProcessWorker extends BaseAppWorker { } send(message: MessageBody): void { + ipcLogger.info(formatIpcMessage(`master->app#${this.workerId}`, message)); sendmessage(this.instance, message); } @@ -48,6 +50,7 @@ export class AppProcessWorker extends BaseAppWorker { static send(message: MessageBody): void { message.senderWorkerId = String(process.pid); + ipcLogger.info(formatIpcMessage(`app#${process.pid}->master`, message)); // cluster won't get `listening` event when reusePort is true, // use cluster `message` event instead if (message.action === 'app-start' && message.reusePort) { @@ -90,7 +93,10 @@ export class AppProcessUtils extends BaseAppUtils { const appWorker = new AppProcessWorker(worker); this.emit('worker_forked', appWorker); appWorker.disableRefork = true; - worker.on('message', (msg) => { + // B. master <- app (recv). Handle is present when master receives a net.Socket + // (e.g. forwarded sticky-session connection). + // Log AFTER forwarding to avoid adding log-serialization latency to the forward path. + worker.on('message', (msg, handle) => { if (typeof msg === 'string') { msg = { action: msg, @@ -99,7 +105,21 @@ export class AppProcessUtils extends BaseAppUtils { } msg.from = 'app'; this.messenger.send(msg); + ipcLogger.info(formatIpcMessage(`master<-app#${worker.process.pid}`, msg, handle)); }); + + // E. cluster internal NODE_CLUSTER messages from worker to master: + // listening / online / queryServer / accepted (fd ack) / close / ... + // Must hook on `worker.process` (ChildProcess) — `cluster.Worker` doesn't forward `internalMessage`. + // `internalMessage` is undocumented but stable across Node.js versions. + // Opt-in via EGG_CLUSTER_IPC_LOG because this is verbose under load. + if (internalIpcLogEnabled) { + worker.process.on('internalMessage', (msg: { cmd?: string; act?: string; ack?: number }, handle: unknown) => { + if (!msg || msg.cmd !== 'NODE_CLUSTER') return; + const label = msg.act ? `cluster:${msg.act}` : `cluster:ack#${msg.ack ?? '?'}`; + ipcLogger.info(formatIpcMessage(`master<-app#${worker.process.pid}`, { action: label, data: msg }, handle)); + }); + } this.log( '[master] app_worker#%s:%s start, state: %s, current workers: %j', appWorker.id,