+++ /dev/null
-#!/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 re
-import subprocess32 as subprocess
-
-import arvados.util
-
-from . import clientactor
-from .config import ARVADOS_ERRORS
-
-
-class ServerCalculator(object):
- """Generate cloud server wishlists from an Arvados job queue.
-
- Instantiate this class with a list of cloud node sizes you're willing to
- use, plus keyword overrides from the configuration. Then you can pass
- job queues to servers_for_queue. It will return a list of node sizes
- that would best satisfy the jobs, choosing the cheapest size that
- satisfies each job, and ignoring jobs that can't be satisfied.
- """
- class InvalidCloudSize(object):
- """
- Dummy CloudSizeWrapper-like class, to be used when a cloud node doesn't
- have a recognizable arvados_node_size tag.
- """
- def __init__(self):
- self.id = 'invalid'
- self.name = 'invalid'
- self.ram = 0
- self.disk = 0
- self.scratch = 0
- self.cores = 0
- self.bandwidth = 0
- # price is multiplied by 1000 to get the node weight
- # the maximum node weight is 4294967280
- # so use invalid node weight 4294967 * 1000 = 4294967000
- self.price = 4294967
- self.preemptible = False
- self.extra = {}
-
- def meets_constraints(self, **kwargs):
- return False
-
-
- class CloudSizeWrapper(object):
- 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')
- # 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)
- self.preemptible = False
- for name, override in kwargs.iteritems():
- if name == 'instance_type': continue
- 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)
- if (have_value != 0) and (have_value < want_value):
- return False
- return True
-
-
- 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.max_nodes = max_nodes or float('inf')
- self.max_price = max_price or float('inf')
- self.logger = logging.getLogger('arvnodeman.jobqueue')
-
- 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:
- return int(x)
- except (TypeError, ValueError):
- return fallback
-
- def cloud_size_for_constraints(self, constraints):
- specified_size = constraints.get('instance_type')
- want_value = lambda key: self.coerce_int(constraints.get(key), 0)
- wants = {'cores': want_value('min_cores_per_node'),
- 'ram': want_value('min_ram_mb_per_node'),
- 'scratch': want_value('min_scratch_mb_per_node')}
- # EC2 node sizes are identified by id. GCE sizes are identified by name.
- for size in self.cloud_sizes:
- if (size.meets_constraints(**wants) and
- (specified_size is None or
- size.id == specified_size or size.name == specified_size)):
- return size
- return None
-
- def servers_for_queue(self, queue):
- servers = []
- unsatisfiable_jobs = {}
- for job in queue:
- 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:
- unsatisfiable_jobs[job['uuid']] = (
- "Constraints cannot be satisfied by any node type")
- 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] * want_count)
- 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]
-
- def find_size(self, sizeid):
- for s in self.cloud_sizes:
- if s.id == sizeid:
- return s
- return self.InvalidCloudSize()
-
-
-class JobQueueMonitorActor(clientactor.RemotePollLoopActor):
- """Actor to generate server wishlists from the job queue.
-
- This actor regularly polls Arvados' job queue, and uses the provided
- ServerCalculator to turn that into a list of requested node sizes. That
- list is sent to subscribers on every poll.
- """
-
- CLIENT_ERRORS = ARVADOS_ERRORS
-
- 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):
- queuelist = []
- if self.slurm_queue:
- # cpus, memory, tempory disk space, reason, job name, feature constraints, priority
- squeue_out = subprocess.check_output(["squeue", "--state=PENDING", "--noheader", "--format=%c|%m|%d|%r|%j|%f|%Q"])
- for out in squeue_out.splitlines():
- try:
- cpu, ram, disk, reason, jobname, features, priority = out.split("|", 6)
- except ValueError:
- self._logger.warning("ignored malformed line in squeue output: %r", out)
- continue
- if '-dz642-' not in jobname:
- continue
- if not re.search(r'BadConstraints|ReqNodeNotAvail|Resources|Priority', reason):
- continue
-
- for feature in features.split(','):
- m = re.match(r'instancetype=(.*)', feature)
- if not m:
- continue
- instance_type = m.group(1)
- # Ignore cpu/ram/scratch requirements, bring up
- # the requested node type.
- queuelist.append({
- "uuid": jobname,
- "runtime_constraints": {
- "instance_type": instance_type,
- },
- "priority": int(priority)
- })
- break
- else:
- # No instance type specified. Choose a node type
- # to suit cpu/ram/scratch requirements.
- 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)
- },
- "priority": int(priority)
- })
- queuelist.sort(key=lambda x: x.get('priority', 1), reverse=True)
-
- if self.jobs_queue:
- queuelist.extend(self._client.jobs().queue().execute()['items'])
-
- return queuelist
-
- def _got_response(self, queue):
- server_list, unsatisfiable_jobs = self._calculator.servers_for_queue(queue)
- # Cancel any job/container with unsatisfiable requirements, emitting
- # a log explaining why.
- for job_uuid, reason in unsatisfiable_jobs.iteritems():
- try:
- self._client.logs().create(body={
- 'object_uuid': job_uuid,
- 'event_type': 'stderr',
- 'properties': {'text': reason},
- }).execute()
- # Cancel the job depending on its type
- if arvados.util.container_uuid_pattern.match(job_uuid):
- subprocess.check_call(['scancel', '--name='+job_uuid])
- elif arvados.util.job_uuid_pattern.match(job_uuid):
- self._client.jobs().cancel(uuid=job_uuid).execute()
- else:
- raise Exception('Unknown job type')
- self._logger.debug("Cancelled unsatisfiable job '%s'", job_uuid)
- except Exception as error:
- self._logger.error("Trying to cancel job '%s': %s",
- job_uuid,
- error)
- self._logger.debug("Calculated wishlist: %s",
- ', '.join(s.id for s in server_list) or "(empty)")
- return super(JobQueueMonitorActor, self)._got_response(server_list)
projects / arvados.git / blobdiff