Coverage for gco / services / api_routes / slurm.py: 100.00%
330 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Slurm endpoints — the authenticated ``/api/v1/slurm/*`` surface.
3The manifest API is the cluster's authenticated ingress (HMAC middleware plus
4the IAM-authorized API Gateway in front), so Slurm is exposed here and
5proxied to slurmrestd inside ``gco-jobs`` through :mod:`gco.services.slurm_client`.
6Callers never reach slurmrestd or hold a Slurm credential: the API acts as
7Slurm's own user with an operator-minted token and stamps the submitting
8caller into each job's ``comment`` for accounting attribution.
10- ``GET /api/v1/slurm/status`` — controller reachability, partitions, node and job counts
11- ``GET /api/v1/slurm/partitions`` — partitions with their state, limits and node counts
12- ``GET /api/v1/slurm/nodes`` — Slurm nodes (the slurmd worker pods) with state and resources
13- ``GET /api/v1/slurm/jobs`` — the queue (filter by state, partition, name)
14- ``POST /api/v1/slurm/jobs`` — submit a batch script (sbatch)
15- ``GET /api/v1/slurm/jobs/{job_id}`` — one job (from slurmdbd once slurmctld forgets it)
16- ``DELETE /api/v1/slurm/jobs/{job_id}`` — cancel (scancel)
17- ``GET /api/v1/slurm/accounting/jobs`` — slurmdbd job history (sacct)
19When Slurm is not deployed (``helm.slurm.enabled`` false) every route answers
20404 with the remedy; an unreachable slurmrestd or missing token answers 503.
21"""
23from __future__ import annotations
25import logging
26import re
27from datetime import UTC, datetime
28from typing import Any
30from fastapi import APIRouter, HTTPException, Query, Request
31from fastapi.responses import JSONResponse, Response
32from pydantic import BaseModel, Field, field_validator
34from gco.services import slurm_client
35from gco.services.structured_logging import sanitize_log_value
36from gco.slurm_config import SLURM_DEFAULT_WORKING_DIRECTORY
38router = APIRouter(prefix="/api/v1/slurm", tags=["Slurm"])
39logger = logging.getLogger(__name__)
41#: Batch scripts are limited well under slurmrestd's request limits; a job that
42#: needs more belongs in a container image or on the shared filesystem.
43MAX_SCRIPT_BYTES = 64 * 1024
44#: Slurm job names: printable, no whitespace control characters.
45_JOB_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:+-]{0,127}$")
46_PARTITION_RE = re.compile(r"^[a-z][a-z0-9-]{0,31}$")
47_TIME_LIMIT_RE = re.compile(
48 r"^(UNLIMITED|INFINITE|\d+|\d+:\d{2}|\d+:\d{2}:\d{2}|\d+-\d{1,2}|\d+-\d{1,2}:\d{2}|\d+-\d{1,2}:\d{2}:\d{2})$"
49)
50_ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
51#: Headers the caller's identity may arrive in: the VPC proxy Lambda forwards
52#: the SigV4 principal of the API Gateway request as ``x-gco-caller`` when it
53#: knows it; otherwise the comment records the API itself.
54_CALLER_HEADERS = ("x-gco-caller", "x-gco-caller-arn")
57class SlurmJobSubmission(BaseModel):
58 """Request body for ``POST /api/v1/slurm/jobs`` (the sbatch equivalent)."""
60 script: str = Field(..., description="The batch script, starting with a #! line")
61 name: str = Field("gco-job", description="Slurm job name")
62 partition: str | None = Field(None, description="Partition (default: the cluster default)")
63 nodes: int = Field(1, ge=1, le=1024, description="Node count (--nodes)")
64 tasks: int = Field(1, ge=1, le=65536, description="Task count (--ntasks)")
65 cpus_per_task: int = Field(1, ge=1, le=1024, description="CPUs per task (--cpus-per-task)")
66 gpus_per_node: int = Field(0, ge=0, le=64, description="GPUs per node (--gpus-per-node)")
67 memory_mb: int | None = Field(None, ge=1, description="Memory per node in MiB (--mem)")
68 time_limit: str | None = Field(None, description="Wall time (--time), e.g. 01:00:00")
69 working_directory: str = Field(
70 SLURM_DEFAULT_WORKING_DIRECTORY,
71 description="Working directory on the worker (default: the slurm user's home on the "
72 "shared filesystem, where slurm-<jobid>.out lands)",
73 )
74 environment: dict[str, str] = Field(default_factory=dict, description="Extra environment")
75 comment: str | None = Field(None, max_length=512, description="Free-text comment")
76 exclusive: bool = Field(False, description="Whole-node allocation (--exclusive)")
78 @field_validator("script")
79 @classmethod
80 def _script_shape(cls, value: str) -> str:
81 if not value.strip():
82 raise ValueError("script must not be empty")
83 if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
84 raise ValueError(f"script exceeds {MAX_SCRIPT_BYTES} bytes")
85 if not value.lstrip().startswith("#!"):
86 raise ValueError("script must start with an interpreter line (#!/bin/bash)")
87 return value
89 @field_validator("name")
90 @classmethod
91 def _name_shape(cls, value: str) -> str:
92 if not _JOB_NAME_RE.match(value):
93 raise ValueError("name must be 1-128 characters of letters, digits, _ . : + -")
94 return value
96 @field_validator("partition")
97 @classmethod
98 def _partition_shape(cls, value: str | None) -> str | None:
99 if value is not None and not _PARTITION_RE.match(value):
100 raise ValueError("partition must be a lowercase partition name")
101 return value
103 @field_validator("time_limit")
104 @classmethod
105 def _time_shape(cls, value: str | None) -> str | None:
106 if value is not None and not _TIME_LIMIT_RE.match(value.strip()):
107 raise ValueError("time_limit must be a Slurm time (minutes, HH:MM:SS, D-HH:MM:SS)")
108 return value.strip() if value else value
110 @field_validator("working_directory")
111 @classmethod
112 def _cwd_shape(cls, value: str) -> str:
113 if not value.startswith("/") or "\n" in value:
114 raise ValueError("working_directory must be an absolute path")
115 return value
117 @field_validator("environment")
118 @classmethod
119 def _environment_shape(cls, value: dict[str, str]) -> dict[str, str]:
120 for key, item in value.items():
121 if not _ENV_NAME_RE.match(key):
122 raise ValueError(f"environment variable name {key!r} is not valid")
123 if "\n" in item or "\x00" in item:
124 raise ValueError(f"environment variable {key!r} must be a single-line value")
125 if len(value) > 64:
126 raise ValueError("at most 64 environment variables")
127 return value
130def _client() -> slurm_client.SlurmRestClient:
131 try:
132 return slurm_client.get_client()
133 except slurm_client.SlurmDisabledError as exc:
134 raise slurm_client.http_exception_for(exc) from exc
137def _raise(error: Exception, context: str) -> HTTPException:
138 """Translate a client error, logging the ones that point at the deployment.
140 4xx answers are the caller's to read (an unknown job, a bad partition);
141 5xx ones are logged with the underlying cause so the operator can find
142 slurmrestd or the token problem the caller only sees as 503/500.
143 """
144 http_error = slurm_client.http_exception_for(error, context=context)
145 if http_error.status_code >= 500:
146 # ``context`` can carry a path parameter (the job id), so it is
147 # sanitized like the error text even though the router already
148 # validated it as an integer.
149 logger.warning(
150 "Slurm %s failed (HTTP %s): %s: %s",
151 sanitize_log_value(context),
152 http_error.status_code,
153 type(error).__name__,
154 sanitize_log_value(str(error)),
155 )
156 return http_error
159def _envelope(payload: dict[str, Any]) -> dict[str, Any]:
160 return {"timestamp": datetime.now(UTC).isoformat(), **payload}
163def _number(value: Any) -> int | None:
164 """slurmrestd wraps many numbers as {set, infinite, number}; unwrap them."""
165 if isinstance(value, dict):
166 if value.get("infinite"):
167 return None
168 return int(value["number"]) if value.get("set") and "number" in value else None
169 if isinstance(value, bool) or value is None:
170 return None
171 try:
172 return int(value)
173 except TypeError, ValueError:
174 return None
177def _states(value: Any) -> list[str]:
178 if isinstance(value, list):
179 return [str(item) for item in value]
180 if isinstance(value, str):
181 return [value]
182 return []
185def _mapping(value: Any) -> dict[str, Any]:
186 """The value when it is a JSON object, else an empty one (sparse payloads)."""
187 return value if isinstance(value, dict) else {}
190def _summarize_partition(item: dict[str, Any]) -> dict[str, Any]:
191 nodes = _mapping(item.get("nodes"))
192 partition_state = _mapping(item.get("partition"))
193 maximums = _mapping(item.get("maximums"))
194 cpus = item.get("cpus")
195 states = _states(partition_state.get("state"))
196 return {
197 "name": item.get("name"),
198 "state": states,
199 "default": "DEFAULT" in {s.upper() for s in states},
200 "total_nodes": _number(nodes.get("total")),
201 "node_names": nodes.get("configured") or nodes.get("allowed_allocation") or "",
202 "max_time_minutes": _number(maximums.get("time")),
203 "total_cpus": _number(cpus.get("total")) if isinstance(cpus, dict) else None,
204 }
207#: The slurm-operator drains a worker before KEDA removes its pod and, in
208#: StatefulSet scaling mode, leaves the node record behind as DOWN+DRAIN with
209#: this reason prefix until the next scale-up re-registers it (observed live).
210_SCALED_IN_REASON_PREFIX = "slurm-operator:"
213def _scaled_in(item: dict[str, Any]) -> bool:
214 """A node whose only problem is that its pod was scaled away."""
215 states = {s.upper() for s in _states(item.get("state"))}
216 reason = str(item.get("reason") or "")
217 return "DRAIN" in states and "DOWN" in states and reason.startswith(_SCALED_IN_REASON_PREFIX)
220def _summarize_node(item: dict[str, Any]) -> dict[str, Any]:
221 return {
222 "name": item.get("name"),
223 "hostname": item.get("hostname"),
224 "state": _states(item.get("state")),
225 "scaled_in": _scaled_in(item),
226 "partitions": list(item.get("partitions") or []),
227 "cpus": _number(item.get("cpus")),
228 "cpus_allocated": _number(item.get("alloc_cpus")),
229 "real_memory_mb": _number(item.get("real_memory")),
230 "memory_allocated_mb": _number(item.get("alloc_memory")),
231 "gres": item.get("gres") or "",
232 "gres_used": item.get("gres_used") or "",
233 "reason": item.get("reason") or "",
234 }
237def _summarize_job(item: dict[str, Any]) -> dict[str, Any]:
238 state = item.get("job_state")
239 return {
240 "job_id": item.get("job_id"),
241 "name": item.get("name"),
242 "user": item.get("user_name"),
243 "account": item.get("account"),
244 "partition": item.get("partition"),
245 "state": _states(state),
246 "state_reason": item.get("state_reason"),
247 "nodes": item.get("nodes") or "",
248 "node_count": _number(item.get("node_count")),
249 "cpus": _number(item.get("cpus")),
250 "tres_requested": item.get("tres_req_str") or "",
251 "tres_allocated": item.get("tres_alloc_str") or "",
252 "submit_time": _number(item.get("submit_time")),
253 "start_time": _number(item.get("start_time")),
254 "end_time": _number(item.get("end_time")),
255 "time_limit_minutes": _number(item.get("time_limit")),
256 "exit_code": _exit_code(item.get("exit_code")),
257 "comment": item.get("comment") or "",
258 "working_directory": item.get("current_working_directory"),
259 "standard_output": _output_path(item, item.get("standard_output")),
260 "standard_error": _output_path(item, item.get("standard_error")),
261 }
264def _output_path(item: dict[str, Any], pattern: Any) -> str | None:
265 """slurmctld reports stdout/stderr as the submitted pattern (``slurm-%j.out``,
266 relative to the working directory); expand the common filename patterns and
267 anchor the path so ``gco slurm output`` can read it from the login pod."""
268 if not isinstance(pattern, str) or not pattern:
269 return None
270 replacements = {
271 "%j": item.get("job_id"),
272 "%J": item.get("job_id"),
273 "%u": item.get("user_name"),
274 "%x": item.get("name"),
275 }
276 array = item.get("array_job_id"), item.get("array_task_id")
277 if _number(array[0]) and _number(array[1]) is not None:
278 replacements["%A"] = _number(array[0])
279 replacements["%a"] = _number(array[1])
280 expanded = pattern.replace("%%", "\x00")
281 for token, value in replacements.items():
282 if value is not None:
283 expanded = expanded.replace(token, str(value))
284 expanded = expanded.replace("\x00", "%")
285 if not expanded.startswith("/"):
286 cwd = str(item.get("current_working_directory") or "").rstrip("/")
287 expanded = f"{cwd}/{expanded}" if cwd else expanded
288 return expanded
291def _exit_code(value: Any) -> dict[str, Any] | None:
292 if not isinstance(value, dict):
293 return None
294 return {
295 "status": _states(value.get("status")),
296 "return_code": _number(value.get("return_code")),
297 }
300def _summarize_accounting_job(item: dict[str, Any]) -> dict[str, Any]:
301 state = _mapping(item.get("state"))
302 time = _mapping(item.get("time"))
303 exit_code = _mapping(item.get("exit_code"))
304 tres = _mapping(item.get("tres"))
305 comment = item.get("comment")
306 return {
307 "job_id": item.get("job_id"),
308 "name": item.get("name"),
309 "user": item.get("user"),
310 "account": item.get("account"),
311 "partition": item.get("partition"),
312 "cluster": item.get("cluster"),
313 "state": _states(state.get("current")),
314 "state_reason": state.get("reason"),
315 "nodes": item.get("nodes") or "",
316 "submit_time": _number(time.get("submission")),
317 "start_time": _number(time.get("start")),
318 "end_time": _number(time.get("end")),
319 "elapsed_seconds": _number(time.get("elapsed")),
320 "time_limit_minutes": _number(time.get("limit")),
321 "exit_code": {
322 "status": _states(exit_code.get("status")),
323 "return_code": _number(exit_code.get("return_code")),
324 },
325 "tres_allocated": tres.get("allocated") or [],
326 "comment": comment.get("job") if isinstance(comment, dict) else (comment or ""),
327 }
330def _accounting_record_as_job(item: dict[str, Any]) -> dict[str, Any]:
331 """The slurmctld job shape, filled from a slurmdbd record.
333 slurmctld forgets a finished job after ``MinJobAge`` (ten minutes here);
334 slurmdbd keeps it, including the expanded stdout/stderr paths and the
335 working directory that ``gco slurm output`` needs. Fields slurmctld alone
336 knows (``state_reason`` for a pending job, ``tres_requested``) are what the
337 record offers, and ``source`` says which daemon answered.
338 """
339 record = _summarize_accounting_job(item)
340 tres = ",".join(
341 f"{entry.get('type')}={entry.get('count')}"
342 for entry in record["tres_allocated"]
343 if isinstance(entry, dict) and entry.get("type")
344 )
345 return {
346 "job_id": record["job_id"],
347 "name": record["name"],
348 "user": record["user"],
349 "account": record["account"],
350 "partition": record["partition"],
351 "state": record["state"],
352 "state_reason": record["state_reason"],
353 "nodes": record["nodes"],
354 "node_count": _number(item.get("allocation_nodes")),
355 "cpus": _number(_mapping(item.get("required")).get("CPUs")),
356 "tres_requested": tres,
357 "tres_allocated": tres,
358 "submit_time": record["submit_time"],
359 "start_time": record["start_time"],
360 "end_time": record["end_time"],
361 "time_limit_minutes": record["time_limit_minutes"],
362 "exit_code": record["exit_code"],
363 "comment": record["comment"],
364 "working_directory": item.get("working_directory"),
365 "standard_output": item.get("stdout_expanded") or None,
366 "standard_error": item.get("stderr_expanded") or None,
367 "source": "slurmdbd",
368 }
371def _ping_up(ping: dict[str, Any]) -> bool:
372 """slurmrestd reports controller reachability as ``pinged: "UP"`` (older
373 plugin versions) or ``responding: true`` (newer ones)."""
374 if ping.get("responding") is True:
375 return True
376 return str(ping.get("pinged") or "").strip().upper() == "UP"
379def _caller_identity(request: Request) -> str:
380 for header in _CALLER_HEADERS:
381 value = request.headers.get(header)
382 if value:
383 return value.strip()[:200]
384 return "gco-api"
387@router.get("/status")
388async def slurm_status() -> Response:
389 """Controller reachability plus partition, node and job counts for this region."""
390 client = _client()
391 try:
392 ping = await client.ping()
393 partitions = await client.partitions()
394 nodes = await client.nodes()
395 jobs = await client.jobs()
396 except Exception as exc:
397 raise _raise(exc, "reading status") from exc
398 node_items = [n for n in nodes.get("nodes", []) if isinstance(n, dict)]
399 job_items = [j for j in jobs.get("jobs", []) if isinstance(j, dict)]
400 job_states: dict[str, int] = {}
401 for job in job_items:
402 for state in _states(job.get("job_state")) or ["UNKNOWN"]:
403 job_states[state] = job_states.get(state, 0) + 1
404 node_states: dict[str, int] = {}
405 scaled_in = 0
406 for node in node_items:
407 if _scaled_in(node):
408 scaled_in += 1
409 continue
410 for state in _states(node.get("state")) or ["UNKNOWN"]:
411 node_states[state] = node_states.get(state, 0) + 1
412 pings = [p for p in ping.get("pings", []) if isinstance(p, dict)]
413 return JSONResponse(
414 status_code=200,
415 content=_envelope(
416 {
417 "enabled": True,
418 "controller": [
419 {
420 "hostname": p.get("hostname"),
421 "responding": _ping_up(p),
422 "latency_us": _number(p.get("latency")),
423 "mode": p.get("mode"),
424 }
425 for p in pings
426 ],
427 "partitions": [
428 _summarize_partition(p)
429 for p in partitions.get("partitions", [])
430 if isinstance(p, dict)
431 ],
432 "nodes": {
433 "total": len(node_items),
434 "by_state": node_states,
435 "scaled_in": scaled_in,
436 },
437 "jobs": {"total": len(job_items), "by_state": job_states},
438 "api_version": client.version,
439 }
440 ),
441 )
444@router.get("/partitions")
445async def slurm_partitions() -> Response:
446 """Every Slurm partition with its state, time limit and node count."""
447 client = _client()
448 try:
449 payload = await client.partitions()
450 except Exception as exc:
451 raise _raise(exc, "listing partitions") from exc
452 return JSONResponse(
453 status_code=200,
454 content=_envelope(
455 {
456 "partitions": [
457 _summarize_partition(p)
458 for p in payload.get("partitions", [])
459 if isinstance(p, dict)
460 ]
461 }
462 ),
463 )
466@router.get("/nodes")
467async def slurm_nodes(
468 partition: str | None = Query(None, description="Only nodes in this partition"),
469) -> Response:
470 """Slurm nodes (the slurmd worker pods) with state, resources and gres."""
471 client = _client()
472 try:
473 payload = await client.nodes()
474 except Exception as exc:
475 raise _raise(exc, "listing nodes") from exc
476 nodes = [_summarize_node(n) for n in payload.get("nodes", []) if isinstance(n, dict)]
477 if partition:
478 nodes = [n for n in nodes if partition in n["partitions"]]
479 return JSONResponse(status_code=200, content=_envelope({"nodes": nodes}))
482@router.get("/jobs")
483async def slurm_jobs(
484 state: str | None = Query(None, description="Job state filter (PENDING, RUNNING, ...)"),
485 partition: str | None = Query(None, description="Partition filter"),
486 name: str | None = Query(None, description="Exact job name filter"),
487 limit: int = Query(200, ge=1, le=5000, description="Maximum jobs returned"),
488) -> Response:
489 """The live queue as slurmctld sees it (pending, running and recently finished jobs)."""
490 client = _client()
491 try:
492 payload = await client.jobs()
493 except Exception as exc:
494 raise _raise(exc, "listing jobs") from exc
495 jobs = [_summarize_job(j) for j in payload.get("jobs", []) if isinstance(j, dict)]
496 if state:
497 wanted = state.upper()
498 jobs = [j for j in jobs if wanted in {s.upper() for s in j["state"]}]
499 if partition:
500 jobs = [j for j in jobs if j["partition"] == partition]
501 if name:
502 jobs = [j for j in jobs if j["name"] == name]
503 return JSONResponse(
504 status_code=200, content=_envelope({"jobs": jobs[:limit], "total": len(jobs)})
505 )
508@router.post("/jobs", status_code=201)
509async def slurm_submit(submission: SlurmJobSubmission, request: Request) -> Response:
510 """Submit a batch script to Slurm (sbatch through slurmrestd)."""
511 client = _client()
512 caller = _caller_identity(request)
513 comment = f"gco:{caller}"
514 if submission.comment:
515 comment = f"{comment} {submission.comment}"
516 # slurmrestd's job_desc_msg (v0.0.44): environment is a list of NAME=VALUE
517 # strings, a whole-node request is shared=none (sbatch --exclusive), and
518 # PATH must be given because the batch step inherits no environment.
519 environment = {"PATH": "/usr/local/bin:/usr/bin:/bin", **submission.environment}
520 job: dict[str, Any] = {
521 "name": submission.name,
522 "nodes": str(submission.nodes),
523 "tasks": submission.tasks,
524 "cpus_per_task": submission.cpus_per_task,
525 "current_working_directory": submission.working_directory,
526 "comment": comment[:1024],
527 "environment": [f"{key}={value}" for key, value in environment.items()],
528 }
529 if submission.partition:
530 job["partition"] = submission.partition
531 if submission.gpus_per_node:
532 job["tres_per_node"] = f"gres/gpu:{submission.gpus_per_node}"
533 if submission.memory_mb is not None:
534 job["memory_per_node"] = {"set": True, "number": submission.memory_mb}
535 if submission.time_limit:
536 job["time_limit"] = _time_limit_minutes(submission.time_limit)
537 if submission.exclusive:
538 job["shared"] = ["none"]
539 try:
540 payload = await client.submit(job, submission.script)
541 except Exception as exc:
542 raise _raise(exc, "submitting job") from exc
543 job_id = payload.get("job_id")
544 logger.info(
545 "Submitted Slurm job %s (%s) for %s",
546 job_id,
547 sanitize_log_value(submission.name),
548 sanitize_log_value(caller),
549 )
550 return JSONResponse(
551 status_code=201,
552 content=_envelope(
553 {
554 "job_id": job_id,
555 "step_id": payload.get("step_id"),
556 "name": submission.name,
557 "partition": submission.partition,
558 "submitted_by": caller,
559 "message": payload.get("job_submit_user_msg") or "",
560 }
561 ),
562 )
565def _time_limit_minutes(value: str) -> dict[str, Any]:
566 """Convert a Slurm time string to the minutes slurmrestd's time_limit wants."""
567 text = value.strip().upper()
568 if text in {"UNLIMITED", "INFINITE"}:
569 return {"set": True, "infinite": True}
570 days = 0
571 if "-" in text:
572 day_part, text = text.split("-", 1)
573 days = int(day_part)
574 parts = [int(p) for p in text.split(":")]
575 hours, minutes, seconds = [*parts, 0, 0, 0][:3]
576 total = days * 1440 + hours * 60 + minutes + (1 if seconds else 0)
577 else:
578 parts = [int(p) for p in text.split(":")]
579 if len(parts) == 1:
580 total = parts[0]
581 elif len(parts) == 2:
582 total = parts[0] + (1 if parts[1] else 0)
583 else:
584 hours, minutes, seconds = parts[:3]
585 total = hours * 60 + minutes + (1 if seconds else 0)
586 return {"set": True, "number": max(total, 1)}
589@router.get("/jobs/{job_id}")
590async def slurm_job(job_id: int) -> Response:
591 """One job: slurmctld's live view, else slurmdbd's record once the controller forgot it.
593 slurmctld answers "Invalid job id" (rc 2017) for a job older than
594 ``MinJobAge``; the accounting record then carries the same summary with
595 ``source: slurmdbd`` so ``gco slurm job`` and ``gco slurm output`` keep
596 working for finished jobs (verified live: a ten-minute-old job).
597 """
598 client = _client()
599 try:
600 payload = await client.job(job_id)
601 except slurm_client.SlurmApiError as exc:
602 if exc.status_code != 404:
603 raise _raise(exc, f"reading job {job_id}") from exc
604 payload = {"jobs": []}
605 except Exception as exc:
606 raise _raise(exc, f"reading job {job_id}") from exc
607 jobs = [j for j in payload.get("jobs", []) if isinstance(j, dict)]
608 if jobs:
609 job = {**_summarize_job(jobs[0]), "source": "slurmctld"}
610 return JSONResponse(status_code=200, content=_envelope({"job": job}))
611 try:
612 accounted = await client.accounting_job(job_id)
613 except Exception as exc:
614 raise _raise(exc, f"reading job {job_id} from accounting") from exc
615 records = [j for j in accounted.get("jobs", []) if isinstance(j, dict)]
616 if not records:
617 raise HTTPException(status_code=404, detail=f"Slurm job {job_id} not found")
618 return JSONResponse(
619 status_code=200, content=_envelope({"job": _accounting_record_as_job(records[0])})
620 )
623@router.delete("/jobs/{job_id}")
624async def slurm_cancel(
625 job_id: int,
626 signal: str | None = Query(None, pattern=r"^[A-Z0-9]{1,16}$", description="Signal name"),
627) -> Response:
628 """Cancel a job (scancel), optionally with a specific signal."""
629 client = _client()
630 try:
631 await client.cancel(job_id, signal=signal)
632 except Exception as exc:
633 raise _raise(exc, f"cancelling job {job_id}") from exc
634 logger.info("Cancelled Slurm job %s", sanitize_log_value(job_id))
635 return JSONResponse(status_code=200, content=_envelope({"job_id": job_id, "cancelled": True}))
638@router.get("/accounting/jobs")
639async def slurm_accounting_jobs(
640 start: str | None = Query(None, description="Start of the window (Slurm time or epoch)"),
641 end: str | None = Query(None, description="End of the window"),
642 user: str | None = Query(None, pattern=r"^[A-Za-z0-9_.-]{1,64}$"),
643 partition: str | None = Query(None, pattern=r"^[a-z][a-z0-9-]{0,31}$"),
644 state: str | None = Query(None, pattern=r"^[A-Za-z_,]{1,64}$"),
645 job_name: str | None = Query(None, pattern=r"^[A-Za-z0-9_.:+-]{1,128}$"),
646 limit: int = Query(200, ge=1, le=5000),
647) -> Response:
648 """Job history from slurmdbd (sacct); requires accounting to be enabled.
650 As with ``sacct -s``, a ``state`` filter without an explicit window end
651 matches only jobs in that state right now; the end defaults to ``now`` so
652 ``state=COMPLETED`` returns the window's completed jobs (verified live).
653 """
654 client = _client()
655 params: dict[str, str] = {}
656 if start:
657 params["start_time"] = start
658 if end:
659 params["end_time"] = end
660 elif state:
661 params["end_time"] = "now"
662 if user:
663 params["users"] = user
664 if partition:
665 params["partition"] = partition
666 if state:
667 params["state"] = state
668 if job_name:
669 params["job_name"] = job_name
670 try:
671 payload = await client.accounting_jobs(params)
672 except Exception as exc:
673 raise _raise(exc, "reading accounting") from exc
674 jobs = [_summarize_accounting_job(j) for j in payload.get("jobs", []) if isinstance(j, dict)]
675 return JSONResponse(
676 status_code=200, content=_envelope({"jobs": jobs[:limit], "total": len(jobs)})
677 )