Skip to content

LSF: key array elements by JOBINDEX so parent status aggregates correctly #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
krokicki merged 3 commits into from
Oct 16, 2026
Merged

LSF: key array elements by JOBINDEX so parent status aggregates correctly #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
732f879
LSF: key array elements by JOBINDEX so parent status aggregates corre…
krokicki Oct 16, 2026
b2f187d
Tests: use real bjobs -json format for array element fixtures
krokicki Oct 16, 2026
5633e33
Revert unrelated _collect_extra_args change accidentally included
krokicki Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion cluster_api/executors/lsf.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@
}

_BJOBS_FIELDS = (
"jobid job_name stat exit_code exec_host max_mem "
"jobid jobindex job_name stat exit_code exec_host max_mem "
"submit_time start_time finish_time"
)

Expand Down Expand Up @@ -305,6 +305,12 @@ def _parse_job_statuses(
job_id = str(rec.get("JOBID", "")).strip()
if not job_id:
continue
# bjobs reports the bare parent ID for every array element; the
# index lives in JOBINDEX ("0" for non-array jobs). Re-key as
# "parent[idx]" so poll()/reconnect() can tell elements apart.
job_index = str(rec.get("JOBINDEX", "")).strip()
if job_index and job_index != "0" and "[" not in job_id:
job_id = f"{job_id}[{job_index}]"

stat = rec.get("STAT", "").strip()
status = _LSF_STATUS_MAP.get(stat)
Expand Down
74 changes: 63 additions & 11 deletions tests/test_lsf.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -599,10 +599,10 @@ def _make_bjobs_json(self, records):
def test_element_ids_as_keys(self, lsf_config):
executor = LSFExecutor(lsf_config)
output = self._make_bjobs_json([
{"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node01", "MAX_MEM": "128 MB",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[2]", "STAT": "RUN", "EXIT_CODE": "-",
{"JOBID": "12345", "JOBINDEX": "2", "STAT": "RUN", "EXIT_CODE": "-",
"EXEC_HOST": "node02", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
])
Expand All @@ -618,7 +618,7 @@ def test_mixed_parent_and_element(self, lsf_config):
{"JOBID": "12345", "STAT": "RUN", "EXIT_CODE": "-",
"EXEC_HOST": "-", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node01", "MAX_MEM": "256 MB",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
])
Expand All @@ -627,6 +627,24 @@ def test_mixed_parent_and_element(self, lsf_config):
assert "12345[1]" in result


def test_real_lsf_format_uses_jobindex(self, lsf_config):
"""Real bjobs -json reports the bare parent ID in JOBID and the
element index only in JOBINDEX (regression for issue #8)."""
executor = LSFExecutor(lsf_config)
output = self._make_bjobs_json([
{"JOBID": "154541712", "JOBINDEX": "2", "STAT": "RUN",
"JOB_NAME": "wells[2]", "EXIT_CODE": "-"},
{"JOBID": "154541712", "JOBINDEX": "1", "STAT": "DONE",
"JOB_NAME": "wells[1]", "EXIT_CODE": ""},
{"JOBID": "99", "JOBINDEX": "0", "STAT": "RUN",
"JOB_NAME": "single", "EXIT_CODE": "-"},
])
result = executor._parse_job_statuses(output)
assert set(result) == {"154541712[1]", "154541712[2]", "99"}
assert result["154541712[1]"][0] == JobStatus.DONE
assert result["154541712[2]"][0] == JobStatus.RUNNING


class TestArrayStatusComputation:
def test_no_elements(self):
record = JobRecord(
Expand Down Expand Up @@ -730,13 +748,13 @@ async def test_poll_populates_elements(self, lsf_config, work_dir):
assert len(job.array_elements) == 0

bjobs_output = json.dumps({"RECORDS": [
{"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node01", "MAX_MEM": "128 MB",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[2]", "STAT": "RUN", "EXIT_CODE": "-",
{"JOBID": "12345", "JOBINDEX": "2", "STAT": "RUN", "EXIT_CODE": "-",
"EXEC_HOST": "node02", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[3]", "STAT": "PEND", "EXIT_CODE": "-",
{"JOBID": "12345", "JOBINDEX": "3", "STAT": "PEND", "EXIT_CODE": "-",
"EXEC_HOST": "-", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
]})
Expand All @@ -752,6 +770,40 @@ async def test_poll_populates_elements(self, lsf_config, work_dir):
assert job.status == JobStatus.RUNNING
assert statuses["12345"] == JobStatus.RUNNING

async def test_poll_real_lsf_format_one_element_done(self, lsf_config, work_dir):
"""Issue #8: bjobs lists DONE elements last and reports the bare parent
ID for every element; the parent must stay RUNNING until all are done."""
executor = LSFExecutor(lsf_config)
with patch.object(
executor, "_call",
new_callable=AsyncMock,
return_value="Job <154541712> is submitted to queue <normal>.",
):
job = await executor.submit_array(
command="echo hello", name="wells", array_range=(1, 4),
resources=ResourceSpec(work_dir=work_dir),
)

def rec(idx, stat):
return {"JOBID": "154541712", "JOBINDEX": str(idx), "STAT": stat,
"JOB_NAME": f"wells[{idx}]", "EXIT_CODE": "-",
"EXEC_HOST": "-", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}

bjobs_output = json.dumps({"RECORDS": [
rec(2, "RUN"), rec(3, "RUN"), rec(4, "RUN"), rec(1, "DONE"),
]})
with patch.object(executor, "_call", new_callable=AsyncMock, return_value=bjobs_output) as m:
await executor.poll()
args = m.call_args.args[0]
assert "jobindex" in args[args.index("-o") + 1]

assert job.status == JobStatus.RUNNING
assert not job.is_terminal
assert len(job.array_elements) == 4
assert job.array_elements[1].status == JobStatus.DONE
assert job.array_elements[4].status == JobStatus.RUNNING

async def test_poll_all_done(self, lsf_config, work_dir):
executor = LSFExecutor(lsf_config)
with patch.object(
Expand All @@ -767,10 +819,10 @@ async def test_poll_all_done(self, lsf_config, work_dir):
)

bjobs_output = json.dumps({"RECORDS": [
{"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node01", "MAX_MEM": "128 MB",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[2]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "2", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node02", "MAX_MEM": "256 MB",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
]})
Expand All @@ -796,13 +848,13 @@ async def test_poll_partial_failure(self, lsf_config, work_dir):
)

bjobs_output = json.dumps({"RECORDS": [
{"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node01", "MAX_MEM": "128 MB",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[2]", "STAT": "EXIT", "EXIT_CODE": "1",
{"JOBID": "12345", "JOBINDEX": "2", "STAT": "EXIT", "EXIT_CODE": "1",
"EXEC_HOST": "node02", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
{"JOBID": "12345[3]", "STAT": "DONE", "EXIT_CODE": "0",
{"JOBID": "12345", "JOBINDEX": "3", "STAT": "DONE", "EXIT_CODE": "0",
"EXEC_HOST": "node03", "MAX_MEM": "-",
"SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"},
]})
Expand Down
7 changes: 7 additions & 0 deletions tests/test_reconnect.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,15 @@ def _make_record(
start_time="Jan 5 10:01:00 2024",
finish_time="-",
):
# Real bjobs -json: elements carry the bare parent ID in JOBID, the
# index in JOBINDEX, and "name[idx]" in JOB_NAME.
job_index = "0"
if "[" in job_id:
job_id, job_index = job_id.rstrip("]").split("[")
job_name = f"{job_name}[{job_index}]"
return {
"JOBID": job_id,
"JOBINDEX": job_index,
"JOB_NAME": job_name,
"STAT": stat,
"EXIT_CODE": exit_code,
Expand Down
Loading