[ARVADOS] created: 3dad67f271492790f63e72ffcbba432cf8e00fa5

[Git user]

        at  3dad67f271492790f63e72ffcbba432cf8e00fa5 (commit)


commit 3dad67f271492790f63e72ffcbba432cf8e00fa5
Author: Lucas Di Pentima <lucas at curoverse.com>
Date:   Mon Jul 3 17:03:53 2017 -0300

    7475: Cancel jobs that cannot be satisfied instead of endlessly retry to run it.
    Add a log entry for every cancelled job explaining why it isn't possible
    to run it.
    Updated tests.
    
    Arvados-DCO-1.1-Signed-off-by: Lucas Di Pentima <lucas at curoverse.com>

diff --git a/services/nodemanager/arvnodeman/jobqueue.py b/services/nodemanager/arvnodeman/jobqueue.py
index ca914e1..dbcbd92 100644
--- a/services/nodemanager/arvnodeman/jobqueue.py
+++ b/services/nodemanager/arvnodeman/jobqueue.py
@@ -11,6 +11,7 @@ import subprocess
 from . import clientactor
 from .config import ARVADOS_ERRORS
 
+
 class ServerCalculator(object):
     """Generate cloud server wishlists from an Arvados job queue.
 
@@ -58,7 +59,6 @@ class ServerCalculator(object):
         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()
 
         self.logger.info("Using cloud node sizes:")
         for s in self.cloud_sizes:
@@ -83,20 +83,20 @@ class ServerCalculator(object):
 
     def servers_for_queue(self, queue):
         servers = []
-        seen_jobs = set()
+        unsatisfiable_jobs = {}
         for job in queue:
-            seen_jobs.add(job['uuid'])
             constraints = job['runtime_constraints']
             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) and (want_count*cloud_size.price <= self.max_price):
+                unsatisfiable_jobs[job['uuid']] = 'Requirements for a single node exceed the available cloud node size'
+            elif (want_count > self.max_nodes):
+                unsatisfiable_jobs[job['uuid']] = "Job's min_nodes constraint is greater than the configured max_nodes (%d)" % self.max_nodes
+            elif (want_count*cloud_size.price <= self.max_price):
                 servers.extend([cloud_size.real] * want_count)
-        self.logged_jobs.intersection_update(seen_jobs)
-        return servers
+            else:
+                unsatisfiable_jobs[job['uuid']] = "Job's price (%d) is above system's max_price limit (%d)" % (want_count*cloud_size.price, self.max_price)
+        return (servers, unsatisfiable_jobs)
 
     def cheapest_size(self):
         return self.cloud_sizes[0]
@@ -107,6 +107,7 @@ class ServerCalculator(object):
                 return s
         return None
 
+
 class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
     """Actor to generate server wishlists from the job queue.
 
@@ -165,7 +166,17 @@ class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
         return queuelist
 
     def _got_response(self, queue):
-        server_list = self._calculator.servers_for_queue(queue)
+        server_list, unsatisfiable_jobs = self._calculator.servers_for_queue(queue)
+        # Cancel any job with unsatisfiable requirements, emitting a log
+        # explaining why.
+        for job_uuid, reason in unsatisfiable_jobs.iteritems():
+            self._client.logs().create(body={
+                'object_uuid': job_uuid,
+                'event_type': 'stderr',
+                'properties': {'text': reason},
+            }).execute()
+            self._client.jobs().cancel(uuid=job['uuid']).execute()
+            self._logger.debug("Unsatisfiable job '%s' cancelled", job_uuid)
         self._logger.debug("Calculated wishlist: %s",
                            ', '.join(s.name for s in server_list) or "(empty)")
         return super(JobQueueMonitorActor, self)._got_response(server_list)
diff --git a/services/nodemanager/tests/test_jobqueue.py b/services/nodemanager/tests/test_jobqueue.py
index 8aa0835..6b4f4b6 100644
--- a/services/nodemanager/tests/test_jobqueue.py
+++ b/services/nodemanager/tests/test_jobqueue.py
@@ -17,14 +17,15 @@ class ServerCalculatorTestCase(unittest.TestCase):
             [(testutil.MockSize(n), {'cores': n}) for n in factors], **kwargs)
 
     def calculate(self, servcalc, *constraints):
-        return servcalc.servers_for_queue(
+        servlist, _ = servcalc.servers_for_queue(
             [{'uuid': 'zzzzz-jjjjj-{:015x}'.format(index),
               'runtime_constraints': cdict}
              for index, cdict in enumerate(constraints)])
+        return servlist
 
     def test_empty_queue_needs_no_servers(self):
         servcalc = self.make_calculator([1])
-        self.assertEqual([], servcalc.servers_for_queue([]))
+        self.assertEqual(([], {}), servcalc.servers_for_queue([]))
 
     def test_easy_server_count(self):
         servcalc = self.make_calculator([1])
@@ -134,7 +135,7 @@ class JobQueueMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
     class MockCalculator(object):
         @staticmethod
         def servers_for_queue(queue):
-            return [testutil.MockSize(n) for n in queue]
+            return ([testutil.MockSize(n) for n in queue], {})
 
 
     def build_monitor(self, side_effect, *args, **kwargs):

-----------------------------------------------------------------------


hooks/post-receive
--