diff --git a/packages/engine-multi/README.md b/packages/engine-multi/README.md index bc7b1b9a9..133fc5122 100644 --- a/packages/engine-multi/README.md +++ b/packages/engine-multi/README.md @@ -81,6 +81,26 @@ engine.execute(plan) For a full list of events, see `src/events/ts` (the top-level API events are listed at the top) +## Memory Limits + +The engine enforces a memory limits on each run via `memoryLimitMb`. There are two modes of enforcement: + +**Heap limit**: sets V8's max old space size on the child process (and the worker thread's resource limits). If a run blows this, V8 aborts and the engine reports an OOMError. This only bounds the JavaScript heap - buffers and other native allocations don't count towards it. This is enabled by default. + +**cgroup limit**: a hard ceiling on the child process's total memory (including native allocations), enforced by the kernel through a cgroup v2 leaf. Each pooled child process is placed in its own cgroup under `cgroupParent` with `memory.max` set and swap disabled. If a run exceeds the ceiling, the kernel OOM-kills the child; the engine detects this from the cgroup's `memory.events` and reports an OOMError. + +An OOMError carries a `source` property (`'heap'` or `'cgroup'`) saying which limit was breached. + +### cgroup setup + +The engine does not provision the cgroup hierarchy itself. Rather, the host runtime must be started within a writable, cgroup v2 subtree. + +The engine creates one leaf per child process within it, applying the memory limit to each. + +cgroups must be enabled explicitly: the are off by default because unless properly configured, an OOM exception can kill the parenting process group hierarchy. + +See Worker documentation for notes about how to start the engine with cgroups enabled. + ## Module Loader Whitelist A whitelist controls what modules a job is allowed to import. At the moment this is hardcoded in the Engine to modules starting with @openfn. diff --git a/packages/engine-multi/ava.config.cjs b/packages/engine-multi/ava.config.cjs index 21d373937..842ebd7d6 100644 --- a/packages/engine-multi/ava.config.cjs +++ b/packages/engine-multi/ava.config.cjs @@ -2,5 +2,11 @@ const baseConfig = require('../../ava.config'); module.exports = { ...baseConfig, - files: ['test/**/*.test.ts'], + files: [ + 'test/**/*.test.ts', + + // Enforcement tests must be rub in an isolated cgroup + // via pnpm test:cgroup + '!test/worker/cgroup-enforcement.test.ts', + ], }; diff --git a/packages/engine-multi/package.json b/packages/engine-multi/package.json index 35ef12543..e38df3112 100644 --- a/packages/engine-multi/package.json +++ b/packages/engine-multi/package.json @@ -6,6 +6,7 @@ "type": "module", "scripts": { "test": "pnpm ava --serial", + "test:cgroup": "systemd-run --user --scope --unit=cgroup-test -p Delegate=yes -p MemoryMax=2G -- pnpm exec ava test/worker/cgroup-enforcement.test.ts", "test:types": "pnpm tsc --noEmit --project tsconfig.json", "test:mem": "NODE_OPTIONS=\"--max-old-space-size=90 --experimental-vm-modules\" pnpm exec tsx test/memtest.ts", "build": "tsup --config ./tsup.config.js", diff --git a/packages/engine-multi/src/api.ts b/packages/engine-multi/src/api.ts index 29450b16a..593196b62 100644 --- a/packages/engine-multi/src/api.ts +++ b/packages/engine-multi/src/api.ts @@ -58,6 +58,7 @@ const createAPI = async function ( maxWorkers: options.maxWorkers, memoryLimitMb: options.memoryLimitMb || DEFAULT_MEMORY_LIMIT, + memoryEnforcement: options.memoryEnforcement, runTimeoutMs: options.runTimeoutMs, statePropsToRemove: options.statePropsToRemove ?? [ diff --git a/packages/engine-multi/src/api/call-worker.ts b/packages/engine-multi/src/api/call-worker.ts index c3a062407..d2f648b0c 100644 --- a/packages/engine-multi/src/api/call-worker.ts +++ b/packages/engine-multi/src/api/call-worker.ts @@ -1,5 +1,5 @@ import { Logger } from '@openfn/logger'; -import createPool from '../worker/pool'; +import createPool, { MemoryEnforcement } from '../worker/pool'; import { CallWorker } from '../types'; // All events coming out of the worker need to include a type key @@ -13,6 +13,7 @@ type WorkerOptions = { env?: any; timeout?: number; // ms memoryLimitMb?: number; + memoryEnforcement?: MemoryEnforcement; proxyStdout?: boolean; // print internal stdout to console }; @@ -27,6 +28,7 @@ export default function initWorkers( env = {}, maxWorkers = 5, memoryLimitMb, + memoryEnforcement, proxyStdout = false, } = options; @@ -36,6 +38,7 @@ export default function initWorkers( maxWorkers, env, memoryLimitMb, + memoryEnforcement, proxyStdout, }, logger diff --git a/packages/engine-multi/src/api/lifecycle.ts b/packages/engine-multi/src/api/lifecycle.ts index 817140899..bd1a6254f 100644 --- a/packages/engine-multi/src/api/lifecycle.ts +++ b/packages/engine-multi/src/api/lifecycle.ts @@ -155,5 +155,7 @@ export const error = ( message: error.message || error.toString(), // default to exception because if we don't know, it's our fault severity: error.severity || 'exception', + // @ts-ignore for OOM errors, say which limit was breached (heap/cgroup) + source: error.source, }); }; diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index 38467b97c..c66c1d388 100644 --- a/packages/engine-multi/src/engine.ts +++ b/packages/engine-multi/src/engine.ts @@ -13,6 +13,7 @@ import { WORKFLOW_START, } from './events'; import initWorkers from './api/call-worker'; +import type { MemoryEnforcement } from './worker/pool'; import createState from './util/create-state'; import execute from './api/execute'; import validateWorker from './api/validate-worker'; @@ -74,6 +75,7 @@ export type EngineOptions = { logger: Logger; maxWorkers?: number; memoryLimitMb?: number; + memoryEnforcement?: MemoryEnforcement; stateLimitMb?: number; payloadLimitMb?: number; logPayloadLimitMb?: number; @@ -142,6 +144,7 @@ const createEngine = async ( { maxWorkers: options.maxWorkers, memoryLimitMb: defaultMemoryLimit, + memoryEnforcement: options.memoryEnforcement, proxyStdout: options.proxyStdout, }, options.logger diff --git a/packages/engine-multi/src/errors.ts b/packages/engine-multi/src/errors.ts index ca88cce83..9ac2e4eef 100644 --- a/packages/engine-multi/src/errors.ts +++ b/packages/engine-multi/src/errors.ts @@ -71,14 +71,21 @@ export class AutoinstallError extends EngineError { } } +// Where an OOM kill came from: +// - 'heap' the V8 heap limit (thread resourceLimits or --max-old-space-size) +// - 'cgroup' the OS cgroup memory.max ceiling (kernel SIGKILL) +export type OOMSource = 'heap' | 'cgroup'; + export class OOMError extends EngineError { severity = 'kill'; name = 'OOMError'; message; + source: OOMSource; - constructor() { + constructor(source: OOMSource = 'heap') { super(); + this.source = source; this.message = `Run exceeded maximum memory usage`; } } diff --git a/packages/engine-multi/src/events.ts b/packages/engine-multi/src/events.ts index f9a81c32d..7d62230b2 100644 --- a/packages/engine-multi/src/events.ts +++ b/packages/engine-multi/src/events.ts @@ -71,6 +71,9 @@ export interface WorkflowErrorPayload extends ExternalEvent { type: string; message: string; severity: string; + // where the error originated; 'engine' generally, but OOM errors narrow + // this to the limit that was breached ('heap' or 'cgroup') + source?: string; } export interface JobStartPayload extends ExternalEvent { diff --git a/packages/engine-multi/src/test/worker-functions.ts b/packages/engine-multi/src/test/worker-functions.ts index 8a0068827..a56f9d5be 100644 --- a/packages/engine-multi/src/test/worker-functions.ts +++ b/packages/engine-multi/src/test/worker-functions.ts @@ -119,6 +119,17 @@ const tasks = { // Array(1e9).fill('mario') }, + // Allocate memory OUTSIDE the V8 heap. Buffer.alloc lives in native memory, + // so --max-old-space-size can't catch this — only an OS-level limit (a + // cgroup memory.max ceiling) will. Zero-filling forces the pages resident so + // they count against the cgroup's accounting. + blowNativeMemory: async () => { + const chunks = []; + while (true) { + chunks.push(Buffer.alloc(10 * 1024 * 1024, 1)); // 10mb, resident + } + }, + // Some useful code // const stats = v8.getHeapStatistics(); // console.log( diff --git a/packages/engine-multi/src/util/serialize-error.ts b/packages/engine-multi/src/util/serialize-error.ts index 73b8a3e53..48718be73 100644 --- a/packages/engine-multi/src/util/serialize-error.ts +++ b/packages/engine-multi/src/util/serialize-error.ts @@ -4,6 +4,7 @@ export default (error: any) => { name: error.name, type: error.type, subtype: error.subtype, + source: error.source, severity: error.severity || 'crash', }; }; diff --git a/packages/engine-multi/src/worker/cgroup.ts b/packages/engine-multi/src/worker/cgroup.ts new file mode 100644 index 000000000..1c4420aa3 --- /dev/null +++ b/packages/engine-multi/src/worker/cgroup.ts @@ -0,0 +1,284 @@ +/** + * cgroup v2 memory enforcement for pooled child processes. + * + * cgroup level memory enforcement allows us to enforce memory limits at the kernel level, + * improving our ability to OOMKill runs. + * + * cgroup enforcement is disabled by default because the worker needs to run in an appropriately delegated runtime. See readme. + * + */ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import type { Logger } from '@openfn/logger'; + +const CGROUP_ROOT = '/sys/fs/cgroup'; + +export type CgroupHandle = { + path: string; // absolute path to the leaf cgroup directory + eventsPath: string; // absolute path to the leaf's memory.events file +}; + +// Extract the cgroup v2 path (the "0::" entry) from a /proc/self/cgroup +// listing and resolve it against the cgroup mount. Returns null if the process +// has no v2 cgroup (ie, a pure cgroup v1 host) +export const parseProcSelfCgroup = (content: string): string | null => { + for (const line of content.split('\n')) { + const match = line.match(/^0::(.+)$/); + if (match) { + // strip the trailing slash left by the namespace root case ("0::/") + return path.join(CGROUP_ROOT, match[1]).replace(/(.)\/$/, '$1'); + } + } + return null; +}; + +// The cgroup this process was started in. This is the default parent for +// per-child leaves: under systemd Delegate= or in a container it is exactly +// the subtree that was handed to us. Returns null when it can't be determined +export const resolveSelfCgroup = (): string | null => { + if (os.platform() !== 'linux') { + return null; + } + try { + const self = parseProcSelfCgroup( + fs.readFileSync('/proc/self/cgroup', 'utf8') + ); + // If we were started inside a leader leaf (eg, restarted from within an + // already-prepared cgroup), the delegated parent is the enclosing cgroup + if (self && path.basename(self) === LEADER_NAME) { + return path.dirname(self); + } + return self; + } catch (e) { + return null; + } +}; + +// Returns true when cgroup v2 memory enforcement can be used under `parent`. +// On first success this also prepares the parent (delegates the memory +// controller to its leaves). Caches the result and warns (once per parent) +// when unavailable +export const isCgroupV2Available = ( + parent: string | null, + logger: Logger +): boolean => { + const key = parent ?? ''; + if (key in availabilityCache) { + return availabilityCache[key]; + } + + let available = false; + if (parent && detect(parent, logger)) { + try { + prepare(parent, logger); + available = true; + } catch (e) { + logger.debug( + `cgroup: could not delegate memory controller in ${parent}: ${ + (e as Error).message + }` + ); + } + } + + availabilityCache[key] = available; + if (!available) { + logger.warn( + 'cgroup: v2 memory enforcement unavailable; falling back to heap-limit only. ' + + 'To enable it, start the worker inside a writable, delegated cgroup ' + + '(see the engine-multi README)' + ); + } + return available; +}; + +// Create a leaf cgroup for `pid`, set its hard memory ceiling and move the +// process into it. Returns a handle, or null on any failure (best-effort). +export const createChildCgroup = ( + parent: string, + pid: number, + limitBytes: number, + logger: Logger +): CgroupHandle | null => { + const dir = path.join(parent, `run-${pid}`); + try { + if (!fs.existsSync(dir)) { + fs.mkdirSync(dir); + } + fs.writeFileSync(path.join(dir, 'memory.max'), String(limitBytes)); + // Stop the process escaping the limit via swap. The swap controller may be + // absent, so this is best-effort. + try { + fs.writeFileSync(path.join(dir, 'memory.swap.max'), '0'); + } catch (e) { + // no swap controller; ignore + } + // Moving the pid in must come last: a cgroup with member processes cannot + // have its controllers reconfigured. + fs.writeFileSync(path.join(dir, 'cgroup.procs'), String(pid)); + logger.debug( + `cgroup: constrained process ${pid} to ${limitBytes} bytes at ${dir}` + ); + return { path: dir, eventsPath: path.join(dir, 'memory.events') }; + } catch (e) { + logger.warn( + `cgroup: failed to constrain process ${pid}: ${(e as Error).message}` + ); + return null; + } +}; + +// Returns true if the kernel OOM-killed anything in this cgroup. Used to tell a +// cgroup memory kill (a bare SIGKILL with no V8 message) apart from other crashes. +export const hasOomKill = (handle: CgroupHandle): boolean => { + try { + const content = fs.readFileSync(handle.eventsPath, 'utf8'); + for (const line of content.split('\n')) { + const [key, value] = line.trim().split(/\s+/); + if ( + (key === 'oom_kill' || key === 'oom_group_kill') && + parseInt(value, 10) > 0 + ) { + return true; + } + } + } catch (e) { + // events file gone (cgroup already removed) or unreadable + } + return false; +}; + +// Remove a leaf cgroup. The cgroup can only be removed once empty, so we retry +// a handful of times while the killed process is reaped. Best-effort and +// non-blocking. +export const removeChildCgroup = ( + handle: CgroupHandle | null, + logger: Logger, + attempts = 10 +) => { + if (!handle) { + return; + } + try { + fs.rmdirSync(handle.path); + } catch (e) { + const err = e as NodeJS.ErrnoException; + if (attempts > 0 && (err.code === 'EBUSY' || err.code === 'ENOTEMPTY')) { + setTimeout( + () => removeChildCgroup(handle, logger, attempts - 1), + 50 + ).unref(); + return; + } + if (err.code !== 'ENOENT') { + logger.debug(`cgroup: could not remove ${handle.path}: ${err.message}`); + } + } +}; + +// For tests: clear the cached availability probes. +export const _resetAvailabilityCache = () => { + for (const key of Object.keys(availabilityCache)) { + delete availabilityCache[key]; + } +}; + +// Leaf cgroup into which the worker relocates the processes in its own cgroup +// (typically just itself), so the memory controller can be delegated to the +// run leaves (see prepare) +const LEADER_NAME = 'leader'; + +// Cache availability per parent so we only probe the filesystem (and warn) once. +const availabilityCache: Record = {}; + +// True if a cgroup interface file (e.g. cgroup.controllers, subtree_control) +// lists the given whitespace-separated token. +function fileLists(file: string, token: string) { + return fs.readFileSync(file, 'utf8').split(/\s+/).includes(token); +} + +// Relocate every process in `parent` into a leader leaf, so `parent` itself +// becomes empty and its controllers can be delegated. Within a delegated +// subtree the only member processes are our own (typically just the worker, +// when it was started inside `parent`) +function moveProcsToLeader(parent: string, logger: Logger) { + const leader = path.join(parent, LEADER_NAME); + if (!fs.existsSync(leader)) { + fs.mkdirSync(leader); + } + const procs = fs + .readFileSync(path.join(parent, 'cgroup.procs'), 'utf8') + .trim() + .split('\n') + .filter(Boolean); + for (const pid of procs) { + try { + fs.writeFileSync(path.join(leader, 'cgroup.procs'), pid); + } catch (e) { + // A process may exit between the read and the move; skip it and let the + // subsequent subtree_control write decide whether the parent is empty + logger.debug( + `cgroup: could not move pid ${pid} into leader: ${(e as Error).message}` + ); + } + } +} + +// Enable the memory controller for the leaves of `parent` (idempotent). +// cgroup v2's "no internal processes" rule forbids delegating controllers +// from a populated cgroup, and in the recommended setup the worker itself +// lives in `parent` - so on EBUSY, move its processes into a leader leaf +// and retry +function prepare(parent: string, logger: Logger) { + const subtree = path.join(parent, 'cgroup.subtree_control'); + if (fileLists(subtree, 'memory')) { + return; + } + try { + fs.writeFileSync(subtree, '+memory'); + } catch (e) { + if ((e as NodeJS.ErrnoException).code === 'EBUSY') { + logger.debug( + `cgroup: ${parent} is populated; relocating processes to a leader leaf` + ); + moveProcsToLeader(parent, logger); + fs.writeFileSync(subtree, '+memory'); + } else { + throw e; + } + } +} + +// Read-only check that `parent` is a usable delegated cgroup: a cgroup v2 +// directory under the cgroup mount, writable by this process, with the memory +// controller available +function detect(parent: string, logger: Logger): boolean { + if (os.platform() !== 'linux') { + return false; + } + + const rel = path.relative(CGROUP_ROOT, parent); + if (rel.startsWith('..') || path.isAbsolute(rel)) { + logger.debug(`cgroup: parent ${parent} is not under ${CGROUP_ROOT}`); + return false; + } + + try { + // cgroup v2 exposes cgroup.controllers in every cgroup; v1 does not + const controllers = path.join(parent, 'cgroup.controllers'); + if (!fileLists(controllers, 'memory')) { + logger.debug(`cgroup: memory controller not available in ${parent}`); + return false; + } + // We need to create leaf dirs and move pids around within the parent + fs.accessSync(parent, fs.constants.W_OK); + fs.accessSync(path.join(parent, 'cgroup.procs'), fs.constants.W_OK); + return true; + } catch (e) { + logger.debug( + `cgroup: parent ${parent} is not usable: ${(e as Error).message}` + ); + return false; + } +} diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 48a268803..28fb2546d 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -13,12 +13,34 @@ import { import { HANDLED_EXIT_CODE } from '../events'; import { Logger } from '@openfn/logger'; import type { PayloadLimits } from './thread/runtime'; +import { + CgroupHandle, + createChildCgroup, + hasOomKill, + isCgroupV2Available, + removeChildCgroup, + resolveSelfCgroup, +} from './cgroup'; + +// Constant memory overhead to apply to the cgroup ceiling +const CGROUP_MEMORY_OVERHEAD_MB = 128; + +export type MemoryEnforcement = { + // Sets node's --max-old-space-size on child processes + // Default true + oldspace?: boolean; + + // Uses cgroups to set a hard ceiling on child processes through the kernel + // Default false. + cgroup?: boolean; +}; export type PoolOptions = { capacity?: number; // defaults to 5 maxWorkers?: number; // alias for capacity. Which is best? env?: Record; // default environment for workers - memoryLimitMb?: number; // --max-old-space-size for child processes + memoryLimitMb?: number; // Set the maximum runtime memory a child process can consume + memoryEnforcement?: MemoryEnforcement; proxyStdout?: boolean; // print internal stdout to console }; @@ -80,12 +102,32 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // Keep track of all the workers we created const allWorkers: Record = {}; + const enforceOldspace = options.memoryEnforcement?.oldspace ?? true; + + // cgroup v2 leaf per child (keyed by pid), when cgroup memory enforcement is enabled + const cgroupParent = resolveSelfCgroup(); + const cgroupMemoryLimitMb = + options.memoryLimitMb && options.memoryLimitMb + CGROUP_MEMORY_OVERHEAD_MB; + const cgroupEnabled = + options.memoryEnforcement?.cgroup && + !!cgroupMemoryLimitMb && + isCgroupV2Available(cgroupParent, logger); + const cgroups: Record = {}; + + // Tear down a child's leaf cgroup once it has been killed (best-effort). + const cleanupCgroup = (worker: ChildProcess | false) => { + if (worker && worker.pid && cgroups[worker.pid]) { + removeChildCgroup(cgroups[worker.pid], logger); + delete cgroups[worker.pid]; + } + }; + const init = (maybeChild: ChildProcess | false) => { let child: ChildProcess; if (!maybeChild) { // create a new child process and load the module script into it const execArgv = ['--experimental-vm-modules', '--no-warnings']; - if (options.memoryLimitMb) { + if (enforceOldspace && options.memoryLimitMb) { execArgv.push(`--max-old-space-size=${options.memoryLimitMb}`); } @@ -107,6 +149,17 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { logger.debug('pool: Created new child process', child.pid); allWorkers[child.pid!] = child; + + // Place the child in its own cgroup with a hard memory ceiling. The leaf + // lives for the lifetime of the (reused) child and is removed when it dies. + if (cgroupEnabled && child.pid) { + cgroups[child.pid] = createChildCgroup( + cgroupParent!, + child.pid, + cgroupMemoryLimitMb! * 1024 * 1024, + logger + ); + } } else { child = maybeChild as ChildProcess; logger.debug('pool: Using existing child process', child.pid); @@ -153,6 +206,21 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { logger.debug(`pool: Worker exited unexpectedly with code ${code}`); clearTimeout(timeout); + // A cgroup memory kill is a bare SIGKILL with no V8 message, so check + // the cgroup's OOM counter before falling back to scraping stderr. + const handle = worker.pid ? cgroups[worker.pid] : null; + if (handle && hasOomKill(handle)) { + logger.error( + `pool: worker ${worker.pid} was killed by the OS for exceeding ` + + `its cgroup memory limit (${cgroupMemoryLimitMb}mb)` + ); + killWorker(worker); + // restore a placeholder to the queue + finish(false); + reject(new OOMError('cgroup')); + return; + } + // Read the stderr stream from the worked to see if this looks like an OOM error const rl = readline.createInterface({ input: worker.stderr!, @@ -163,6 +231,9 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (worker.stderr && worker.stderr?.readableLength > 0) { for await (const line of rl) { if (line.match(/JavaScript heap out of memory/)) { + logger.error( + `pool: worker ${worker.pid} exceeded the V8 heap limit` + ); killWorker(worker); // restore a placeholder to the queue finish(false); @@ -250,6 +321,8 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // @ts-ignore e.severity = evt.error.severity; e.name = evt.error.name; + // @ts-ignore preserve the OOM source ('heap'/'cgroup') across IPC + e.source = evt.error.source; reject(e); finish(worker); @@ -265,6 +338,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (worker) { logger.debug('pool: destroying worker ', worker.pid); worker.kill(); + cleanupCgroup(worker); delete allWorkers[worker.pid!]; } }; @@ -305,6 +379,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const worker = pool.pop(); if (worker) { killPromises.push(waitForWorkerExit(worker)); + cleanupCgroup(worker); delete allWorkers[worker.pid!]; } } @@ -312,6 +387,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (immediate) { Object.values(allWorkers).forEach((worker) => { killPromises.push(waitForWorkerExit(worker, 1)); + cleanupCgroup(worker); delete allWorkers[worker.pid!]; }); } diff --git a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts new file mode 100644 index 000000000..beb53028e --- /dev/null +++ b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts @@ -0,0 +1,69 @@ +/** + * These tests prove the kernel actually OOM-kills a runaway run via the cgroup's memory.max ceiling. + * Tests run with node's oldspace enforcement disabled: cgroups are the only lever we can use to trigger an OOM error. + * + * !!!IMPORTANT!! + * + * This is NOT safe to run directly from a terminal as the owning cgroup will be terminated + * Use `pnpm test:cgroup` to run the test in a detatched process + */ +import test from 'ava'; +import path from 'node:path'; +import { createMockLogger } from '@openfn/logger'; + +import createPool from '../../src/worker/pool'; +import { + isCgroupV2Available, + resolveSelfCgroup, + _resetAvailabilityCache, +} from '../../src/worker/cgroup'; + +const logger = createMockLogger(); + +const available = isCgroupV2Available(resolveSelfCgroup(), logger); +_resetAvailabilityCache(); + +const cgroupTest = available ? test.serial : test.serial.skip; + +const workerPath = path.resolve('dist/test/worker-functions.js'); + +const memoryLimitMb = 72; // + 128mb headroom = 200mb effective cgroup ceiling + +// Disable node's oldspace memory limit so that croup is the only variable in play +const memoryEnforcement = { cgroup: true, oldspace: false }; + +cgroupTest( + 'cgroup OOM-kills a run that exceeds memory.max and surfaces OOMError', + async (t) => { + const pool = createPool( + workerPath, + { memoryLimitMb, memoryEnforcement }, + logger + ); + + const err = await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { + name: 'OOMError', + }); + t.is((err as any).source, 'cgroup'); + + await pool.destroy(); + } +); + +cgroupTest('pool recovers after a cgroup OOM kill', async (t) => { + const pool = createPool( + workerPath, + { memoryLimitMb, memoryEnforcement }, + logger + ); + + await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { + name: 'OOMError', + }); + + // The pool should restore the dead worker's slot and keep working. + const result = await pool.exec('test', []); + t.is(result, 42); + + await pool.destroy(); +}); diff --git a/packages/engine-multi/test/worker/cgroup.test.ts b/packages/engine-multi/test/worker/cgroup.test.ts new file mode 100644 index 000000000..eab90ed04 --- /dev/null +++ b/packages/engine-multi/test/worker/cgroup.test.ts @@ -0,0 +1,118 @@ +import test from 'ava'; +import os from 'node:os'; +import { createMockLogger } from '@openfn/logger'; + +import { + createChildCgroup, + hasOomKill, + isCgroupV2Available, + parseProcSelfCgroup, + removeChildCgroup, + resolveSelfCgroup, + _resetAvailabilityCache, +} from '../../src/worker/cgroup'; + +const logger = createMockLogger(); + +const isLinux = os.platform() === 'linux'; + +test.beforeEach(() => { + _resetAvailabilityCache(); +}); + +test('parseProcSelfCgroup resolves the v2 entry against the cgroup mount', (t) => { + t.is( + parseProcSelfCgroup('0::/system.slice/openfn.service\n'), + '/sys/fs/cgroup/system.slice/openfn.service' + ); +}); + +test('parseProcSelfCgroup handles the namespace root', (t) => { + t.is(parseProcSelfCgroup('0::/\n'), '/sys/fs/cgroup'); +}); + +test('parseProcSelfCgroup returns null for a v1-only listing', (t) => { + const v1 = ['12:memory:/user.slice', '3:cpu,cpuacct:/user.slice', ''].join( + '\n' + ); + t.is(parseProcSelfCgroup(v1), null); +}); + +if (!isLinux) { + test('resolveSelfCgroup returns null on non-linux hosts', (t) => { + t.is(resolveSelfCgroup(), null); + }); + + test('isCgroupV2Available returns false on non-linux hosts', (t) => { + t.false(isCgroupV2Available('/sys/fs/cgroup/test', logger)); + }); +} + +if (isLinux) { + test('resolveSelfCgroup returns a path under the cgroup mount', (t) => { + const self = resolveSelfCgroup(); + // null is legitimate on a cgroup v1 host; otherwise the path must be + // inside the mount and never a leader leaf (the enclosing cgroup is + // returned instead) + if (self !== null) { + t.true(self.startsWith('/sys/fs/cgroup')); + t.not(self.split('/').pop(), 'leader'); + } else { + t.pass(); + } + }); +} + +test('isCgroupV2Available returns false for a null parent', (t) => { + t.false(isCgroupV2Available(null, logger)); +}); + +test('isCgroupV2Available returns false for a parent outside the cgroup mount', (t) => { + t.false(isCgroupV2Available('/tmp/not-a-cgroup', logger)); +}); + +test('isCgroupV2Available returns false for a missing parent', (t) => { + t.false(isCgroupV2Available('/sys/fs/cgroup/this/does/not/exist', logger)); +}); + +test('availability probe is cached and only warns once', (t) => { + const l = createMockLogger('test', { level: 'debug' }); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + + const warnings = l._history.filter((h: any) => h[0] === 'warn'); + t.is(warnings.length, 1); +}); + +test('createChildCgroup returns null when the parent is not writable', (t) => { + const handle = createChildCgroup( + '/this/path/does/not/exist', + 12345, + 256 * 1024 * 1024, + logger + ); + t.is(handle, null); +}); + +test('hasOomKill returns false when the events file is missing', (t) => { + t.false( + hasOomKill({ + path: '/no/such/cgroup', + eventsPath: '/no/such/cgroup/memory.events', + }) + ); +}); + +test('removeChildCgroup tolerates a null handle', (t) => { + t.notThrows(() => removeChildCgroup(null, logger)); +}); + +test('removeChildCgroup tolerates a non-existent cgroup', (t) => { + t.notThrows(() => + removeChildCgroup( + { path: '/no/such/cgroup', eventsPath: '/no/such/cgroup/memory.events' }, + logger + ) + ); +}); diff --git a/packages/engine-multi/test/worker/pool.test.ts b/packages/engine-multi/test/worker/pool.test.ts index 5a3b8d9dc..6c02ccc0d 100644 --- a/packages/engine-multi/test/worker/pool.test.ts +++ b/packages/engine-multi/test/worker/pool.test.ts @@ -179,9 +179,11 @@ test('throw if memory limit is exceeded', async (t) => { try { await pool.exec('blowMemory', [], { memoryLimitMb: 100 }); + t.fail('expected the run to OOM'); } catch (e: any) { t.is(e.message, 'Run exceeded maximum memory usage'); t.is(e.name, 'OOMError'); + t.is(e.source, 'heap'); } }); @@ -200,9 +202,13 @@ test('child process should not have --max-old-space-size when memoryLimitMb is n test('pool recovers after process-level OOM', async (t) => { const pool = createPool(workerPath, { memoryLimitMb: 50 }, logger); - await t.throwsAsync(() => pool.exec('blowMemory', [], { memoryLimitMb: 20 }), { - name: 'OOMError', - }); + const err = await t.throwsAsync( + () => pool.exec('blowMemory', [], { memoryLimitMb: 20 }), + { + name: 'OOMError', + } + ); + t.is((err as any).source, 'heap'); // Pool should still be functional after the OOM const result = await pool.exec('test', [42]); diff --git a/packages/ws-worker/README.md b/packages/ws-worker/README.md index 9a82e8209..f9394b844 100644 --- a/packages/ws-worker/README.md +++ b/packages/ws-worker/README.md @@ -51,6 +51,44 @@ Use `-l mock` to connect to a lightning mock server (on the default port). For a list of supported worker and engine options, see src/start.ts +## Enforcing memory limits with cgroups + +Each run's memory limit is enforced by default through node's max-old-space-size, which only constrains heap size. Native and buffer allocations bypass this limit. This can cause the worker to consume more memory than it is technically allowed, which can in turn cause the whole worker process to be killed by its container (ie, kubernetes). + +For tighter enforcement on linux environments, cgroups can be utilised. This approach make the kernel itself enforce the memory size of all run processes. If a single run exceeds its alloted size, that process is terminated and the run is marked as Killed. See engine docs for more details about this. + +To enable cgroup enforcement, pass `--cgroups` or set `WORKER_ENABLE_CGROUP_ENFORCEMENT`. + +### Local Development + +If you try and start a local worker with cgroup enforcement, and a process exceeds its memory, you'll find that the whole owning process is taken down (ie, the hosting terminal). This is because the worker doesn't create its own isolated cgroup — it enforces limits using whatever cgroup it happens to be started in. In an ordinary dev terminal, that cgroup is shared with other things (your shell, other tools, sometimes your whole desktop session), so an OOM kill there can take out more than just the run that went over its limit. + +To safely run with cgroup enforcement, you have to spawn a detached process and delegate the cgroup: + +``` +systemd-run --user --scope --unit=openfn-worker -p Delegate=yes -- pnpm start +``` + +To close the process press CTRL-C or run: + +``` +systemctl --user stop openfn-worker.scope +``` + +### With Docker + +Cgroup enforcement is off by default even inside a container — you still need to explicitly set `WORKER_ENABLE_CGROUP_ENFORCEMENT` (or pass `--cgroups`) for it to activate. + +Setting the env var isn't enough on its own, though: the container also needs write access to its own cgroup files, which an ordinary `docker run` doesn't grant. Without it, enforcement quietly falls back to heap-limit-only, same as above. + +For local testing, you can grant that access with `--privileged`: + +```bash +docker run --privileged -e WORKER_ENABLE_CGROUP_ENFORCEMENT=true -e WORKER_SECRET=$WORKER_SECRET openfn-worker +``` + +**`--privileged` is not suitable for production.** It grants far more than cgroup write access — full device access, every Linux capability, and disabled seccomp/AppArmor confinement — which is a big escalation for a worker that executes arbitrary job code. For a real deployment, work out the narrowest way to grant cgroup write access with whoever owns your container/orchestration setup, rather than reaching for `--privileged`. + ## Watched Server You can start a dev server (which rebuilds on save) by running: diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts index 4e08e2601..e8650ee72 100644 --- a/packages/ws-worker/src/start.ts +++ b/packages/ws-worker/src/start.ts @@ -111,6 +111,7 @@ if (args.mock) { const engineOptions = { repoDir: args.repoDir, memoryLimitMb: args.runMemory, + memoryEnforcement: { cgroup: args.cgroup }, stateLimitMb: args.stateMemory, maxWorkers: effectiveCapacity, statePropsToRemove: args.statePropsToRemove, diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index f2850a383..e6bd9d374 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -16,6 +16,7 @@ type Args = { batchLimit?: number; batchLogs: boolean; capacity: number; + cgroup?: boolean; workloops?: string; claimTimeoutSeconds?: number; collectionsUrl?: string; @@ -82,6 +83,7 @@ export default function parseArgs(argv: string[]): Args { WORKER_BATCH_LIMIT, WORKER_BATCH_LOGS, WORKER_CAPACITY, + WORKER_ENABLE_CGROUP_ENFORCEMENT, WORKER_CLAIM_TIMEOUT_SECONDS, WORKER_COLLECTIONS_URL, WORKER_COLLECTIONS_VERSION, @@ -231,7 +233,12 @@ export default function parseArgs(argv: string[]): Args { 'Maximum memory allocated to a single run, in mb. Env: WORKER_MAX_PAYLOAD_MB', type: 'number', }) - + .option('cgroup', { + alias: ['enable-cgroup-enforcement', 'cgroups'], + description: + 'Enforce run memory limits applied via cgroups for extra memory hardening. Linux only. Default false. Env: WORKER_ENABLE_CGROUP_ENFORCEMENT', + type: 'boolean', + }) .option('max-run-duration-seconds', { alias: 't', description: @@ -327,6 +334,7 @@ export default function parseArgs(argv: string[]): Args { log: setArg(args.log, WORKER_LOG_LEVEL as LogLevel, 'debug'), backoff: setArg(args.backoff, WORKER_BACKOFF, '1/10'), capacity: setArg(args.capacity, WORKER_CAPACITY, DEFAULT_WORKER_CAPACITY), + cgroup: setArg(args.cgroup, WORKER_ENABLE_CGROUP_ENFORCEMENT, false), statePropsToRemove: setArg( args.statePropsToRemove, WORKER_STATE_PROPS_TO_REMOVE,