From 1613b5a0cf3bb03b50aec9d4d2b4354931ff936a Mon Sep 17 00:00:00 2001 From: Nick Van Date: Tue, 14 Jul 2026 13:04:56 -0700 Subject: [PATCH] fix(convert-to-dss): size worker pool from cgroup budget, not host CPUs convert-to-dss fell back to min(len(work), os.cpu_count()) when DSS_WORKERS was unset. Inside a container os.cpu_count() reports the node's CPU count (8), not a memory-aware number, so eight concurrent rio.reproject workers (72-hr AORC -> SHG 4 km) exceeded the 12000Mi cgroup within seconds and the pod was OOMKilled (exit 137), discarding an 18-hour process-storms catalog. Route the auto path through worker_sizing.resolve_num_workers instead -- the same cgroup-derived cap process-storms already trusts (PER_WORKER_MB budget, plus num_workers / CC_NUM_WORKERS overrides). Neither stage can now silently scale to host-CPU count. DSS_WORKERS stays as an explicit override. --- src/actions/convert_to_dss.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/actions/convert_to_dss.py b/src/actions/convert_to_dss.py index 2534856..e540042 100644 --- a/src/actions/convert_to_dss.py +++ b/src/actions/convert_to_dss.py @@ -11,6 +11,7 @@ from typing import Any, Optional from actions import dss_filename, parse_storm_datetime, storm_rank +from worker_sizing import resolve_num_workers log = logging.getLogger(__name__) @@ -105,9 +106,17 @@ def convert_to_dss(ctx: dict[str, Any], action: Any) -> None: failed: list[str] = list(skipped) if work: - workers = ( - DSS_WORKERS if DSS_WORKERS > 0 else min(len(work), os.cpu_count() or 1) - ) + # Size the pool from the cgroup memory budget, not os.cpu_count(): inside + # a container cpu_count() reports the *host* CPU count, so the old + # fallback spawned 8 rio.reproject workers on an 8-core node and blew the + # 12000Mi cgroup in seconds (OOMKill, exit 137). resolve_num_workers is + # the same memory-aware sizing process-storms already trusts, and it + # honors the num_workers payload attr / CC_NUM_WORKERS env. DSS_WORKERS + # remains an explicit override for a fatter host. + if DSS_WORKERS > 0: + workers = DSS_WORKERS + else: + workers = min(len(work), resolve_num_workers(attrs)) log.info("Running %d conversions with %d workers", len(work), workers) # Explicit spawn context: the worker's first act is an fsspec/s3fs AORC