Merge branch '5353-node-sizes' closes #5353
[arvados.git] / services / nodemanager / arvnodeman / jobqueue.py
1 #!/usr/bin/env python
2
3 from __future__ import absolute_import, print_function
4
5 import logging
6
7 from . import clientactor
8 from .config import ARVADOS_ERRORS
9
10 class ServerCalculator(object):
11     """Generate cloud server wishlists from an Arvados job queue.
12
13     Instantiate this class with a list of cloud node sizes you're willing to
14     use, plus keyword overrides from the configuration.  Then you can pass
15     job queues to servers_for_queue.  It will return a list of node sizes
16     that would best satisfy the jobs, choosing the cheapest size that
17     satisfies each job, and ignoring jobs that can't be satisfied.
18     """
19
20     class CloudSizeWrapper(object):
21         def __init__(self, real_size, **kwargs):
22             self.real = real_size
23             for name in ['id', 'name', 'ram', 'disk', 'bandwidth', 'price',
24                          'extra']:
25                 setattr(self, name, getattr(self.real, name))
26             self.cores = kwargs.pop('cores')
27             self.scratch = self.disk
28             for name, override in kwargs.iteritems():
29                 if not hasattr(self, name):
30                     raise ValueError("unrecognized size field '%s'" % (name,))
31                 setattr(self, name, override)
32
33         def meets_constraints(self, **kwargs):
34             for name, want_value in kwargs.iteritems():
35                 have_value = getattr(self, name)
36                 if (have_value != 0) and (have_value < want_value):
37                     return False
38             return True
39
40
41     def __init__(self, server_list, max_nodes=None, max_price=None):
42         self.cloud_sizes = [self.CloudSizeWrapper(s, **kws)
43                             for s, kws in server_list]
44         self.cloud_sizes.sort(key=lambda s: s.price)
45         self.max_nodes = max_nodes or float('inf')
46         self.max_price = max_price or float('inf')
47         self.logger = logging.getLogger('arvnodeman.jobqueue')
48         self.logged_jobs = set()
49
50     @staticmethod
51     def coerce_int(x, fallback):
52         try:
53             return int(x)
54         except (TypeError, ValueError):
55             return fallback
56
57     def cloud_size_for_constraints(self, constraints):
58         want_value = lambda key: self.coerce_int(constraints.get(key), 0)
59         wants = {'cores': want_value('min_cores_per_node'),
60                  'ram': want_value('min_ram_mb_per_node'),
61                  'scratch': want_value('min_scratch_mb_per_node')}
62         for size in self.cloud_sizes:
63             if size.meets_constraints(**wants):
64                 return size
65         return None
66
67     def servers_for_queue(self, queue):
68         servers = []
69         seen_jobs = set()
70         for job in queue:
71             seen_jobs.add(job['uuid'])
72             constraints = job['runtime_constraints']
73             want_count = self.coerce_int(constraints.get('min_nodes'), 1)
74             cloud_size = self.cloud_size_for_constraints(constraints)
75             if cloud_size is None:
76                 if job['uuid'] not in self.logged_jobs:
77                     self.logged_jobs.add(job['uuid'])
78                     self.logger.debug("job %s not satisfiable", job['uuid'])
79             elif (want_count <= self.max_nodes) and (want_count*cloud_size.price <= self.max_price):
80                 servers.extend([cloud_size.real] * max(1, want_count))
81         self.logged_jobs.intersection_update(seen_jobs)
82         return servers
83
84     def cheapest_size(self):
85         return self.cloud_sizes[0]
86
87
88 class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
89     """Actor to generate server wishlists from the job queue.
90
91     This actor regularly polls Arvados' job queue, and uses the provided
92     ServerCalculator to turn that into a list of requested node sizes.  That
93     list is sent to subscribers on every poll.
94     """
95
96     CLIENT_ERRORS = ARVADOS_ERRORS
97     LOGGER_NAME = 'arvnodeman.jobqueue'
98
99     def __init__(self, client, timer_actor, server_calc, *args, **kwargs):
100         super(JobQueueMonitorActor, self).__init__(
101             client, timer_actor, *args, **kwargs)
102         self._calculator = server_calc
103
104     def _send_request(self):
105         return self._client.jobs().queue().execute()['items']
106
107     def _got_response(self, queue):
108         server_list = self._calculator.servers_for_queue(queue)
109         self._logger.debug("Sending server wishlist: %s",
110                            ', '.join(s.name for s in server_list) or "(empty)")
111         return super(JobQueueMonitorActor, self)._got_response(server_list)