X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/9f37a732c9bfb3892231765f37e213af4e037e68..25af6c40181b95f13ec3e9e366c53cb50868d065:/services/nodemanager/arvnodeman/jobqueue.py

diff --git a/services/nodemanager/arvnodeman/jobqueue.py b/services/nodemanager/arvnodeman/jobqueue.py
index 59659fec0e..abe8852380 100644
--- a/services/nodemanager/arvnodeman/jobqueue.py
+++ b/services/nodemanager/arvnodeman/jobqueue.py
@@ -3,6 +3,7 @@
 from __future__ import absolute_import, print_function
 
 import logging
+import subprocess
 
 from . import clientactor
 from .config import ARVADOS_ERRORS
@@ -30,6 +31,9 @@ class ServerCalculator(object):
                     raise ValueError("unrecognized size field '%s'" % (name,))
                 setattr(self, name, override)
 
+            if self.price is None:
+                raise ValueError("Required field 'price' is None")
+
         def meets_constraints(self, **kwargs):
             for name, want_value in kwargs.iteritems():
                 have_value = getattr(self, name)
@@ -38,11 +42,12 @@ class ServerCalculator(object):
             return True
 
 
-    def __init__(self, server_list, max_nodes=None):
+    def __init__(self, server_list, max_nodes=None, max_price=None):
         self.cloud_sizes = [self.CloudSizeWrapper(s, **kws)
                             for s, kws in server_list]
         self.cloud_sizes.sort(key=lambda s: s.price)
         self.max_nodes = max_nodes or float('inf')
+        self.max_price = max_price or float('inf')
         self.logger = logging.getLogger('arvnodeman.jobqueue')
         self.logged_jobs = set()
 
@@ -69,17 +74,25 @@ class ServerCalculator(object):
         for job in queue:
             seen_jobs.add(job['uuid'])
             constraints = job['runtime_constraints']
-            want_count = self.coerce_int(constraints.get('min_nodes'), 1)
+            want_count = max(1, self.coerce_int(constraints.get('min_nodes'), 1))
             cloud_size = self.cloud_size_for_constraints(constraints)
             if cloud_size is None:
                 if job['uuid'] not in self.logged_jobs:
                     self.logged_jobs.add(job['uuid'])
                     self.logger.debug("job %s not satisfiable", job['uuid'])
-            elif (want_count <= self.max_nodes):
-                servers.extend([cloud_size.real] * max(1, want_count))
+            elif (want_count <= self.max_nodes) and (want_count*cloud_size.price <= self.max_price):
+                servers.extend([cloud_size.real] * want_count)
         self.logged_jobs.intersection_update(seen_jobs)
         return servers
 
+    def cheapest_size(self):
+        return self.cloud_sizes[0]
+
+    def find_size(self, sizeid):
+        for s in self.cloud_sizes:
+            if s.id == sizeid:
+                return s
+        return None
 
 class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
     """Actor to generate server wishlists from the job queue.
@@ -90,7 +103,6 @@ class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
     """
 
     CLIENT_ERRORS = ARVADOS_ERRORS
-    LOGGER_NAME = 'arvnodeman.jobqueue'
 
     def __init__(self, client, timer_actor, server_calc, *args, **kwargs):
         super(JobQueueMonitorActor, self).__init__(
@@ -98,10 +110,27 @@ class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
         self._calculator = server_calc
 
     def _send_request(self):
-        return self._client.jobs().queue().execute()['items']
+        # 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": ram,
+                        "min_scratch_mb_per_node": disk
+                    }
+                })
+
+        queuelist.extend(self._client.jobs().queue().execute()['items'])
+
+        return queuelist
 
     def _got_response(self, queue):
         server_list = self._calculator.servers_for_queue(queue)
-        self._logger.debug("Sending server wishlist: %s",
+        self._logger.debug("Calculated wishlist: %s",
                            ', '.join(s.name for s in server_list) or "(empty)")
         return super(JobQueueMonitorActor, self)._got_response(server_list)