feat(core): collect resource usage

Author: leosvelperez

packages/nx/src/daemon/client/client.ts

@@ -623,14 +623,20 @@ export class DaemonClient {
     if (this._daemonStatus == DaemonStatus.DISCONNECTED) {
       this._daemonStatus = DaemonStatus.CONNECTING;
 
+      let daemonPid: number | null = null;
       if (!(await this.isServerAvailable())) {
-        await this.startInBackground();
+        daemonPid = await this.startInBackground();
       }
       this.setUpConnection();
       this._daemonStatus = DaemonStatus.CONNECTED;
       this._daemonReady();
+
+      daemonPid ??= getDaemonProcessIdSync();
+      await this.registerDaemonProcessWithMetricsService(daemonPid);
     } else if (this._daemonStatus == DaemonStatus.CONNECTING) {
       await this._waitForDaemonReady;
+      const daemonPid = getDaemonProcessIdSync();
+      await this.registerDaemonProcessWithMetricsService(daemonPid);
     }
     // An open promise isn't enough to keep the event loop
     // alive, so we set a timeout here and clear it when we hear
@@ -649,6 +655,23 @@ export class DaemonClient {
     });
   }
 
+  private async registerDaemonProcessWithMetricsService(
+    daemonPid: number | null
+  ) {
+    if (!daemonPid) {
+      return;
+    }
+
+    try {
+      const { getProcessMetricsService } = await import(
+        '../../tasks-runner/process-metrics-service'
+      );
+      getProcessMetricsService().registerDaemonProcess(daemonPid);
+    } catch {
+      // don't error, this is a secondary concern that should not break task execution
+    }
+  }
+
   private retryMessageAfterNewDaemonStarts() {
     const [msg, res, rej] = [
       this.currentMessage,

packages/nx/src/executors/run-commands/run-commands.impl.ts

@@ -2,6 +2,7 @@ import * as yargsParser from 'yargs-parser';
 import { ExecutorContext } from '../../config/misc-interfaces';
 import { isTuiEnabled } from '../../tasks-runner/is-tui-enabled';
 import { PseudoTerminal } from '../../tasks-runner/pseudo-terminal';
+import { createTaskId } from '../../tasks-runner/utils';
 import { NoopChildProcess } from '../../tasks-runner/running-tasks/noop-child-process';
 import {
   ParallelRunningTasks,
@@ -97,7 +98,8 @@ export default async function (
 
 export async function runCommands(
   options: RunCommandsOptions,
-  context: ExecutorContext
+  context: ExecutorContext,
+  taskId?: string
 ) {
   const normalized = normalizeOptions(options);
 
@@ -138,11 +140,27 @@ export async function runCommands(
   const tuiEnabled = isTuiEnabled();
 
   try {
+    const resolvedTaskId =
+      taskId ??
+      createTaskId(
+        context.projectName,
+        context.targetName,
+        context.configurationName
+      );
     const runningTask = isSingleCommandAndCanUsePseudoTerminal
-      ? await runSingleCommandWithPseudoTerminal(normalized, context)
+      ? await runSingleCommandWithPseudoTerminal(
+          normalized,
+          context,
+          resolvedTaskId
+        )
       : options.parallel
-      ? new ParallelRunningTasks(normalized, context)
-      : new SeriallyRunningTasks(normalized, context, tuiEnabled);
+      ? new ParallelRunningTasks(normalized, context, resolvedTaskId)
+      : new SeriallyRunningTasks(
+          normalized,
+          context,
+          tuiEnabled,
+          resolvedTaskId
+        );
     return runningTask;
   } catch (e) {
     if (process.env.NX_VERBOSE_LOGGING === 'true') {

packages/nx/src/executors/run-commands/running-tasks.ts

@@ -14,6 +14,7 @@ import {
   loadAndExpandDotEnvFile,
   unloadDotEnvFile,
 } from '../../tasks-runner/task-env';
+import { getProcessMetricsService } from '../../tasks-runner/process-metrics-service';
 import { signalToCode } from '../../utils/exit-codes';
 import {
   LARGE_BUFFER,
@@ -30,7 +31,11 @@ export class ParallelRunningTasks implements RunningTask {
     [];
   private outputCallbacks: Array<(terminalOutput: string) => void> = [];
 
-  constructor(options: NormalizedRunCommandsOptions, context: ExecutorContext) {
+  constructor(
+    options: NormalizedRunCommandsOptions,
+    context: ExecutorContext,
+    taskId: string
+  ) {
     this.childProcesses = options.commands.map(
       (commandConfig) =>
         new RunningNodeProcess(
@@ -40,7 +45,8 @@ export class ParallelRunningTasks implements RunningTask {
           options.env ?? {},
           options.readyWhenStatus,
           options.streamOutput,
-          options.envFile
+          options.envFile,
+          taskId
         )
     );
     this.readyWhenStatus = options.readyWhenStatus;
@@ -228,7 +234,8 @@ export class SeriallyRunningTasks implements RunningTask {
   constructor(
     options: NormalizedRunCommandsOptions,
     context: ExecutorContext,
-    private readonly tuiEnabled: boolean
+    private readonly tuiEnabled: boolean,
+    private readonly taskId: string
   ) {
     this.run(options, context)
       .catch((e) => {
@@ -279,6 +286,7 @@ export class SeriallyRunningTasks implements RunningTask {
         options.color,
         calculateCwd(options.cwd, context),
         options.processEnv ?? options.env ?? {},
+        this.taskId,
         options.usePty,
         options.streamOutput,
         options.tty,
@@ -314,6 +322,7 @@ export class SeriallyRunningTasks implements RunningTask {
     color: boolean,
     cwd: string,
     env: Record<string, string>,
+    taskId: string,
     usePty: boolean = true,
     streamOutput: boolean = true,
     tty: boolean,
@@ -330,7 +339,7 @@ export class SeriallyRunningTasks implements RunningTask {
       const pseudoTerminal = createPseudoTerminal();
       registerProcessListener(this, pseudoTerminal);
 
-      return createProcessWithPseudoTty(
+      const pseudoTtyProcess = await createProcessWithPseudoTty(
         pseudoTerminal,
         commandConfig,
         color,
@@ -340,6 +349,15 @@ export class SeriallyRunningTasks implements RunningTask {
         tty,
         envFile
       );
+
+      // Register process for metrics collection (direct run-commands execution)
+      // Skip registration if we're in a forked executor - the fork wrapper already registered
+      const pid = pseudoTtyProcess.getPid();
+      if (pid && !process.env.NX_FORKED_TASK_EXECUTOR) {
+        getProcessMetricsService().registerTaskProcess(taskId, pid);
+      }
+
+      return pseudoTtyProcess;
     }
 
     return new RunningNodeProcess(
@@ -349,7 +367,8 @@ export class SeriallyRunningTasks implements RunningTask {
       env,
       [],
       streamOutput,
-      envFile
+      envFile,
+      taskId
     );
   }
 }
@@ -369,7 +388,8 @@ class RunningNodeProcess implements RunningTask {
     env: Record<string, string>,
     private readyWhenStatus: { stringToMatch: string; found: boolean }[],
     streamOutput = true,
-    envFile: string
+    envFile: string,
+    private taskId: string
   ) {
     env = processEnv(color, cwd, env, envFile);
     this.command = commandConfig.command;
@@ -384,6 +404,15 @@ class RunningNodeProcess implements RunningTask {
       windowsHide: false,
     });
 
+    // Register process for metrics collection
+    // Skip registration if we're in a forked executor - the fork wrapper already registered
+    if (this.childProcess.pid && !process.env.NX_FORKED_TASK_EXECUTOR) {
+      getProcessMetricsService().registerTaskProcess(
+        this.taskId,
+        this.childProcess.pid
+      );
+    }
+
     this.addListeners(commandConfig, streamOutput);
   }
 
@@ -508,7 +537,8 @@ class RunningNodeProcess implements RunningTask {
 
 export async function runSingleCommandWithPseudoTerminal(
   normalized: NormalizedRunCommandsOptions,
-  context: ExecutorContext
+  context: ExecutorContext,
+  taskId: string
 ): Promise<PseudoTtyProcess> {
   const pseudoTerminal = createPseudoTerminal();
   const pseudoTtyProcess = await createProcessWithPseudoTty(
@@ -521,6 +551,14 @@ export async function runSingleCommandWithPseudoTerminal(
     pseudoTerminal ? normalized.isTTY : false,
     normalized.envFile
   );
+
+  // Register process for metrics collection (direct run-commands execution)
+  // Skip registration if we're in a forked executor - the fork wrapper already registered
+  const pid = pseudoTtyProcess.getPid();
+  if (pid && !process.env.NX_FORKED_TASK_EXECUTOR) {
+    getProcessMetricsService().registerTaskProcess(taskId, pid);
+  }
+
   registerProcessListener(pseudoTtyProcess, pseudoTerminal);
   return pseudoTtyProcess;
 }

packages/nx/src/native/index.d.ts

@@ -27,6 +27,7 @@ export declare class AppLifeCycle {
 
 export declare class ChildProcess {
   getParserAndWriter(): ExternalObject<[ParserArc, WriterArc]>
+  getPid(): number
   kill(signal?: NodeJS.Signals): void
   onExit(callback: (message: string) => void): void
   onOutput(callback: (message: string) => void): void
@@ -92,6 +93,38 @@ export declare class NxTaskHistory {
   getEstimatedTaskTimings(targets: Array<TaskTarget>): Record<string, number>
 }
 
+/**
+ * NAPI wrapper for the process metrics collector
+ * Provides a JavaScript-friendly interface for metrics collection
+ */
+export declare class ProcessMetricsCollector {
+  /** Create a new metrics collector with default configuration */
+  constructor()
+  /**
+   * Start metrics collection
+   * Idempotent - safe to call multiple times
+   */
+  startCollection(): void
+  /**
+   * Stop metrics collection
+   * Returns true if collection was stopped, false if not running
+   */
+  stopCollection(): boolean
+  /** Register the main CLI process for metrics collection */
+  registerMainCliProcess(pid: number): void
+  /** Register the daemon process for metrics collection */
+  registerDaemonProcess(pid: number): void
+  /**
+   * Register a process for a specific task
+   * Automatically creates the task if it doesn't exist
+   */
+  registerTaskProcess(taskId: string, pid: number): void
+  /** Register a batch with multiple tasks sharing a worker */
+  registerBatch(batchId: string, taskIds: Array<string>, pid: number): void
+  /** Subscribe to push-based metrics notifications from TypeScript */
+  subscribe(callback: (err: Error | null, event: MetricsUpdate) => void): void
+}
+
 export declare class RunningTasksService {
   constructor(db: ExternalObject<NxDbConnection>)
   getRunningTasks(ids: Array<string>): Array<string>
@@ -153,6 +186,13 @@ export declare class WorkspaceContext {
   getFilesInDirectory(directory: string): Array<string>
 }
 
+/** Batch metrics snapshot */
+export interface BatchMetricsSnapshot {
+  batchId: string
+  taskIds: Array<string>
+  processes: Array<ProcessMetrics>
+}
+
 export interface CachedResult {
   code: number
   terminalOutput?: string
@@ -170,6 +210,12 @@ export declare export declare function connectToNxDb(cacheDir: string, nxVersion
 
 export declare export declare function copy(src: string, dest: string): number
 
+/** Daemon metrics with main process and subprocesses */
+export interface DaemonMetrics {
+  main: ProcessMetrics
+  subprocesses: Array<ProcessMetrics>
+}
+
 export interface DepsOutputsInput {
   dependentTasksOutputFiles: string
   transitive?: boolean
@@ -264,6 +310,20 @@ export declare export declare function isEditorInstalled(editor: SupportedEditor
 
 export declare export declare function logDebug(message: string): void
 
+/** Metrics data for collection cycle */
+export interface MetricsData {
+  timestamp: number
+  mainCliProcess?: ProcessSnapshot
+  daemonProcesses: Array<ProcessSnapshot>
+  tasks: Record<string, Array<ProcessSnapshot>>
+}
+
+/** Metrics update sent every collection cycle */
+export interface MetricsUpdate {
+  metrics: ProcessMetricsSnapshot
+  metadata?: Record<string, ProcessMetadata>
+}
+
 /** Stripped version of the NxJson interface for use in rust */
 export interface NxJson {
   namedInputs?: Record<string, Array<JsInputs>>
@@ -283,6 +343,43 @@ export interface NxWorkspaceFilesExternals {
 
 export declare export declare function parseTaskStatus(stringStatus: string): TaskStatus
 
+/** Process metadata (static, doesn't change during process lifetime) */
+export interface ProcessMetadata {
+  ppid: number
+  name: string
+  command: string
+  exePath: string
+  cwd: string
+}
+
+/** Process metrics (dynamic, changes every collection) */
+export interface ProcessMetrics {
+  pid: number
+  cpu: number
+  memory: number
+}
+
+/** Organized collection of process metrics with timestamp */
+export interface ProcessMetricsSnapshot {
+  timestamp: number
+  mainCli?: ProcessMetrics
+  daemon?: DaemonMetrics
+  tasks: Record<string, Array<ProcessMetrics>>
+  batches: Record<string, BatchMetricsSnapshot>
+}
+
+/** Process snapshot with full metadata and metrics */
+export interface ProcessSnapshot {
+  pid: number
+  ppid: number
+  name: string
+  command: string
+  exePath: string
+  cwd: string
+  cpu: number
+  memory: number
+}
+
 export interface Project {
   root: string
   namedInputs?: Record<string, Array<JsInputs>>