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..bfa6adb5 --- /dev/null +++ b/osg-htc/nrp-provisioner/README.md @@ -0,0 +1,330 @@ +# 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_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 | +| `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..ad65eb4d --- /dev/null +++ b/osg-htc/nrp-provisioner/provisioner_startup.sh @@ -0,0 +1,42 @@ +#!/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}" +exit "$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..7a479baf --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_clustering.py @@ -0,0 +1,244 @@ +# +# 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']='' + # 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 + + 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() and k in self.attrs.supported_job_attributes: + 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..147407ad --- /dev/null +++ b/osg-htc/nrp-provisioner/src/provisioner_k8s.py @@ -0,0 +1,568 @@ +# +# 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_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() 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..98e52c08 --- /dev/null +++ b/osg-htc/nrp-provisioner/tests/test_k8s_submit.py @@ -0,0 +1,87 @@ +# +# 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']) + + 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() 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()