Feat: Submit and status reporting for HTCondor

.github/workflows/integration.yml

@@ -10,6 +10,7 @@ on:
       - ".github/workflows/integration.yml"
       - "tests/conftest.py"
       - "pyproject.toml"
+      - "src/**"
 
 permissions:
   contents: read
@@ -66,10 +67,12 @@ jobs:
             # TODO(#13/#14): ship the suite in the image and exec as an unprivileged user.
             runner=(docker compose -f "$COMPOSE_FILE" exec -T backend pytest)
           else
-            runner=(pixi run -e py313 pytest)
+            # Independent tests, run side by side (stacks.yml `workers`).
+            runner=(pixi run -e py313 pytest -n "${{ matrix.workers }}")
           fi
           set +e
-          "${runner[@]}" tests/integration -m "${{ matrix.markers }}" \
+          # -rs --no-fold-skipped: name every skipped test and why.
+          "${runner[@]}" tests/integration -m "${{ matrix.markers }}" -rs --no-fold-skipped \
             --stack="${{ matrix.stack }}" --config="${{ matrix.config }}"
           rc=$?
           # Exit 5 = no tests selected; tolerated until the contract suite lands.

docs/dev/how-to/add-an-integration-stack.md

@@ -24,7 +24,7 @@ You do not need to change any CI workflow. CI finds the stack in `stacks.yml`.
 
    ```sh
    docker compose -f tests/integration/stacks/<id>/compose.yml up --wait
-   pixi run -e py313 pytest tests/integration -m "<markers>" --stack=<id> --config=basic
+   pixi run -e py313 pytest tests/integration -n 4 -m "<markers>" --stack=<id> --config=basic
    pixi run -e py313 pytest tests/integration/test_manifest.py
    ```
 

docs/dev/how-to/run-integration-tests-locally.md

@@ -16,9 +16,12 @@ CI runs the same commands as below, so you can reproduce any CI job on your mach
 2. Run the tests. Copy `<markers>` from the stack's entry in `stacks.yml`:
 
    ```sh
-   pixi run -e py313 pytest tests/integration -m "<markers>" --stack=<id> --config=basic
+   pixi run -e py313 pytest tests/integration -n 4 -m "<markers>" --stack=<id> --config=basic
    ```
 
+`-n 4` runs four tests at the same time, as CI does. Use the stack's `workers` value from
+`stacks.yml` if it sets one, or leave `-n` out to run one test at a time while debugging.
+
 For stacks with `exec_in_container: true` (the `local-*` stacks), run the tests inside the
 container instead:
 

docs/dev/reference/integration-stacks.md

@@ -44,6 +44,7 @@ tests/integration/
 | `versions` | no | Backend versions to test, for example `[lts, latest]`. Default: `[latest]`. |
 | `markers` | yes | `pytest -m` expression that selects the tests for this stack. |
 | `exec_in_container` | no | `true`: run pytest inside the backend container. Default: `false`. |
+| `workers` | no | How many tests run at the same time (pytest-xdist). Default: `4`. Set `1` for a backend that cannot take several jobs at once. |
 
 ## When a stack gets CI jobs
 

docs/user/reference/api.md

@@ -19,6 +19,14 @@ Anything else, including every name starting with `_`, is internal.
 
 ::: intercede.registry
 
+## Credentials
+
+::: intercede.auth.models
+
+::: intercede.auth.protocols
+
+::: intercede.auth.providers
+
 ## Exceptions
 
 ::: intercede.exceptions

src/intercede/_backends/_htcondor/__init__.py

@@ -0,0 +1,11 @@
+"""What interCEde knows about HTCondor, shared by every backend that drives it.
+
+Plain functions and data only, with no dependency on any backend class: the
+HTCondor-CE backend uses them directly, and the HTCondor scheduler of the
+composed backend (issue #21) will use the same ones over its transport. They
+share these functions, never a subtype.
+
+- ``submit``: a submit description built from a ``SubmissionSpec``
+- ``commands``: the argv of each HTCondor tool, and parsers for their output
+- ``status``: the state mapping from job ads to ``JobStatus``
+"""

src/intercede/_backends/_htcondor/commands.py

