added partition
This commit is contained in:
+27
-6
@@ -42,7 +42,7 @@ from typing import Any, Iterable, Mapping
|
||||
import msgpack
|
||||
import gzip
|
||||
|
||||
VERSION=0.22
|
||||
VERSION=0.23
|
||||
|
||||
# Default GB per CPUs for the cluster
|
||||
default_mempercpu_gb = 2.2
|
||||
@@ -51,6 +51,7 @@ SACCT_FIELDS = [
|
||||
"JobIDRaw",
|
||||
"JobID",
|
||||
"User",
|
||||
"Partition",
|
||||
"JobName",
|
||||
"State",
|
||||
"AllocCPUS",
|
||||
@@ -72,6 +73,7 @@ ALL_COLUMNS = [
|
||||
"JobID",
|
||||
"state",
|
||||
"Count",
|
||||
"Partition",
|
||||
"NTasks",
|
||||
"CPUs",
|
||||
"Nodes",
|
||||
@@ -130,6 +132,7 @@ PRESET_COLUMNS = {
|
||||
"JobID",
|
||||
"state",
|
||||
"Count",
|
||||
"Partition",
|
||||
"NTasks",
|
||||
"CPUs",
|
||||
"Nodes",
|
||||
@@ -145,6 +148,7 @@ PRESET_COLUMNS = {
|
||||
"JobID",
|
||||
"state",
|
||||
"Count",
|
||||
"Partition",
|
||||
"NTasks",
|
||||
"CPUs",
|
||||
"Nodes",
|
||||
@@ -156,6 +160,7 @@ PRESET_COLUMNS = {
|
||||
"JobID",
|
||||
"state",
|
||||
"Count",
|
||||
"Partition",
|
||||
"NTasks",
|
||||
"CPUs",
|
||||
"Nodes",
|
||||
@@ -358,6 +363,7 @@ class JobRecord:
|
||||
jobid: str
|
||||
jobidraw: str
|
||||
username: str
|
||||
partition: str
|
||||
jobname: str
|
||||
reqtasks: int
|
||||
cpus: int
|
||||
@@ -437,6 +443,7 @@ class OutputRow:
|
||||
username: str
|
||||
JobID: str
|
||||
state: str
|
||||
Partition: str
|
||||
NTasks: int
|
||||
CPUs: int
|
||||
Nodes: int
|
||||
@@ -467,6 +474,7 @@ class OutputRow:
|
||||
"username": self.username,
|
||||
"JobID": self.JobID,
|
||||
"state": self.state,
|
||||
"Partition": self.Partition,
|
||||
"NTasks": self.NTasks,
|
||||
"CPUs": self.CPUs,
|
||||
"Nodes": self.Nodes,
|
||||
@@ -843,6 +851,7 @@ def build_job_records(rows: list[dict[str, str]]) -> list[JobRecord]:
|
||||
jobid=top.get("JobID", ""),
|
||||
jobidraw=top.get("JobIDRaw", ""),
|
||||
username=top.get("User", ""),
|
||||
partition=top.get("Partition", ""),
|
||||
jobname=top.get("JobName", ""),
|
||||
reqtasks=req_ntasks,
|
||||
cpus=alloc_cpus,
|
||||
@@ -900,6 +909,7 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis
|
||||
if rx.search(rec.jobname):
|
||||
key = (pat,
|
||||
rec.username,
|
||||
rec.partition,
|
||||
rec.state,
|
||||
rec.reqtasks,
|
||||
rec.cpus,
|
||||
@@ -913,6 +923,7 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis
|
||||
unmatched.append(rec)
|
||||
|
||||
out = [make_aggregate_row(v, username=k[1], jobname=k[0], \
|
||||
partition=k[2],
|
||||
dflt_mpcpu=default_mempercpu_gb) \
|
||||
for k, v in buckets.items()]
|
||||
out.extend(make_single_row(r, dflt_mpcpu=default_mempercpu_gb) \
|
||||
@@ -923,6 +934,7 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis
|
||||
buckets = defaultdict(list)
|
||||
for rec in records:
|
||||
key = (rec.username,
|
||||
rec.partition,
|
||||
rec.state,
|
||||
rec.reqtasks,
|
||||
rec.cpus,
|
||||
@@ -930,7 +942,7 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis
|
||||
rec.reqmem_gb,
|
||||
rec.reqwall_hours)
|
||||
buckets[key].append(rec)
|
||||
return [make_aggregate_row(v, username=k[0],jobname=f"{common_prefix([r.jobname for r in v])}", dflt_mpcpu=default_mempercpu_gb) for k, v in buckets.items()]
|
||||
return [make_aggregate_row(v, username=k[0],partition=k[1],jobname=f"{common_prefix([r.jobname for r in v])}", dflt_mpcpu=default_mempercpu_gb) for k, v in buckets.items()]
|
||||
|
||||
return [make_single_row(r, dflt_mpcpu=default_mempercpu_gb) for r in records]
|
||||
|
||||
@@ -964,6 +976,7 @@ def make_single_row(rec: JobRecord, dflt_mpcpu: float) -> OutputRow:
|
||||
username=rec.username,
|
||||
JobID=jobid,
|
||||
state=state_mappings_inv[rec.state],
|
||||
Partition=rec.partition,
|
||||
NTasks=rec.reqtasks,
|
||||
CPUs=rec.cpus,
|
||||
Nodes=rec.nodes,
|
||||
@@ -991,7 +1004,7 @@ def make_single_row(rec: JobRecord, dflt_mpcpu: float) -> OutputRow:
|
||||
)
|
||||
|
||||
|
||||
def make_aggregate_row(records: list[JobRecord], username: str,
|
||||
def make_aggregate_row(records: list[JobRecord], username: str, partition: str,
|
||||
jobname: str, dflt_mpcpu: float) -> OutputRow:
|
||||
"""Returns an OutputRow based on the given list of job records."""
|
||||
first = records[0]
|
||||
@@ -1005,8 +1018,15 @@ def make_aggregate_row(records: list[JobRecord], username: str,
|
||||
|
||||
planned_sec = mean_or_none([r.planned_sec for r in records
|
||||
if r.planned_sec is not None])
|
||||
planned_time_h = None
|
||||
if planned_sec is not None:
|
||||
planned_time_h = planned_sec/3600
|
||||
|
||||
elig_qtime_sec = mean_or_none([r.elig_qtime_sec for r in records
|
||||
if r.elig_qtime_sec is not None])
|
||||
elig_qtime_h = None
|
||||
if elig_qtime_sec is not None:
|
||||
elig_qtime_h = elig_qtime_sec/3600
|
||||
|
||||
# We average over all jobs' allocated memory. Some jobs could have received
|
||||
# different allocations, even though all of them had the same user required
|
||||
@@ -1035,11 +1055,12 @@ def make_aggregate_row(records: list[JobRecord], username: str,
|
||||
waste_cpu = count * walltime * (100-cpu_efficiency)/100 * first.cpus
|
||||
|
||||
waste_total = waste_cpu + waste_mem/dflt_mpcpu
|
||||
|
||||
|
||||
return OutputRow(
|
||||
username=username,
|
||||
JobID="",
|
||||
state=state_mappings_inv[first.state],
|
||||
Partition=partition,
|
||||
NTasks=first.reqtasks,
|
||||
CPUs=first.cpus,
|
||||
Nodes=first.nodes,
|
||||
@@ -1048,8 +1069,8 @@ def make_aggregate_row(records: list[JobRecord], username: str,
|
||||
UsedMem=used_mem,
|
||||
MemPerCPU=first.mem_per_cpu_gb,
|
||||
ReqWalltime=first.reqwall_hours,
|
||||
Planned_Time=planned_sec/3600,
|
||||
Elig_Qtime=elig_qtime_sec/3600,
|
||||
Planned_Time=planned_time_h,
|
||||
Elig_Qtime=elig_qtime_h,
|
||||
Count=count,
|
||||
CPU_Eff=cpu_efficiency,
|
||||
Mem_Eff=memory_efficiency,
|
||||
|
||||
Reference in New Issue
Block a user