#!/usr/bin/env python
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
from __future__ import absolute_import, print_function
self.cores = kwargs.pop('cores')
# libcloud disk sizes are in GB, Arvados/SLURM are in MB
# multiply by 1000 instead of 1024 to err on low side
+ if self.disk is None:
+ self.disk = 0
self.scratch = self.disk * 1000
self.ram = int(self.ram * node_mem_scaling)
for name, override in kwargs.iteritems():
CLIENT_ERRORS = ARVADOS_ERRORS
- def __init__(self, client, timer_actor, server_calc, *args, **kwargs):
+ def __init__(self, client, timer_actor, server_calc,
+ jobs_queue, slurm_queue, *args, **kwargs):
super(JobQueueMonitorActor, self).__init__(
client, timer_actor, *args, **kwargs)
+ self.jobs_queue = jobs_queue
+ self.slurm_queue = slurm_queue
self._calculator = server_calc
@staticmethod
if u in ("M", "m"):
return int(v)
elif u in ("G", "g"):
- return int(v) * 2**10
+ return float(v) * 2**10
elif u in ("T", "t"):
- return int(v) * 2**20
+ return float(v) * 2**20
elif u in ("P", "p"):
- return int(v) * 2**30
+ return float(v) * 2**30
else:
return int(x)
def _send_request(self):
- # cpus, memory, tempory disk space, reason, job name
- squeue_out = subprocess.check_output(["squeue", "--state=PENDING", "--noheader", "--format=%c %m %d %r %j"])
queuelist = []
- for out in squeue_out.splitlines():
- cpu, ram, disk, reason, jobname = out.split(" ", 4)
- if reason in ("Resources", "ReqNodeNotAvail"):
- queuelist.append({
- "uuid": jobname,
- "runtime_constraints": {
- "min_cores_per_node": cpu,
- "min_ram_mb_per_node": self.coerce_to_mb(ram),
- "min_scratch_mb_per_node": self.coerce_to_mb(disk)
- }
- })
-
- queuelist.extend(self._client.jobs().queue().execute()['items'])
+ if self.slurm_queue:
+ # cpus, memory, tempory disk space, reason, job name
+ squeue_out = subprocess.check_output(["squeue", "--state=PENDING", "--noheader", "--format=%c|%m|%d|%r|%j"])
+ for out in squeue_out.splitlines():
+ try:
+ cpu, ram, disk, reason, jobname = out.split("|", 4)
+ if ("ReqNodeNotAvail" in reason) or ("Resources" in reason):
+ queuelist.append({
+ "uuid": jobname,
+ "runtime_constraints": {
+ "min_cores_per_node": cpu,
+ "min_ram_mb_per_node": self.coerce_to_mb(ram),
+ "min_scratch_mb_per_node": self.coerce_to_mb(disk)
+ }
+ })
+ except ValueError:
+ pass
+
+ if self.jobs_queue:
+ queuelist.extend(self._client.jobs().queue().execute()['items'])
return queuelist