@@ -0,0 +1,117 @@
+"""The argv of each HTCondor tool, and parsers for what they print.
+
+``remote`` is how a tool reaches its schedd: ``schedd_options(pool, name)``
+for a remote schedd, as an HTCondor-CE is reached, or ``()`` for the schedd
+of the host the tool runs on, as the composed backend reaches it.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Sequence
+
+from intercede.models import JobID
+
+# Ids per command line, so a status poll of thousands of jobs stays bounded.
+CHUNK = 100
+
+
+def schedd_options(pool: str, name: str) -> list[str]:
+    """Reach the schedd ``name`` through the collector ``pool`` (host:port)."""
+    return ["-pool", pool, "-name", name]
+
+
+def submit(
+    description: str, pool: str | None = None, name: str | None = None
+) -> list[str]:
+    """Submit a description file and spool its inputs to the schedd.
+
+    Spooling copies the inputs at submit time, so nothing on the submitting
+    host is needed afterwards. With ``pool`` and ``name`` the schedd is a
+    remote one (``-remote`` implies spooling); without, it is the local one.
+    """
+    if (pool is None) != (name is None):
+        raise ValueError("a remote schedd needs both pool and name")
+
+    command = ["condor_submit", "-terse"]
+
+    if pool is None or name is None:
+        command.append("-spool")
+    else:
+        command.extend(["-pool", pool, "-remote", name])
+
+    command.append(description)
+    return command
+
+
+def query(
+    ids: Sequence[JobID], attributes: Sequence[str], remote: Sequence[str]
+) -> list[str]:
+    """Read the ads of jobs still in the queue."""
+    return ["condor_q", *remote, *ids, "-attributes", ",".join(attributes), "-json"]
+
+
+def history(
+    ids: Sequence[JobID], attributes: Sequence[str], remote: Sequence[str]
+) -> list[str]:
+    """Read the ads of jobs that have left the queue."""
+    return [
+        "condor_history",
+        *remote,
+        *ids,
+        "-attributes",
+        ",".join(attributes),
+        "-json",
+    ]
+
+
+def remove(ids: Sequence[JobID], remote: Sequence[str]) -> list[str]:
+    """Remove jobs: cancels active ones, releases finished ones."""
+    return ["condor_rm", *remote, *ids]
+
+
+def transfer_data(ids: Sequence[JobID], remote: Sequence[str]) -> list[str]:
+    """Fetch the spooled output of jobs into the directory they were submitted from."""
+    return ["condor_transfer_data", *remote, *ids]
+
+
+def chunks(ids: Sequence[JobID]) -> list[Sequence[JobID]]:
+    """Split ids into groups of at most ``CHUNK``."""
+    return [ids[start : start + CHUNK] for start in range(0, len(ids), CHUNK)]
+
+
+_TERSE = re.compile(
+    r"\s*(?P<cluster>\d+)\.(?P<first>\d+)"
+    r"\s*-\s*"
+    r"(?P<last_cluster>\d+)\.(?P<last>\d+)\s*"
+)
+
+
+def parse_submit(raw: str) -> list[JobID]:
+    """Parse job IDs from condor_submit -terse output, e.g. '12.0 - 12.3'."""
+    lines = [line for line in raw.splitlines() if line.strip()]
+
+    if len(lines) != 1:
+        raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}")
+
+    match = _TERSE.fullmatch(lines[0])
+    if match is None:
+        raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}")
+
+    cluster = int(match["cluster"])
+    first = int(match["first"])
+    last_cluster = int(match["last_cluster"])
+    last = int(match["last"])
+
+    if cluster != last_cluster or last < first:
+        raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}")
+
+    return [JobID(f"{cluster}.{proc}") for proc in range(first, last + 1)]
+
+
+_REMOVED = re.compile(r"^Job (\d+\.\d+) marked for removal", re.MULTILINE)
+
+
+def parse_remove(raw: str) -> set[JobID]:
+    """Return the ids ``condor_rm`` reports as removed; any other id was not."""
+    return {JobID(job) for job in _REMOVED.findall(raw)}

src/intercede/_backends/_htcondor/status.py

@@ -0,0 +1,124 @@
+"""From HTCondor job ads to ``JobStatus``.
+
+Reading only: the state is taken from what HTCondor records about a job, never
+from a policy interCEde sets on it. A finished job reads as FAILED when its
+payload exited non-zero or was killed by a signal, from ``ExitCode`` and
+``ExitBySignal``.
+
+Ported from DIRAC's ``BatchSystems/Condor.py`` (``STATES_MAP`` and
+``getCondorStatus``), with differences: DIRAC puts a payload that fails on
+hold to recognise it, where the exit code is read here instead; the states
+DIRAC leaves unmapped (6, 7) are mapped; and a removed job reports what it did
+before it was removed, so a purged job keeps its result.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from intercede.models import JobID, JobStatus
+
+# The ad attributes every status query asks for.
+STATE_ATTRIBUTES: Sequence[str] = (
+    "ClusterId",
+    "ProcId",
+    "JobStatus",
+    "LastJobStatus",
+    "ExitCode",
+    "ExitBySignal",
+    "HoldReasonCode",
+    "HoldReason",
+)
+
+# HTCondor's JobStatus codes.
+_IDLE, _RUNNING, _REMOVED, _COMPLETED, _HELD, _TRANSFERRING, _SUSPENDED = range(1, 8)
+
+_STATES: Mapping[int, JobStatus] = {
+    _IDLE: JobStatus.WAITING,
+    _RUNNING: JobStatus.RUNNING,
+    _TRANSFERRING: JobStatus.RUNNING,  # transferring output after running
+    _SUSPENDED: JobStatus.WAITING,
+}
+
+_HOLD_SPOOLING = 16  # HoldReasonCode: input files are still being spooled
+
+
+def job_id(ad: Mapping[str, Any]) -> JobID:
+    """Return the ``cluster.proc`` id of a job ad."""
+    return JobID(f"{ad['ClusterId']}.{ad['ProcId']}")
+
+
+def job_status(ad: Mapping[str, Any]) -> JobStatus:
+    """Map one job ad to the common status."""
+    code = ad.get("JobStatus")
+    if code == _COMPLETED:
+        return _result(ad)
+    if code == _REMOVED:
+        # purge() removes finished jobs to release them, after which the
+        # history only says "removed": the state it left tells the result.
+        return (
+            _result(ad) if ad.get("LastJobStatus") == _COMPLETED else JobStatus.ABORTED
+        )
+    if code == _HELD:
+        # Held by the site or by HTCondor, not by interCEde: the job is stuck,
+        # except while its inputs are still being spooled.
+        return (
+            JobStatus.WAITING
+            if ad.get("HoldReasonCode") == _HOLD_SPOOLING
+            else JobStatus.ABORTED
+        )
+    return (
+        _STATES.get(code, JobStatus.UNKNOWN)
+        if isinstance(code, int)
+        else JobStatus.UNKNOWN
+    )
+
+
+def is_removed(ad: Mapping[str, Any]) -> bool:
+    """Tell whether a job was removed, even while it is still in the queue.
+
+    ``condor_rm`` only marks a job for removal: the schedd drops it from the
+    queue a moment later. In between it is already released.
+
+    Args:
+        ad: A job ad with ``JobStatus``.
+
+    Returns:
+        True when the job has been removed.
+
+    """
+    return ad.get("JobStatus") == _REMOVED
+
+
+def _result(ad: Mapping[str, Any]) -> JobStatus:
+    """Tell DONE from FAILED for a job whose payload has finished."""
+    if ad.get("ExitBySignal") is True:
+        return JobStatus.FAILED
+    exit_code = ad.get("ExitCode")
+    return (
+        JobStatus.FAILED
+        if isinstance(exit_code, int) and exit_code != 0
+        else JobStatus.DONE
+    )
+
+
+def hold_reason(ad: Mapping[str, Any]) -> str | None:
+    """Return why a held job is held, if it is."""
+    reason = ad.get("HoldReason")
+    return reason if ad.get("JobStatus") == _HELD and isinstance(reason, str) else None
+
+
+def parse_ads(raw: str) -> list[dict[str, Any]]:
+    """Parse ``condor_q -json`` or ``condor_history -json`` output.
+
+    ``condor_q`` prints nothing at all when no job matches, and
+    ``condor_history`` prints an empty list.
+    """
+    if not raw.strip():
+        return []
+    ads = json.loads(raw)
+    if not isinstance(ads, list) or not all(isinstance(ad, dict) for ad in ads):
+        raise ValueError("expected a JSON list of job ads")
+    return ads