X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/5141c3ee23e89696773e227a93236ef2a51543c2..0561bd0c3c07257fd58ded6c7cfa5feeae97af57:/services/nodemanager/arvnodeman/jobqueue.py diff --git a/services/nodemanager/arvnodeman/jobqueue.py b/services/nodemanager/arvnodeman/jobqueue.py index 239934f529..ca914e1096 100644 --- a/services/nodemanager/arvnodeman/jobqueue.py +++ b/services/nodemanager/arvnodeman/jobqueue.py @@ -1,8 +1,12 @@ #!/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 import logging +import subprocess from . import clientactor from .config import ARVADOS_ERRORS @@ -18,18 +22,26 @@ class ServerCalculator(object): """ class CloudSizeWrapper(object): - def __init__(self, real_size, **kwargs): + def __init__(self, real_size, node_mem_scaling, **kwargs): self.real = real_size for name in ['id', 'name', 'ram', 'disk', 'bandwidth', 'price', 'extra']: setattr(self, name, getattr(self.real, name)) self.cores = kwargs.pop('cores') - self.scratch = self.disk + # 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(): if not hasattr(self, name): 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,15 +50,20 @@ class ServerCalculator(object): return True - def __init__(self, server_list, min_nodes=0, max_nodes=None): - self.cloud_sizes = [self.CloudSizeWrapper(s, **kws) + def __init__(self, server_list, max_nodes=None, max_price=None, + node_mem_scaling=0.95): + self.cloud_sizes = [self.CloudSizeWrapper(s, node_mem_scaling, **kws) for s, kws in server_list] self.cloud_sizes.sort(key=lambda s: s.price) - self.min_nodes = min_nodes 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: + self.logger.info(str(s.__dict__)) + @staticmethod def coerce_int(x, fallback): try: @@ -70,24 +87,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) - - # Make sure the server queue has at least enough entries to - # satisfy min_nodes. - node_shortfall = self.min_nodes - len(servers) - if node_shortfall > 0: - basic_node = self.cloud_size_for_constraints({}) - servers.extend([basic_node.real] * node_shortfall) 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. @@ -98,18 +116,56 @@ class JobQueueMonitorActor(clientactor.RemotePollLoopActor): """ CLIENT_ERRORS = ARVADOS_ERRORS - LOGGER_NAME = 'arvnodeman.jobqueue' - 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 + def coerce_to_mb(x): + v, u = x[:-1], x[-1] + if u in ("M", "m"): + return int(v) + elif u in ("G", "g"): + return float(v) * 2**10 + elif u in ("T", "t"): + return float(v) * 2**20 + elif u in ("P", "p"): + return float(v) * 2**30 + else: + return int(x) + def _send_request(self): - return self._client.jobs().queue().execute()['items'] + queuelist = [] + 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 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)