Merge pull request #420 from OpenFn/dockerize

Dockerize ws-worker

.dockerignore

@@ -0,0 +1,5 @@
+node_modules
+.git
+.gitignore
+*.md
+dist

.github/workflows/dockerize.yaml

@@ -0,0 +1,35 @@
+name: Build and push Docker images
+
+on:
+  push:
+    tags: '@openfn/ws-worker@**'
+
+jobs:
+  build-and-push-worker-image:
+    name: Push Docker image to Docker Hub
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Docker meta
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: openfn/ws-worker
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+      - name: Login to Docker Hub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Build and push
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          push: ${{ github.event_name != 'pull_request' }}
+          target: ws-worker
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

.tool-versions

@@ -1 +1 @@
-nodejs 18.18.2
+nodejs 18.12.1

Dockerfile

@@ -0,0 +1,30 @@
+FROM node:18-alpine AS base
+ENV PNPM_HOME="/pnpm"
+ENV PATH="$PNPM_HOME:$PATH"
+RUN corepack enable
+COPY . /app
+WORKDIR /app
+
+# TODO: use prod optimized build? ----------------------------------------------
+# FROM base AS build
+# RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile
+# RUN pnpm build
+# RUN pnpm deploy --filter=ws-worker --prod /prod/ws-worker
+
+# FROM base AS ws-worker
+# COPY --from=build /prod/ws-worker /prod/ws-worker
+# WORKDIR /prod/ws-worker
+# ------------------------------------------------------------------------------
+
+# TODO: remove simple build once prod optimized build is working ---------------
+FROM base AS ws-worker
+RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile
+RUN pnpm build
+WORKDIR /app/packages/ws-worker
+# ------------------------------------------------------------------------------
+
+EXPOSE 2222
+CMD [ "pnpm", "start:prod"]
+
+# TODO: determine how to pass in the -l from `docker run` for running on mac m1
+# CMD [ "pnpm", "start", "-l", "ws://host.docker.internal:4000"]

docker-compose.yaml

@@ -0,0 +1,8 @@
+services:
+  ws-worker:
+    build:
+      context: .
+    environment:
+      NODE_ENV: production
+    ports:
+      - 2222:2222

integration-tests/worker/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @openfn/integration-tests-worker
 
+## 1.0.2
+
+### Patch Changes
+
+- Updated dependencies
+  - @openfn/engine-multi@0.1.1
+  - @openfn/ws-worker@0.1.1
+  - @openfn/lightning-mock@1.0.2
+
 ## 1.0.1
 
 ### Patch Changes

integration-tests/worker/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@openfn/integration-tests-worker",
   "private": true,
-  "version": "1.0.1",
+  "version": "1.0.2",
   "description": "Lightning WOrker integration tests",
   "author": "Open Function Group <[email protected]>",
   "license": "ISC",

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,11 @@
 # engine-multi
 
+## 0.1.1
+
+### Patch Changes
+
+- Destroy workers on close
+
 ## 0.1.0
 
 First release of the multi-threaded runtime engine.

packages/engine-multi/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/engine-multi",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "description": "Multi-process runtime engine",
   "main": "dist/index.js",
   "type": "module",

packages/engine-multi/src/api.ts

@@ -72,6 +72,7 @@ const createAPI = async function (options: RTEOptions = {}) {
   return {
     execute: engine.execute,
     listen: engine.listen,
+    destroy: engine.destroy,
     on: (evt: string, fn: (...args: any[]) => void) => engine.on(evt, fn),
   };
 };

packages/engine-multi/src/api/call-worker.ts

@@ -59,7 +59,8 @@ export default function initWorkers(
     return promise;
   };
 
