Worker: add metadata to runs

.changeset/late-games-deny.md

@@ -0,0 +1,5 @@
+---
+'@openfn/ws-worker': minor
+---
+
+Pass a meta object to each run

.changeset/pink-bobcats-study.md

@@ -0,0 +1,5 @@
+---
+'@openfn/engine-multi': patch
+---
+
+Allow global scope context to be fed through to individual runs

packages/engine-multi/src/api/execute.ts

@@ -40,6 +40,7 @@ const execute = async (context: ExecutionContext) => {
     const whitelist = options.whitelist?.map((w) => w.toString());
 
     const runOptions = {
+      globals: options.globals,
       statePropsToRemove: options.statePropsToRemove,
       whitelist,
       jobLogLevel: options.jobLogLevel,

packages/engine-multi/src/engine.ts

@@ -178,6 +178,7 @@ const createEngine = async (
         payloadLimitMb: opts.payloadLimitMb ?? defaultPayloadLimit,
         logPayloadLimitMb: opts.logPayloadLimitMb ?? defaultLogPayloadLimit,
         jobLogLevel: opts.jobLogLevel,
+        globals: opts.globals,
         profile: defaultProfile,
         profilePollInterval: defaultProfilePollInterval,
       },

packages/engine-multi/src/types.ts

@@ -54,6 +54,8 @@ export type ExecuteOptions = {
   runTimeoutMs?: number;
   sanitize?: SanitizePolicies;
   jobLogLevel?: string;
+  // inject globals into the environment
+  globals?: any;
 };
 
 export type ExecutionContextOptions = ExecuteOptions & EngineOptions;

packages/engine-multi/src/worker/thread/run.ts

@@ -21,6 +21,7 @@ export type RunOptions = {
   profile?: boolean;
   profilePollInterval?: number;
   stateLimitMb?: number;
+  globals?: any;
 };
 
 const eventMap = {
@@ -41,6 +42,7 @@ register({
       profile,
       profilePollInterval,
       stateLimitMb,
+      globals,
     } = runOptions;
     const { logger, jobLogger, adaptorLogger } = createLoggers(
       plan.id!,
@@ -74,6 +76,7 @@ register({
       profilePollInterval,
       statePropsToRemove,
       stateLimitMb,
+      globals,
       callbacks: {
         // TODO: this won't actually work across the worker boundary
         // For now I am preloading credentials

packages/engine-multi/test/api/execute.test.ts

@@ -403,3 +403,29 @@ test.serial('should forward the jobLogLevel option', async (t) => {
   t.truthy(passedOptions);
   t.deepEqual(passedOptions.jobLogLevel, 'none');
 });
+
+test.serial('should forward the globals option', async (t) => {
+  let passedOptions: any;
+
+  const state = {
+    id: 'x',
+    plan,
+  } as WorkflowState;
+
+  const context = createContext({
+    state,
+    options: {
+      ...options,
+      globals: { x: 22 },
+    },
+  });
+  // @ts-ignore
+  context.callWorker = (_command, args) => {
+    passedOptions = args[2];
+  };
+
+  await execute(context);
+
+  t.truthy(passedOptions);
+  t.deepEqual(passedOptions.globals, { x: 22 });
+});

packages/runtime/src/runtime.ts

@@ -87,7 +87,6 @@ const run = (
   }
 
   const logger = opts.logger || defaultLogger;
-
   if (typeof xplan === 'string') {
     xplan = loadPlanFromString(
       xplan,

packages/ws-worker/src/util/convert-lightning-plan.ts

@@ -133,6 +133,14 @@ export default (
 
   // But some need to get passed down into the engine's options
   const engineOpts: WorkerRunOptions = {};
+
+  engineOpts.globals = {
+    meta: {
+      runId: run.id,
+      startTime: Date.now(),
+    },
+  };
+
   if (run.options) {
     if ('run_timeout_ms' in run.options) {
       engineOpts.runTimeoutMs = run.options.run_timeout_ms;

packages/ws-worker/test/util/convert-lightning-plan.test.ts

@@ -55,6 +55,15 @@ const createJob = (props = {}) => ({
   ...props,
 });
 
+// Pulls the generated globals out of the options object and checks them,
+// returning the remainder so callers can deepEqual the rest as before
+const checkGlobals = (t: any, options: any, runId: string) => {
+  const { globals, ...rest } = options;
+  t.is(globals.meta.runId, runId);
+  t.is(typeof globals.meta.startTime, 'number');
+  return rest;
+};
+
 test('convert a single job', (t) => {
   const run: Partial<LightningPlan> = {
     id: 'w',
@@ -115,7 +124,7 @@ test('convert a single job with options', (t) => {
       steps: [createJob()],
     },
   });
-  t.deepEqual(options, {
+  t.deepEqual(checkGlobals(t, options, 'w'), {
     runTimeoutMs: 10,
     memoryLimitMb: 500,
     payloadLimitMb: 20,
@@ -136,7 +145,7 @@ test('convert a single job with state_limit_mb', (t) => {
   };
   const { options } = convertPlan(run as LightningPlan);
 
-  t.deepEqual(options, {
+  t.deepEqual(checkGlobals(t, options, 'w'), {
     memoryLimitMb: 500,
     stateLimitMb: 50,
   });
@@ -165,7 +174,7 @@ test('convert a single job with log_payload_limit_mb', (t) => {
       steps: [createJob()],
     },
   });
-  t.deepEqual(options, {
+  t.deepEqual(checkGlobals(t, options, 'w'), {
     runTimeoutMs: 10,
     memoryLimitMb: 500,
     payloadLimitMb: 20,
@@ -192,7 +201,7 @@ test('convert a single job with data', (t) => {
       steps: [createJob({ state: { data: { x: 22 } } })],
     },
   });
-  t.deepEqual(options, {});
+  t.deepEqual(checkGlobals(t, options, 'w'), {});
 });
 
 test('Accept a partial run object', (t) => {
@@ -208,7 +217,7 @@ test('Accept a partial run object', (t) => {
       steps: [],
     },
   });
-  t.deepEqual(options, {});
+  t.deepEqual(checkGlobals(t, options, 'w'), {});
 });
 
 test('handle dataclip_id as input', (t) => {
@@ -241,7 +250,7 @@ test('handle output_dataclip as options', (t) => {
     },
   };
   const { options } = convertPlan(run as LightningPlan);
-  t.deepEqual(options, {
+  t.deepEqual(checkGlobals(t, options, 'w'), {
     outputDataclips: false,
   });
 });
@@ -747,3 +756,20 @@ test("ignore globals when it isn't a string", (t) => {
   const { plan } = convertPlan(run as LightningPlan);
   t.deepEqual(plan.workflow.globals, undefined);
 });
+
+test('sets runId and startTime on the engine options globals meta', (t) => {
+  const run: Partial<LightningPlan> = {
+    id: 'some-run-id',
+    jobs: [createNode()],
+    triggers: [],
+    edges: [],
+  };
+
+  const before = Date.now();
+  const { options } = convertPlan(run as LightningPlan);
+  const after = Date.now();
+
+  t.is(options.globals!.meta.runId, 'some-run-id');
+  t.true(options.globals!.meta.startTime >= before);
+  t.true(options.globals!.meta.startTime <= after);
+});

packages/ws-worker/test/worker.test.ts

@@ -42,8 +42,8 @@ const execute = async (plan: ExecutionPlan, input = {}, options = {}) =>
     const channel = mockChannel({
       [RUN_START]: async () => true,
       [STEP_START]: async () => true,
-      [RUN_LOG]: async (_evt) => {
-        //console.log(evt.source, evt.message)
+      [RUN_LOG]: async (evt) => {
+        console.log(evt.source, evt.message);
         return true;
       },
       [STEP_COMPLETE]: async () => true,