From cc557bf7a912213c901342f0afbab8e905cbe296 Mon Sep 17 00:00:00 2001 From: Ashton Graves Date: Thu, 22 Oct 2026 09:21:41 -0500 Subject: [PATCH 1/6] Add osg-htc/nrp-provisioner: the HTCondor provisioner for NRP Nautilus The demand-driven provisioner that launches osg-htc/nrp-ospool-ep worker pods on the NRP Nautilus Kubernetes cluster for idle OSPool jobs. Moved here from github.com/ashtongraves/nrp-provisioner, whose provisioner/ directory is kept identical to this one; deployment configs stay there. Rocky 10, OSG 25, HTCondor 25 (htcondor2 bindings). Includes its README, unit tests and an image test (tests/image.sh). Co-Authored-By: Claude Opus 5.5 --- osg-htc/nrp-provisioner/03_condor_sec.config | 4 + osg-htc/nrp-provisioner/Dockerfile | 52 ++ .../BSD-2-Clause-prp-htcondor-portal.txt | 25 + .../LICENSES/MIT-prp-osg-pool.txt | 21 + osg-htc/nrp-provisioner/README.md | 329 +++++++++++ osg-htc/nrp-provisioner/build-config.json | 7 + osg-htc/nrp-provisioner/provisioner_info.py | 63 ++ osg-htc/nrp-provisioner/provisioner_main.py | 78 +++ .../nrp-provisioner/provisioner_startup.sh | 41 ++ .../nrp-provisioner/setup_htcondor_creds.sh | 10 + osg-htc/nrp-provisioner/setup_k8s_creds.sh | 8 + osg-htc/nrp-provisioner/src/__init__.py | 1 + osg-htc/nrp-provisioner/src/event_loop.py | 148 +++++ .../src/provisioner_backoff.py | 118 ++++ .../src/provisioner_clustering.py | 240 ++++++++ .../src/provisioner_config_parser.py | 48 ++ .../src/provisioner_htcondor.py | 236 ++++++++ .../nrp-provisioner/src/provisioner_k8s.py | 558 ++++++++++++++++++ .../src/provisioner_logging.py | 74 +++ osg-htc/nrp-provisioner/tests/image.sh | 61 ++ .../tests/sample_osg_provisioner.conf | 24 + osg-htc/nrp-provisioner/tests/test_backoff.py | 213 +++++++ .../tests/test_htcondor_ads.py | 131 ++++ .../nrp-provisioner/tests/test_k8s_submit.py | 80 +++ osg-htc/nrp-provisioner/tests/test_logging.py | 64 ++ .../tests/test_schedd_query.py | 141 +++++ 26 files changed, 2775 insertions(+) create mode 100644 osg-htc/nrp-provisioner/03_condor_sec.config create mode 100644 osg-htc/nrp-provisioner/Dockerfile create mode 100644 osg-htc/nrp-provisioner/LICENSES/BSD-2-Clause-prp-htcondor-portal.txt create mode 100644 osg-htc/nrp-provisioner/LICENSES/MIT-prp-osg-pool.txt create mode 100644 osg-htc/nrp-provisioner/README.md create mode 100644 osg-htc/nrp-provisioner/build-config.json create mode 100644 osg-htc/nrp-provisioner/provisioner_info.py create mode 100644 osg-htc/nrp-provisioner/provisioner_main.py create mode 100755 osg-htc/nrp-provisioner/provisioner_startup.sh create mode 100755 osg-htc/nrp-provisioner/setup_htcondor_creds.sh create mode 100755 osg-htc/nrp-provisioner/setup_k8s_creds.sh create mode 100644 osg-htc/nrp-provisioner/src/__init__.py create mode 100644 osg-htc/nrp-provisioner/src/event_loop.py create mode 100644 osg-htc/nrp-provisioner/src/provisioner_backoff.py create mode 100644 osg-htc/nrp-provisioner/src/provisioner_clustering.py create mode 100644 osg-htc/nrp-provisioner/src/provisioner_config_parser.py create mode 100644 osg-htc/nrp-provisioner/src/provisioner_htcondor.py create mode 100644 osg-htc/nrp-provisioner/src/provisioner_k8s.py create mode 100644 osg-htc/nrp-provisioner/src/provisioner_logging.py create mode 100755 osg-htc/nrp-provisioner/tests/image.sh create mode 100644 osg-htc/nrp-provisioner/tests/sample_osg_provisioner.conf create mode 100644 osg-htc/nrp-provisioner/tests/test_backoff.py create mode 100644 osg-htc/nrp-provisioner/tests/test_htcondor_ads.py create mode 100644 osg-htc/nrp-provisioner/tests/test_k8s_submit.py create mode 100644 osg-htc/nrp-provisioner/tests/test_logging.py create mode 100644 osg-htc/nrp-provisioner/tests/test_schedd_query.py diff --git a/osg-htc/nrp-provisioner/03_condor_sec.config b/osg-htc/nrp-provisioner/03_condor_sec.config new file mode 100644 index 00000000..33b1383a --- /dev/null +++ b/osg-htc/nrp-provisioner/03_condor_sec.config @@ -0,0 +1,4 @@ +# Enable authenticated SSL +# using OSG certificates +AUTH_SSL_SERVER_CADIR = /etc/grid-security/certificates +AUTH_SSL_CLIENT_CADIR = /etc/grid-security/certificates diff --git a/osg-htc/nrp-provisioner/Dockerfile b/osg-htc/nrp-provisioner/Dockerfile new file mode 100644 index 00000000..7831d12d --- /dev/null +++ b/osg-htc/nrp-provisioner/Dockerfile @@ -0,0 +1,52 @@ +FROM rockylinux/rockylinux:10 + +# dependency needed by OSG +RUN dnf install -y 'dnf-command(config-manager)' && \ + dnf config-manager --set-enabled crb && \ + yum clean all + +# OSG RPMs +# osg-ca-certs must be recent enough for the current Let's Encrypt chains (YE1, ISRG Root YE) +RUN yum -y install https://repo.opensciencegrid.org/osg/25-main/osg-25-main-el10-release-latest.rpm && \ + yum -y install https://dl.fedoraproject.org/pub/epel/epel-release-latest-10.noarch.rpm && \ + yum -y install python3-pip less which && \ + yum -y install osg-ca-certs condor python3-condor && \ + yum clean all + + +RUN pip3 install --no-cache-dir kubernetes==37.0.1 + + +# Tell condor to use osg-ca-certs +ADD 03_condor_sec.config /etc/condor/config.d/03_condor_sec.config + +# configure python packages +ADD src /opt/provisioner/src + +# Copy /etc/skel explicitly: when built under amd64 emulation (e.g. on an arm64 Mac), EL10's useradd +# leaves the home empty, and without ~/.bash_profile 'su -l' never reads the ~/.bashrc the setup scripts write +RUN useradd provisioner && cp -a /etc/skel/. /home/provisioner/ && chown -R provisioner: /home/provisioner && chown -R provisioner /opt/provisioner + +RUN su provisioner -c '/usr/bin/ln -s /opt/provisioner/src /home/provisioner/src' + +ADD provisioner_main.py /home/provisioner/provisioner_main.py +ADD provisioner_info.py /home/provisioner/provisioner_info.py +RUN chown provisioner /home/provisioner/provisioner_main.py; chmod a+x /home/provisioner/provisioner_info.py; chown provisioner /home/provisioner/provisioner_info.py + +ADD setup_k8s_creds.sh /opt/provisioner/setup_k8s_creds.sh +ADD setup_htcondor_creds.sh /opt/provisioner/setup_htcondor_creds.sh +ADD provisioner_startup.sh /opt/provisioner/provisioner_startup.sh + +# +# Requires env variables +# CONDOR_HOST +# K8S_NAMESPACE +# CVMFS_MOUNTS + +# +# Optional env variable +# HTCONDOR_QUERY_INSECURE +# + +CMD ["/opt/provisioner/provisioner_startup.sh"] + diff --git a/osg-htc/nrp-provisioner/LICENSES/BSD-2-Clause-prp-htcondor-portal.txt b/osg-htc/nrp-provisioner/LICENSES/BSD-2-Clause-prp-htcondor-portal.txt new file mode 100644 index 00000000..4bb205df --- /dev/null +++ b/osg-htc/nrp-provisioner/LICENSES/BSD-2-Clause-prp-htcondor-portal.txt @@ -0,0 +1,25 @@ +BSD 2-Clause License + +Copyright (c) 2021, Igor Sfiligoi +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this + list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, + this list of conditions and the following disclaimer in the documentation + and/or other materials provided with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/osg-htc/nrp-provisioner/LICENSES/MIT-prp-osg-pool.txt b/osg-htc/nrp-provisioner/LICENSES/MIT-prp-osg-pool.txt new file mode 100644 index 00000000..90e9ac55 --- /dev/null +++ b/osg-htc/nrp-provisioner/LICENSES/MIT-prp-osg-pool.txt @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2019 Igor Sfiligoi + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/osg-htc/nrp-provisioner/README.md b/osg-htc/nrp-provisioner/README.md new file mode 100644 index 00000000..f57e529b --- /dev/null +++ b/osg-htc/nrp-provisioner/README.md @@ -0,0 +1,329 @@ +# nrp-provisioner + +Demand-driven HTCondor provisioner for OSG workloads on the NRP Nautilus Kubernetes cluster. One +provisioner pod per namespace reads idle jobs from whitelisted HTCondor schedds, groups them by +resource shape, and creates Kubernetes Jobs whose pods run the OSPool pilot (one static slot per pod). +The worker pods run `osg-htc/nrp-ospool-ep`. + +**Source:** `osg-htc/nrp-provisioner` in [opensciencegrid/images](https://github.com/opensciencegrid/images), +published as `hub.opensciencegrid.org/osg-htc/nrp-provisioner:25-release`. The deployment (per-namespace +configs and manifests), operations notes and incident history live in +[ashtongraves/nrp-provisioner](https://github.com/ashtongraves/nrp-provisioner), whose `provisioner/` +directory is kept identical to this one. Facts were checked against the live system in October 2026 +unless marked as inference. + +- [Layout](#layout) +- [How it works](#how-it-works) +- [Configuration](#configuration) +- [Authentication, and why queue reads fail](#authentication-and-why-queue-reads-fail) +- [Troubleshooting](#troubleshooting) +- [Building](#building) +- [Tests](#tests) +- [Background and pitfalls](#background-and-pitfalls) + +## Layout + +``` +Dockerfile, *.sh, *.py the image: startup scripts and provisioner_main.py +src/ Python library (the `src` package; it was `prp_provisioner` until 2026-10-22). + provisioner_htcondor.py is the only file that touches the HTCondor + bindings, provisioner_k8s.py the only one that builds Job bodies. +tests/ unit tests, the image test (image.sh) and a sample config +build-config.json build matrix for the images repo +LICENSES/ the original repositories' licenses (BSD-2-Clause, MIT); the code keeps + their copyright headers +``` + +## How it works + +``` + OSPool access points (schedds) OSPool central managers (cm-1/cm-2.ospool.osg-htc.org) + ap40, ap41, ap23, JLab, IceCube, PATh… collector = directory of ads (schedds, submitters, slots) + hold the actual job queues negotiator = matches jobs to slots each cycle + │ ▲ ▲ + provisioner reads idle jobs │ slots advertise themselves + (QUERY_JOB_ADS, anonymous) │ + ▼ │ + nrp-provisioner (one pod per namespace) ──creates──▶ k8s Jobs ──▶ worker pods (OSPool pilot) + groups idle jobs by shape, sizes pods one static slot per pod +``` + +- **There is no central OSPool job queue.** Jobs exist only in each schedd's queue. The collector + holds a summary ad per schedd and a **submitter ad per user per schedd** with idle and running + *counts*, but never the job ads. The negotiator pulls resource requests from schedds during each + cycle and keeps nothing you can query. So the provisioner must be able to read each schedd it + serves; collector access alone is never enough. +- **Worker pods join the pool** with the pool token. Any schedd in the pool can then claim them through + normal matchmaking. Schedds don't need to trust the provisioner, and the provisioner doesn't need + anything from a schedd to *run* jobs, only to *see* them. Jobs from schedds it can't read still land + on its pods when the shapes match. + +### Each loop (about every 120 s) + +1. Query idle jobs from the whitelisted schedds, filtered by `additional_requirements` and the + `apptainer` setting. +2. Query the collector for its own slots (`K8SProvisionerName`, `K8SNamespace`, `K8SDomain`), and + list its pods. +3. Group jobs and pods by shape: `CPUs;Memory;Disk;DiskVolumes;GPUs;GPUTypes;Labels`. Job memory is + rounded up to 1 GB and disk to 4 GB, so similar jobs share a group. Requests below 4096 MB or + 8000000 KB count as unset. The worker's `START` accepts memory within 1.5 GB and disk within 4.5 GB + of the request, deliberately consistent with the rounding. +4. Per group, keep `1 + idle/4` unclaimed pods (slowed further above 20), at most + `max_pods_per_cluster`. Stop adding when the group has `max_submit_pods_per_cluster` pods in total. +5. Submit a k8s Job for the missing pods. + +Worker pods exit after 20 minutes unclaimed (`STARTD_NOCLAIM_SHUTDOWN=1200`) and the provisioner +replaces them. A group that can never match would burn up to 20 pods every 20 minutes, which is what +the next two mechanisms limit. + +### Trial pods + +A group with **no claimed pods right now** gets at most `trial_pods` (default 2) instead of up to 20. +Once one is claimed, the group gets the usual number on the next loop. It's stateless (it looks only +at the current pods), so it also limits a mixed group whose matching jobs have run out, and a group +that matches again recovers on its first claim. The cost: a large new group ramps up one loop after +its first claim. + +### The back-off + +The provisioner can't see *why* a group never matches (job `Requirements` typos, site pinning, annex +jobs, GPU models the node affinity doesn't include), so it judges by outcome. A pod that ends +`Succeeded` with a startd that was never claimed (`State` never `Claimed`, `JobStarts` never above 0) +counts as an unclaimed exit. Two in a row with no claim pause the group for 1 hour, then 2, 4, up to +8. Any claim clears it. Failed or deleted pods (preemption, node problems) don't count. The state is +in memory and starts fresh when the provisioner restarts. + +We chose this over pre-matching jobs against a template slot because the template approach got GPU +groups wrong: a test slot's GPU isn't the GPU a real pod lands on, and one GPU group it predicted +"can't match" was in fact matched on real nodes. The back-off's blind spot is mixed groups: a shape +where *some* jobs match is never paused. Trial pods cover part of that. + +## Configuration + +Each namespace's `osg_provisioner.conf` is mounted from a ConfigMap **with `subPath`**, so changes need +a provisioner restart. Sections: `[DEFAULT]`, `[k8s]` (falls back to `[DEFAULT]` if missing) and +`[htcondor]`. + +| Setting | Section | Default | Meaning | +|---|---|---|---| +| `apptainer` | `[DEFAULT]` | `yes` | Whether worker pods can run containers; see below | +| `schedd_whitelist_regexp` | `[htcondor]` | `.*` | Schedds to read | +| `additional_requirements` | `[htcondor]` / `[k8s]` | | Extra job constraint for the schedd query. In `[k8s]`/`[DEFAULT]` it's also added to the worker's `START` | +| `force_k8s_namespace_matching` | any | `yes` | Only jobs whose `RequestK8SNamespace` matches the namespace | +| `app_name`, `k8s_domain` | `[DEFAULT]` | | Identify this provisioner's pods and slots. Two provisioners with the same `app_name` in one namespace count each other's pods | +| `k8s_image` | `[k8s]` | | Worker image | +| `max_pods_per_cluster` | `[k8s]` | `20` | Most unclaimed pods per group | +| `max_submit_pods_per_cluster` | `[k8s]` | `1000` | No new pods once a group has this many in total | +| `trial_pods` | `[k8s]` | `2` | A group with no claimed pods gets at most this many (`0` disables) | +| `backoff_unclaimed_exits` | `[k8s]` | `2` | Pause a group after this many unclaimed exits in a row (`0` disables) | +| `backoff_time` | `[k8s]` | `3600` | First pause, in seconds; doubled each time | +| `backoff_max_time` | `[k8s]` | `28800` | Longest pause, in seconds | +| `k8s_job_ttl` | `[k8s]` | `86400` | Delete worker Jobs this long after they finish | +| `sleep_time` | `[DEFAULT]` | `120` | Seconds between loops | +| `envs_dict`, `labels_dict`, `annotations_dict`, `tolerations_list`, `node_affinity_dict`, `config_volumes_dict`, `pvc_volumes_dict`, `priority_class`, `service_account_name`, `use_large_limit` | `[k8s]` | | Passed into the worker pod spec | + +### Apptainer, containers and user namespaces + +**Unprivileged user namespaces are disabled on the Nautilus nodes on purpose** (a CVE response). It's a +standing decision, not a bug. Apptainer in the worker image has no setuid helper, so it can't run, +and container jobs can't run here: the OSPool `START` clause +(`isUndefined(TARGET.SingularityImage) || MY.SINGULARITY_START_CLAUSE`) rejects any job with a +`SingularityImage`, and many jobs' own `Requirements` ask for `HasSingularity`/`HasSIF` too. + +`apptainer=yes|no` makes the provisioner, the worker validation and the pilot's image download agree. +The provisioner passes it to the workers as `PROVISIONER_APPTAINER`. All namespaces use `no`. + +| | `apptainer=yes` (default) | `apptainer=no` | +|---|---|---| +| Jobs provisioned for | only jobs with a `SingularityImage` | only jobs without one | +| Worker apptainer check | runs; failure aborts the pilot | skipped | +| Default container image | downloaded as a `.sif` | not downloaded (`OSG_DEFAULT_CONTAINER_CVMFS=1`) | + +Before the option existed, the config hard-coded `!isUndefined(SingularityImage)` while apptainer was +off, so it provisioned only for jobs that could never run, and nearly every pod idled out. + +How much work there is without containers depends on the community: + +| Namespace | Container-free share of matching idle jobs | +|---|---| +| `osg-opportunistic` | a few thousand of ~145k (October 2026) | +| `osg-icecube` | none (all of IceCube's matching GPU jobs set `SingularityImage`) | +| `osg-ligo` | about 1 in 1,100 | +| `osg-nrao` | about half | + +**CVMFS is needed either way.** It carries job software (`oasis`, community repos), the pilot's config +(`config-osg`) and the `HAS_CVMFS_*` attributes jobs match on. Only `singularity.opensciencegrid.org` +is container-specific. The provisioner adds the `cvmfs` PVC to every worker pod in code, not in the +deployment, so it looks unused. Don't delete it. + +## Authentication, and why queue reads fail + +- **The pool token** is an OSPool IDTOKEN: issuer `flock.opensciencegrid.org`, HS256, scopes `READ`, + `ADVERTISE_MASTER` and `ADVERTISE_STARTD`, no expiry. It's signed with the OSPool's **shared secret + key**, so only daemons holding that key (the OSPool central managers) can verify it. An access point + can't. +- **Reading a schedd's queue is therefore anonymous in practice.** The provisioner sets client + authentication, integrity and encryption to `OPTIONAL` (`HTCONDOR_QUERY_INSECURE=yes`). Anonymous + access normally goes over **SSL without a client certificate**: the client still verifies the + *server's* certificate. A schedd with `ALLOW_READ = *` lets that through. So a stale CA bundle on + our side looks exactly like "the site locked us out", and that has twice been the real cause. +- **Another OSPool `READ` token doesn't help;** the AP still can't verify it. If an AP truly requires + authentication, the options are: + 1. the AP re-allows anonymous reads; + 2. the AP creates its own **named** signing key (not `POOL`) and issues a token with `-authz READ` + from it, mounted next to the pool token. HTCondor picks the token whose issuer matches the server, + which only works if the AP's `TRUST_DOMAIN` differs from the OSPool's; + 3. SSL with a client certificate the AP maps to a read-only identity. + +## Troubleshooting + +Read the provisioner log with `kubectl logs -f deploy/ -n ` (recent lines; the log is +mirrored to stdout), or `/var/log/provisioner/logs/provisioner.log.YYYYMMDD` on the log PVC (full +history, one file per day). Each loop logs `Iteration started`, then one line per job group: +`Cluster '' n_jobs_idle … n_pods_unclaimed … min_pods …`, and `Submitted N pods as job …` when +it adds pods. The provisioner logs failures, never successful queries. + +### `Failed to query HTCondor schedd ''` + +If **every** whitelisted schedd fails, the loop fails (`Failed to query HTCondor`) and nothing is +provisioned. Confirm from the provisioner pod; it shows which auth methods were tried and why they +failed: + +```bash +kubectl exec -n deploy/ -- su -l provisioner -c '_CONDOR_TOOL_DEBUG="D_SECURITY:2" condor_q -debug -name -pool cm-1.ospool.osg-htc.org -totals 2>&1 | grep -E "my_methods|server replied|AUTHENTICATE|certificate|issuer|err [0-9]+:|Total for"' +``` + +`-debug` is required; `_CONDOR_TOOL_DEBUG` alone prints nothing. [nrp-provisioner](https://github.com/ashtongraves/nrp-provisioner)'s +`tests/live/check_schedds.py` does this for every whitelisted schedd. + +| Cause | What `-debug` shows | Fix | +|---|---|---| +| **Outdated CA bundle in the provisioner image** | `Error with certificate at depth: 0 … issuer = /C=US/O=Let's Encrypt/CN=YE1 … err 20: unable to get local issuer certificate` | Build on a newer OSG series. `osg-ca-certs` ≥ 1.144 (OSG 24/25) has Let's Encrypt's "Y" chains; OSG 23's last bundle (1.137) doesn't | +| **Old HTCondor client vs newer schedd** | `my_methods = 'SCITOKENS,FS'` … `required authentication … failed, so aborting command QUERY_JOB_ADS` | HTCondor 24/25 client: it reads the queue without starting an authentication handshake | +| **Schedd advertises a private address** | connection timeouts, ~20 s each | Unreachable from NRP. Harmless apart from slowing each loop | + +**When did it start?** Daily log files sort by date, so binary-search them for the first failure. +Don't `grep` every file: the logs go back years on network storage, and a full scan takes many +minutes. This assumes the failures are continuous once they start: + +```bash +cd /var/log/provisioner/logs; files=($(ls provisioner.log.* | sort)); s=; has() { head -c 20M "$1" | LC_ALL=C grep -q -F "Failed to query HTCondor schedd '$s'"; }; lo=0; hi=$((${#files[@]}-1)); if ! has "${files[$hi]}"; then echo "not failing in the latest file"; else while [ $lo -lt $hi ]; do mid=$(((lo+hi)/2)); if has "${files[$mid]}"; then hi=$mid; else lo=$((mid+1)); fi; done; echo "first failing file: ${files[$lo]}"; fi +``` + +### `Failed to submit N pods` + +Creating the worker Job failed, and the event loop doesn't log why. Reproduce it with a server-side +dry run (nrp-provisioner's `tests/live/dryrun_submit.py`, in a provisioner pod), which validates +permissions, quotas and admission webhooks without creating anything. Seen so far: the `kubernetes` client ≥ 37 validates +bodies client-side and rejected numeric quantities and a misplaced field (see +[Background](#background-and-pitfalls)). + +### Provisioner exits at start: `ConfigException: Service host/port is not set` + +The provisioner runs as the `provisioner` user via `su -l`, a clean login shell. The setup scripts +write the Kubernetes and HTCondor settings into `~/.bashrc`, which only `~/.bash_profile` sources. +EL10's `useradd` under amd64 emulation (Docker on an arm64 Mac) left the home empty, so the Dockerfile +copies `/etc/skel` itself. Check: `docker run --rm --entrypoint bash -c 'ls -A /home/provisioner'` +must include `.bash_profile`. + +### Pods start, idle about 20 minutes, exit unclaimed (churn) + +Common reasons nothing claims them: + +- **Container jobs while apptainer is unavailable.** Set `apptainer=no`. +- **Jobs aimed at an HTCondor annex** (`TargetAnnexName` set). These only match their owner's annex. + Excluded in `osg-opportunistic` by `additional_requirements`. +- **Jobs whose own `Requirements` can't match here:** pinned to another site (`GLIDEIN_Site == "…"`), + needing a GPU model the node affinity doesn't include, or typos. + +Trial pods and the back-off limit the cost; look for `suspending provisioning` / `back-off cleared` +in the log. To see *why* a slot and a job don't match, evaluate the job's `Requirements` and the +slot's `Start` clause by clause against a real slot ad from the collector, resized to the group's +shape. Expect false positives on GPU-specific clauses. + +### Pods stuck `Pending`: `Insufficient smarter-devices/fuse` / node affinity + +Usually large shapes (many CPUs or GPUs) that fit few nodes, combined with the FUSE device request and +the GPU node affinity. They wait until a node frees up. + +### Jobs with no status that never go away + +Worker Jobs have `backoffLimit: 0` and are deleted `k8s_job_ttl` after they *finish*. A Job whose pods +were deleted while Pending can be left with no active, succeeded or failed count, so it never +finishes. List and delete those older than a day: + +```bash +kubectl get jobs -n -o json | python3 -c 'import json,sys,datetime as d; n=d.datetime.now(d.timezone.utc); print("\n".join(j["metadata"]["name"] for j in json.load(sys.stdin)["items"] if not any(j["status"].get(k) for k in ("active","succeeded","failed")) and (n-d.datetime.fromisoformat(j["metadata"]["creationTimestamp"].replace("Z","+00:00"))).days>=1))' | xargs -r kubectl delete job -n +``` + +Worker pod problems (exit 143 after 5 minutes, `KeyError: 'ca'`, user namespace errors) are covered +with the worker image, `osg-htc/nrp-ospool-ep`. + +## Building + +```bash +docker build -t nrp-provisioner:local . +``` + +The image is Rocky 10, OSG 25, HTCondor 25 with the `htcondor2` bindings, Python 3.12 and a pinned +`kubernetes` client. It doesn't use the `BASE_*` build arguments; `build-config.json` (el10, OSG 25, +release) only names the build for the images repo's CI. + +## Tests + +**Unit tests** (`tests/`): `python3 -m unittest discover -s tests`, from this directory. +Without the HTCondor bindings (e.g. on a laptop) the `htcondor2` tests are skipped; inside the image +nothing is. + +| File | What it covers | +|---|---| +| `test_backoff.py` | the back-off state machine, trial pods, and the event loop skipping paused groups | +| `test_htcondor_ads.py` | job/startd ad conversion (rounding, "too small" means unset, expressions, stringified values) and job grouping | +| `test_schedd_query.py` | the schedd list (whitelist, identity filter, the location attributes `htcondor2.Schedd` needs), the query constraint (apptainer, namespace matching), failure handling, every config | +| `test_k8s_submit.py` | Job submission through the **real** `kubernetes` client (only the network send is intercepted), for every config | +| `test_logging.py` | log file and stdout mirroring | + +**Image test**, on every image before deploying it: + +```bash +tests/image.sh +``` + +It runs the unit tests in the image, checks the `provisioner` user's home has `.bash_profile` and that +the setup scripts' environment reaches `su -l provisioner`, then starts the provisioner with a **fake +service account** and requires it to reach the loop with no traceback and no deprecation warnings. +This makes real read-only queries to the OSPool. + +The config-driven tests use `tests/sample_osg_provisioner.conf`. Inside the +[nrp-provisioner](https://github.com/ashtongraves/nrp-provisioner) repo they also check every +namespace config under `k8s/`. + +Writing tests: don't put real job ads or logs in fixtures (they carry user and project names); write +synthetic ads with the same attribute names and expression forms. Test the client path you actually +use. Make smoke tests get past the expected failure point: an "expected" error once hid a real bug. + +## Background and pitfalls + +### Kubernetes + +- **The `kubernetes` client ≥ 37 validates request bodies client-side** (pydantic). Resource amounts + must be strings, and unknown fields are rejected. The API server used to accept numbers and silently + drop a `defaultMode` inside a secret volume item, which is why the worker token file has the default + mode (0644). Test submission through the real client, never a fake API object. +- **The ConfigMap is mounted with `subPath`**, so config changes need a restart. + +### HTCondor Python bindings + +- **HTCondor 25 ships only `htcondor2`/`classad2`.** 24 ships both; 23 only the classic `htcondor`. + The port meant `Schedd.xquery()` → `Schedd.query()` and `htcondor.AdTypes` → `htcondor2.AdType`. +- **`htcondor2.Schedd(ad)` reads `ad['CondorVersion']`.** Location ads need `MyAddress`, `AddressV1`, + `CondorVersion`, `CondorPlatform`, `Name` and `Machine`. A narrower collector projection makes every + schedd query raise `KeyError`. +- **`classad2` `eval()` stringifies the same as the classic module**, which the grouping depends on + (verified on 1,000 real job ads). + +### Building on an arm64 Mac + +Images build as amd64 under emulation: slow, and sometimes different from native (the `useradd` +case). `--pull` forces a full rebuild. macOS has no `timeout` command. A file in a mounted script +directory can shadow stdlib modules (a stray `ssl.py` broke `import urllib3`). diff --git a/osg-htc/nrp-provisioner/build-config.json b/osg-htc/nrp-provisioner/build-config.json new file mode 100644 index 00000000..524843a0 --- /dev/null +++ b/osg-htc/nrp-provisioner/build-config.json @@ -0,0 +1,7 @@ +{ + "standard_build": true, + "repo_build": false, + "base_os": ["el10"], + "osg_series": ["25"], + "base_repo": ["release"] + } diff --git a/osg-htc/nrp-provisioner/provisioner_info.py b/osg-htc/nrp-provisioner/provisioner_info.py new file mode 100644 index 00000000..ece030ba --- /dev/null +++ b/osg-htc/nrp-provisioner/provisioner_info.py @@ -0,0 +1,63 @@ +#!/bin/env python3 +# +# prp-osg-pool/provisioner +# +# BSD license, copyright Igor Sfiligoi 2022 +# +# Main entry point of the provisioner process +# + +import sys,os +import time +import configparser + +import src.provisioner_k8s as provisioner_k8s +import src.provisioner_logging as provisioner_logging +import src.provisioner_htcondor as provisioner_htcondor +import src.event_loop as event_loop + +def main(namespace): + fconfig = configparser.ConfigParser() + fconfig.read(('pod.conf','osg_provisioner.conf')) + kconfig = provisioner_k8s.ProvisionerK8SConfig(namespace) + cconfig = provisioner_htcondor.ProvisionerHTCConfig(namespace) + + # we will distinguish this class by these attrs + kconfig.additional_labels['osg-provisioner'] = 'wn' + kconfig.app_name = 'osg-wn' + cconfig.app_name = 'osg-wn' + + if 'k8s' in fconfig: + kconfig.parse(fconfig['k8s']) + else: + kconfig.parse(fconfig['DEFAULT']) + + hfconfig = fconfig['htcondor'] if ('htcondor' in fconfig) else fconfig['DEFAULT'] + cconfig.parse(hfconfig) + + log_obj = provisioner_logging.ProvisionerStdoutLogging(want_log_debug=True) + # TBD: Strong security + schedd_whitelist=hfconfig.get('schedd_whitelist_regexp','.*') + schedd_obj = provisioner_htcondor.ProvisionerSchedd(log_obj, {schedd_whitelist:'.*'}, cconfig) + collector_obj = provisioner_htcondor.ProvisionerCollector(log_obj, '.*', cconfig) + k8s_obj = provisioner_k8s.ProvisionerK8S(kconfig) + k8s_obj.authenticate() + + el = event_loop.ProvisionerEventLoop(log_obj, schedd_obj, collector_obj, k8s_obj, 1, 10000) + (schedd_clusters, k8s_clusters) = el.query_system() + all_clusters_set = set(schedd_clusters.keys())|set(k8s_clusters.keys()) + print("%-39s %9s %9s %9s %9s %9s %9s"%("Cluster (CPUs;MEM;DISK;;GPUs;;)","Idle Jobs","Wait Pods","Unmatched","Run Pods","Failed P","Unknown P")) + print("="*(39+1+9+1+9+1+9+1+9+1+9+1+9)) + ckeys=list(all_clusters_set) + ckeys.sort() + for ckey in ckeys: + sched_cnt=schedd_clusters[ckey].count_idle() if (ckey in schedd_clusters) else 0 + (waiting_cnt,unmatched_cnt,claimed_cnt,failed_cnt,unknown_cnt) = k8s_clusters[ckey].count_states() if (ckey in k8s_clusters) else (0,0,0,0,0) + if (sched_cnt+waiting_cnt+unmatched_cnt+claimed_cnt+failed_cnt+unknown_cnt)>0: + # do not report all zeros + print("%-39s %9i %9i %9i %9i %9i %9i"%(ckey, sched_cnt,waiting_cnt,unmatched_cnt,claimed_cnt,failed_cnt,unknown_cnt)) + +if __name__ == "__main__": + # execute only if run as a script + main(sys.argv[1] if len(sys.argv)>1 else os.environ['K8S_NAMESPACE']) + diff --git a/osg-htc/nrp-provisioner/provisioner_main.py b/osg-htc/nrp-provisioner/provisioner_main.py new file mode 100644 index 00000000..8fe62e90 --- /dev/null +++ b/osg-htc/nrp-provisioner/provisioner_main.py @@ -0,0 +1,78 @@ +# +# prp-osg-pool/provisioner +# +# BSD license, copyright Igor Sfiligoi 2022 +# +# Main entry point of the provisioner process +# + +import sys +import time +import configparser + +import src.provisioner_k8s as provisioner_k8s +import src.provisioner_logging as provisioner_logging +import src.provisioner_htcondor as provisioner_htcondor +import src.event_loop as event_loop +import src.provisioner_backoff as provisioner_backoff + +def main(log_fname, namespace, cvmfs_mounts): + fconfig = configparser.ConfigParser() + fconfig.read(('pod.conf','osg_provisioner.conf')) + kconfig = provisioner_k8s.ProvisionerK8SConfig(namespace) + cconfig = provisioner_htcondor.ProvisionerHTCConfig(namespace) + + # we will distinguish this class by these attrs + kconfig.additional_labels['osg-provisioner'] = 'wn' + kconfig.app_name = 'osg-wn' + cconfig.app_name = 'osg-wn' + + kfconfig = fconfig['k8s'] if ('k8s' in fconfig) else fconfig['DEFAULT'] + kconfig.parse(kfconfig) + + hfconfig = fconfig['htcondor'] if ('htcondor' in fconfig) else fconfig['DEFAULT'] + cconfig.parse(hfconfig) + + kconfig.base_pvc_volumes["cvmfs"] = "#/cvmfs" + cvmfsenvs=[] + + # we always need the osg-config and oasis + for c in (['config-osg','oasis'] + cvmfs_mounts): + ext='opensciencegrid.org' if c!='stash' else 'osgstorage.org' + cvmfsenvs.append( "%s.%s"%(c,ext) ) + + kconfig.additional_envs['CVMFS_REPOS'] = ",".join(cvmfsenvs) + + log_obj = provisioner_logging.ProvisionerFileLogging(log_fname, want_log_debug=True, also_stdout=True) + # TBD: Strong security + schedd_whitelist=hfconfig.get('schedd_whitelist_regexp','.*') + schedd_obj = provisioner_htcondor.ProvisionerSchedd(log_obj, {schedd_whitelist:'.*'}, cconfig) + collector_obj = provisioner_htcondor.ProvisionerCollector(log_obj, '.*', cconfig) + k8s_obj = provisioner_k8s.ProvisionerK8S(kconfig) + k8s_obj.authenticate() + + max_pods_per_cluster = int(kfconfig.get('max_pods_per_cluster','20')) + max_submit_pods_per_cluster = int(kfconfig.get('max_submit_pods_per_cluster','1000')) + # clusters with no claimed pods get only this many, until one is claimed + trial_pods = int(kfconfig.get('trial_pods','2')) + sleep_time = int(fconfig['DEFAULT'].get('sleep_time','120')) + + # stop provisioning for a while for clusters whose pods keep exiting unclaimed + backoff_obj = provisioner_backoff.ProvisionerBackoff(int(kfconfig.get('backoff_unclaimed_exits','2')), + int(kfconfig.get('backoff_time','3600')), + int(kfconfig.get('backoff_max_time','28800'))) + + el = event_loop.ProvisionerEventLoop(log_obj, schedd_obj, collector_obj, k8s_obj, max_pods_per_cluster, max_submit_pods_per_cluster, backoff_obj, trial_pods) + while True: + log_obj.log_debug("[Main] Iteration started, schedd whitelist='%s'"%schedd_whitelist) + try: + el.one_iteration() + except: + log_obj.log_debug("[Main] Exception in one_iteration") + log_obj.sync() + time.sleep(sleep_time) + +if __name__ == "__main__": + # execute only if run as a script + main(sys.argv[1], sys.argv[2], sys.argv[3].split(",")) + diff --git a/osg-htc/nrp-provisioner/provisioner_startup.sh b/osg-htc/nrp-provisioner/provisioner_startup.sh new file mode 100755 index 00000000..1ac6c626 --- /dev/null +++ b/osg-htc/nrp-provisioner/provisioner_startup.sh @@ -0,0 +1,41 @@ +#!/bin/bash + +# basic setup +/opt/provisioner/setup_k8s_creds.sh +/opt/provisioner/setup_htcondor_creds.sh + +mkdir -p /var/log/provisioner/logs +chown provisioner:provisioner /var/log/provisioner/logs + +if [ "x${CONDOR_HOST}" == "x" ]; then + echo "Missing CONDOR_HOST" + sleep 15 + exit 1 +fi + +if [ "x${K8S_NAMESPACE}" == "x" ]; then + echo "Missing K8S_NAMESPACE" + sleep 15 + exit 1 +fi + +if [ "x${CVMFS_MOUNTS}" == "x" ]; then + echo "Missing CVMFS_MOUNTS" + sleep 15 + exit 1 +fi + +echo "CONDOR_HOST=${CONDOR_HOST}" > /etc/condor/config.d/02_condor_host.config + +echo "[DEFAULT]" > /home/provisioner/pod.conf +echo "condor_host=${CONDOR_HOST}" >> /home/provisioner/pod.conf +echo "namespace=${K8S_NAMESPACE}" >> /home/provisioner/pod.conf + +trap 'echo signal received!; kill $(jobs -p); wait' SIGINT SIGTERM + +echo "`date` Starting provisioner_main.py" +su -l provisioner -c "cd /home/provisioner && python3 provisioner_main.py /var/log/provisioner/logs/provisioner.log ${K8S_NAMESPACE} ${CVMFS_MOUNTS}" & +wait +rc=$? +echo "`date` End of provisioner_main.py, rc=${rc}" + diff --git a/osg-htc/nrp-provisioner/setup_htcondor_creds.sh b/osg-htc/nrp-provisioner/setup_htcondor_creds.sh new file mode 100755 index 00000000..41cd73c2 --- /dev/null +++ b/osg-htc/nrp-provisioner/setup_htcondor_creds.sh @@ -0,0 +1,10 @@ +#!/bin/bash +su provisioner -c "mkdir -p /home/provisioner/.condor/tokens.d && chmod -R go-rwx /home/provisioner/.condor" +cp /etc/condor/tokens.d/* /home/provisioner/.condor/tokens.d/ && chown -R provisioner /home/provisioner/.condor + +if [ "x${HTCONDOR_QUERY_INSECURE}" == "xyes" ]; then + su provisioner -c "echo export _CONDOR_SEC_CLIENT_AUTHENTICATION=OPTIONAL >> /home/provisioner/.bashrc" + su provisioner -c "echo export _CONDOR_SEC_CLIENT_INTEGRITY=OPTIONAL >> /home/provisioner/.bashrc" + su provisioner -c "echo export _CONDOR_SEC_CLIENT_ENCRYPTION=OPTIONAL >> /home/provisioner/.bashrc" +fi + diff --git a/osg-htc/nrp-provisioner/setup_k8s_creds.sh b/osg-htc/nrp-provisioner/setup_k8s_creds.sh new file mode 100755 index 00000000..2f9f86d1 --- /dev/null +++ b/osg-htc/nrp-provisioner/setup_k8s_creds.sh @@ -0,0 +1,8 @@ +#!/bin/bash +su provisioner -c "touch /home/provisioner/.bashrc" + +su provisioner -c "echo export KUBERNETES_SERVICE_PORT=${KUBERNETES_SERVICE_PORT} >> /home/provisioner/.bashrc" +su provisioner -c "echo export KUBERNETES_SERVICE_HOST=${KUBERNETES_SERVICE_HOST} >> /home/provisioner/.bashrc" + +#also add the other kubernetes-related env variables +su provisioner -c "echo export K8S_NAMESPACE=${K8S_NAMESPACE} >> /home/provisioner/.bashrc" diff --git a/osg-htc/nrp-provisioner/src/__init__.py b/osg-htc/nrp-provisioner/src/__init__.py new file mode 100644 index 00000000..769ad1f1 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/__init__.py @@ -0,0 +1 @@ +# provisioner library (was prp_provisioner) diff --git a/osg-htc/nrp-provisioner/src/event_loop.py b/osg-htc/nrp-provisioner/src/event_loop.py new file mode 100644 index 00000000..cdc1f165 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/event_loop.py @@ -0,0 +1,148 @@ +# +# prp-htcondor-portal/provisioner +# +# BSD license, copyright Igor Sfiligoi 2021 +# +# Implement the event loop +# + +from . import provisioner_clustering + +class ProvisionerEventLoop: + def __init__(self, log_obj, schedd_obj, collector_obj, k8s_obj, max_pods_per_cluster, max_submit_pods_per_cluster, backoff_obj=None, trial_pods=0): + self.log_obj = log_obj + self.schedd = schedd_obj + self.collector = collector_obj + self.k8s = k8s_obj + self.max_pods_per_cluster = max_pods_per_cluster + self.max_submit_pods_per_cluster = max_submit_pods_per_cluster + self.backoff = backoff_obj + self.trial_pods = trial_pods # 0 means no limit + + def query_system(self): + schedd_attrs = provisioner_clustering.ProvisionerClusteringAttributes().get_schedd_attributes() + try: + schedd_jobs = self.schedd.query_idle(projection=schedd_attrs) + startd_pods = self.collector.query(projection=['JobStarts']) + except: + self.log_obj.log_error("[ProvisionerEventQuery] Failed to query HTCondor") + self.log_obj.sync() + raise + del schedd_attrs + + try: + k8s_pods = self.k8s.query() + except: + self.log_obj.log_error("[ProvisionerEventQuery] Failed to query k8s") + self.log_obj.sync() + raise + + clustering = provisioner_clustering.ProvisionerClustering() + schedd_clusters = clustering.cluster_schedd_jobs(schedd_jobs) + k8s_clusters = clustering.cluster_k8s_pods(k8s_pods, startd_pods) + + return (schedd_clusters, k8s_clusters) + + def one_iteration(self): + try: + (schedd_clusters, k8s_clusters) = self.query_system() + except: + self.log_obj.log_error("[ProvisionerEventLoop] Failed to query") + self.log_obj.sync() + return + + if self.backoff!=None: + try: + for msg in self.backoff.update(k8s_clusters): + self.log_obj.log_info("[ProvisionerEventLoop] %s"%msg) + except: + self.log_obj.log_error("[ProvisionerEventLoop] Failed to update the provisioning back-off") + + all_clusters_set = set(schedd_clusters.keys())|set(k8s_clusters.keys()) + for ckey in all_clusters_set: + try: + k8s_cluster = k8s_clusters[ckey] if ckey in k8s_clusters else None + if ckey in schedd_clusters: + self._provision_cluster(ckey, schedd_clusters[ckey], k8s_cluster ) + else: + # we have k8s cluster, but no condor jobs + self._report_cluster(ckey, k8s_cluster) + except: + self.log_obj.log_debug("[ProvisionerEventLoop] Exception in cluster '%s'"%ckey) + + # explicit cleanup to avoid accidental reuse + del all_clusters_set + del schedd_clusters + del k8s_clusters + + self.log_obj.sync() + + + # INTERNAL + def _check_and_report_cluster(self, cluster_id, k8s_cluster, n_jobs_idle, min_pods): + "Check if we have enough pods already, and also report" + n_pods_statearr=k8s_cluster.count_states() if k8s_cluster!=None else (0,0,0,0,0) + n_pods_waiting=n_pods_statearr[0] + n_pods_unmatched=n_pods_statearr[1] + n_pods_claimed=n_pods_statearr[2] + n_pods_failed=n_pods_statearr[3] + n_pods_unclaimed = n_pods_waiting+n_pods_unmatched + n_pods_total = n_pods_unclaimed+n_pods_claimed+n_pods_failed + + if n_pods_total>=self.max_submit_pods_per_cluster: + min_pods = 0 + + if (n_jobs_idle+n_pods_unclaimed+min_pods+n_pods_waiting+n_pods_unmatched+n_pods_claimed+n_pods_failed)>0: + # do not report all zeros (e.g. only completed pods in the system) + self.log_obj.log_debug("[ProvisionerEventLoop] Cluster '%s' n_jobs_idle %i n_pods_unclaimed %i min_pods %i (pods wait %i unmatched %i claimed %i failed %i max %i)"% + (cluster_id, n_jobs_idle, n_pods_unclaimed, min_pods, n_pods_waiting, n_pods_unmatched, n_pods_claimed, n_pods_failed, self.max_submit_pods_per_cluster)) + + return (min_pods,n_pods_unclaimed) + + def _report_cluster(self, cluster_id, k8s_cluster): + self._check_and_report_cluster(cluster_id, k8s_cluster, 0, 0) + + def _provision_cluster(self, cluster_id, schedd_cluster, k8s_cluster): + "Check if we have enough k8s clusters. Submit more if needed" + n_jobs_idle = schedd_cluster.count_idle() + if n_jobs_idle==0: + self.log_obj.log_debug("[ProvisionerEventLoop] Cluster '%s' n_jobs_idle==0 found!"%cluster_id) + return # should never get in here, but just in case (nothing to do, we are done) + + # assume some latency and pod reuse + min_pods = 1 + int(n_jobs_idle/4) + if min_pods>20: + # when we have a lot of jobs, slow futher + min_pods = 20 + int((min_pods-20)/4) + + if min_pods>self.max_pods_per_cluster: + min_pods = self.max_pods_per_cluster + + if self.trial_pods>0 and min_pods>self.trial_pods: + n_pods_claimed = k8s_cluster.count_states()[2] if k8s_cluster!=None else 0 + if n_pods_claimed==0: + # none of our pods for this cluster is running a job, + # so try only a few until we know the jobs can match them + min_pods = self.trial_pods + + (min_pods,n_pods_unclaimed) = self._check_and_report_cluster(cluster_id, k8s_cluster, n_jobs_idle, min_pods) + + suspended_for = self.backoff.suspended_for(cluster_id) if self.backoff!=None else 0 + + if n_pods_unclaimed>=min_pods: + pass # we have enough pods, do nothing for now + # we may want to do some sanity checks here, eventually + elif suspended_for>0: + (n_exits,n_suspensions) = self.backoff.get_state(cluster_id) + self.log_obj.log_debug("[ProvisionerEventLoop] Cluster '%s' provisioning suspended for %i more seconds (%i unclaimed exits, suspension %i)"% + (cluster_id, suspended_for, n_exits, n_suspensions)) + else: + try: + job_name = self.k8s.submit(schedd_cluster.get_attr_dict(), min_pods-n_pods_unclaimed) + self.log_obj.log_info("[ProvisionerEventLoop] Cluster '%s' Submitted %i pods as job %s"% + (cluster_id,min_pods-n_pods_unclaimed, job_name)) + except: + self.log_obj.log_error("[ProvisionerEventLoop] Cluster '%s' Failed to submit %i pods"% + (cluster_id,min_pods-n_pods_unclaimed)) + + return diff --git a/osg-htc/nrp-provisioner/src/provisioner_backoff.py b/osg-htc/nrp-provisioner/src/provisioner_backoff.py new file mode 100644 index 00000000..dc6d93fa --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_backoff.py @@ -0,0 +1,118 @@ +# +# nrp-provisioner +# +# BSD license +# +# Stop provisioning for clusters whose pods keep exiting without ever being claimed +# +# A pod that reaches the Succeeded phase without its startd ever being claimed +# most likely idled out because the jobs it was provisioned for cannot match it +# (e.g. pinned to another site, or targeting an annex). +# Failed or vanished pods are ignored, since preemption and node problems +# say nothing about the jobs. +# + +import time + +class ProvisionerBackoff: + def __init__(self, max_unclaimed_exits=2, backoff_time=3600, max_backoff_time=8*3600): + """ + Arguments: + max_unclaimed_exits: int + Suspend a cluster after this many unclaimed exits without a claim in between (0 disables) + backoff_time: int + Length of the first suspension, in seconds; doubled on each repeat + max_backoff_time: int + Upper limit for the suspension length, in seconds + """ + self.max_unclaimed_exits = max_unclaimed_exits + self.backoff_time = backoff_time + self.max_backoff_time = max_backoff_time + self.pods = {} # pod name -> {'cluster','started','claimed'} + self.clusters = {} # cluster id -> {'exits','level','until'} + + def update(self, k8s_clusters, now=None): + """Process the latest pod states; returns a list of messages worth logging""" + msgs = [] + if self.max_unclaimed_exits<1: + return msgs + if now==None: + now = time.time() + + seen = set() + for cluster_id in k8s_clusters.keys(): + for (pod_el, startd_el) in k8s_clusters[cluster_id].elements: + name = pod_el['Name'] + phase = "%s"%pod_el['Phase'] + seen.add(name) + rec = self.pods.get(name) + if phase in ("Pending","Running"): + if rec==None: + rec = {'cluster': cluster_id, 'started': False, 'claimed': False} + self.pods[name] = rec + if startd_el!=None: + rec['started'] = True + if self._is_claimed(startd_el): + rec['claimed'] = True + msgs += self._reset(cluster_id) + elif phase=="Succeeded": + if rec!=None: + if rec['started'] and not rec['claimed']: + msgs += self._unclaimed_exit(rec['cluster'], now) + del self.pods[name] + elif rec!=None: + # Failed or unknown, not the fault of the jobs + del self.pods[name] + + for name in list(self.pods.keys()): + if name not in seen: + # vanished without us seeing the end, ignore + del self.pods[name] + return msgs + + def suspended_for(self, cluster_id, now=None): + """Return the number of seconds the cluster is still suspended for (0 if not suspended)""" + if self.max_unclaimed_exits<1: + return 0 + if now==None: + now = time.time() + el = self.clusters.get(cluster_id) + if (el==None) or (el['until']<=now): + return 0 + return int(el['until']-now) + + def get_state(self, cluster_id): + """Return (unclaimed exits since the last claim, number of suspensions so far)""" + el = self.clusters.get(cluster_id) + if el==None: + return (0,0) + return (el['exits'],el['level']) + + # INTERNAL + def _is_claimed(self, startd_el): + if "%s"%startd_el.get('State')=="Claimed": + return True + try: + return int("%s"%startd_el.get('JobStarts'))>0 + except ValueError: + return False + + def _reset(self, cluster_id): + el = self.clusters.get(cluster_id) + if el==None: + return [] + del self.clusters[cluster_id] + if el['level']>0: + return ["Cluster '%s' had a pod claimed, provisioning back-off cleared"%cluster_id] + return [] + + def _unclaimed_exit(self, cluster_id, now): + el = self.clusters.setdefault(cluster_id, {'exits': 0, 'level': 0, 'until': 0}) + el['exits'] += 1 + if (el['exits']>=self.max_unclaimed_exits) and (el['until']<=now): + # once over the limit, each further unclaimed exit outside a suspension starts a longer one + duration = min(self.backoff_time*(2**el['level']), self.max_backoff_time) + el['until'] = now+duration + el['level'] += 1 + return ["Cluster '%s' had %i pods exit unclaimed, suspending provisioning for %i seconds"%(cluster_id, el['exits'], duration)] + return [] diff --git a/osg-htc/nrp-provisioner/src/provisioner_clustering.py b/osg-htc/nrp-provisioner/src/provisioner_clustering.py new file mode 100644 index 00000000..45cd1ef5 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_clustering.py @@ -0,0 +1,240 @@ +# +# prp-htcondor-portal/provisioner +# +# BSD license, copyright Igor Sfiligoi 2021 +# +# Implement the clusterinng +# + +from collections import OrderedDict +import copy +import re +import datetime + +# +# GPUTypes is a comma-separated list of GPU names +# Note: The GPUType is just a special case of a label +# with the hardcoded key of gpu-type +# +# DiskVolumes are an encoded comma-separated list +# with each element a colon separated pair +# PVC_NAME:MOUNT_PATH +# +# Labels are an encoded comma-separated list +# with each element a colon spearated pair +# LABEL_KEY:LABEL_VALUE +# + +class ProvisionerClusteringAttributes: + + def __init__(self): + self.attributes=OrderedDict() + self.attributes['CPUs']=1 + self.attributes['Memory']=4096 + self.attributes['Disk']=8000000 + self.attributes['DiskVolumes']='' + self.attributes['GPUs']=0 + self.attributes['GPUTypes']='' + self.attributes['Labels']='' + + def expand_schedd_attr(self, attr): + return "Request%s"%attr + + def get_schedd_attributes(self): + attrs=[] + for k in self.attributes.keys(): + attrs.append(self.expand_schedd_attr(k)) + return attrs + + # Note: The startd attributes have no prefix + + def expand_k8s_attr(self, attr): + return "Pod%s"%attr + + def get_k8s_attributes(self): + attrs=[] + for k in self.attributes.keys(): + attrs.append(self.expand_k8s_attr(k)) + return attrs + + +class ProvisionerCluster: + def __init__(self, key, attr_vals): + self.key = key + self.attr_vals = attr_vals + self.elements = [] + + def append(self, el): + self.elements.append(el) + + def append_list(self, ellist): + for el in ellist: + self.elements.append(el) + + def get_attr_dict(self): + p=ProvisionerClusteringAttributes() + attrs={} + i=0 + for k in p.attributes.keys(): + attrs[k] = self.attr_vals[i] + i = i+1 + return attrs + + +class ProvisionerScheddCluster(ProvisionerCluster): + def __init__(self, key, attr_vals, schedd_attrs): + ProvisionerCluster.__init__(self, key, attr_vals) + self.schedd_attrs = schedd_attrs + + def count_idle(self): + cnt = 0 + for job in self.elements: + status="%s"%job['JobStatus'] + if status=="1": + cnt+=1 + return cnt + +class ProvisionerK8SCluster(ProvisionerCluster): + def __init__(self, key, attr_vals, pod_attrs, max_wait_onerror=1000): + ProvisionerCluster.__init__(self, key, attr_vals) + self.pod_attrs = pod_attrs + self.max_wait_onerror=max_wait_onerror # in seconds + + def count_states(self, max_wait_onerror=None): + "Returns (waiting,unmatched,claimed,failed,unknown) counts" + waiting_cnt = 0 + unmatched_cnt = 0 + claimed_cnt = 0 + failed_cnt = 0 + unknown_cnt = 0 + + if max_wait_onerror==None: + max_wait_onerror=self.max_wait_onerror + + err_deadline = datetime.datetime.now(datetime.timezone.utc)-datetime.timedelta(seconds=max_wait_onerror) + for el in self.elements: + pod_el = el[0] + phase="%s"%pod_el['Phase'] + if phase=="Running": + startd_el = el[1] + if startd_el!=None: + state = "%s"%startd_el['State'] + else: + state = "None" + # we will count all Running pods that are not yet claimed + if state=="None": + waiting_cnt+=1 + elif state!="Claimed": + unmatched_cnt+=1 + else: + claimed_cnt+=1 + elif phase=="Pending": + # we can safely count these as waiting at all times + waiting_cnt+=1 + elif (phase=="Terminating"): + # we can safely ignore these + pass + elif (phase=="Succeeded"): + # we can safely ignore these + pass + elif (phase=="Failed"): + if pod_el['HasContainers']: + failed_cnt+=1 + else: + unknown_cnt+=1 + else: + unknown_cnt+=1 + # All other states should be transitory + # if not, they are problematic and should be ignored + err_time = None + try: + err_time=pod_el['StartTime'] + except: + pass + if (err_time==None): + # no good estimate, just count as waiting + waiting_cnt+=1 + unknown_cnt-=1 + elif err_time>err_deadline: + # assume it is transitory, count as waiting + waiting_cnt+=1 + unknown_cnt-=1 + #else, this is unlikely to recover, count as unknown + return (waiting_cnt,unmatched_cnt,claimed_cnt,failed_cnt,unknown_cnt) + + def count_unclaimed(self, max_wait_onerror=None): + (waiting_cnt,unmatched_cnt,claimed_cnt,failed_cnt,unknown_cnt)=self.count_states(max_wait_onerror) + # "logically unclaimed" = waiting+"running unmatched" + return waiting_cnt+unmatched_cnt + +class ProvisionerClustering: + def __init__(self): + self.attrs=ProvisionerClusteringAttributes() + + def cluster_schedd_jobs(self, schedd_jobs): + clusters={} + for job in schedd_jobs: + job_attrs=[] + key_attrs={} + for k in self.attrs.attributes.keys(): + jobk = self.attrs.expand_schedd_attr(k) + if jobk in job.keys(): + val = job[jobk] + else: + val = self.attrs.attributes[k] + + sval="%s"%val + if sval=="Undefined": + val = self.attrs.attributes[k] + sval="%s"%val + + job_attrs.append(sval) + key_attrs[jobk]=val + del sval + del val + job_key=";".join(job_attrs) + if job_key not in clusters: + clusters[job_key] = ProvisionerScheddCluster(job_key, job_attrs, key_attrs) + clusters[job_key].append(job) + # cleanup to avoid accidental reuse + del job_attrs + del key_attrs + + return clusters + + + def cluster_k8s_pods(self, k8s_pods, startd_ads): + startd_dict={} + for ad in startd_ads: + k=ad["K8SPodName"] + startd_dict[k]=ad + del k + + clusters={} + for pod in k8s_pods: + if pod['Name'] in startd_dict: + pod_ad = startd_dict[pod['Name'] ] + else: + pod_ad = None + pod_attrs=[] + key_attrs={} + for k in self.attrs.attributes.keys(): + podk = self.attrs.expand_k8s_attr(k) + if podk in pod.keys(): + val = pod[podk] + else: + val = self.attrs.attributes[k] + pod_attrs.append("%s"%val) + key_attrs[podk]=val + pod_key=";".join(pod_attrs) + if pod_key not in clusters: + clusters[pod_key] = ProvisionerK8SCluster(pod_key, pod_attrs, key_attrs) + clusters[pod_key].append( (pod,pod_ad) ) + # cleanup to avoid accidental reuse + del pod_attrs + del key_attrs + del pod_ad + + return clusters + + diff --git a/osg-htc/nrp-provisioner/src/provisioner_config_parser.py b/osg-htc/nrp-provisioner/src/provisioner_config_parser.py new file mode 100644 index 00000000..3f0d43d8 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_config_parser.py @@ -0,0 +1,48 @@ +# +# prp-htcondor-portal/provisioner +# +# BSD license, copyright Igor Sfiligoi@UCSD 2022 +# +# Implement the config parser helper function +# + +import copy + +def encode_el(v): + # need a way to propagate : and , + while v.find(":")>=0: + v=v.replace(":",":") + while v.find(",")>=0: + v=v.replace(",",",") + while v.find("&")>=0: + v=v.replace("&","&") + return v + +def parse_list(list_str): + out=[] + for v in list_str.split(','): + out.append(encode_el(v)) + return out + +def parse_dict(dict_str): + els = dict_str.split(",") + out = {} + for el in els: + k,v = el.split(":") + out[k] = encode_el(v) + return out + +def update_parse(var, field, ftype, + fields, dict): + """Update var if filed in both fields and dice, and parse according to ftype""" + if (field in fields) and (field in dict): + rval=dict[field] + if (ftype=="str"): + var = copy.deepcopy(rval) + elif (ftype=="int"): + var = int(rval) + elif (ftype=="list"): + var = parse_list(rval) + elif (ftype=="dict"): + var = parse_dict(rval) + return var diff --git a/osg-htc/nrp-provisioner/src/provisioner_htcondor.py b/osg-htc/nrp-provisioner/src/provisioner_htcondor.py new file mode 100644 index 00000000..6ff16061 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_htcondor.py @@ -0,0 +1,236 @@ +# +# prp-htcondor-portal/provisioner +# +# BSD license, copyright Igor Sfiligoi 2021 +# +# Implement the htcondor interface +# + +import copy +import re +import htcondor2 + +from . import provisioner_config_parser + +ProvisionerHTCConfigFields = ('namespace','condor_host', + 'app_name','k8s_domain', + 'force_k8s_namespace_matching', + 'additional_requirements', + 'apptainer') + +class ProvisionerHTCConfig: + """Config file for HTCOndor provisioning classes""" + + def __init__(self, namespace, + condor_host="prp-cm-htcondor.htcondor-portal.svc.cluster.local", + app_name = 'prp-wn', + k8s_domain='optiputer.net', + force_k8s_namespace_matching = "yes", + additional_requirements = "", + apptainer = "yes"): + self.namespace = copy.deepcopy(namespace) + self.condor_host = copy.deepcopy(condor_host) + self.app_name = copy.deepcopy(app_name) + self.k8s_domain = copy.deepcopy(k8s_domain) + self.force_k8s_namespace_matching = copy.deepcopy(force_k8s_namespace_matching) + self.additional_requirements = copy.deepcopy(additional_requirements) + self.apptainer = copy.deepcopy(apptainer) + + def parse(self, + dict, + fields=ProvisionerHTCConfigFields): + """Parse the valuies from a dictionary""" + self.namespace = provisioner_config_parser.update_parse(self.namespace, 'namespace', 'str', fields, dict) + self.condor_host = provisioner_config_parser.update_parse(self.condor_host, 'condor_host', 'str', fields, dict) + self.k8s_domain = provisioner_config_parser.update_parse(self.k8s_domain, 'k8s_domain', 'str', fields, dict) + self.app_name = provisioner_config_parser.update_parse(self.app_name, 'app_name', 'str', fields, dict) + self.force_k8s_namespace_matching = provisioner_config_parser.update_parse(self.force_k8s_namespace_matching, 'force_k8s_namespace_matching', 'str', fields, dict) + self.additional_requirements = provisioner_config_parser.update_parse(self.additional_requirements, 'additional_requirements', 'str', fields, dict) + self.apptainer = provisioner_config_parser.update_parse(self.apptainer, 'apptainer', 'str', fields, dict) + +class ProvisionerSchedd: + """HTCondor schedd interface""" + + def __init__(self, log_obj, trusted_schedds, config): + """ + Arguments: + log_obj: object + Logging object + namespace: string + Monitored namespace + trusted_schedds: dictionary, NameRegexp:AuthenticatedIdentityRegexp + Set of schedds to query. Both name and AuthenticatedIdentity are regexp. + """ + self.log_obj = log_obj + self.trusted_schedds = copy.deepcopy(trusted_schedds) + self.namespace = copy.deepcopy(config.namespace) + self.force_k8s_namespace_matching = copy.deepcopy(config.force_k8s_namespace_matching) + self.additional_requirements = copy.deepcopy(config.additional_requirements) + self.apptainer = copy.deepcopy(config.apptainer) + + def query_idle(self, projection=[]): + """Return the list of idle jobs for my namespace""" + return self.query(job_status=1, projection=projection) + + + def query(self, job_status, projection=[]): + """Return the list of jobs for my namespace""" + + full_projection=['ClusterId','ProcId','JobStatus','RequestK8SNamespace']+projection + query_str='(JobStatus=?=%i)'%job_status + if self.force_k8s_namespace_matching!="no": + query_str += ' && regexp(RequestK8SNamespace,"%s")'%self.namespace + if self.additional_requirements != "": + query_str += ' && (%s)'%self.additional_requirements + # without apptainer, the WNs cannot run jobs that ask for a container image + if self.apptainer=="no": + query_str += ' && isUndefined(SingularityImage)' + else: + query_str += ' && (!isUndefined(SingularityImage))' + + jobs=[] + + sobjs=self._get_schedd_objs() + ok_schedds=0 + bad_schedds=0 + for sclassad in sobjs: + s=htcondor2.Schedd(sclassad) + sname=sclassad['Name'] + have_myjobs=False + try: + myjobs=s.query(query_str, full_projection) + have_myjobs=True + except: + self.log_obj.log_debug("[ProvisionerSchedd] Failed to query HTCondor schedd '%s'"%sname) + bad_schedds = bad_schedds + 1 + + if have_myjobs: + ok_schedds = ok_schedds + 1 + self._append_jobs(sname, jobs, myjobs) + + if (bad_schedds>0) and (ok_schedds==0): + # only raise here if all the schedds have failed + raise Exception("[ProvisionerSchedd] All HTCondor schedds failed") + + return jobs + + + # INTERNAL + def _append_jobs(self, schedd_name, jobs, myjobs): + """jobs is a list and will be updated in-place""" + minvals={'RequestMemory':(4096,1024),'RequestDisk':(8000000,4000000)} + for job in myjobs: + jobattrs={'ScheddName':schedd_name} + for k in job.keys(): + if k in minvals.keys(): + # the default RequestMemory and RequestDisk in condor is dynamic + # and the initial value (after eval) is way too low + # Treat very low values as undefines + val = int(job.eval(k)) + mval = minvals[k][0] + if val>=mval: + rval = minvals[k][1] + val = ((val + rval-1)//rval)*rval # round up + jobattrs[k]="%s"%val + #else pretend it is not there + else: + # convert all values to strings, for easier management + # also expand all expresstions + jobattrs[k]="%s"%job.eval(k) + jobs.append(jobattrs) + return + + def _get_schedd_objs(self): + sobjs=[] + c = htcondor2.Collector() + try: + # htcondor2.Schedd() also needs CondorVersion and CondorPlatform from the location ad + slist=c.query(ad_type=htcondor2.AdType.Schedd,projection=['Name','AuthenticatedIdentity','MyAddress','AddressV1','Machine','CondorVersion','CondorPlatform']) + except: + self.log_obj.log_debug("[ProvisionerSchedd] Failed to retrieve HTCondor schedd list") + raise + + for s in slist: + try: + sname=s['Name'] + sauthid=s['AuthenticatedIdentity'] + except: + # if I cannot find all, it is invalid + continue + if not self._is_valid_schedd(sname,sauthid): + continue # ignore untrusted schedds + sobjs.append(s) + # cleaup to avoid accidental reuse + del sname + del sauthid + return sobjs + + def _is_valid_schedd(self, schedd_name, schedd_authid): + found = False + for nameregex in self.trusted_schedds.keys(): + if re.fullmatch(nameregex,schedd_name): + # found a valid schedd name + authregex=self.trusted_schedds[nameregex] + if re.fullmatch(authregex,schedd_authid): + #it also matches the identity + found = True + break # found, we are done + # ignore all others + return found + + +class ProvisionerCollector: + """HTCondor Collector/startd interface""" + + def __init__(self, log_obj, startd_identity, config): + """ + Arguments: + log_obj: object + Logging object + namespace: string + Monitored namespace + startd_identity: string + AuthenticatedIdentity Regexp used as a whitelist + """ + self.log_obj = log_obj + self.startd_identity = copy.deepcopy(startd_identity) + self.namespace = copy.deepcopy(config.namespace) + self.k8s_domain = copy.deepcopy(config.k8s_domain) + self.app_name = copy.deepcopy(config.app_name) + + def query(self, projection=[]): + """Return the list of startds for my namespace""" + + full_projection=['Name','AuthenticatedIdentity','State','Activity','K8SPodName','K8SNamespace', 'K8SDomain', 'K8SProvisionerType', 'K8SProvisionerName']+projection + startds=[] + + c = htcondor2.Collector() + try: + slist=c.query(ad_type=htcondor2.AdType.Startd,projection=full_projection, + constraint='(K8SProvisionerType=?="PRPHTCondorProvisioner")&&(K8SProvisionerName=?="%s")&&(K8SNamespace=?="%s")&&(K8SDomain=?="%s")'%(self.app_name,self.namespace,self.k8s_domain)) + except: + self.log_obj.log_debug("[ProvisionerCollector] Failed to retrieve HTCondor startd list") + raise + + for s in slist: + try: + sname=s['Name'] + sauthid=s['AuthenticatedIdentity'] + except: + # if I cannot find all, it is invalid + continue + if not re.fullmatch(self.startd_identity,sauthid): + # not trusted, ignore + continue + adattrs={} + for k in s.keys(): + # convert all values to strings, for easier management + # also eval any expressions + adattrs[k]="%s"%s.eval(k) + startds.append(adattrs) + # cleaup to avoid accidental reuse + del sname + del sauthid + + return startds + diff --git a/osg-htc/nrp-provisioner/src/provisioner_k8s.py b/osg-htc/nrp-provisioner/src/provisioner_k8s.py new file mode 100644 index 00000000..85579eb6 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_k8s.py @@ -0,0 +1,558 @@ +# +# prp-htcondor-portal/provisioner +# +# BSD license, copyright Igor Sfiligoi@UCSD 2021 +# +# Implement the Kubernetes interface +# + +import copy +import re +import kubernetes +import time + +from . import provisioner_config_parser + +ProvisionerK8SConfigFields = ('namespace','condor_host', + 'k8s_image','k8s_image_pull_policy', + 'priority_class','priority_class_cpu','priority_class_gpu', + 'service_account_name', + 'tolerations_list', 'node_selectors_dict', 'node_affinity_dict', + 'labels_dict', 'annotations_dict', 'envs_dict', + 'pvc_volumes_dict', 'config_volumes_dict', + 'app_name','k8s_job_ttl','k8s_domain', + 'force_k8s_namespace_matching', + 'request_fuse','storage_mbs','resources_dict', + 'additional_requirements', + 'use_large_limit', + 'apptainer') + +class ProvisionerK8SConfig: + """Config fie for ProvisionerK8S""" + + def __init__(self, namespace, + condor_host="prp-cm-htcondor.htcondor-portal.svc.cluster.local", + k8s_image='sfiligoi/prp-portal-wn', + k8s_image_pull_policy='Always', + priority_class = None, + priority_class_cpu = None, + priority_class_gpu = None, + base_tolerations = [], + base_pvc_volumes = {}, + additional_labels = {}, + additional_envs = {}, + additional_volumes = {}, + additional_tolerations = [], + additional_node_selectors = {}, + additional_annotations = {}, + app_name = 'prp-wn', + k8s_job_ttl = 24*3600, # clean after 1 day + k8s_domain = 'optiputer.net', + force_k8s_namespace_matching = "yes", + additional_requirements = "", + node_affinity_dict = {}): + """ + Arguments: + namespace: string + Monitored namespace + condor_host: string (Optional) + DNS address of the HTCOndor collector + k8s_image: string (Optional) + WN Container image to use in the pod + k8s_image_pull_policy: string (Optional) + WN Container image pull policy + priority_class: string (Optional) + priorityClassName to associate with the pod + priority_class_cpu: string (Optional) + priorityClassName to associate with CPU pod + priority_class_cpu: string (Optional) + priorityClassName to associate with GPU pod + base_tolerations: list of strings (Optional) + Tolerations of the form NoSchedule/Exists to add to the container + base_pvc_volumes: list of strings (Optional) + PersistentVolumeClaims to add to the container, vvalues is mountpoint + additional_labels: dictionary of strings (Optional) + Labels to attach to the pod + additional_envs: dictionary of strings (Optional) + Environment values to add to the container + additional_volumes: dictionary of (volume,mount) pairs (Optional) + Volumes to mount in the pod. Both volume and mount must be a dictionary. + additional_tolerations: list of dictionaries (Optional) + Tolerations to add to the container + additional_node_selectors: dictionary of strings (Optional) + nodeSelectors to attach to the pod + additional_annotations: dictionary of strings (Optional) + Annotations to attach to the pod + k8s_domain: string (Optional) + Kubernetes domain to advertise + node_affinity_dict: dictionary of node affinity (Optional) + If key is prepended with ^, NotIn is used (In else) + Each value is either a tuple of strings + or a | separated string + """ + self.namespace = copy.deepcopy(namespace) + self.condor_host = copy.deepcopy(condor_host) + self.k8s_image = copy.deepcopy(k8s_image) + self.k8s_image_pull_policy = copy.deepcopy(k8s_image_pull_policy) + self.priority_class = copy.deepcopy(priority_class) + self.priority_class_cpu = copy.deepcopy(priority_class_cpu) + self.priority_class_gpu = copy.deepcopy(priority_class_gpu) + self.service_account_name = None + self.base_tolerations = copy.deepcopy(base_tolerations) + self.base_pvc_volumes = copy.deepcopy(base_pvc_volumes) + self.base_config_volumes = {} + self.additional_labels = copy.deepcopy(additional_labels) + self.additional_envs = copy.deepcopy(additional_envs) + self.additional_volumes = copy.deepcopy(additional_volumes) + self.additional_tolerations = copy.deepcopy(additional_tolerations) + self.additional_node_selectors = additional_node_selectors + self.additional_annotations = copy.deepcopy(additional_annotations) + self.app_name = copy.deepcopy(app_name) + self.k8s_job_ttl = k8s_job_ttl + self.k8s_domain = copy.deepcopy(k8s_domain) + self.force_k8s_namespace_matching = copy.deepcopy(force_k8s_namespace_matching) + self.additional_requirements = copy.deepcopy(additional_requirements) + self.node_affinity_dict = copy.deepcopy(node_affinity_dict) + self.request_fuse = "yes" + self.storage_mbs = "20000" + self.additional_resources = {} + self.use_large_limit = "yes" + self.apptainer = "yes" + + def parse(self, + dict, + fields=ProvisionerK8SConfigFields): + """Parse the valuies from a dictionary""" + self.namespace = provisioner_config_parser.update_parse(self.namespace, 'namespace', 'str', fields, dict) + self.condor_host = provisioner_config_parser.update_parse(self.condor_host, 'condor_host', 'str', fields, dict) + self.k8s_domain = provisioner_config_parser.update_parse(self.k8s_domain, 'k8s_domain', 'str', fields, dict) + self.k8s_image = provisioner_config_parser.update_parse(self.k8s_image, 'k8s_image', 'str', fields, dict) + self.k8s_image_pull_policy = provisioner_config_parser.update_parse(self.k8s_image_pull_policy, 'k8s_image_pull_policy', 'str', fields, dict) + self.priority_class = provisioner_config_parser.update_parse(self.priority_class, 'priority_class', 'str', fields, dict) + self.priority_class_cpu = provisioner_config_parser.update_parse(self.priority_class_cpu, 'priority_class_cpu', 'str', fields, dict) + self.priority_class_gpu = provisioner_config_parser.update_parse(self.priority_class_gpu, 'priority_class_gpu', 'str', fields, dict) + self.service_account_name = provisioner_config_parser.update_parse(self.service_account_name, 'service_account_name', 'str', fields, dict) + self.base_tolerations = provisioner_config_parser.update_parse(self.base_tolerations, 'tolerations_list', 'list', fields, dict) + self.base_pvc_volumes = provisioner_config_parser.update_parse(self.base_pvc_volumes, 'pvc_volumes_dict', 'dict', fields, dict) + self.base_config_volumes = provisioner_config_parser.update_parse(self.base_config_volumes, 'config_volumes_dict', 'dict', fields, dict) + self.additional_labels = provisioner_config_parser.update_parse(self.additional_labels, 'labels_dict', 'dict', fields, dict) + self.additional_envs = provisioner_config_parser.update_parse(self.additional_envs, 'envs_dict', 'dict', fields, dict) + self.additional_node_selectors = provisioner_config_parser.update_parse(self.additional_node_selectors, 'node_selectors_dict', 'dict', fields, dict) + self.additional_annotations = provisioner_config_parser.update_parse(self.additional_annotations, 'annotations_dict', 'dict', fields, dict) + self.app_name = provisioner_config_parser.update_parse(self.app_name, 'app_name', 'str', fields, dict) + self.k8s_job_ttl = provisioner_config_parser.update_parse(self.k8s_job_ttl, 'k8s_job_ttl', 'int', fields, dict) + self.force_k8s_namespace_matching = provisioner_config_parser.update_parse(self.force_k8s_namespace_matching, 'force_k8s_namespace_matching', 'str', fields, dict) + self.additional_requirements = provisioner_config_parser.update_parse(self.additional_requirements, 'additional_requirements', 'str', fields, dict) + self.node_affinity_dict = provisioner_config_parser.update_parse(self.node_affinity_dict, 'node_affinity_dict', 'dict', fields, dict) + self.request_fuse = provisioner_config_parser.update_parse(self.request_fuse, 'request_fuse', 'str', fields, dict) + self.storage_mbs = provisioner_config_parser.update_parse(self.storage_mbs, 'storage_mbs', 'str', fields, dict) + self.additional_resources = provisioner_config_parser.update_parse(self.additional_resources, 'resources_dict', 'dict', fields, dict) + self.use_large_limit = provisioner_config_parser.update_parse(self.use_large_limit, 'use_large_limit', 'str', fields, dict) + self.apptainer = provisioner_config_parser.update_parse(self.apptainer, 'apptainer', 'str', fields, dict) + +class ProvisionerK8S: + """Kubernetes Query interface""" + + def __init__(self, config): + self.start_time = int(time.time()) + self.submitted = 0 + # use deepcopy to avoid surprising changes at runtime + self.app_name = copy.deepcopy(config.app_name) + self.k8s_job_ttl = config.k8s_job_ttl + self.namespace = copy.deepcopy(config.namespace) + self.condor_host = copy.deepcopy(config.condor_host) + self.k8s_image = copy.deepcopy(config.k8s_image) + self.k8s_image_pull_policy = copy.deepcopy(config.k8s_image_pull_policy) + self.priority_class = copy.deepcopy(config.priority_class) + self.priority_class_cpu = copy.deepcopy(config.priority_class_cpu) + self.priority_class_gpu = copy.deepcopy(config.priority_class_gpu) + self.service_account_name = copy.deepcopy(config.service_account_name) + self.base_tolerations = copy.deepcopy(config.base_tolerations) + self.base_pvc_volumes = copy.deepcopy(config.base_pvc_volumes) + self.base_config_volumes = copy.deepcopy(config.base_config_volumes) + self.additional_labels = copy.deepcopy(config.additional_labels) + self.additional_envs = copy.deepcopy(config.additional_envs) + self.additional_volumes = copy.deepcopy(config.additional_volumes) + self.additional_tolerations = copy.deepcopy(config.additional_tolerations) + self.additional_node_selectors = copy.deepcopy(config.additional_node_selectors) + self.additional_annotations = copy.deepcopy(config.additional_annotations) + self.k8s_domain = copy.deepcopy(config.k8s_domain) + self.force_k8s_namespace_matching = copy.deepcopy(config.force_k8s_namespace_matching) + self.additional_requirements = copy.deepcopy(config.additional_requirements) + self.node_affinity_dict = copy.deepcopy(config.node_affinity_dict) + self.request_fuse = copy.deepcopy(config.request_fuse) + self.storage_mbs = copy.deepcopy(config.storage_mbs) + self.additional_resources = copy.deepcopy(config.additional_resources) + self.use_large_limit = copy.deepcopy(config.use_large_limit) + self.apptainer = copy.deepcopy(config.apptainer) + return + + def authenticate(self, use_service_account=True): + """Load credentials needed for authentication""" + if use_service_account: + # The pod must have a service accoun associated with it + # https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/ + # It also relies on KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT beign set + kubernetes.config.load_incluster_config() + else: + # Read config from ~/.kube/config + kubernetes.config.load_kube_config() + + def query(self): + """Return the list of jobs for my namespace""" + + pods=[] + + k8s = kubernetes.client.CoreV1Api() + plist = k8s.list_namespaced_pod(namespace=self.namespace, + label_selector="prp-htcondor-portal=wn,k8s-app=%s"%self.app_name) + # TBD: We may need to check for continuation flag + self._append_pods(pods,plist.items) + + return pods + + def submit(self, attrs, n_pods=1): + # first ensure that the basic int values are valid + int_vals={} + for k in ('CPUs','GPUs','Memory','Disk'): + int_vals[k] = int(attrs[k]) + + labels = { + 'k8s-app': self.app_name, + 'prp-htcondor-portal': 'wn', + 'PodCPUs': "%i"%int_vals['CPUs'], + 'PodGPUs': "%i"%int_vals['GPUs'], + 'PodMemory': "%i"%int_vals['Memory'], + 'PodDisk': "%i"%int_vals['Disk'] + } + for k in self.additional_labels.keys(): + labels[k] = copy.copy(self.additional_labels[k]) + self._augment_labels(labels, attrs) + + if self.use_large_limit == "yes": + # CPU is hardly ever used 100%... request 75% + # Similarly for memory, but request 80% there + req = { + 'memory': '%iMi'%int(int_vals['Memory']*0.8), + 'cpu': self._quantity(int_vals['CPUs']*0.75) + } + # but enforce Memory limit + delta + lim = { + 'memory': '%iMi'%(int_vals['Memory']+1000), + 'cpu': self._quantity(int_vals['CPUs']+0.25) + } + else: + # Stick to the needs, with just small memory delta + req = { + 'memory': '%iMi'%(int_vals['Memory']+1000), + 'cpu': self._quantity(int_vals['CPUs']) + } + lim = { + 'memory': '%iMi'%(int_vals['Memory']+1000), + 'cpu': self._quantity(int_vals['CPUs']) + } + + if int_vals['GPUs']>0: + req['nvidia.com/gpu'] = self._quantity(int_vals['GPUs']) + lim['nvidia.com/gpu'] = self._quantity(int_vals['GPUs']) + + if self.request_fuse != "no": + req['smarter-devices/fuse'] = "1" + lim['smarter-devices/fuse'] = "1" + + eph_storage = int_vals['Disk'] + if self.storage_mbs != "": + # many jobs do not define reasonable Disk requests + # keep a reasonable minimum + min_disk = int(self.storage_mbs)*1024 + if eph_storage "2", 1.5 -> "1.5")""" + if float(val).is_integer(): + return "%i"%int(val) + return "%s"%val + + def _append_pods(self, pods, mypods): + """pods is a list and will be updated in-place""" + for pod in mypods: + podattrs={} + labels=pod.metadata.labels + for k in labels.keys(): + # convert all values to strings, for easier management + podattrs[k]="%s"%labels[k] + del labels + podattrs['Name'] = pod.metadata.name + podattrs['Phase'] = pod.status.phase + podattrs['StartTime'] = pod.status.start_time + podattrs['HasContainers'] = pod.status.container_statuses!=None + pods.append(podattrs) + return + + # These can be re-implemented by derivative classes + def _get_k8s_image(self, attrs): + return (self.k8s_image,self.k8s_image_pull_policy) + + def _get_priority_class(self, attrs): + if int(attrs['GPUs'])>0: + pc = self.priority_class_gpu + else: + pc = self.priority_class_cpu + if pc==None: + pc = self.priority_class + return pc + + def _augment_labels(self, labels, attrs): + """Add any additional labels to the dictionary (attrs is read-only)""" + return + + def _augment_environment(self, env_list, attrs): + """Add any additional (name,value) pairs to the list (attrs is read-only)""" + return + + def _augment_volumes(self, volumes, attrs): + """Add any additional (volume,mount) pair to the dictionary (attrs is read-only)""" + + for v in self.base_pvc_volumes: + # Special characters in the volume path + # ! - read only + # # - HostToContainer mountPropagation + mro=False + mhp=False + mpath = self.base_pvc_volumes[v] + while mpath[0] in ('!','#'): + sc = mpath[0] + mpath = mpath[1:] + mro |= (sc=='!') + mhp |= (sc=='#') + + vdict = { + 'mountPath': mpath + } + if mro: + vdict['readOnly'] = True + if mhp: + vdict['mountPropagation'] = 'HostToContainer' + + volumes[v] = \ + ( + { + 'persistentVolumeClaim': { + 'claimName': v + } + }, + vdict + ) + + for v in self.base_config_volumes: + path = self.base_config_volumes[v] + fname = path.rsplit("/")[-1] + volumes[v] = \ + ( + { + 'configMap': { + 'name': v, + 'items': [{ + 'key': fname, + 'path': fname + }] + } + }, + { + 'mountPath': path, + 'subPath': fname + } + ) + + + # by default, we mount the token secret + volumes['configpasswd'] = \ + ( + { + 'secret': { + 'secretName': 'prp-htcondor-wn-secret', + # note: defaultMode is not valid in an item, so the API server always + # dropped the 'defaultMode': 256 that used to be here, and the client + # now rejects it; the file keeps the default mode, as before + 'items': [{ + 'key': 'prp-wn.token', + 'path': 'prp-wn.token' + }] + } + }, + { + 'mountPath': '/etc/condor/tokens.d/prp-wn.token', + 'subPath': 'prp-wn.token', + 'readOnly': True + } + ) + + return + + def _augment_tolerations(self, t_list, attrs): + """Add any additional dictionaries to the list (attrs is read-only)""" + for t in self.base_tolerations: + t_list.append({'effect': 'NoSchedule', + 'key': t, + 'operator': 'Exists'}) + + return + + def _augment_node_selectors(self, node_selectors, attrs): + """Add any additional elements to the dictionary (attrs is read-only)""" + return diff --git a/osg-htc/nrp-provisioner/src/provisioner_logging.py b/osg-htc/nrp-provisioner/src/provisioner_logging.py new file mode 100644 index 00000000..c64ab651 --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_logging.py @@ -0,0 +1,74 @@ +# +# prp-htcondor-portal/provisioner +# +# BSD license, copyright Igor Sfiligoi 2021 +# + +import sys +import time + +# abstract class, missing log method +class ProvisionerBaseLogging: + def __init__(self, want_log_debug=False, want_log_info=True): + self.want_log_debug = want_log_debug + self.want_log_info = want_log_info + + def log_info(self, message): + if self.want_log_info: + self.log("INFO", message) + + def log_error(self, message): + self.log("ERROR", message) + + def log_debug(self, message): + if self.want_log_debug: + self.log("DEBUG", message) + + # logs are not guaranteed to be saved until sync is called + def sync(self): + return #derivative classes can update this + +class ProvisionerStdoutLogging(ProvisionerBaseLogging): + def __init__(self, want_log_debug=False, want_log_info=True): + ProvisionerBaseLogging.__init__(self, want_log_debug, want_log_info) + + def log(self, level_str, message): + now = time.ctime() + print("%s [%s] %s"%(now, level_str, message) ) + + +class ProvisionerFileLogging(ProvisionerBaseLogging): + def __init__(self, fname, dated=True, want_log_debug=False, want_log_info=True, also_stdout=False): + ProvisionerBaseLogging.__init__(self, want_log_debug, want_log_info) + self.buffer = [] + self.fname = fname + self.dated = dated + self.also_stdout = also_stdout # e.g. so that kubectl logs shows the same lines + + def log(self, level_str, message): + now = time.ctime() + self.buffer.append("%s [%s] %s\n"%(now, level_str, message) ) + + def sync(self): + if len(self.buffer)==0: + return # nothing to do + fname = self._getfname() + blob = "".join(self.buffer); + with open(fname,"a") as fd: + fd.write(blob); + if self.also_stdout: + # stdout is block-buffered when not a terminal, so flush to show up right away + sys.stdout.write(blob) + sys.stdout.flush() + # we can now reset the buffer + self.buffer = [] + + # internal + def _getfname(self): + if self.dated: + return "%s.%s"%(self.fname,time.strftime("%Y%m%d")) + else: + return self.fname + + + diff --git a/osg-htc/nrp-provisioner/tests/image.sh b/osg-htc/nrp-provisioner/tests/image.sh new file mode 100755 index 00000000..9486c738 --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/image.sh @@ -0,0 +1,61 @@ +#!/bin/bash +# +# Checks a built provisioner image. Usage: +# +# tests/image.sh (from the provisioner directory) +# +# Needs docker. The startup check makes real (read-only) queries to the OSPool collector and +# to IceCube's schedd (tests/sample_osg_provisioner.conf), so it needs network access; without +# it the loop still starts, but logs query failures. +# +set -u +IMG=${1:?usage: $0 } +TESTS=$(cd "$(dirname "$0")" && pwd) +# inside the nrp-provisioner repo, the unit tests also check the namespace configs in k8s/ +K8S=$(cd "$TESTS/../../k8s" 2>/dev/null && pwd) +PLATFORM=${PLATFORM:-linux/amd64} +fails=0 +pass() { echo "PASS $*"; } +fail() { echo "FAIL $*"; fails=$((fails+1)); } +run() { docker run --rm --platform "$PLATFORM" "$@"; } + +echo "== $IMG" +run --entrypoint bash "$IMG" -c 'grep PRETTY_NAME /etc/os-release; python3 --version; rpm -q condor python3-condor osg-ca-certs; python3 -c "import kubernetes; print(\"kubernetes\", kubernetes.__version__)"' | sed 's/^/ /' + +# 1) unit tests, inside the image (they need the htcondor2 bindings and the kubernetes client) +out=$(run -v "$TESTS:/t:ro" ${K8S:+-v "$K8S:/k8s:ro"} --entrypoint bash "$IMG" -c \ + 'mkdir -p /tmp/w/p && ln -s /opt/provisioner/src /tmp/w/p/src && cp -r /t /tmp/w/p/tests && { [ ! -d /k8s ] || ln -s /k8s /tmp/w/k8s; } && cd /tmp/w/p && python3 -B -m unittest discover -s tests 2>&1') +echo "$out" | grep -q "^OK" && pass "unit tests: $(echo "$out" | grep '^Ran')" || { fail "unit tests"; echo "$out" | tail -20; } +echo "$out" | grep -q "skipped" && fail "unit tests were skipped inside the image: $(echo "$out" | grep -o 'skipped=[0-9]*')" + +# 2) the provisioner user's home has the shell profile that 'su -l' needs +home=$(run --entrypoint bash "$IMG" -c 'ls -A /home/provisioner') +echo "$home" | grep -qx ".bash_profile" && pass "home has .bash_profile" || fail "home has no .bash_profile (useradd skipped /etc/skel?)" + +# 3) the setup scripts' environment reaches the provisioner's login shell +envout=$(run -e HTCONDOR_QUERY_INSECURE=yes -e KUBERNETES_SERVICE_HOST=192.0.2.1 -e KUBERNETES_SERVICE_PORT=443 -e K8S_NAMESPACE=osg-test --entrypoint bash "$IMG" -c \ + 'mkdir -p /etc/condor/tokens.d && touch /etc/condor/tokens.d/test.token; /opt/provisioner/setup_k8s_creds.sh; /opt/provisioner/setup_htcondor_creds.sh; su -l provisioner -c "echo \$KUBERNETES_SERVICE_HOST \$KUBERNETES_SERVICE_PORT \$K8S_NAMESPACE \$_CONDOR_SEC_CLIENT_AUTHENTICATION; ls ~/.condor/tokens.d"') +[ "$(echo "$envout" | head -1)" = "192.0.2.1 443 osg-test OPTIONAL" ] && pass "environment reaches 'su -l provisioner'" || fail "environment did not reach the provisioner: $envout" +echo "$envout" | grep -qx "test.token" && pass "tokens copied to the provisioner user" || fail "tokens not copied" + +# 4) startup with a fake service account: must get into the provisioning loop +sa=$(mktemp -d); echo fake > "$sa/token"; echo osg-icecube > "$sa/namespace" +printf -- '-----BEGIN CERTIFICATE-----\nZmFrZQ==\n-----END CERTIFICATE-----\n' > "$sa/ca.crt" +chmod 755 "$sa"; chmod 644 "$sa"/* # mktemp -d is 0700; the provisioner user in the container must read it +start=$(run -e CONDOR_HOST=cm-1.ospool.osg-htc.org,cm-2.ospool.osg-htc.org -e K8S_NAMESPACE=osg-icecube -e CVMFS_MOUNTS=icecube,singularity \ + -e HTCONDOR_QUERY_INSECURE=yes -e KUBERNETES_SERVICE_HOST=127.0.0.1 -e KUBERNETES_SERVICE_PORT=6443 \ + -v "$sa:/var/run/secrets/kubernetes.io/serviceaccount:ro" -v "$TESTS/sample_osg_provisioner.conf:/cfg/osg_provisioner.conf:ro" \ + --entrypoint bash "$IMG" -c ' + mkdir -p /etc/condor/tokens.d && touch /etc/condor/tokens.d/none.token + cp /cfg/osg_provisioner.conf /home/provisioner/ && chown provisioner /home/provisioner/osg_provisioner.conf + /opt/provisioner/provisioner_startup.sh > /tmp/out 2>&1 & + for i in $(seq 1 90); do grep -q "Iteration started" /tmp/out 2>/dev/null && break; sleep 1; done + sleep 3; cat /tmp/out') +rm -rf "$sa" +echo "$start" | grep -q "Iteration started" && pass "startup reaches the provisioning loop (and logs it to stdout)" || { fail "startup did not reach the loop"; echo "$start" | tail -20 | sed 's/^/ /'; } +echo "$start" | grep -q "Traceback" && { fail "traceback during startup"; echo "$start" | grep -A8 Traceback | sed 's/^/ /'; } || pass "no traceback during startup" +echo "$start" | grep -q "FutureWarning\|DeprecationWarning" && fail "deprecation warnings during startup" || pass "no deprecation warnings" +echo "$start" | grep -q "Failed to query HTCondor schedd" && echo "NOTE a schedd query failed (network, CA bundle or schedd policy?):" && echo "$start" | grep "Failed to query HTCondor schedd" | head -3 | sed 's/^/ /' + +echo; [ $fails -eq 0 ] && echo "ALL PASSED" || echo "$fails FAILED" +exit $fails diff --git a/osg-htc/nrp-provisioner/tests/sample_osg_provisioner.conf b/osg-htc/nrp-provisioner/tests/sample_osg_provisioner.conf new file mode 100644 index 00000000..b321f5dd --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/sample_osg_provisioner.conf @@ -0,0 +1,24 @@ +# Sample provisioner config for the tests, modeled on the osg-icecube one. The startup check in +# image.sh uses it for real (read-only) queries to the OSPool and IceCube's schedd. +[DEFAULT] +apptainer=no +k8s_domain=nrp-nautilus.io +app_name=osg-direct +force_k8s_namespace_matching=no +additional_requirements=Group=?="IceCube"&&RequestGPUs=?=1&&(ProjectName=?="IceCube") + +[htcondor] +schedd_whitelist_regexp=(.*sub.*\.icecube\.wisc\.edu) + +[k8s] +k8s_image=hub.opensciencegrid.org/osg-htc/nrp-ospool-ep:25-release +tolerations_list=nautilus.io/noceph +node_affinity_dict=^nautilus.io/low-power:true,nvidia.com/gpu.product:NVIDIA-A10|NVIDIA-GeForce-RTX-3090 +priority_class=opportunistic +max_submit_pods_per_cluster=300 +k8s_job_ttl=7200 +use_large_limit=no +service_account_name=prp-htcondor-wn-ro +envs_dict=GLIDEIN_Site:SDSC-PRP,GLIDEIN_ResourceName:SDSC-PRP-OSPool-Provisioner +annotations_dict=container.apparmor.security.beta.kubernetes.io/htcondor:unconfined +labels_dict=applicationId:osg-test-wn diff --git a/osg-htc/nrp-provisioner/tests/test_backoff.py b/osg-htc/nrp-provisioner/tests/test_backoff.py new file mode 100644 index 00000000..82961b1c --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_backoff.py @@ -0,0 +1,213 @@ +# +# Unit tests for the provisioning back-off +# +# Run from the provisioner directory: +# python3 -m unittest discover -s tests +# + +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) + +from src import provisioner_backoff +from src import provisioner_clustering +from src import event_loop + +SHAPE = {'PodCPUs': '1', 'PodMemory': '4096', 'PodDisk': '8000000', 'PodGPUs': '0'} +OTHER_SHAPE = {'PodCPUs': '19', 'PodMemory': '4096', 'PodDisk': '8000000', 'PodGPUs': '0'} + +def pod(name, phase, shape=SHAPE): + el = dict(shape) + el.update({'Name': name, 'Phase': phase, 'StartTime': None, 'HasContainers': True}) + return el + +def startd(name, state='Unclaimed', job_starts=None): + ad = {'Name': 'slot1@%s'%name, 'K8SPodName': name, 'State': state} + if job_starts!=None: + ad['JobStarts'] = '%i'%job_starts + return ad + +def clusters(pods, startds): + return provisioner_clustering.ProvisionerClustering().cluster_k8s_pods(pods, startds) + +def cluster_id(shape=SHAPE): + return list(clusters([pod('x', 'Running', shape)], []).keys())[0] + +class TestBackoff(unittest.TestCase): + def setUp(self): + self.b = provisioner_backoff.ProvisionerBackoff(max_unclaimed_exits=2, backoff_time=3600, max_backoff_time=8*3600) + self.cid = cluster_id() + + def run_pod(self, name, now, claimed=False, job_starts=None, end_phase='Succeeded', shape=SHAPE): + """One pod lifecycle: running with a startd, then gone into end_phase""" + state = 'Claimed' if claimed else 'Unclaimed' + self.b.update(clusters([pod(name, 'Running', shape)], [startd(name, state, job_starts)]), now) + return self.b.update(clusters([pod(name, end_phase, shape)], []), now+60) + + def test_suspends_after_unclaimed_exits(self): + self.run_pod('p1', 1000) + self.assertEqual(self.b.suspended_for(self.cid, 1100), 0) + msgs = self.run_pod('p2', 1100) + self.assertEqual(len(msgs), 1) + self.assertEqual(self.b.suspended_for(self.cid, 1160), 3600) + self.assertEqual(self.b.get_state(self.cid), (2, 1)) + + def test_claim_resets(self): + self.run_pod('p1', 1000) + self.run_pod('p2', 1100, claimed=True) + self.run_pod('p3', 1200) + self.assertEqual(self.b.suspended_for(self.cid, 1300), 0) + self.assertEqual(self.b.get_state(self.cid), (1, 0)) + + def test_job_starts_counts_as_claimed(self): + # a job may come and go between two polls; JobStarts still shows it + self.run_pod('p1', 1000, job_starts=1) + self.run_pod('p2', 1100, job_starts=1) + self.assertEqual(self.b.suspended_for(self.cid, 1200), 0) + + def test_claim_during_suspension_clears_it(self): + self.run_pod('p1', 1000) + self.run_pod('p2', 1100) + self.assertGreater(self.b.suspended_for(self.cid, 1200), 0) + msgs = self.b.update(clusters([pod('p3', 'Running')], [startd('p3', 'Claimed')]), 1300) + self.assertEqual(len(msgs), 1) + self.assertEqual(self.b.suspended_for(self.cid, 1300), 0) + self.assertEqual(self.b.get_state(self.cid), (0, 0)) + + def test_failed_and_vanished_pods_are_neutral(self): + self.run_pod('p1', 1000, end_phase='Failed') + self.b.update(clusters([pod('p2', 'Running')], [startd('p2')]), 1100) + self.b.update(clusters([], []), 1200) # p2 vanished + self.run_pod('p3', 1300) + self.assertEqual(self.b.suspended_for(self.cid, 1400), 0) + self.assertEqual(self.b.get_state(self.cid), (1, 0)) + + def test_pods_that_never_started_a_startd_are_neutral(self): + for name in ('p1', 'p2', 'p3'): + self.b.update(clusters([pod(name, 'Pending')], []), 1000) + self.b.update(clusters([pod(name, 'Succeeded')], []), 1100) + self.assertEqual(self.b.suspended_for(self.cid, 1200), 0) + + def test_pods_first_seen_finished_are_ignored(self): + # e.g. pods from before a provisioner restart + self.b.update(clusters([pod('old1', 'Succeeded'), pod('old2', 'Succeeded')], []), 1000) + self.assertEqual(self.b.suspended_for(self.cid, 1000), 0) + + def test_no_extension_during_suspension_then_doubling_and_cap(self): + self.run_pod('p1', 1000) + self.run_pod('p2', 1000) + first = self.b.suspended_for(self.cid, 1060) + self.run_pod('p3', 2000) # leftover pod idling out during the suspension + self.assertEqual(self.b.suspended_for(self.cid, 1060), first) + t = 1060+first + durations = [] + for i in range(6): + self.run_pod('q%i'%i, t) + d = self.b.suspended_for(self.cid, t+60) + durations.append(d) + t += 60+d + self.assertEqual(durations, [7200, 14400, 28800, 28800, 28800, 28800]) + + def test_clusters_are_independent(self): + other = cluster_id(OTHER_SHAPE) + self.run_pod('a1', 1000, shape=OTHER_SHAPE) + self.run_pod('a2', 1100, shape=OTHER_SHAPE) + self.assertGreater(self.b.suspended_for(other, 1200), 0) + self.assertEqual(self.b.suspended_for(self.cid, 1200), 0) + + def test_disabled(self): + b = provisioner_backoff.ProvisionerBackoff(max_unclaimed_exits=0) + for name in ('p1', 'p2', 'p3'): + b.update(clusters([pod(name, 'Running')], [startd(name)]), 1000) + self.assertEqual(b.update(clusters([pod(name, 'Succeeded')], []), 1100), []) + self.assertEqual(b.suspended_for(self.cid, 1200), 0) + + +class FakeLog: + def __init__(self): + self.lines = [] + def log_debug(self, m): self.lines.append(m) + def log_info(self, m): self.lines.append(m) + def log_error(self, m): self.lines.append(m) + def sync(self): pass + +class FakeSchedd: + def __init__(self, n_idle): self.n_idle = n_idle + def query_idle(self, projection=[]): + return [{'JobStatus': 1, 'RequestCPUs': 1, 'RequestMemory': 4096, 'RequestDisk': 8000000} for i in range(self.n_idle)] + +class FakeCollector: + def __init__(self): self.ads = [] + def query(self, projection=[]): + self.projection = projection + return self.ads + +class FakeK8S: + def __init__(self): + self.pods = [] + self.submitted = [] + def query(self): return self.pods + def submit(self, attrs, n_pods=1): + self.submitted.append(n_pods) + return 'job-%i'%len(self.submitted) + +class TestEventLoopBackoff(unittest.TestCase): + def make_loop(self, backoff, trial_pods=0, n_idle=8): + self.log = FakeLog() + self.collector = FakeCollector() + self.k8s = FakeK8S() + return event_loop.ProvisionerEventLoop(self.log, FakeSchedd(n_idle), self.collector, self.k8s, 20, 1000, backoff, trial_pods) + + def test_suspended_cluster_is_not_provisioned(self): + b = provisioner_backoff.ProvisionerBackoff(max_unclaimed_exits=2, backoff_time=3600) + el = self.make_loop(b) + el.one_iteration() + self.assertEqual(self.k8s.submitted, [3]) # 1+8/4 pods for 8 idle jobs + self.assertEqual(self.collector.projection, ['JobStarts']) + # two of those pods idle out without ever being claimed + self.k8s.pods = [pod('p1', 'Running'), pod('p2', 'Running')] + self.collector.ads = [startd('p1'), startd('p2')] + el.one_iteration() + self.k8s.pods = [pod('p1', 'Succeeded'), pod('p2', 'Succeeded')] + self.collector.ads = [] + self.k8s.submitted = [] + el.one_iteration() + self.assertEqual(self.k8s.submitted, []) + self.assertTrue(any('suspending provisioning' in l for l in self.log.lines)) + self.assertTrue(any('provisioning suspended for' in l for l in self.log.lines)) + + def test_no_backoff_keeps_old_behaviour(self): + el = self.make_loop(None) + self.k8s.pods = [pod('p1', 'Succeeded'), pod('p2', 'Succeeded')] + el.one_iteration() + self.assertEqual(self.k8s.submitted, [3]) + + def test_trial_pods_until_claimed(self): + el = self.make_loop(None, trial_pods=2, n_idle=40) + el.one_iteration() + self.assertEqual(self.k8s.submitted, [2]) # not 1+40/4=11 + # the trial pods wait unclaimed: nothing more + self.k8s.pods = [pod('p1', 'Running'), pod('p2', 'Running')] + self.collector.ads = [startd('p1'), startd('p2')] + self.k8s.submitted = [] + el.one_iteration() + self.assertEqual(self.k8s.submitted, []) + # one gets claimed: the cluster gets the full number + self.collector.ads = [startd('p1', 'Claimed'), startd('p2')] + el.one_iteration() + self.assertEqual(self.k8s.submitted, [10]) # 11 wanted, p2 still unclaimed + + def test_trial_pods_below_limit_changes_nothing(self): + el = self.make_loop(None, trial_pods=2, n_idle=2) + el.one_iteration() + self.assertEqual(self.k8s.submitted, [1]) + + def test_trial_pods_disabled(self): + el = self.make_loop(None, trial_pods=0, n_idle=40) + el.one_iteration() + self.assertEqual(self.k8s.submitted, [11]) + +if __name__ == '__main__': + unittest.main() diff --git a/osg-htc/nrp-provisioner/tests/test_htcondor_ads.py b/osg-htc/nrp-provisioner/tests/test_htcondor_ads.py new file mode 100644 index 00000000..1a0f1e0d --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_htcondor_ads.py @@ -0,0 +1,131 @@ +# +# Unit tests for turning HTCondor ads into the provisioner's job/startd dictionaries, +# and for the job clustering built on top of them. +# +# The ads are synthetic (no real users or paths), but use the same attribute names +# and expression forms that real schedds return for the provisioner's projection. +# Needs the htcondor2/classad2 bindings (present in the provisioner image). +# +# Run from the provisioner directory: +# python3 -m unittest discover -s tests +# + +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) + +try: + import classad2 + from src import provisioner_htcondor + from src import provisioner_clustering +except ImportError: + classad2 = None + +JOBS = """ +[ ClusterId = 1; ProcId = 0; JobStatus = 1; RequestCPUs = 1; + RequestMemory = ifThenElse(MemoryUsage isnt undefined,MemoryUsage,500); + RequestDisk = 8000001 ] +[ ClusterId = 2; ProcId = 0; JobStatus = 1; RequestCPUs = 2; + RequestMemory = 5000; RequestDisk = 20971520; RequestGPUs = 1 ] +[ ClusterId = 3; ProcId = 0; JobStatus = 1; RequestCPUs = 2; + RequestMemory = 4096; RequestDisk = 8000000 ] +[ ClusterId = 4; ProcId = 0; JobStatus = 1; RequestCPUs = 2; + RequestMemory = 4100; DiskUsage = 9000000; RequestDisk = DiskUsage; RequestGPUs = 1 ] +""" + +STARTDS = """ +[ Name = "slot1@pod-a"; AuthenticatedIdentity = "pool@cm.example"; State = "Claimed"; + Activity = "Busy"; K8SPodName = "pod-a"; JobStarts = 2; HAS_SINGULARITY = false ] +[ Name = "slot1@pod-b"; AuthenticatedIdentity = "pool@cm.example"; State = "Unclaimed"; + Activity = "Idle"; K8SPodName = "pod-b" ] +[ Name = "slot1@pod-c"; AuthenticatedIdentity = "someone@elsewhere"; State = "Unclaimed"; + K8SPodName = "pod-c" ] +[ Name = "slot1@pod-d"; State = "Unclaimed"; K8SPodName = "pod-d" ] +""" + +class Log: + def __init__(self): self.lines = [] + def log_debug(self, m): self.lines.append(m) + log_info = log_error = log_debug + def sync(self): pass + +@unittest.skipIf(classad2 is None, "htcondor2/classad2 bindings not installed") +class TestJobConversion(unittest.TestCase): + def convert(self): + config = provisioner_htcondor.ProvisionerHTCConfig("ns") + schedd = provisioner_htcondor.ProvisionerSchedd(Log(), {".*": ".*"}, config) + jobs = [] + schedd._append_jobs("schedd.example", jobs, list(classad2.parseAds(JOBS))) + return {j['ClusterId']: j for j in jobs} + + def test_values_become_strings_and_schedd_is_recorded(self): + j = self.convert()['2'] + self.assertEqual(j['ScheddName'], "schedd.example") + self.assertEqual(j['RequestCPUs'], "2") + self.assertEqual(j['JobStatus'], "1") + self.assertEqual(j['RequestGPUs'], "1") + + def test_memory_rounds_up_to_1gb_and_small_values_are_dropped(self): + jobs = self.convert() + self.assertNotIn('RequestMemory', jobs['1']) # expression evaluates to 500 < 4096: treated as unset + self.assertEqual(jobs['2']['RequestMemory'], "5120") + self.assertEqual(jobs['3']['RequestMemory'], "4096") # exact multiple stays + self.assertEqual(jobs['4']['RequestMemory'], "5120") + + def test_disk_rounds_up_to_4gb_and_expressions_are_evaluated(self): + jobs = self.convert() + self.assertEqual(jobs['1']['RequestDisk'], "12000000") + self.assertEqual(jobs['2']['RequestDisk'], "24000000") + self.assertEqual(jobs['3']['RequestDisk'], "8000000") + self.assertEqual(jobs['4']['RequestDisk'], "12000000") # RequestDisk = DiskUsage = 9000000 + + def test_clustering_by_shape_with_defaults(self): + jobs = list(self.convert().values()) + groups = provisioner_clustering.ProvisionerClustering().cluster_schedd_jobs(jobs) + counts = {k: g.count_idle() for k, g in groups.items()} + self.assertEqual(counts, { + "1;4096;12000000;;0;;": 1, # memory unset -> default 4096, GPUs unset -> 0 + "2;5120;24000000;;1;;": 1, + "2;4096;8000000;;0;;": 1, + "2;5120;12000000;;1;;": 1, + }) + attrs = groups["2;5120;24000000;;1;;"].get_attr_dict() + self.assertEqual(attrs['CPUs'], "2") + self.assertEqual(attrs['GPUs'], "1") + +@unittest.skipIf(classad2 is None, "htcondor2/classad2 bindings not installed") +class TestCollectorConversion(unittest.TestCase): + def setUp(self): + self.ads = list(classad2.parseAds(STARTDS)) + self.queries = [] + outer = self + class FakeCollector: + def query(self, **kwargs): + outer.queries.append(kwargs) + return outer.ads + self.orig = provisioner_htcondor.htcondor2.Collector + provisioner_htcondor.htcondor2.Collector = lambda *a, **k: FakeCollector() + + def tearDown(self): + provisioner_htcondor.htcondor2.Collector = self.orig + + def test_startd_ads(self): + config = provisioner_htcondor.ProvisionerHTCConfig("osg-test") + config.app_name = "osg-direct"; config.k8s_domain = "example.org" + coll = provisioner_htcondor.ProvisionerCollector(Log(), "pool@cm\\.example", config) + startds = {s['K8SPodName']: s for s in coll.query(projection=['JobStarts'])} + # only ads with a trusted identity survive; ads without one are ignored + self.assertEqual(sorted(startds), ["pod-a", "pod-b"]) + self.assertEqual(startds['pod-a']['State'], "Claimed") + self.assertEqual(startds['pod-a']['JobStarts'], "2") + self.assertEqual(startds['pod-a']['HAS_SINGULARITY'], "False") + self.assertNotIn('JobStarts', startds['pod-b']) + q = self.queries[0] + self.assertIn('JobStarts', q['projection']) + self.assertIn('K8SNamespace=?="osg-test"', q['constraint']) + self.assertIn('K8SProvisionerName=?="osg-direct"', q['constraint']) + +if __name__ == '__main__': + unittest.main() diff --git a/osg-htc/nrp-provisioner/tests/test_k8s_submit.py b/osg-htc/nrp-provisioner/tests/test_k8s_submit.py new file mode 100644 index 00000000..5f44acab --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_k8s_submit.py @@ -0,0 +1,80 @@ +# +# Unit tests for the k8s Job submission +# +# The kubernetes client (>= 34) validates request bodies before sending them, +# so run submit() through the real client and only intercept the network send. +# +# Run from the provisioner directory: +# python3 -m unittest discover -s tests +# + +import configparser +import glob +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) + +try: + import kubernetes + from src import provisioner_k8s +except ImportError: + kubernetes = None + +TESTS_DIR = os.path.dirname(os.path.abspath(__file__)) +# the sample config, plus every namespace config when run inside the nrp-provisioner repo (k8s/) +CONFIGS = {'sample': os.path.join(TESTS_DIR, 'sample_osg_provisioner.conf')} +CONFIGS.update((os.path.basename(os.path.dirname(p)), p) + for p in sorted(glob.glob(os.path.join(TESTS_DIR, '..', '..', 'k8s', 'osg-*', 'osg_provisioner.conf')))) + +class Sent(Exception): + pass + +@unittest.skipIf(kubernetes is None, "kubernetes client not installed") +class TestSubmit(unittest.TestCase): + def setUp(self): + self.sent = [] + self.orig_call_api = kubernetes.client.ApiClient.call_api + def fake_call_api(api_client, method, url, header_params=None, body=None, *args, **kwargs): + self.sent.append(body) + raise Sent() + kubernetes.client.ApiClient.call_api = fake_call_api + + def tearDown(self): + kubernetes.client.ApiClient.call_api = self.orig_call_api + + def make_k8s(self, name, use_large_limit): + f = configparser.ConfigParser() + f.read(CONFIGS[name]) + k = provisioner_k8s.ProvisionerK8SConfig(name if name.startswith('osg-') else 'osg-test') + k.parse(f['k8s'] if 'k8s' in f else f['DEFAULT']) + k.base_pvc_volumes['cvmfs'] = "#/cvmfs" + k.use_large_limit = use_large_limit + return provisioner_k8s.ProvisionerK8S(k) + + def submit(self, k8s, cpus, gpus): + with self.assertRaises(Sent): # Sent means the body passed the client validation + k8s.submit({'CPUs': cpus, 'GPUs': gpus, 'Memory': "8192", 'Disk': "8000000"}, 2) + return self.sent[-1] + + def test_all_configs_pass_client_validation(self): + for name in CONFIGS: + for large in ('yes', 'no'): + for gpus in ('0', '1'): + with self.subTest(config=name, use_large_limit=large, gpus=gpus): + self.submit(self.make_k8s(name, large), "3", gpus) + + def test_quantities_are_strings(self): + body = self.submit(self.make_k8s('sample', 'yes'), "3", "1") + res = body['spec']['template']['spec']['containers'][0]['resources'] + self.assertEqual(res['requests']['cpu'], "2.25") + self.assertEqual(res['limits']['cpu'], "3.25") + self.assertEqual(res['requests']['nvidia.com/gpu'], "1") + body = self.submit(self.make_k8s('sample', 'no'), "3", "0") + res = body['spec']['template']['spec']['containers'][0]['resources'] + self.assertEqual(res['requests']['cpu'], "3") + self.assertNotIn('nvidia.com/gpu', res['requests']) + +if __name__ == '__main__': + unittest.main() diff --git a/osg-htc/nrp-provisioner/tests/test_logging.py b/osg-htc/nrp-provisioner/tests/test_logging.py new file mode 100644 index 00000000..6c132dfd --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_logging.py @@ -0,0 +1,64 @@ +# +# Unit tests for the provisioner file logging +# +# Run from the provisioner directory: +# python3 -m unittest discover -s tests +# + +import io +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) + +from src import provisioner_logging + +class TestFileLogging(unittest.TestCase): + def setUp(self): + self.dir = tempfile.TemporaryDirectory() + self.fname = os.path.join(self.dir.name, 'provisioner.log') + self.stdout = sys.stdout + sys.stdout = io.StringIO() + + def tearDown(self): + sys.stdout = self.stdout + self.dir.cleanup() + + def read_file(self): + with open(self.fname) as fd: + return fd.read() + + def test_also_stdout_mirrors_the_file(self): + log = provisioner_logging.ProvisionerFileLogging(self.fname, dated=False, want_log_debug=True, also_stdout=True) + log.log_debug("first") + log.log_error("second") + self.assertEqual(sys.stdout.getvalue(), "") # nothing until sync + log.sync() + self.assertEqual(sys.stdout.getvalue(), self.read_file()) + self.assertIn("[DEBUG] first", self.read_file()) + self.assertIn("[ERROR] second", self.read_file()) + log.log_info("third") + log.sync() + self.assertEqual(sys.stdout.getvalue(), self.read_file()) + self.assertEqual(self.read_file().count("\n"), 3) + + def test_default_is_file_only(self): + log = provisioner_logging.ProvisionerFileLogging(self.fname, dated=False, want_log_debug=True) + log.log_info("only in the file") + log.sync() + self.assertEqual(sys.stdout.getvalue(), "") + self.assertIn("[INFO] only in the file", self.read_file()) + + def test_debug_filtering_applies_to_both(self): + log = provisioner_logging.ProvisionerFileLogging(self.fname, dated=False, want_log_debug=False, also_stdout=True) + log.log_debug("hidden") + log.log_info("shown") + log.sync() + self.assertNotIn("hidden", sys.stdout.getvalue()) + self.assertNotIn("hidden", self.read_file()) + self.assertIn("shown", sys.stdout.getvalue()) + +if __name__ == '__main__': + unittest.main() diff --git a/osg-htc/nrp-provisioner/tests/test_schedd_query.py b/osg-htc/nrp-provisioner/tests/test_schedd_query.py new file mode 100644 index 00000000..798764a2 --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_schedd_query.py @@ -0,0 +1,141 @@ +# +# Unit tests for ProvisionerSchedd: the schedd list, the job query constraint and +# failure handling. The collector and schedds are faked; no network. +# Needs the htcondor2/classad2 bindings (present in the provisioner image). +# +# Run from the provisioner directory: +# python3 -m unittest discover -s tests +# + +import configparser +import glob +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) + +try: + import classad2 + from src import provisioner_htcondor +except ImportError: + classad2 = None + +TESTS_DIR = os.path.dirname(os.path.abspath(__file__)) +# the sample config, plus every namespace config when run inside the nrp-provisioner repo (k8s/) +CONFIGS = {'sample': os.path.join(TESTS_DIR, 'sample_osg_provisioner.conf')} +CONFIGS.update((os.path.basename(os.path.dirname(p)), p) + for p in sorted(glob.glob(os.path.join(TESTS_DIR, '..', '..', 'k8s', 'osg-*', 'osg_provisioner.conf')))) + +SCHEDDS = """ +[ Name = "ap1.example.org"; AuthenticatedIdentity = "ap1@pool"; MyAddress = "<192.0.2.1:9618>"; + CondorVersion = "$CondorVersion: 25.0.0 $"; CondorPlatform = "$CondorPlatform: X86_64 $" ] +[ Name = "ap2.example.org"; AuthenticatedIdentity = "ap2@pool"; MyAddress = "<192.0.2.2:9618>"; + CondorVersion = "$CondorVersion: 25.0.0 $"; CondorPlatform = "$CondorPlatform: X86_64 $" ] +[ Name = "other.example.com"; AuthenticatedIdentity = "x@pool"; MyAddress = "<192.0.2.3:9618>" ] +[ Name = "noidentity.example.org"; MyAddress = "<192.0.2.4:9618>" ] +""" + +JOB = "[ ClusterId = 7; ProcId = 0; JobStatus = 1; RequestCPUs = 1 ]" + +class Log: + def __init__(self): self.lines = [] + def log_debug(self, m): self.lines.append(m) + log_info = log_error = log_debug + +@unittest.skipIf(classad2 is None, "htcondor2/classad2 bindings not installed") +class TestScheddQuery(unittest.TestCase): + def setUp(self): + self.collector_queries = [] + self.schedd_queries = [] + self.failing = set() + outer = self + class FakeCollector: + def query(self, **kwargs): + outer.collector_queries.append(kwargs) + return list(classad2.parseAds(SCHEDDS)) + class FakeSchedd: + def __init__(self, location_ad): + self.name = location_ad['Name'] + def query(self, constraint, projection): + outer.schedd_queries.append((self.name, constraint, projection)) + if self.name in outer.failing: + raise RuntimeError("Failed to fetch ads from schedd") + return list(classad2.parseAds(JOB)) + self.orig = (provisioner_htcondor.htcondor2.Collector, provisioner_htcondor.htcondor2.Schedd) + provisioner_htcondor.htcondor2.Collector = lambda *a, **k: FakeCollector() + provisioner_htcondor.htcondor2.Schedd = FakeSchedd + + def tearDown(self): + provisioner_htcondor.htcondor2.Collector, provisioner_htcondor.htcondor2.Schedd = self.orig + + def schedd(self, whitelist=r".*\.example\.org", **settings): + config = provisioner_htcondor.ProvisionerHTCConfig("osg-test") + config.parse(dict(settings)) + log = Log() + return provisioner_htcondor.ProvisionerSchedd(log, {whitelist: ".*"}, config), log + + def constraint(self): + return self.schedd_queries[-1][1] + + def test_location_projection_has_what_htcondor2_needs(self): + s, _ = self.schedd() + s.query_idle() + proj = self.collector_queries[0]['projection'] + for attr in ('Name', 'AuthenticatedIdentity', 'MyAddress', 'AddressV1', 'CondorVersion', 'CondorPlatform', 'Machine'): + self.assertIn(attr, proj) + + def test_whitelist_and_identity_filtering(self): + s, _ = self.schedd() + jobs = s.query_idle() + self.assertEqual(sorted(q[0] for q in self.schedd_queries), ["ap1.example.org", "ap2.example.org"]) + self.assertEqual(sorted(j['ScheddName'] for j in jobs), ["ap1.example.org", "ap2.example.org"]) + + def test_apptainer_toggle(self): + s, _ = self.schedd(apptainer="no") + s.query_idle() + self.assertTrue(self.constraint().endswith(" && isUndefined(SingularityImage)")) + s, _ = self.schedd(apptainer="yes") + s.query_idle() + self.assertTrue(self.constraint().endswith(" && (!isUndefined(SingularityImage))")) + s, _ = self.schedd() # default is yes + s.query_idle() + self.assertTrue(self.constraint().endswith(" && (!isUndefined(SingularityImage))")) + + def test_namespace_matching_and_additional_requirements(self): + s, _ = self.schedd(additional_requirements="RequestGPUs=?=1") + s.query_idle(projection=['RequestCPUs']) + self.assertIn('regexp(RequestK8SNamespace,"osg-test")', self.constraint()) + self.assertIn('(RequestGPUs=?=1)', self.constraint()) + self.assertIn('RequestCPUs', self.schedd_queries[-1][2]) + s, _ = self.schedd(force_k8s_namespace_matching="no") + s.query_idle() + self.assertNotIn('RequestK8SNamespace', self.constraint()) + + def test_one_failing_schedd_is_logged_and_skipped(self): + self.failing = {"ap2.example.org"} + s, log = self.schedd() + jobs = s.query_idle() + self.assertEqual([j['ScheddName'] for j in jobs], ["ap1.example.org"]) + self.assertIn("Failed to query HTCondor schedd 'ap2.example.org'", "\n".join(log.lines)) + + def test_all_schedds_failing_raises(self): + self.failing = {"ap1.example.org", "ap2.example.org"} + s, _ = self.schedd() + with self.assertRaises(Exception): + s.query_idle() + + def test_configs_have_exactly_one_container_condition(self): + for name, path in CONFIGS.items(): + f = configparser.ConfigParser() + f.read(path) + h = f['htcondor'] if 'htcondor' in f else f['DEFAULT'] + with self.subTest(config=name): + config = provisioner_htcondor.ProvisionerHTCConfig(name) + config.parse(h) + s = provisioner_htcondor.ProvisionerSchedd(Log(), {".*": ".*"}, config) + s.query_idle() + self.assertEqual(self.constraint().count("SingularityImage"), 1) + +if __name__ == '__main__': + unittest.main() From 62e3c9961ece45a386fdce02622bc37556bb15f7 Mon Sep 17 00:00:00 2001 From: Ashton Graves Date: Sat, 24 Oct 2026 17:26:22 -0500 Subject: [PATCH 2/6] Potential fix for pull request finding 'Omit empty node affinity to avoid matching no nodes' Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- osg-htc/nrp-provisioner/src/provisioner_k8s.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/osg-htc/nrp-provisioner/src/provisioner_k8s.py b/osg-htc/nrp-provisioner/src/provisioner_k8s.py index 85579eb6..76e95c3a 100644 --- a/osg-htc/nrp-provisioner/src/provisioner_k8s.py +++ b/osg-htc/nrp-provisioner/src/provisioner_k8s.py @@ -381,13 +381,13 @@ def submit(self, attrs, n_pods=1): 'restartPolicy': 'Never', 'tolerations' : tolerations, 'nodeSelector' : node_selectors, - "affinity": { + **({"affinity": { "nodeAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": { "nodeSelectorTerms": [ { "matchExpressions": node_affinity_selectors } ] } } - }, + }} if node_affinity_selectors else {}), 'containers': [{ 'name': 'htcondor', 'image': k8s_image, From 1479ebcce3cfe12f4fb4908511655f7a09a164e1 Mon Sep 17 00:00:00 2001 From: Ashton Graves Date: Sat, 24 Oct 2026 17:26:45 -0500 Subject: [PATCH 3/6] Potential fix for pull request finding 'Propagate background process failures from the worker script' Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- osg-htc/nrp-provisioner/provisioner_startup.sh | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/osg-htc/nrp-provisioner/provisioner_startup.sh b/osg-htc/nrp-provisioner/provisioner_startup.sh index 1ac6c626..ad65eb4d 100755 --- a/osg-htc/nrp-provisioner/provisioner_startup.sh +++ b/osg-htc/nrp-provisioner/provisioner_startup.sh @@ -35,7 +35,8 @@ trap 'echo signal received!; kill $(jobs -p); wait' SIGINT SIGTERM echo "`date` Starting provisioner_main.py" su -l provisioner -c "cd /home/provisioner && python3 provisioner_main.py /var/log/provisioner/logs/provisioner.log ${K8S_NAMESPACE} ${CVMFS_MOUNTS}" & -wait +wait "$!" rc=$? echo "`date` End of provisioner_main.py, rc=${rc}" +exit "$rc" From 19ba582b050f5ae2ebbb062dfb35339b23031ae2 Mon Sep 17 00:00:00 2001 From: Ashton Graves Date: Sat, 24 Oct 2026 17:26:55 -0500 Subject: [PATCH 4/6] Potential fix for pull request finding 'Use Ki units for HTCondor disk resource limits' Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- osg-htc/nrp-provisioner/src/provisioner_k8s.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/osg-htc/nrp-provisioner/src/provisioner_k8s.py b/osg-htc/nrp-provisioner/src/provisioner_k8s.py index 76e95c3a..d17a063b 100644 --- a/osg-htc/nrp-provisioner/src/provisioner_k8s.py +++ b/osg-htc/nrp-provisioner/src/provisioner_k8s.py @@ -268,8 +268,8 @@ def submit(self, attrs, n_pods=1): if eph_storage Date: Sat, 24 Oct 2026 19:06:04 -0500 Subject: [PATCH 5/6] nrp-provisioner: group jobs only on the attributes pods are created with Jobs were grouped by RequestDiskVolumes, RequestGPUTypes and RequestLabels too, but submit() never puts those on the pods. A job setting one would get pods every loop that its group never counted, with no cap and no back-off. Group as if they were unset: trial pods and the back-off then limit the cost if such a job's pods don't match it. No job in the deployed namespaces sets them today. Co-Authored-By: Claude Opus 5.5 --- osg-htc/nrp-provisioner/README.md | 1 + .../src/provisioner_clustering.py | 6 ++- .../nrp-provisioner/tests/test_clustering.py | 42 +++++++++++++++++++ 3 files changed, 48 insertions(+), 1 deletion(-) create mode 100644 osg-htc/nrp-provisioner/tests/test_clustering.py diff --git a/osg-htc/nrp-provisioner/README.md b/osg-htc/nrp-provisioner/README.md index f57e529b..bfa6adb5 100644 --- a/osg-htc/nrp-provisioner/README.md +++ b/osg-htc/nrp-provisioner/README.md @@ -278,6 +278,7 @@ nothing is. | File | What it covers | |---|---| | `test_backoff.py` | the back-off state machine, trial pods, and the event loop skipping paused groups | +| `test_clustering.py` | grouping with plain dictionaries: jobs that request GPU types, disk volumes or labels group with the pods made for them | | `test_htcondor_ads.py` | job/startd ad conversion (rounding, "too small" means unset, expressions, stringified values) and job grouping | | `test_schedd_query.py` | the schedd list (whitelist, identity filter, the location attributes `htcondor2.Schedd` needs), the query constraint (apptainer, namespace matching), failure handling, every config | | `test_k8s_submit.py` | Job submission through the **real** `kubernetes` client (only the network send is intercepted), for every config | diff --git a/osg-htc/nrp-provisioner/src/provisioner_clustering.py b/osg-htc/nrp-provisioner/src/provisioner_clustering.py index 45cd1ef5..7a479baf 100644 --- a/osg-htc/nrp-provisioner/src/provisioner_clustering.py +++ b/osg-htc/nrp-provisioner/src/provisioner_clustering.py @@ -36,6 +36,10 @@ def __init__(self): self.attributes['GPUs']=0 self.attributes['GPUTypes']='' self.attributes['Labels']='' + # Pods are only created with these (ProvisionerK8S.submit). DiskVolumes, GPUTypes and + # Labels are not implemented there, so pods never carry them: jobs that request them + # are grouped as if they didn't, or their group would never see its own pods. + self.supported_job_attributes=('CPUs','Memory','Disk','GPUs') def expand_schedd_attr(self, attr): return "Request%s"%attr @@ -178,7 +182,7 @@ def cluster_schedd_jobs(self, schedd_jobs): key_attrs={} for k in self.attrs.attributes.keys(): jobk = self.attrs.expand_schedd_attr(k) - if jobk in job.keys(): + if jobk in job.keys() and k in self.attrs.supported_job_attributes: val = job[jobk] else: val = self.attrs.attributes[k] diff --git a/osg-htc/nrp-provisioner/tests/test_clustering.py b/osg-htc/nrp-provisioner/tests/test_clustering.py new file mode 100644 index 00000000..c8465b6c --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_clustering.py @@ -0,0 +1,42 @@ +# +# Unit tests for grouping jobs and pods by shape, with plain dictionaries +# (no HTCondor bindings needed). +# +# Run from the provisioner directory: +# python3 -m unittest discover -s tests +# + +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')) + +from src import provisioner_clustering + +def job(**extra): + j = {'JobStatus': '1', 'RequestCPUs': '1', 'RequestMemory': '4096', 'RequestDisk': '8000000', 'RequestGPUs': '1'} + j.update(extra) + return j + +# what ProvisionerK8S.submit labels a pod with, for the job shape above +POD = {'Name': 'p1', 'Phase': 'Running', 'StartTime': None, 'HasContainers': True, + 'PodCPUs': '1', 'PodMemory': '4096', 'PodDisk': '8000000', 'PodGPUs': '1'} + +class TestUnsupportedAttributes(unittest.TestCase): + def setUp(self): + self.c = provisioner_clustering.ProvisionerClustering() + + def test_jobs_requesting_unsupported_attributes_group_with_their_pods(self): + jobs = [job(), job(RequestGPUTypes='NVIDIA-A10'), job(RequestLabels='a:b'), job(RequestDiskVolumes='pvc:/data')] + job_groups = self.c.cluster_schedd_jobs(jobs) + pod_groups = self.c.cluster_k8s_pods([POD], []) + self.assertEqual(list(job_groups.keys()), list(pod_groups.keys())) + self.assertEqual(job_groups[list(job_groups)[0]].count_idle(), 4) + + def test_supported_attributes_still_split_groups(self): + job_groups = self.c.cluster_schedd_jobs([job(), job(RequestGPUs='2'), job(RequestCPUs='4')]) + self.assertEqual(len(job_groups), 3) + +if __name__ == '__main__': + unittest.main() From 4edfe2640a4f3631c0d1fee5808ead7e01f76973 Mon Sep 17 00:00:00 2001 From: Ashton Graves Date: Sat, 24 Oct 2026 19:08:49 -0500 Subject: [PATCH 6/6] nrp-provisioner: don't let a preempted pod take down its Job's other pods Worker Jobs run several pods with backoffLimit 0, so one failed pod fails the whole Job and the Job controller deletes the others, claimed ones included. In osg-opportunistic, 17 such failures in an hour, all preemptions, deleted 141 sibling pods. Ignore pods stopped for preemption (DisruptionTarget) in a podFailurePolicy: the Job controller replaces them instead. Other failures still fail the Job. Co-Authored-By: Claude Opus 5.5 --- osg-htc/nrp-provisioner/src/provisioner_k8s.py | 10 ++++++++++ osg-htc/nrp-provisioner/tests/test_k8s_submit.py | 7 +++++++ 2 files changed, 17 insertions(+) diff --git a/osg-htc/nrp-provisioner/src/provisioner_k8s.py b/osg-htc/nrp-provisioner/src/provisioner_k8s.py index d17a063b..147407ad 100644 --- a/osg-htc/nrp-provisioner/src/provisioner_k8s.py +++ b/osg-htc/nrp-provisioner/src/provisioner_k8s.py @@ -372,6 +372,16 @@ def submit(self, attrs, n_pods=1): 'parallelism': n_pods, 'completions': n_pods, 'backoffLimit': 0, + # With backoffLimit 0, one failed pod fails the whole Job and the Job controller + # deletes its other pods, claimed ones included. Preemption (common at the + # opportunistic priority) isn't a worker failure: ignore it, and the Job + # controller just replaces the preempted pod. + 'podFailurePolicy': { + 'rules': [{ + 'action': 'Ignore', + 'onPodConditions': [{'type': 'DisruptionTarget'}] + }] + }, 'ttlSecondsAfterFinished': self.k8s_job_ttl, 'template': { 'metadata': { diff --git a/osg-htc/nrp-provisioner/tests/test_k8s_submit.py b/osg-htc/nrp-provisioner/tests/test_k8s_submit.py index 5f44acab..98e52c08 100644 --- a/osg-htc/nrp-provisioner/tests/test_k8s_submit.py +++ b/osg-htc/nrp-provisioner/tests/test_k8s_submit.py @@ -76,5 +76,12 @@ def test_quantities_are_strings(self): self.assertEqual(res['requests']['cpu'], "3") self.assertNotIn('nvidia.com/gpu', res['requests']) + def test_preemption_does_not_fail_the_job(self): + body = self.submit(self.make_k8s('sample', 'no'), "1", "0") + self.assertEqual(body['spec']['backoffLimit'], 0) + self.assertEqual(body['spec']['template']['spec']['restartPolicy'], 'Never') # required by podFailurePolicy + self.assertEqual(body['spec']['podFailurePolicy']['rules'], + [{'action': 'Ignore', 'onPodConditions': [{'type': 'DisruptionTarget'}]}]) + if __name__ == '__main__': unittest.main()