|
#!/bin/bash |
|
set -euo pipefail |
|
|
|
# Run the ICAL self-calibration pipeline via Rapthor. |
|
# |
|
# This script: |
|
# 1. Generates a strategy file from environment variables |
|
# 2. Templates the parset with concrete paths |
|
# 3. Runs `rapthor <parset>` |
|
# |
|
# Required env vars: |
|
# DATA_DIR – directory containing the MS and skymodel |
|
# SCRIPTS_DIR – directory containing parset templates and this script |
|
# |
|
# Optional env vars (with sensible defaults for the dev target): |
|
# DATA – dataset name stem [ska-low-aa2-mildtec-small] |
|
# SKYMODEL_FILE – skymodel filename in DATA_DIR [lotss_p207+24_negdec.skymodel] |
|
# SKYMODEL_MAX_SOURCES – max brightest skymodel components [200] |
|
# SKYMODEL_MIN_FLUX_JY – min Stokes-I flux density [Jy] [0.0] |
|
# NCYCLES – number of DD self-cal cycles [3] |
|
# MAX_DIRECTIONS – max DD directions per cycle [3] |
|
# MAXITER – DDECal max iterations [150] |
|
# CELLSIZE_ARCSEC – imaging cell size [5.0] |
|
# ROBUST – imaging robust weight [-0.5] |
|
# IDG_MODE – IDG mode (cpu/hybrid/gpu) [cpu] |
|
# SAVE_VISIBILITIES – save corrected visibilities [False] |
|
# FAST_TIMESTEP_SEC – fast phase solution interval (s) [20] |
|
# SLOW_TIMESTEP_SEC – slow gain solution interval (s) [120] |
|
# GENERATE_SKYMODEL – let Rapthor image+extract skymodel [false] |
|
# DDE_METHOD – imaging DD correction method [full] |
|
# NUMTHREADS – max threads for Rapthor [nproc/2] |
|
# MEM_PER_NODE_GB – GB of memory per worker node [0 = all] |
|
# RAPTHOR_DEBUG_WORKFLOW – enable CWL debug mode (True/False) [False] |
|
# SELFCAL_DATA_FRACTION – fraction of data for self-cal steps [1.0] |
|
# FINAL_DATA_FRACTION – fraction for the optional final Rapthor cycle [same as SELFCAL_DATA_FRACTION] |
|
# Set to 1.0 (with SELFCAL < 1) to run self-cal on a subset then a full-data final pass. |
|
|
|
DATA_DIR="${DATA_DIR:-data}" |
|
OUTPUT_DIR="${OUTPUT_DIR:-${DATA_DIR}}" # Separate writable output dir, defaults to DATA_DIR for backward compatibility |
|
SCRIPTS_DIR="${SCRIPTS_DIR:-scripts}" |
|
# HTCondor pilots run as nobody (HOME=/nonexistent); apptainer/Toil need a writable home. |
|
export HOME="${OUTPUT_DIR}/.docker_home" |
|
mkdir -p "${HOME}" "${HOME}/.casa" |
|
export TOIL_WORKDIR="${TOIL_WORKDIR:-${OUTPUT_DIR}/.toil-workdir}" |
|
export TMPDIR="${TMPDIR:-${TOIL_WORKDIR}}" |
|
mkdir -p "${TOIL_WORKDIR}" "${TMPDIR}" |
|
DATA="${DATA:-ska-low-aa2-mildtec-small}" |
|
SKYMODEL_FILE="${SKYMODEL_FILE:-lotss_p207+24_negdec.skymodel}" |
|
SKYMODEL_MAX_SOURCES="${SKYMODEL_MAX_SOURCES:-200}" |
|
SKYMODEL_MIN_FLUX_JY="${SKYMODEL_MIN_FLUX_JY:-0.0}" |
|
|
|
# Resolve MS |
|
MS="${DATA_DIR}/${DATA}.ms" |
|
if [ ! -d "${MS}" ]; then |
|
echo "ERROR: MS not found: ${MS}" >&2 |
|
echo " Run 'make get-data' first, or set DATA= to match the extracted MS name." >&2 |
|
exit 2 |
|
fi |
|
|
|
# Resolve skymodel |
|
SKYMODEL="${DATA_DIR}/${SKYMODEL_FILE}" |
|
GENERATE_SKYMODEL="${GENERATE_SKYMODEL:-false}" |
|
|
|
# Pipeline parameters |
|
NCYCLES="${NCYCLES:-3}" |
|
MAX_DIRECTIONS="${MAX_DIRECTIONS:-3}" |
|
MAXITER="${MAXITER:-150}" |
|
CELLSIZE_ARCSEC="${CELLSIZE_ARCSEC:-5.0}" |
|
ROBUST="${ROBUST:--0.5}" |
|
IDG_MODE="${IDG_MODE:-cpu}" |
|
SAVE_VISIBILITIES="${SAVE_VISIBILITIES:-False}" |
|
FAST_TIMESTEP_SEC="${FAST_TIMESTEP_SEC:-20}" |
|
SLOW_TIMESTEP_SEC="${SLOW_TIMESTEP_SEC:-120}" |
|
DDE_METHOD="${DDE_METHOD:-full}" |
|
NUMTHREADS="${NUMTHREADS:-$(($(nproc) / 2))}" |
|
case "${NUMTHREADS}" in |
|
'' | *[!0-9]*) |
|
echo "ERROR: NUMTHREADS must be a positive integer, got '${NUMTHREADS}'" >&2 |
|
exit 2 |
|
;; |
|
esac |
|
if [ "${NUMTHREADS}" -lt 1 ]; then |
|
NUMTHREADS=1 |
|
fi |
|
SELFCAL_DATA_FRACTION="${SELFCAL_DATA_FRACTION:-1.0}" |
|
FINAL_DATA_FRACTION="${FINAL_DATA_FRACTION:-$SELFCAL_DATA_FRACTION}" |
|
|
|
# Batch system for Rapthor's internal CWL/Toil steps. |
|
# single_machine – all steps run in this container (default, no extra deps) |
|
# kubernetes – steps run as k8s pods; requires KUBECONFIG, a shared |
|
# filesystem accessible to pods (TOIL_KUBERNETES_EXTRA_HOSTPATH), |
|
# and TOIL_APPLIANCE_SELF set to the rapthor image. |
|
# See wes-mildtec-k8s-smoke.sh for required env vars. |
|
RAPTHOR_BATCH_SYSTEM="${RAPTHOR_BATCH_SYSTEM:-single_machine}" |
|
|
|
# Max parallel CWL steps (Rapthor max_nodes / Toil --maxLocalJobs). |
|
# single_machine: 1 = serial; >1 = parallel threads on same node. |
|
# slurm_rest: each step becomes an independent Slurm job; set to the |
|
# number of MS chunks (typically 5 for the mildtec smoke test). |
|
# kubernetes: max concurrent k8s pods. |
|
RAPTHOR_MAX_NODES="${RAPTHOR_MAX_NODES:-1}" |
|
|
|
# CPUs requested per CWL step (Rapthor cpus_per_task). |
|
# 0 = use all available on the node (default for single_machine). |
|
# For slurm_rest, set this to match the actual CPU resources per Slurm job. |
|
RAPTHOR_CPUS_PER_TASK="${RAPTHOR_CPUS_PER_TASK:-0}" |
|
|
|
# Memory per node in GB used for Rapthor imaging/calibration steps. |
|
# 0 means "use all available" (single_machine default). |
|
# For kubernetes, set this to the memory you want each worker pod to have |
|
# (e.g. 32 for 32 GB); Toil uses it as --defaultMemory for CWL steps. |
|
MEM_PER_NODE_GB="${MEM_PER_NODE_GB:-0}" |
|
|
|
# Enable Rapthor workflow debugging (generates verbose CWL logs and keeps temp files). |
|
# Note: debug_workflow=True is incompatible with slurm_rest and slurm_static batch systems. |
|
# Set to "True" for debugging, defaults to "False" for production use. |
|
RAPTHOR_DEBUG_WORKFLOW="${RAPTHOR_DEBUG_WORKFLOW:-False}" |
|
# Leave the successful workload process alive briefly for an external |
|
# profiler to flush its perf capture before the HTCondor pilot is reaped. |
|
BENCHMON_DRAIN_S="${BENCHMON_DRAIN_S:-0}" |
|
case "${BENCHMON_DRAIN_S}" in |
|
'' | *[!0-9]*) |
|
echo "ERROR: BENCHMON_DRAIN_S must be a non-negative integer, got '${BENCHMON_DRAIN_S}'" >&2 |
|
exit 2 |
|
;; |
|
esac |
|
if [ "${RAPTHOR_BATCH_SYSTEM}" = "kubernetes" ]; then |
|
# Kubernetes workers create tmp-out directories as root; the parent Rapthor |
|
# process may run as a different UID and cannot safely clean them up. |
|
RAPTHOR_KEEP_TEMPORARY_FILES="${RAPTHOR_KEEP_TEMPORARY_FILES:-True}" |
|
else |
|
RAPTHOR_KEEP_TEMPORARY_FILES="${RAPTHOR_KEEP_TEMPORARY_FILES:-False}" |
|
fi |
|
|
|
# Working directory |
|
WORK_DIR="${OUTPUT_DIR}/rapthor_${DATA}" |
|
LOG_DIR="${WORK_DIR}/logs" |
|
mkdir -p "${LOG_DIR}" |
|
# The pilot filesystem is read-only at /var/lib/toil. Keep Toil's local |
|
# workflow state and temporary files in the shared, writable run directory. |
|
export TOIL_WORKDIR="${TOIL_WORKDIR:-${WORK_DIR}/toil-workdir}" |
|
export TMPDIR="${TMPDIR:-${WORK_DIR}/tmp}" |
|
mkdir -p "${TOIL_WORKDIR}" "${TMPDIR}" |
|
# Preserve the HTCondor stdout/stderr streams. The setup phase is copied into |
|
# the shared log below; Rapthor writes its own structured log there directly. |
|
exec 3>&1 4>&2 |
|
exec > >(tee -a "${LOG_DIR}/rapthor.log") 2>&1 |
|
|
|
if [ "${RAPTHOR_BATCH_SYSTEM}" = "kubernetes" ]; then |
|
# Keep nested Toil worker pods on the same uid/gid as the parent Rapthor |
|
# process. Otherwise cwltool creates root-owned 0700 tmp-out directories and |
|
# the parent fails during output collection after the workflow has succeeded. |
|
RAPTHOR_UID="$(id -u)" |
|
RAPTHOR_GID="$(id -g)" |
|
cat >"${WORK_DIR}/toil-k8s-pod-security-context.yaml" <<EOF |
|
runAsUser: ${RAPTHOR_UID} |
|
runAsGroup: ${RAPTHOR_GID} |
|
fsGroup: ${RAPTHOR_GID} |
|
fsGroupChangePolicy: OnRootMismatch |
|
EOF |
|
cat >"${WORK_DIR}/toil-k8s-container-security-context.yaml" <<EOF |
|
runAsUser: ${RAPTHOR_UID} |
|
runAsGroup: ${RAPTHOR_GID} |
|
EOF |
|
cp "${WORK_DIR}/toil-k8s-pod-security-context.yaml" "${WORK_DIR}/pod.yaml" |
|
cp "${WORK_DIR}/toil-k8s-container-security-context.yaml" "${WORK_DIR}/container.yaml" |
|
export TOIL_KUBERNETES_POD_SECURITY_CONTEXT="${WORK_DIR}/toil-k8s-pod-security-context.yaml" |
|
export TOIL_KUBERNETES_SECURITY_CONTEXT="${WORK_DIR}/toil-k8s-container-security-context.yaml" |
|
# Autoscaling nodes spend minutes in ContainerCreating (CSI + image |
|
# pull). Toil's default 120s timeout treats that as a stuck mount. |
|
export TOIL_KUBERNETES_POD_TIMEOUT="${TOIL_KUBERNETES_POD_TIMEOUT:-600}" |
|
fi |
|
|
|
# Rapthor invokes toil with --restart for pipeline steps. If a previous local |
|
# run left job stores behind, newer toil/cwltool combinations can fail during |
|
# restart before any work is redone. Start each local task run from a clean |
|
# pipeline state inside WORK_DIR. |
|
python3 - "${WORK_DIR}" <<'PYEOF' |
|
from pathlib import Path |
|
import shutil |
|
import sys |
|
|
|
work_dir = Path(sys.argv[1]) |
|
for jobstore in work_dir.glob("pipelines/**/jobstore"): |
|
shutil.rmtree(jobstore, ignore_errors=True) |
|
for output_json in work_dir.glob("pipelines/**/pipeline_outputs.json"): |
|
output_json.unlink(missing_ok=True) |
|
PYEOF |
|
|
|
# Astropy (and other XDG-aware tools) write cache to ~/.astropy by default, |
|
# which is not writable inside the container. Pin everything under DATA_DIR |
|
# so the same writable volume is used throughout, including in toil sub-processes. |
|
export XDG_CACHE_HOME="${OUTPUT_DIR}/.cache" |
|
export ASTROPY_CACHE_DIR="${XDG_CACHE_HOME}/astropy" |
|
mkdir -p "${ASTROPY_CACHE_DIR}" |
|
|
|
# DP3 (and its OpenBLAS dependency) must not spawn multiple threads — multi- |
|
# threaded OpenBLAS interferes with DP3's own threading and causes an immediate |
|
# fatal exception ("std exception detected: …OpenBLAS multi-threading…") that |
|
# terminates the process before any output is written. toil uses |
|
# --preserve-entire-environment so this value is inherited by all CWL steps. |
|
export OPENBLAS_NUM_THREADS=1 |
|
|
|
# Toil's workflow-history SQLite bookkeeping currently crashes on repeated local |
|
# runs in this environment before CWL steps can execute. Disable history |
|
# recording; this does not affect the actual pipeline work. |
|
export TOIL_HISTORY=False |
|
export TOIL_JOB_HISTORY=False |
|
|
|
# ── Generate strategy file ─────────────────────────────────────────────────── |
|
# Follows the ICAL pipeline workflow (slides 28-29 of the Gecko presentation): |
|
# - DD cycles 1-2: fast phase + medium phase only |
|
# - DD cycle 3+: fast phase + medium phase + slow gain |
|
# - Convergence check on middle cycles |
|
|
|
MEDIUM_TIMESTEP_SEC=$(awk "BEGIN {printf \"%.1f\", ${FAST_TIMESTEP_SEC} * 2}") |
|
FULLJONES_TIMESTEP_SEC="${SLOW_TIMESTEP_SEC}" |
|
|
|
STRATEGY_FILE="${WORK_DIR}/strategy.py" |
|
cat >"${STRATEGY_FILE}" <<STRATEGY_EOF |
|
"""Custom Rapthor strategy generated by run-ical.sh""" |
|
strategy_steps = [] |
|
n_cycles = ${NCYCLES} |
|
|
|
for i in range(n_cycles): |
|
strategy_steps.append({}) |
|
|
|
idx = i |
|
strategy_steps[idx]['do_calibrate'] = True |
|
strategy_steps[idx]['do_slowgain_solve'] = (i >= 2) # slow gain from cycle 3 (1-indexed) |
|
strategy_steps[idx]['do_fulljones_solve'] = False |
|
strategy_steps[idx]['peel_outliers'] = (i == 0) |
|
strategy_steps[idx]['peel_bright_sources'] = False |
|
strategy_steps[idx]['max_normalization_delta'] = 0.3 |
|
strategy_steps[idx]['scale_normalization_delta'] = True |
|
|
|
strategy_steps[idx]['solve_min_uv_lambda'] = 50 |
|
strategy_steps[idx]['fast_timestep_sec'] = ${FAST_TIMESTEP_SEC} |
|
strategy_steps[idx]['medium_timestep_sec'] = ${MEDIUM_TIMESTEP_SEC} |
|
strategy_steps[idx]['slow_timestep_joint_sec'] = ${SLOW_TIMESTEP_SEC} |
|
strategy_steps[idx]['slow_timestep_separate_sec'] = ${SLOW_TIMESTEP_SEC} |
|
strategy_steps[idx]['fulljones_timestep_sec'] = ${FULLJONES_TIMESTEP_SEC} |
|
|
|
strategy_steps[idx]['do_normalize'] = False |
|
strategy_steps[idx]['do_image'] = True |
|
strategy_steps[idx]['auto_mask'] = 3.0 |
|
strategy_steps[idx]['auto_mask_nmiter'] = 2 |
|
strategy_steps[idx]['max_nmiter'] = 8 |
|
strategy_steps[idx]['channel_width_hz'] = 4e6 |
|
strategy_steps[idx]['threshisl'] = 4.0 |
|
strategy_steps[idx]['threshpix'] = 5.0 |
|
|
|
strategy_steps[idx]['regroup_model'] = (i > 0) |
|
strategy_steps[idx]['max_directions'] = ${MAX_DIRECTIONS} |
|
strategy_steps[idx]['max_distance'] = None |
|
# target_flux must be a positive float; None triggers a rapthor bug when |
|
# max_directions is also set (total_flux < None comparison before the |
|
# target_number branch populates target_flux from the N-brightest sort). |
|
# A small value is fine: rapthor raises it to the Nth-brightest-source flux |
|
# when max_directions drives selection. |
|
# Use 0.01 Jy to ensure sources are found even in shallow images |
|
strategy_steps[idx]['target_flux'] = 0.01 |
|
|
|
if i == 0 or i == n_cycles - 1: |
|
strategy_steps[idx]['do_check'] = False |
|
else: |
|
strategy_steps[idx]['do_check'] = True |
|
strategy_steps[idx]['convergence_ratio'] = 0.95 |
|
strategy_steps[idx]['divergence_ratio'] = 1.1 |
|
strategy_steps[idx]['failure_ratio'] = 10.0 |
|
|
|
# Final pass: duplicate last selfcal step |
|
strategy_steps.append(strategy_steps[-1]) |
|
STRATEGY_EOF |
|
|
|
# ── Skymodel configuration ─────────────────────────────────────────────────── |
|
if [[ "${GENERATE_SKYMODEL}" == "true" ]] || [[ ! -f "${SKYMODEL}" ]]; then |
|
if [[ ! -f "${SKYMODEL}" ]]; then |
|
echo "WARNING: skymodel not found: ${SKYMODEL}" >&2 |
|
echo " Rapthor will generate one from the data." >&2 |
|
fi |
|
SKYMODEL_LINES="generate_initial_skymodel = True |
|
generate_initial_skymodel_data_fraction = 0.2" |
|
else |
|
REDUCE_SKYMODEL=0 |
|
if [[ "${SKYMODEL_MAX_SOURCES}" != "0" ]]; then |
|
REDUCE_SKYMODEL=1 |
|
elif awk "BEGIN { exit !(${SKYMODEL_MIN_FLUX_JY} > 0.0) }"; then |
|
REDUCE_SKYMODEL=1 |
|
fi |
|
|
|
if [[ "${REDUCE_SKYMODEL}" -eq 1 ]]; then |
|
REDUCED_SKYMODEL="${WORK_DIR}/$(basename "${SKYMODEL_FILE%.skymodel}")_top${SKYMODEL_MAX_SOURCES}.skymodel" |
|
python3 "${SCRIPTS_DIR}/reduce-skymodel.py" \ |
|
--input "${SKYMODEL}" \ |
|
--output "${REDUCED_SKYMODEL}" \ |
|
--max-sources "${SKYMODEL_MAX_SOURCES}" \ |
|
--min-flux "${SKYMODEL_MIN_FLUX_JY}" |
|
SKYMODEL="${REDUCED_SKYMODEL}" |
|
fi |
|
echo "Using external skymodel: ${SKYMODEL}" |
|
SKYMODEL_LINES="generate_initial_skymodel = False |
|
input_skymodel = ${SKYMODEL} |
|
regroup_input_skymodel = True" |
|
fi |
|
|
|
# ── Generate parset ────────────────────────────────────────────────────────── |
|
# Based on rapthor_di_easy.parset but adapted for combined DI+DD flow |
|
# and parameterised for the M2 Max dev target. |
|
|
|
PARSET="${WORK_DIR}/rapthor.parset" |
|
cat >"${PARSET}" <<PARSET_EOF |
|
# Generated by run-ical.sh — $(date -u +%Y-%m-%dT%H:%M:%SZ) |
|
# Based on rapthor_di_easy.parset (Vijay Mahatma, ska-sdp-ical) |
|
|
|
[global] |
|
dir_working = ${WORK_DIR} |
|
input_ms = ${MS} |
|
data_colname = DATA |
|
|
|
${SKYMODEL_LINES} |
|
|
|
download_initial_skymodel = False |
|
strategy = ${STRATEGY_FILE} |
|
|
|
selfcal_data_fraction = ${SELFCAL_DATA_FRACTION} |
|
final_data_fraction = ${FINAL_DATA_FRACTION} |
|
|
|
facet_layout = None |
|
dde_mode = faceting |
|
|
|
|
|
[calibration] |
|
use_included_skymodels = False |
|
use_image_based_predict = False |
|
|
|
dd_interval_factor = 3 |
|
dd_smoothness_factor = 3 |
|
|
|
llssolver = qr |
|
maxiter = ${MAXITER} |
|
propagatesolutions = True |
|
solveralgorithm = directioniterative |
|
onebeamperpatch = False |
|
stepsize = 0.02 |
|
stepsigma = 2.0 |
|
tolerance = 5e-3 |
|
|
|
fast_freqstep_hz = 1e6 |
|
fast_smoothnessconstraint = 3e6 |
|
fast_datause = single |
|
|
|
medium_freqstep_hz = 1e6 |
|
medium_smoothnessconstraint = 6e6 |
|
medium_datause = single |
|
|
|
slow_freqstep_hz = 1e6 |
|
slow_smoothnessconstraint = 3e6 |
|
slow_datause = dual |
|
|
|
fulljones_freqstep_hz = 1e6 |
|
fulljones_smoothnessconstraint = 0.0 |
|
|
|
correct_time_frequency_smearing = False |
|
|
|
|
|
[imaging] |
|
cellsize_arcsec = ${CELLSIZE_ARCSEC} |
|
robust = ${ROBUST} |
|
min_uv_lambda = 50 |
|
max_uv_lambda = 5000 |
|
mgain = 0.8 |
|
do_multiscale_clean = True |
|
|
|
dde_method = ${DDE_METHOD} |
|
|
|
filter_skymodel = True |
|
source_finder = bdsf |
|
|
|
save_visibilities = ${SAVE_VISIBILITIES} |
|
save_supplementary_images = False |
|
compress_selfcal_images = True |
|
compress_final_images = True |
|
|
|
idg_mode = ${IDG_MODE} |
|
mem_gb = ${MEM_PER_NODE_GB} |
|
|
|
apply_diagonal_solutions = True |
|
make_quv_images = False |
|
use_mpi = False |
|
reweight = False |
|
|
|
skip_final_major_iteration = True |
|
skip_corner_sectors = False |
|
|
|
# Use the input skymodel for diagnostics since Pan-STARRS doesn't cover |
|
# southern declinations (SKA-Low at dec ~ -32 deg) |
|
photometry_skymodel = ${SKYMODEL} |
|
astrometry_skymodel = ${SKYMODEL} |
|
|
|
|
|
[cluster] |
|
batch_system = ${RAPTHOR_BATCH_SYSTEM} |
|
max_nodes = ${RAPTHOR_MAX_NODES} |
|
cpus_per_task = ${RAPTHOR_CPUS_PER_TASK} |
|
max_cores = ${NUMTHREADS} |
|
max_threads = ${NUMTHREADS} |
|
deconvolution_threads = ${NUMTHREADS} |
|
mem_per_node_gb = ${MEM_PER_NODE_GB} |
|
# Without this, cwlrunner.py's _get_tmp_outdir_prefix() returns None and |
|
# --tmp-outdir-prefix is never passed to toil-cwl-runner. With --bypass-file-store |
|
# (always on — see cwlrunner.py's CWLRunner.setup()), that leaves each CWL job's |
|
# output location to cwltool's own per-job default, which is NOT guaranteed to be |
|
# the same shared, persistent path a later job on a *different* worker pod looks |
|
# for it at — surfacing as e.g. "FileNotFoundError: .../tmpXXXXXXXX/medium1_phase_0.h5parm" |
|
# once a downstream collect_h5parms-style step runs on a different pod than the |
|
# ddecal_solve step that produced it. Point it at shared storage so every pod sees |
|
# the same location regardless of which one produced/consumes a given file. |
|
global_scratch_dir = ${WORK_DIR}/.global_scratch |
|
|
|
cwl_runner = toil |
|
debug_workflow = ${RAPTHOR_DEBUG_WORKFLOW} |
|
keep_temporary_files = ${RAPTHOR_KEEP_TEMPORARY_FILES} |
|
|
|
# Disable internet access to skip downloading catalogs for diagnostics |
|
# (Pan-STARRS doesn't cover southern declinations like SKA-Low at dec ~ -32 deg) |
|
allow_internet_access = False |
|
PARSET_EOF |
|
|
|
# ── Log configuration ──────────────────────────────────────────────────────── |
|
echo "═══════════════════════════════════════════════════════════" |
|
echo " ICAL self-calibration via Rapthor" |
|
echo "═══════════════════════════════════════════════════════════" |
|
echo " MS: ${MS}" |
|
echo " Skymodel: ${SKYMODEL_FILE} (generate=${GENERATE_SKYMODEL})" |
|
echo " Reduced skymodel: ${SKYMODEL}" |
|
echo " Skymodel cap: ${SKYMODEL_MAX_SOURCES} brightest components" |
|
echo " Work dir: ${WORK_DIR}" |
|
echo " Strategy: ${STRATEGY_FILE}" |
|
echo " DD cycles: ${NCYCLES}" |
|
echo " Max directions: ${MAX_DIRECTIONS}" |
|
echo " Max iterations: ${MAXITER}" |
|
echo " Cell size: ${CELLSIZE_ARCSEC} arcsec" |
|
echo " Robust: ${ROBUST}" |
|
echo " Fast timestep: ${FAST_TIMESTEP_SEC} s" |
|
echo " Slow timestep: ${SLOW_TIMESTEP_SEC} s" |
|
echo " DDE method: ${DDE_METHOD}" |
|
echo " Save vis: ${SAVE_VISIBILITIES}" |
|
echo " Threads: ${NUMTHREADS}" |
|
echo "═══════════════════════════════════════════════════════════" |
|
echo "" |
|
echo "── Parset ──" |
|
cat "${PARSET}" |
|
echo "" |
|
echo "── Strategy ──" |
|
cat "${STRATEGY_FILE}" |
|
echo "═══════════════════════════════════════════════════════════" |
|
|
|
# ── Writable package overlay (Docker --user host uid cannot write site-packages) ─ |
|
# Rapthor/Toil spawn child Python processes, so patches must live on disk. Copy |
|
# lsmtool + rapthor + toil into WORK_DIR and prepend PYTHONPATH so edits apply |
|
# everywhere. toil is included (not just lsmtool/rapthor) because the Kubernetes |
|
# batch system patch below writes into toil/batchSystems/kubernetes.py; the real |
|
# dist-packages copy is read-only in this container (Errno 30), same reason the |
|
# other two packages are copied here rather than patched in place. |
|
OVERLAY="${WORK_DIR}/.python_site_overlay" |
|
LS_PKG=$(PYTHONPATH= python3 -c 'import lsmtool, pathlib; print(pathlib.Path(lsmtool.__file__).parent)') |
|
RT_PKG=$(PYTHONPATH= python3 -c 'import rapthor, pathlib; print(pathlib.Path(rapthor.__file__).parent)') |
|
TOIL_PKG=$(PYTHONPATH= python3 -c 'import toil, pathlib; print(pathlib.Path(toil.__file__).parent)') |
|
rm -rf "${OVERLAY}" |
|
mkdir -p "${OVERLAY}" "${WORK_DIR}/.global_scratch" |
|
# Spack layouts symlink site-packages into the store; copy real files so patches can write. |
|
cp -aL "${LS_PKG}" "${OVERLAY}/lsmtool" |
|
cp -aL "${RT_PKG}" "${OVERLAY}/rapthor" |
|
cp -aL "${TOIL_PKG}" "${OVERLAY}/toil" |
|
export PYTHONPATH="${OVERLAY}${PYTHONPATH:+:${PYTHONPATH}}" |
|
|
|
# ── Patch cwlrunner to pass an explicit Toil --maxCores ───────────────────── |
|
# Without it Toil derives maxCores from the *local* machine (cgroup cpu quota / |
|
# affinity). Inside a Slurm-launched pilot the coordinator's cgroup is sized to |
|
# the outer job's request_cpus (2), so any CWL step asking for cpus_per_task=4 |
|
# is rejected with "requesting 4 cores, more than the maximum of 2". Kubernetes |
|
# pilots dodge this only because their pods carry no cpu *limit*. TOIL_MAX_CORES |
|
# bounds the whole pool, not one node, so cap = max_nodes * cpus_per_task. |
|
# ``cpus_per_task=0`` means use available cores in Rapthor, but Toil rejects |
|
# zero; use the explicitly configured thread count as the corresponding cap. |
|
if [ -z "${TOIL_MAX_CORES:-}" ]; then |
|
if [ "${RAPTHOR_CPUS_PER_TASK:-0}" -gt 0 ]; then |
|
TOIL_MAX_CORES=$((${RAPTHOR_MAX_NODES:-4} * ${RAPTHOR_CPUS_PER_TASK})) |
|
else |
|
TOIL_MAX_CORES="${NUMTHREADS:-1}" |
|
fi |
|
fi |
|
export TOIL_MAX_CORES |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
path = pathlib.Path(sys.argv[1]) / "rapthor/lib/cwlrunner.py" |
|
src = path.read_text() |
|
# Quote style differs between rapthor builds (pre/post black formatting). |
|
markers = [ |
|
"self.args.extend(['--maxJobs', str(self.operation.max_nodes)])", |
|
'self.args.extend(["--maxJobs", str(self.operation.max_nodes)])', |
|
] |
|
if "tiger-patch: explicit maxCores" not in src: |
|
marker = next((m for m in markers if m in src), None) |
|
assert marker, "cwlrunner.py maxJobs marker not found" |
|
patch = ( |
|
marker |
|
+ "\n # [tiger-patch: explicit maxCores — see run-ical.sh]" |
|
+ "\n _max_cores = os.environ.get('TOIL_MAX_CORES')" |
|
+ "\n if _max_cores:" |
|
+ "\n self.args.extend(['--maxCores', _max_cores])" |
|
) |
|
src = src.replace(marker, patch, 1) |
|
path.write_text(src) |
|
PYEOF |
|
|
|
# ── Pin tool PYTHONPATH via EnvVarRequirement in every step CWL ────────────── |
|
# The leader exports PYTHONPATH=<overlay>:... and toil-cwl-runner runs with |
|
# --preserve-entire-environment, so worker tool processes can inherit the |
|
# overlay. That is poison for the /opt/view spack-env tools (e.g. |
|
# make_region_file.py): the overlay lsmtool is copied from the *toil* env's |
|
# older/partial lsmtool (no make_ds9_region_file), giving |
|
# ImportError: cannot import name 'make_ds9_region_file' from 'lsmtool.facet' |
|
# Pin the tool env at the CWL layer (cwltool applies EnvVarRequirement *after* |
|
# preserved env, so this wins regardless of how the worker was launched): the |
|
# CURRENT run's overlay first — tools must see the patched lsmtool (voronoi |
|
# empty-points guard) and rapthor — then pysite for the fileJobStore fix. The |
|
# overlay is copied from this same SIF, so tool/library versions always agree; |
|
# a stale inherited PYTHONPATH from an older run's overlay cannot leak in. |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import ruamel.yaml |
|
|
|
TOOL_PYTHONPATH = f"{sys.argv[1]}:/srv/storage/pysite" |
|
yaml = ruamel.yaml.YAML() |
|
yaml.preserve_quotes = True |
|
|
|
steps_dir = pathlib.Path(sys.argv[1]) / "rapthor/pipeline/steps" |
|
patched = 0 |
|
for path in sorted(steps_dir.glob("*.cwl")): |
|
doc = yaml.load(path.read_text()) |
|
if not isinstance(doc, dict) or doc.get("class") != "CommandLineTool": |
|
continue |
|
reqs = doc.setdefault("requirements", {}) |
|
if isinstance(reqs, list): |
|
env_req = next((r for r in reqs if r.get("class") == "EnvVarRequirement"), None) |
|
if env_req is None: |
|
env_req = {"class": "EnvVarRequirement", "envDef": {}} |
|
reqs.append(env_req) |
|
env_def = env_req.setdefault("envDef", {}) |
|
else: |
|
env_req = reqs.setdefault("EnvVarRequirement", {"envDef": {}}) |
|
env_def = env_req.setdefault("envDef", {}) |
|
if env_def.get("PYTHONPATH") == TOOL_PYTHONPATH: |
|
continue |
|
env_def["PYTHONPATH"] = TOOL_PYTHONPATH |
|
with path.open("w") as fh: |
|
yaml.dump(doc, fh) |
|
patched += 1 |
|
print(f"Pinned tool PYTHONPATH={TOOL_PYTHONPATH} in {patched} step CWLs", |
|
file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch Rapthor CWL memory hints for Kubernetes workers ─────────────────── |
|
# Several Rapthor step tools ship without ResourceRequirement memory hints, so |
|
# toil-cwl-runner starts them at 256 MiB and only retries at a higher value after |
|
# an OOM. Give the heavy DP3 steps an explicit request up front. |
|
if [ "${RAPTHOR_BATCH_SYSTEM}" = "kubernetes" ] && [ "${MEM_PER_NODE_GB}" != "0" ]; then |
|
python3 - "${OVERLAY}" "${MEM_PER_NODE_GB}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
overlay = pathlib.Path(sys.argv[1]) |
|
ram_mib = int(float(sys.argv[2]) * 1024) |
|
targets = [ |
|
overlay / "rapthor/pipeline/steps/ddecal_solve.cwl", |
|
overlay / "rapthor/pipeline/steps/predict_model_data.cwl", |
|
# subtract_sector_models.cwl has no ResourceRequirement at all (confirmed: |
|
# no ramMin/ramMax anywhere in the file) and got OOMKilled (exit 137) at |
|
# the default few-hundred-MiB cwltool/Toil fall back to — it does the same |
|
# class of full-MS-data work as the two steps above. |
|
overlay / "rapthor/pipeline/steps/subtract_sector_models.cwl", |
|
] |
|
for path in targets: |
|
try: |
|
src = path.read_text() |
|
except FileNotFoundError: |
|
print(f"WARNING: {path} not found — skipping ResourceRequirement patch", file=sys.stderr) |
|
continue |
|
if "class: ResourceRequirement" in src: |
|
continue |
|
marker = "hints:\n" |
|
if marker not in src: |
|
# No existing hints: block at all — 0-indent list items are valid CWL/YAML. |
|
block = ( |
|
"- class: ResourceRequirement\n" |
|
f" ramMin: {ram_mib}\n" |
|
f" ramMax: {ram_mib}\n" |
|
) |
|
src = src.rstrip() + "\n\nhints:\n" + block |
|
else: |
|
# Match this file's own indent for existing "- class: ..." items under |
|
# hints: (observed to vary between files — e.g. 0-indent in |
|
# ddecal_solve.cwl/predict_model_data.cwl, 2-indent in |
|
# subtract_sector_models.cwl). Mixing indents under one mapping key is |
|
# invalid YAML ("expected <block end>, but found '-'"). |
|
after_marker = src[src.index(marker) + len(marker) :] |
|
indent = after_marker[: len(after_marker) - len(after_marker.lstrip(" "))] |
|
block = ( |
|
f"{indent}- class: ResourceRequirement\n" |
|
f"{indent} ramMin: {ram_mib}\n" |
|
f"{indent} ramMax: {ram_mib}\n" |
|
) |
|
src = src.replace(marker, marker + block, 1) |
|
path.write_text(src) |
|
print(f"Patched {path} ResourceRequirement ram={ram_mib} MiB", file=sys.stderr) |
|
PYEOF |
|
fi |
|
|
|
# ── Patch CWL cores/ram so Toil worker pods request cpus_per_task ────────── |
|
# Rapthor only puts a ResourceRequirement on some steps. The rest start at |
|
# Toil's default 1 CPU, so five "8-core" workers still fit on one 36-vCPU |
|
# pool2 node and NUMTHREADS is cgroup-throttled to 1. Merge coresMin/ramMin |
|
# into an existing hints: list — never append a second hints: key (invalid YAML). |
|
if [ "${RAPTHOR_BATCH_SYSTEM}" = "kubernetes" ] && [ "${RAPTHOR_CPUS_PER_TASK:-0}" -gt 0 ]; then |
|
python3 - "${OVERLAY}" "${RAPTHOR_CPUS_PER_TASK}" "${MEM_PER_NODE_GB:-0}" <<'PYEOF' |
|
import pathlib, re, sys |
|
|
|
overlay = pathlib.Path(sys.argv[1]) |
|
cores = int(sys.argv[2]) |
|
ram_mib = int(float(sys.argv[3]) * 1024) if float(sys.argv[3]) > 0 else 0 |
|
steps = overlay / "rapthor/pipeline/steps" |
|
if not steps.is_dir(): |
|
print(f"WARNING: {steps} not found — skipping coresMin patch", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
|
|
def _stamp(src: str) -> str: |
|
extra_fields = [] |
|
if not re.search(r"^\s*coresMin:", src, re.M): |
|
extra_fields += [f"coresMin: {cores}", f"coresMax: {cores}"] |
|
if ram_mib and not re.search(r"^\s*ramMin:", src, re.M): |
|
extra_fields += [f"ramMin: {ram_mib}", f"ramMax: {ram_mib}"] |
|
if not extra_fields: |
|
return src |
|
m = re.search(r"^([ \t]*)- class: ResourceRequirement\n", src, re.M) |
|
if m: |
|
indent = m.group(1) |
|
extra = "".join(f"{indent} {line}\n" for line in extra_fields) |
|
return src[: m.end()] + extra + src[m.end() :] |
|
rr = "- class: ResourceRequirement\n" + "".join( |
|
f" {line}\n" for line in extra_fields |
|
) |
|
hm = re.search(r"^hints:\n", src, re.M) |
|
if hm: |
|
after = src[hm.end() :] |
|
indent = after[: len(after) - len(after.lstrip(" "))] |
|
block = "".join( |
|
(indent + line + "\n") if line else "\n" for line in rr.splitlines() |
|
) |
|
return src[: hm.end()] + block + src[hm.end() :] |
|
return src.rstrip() + "\n\nhints:\n" + rr |
|
|
|
|
|
n = 0 |
|
for path in sorted(steps.glob("*.cwl")): |
|
src = path.read_text() |
|
new = _stamp(src) |
|
if new == src: |
|
continue |
|
if len(re.findall(r"^hints:\s*$", new, re.M)) > 1: |
|
print(f"WARNING: {path.name} would have duplicate hints — skipped", file=sys.stderr) |
|
continue |
|
path.write_text(new) |
|
n += 1 |
|
print(f"Patched coresMin={cores} ramMin={ram_mib} on {n} CWL steps", file=sys.stderr) |
|
PYEOF |
|
fi |
|
|
|
# ── Python startup hook for nested Toil worker pods ───────────────────────── |
|
# cwltool creates tmp-out directories with tempfile.mkdtemp(..., mode 0700). |
|
# Kubernetes workers run as root while the parent Rapthor/Toil process may not, |
|
# so Toil can finish the workflow but fail while collecting outputs from those |
|
# root-owned directories. PYTHONPATH is preserved into nested workers; make their |
|
# temp dirs group/world traversable as soon as they are created. |
|
cat >"${OVERLAY}/sitecustomize.py" <<'PYEOF' |
|
import importlib.util |
|
import os |
|
from pathlib import Path |
|
import sys |
|
import tempfile |
|
|
|
_this = Path(__file__).resolve() |
|
for _entry in sys.path: |
|
_candidate = Path(_entry or os.getcwd()) / "sitecustomize.py" |
|
try: |
|
if not _candidate.is_file() or _candidate.resolve() == _this: |
|
continue |
|
except OSError: |
|
continue |
|
_spec = importlib.util.spec_from_file_location("_rapthor_upstream_sitecustomize", _candidate) |
|
if _spec and _spec.loader: |
|
_module = importlib.util.module_from_spec(_spec) |
|
_spec.loader.exec_module(_module) |
|
globals().update({k: v for k, v in vars(_module).items() if not k.startswith("__")}) |
|
break |
|
|
|
_original_mkdtemp = tempfile.mkdtemp |
|
|
|
|
|
def _chmod_shared_tmpdir(path: str) -> None: |
|
if "/tmp-out/" not in path: |
|
return |
|
current = path |
|
while "/tmp-out/" in current: |
|
try: |
|
os.chmod(current, 0o770) |
|
except OSError: |
|
pass |
|
parent = os.path.dirname(current) |
|
if parent == current: |
|
break |
|
current = parent |
|
|
|
|
|
def _shared_mkdtemp(*args, **kwargs): |
|
path = _original_mkdtemp(*args, **kwargs) |
|
_chmod_shared_tmpdir(path) |
|
return path |
|
|
|
|
|
os.umask(0o007) |
|
tempfile.mkdtemp = _shared_mkdtemp |
|
PYEOF |
|
|
|
# ── Patch Toil Kubernetes security-context loader ─────────────────────────── |
|
# The dev image has a local Kubernetes batch-system patch where |
|
# _load_kubernetes_object(file, cls) ignores file and always opens |
|
# "container.yaml". That breaks absolute --kubernetes*SecurityContext paths. |
|
# Patch the OVERLAY's copy of toil, not the real dist-packages one: this |
|
# container's site-packages is read-only (Errno 30), which is exactly why |
|
# lsmtool/rapthor/toil are all copied into OVERLAY above instead of patched |
|
# in place. PYTHONPATH already puts OVERLAY first, so the overlay copy is |
|
# what actually gets imported. |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
p = pathlib.Path(sys.argv[1]) / "toil" / "batchSystems" / "kubernetes.py" |
|
if not p.is_file(): |
|
print(f"WARNING: {p} not found — skipping security-context loader patch", file=sys.stderr) |
|
sys.exit(0) |
|
src = p.read_text() |
|
|
|
old = ' object_def = YAML.load(open("container.yaml").read())\n' |
|
new = ' object_def = YAML.load(open(file).read())\n' |
|
if old not in src: |
|
print(f"WARNING: Toil Kubernetes security-context loader patch target not found in {p}", file=sys.stderr) |
|
sys.exit(0) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (security-context loader uses requested file)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor ToilRunner local workdir ─────────────────────────────────── |
|
# Rapthor only provides a work directory for Slurm/Kubernetes batch systems; |
|
# single_machine therefore falls back to read-only /var/lib/toil in a pilot. |
|
# Make its shared _get_workdir() helper return TOIL_WORKDIR for every backend. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import rapthor.lib.cwlrunner as _cr |
|
p = pathlib.Path(_cr.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: explicit toil workdir]" |
|
if MARKER in src: |
|
sys.exit(0) |
|
|
|
old = " else:\n return None\n\n def _add_slurm_options" |
|
new = ( |
|
" " + MARKER + "\n" |
|
+ " _work_dir = os.environ.get('TOIL_WORKDIR')\n" |
|
+ " if _work_dir:\n" |
|
+ " return _work_dir\n" |
|
+ " else:\n" |
|
+ " return None\n\n" |
|
+ " def _add_slurm_options" |
|
) |
|
if old not in src: |
|
print(f"ERROR: Toil workdir patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (explicit Toil workdir)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor ToilRunner Kubernetes security-context args ──────────────── |
|
# Toil supports env vars for these, but passing explicit absolute paths avoids |
|
# stale/default relative names such as container.yaml in restarted WES runs. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import rapthor.lib.cwlrunner as _cr |
|
p = pathlib.Path(_cr.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: explicit kubernetes security context args]" |
|
if MARKER in src: |
|
sys.exit(0) |
|
|
|
old = ( |
|
" for k in k8s_keys:\n" |
|
" _os.environ[k] = _os.environ[k] # ensure visible to child process\n" |
|
) |
|
new = ( |
|
old |
|
+ " " + MARKER + "\n" |
|
+ " _pod_sec = _os.environ.get('TOIL_KUBERNETES_POD_SECURITY_CONTEXT')\n" |
|
+ " if _pod_sec:\n" |
|
+ " self.args.extend(['--kubernetesPodSecurityContext', _pod_sec])\n" |
|
+ " _container_sec = _os.environ.get('TOIL_KUBERNETES_SECURITY_CONTEXT')\n" |
|
+ " if _container_sec:\n" |
|
+ " self.args.extend(['--kubernetesSecurityContext', _container_sec])\n" |
|
) |
|
if old not in src: |
|
# The k8s_keys block is added by patch-rapthor-k8s.py in the old rapthor |
|
# image; the Karabo-based image ships kubernetes support natively and no |
|
# longer has it. The patch only matters for the kubernetes batch system. |
|
print(f"WARNING: Kubernetes security-context patch target not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (kubernetes security-context args)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Inject slurm_rest Toil batch system ────────────────────────────────────── |
|
# Copies the slurmrest dispatcher into the overlay so Rapthor can load it as |
|
# a batch system plugin regardless of whether it ships in the container image. |
|
# Looks first in SCRIPTS_DIR (inside the container), then on shared storage. |
|
_SLURMREST_SRC="${SCRIPTS_DIR}/toil_batch_system_slurmrest.py" |
|
if [ ! -f "${_SLURMREST_SRC}" ]; then |
|
_SLURMREST_SRC="/srv/storage/toil_batch_system_slurmrest.py" |
|
fi |
|
if [ -f "${_SLURMREST_SRC}" ]; then |
|
cp "${_SLURMREST_SRC}" "${OVERLAY}/rapthor/lib/toil_batch_systems/toil_batch_system_slurmrest.py" |
|
echo "Injected slurm_rest batch system from ${_SLURMREST_SRC}" >&2 |
|
else |
|
echo "WARNING: toil_batch_system_slurmrest.py not found; slurm_rest batch system unavailable" >&2 |
|
fi |
|
|
|
# ── Inject htcondor pool Toil batch system ─────────────────────────────────── |
|
# Same pattern as slurm_rest above: prefer the staged copy (tiger-stage-scripts |
|
# stages images/rapthor-htcondor/toil_batch_system_condor.py) over the one baked |
|
# into the container image, so batch-system fixes don't require an image rebuild. |
|
_CONDORBS_SRC="${SCRIPTS_DIR}/toil_batch_system_condor.py" |
|
if [ ! -f "${_CONDORBS_SRC}" ]; then |
|
_CONDORBS_SRC="/srv/storage/toil_batch_system_condor.py" |
|
fi |
|
if [ -f "${_CONDORBS_SRC}" ]; then |
|
cp "${_CONDORBS_SRC}" "${OVERLAY}/rapthor/lib/toil_batch_systems/toil_batch_system_condor.py" |
|
echo "Injected htcondor pool batch system from ${_CONDORBS_SRC}" >&2 |
|
else |
|
echo "WARNING: toil_batch_system_condor.py not found; htcondor pool batch system unavailable" >&2 |
|
fi |
|
|
|
# ── Patch lsmtool: guard voronoi() against empty bounding-box filter ───────── |
|
# lsmtool.facet.prepare_points_for_tessellate already handles the case where |
|
# every calibration source falls outside the image bounding box – it returns |
|
# an empty array early. The caller, voronoi(), never checked for that before |
|
# handing `points` to scipy.spatial.Voronoi, which raises: |
|
# ValueError: No points given |
|
# This happens with MAX_DIRECTIONS=1 because the single calibration patch can |
|
# sit outside the tight pixel footprint used for tessellation plotting. |
|
# Fix: avoid scipy on empty points, but still emit one facet per calibrator so |
|
# WSClean facet count matches the H5 (empty facet list → "1 directions vs 0 facets"). |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import lsmtool.facet as _lf |
|
p = pathlib.Path(_lf.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: guard voronoi against empty points_centre]" |
|
MARKER2 = "# [rapthor-patch: voronoi full-bbox fallback for out-of-box cals]" |
|
MARKER3 = "# [rapthor-patch: voronoi fallback facet centers at bbox middle]" |
|
if MARKER3 in src: |
|
sys.exit(0) # idempotent – already applied |
|
|
|
# Upgrade: cal-pixel centers can sit outside the WCS footprint → invalid polygons / empty centroid. |
|
BUG_RETURN = " return np.asarray(cal_coords, dtype=float).reshape(-1, 2), rect, regions\n" |
|
FIX_RETURN = ( |
|
" cx = 0.5 * (minx + maxx)\n" |
|
" cy = 0.5 * (miny + maxy)\n" |
|
" ncal = len(cal_coords)\n" |
|
" centers = np.tile(np.array([[cx, cy]], dtype=float), (ncal, 1))\n" |
|
" return centers, rect, regions\n" |
|
) |
|
if BUG_RETURN in src: |
|
src = src.replace(BUG_RETURN, FIX_RETURN, 1) |
|
if MARKER3 not in src: |
|
src = src.replace( |
|
MARKER2 + "\n", |
|
MARKER2 + "\n " + MARKER3 + "\n", |
|
1, |
|
) |
|
p.write_text(src) |
|
print(f"Upgraded voronoi fallback centers in {p}", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
# Upgrade path: replace the old "return empty regions" patch body. |
|
old_broken = ( |
|
" " + MARKER + "\n" |
|
" # All calibrators lie outside the bounding box; return an empty\n" |
|
" # tessellation instead of crashing scipy.spatial.Voronoi with a\n" |
|
" # zero-length array (e.g. MAX_DIRECTIONS=1 smoke-test runs).\n" |
|
" if len(points_centre) == 0:\n" |
|
" return points_centre, np.zeros((0, 2)), []\n" |
|
) |
|
new_broken = ( |
|
" " + MARKER + "\n" |
|
" " + MARKER2 + "\n" |
|
" " + MARKER3 + "\n" |
|
" # Calibrators outside the tessellation bbox: scipy Voronoi cannot run on\n" |
|
" # zero points. WSClean still needs one facet per direction in the H5.\n" |
|
" # Use one full-field rectangle per calibrator; facet centers must lie\n" |
|
" # inside the bbox (not raw cal pixels) so Rapthor/Shapely get valid polygons.\n" |
|
" if len(points_centre) == 0:\n" |
|
" if len(cal_coords) == 0:\n" |
|
" return points_centre, np.zeros((0, 2)), []\n" |
|
" minx, maxx, miny, maxy = bounding_box\n" |
|
" rect = np.array(\n" |
|
" [[minx, miny], [maxx, miny], [maxx, maxy], [minx, maxy]], dtype=float\n" |
|
" )\n" |
|
" regions = [list(range(4)) for _ in range(len(cal_coords))]\n" |
|
" cx = 0.5 * (minx + maxx)\n" |
|
" cy = 0.5 * (miny + maxy)\n" |
|
" ncal = len(cal_coords)\n" |
|
" centers = np.tile(np.array([[cx, cy]], dtype=float), (ncal, 1))\n" |
|
" return centers, rect, regions\n" |
|
) |
|
|
|
old_pristine = ( |
|
" points_centre, points = prepare_points_for_tessellate(\n" |
|
" cal_coords, bounding_box\n" |
|
" )\n" |
|
"\n" |
|
" # Compute Voronoi, sorting the output regions to match the order of the\n" |
|
" # input coordinates\n" |
|
" vor = scipy.spatial.Voronoi(points)" |
|
) |
|
new_pristine = ( |
|
" points_centre, points = prepare_points_for_tessellate(\n" |
|
" cal_coords, bounding_box\n" |
|
" )\n" |
|
"\n" |
|
+ new_broken |
|
+ "\n" |
|
" # Compute Voronoi, sorting the output regions to match the order of the\n" |
|
" # input coordinates\n" |
|
" vor = scipy.spatial.Voronoi(points)" |
|
) |
|
|
|
if old_broken in src: |
|
src = src.replace(old_broken, new_broken, 1) |
|
elif old_pristine in src: |
|
src = src.replace(old_pristine, new_pristine, 1) |
|
else: |
|
print(f"WARNING: patch target not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
p.write_text(src) |
|
print(f"Patched {p}", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor facet.py: wrap download_panstarrs call in try/except ─────── |
|
# Pan-STARRS doesn't cover southern declinations (dec < -30°), so download_panstarrs() |
|
# returns an empty file which causes lsmtool to crash with "No data lines found". |
|
# Patch: wrap the call in find_astrometry_offsets with try/except to gracefully skip. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
try: |
|
import rapthor.lib.facet as _rf |
|
except ModuleNotFoundError: |
|
# Newer (Karabo-based) rapthor moved facet code into lsmtool.facet. |
|
print("WARNING: rapthor.lib.facet not present — skipping panstarrs patch", file=sys.stderr) |
|
sys.exit(0) |
|
p = pathlib.Path(_rf.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: wrap download_panstarrs in try/except]" |
|
if MARKER in src: |
|
print(f"Patch already applied to {p}", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
# The code structure is: |
|
# if comparison_skymodel is None: |
|
# comparison_skymodel = self.download_panstarrs() |
|
# We need to wrap the download_panstarrs() call in try/except |
|
old = """ if comparison_skymodel is None: |
|
comparison_skymodel = self.download_panstarrs()""" |
|
|
|
new = """ if comparison_skymodel is None: |
|
""" + MARKER + """ |
|
try: |
|
comparison_skymodel = self.download_panstarrs() |
|
except Exception as e: |
|
import logging |
|
logging.warning(f'Pan-STARRS download failed (likely outside coverage): {e}') |
|
return""" |
|
|
|
if old not in src: |
|
print(f"WARNING: find_astrometry_offsets patch target not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p}", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch lsmtool Facet.__init__ (new home of rapthor.lib.facet.Facet) ─────── |
|
# Same robustness fix as the rapthor.lib.facet block below: WSClean-produced |
|
# facet regions can yield invalid Shapely polygons (NaN pixels, bow-ties); |
|
# .centroid is then an empty Point and calculate_image_diagnostics crashes with |
|
# "Error: getX called on empty Point". buffer(0) repairs the polygon and |
|
# representative_point() always returns an interior point. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import lsmtool.facet as _lf |
|
p = pathlib.Path(_lf.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: lsmtool Facet polygon buffer + representative_point]" |
|
if MARKER in src: |
|
sys.exit(0) |
|
|
|
old_poly = """ polygon_vertices = [(x, y) for x, y in zip(x_values, y_values)] |
|
self.polygon = Polygon(polygon_vertices) |
|
""" |
|
new_poly = """ polygon_vertices = [(x, y) for x, y in zip(x_values, y_values)] |
|
""" + MARKER + """ |
|
_poly = Polygon(polygon_vertices) |
|
if not _poly.is_valid: |
|
_poly = _poly.buffer(0) |
|
self.polygon = _poly |
|
""" |
|
|
|
old_centroid = """ self.ra_centroid, self.dec_centroid = map( |
|
float, |
|
self.wcs.wcs_pix2world( |
|
self.polygon.centroid.x, |
|
self.polygon.centroid.y, |
|
WCS_ORIGIN, |
|
), |
|
) |
|
""" |
|
new_centroid = """ _rp = self.polygon.representative_point() |
|
self.ra_centroid, self.dec_centroid = map( |
|
float, |
|
self.wcs.wcs_pix2world(_rp.x, _rp.y, WCS_ORIGIN), |
|
) |
|
""" |
|
|
|
if old_poly not in src or old_centroid not in src: |
|
print(f"WARNING: lsmtool Facet patch targets not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
src = src.replace(old_poly, new_poly, 1) |
|
src = src.replace(old_centroid, new_centroid, 1) |
|
p.write_text(src) |
|
print(f"Patched {p} (lsmtool Facet robust polygon)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor Facet.__init__: robust pixel polygon for ds9 / diagnostics ── |
|
# WSClean-produced facet regions can yield invalid Shapely polygons (NaN pixels, |
|
# bow-ties); .centroid is then an empty Point and calculate_image_diagnostics crashes. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
try: |
|
import rapthor.lib.facet as _rf |
|
except ModuleNotFoundError: |
|
print("WARNING: rapthor.lib.facet not present — skipping Facet polygon patch", file=sys.stderr) |
|
sys.exit(0) |
|
p = pathlib.Path(_rf.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: Facet polygon buffer + representative_point]" |
|
if MARKER in src: |
|
sys.exit(0) |
|
|
|
old = """ polygon_vertices = [(x, y) for x, y in zip(x_values, y_values)] |
|
self.polygon = Polygon(polygon_vertices) |
|
|
|
# Find the size and center coordinates of the facet |
|
xmin, ymin, xmax, ymax = self.polygon.bounds |
|
self.size = min(0.5, max(xmax-xmin, ymax-ymin) * |
|
abs(self.wcs.wcs.cdelt[0])) # degrees |
|
self.x_center = xmin + (xmax - xmin)/2 |
|
self.y_center = ymin + (ymax - ymin)/2 |
|
self.ra_center, self.dec_center = map( |
|
float, self.wcs.wcs_pix2world(self.x_center, self.y_center, misc.WCS_ORIGIN) |
|
) |
|
|
|
# Find the centroid of the facet |
|
self.ra_centroid, self.dec_centroid = map( |
|
float, self.wcs.wcs_pix2world(self.polygon.centroid.x, |
|
self.polygon.centroid.y, |
|
misc.WCS_ORIGIN) |
|
)""" |
|
|
|
new = """ polygon_vertices = [(x, y) for x, y in zip(x_values, y_values)] |
|
""" + MARKER + """ |
|
_poly = Polygon(polygon_vertices) |
|
if not _poly.is_valid: |
|
_poly = _poly.buffer(0) |
|
self.polygon = _poly |
|
|
|
# Find the size and center coordinates of the facet |
|
xmin, ymin, xmax, ymax = self.polygon.bounds |
|
self.size = min(0.5, max(xmax-xmin, ymax-ymin) * |
|
abs(self.wcs.wcs.cdelt[0])) # degrees |
|
self.x_center = xmin + (xmax - xmin)/2 |
|
self.y_center = ymin + (ymax - ymin)/2 |
|
self.ra_center, self.dec_center = map( |
|
float, self.wcs.wcs_pix2world(self.x_center, self.y_center, misc.WCS_ORIGIN) |
|
) |
|
|
|
# Interior reference point in pixel space (centroid can be empty for invalid polys) |
|
_rp = self.polygon.representative_point() |
|
self.ra_centroid, self.dec_centroid = map( |
|
float, self.wcs.wcs_pix2world(_rp.x, _rp.y, misc.WCS_ORIGIN) |
|
)""" |
|
|
|
if old not in src: |
|
print(f"WARNING: Facet.__init__ patch target not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p}", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor image op: allow missing h5parm without disabling facets ───── |
|
# Some runs have no h5parm for early image steps; keep faceting inputs enabled |
|
# and only null out h5parm serialization when filename is absent. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import rapthor.operations.image as _img |
|
p = pathlib.Path(_img.__file__) |
|
src = p.read_text() |
|
|
|
OLD_MARKER = "# [rapthor-patch: Image set apply_none when h5parm missing]" |
|
NEW_MARKER = "# [rapthor-patch: Image guard missing h5parm path]" |
|
if NEW_MARKER in src: |
|
sys.exit(0) |
|
|
|
# Undo older patch if present, then apply the safer replacement. |
|
src = src.replace( |
|
" " + OLD_MARKER + "\n" |
|
" if self.field.h5parm_filename is None:\n" |
|
" self.apply_none = True\n", |
|
"", |
|
) |
|
|
|
old = " h5parm = CWLFile(self.field.h5parm_filename).to_json() if not self.apply_none else None\n" |
|
new = ( |
|
" " + NEW_MARKER + "\n" |
|
" h5_ok = (self.field.h5parm_filename is not None)\n" |
|
" h5parm = CWLFile(self.field.h5parm_filename).to_json() if (h5_ok and not self.apply_none) else None\n" |
|
) |
|
if old not in src: |
|
print(f"WARNING: image.py h5parm patch target not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
src = src.replace(old, new, 1) |
|
p.write_text(src) |
|
print(f"Patched {p}", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor parset validation: allow batch_system=kubernetes ────────── |
|
# parset.py's Parset.__check_and_adjust only allows single_machine/slurm/ |
|
# slurm_static for [cluster] batch_system, so RAPTHOR_BATCH_SYSTEM=kubernetes |
|
# fails parset validation before any CWL step runs — even though ToilRunner |
|
# elsewhere (see the teardown patch below) already recognizes 'kubernetes' as |
|
# a valid self.operation.batch_system value. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import rapthor.lib.parset as _ps |
|
p = pathlib.Path(_ps.__file__) |
|
src = p.read_text() |
|
|
|
old = '"batch_system": ("single_machine", "slurm", "slurm_static"),' |
|
new = '"batch_system": ("single_machine", "slurm", "slurm_static", "kubernetes"),' |
|
if new in src: |
|
sys.exit(0) # idempotent - already applied |
|
if old not in src: |
|
print(f"ERROR: parset.py batch_system validation target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (batch_system accepts 'kubernetes')", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch Toil appliance-image existence check ────────────────────────────── |
|
# checkDockerImageExists() does an anonymous HEAD request against the |
|
# registry's v2 manifest API before Toil's kubernetes batchSystem launches |
|
# worker pods. Harbor requires auth for that, so it 401s and Toil raises |
|
# ApplianceImageNotFound even though Kubernetes (using its own |
|
# imagePullSecrets) pulls the same image just fine. There is no supported env |
|
# var to skip this (TOIL_SKIP_IMAGE_CHECK is not read anywhere in this Toil |
|
# version) — patch the redundant pre-check out instead. Patched on the |
|
# OVERLAY's toil copy (see the writable overlay at the top of this script). |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
p = pathlib.Path(sys.argv[1]) / "toil" / "__init__.py" |
|
if not p.is_file(): |
|
print(f"WARNING: {p} not found — skipping appliance image check patch", file=sys.stderr) |
|
sys.exit(0) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: skip appliance image existence check]" |
|
if MARKER in src: |
|
sys.exit(0) # idempotent - already applied |
|
|
|
old = ( |
|
" if currentCommit in appliance:\n" |
|
" return appliance\n" |
|
" registryName, imageName, tag = parseDockerAppliance(appliance)\n" |
|
) |
|
new = ( |
|
" if currentCommit in appliance:\n" |
|
" return appliance\n" |
|
" " + MARKER + "\n" |
|
" return appliance\n" |
|
" registryName, imageName, tag = parseDockerAppliance(appliance)\n" |
|
) |
|
if old not in src: |
|
print(f"ERROR: Toil appliance image check patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (skip appliance image existence check)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch Toil kubernetes worker pods to mount shared storage ─────────────── |
|
# Toil's file jobStore (--jobStore file:/srv/storage/...) must be visible at |
|
# the SAME absolute path inside every worker pod it launches, or the worker |
|
# fails instantly with NoSuchJobStoreException. None of Toil's built-in |
|
# TOIL_KUBERNETES_* env vars do this — TOIL_KUBERNETES_HOST_PATH only mounts |
|
# Toil's own internal per-node workDir/coordination dirs, not arbitrary paths. |
|
# wes-mildtec-k8s-smoke.sh already documents/exports a |
|
# TOIL_KUBERNETES_EXTRA_HOSTPATH env var for exactly this, but nothing ever |
|
# taught Toil to read it — add that mount here (idempotent, keyed by env var |
|
# presence so this is a no-op unless the caller opts in). |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
p = pathlib.Path(sys.argv[1]) / "toil" / "batchSystems" / "kubernetes.py" |
|
if not p.is_file(): |
|
print(f"WARNING: {p} not found — skipping extra hostpath mount patch", file=sys.stderr) |
|
sys.exit(0) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: mount TOIL_KUBERNETES_EXTRA_HOSTPATH]" |
|
if MARKER in src: |
|
sys.exit(0) # idempotent - already applied |
|
|
|
# TOIL_KUBERNETES_EXTRA_PVC_CLAIM (+ the existing TOIL_KUBERNETES_EXTRA_HOSTPATH |
|
# as the mount path) binds a PersistentVolumeClaim instead of a hostPath. A |
|
# hostPath is only actually shared storage on a single-node cluster where |
|
# every worker pod lands on the same node as the leader — on a real multi-node |
|
# cluster (this patch's whole reason to exist) each worker pod would get an |
|
# empty, node-local directory disconnected from the leader's data instead. |
|
# Falls back to hostPath when no PVC claim is given, so the local dev cluster |
|
# (which has no such PVC) is unaffected. |
|
old = " if self.host_path is not None:\n" |
|
new = ( |
|
" " + MARKER + "\n" |
|
+ " _extra_hostpath = os.environ.get(\"TOIL_KUBERNETES_EXTRA_HOSTPATH\")\n" |
|
+ " _extra_pvc_claim = os.environ.get(\"TOIL_KUBERNETES_EXTRA_PVC_CLAIM\")\n" |
|
+ " if _extra_pvc_claim and _extra_hostpath:\n" |
|
+ " from kubernetes.client import V1PersistentVolumeClaimVolumeSource\n" |
|
+ " _pvc_source = V1PersistentVolumeClaimVolumeSource(claim_name=_extra_pvc_claim)\n" |
|
+ " volumes.append(V1Volume(name=\"extra-pvc\", persistent_volume_claim=_pvc_source))\n" |
|
+ " mounts.append(V1VolumeMount(mount_path=_extra_hostpath, name=\"extra-pvc\"))\n" |
|
+ " elif _extra_hostpath:\n" |
|
+ " mount_host_path(\"extra-hostpath\", _extra_hostpath, _extra_hostpath)\n" |
|
+ old |
|
) |
|
if old not in src: |
|
print(f"ERROR: Toil kubernetes extra-hostpath patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (mounts TOIL_KUBERNETES_EXTRA_HOSTPATH/EXTRA_PVC_CLAIM into worker pods)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch Toil kubernetes worker pods with a node-local emptyDir scratch ──── |
|
# CephFS PVC I/O showed up as iowait in benchmon. Keep the shared jobStore |
|
# on the PVC, but put TMPDIR / CWL temp on an emptyDir so hot intermediates |
|
# stay on the node. Opt-in via TOIL_KUBERNETES_LOCAL_SCRATCH=1; mount path |
|
# defaults to /scratch (override with TOIL_KUBERNETES_LOCAL_SCRATCH_PATH). |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
p = pathlib.Path(sys.argv[1]) / "toil" / "batchSystems" / "kubernetes.py" |
|
if not p.is_file(): |
|
print(f"WARNING: {p} not found — skipping local scratch patch", file=sys.stderr) |
|
sys.exit(0) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: mount TOIL_KUBERNETES_LOCAL_SCRATCH emptyDir]" |
|
if MARKER in src: |
|
sys.exit(0) |
|
|
|
old = " if self.host_path is not None:\n" |
|
if old not in src: |
|
# Prefer anchoring after the EXTRA_PVC block if the host_path line moved. |
|
old = " " + "# [rapthor-patch: mount TOIL_KUBERNETES_EXTRA_HOSTPATH]\n" |
|
if old not in src: |
|
print(f"ERROR: Toil kubernetes local-scratch patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
new = ( |
|
" " + MARKER + "\n" |
|
# Use module-level V1EmptyDirVolumeSource — a local import would shadow it |
|
# and raise UnboundLocalError on Toil's own emptyDir workdir path when |
|
# LOCAL_SCRATCH is unset. |
|
+ " if os.environ.get(\"TOIL_KUBERNETES_LOCAL_SCRATCH\", \"\").lower() in (\"1\", \"true\", \"yes\"):\n" |
|
+ " _scratch_path = os.environ.get(\"TOIL_KUBERNETES_LOCAL_SCRATCH_PATH\", \"/scratch\")\n" |
|
+ " volumes.append(V1Volume(name=\"local-scratch\", empty_dir=V1EmptyDirVolumeSource()))\n" |
|
+ " mounts.append(V1VolumeMount(mount_path=_scratch_path, name=\"local-scratch\"))\n" |
|
+ old |
|
) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (mounts emptyDir local scratch into worker pods)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch Toil kubernetes worker pods to require a specific node pool ─────── |
|
# Toil's own Placement.required_labels mechanism (kubernetes.py) already |
|
# builds a proper nodeAffinity — it's just only ever populated from CWL |
|
# accelerator/GPU hints. This cluster's small static pool (pool1) also runs |
|
# rook-ceph/Harbor/job-gateway itself, and sizing a task to *accidentally* |
|
# overflow it (the previous approach) only forces a scale-up on the first |
|
# task that doesn't fit — every other task keeps landing on pool1 and |
|
# contending with cluster infra. Read TOIL_KUBERNETES_REQUIRED_NODE_LABEL |
|
# ("key=value") and require it directly so every worker pod targets the |
|
# large autoscaling pool explicitly. No-op (falls back to default scheduling) |
|
# when unset, so clusters without such a label are unaffected. |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib, sys |
|
|
|
p = pathlib.Path(sys.argv[1]) / "toil" / "batchSystems" / "kubernetes.py" |
|
if not p.is_file(): |
|
print(f"WARNING: {p} not found — skipping required-node-label patch", file=sys.stderr) |
|
sys.exit(0) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: require TOIL_KUBERNETES_REQUIRED_NODE_LABEL]" |
|
if MARKER in src: |
|
sys.exit(0) # idempotent - already applied |
|
|
|
old = " placement.set_preemptible(job_desc.preemptible)\n" |
|
new = ( |
|
old |
|
+ " " + MARKER + "\n" |
|
+ " _required_label = os.environ.get(\"TOIL_KUBERNETES_REQUIRED_NODE_LABEL\")\n" |
|
+ " if _required_label and \"=\" in _required_label:\n" |
|
+ " _label_key, _label_value = _required_label.split(\"=\", 1)\n" |
|
+ " placement.required_labels.append((_label_key, [_label_value]))\n" |
|
) |
|
if old not in src: |
|
print(f"ERROR: Toil kubernetes placement patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (requires TOIL_KUBERNETES_REQUIRED_NODE_LABEL on worker pods)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch Toil's appliance-existence check for token-auth registries ──────── |
|
# Toil validates TOIL_APPLIANCE_SELF by GETting the registry manifest URL with |
|
# no auth. Harbor answers 401 with a Bearer challenge even for PUBLIC projects |
|
# (the standard Docker token dance, which Toil does not perform), so the check |
|
# fails and the batch system aborts before it ever creates a worker pod: |
|
# toil.ApplianceImageNotFound: ... The HTTP status returned was 401. |
|
# forceDockerAppliance is Toil's own supported bypass. Worker pods still pull |
|
# the image normally -- they inherit imagePullSecrets from the pilot's |
|
# TOIL_KUBERNETES_SERVICE_ACCOUNT. |
|
python3 - "${OVERLAY}" <<'PYEOF' |
|
import pathlib |
|
import sys |
|
|
|
p = pathlib.Path(sys.argv[1]) / "toil" / "batchSystems" / "kubernetes.py" |
|
src = p.read_text() |
|
|
|
old = "applianceSelf()" |
|
new = "applianceSelf(forceDockerAppliance=True)" |
|
if new in src: |
|
sys.exit(0) |
|
if old not in src: |
|
print(f"ERROR: appliance-check patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(src.replace(old, new, 1)) |
|
print(f"Patched {p} (skip appliance manifest check; registry uses token auth)", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Patch rapthor ToilRunner teardown for Kubernetes cross-UID tmp dirs ────── |
|
# Kubernetes worker pods can create tmp-out directories as root. The parent |
|
# Rapthor process may run as UID 1000 and must not rmtree those directories. |
|
python3 - <<'PYEOF' |
|
import pathlib, sys |
|
|
|
import rapthor.lib.cwlrunner as _cr |
|
p = pathlib.Path(_cr.__file__) |
|
src = p.read_text() |
|
|
|
MARKER = "# [rapthor-patch: skip kubernetes tmp-out cleanup]" |
|
if MARKER in src: |
|
sys.exit(0) |
|
|
|
old = ( |
|
" \"\"\"\n" |
|
" if not self.operation.keep_temporary_files:\n" |
|
" # Remove directories used for storing intermediate job results" |
|
) |
|
new = ( |
|
" \"\"\"\n" |
|
" " + MARKER + "\n" |
|
" if self.operation.batch_system == 'kubernetes':\n" |
|
" return super().teardown()\n" |
|
" if not self.operation.keep_temporary_files:\n" |
|
" # Remove directories used for storing intermediate job results" |
|
) |
|
|
|
start = src.find("class ToilRunner(CWLRunner):") |
|
end = src.find("class CWLToolRunner(CWLRunner):") |
|
if start == -1 or end == -1: |
|
print(f"WARNING: ToilRunner class bounds not found in {p} — skipping", file=sys.stderr) |
|
sys.exit(0) |
|
|
|
before, body, after = src[:start], src[start:end], src[end:] |
|
if old not in body: |
|
print(f"ERROR: ToilRunner teardown patch target not found in {p}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
body = body.replace(old, new, 1) |
|
patched = before + body + after |
|
if MARKER not in patched: |
|
print(f"ERROR: ToilRunner teardown marker missing after patching {p}", file=sys.stderr) |
|
sys.exit(1) |
|
p.write_text(patched) |
|
print(f"Patched {p}", file=sys.stderr) |
|
PYEOF |
|
|
|
# ── Clean up any stale pipeline state from failed image_* runs ────────────── |
|
# toil job stores and CWL pipelines can get into a stuck state if a previous run failed |
|
# Remove the entire image_N pipeline folder so rapthor regenerates with fresh config |
|
for i in 0 1 2 3 4; do |
|
pipeline_dir="${WORK_DIR}/pipelines/image_${i}" |
|
log_dir="${WORK_DIR}/logs/image_${i}" |
|
if [ -d "${pipeline_dir}/jobstore" ]; then |
|
echo "Removing stale pipeline: ${pipeline_dir}" |
|
rm -rf "${pipeline_dir}" |
|
rm -rf "${log_dir}" |
|
fi |
|
done |
|
|
|
# ── Run Rapthor ────────────────────────────────────────────────────────────── |
|
# Do not append Rapthor's console logger to rapthor.log: it already writes the |
|
# same records to that file, which otherwise produces duplicate lines. |
|
exec 1>&3 2>&4 |
|
rapthor "${PARSET}" |
|
|
|
if [ "${BENCHMON_DRAIN_S}" -gt 0 ]; then |
|
echo "Keeping pilot alive for ${BENCHMON_DRAIN_S}s so BenchMon can flush..." |
|
sleep "${BENCHMON_DRAIN_S}" |
|
fi |
|
|
|
echo "Done." |