-  api.closeWorkers = () => workers.terminate();
+  // This will force termination instantly
+  api.closeWorkers = () => workers.terminate(true);
 }
 
 export function createWorkers(workerPath: string, options: WorkerOptions) {
@@ -85,7 +86,6 @@ export function createWorkers(workerPath: string, options: WorkerOptions) {
     minWorkers,
     maxWorkers,
     workerThreadOpts: {
-      // Note that we have to pass this explicitly to run in ava's test runner
       execArgv: ['--no-warnings', '--experimental-vm-modules'],
       // Important to override the child env so that it cannot access the parent env
       env,

packages/engine-multi/src/engine.ts

@@ -218,6 +218,10 @@ const createEngine = async (options: EngineOptions, workerPath?: string) => {
     // How does this work if deferred?
   };
 
+  const destroy = () => {
+    engine.closeWorkers()
+  }
+
   return Object.assign(engine, {
     options,
     workerPath: resolvedWorkerPath,
@@ -227,6 +231,7 @@ const createEngine = async (options: EngineOptions, workerPath?: string) => {
     getWorkflowStatus,
     execute: executeWrapper,
     listen,
+    destroy,
   });
 };
 

packages/engine-multi/src/types.ts

@@ -80,5 +80,7 @@ export interface RuntimeEngine extends EventEmitter {
     options?: Partial<EngineOptions>
   ): Pick<EventEmitter, 'on' | 'off' | 'once'>;
 
+  destroy(): void;
+
   // TODO my want some maintenance APIs, like getStatus. idk
 }

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

@@ -7,28 +7,30 @@ import { PURGE } from '../src/events';
 // so these are very high level tests and don't allow mock workers or anything
 
 const logger = createMockLogger(undefined, { level: 'debug' });
+let api;
 
 test.afterEach(() => {
   logger._reset();
+  api?.destroy()
 });
 
 test.serial('create a default engine api without throwing', async (t) => {
-  await createAPI();
+  api = await createAPI();
   t.pass();
 });
 
 test.serial('create an engine api with options without throwing', async (t) => {
-  await createAPI({ logger });
+  api = await createAPI({ logger });
   // just a token test to see if the logger is accepted and used
   t.true(logger._history.length > 0);
 });
 
 test.serial('create an engine api with a limited surface', async (t) => {
-  const api = await createAPI({ logger });
-  const keys = Object.keys(api);
+  api = await createAPI({ logger });
+  const keys = Object.keys(api).sort();
 
   // TODO the api will actually probably get a bit bigger than this
-  t.deepEqual(keys, ['execute', 'listen', 'on']);
+  t.deepEqual(keys, ['execute', 'listen', 'on', 'destroy'].sort());
 });
 
 // Note that this runs with the actual runtime worker
@@ -39,7 +41,7 @@ test.serial(
   'execute should return an event listener and receive workflow-complete',
   async (t) => {
     return new Promise(async (done) => {
-      const api = await createAPI({
+      api = await createAPI({
         logger,
         // Disable compilation
         compile: {
@@ -68,7 +70,7 @@ test.serial(
 
 test.serial('should listen to workflow-complete', async (t) => {
   return new Promise(async (done) => {
-    const api = await createAPI({
+    api = await createAPI({
       logger,
       // Disable compilation
       compile: {
@@ -98,7 +100,7 @@ test.serial('should listen to workflow-complete', async (t) => {
 
 test.serial('should purge workers after a single run', async (t) => {
   return new Promise(async (done) => {
-    const api = await createAPI({
+    api = await createAPI({
       logger,
       // Disable compilation
       compile: {

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

@@ -266,7 +266,9 @@ test.serial('autoinstall: emit an event on completion', async (t) => {
   t.truthy(event);
   t.is(event.module, '@openfn/language-common');
   t.is(event.version, '1.0.0');
-  t.assert(event.duration >= 50);
+  // Duration could be anything really as timeout is only loose - but so long as it's
+  // more than 10ms that implies it's called handleInstall and returned a reasonable value
+  t.assert(event.duration >= 10);
 });
 
 test.todo('autoinstall: emit on error');

packages/engine-multi/test/api/call-worker.test.ts

@@ -17,8 +17,6 @@ test.before(() => {
 
 test.after(() => api.closeWorkers());
 
-// TODO should I be tearing down the pool each time?
-
 test('initWorkers should add a callWorker function', (t) => {
   t.assert(typeof api.callWorker === 'function');
 });