Runtime memory Metrics (#1103)

josephjclark · web-flow · commit 17e32d6ea0e6 · 2025-10-28T18:21:00.000Z
* Add main thread memory stats to worker

* changeset

* add polling

* don't log condition code

* changeset

* add profiling to steps

* add profile and interval options to the worker

* changeset

* runtime: add profiler for the workflow as a whole

* types

* versions: worker@1.19.0
diff --git a/packages/cli/CHANGELOG.md b/packages/cli/CHANGELOG.md
@@ -1,5 +1,12 @@
 # @openfn/cli
 
+## 1.18.1
+
+### Patch Changes
+
+- Updated dependencies [3cd3cd5]
+  - @openfn/runtime@1.7.4
+
 ## 1.18.0
 
 ### Minor Changes
diff --git a/packages/cli/package.json b/packages/cli/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@openfn/cli",
-  "version": "1.18.0",
+  "version": "1.18.1",
   "description": "CLI devtools for the OpenFn toolchain",
   "engines": {
     "node": ">=18",
diff --git a/packages/engine-multi/CHANGELOG.md b/packages/engine-multi/CHANGELOG.md
@@ -1,5 +1,16 @@
 # engine-multi
 
+## 1.8.0
+
+### Minor Changes
+
+- 475f8cf: Add options to profile memory usage in workflows. This should give a more accurate reading of the memory used by a run, but may have a small impact on performance
+
+### Patch Changes
+
+- Updated dependencies [3cd3cd5]
+  - @openfn/runtime@1.7.4
+
 ## 1.7.1
 
 ### Patch Changes
diff --git a/packages/engine-multi/package.json b/packages/engine-multi/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@openfn/engine-multi",
-  "version": "1.7.1",
+  "version": "1.8.0",
   "description": "Multi-process runtime engine",
   "main": "dist/index.js",
   "type": "module",
diff --git a/packages/engine-multi/src/api.ts b/packages/engine-multi/src/api.ts
@@ -63,6 +63,8 @@ const createAPI = async function (
       'configuration',
       'response',
     ],
+    profile: options.profile,
+    profilePollInterval: options.profilePollInterval,
   };
 
   logger.info(`memory limit set to ${options.memoryLimitMb}mb`);
diff --git a/packages/engine-multi/src/api/execute.ts b/packages/engine-multi/src/api/execute.ts
@@ -43,6 +43,8 @@ const execute = async (context: ExecutionContext) => {
       whitelist,
       jobLogLevel: options.jobLogLevel,
       repoDir: options.repoDir,
+      profile: context.options.profile,
+      profilePollInteval: context.options.profilePollInterval,
     } as RunOptions;
 
     const workerOptions = {
diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts
@@ -27,10 +27,10 @@ import type {
 import type { AutoinstallOptions } from './api/autoinstall';
 
 const DEFAULT_RUN_TIMEOUT = 1000 * 60 * 10; // ms
-
 const DEFAULT_MEMORY_LIMIT_MB = 500;
-
 const DEFAULT_PAYLOAD_LIMIT_MB = 10;
+const DEFAULT_PROFILE = false;
+const DEFAULT_PROFILE_POLL_INTERVAL = 10;
 
 // For each workflow, create an API object with its own event emitter
 // this is a bit weird - what if the emitter went on state instead?
@@ -84,6 +84,9 @@ export type EngineOptions = {
 
   workerValidationTimeout?: number;
   workerValidationRetries?: number;
+
+  profile?: boolean;
+  profilePollInterval?: number;
 };
 
 export type InternalEngine = RuntimeEngine & {
@@ -106,6 +109,9 @@ const createEngine = async (
   const defaultMemoryLimit = options.memoryLimitMb || DEFAULT_MEMORY_LIMIT_MB;
   const defaultPayloadLimit =
     options.payloadLimitMb || DEFAULT_PAYLOAD_LIMIT_MB;
+  const defaultProfile = options.profile ?? DEFAULT_PROFILE;
+  const defaultProfilePollInterval =
+    options.profilePollInterval ?? DEFAULT_PROFILE_POLL_INTERVAL;
 
   let resolvedWorkerPath;
   if (workerPath) {
@@ -165,6 +171,8 @@ const createEngine = async (
         memoryLimitMb: opts.memoryLimitMb ?? defaultMemoryLimit,
         payloadLimitMb: opts.payloadLimitMb ?? defaultPayloadLimit,
         jobLogLevel: opts.jobLogLevel,
+        profile: defaultProfile,
+        profilePollInterval: defaultProfilePollInterval,
       },
     });
 
diff --git a/packages/engine-multi/src/worker/thread/run.ts b/packages/engine-multi/src/worker/thread/run.ts
@@ -15,9 +15,11 @@ import { COMPILE_START, COMPILE_COMPLETE } from '../events';
 export type RunOptions = {
   repoDir: string;
   whitelist?: RegExp[];
-  sanitize: SanitizePolicies;
+  sanitize?: SanitizePolicies;
   statePropsToRemove?: string[];
   jobLogLevel?: LogLevel;
+  profile?: boolean;
+  profilePollInterval?: number;
 };
 
 const eventMap = {
@@ -29,8 +31,15 @@ const eventMap = {
 
 register({
   run: async (plan: ExecutionPlan, input: State, runOptions: RunOptions) => {
-    const { repoDir, whitelist, sanitize, statePropsToRemove, jobLogLevel } =
-      runOptions;
+    const {
+      repoDir,
+      whitelist,
+      sanitize,
+      statePropsToRemove,
+      jobLogLevel,
+      profile,
+      profilePollInterval,
+    } = runOptions;
     const { logger, jobLogger, adaptorLogger } = createLoggers(
       plan.id!,
       sanitize,
@@ -60,6 +69,8 @@ register({
         whitelist,
         cacheKey: plan.id,
       },
+      profile,
+      profilePollInterval,
       statePropsToRemove,
       callbacks: {
         // TODO: this won't actually work across the worker boundary
@@ -75,6 +86,7 @@ register({
         },
       },
     };
+    logger.log(' >>', options.profile, options.profilePollInterval);
 
     return execute(plan.id!, async () => {
       const start = Date.now();
diff --git a/packages/lightning-mock/CHANGELOG.md b/packages/lightning-mock/CHANGELOG.md
@@ -1,5 +1,14 @@
 # @openfn/lightning-mock
 
+## 2.3.3
+
+### Patch Changes
+
+- Updated dependencies [3cd3cd5]
+- Updated dependencies [475f8cf]
+  - @openfn/runtime@1.7.4
+  - @openfn/engine-multi@1.8.0
+
 ## 2.3.2
 
 ### Patch Changes
diff --git a/packages/lightning-mock/package.json b/packages/lightning-mock/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@openfn/lightning-mock",
-  "version": "2.3.2",
+  "version": "2.3.3",
   "private": true,
   "description": "A mock Lightning server",
   "main": "dist/index.js",
diff --git a/packages/runtime/CHANGELOG.md b/packages/runtime/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @openfn/runtime
 
+## 1.7.4
+
+### Patch Changes
+
+- 3cd3cd5: Don't log edge condition code
+
 ## 1.7.3
 
 ### Patch Changes
diff --git a/packages/runtime/package.json b/packages/runtime/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@openfn/runtime",
-  "version": "1.7.3",
+  "version": "1.7.4",
   "description": "Job processing runtime.",
   "type": "module",
   "exports": {
diff --git a/packages/runtime/src/execute/plan.ts b/packages/runtime/src/execute/plan.ts
@@ -7,6 +7,7 @@ import compilePlan from './compile-plan';
 import type { Options } from '../runtime';
 import validatePlan from '../util/validate-plan';
 import createErrorReporter from '../util/log-error';
+import createProfiler from '../util/profile-memory';
 import { NOTIFY_STATE_LOAD } from '../events';
 import { CompiledExecutionPlan, ExecutionContext } from '../types';
 
@@ -59,6 +60,11 @@ const executePlan = async (
   // count how many times each step has been called
   const counts: Record<string, number> = {};
 
+  let profiler = opts.profile
+    ? createProfiler(opts.profilePollInterval ?? 10)
+    : null;
+  profiler?.start();
+
   // Right now this executes in series, even if jobs are parallelised
   while (queue.length) {
     const { stepName, input: prevState } = queue.shift()!;
@@ -94,6 +100,11 @@ const executePlan = async (
     });
   }
 
+  if (profiler) {
+    const peak = profiler?.stop();
+    logger.debug(`Workflow complete: peak memory ${profiler.toMb(peak)}mb`);
+  }
+
   // If there are multiple leaf results, return them
   if (Object.keys(leaves).length > 1) {
     return leaves;
diff --git a/packages/runtime/src/execute/step.ts b/packages/runtime/src/execute/step.ts
@@ -4,7 +4,11 @@ import type { Logger } from '@openfn/logger';
 import executeExpression, { ExecutionErrorWrapper } from './expression';
 import clone from '../util/clone';
 import assembleState from '../util/assemble-state';
-import type { CompiledStep, ExecutionContext } from '../types';
+import type {
+  CompiledStep,
+  ExecutionContext,
+  NotifyJobCompletePayload,
+} from '../types';
 import { EdgeConditionError } from '../errors';
 import {
   NOTIFY_INIT_COMPLETE,
@@ -15,6 +19,7 @@ import {
 } from '../events';
 import { isNullState } from '../util/null-state';
 import sourcemapErrors from '../util/sourcemap-errors';
+import createProfiler from '../util/profile-memory';
 
 const loadCredentials = async (
   job: Job,
@@ -53,15 +58,15 @@ const calculateNext = (job: CompiledStep, result: any, logger: Logger) => {
           try {
             if (!edge.condition(result)) {
               logger.debug(
-                `Edge ${edge.condition.toString()} returned false; ${nextJobId} will NOT be executed`
+                `Edge condition returned false; ${nextJobId} will NOT be executed`
               );
               continue;
             }
           } catch (e: any) {
             throw new EdgeConditionError(e.message);
           }
           logger.debug(
-            `Edge ${edge.condition.toString()} returned true; ${nextJobId} will be executed next`
+            `Edge condition returned true; ${nextJobId} will be executed next`
           );
         }
       }
@@ -121,7 +126,13 @@ const executeStep = async (
   let result: any = input;
   let next: string[] = [];
   let didError = false;
+
   if (step.expression) {
+    let profiler = opts.profile
+      ? createProfiler(opts.profilePollInterval ?? 10)
+      : null;
+    profiler?.start();
+
     const job = step as Job;
     const jobId = job.id!;
     const jobName = job.name || job.id;
@@ -221,20 +232,28 @@ const executeStep = async (
 
       const humanJobMemory = Math.round(jobMemory / 1024 / 1024);
       const humanSystemMemory = Math.round(systemMemory / 1024 / 1024);
-      logger.debug(
-        `Final memory usage: [step ${humanJobMemory}mb] [system ${humanSystemMemory}mb]`
-      );
+
+      const mem: NotifyJobCompletePayload['mem'] = {
+        job: jobMemory,
+        system: systemMemory,
+      };
+
+      if (profiler) {
+        mem.peak = profiler.stop();
+        logger.debug(`Step memory usage: peak ${profiler.toMb(mem.peak)}mb`);
+      } else {
+        logger.debug(
+          `Step memory usage: [step ${humanJobMemory}mb] [system ${humanSystemMemory}mb]`
+        );
+      }
 
       next = calculateNext(step, result, logger);
       notify(NOTIFY_JOB_COMPLETE, {
         duration: Date.now() - duration,
         state: result,
         jobId,
         next,
-        mem: {
-          job: jobMemory,
-          system: systemMemory,
-        },
+        mem,
       });
     }
   } else {
diff --git a/packages/runtime/src/runtime.ts b/packages/runtime/src/runtime.ts
@@ -29,6 +29,12 @@ export type Options = {
   // SourceMap is only passed directly if the expression is a string
   // Usually a sourcemap is passed on a step
   sourceMap?: SourceMapWithOperations;
+
+  /** Profile memory usage through the run */
+  profile?: boolean;
+
+  /** Number of ms to wait before polling memory */
+  profilePollInterval?: number;
 };
 
 type RawOptions = Omit<Options, 'linker'> & {
diff --git a/packages/runtime/src/types.ts b/packages/runtime/src/types.ts
@@ -96,6 +96,7 @@ export type NotifyJobCompletePayload = {
   mem: {
     job: number;
     system: number;
+    peak?: number;
   };
 };
 
diff --git a/packages/runtime/src/util/profile-memory.ts b/packages/runtime/src/util/profile-memory.ts
@@ -0,0 +1,36 @@
+import { getHeapStatistics } from 'node:v8';
+
+/**
+ * Utility to profile memory usage
+ * Will constantly poll memory until stopped and
+ * then return the max tracked heap
+ */
+const profiler = (pollrate = 10) => {
+  let peak = -1;
+
+  let interval: NodeJS.Timeout;
+
+  const test = () => {
+    const now = getHeapStatistics().used_heap_size;
+    peak = Math.max(peak, now);
+  };
+
+  const convertToMb = (mem: number) =>
+    Math.round((mem / 1024 / 1024) * 100) / 100;
+
+  return {
+    start: () => {
+      interval = setInterval(test, pollrate);
+    },
+    stop: (toMb = false) => {
+      clearInterval(interval);
+      // Make one final test now that we've stopped
+      test();
+      return toMb ? convertToMb(peak) : peak;
+    },
+    peak: () => peak,
+    toMb: convertToMb,
+  };
+};
+
+export default profiler;
diff --git a/packages/runtime/test/execute/step.test.ts b/packages/runtime/test/execute/step.test.ts
diff --git a/packages/runtime/test/util/profile-memory.test.ts b/packages/runtime/test/util/profile-memory.test.ts
diff --git a/packages/ws-worker/CHANGELOG.md b/packages/ws-worker/CHANGELOG.md
diff --git a/packages/ws-worker/package.json b/packages/ws-worker/package.json
diff --git a/packages/ws-worker/src/api/claim.ts b/packages/ws-worker/src/api/claim.ts
diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts
diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts
diff --git a/packages/ws-worker/test/util/cli.test.ts b/packages/ws-worker/test/util/cli.test.ts

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@openfn/cli",`
`3`		`- "version": "1.18.0",`
	`3`	`+ "version": "1.18.1",`
`4`	`4`	`"description": "CLI devtools for the OpenFn toolchain",`
`5`	`5`	`"engines": {`
`6`	`6`	`"node": ">=18",`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@openfn/engine-multi",`
`3`		`- "version": "1.7.1",`
	`3`	`+ "version": "1.8.0",`
`4`	`4`	`"description": "Multi-process runtime engine",`
`5`	`5`	`"main": "dist/index.js",`
`6`	`6`	`"type": "module",`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@openfn/lightning-mock",`
`3`		`- "version": "2.3.2",`
	`3`	`+ "version": "2.3.3",`
`4`	`4`	`"private": true,`
`5`	`5`	`"description": "A mock Lightning server",`
`6`	`6`	`"main": "dist/index.js",`