From dde7f0581f05c37ff873b52660cce5ae7410a680 Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 07 2022 12:20:59 +0000 Subject: [PATCH 1/5] move task selection logic to hub --- diff --git a/docs/schema.sql b/docs/schema.sql index 39fdbf7..6a9502f 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -199,7 +199,6 @@ CREATE TABLE host_channels ( UNIQUE (host_id, channel_id, active) ) WITHOUT OIDS; - -- tasks are pretty general and may refer to all sorts of jobs, not -- just package builds. -- tasks may spawn subtasks (hence the parent field) @@ -244,6 +243,14 @@ CREATE INDEX task_by_host ON task (host_id); CREATE INDEX task_by_no_parent_state_method ON task(parent, state, method) WHERE parent IS NULL; +-- helper table for scheduler, see getLoadData +CREATE TABLE skipped_tasks ( + host_id INTEGER NOT NULL REFERENCES host (id), + task_id INTEGER NOT NULL REFERENCES task (id), + seen TIMESTAMPTZ NOT NULL DEFAULT NOW() +) WITHOUT OIDS; + + -- by package, we mean srpm -- we mean the package in general, not an individual build CREATE TABLE package ( diff --git a/hub/kojihub.py b/hub/kojihub.py index 680f9d3..ba05bcb 100644 --- a/hub/kojihub.py +++ b/hub/kojihub.py @@ -14022,30 +14022,137 @@ class Host(object): task['alert'] = True return tasks - def updateHost(self, task_load, ready): + def updateHost(self, task_load, ready, free_resources): host_data = get_host(self.id) task_load = float(task_load) if task_load != host_data['task_load'] or ready != host_data['ready']: update = UpdateProcessor('host', clauses=['id=%(id)i'], values={'id': self.id}, data={'task_load': task_load, 'ready': ready}) update.execute() + free_mem = free_resources.get('memory') + q = """UPDATE host SET + task_load = %(task_load)f, + ready = %(ready)s, + free_space=%(free_space)s, + free_mem = %(free_mem)s + WHERE id=%(id)s""" context.commit_pending = True + def checkAvailDelay(self, task, bin_avail, our_avail): + """Check to see if we should still delay taking a task + + Returns True if we are still in the delay period and should skip the + task. Otherwise False (delay has expired). + """ + + logger.debug("checkAvailDelay: host: %s, task: %s, bin_avail: %s, our_avail: %s" % ( + self.id, task['id'], bin_avail, our_avail)) + q = "SELECT seen FROM skipped_tasks WHERE task_id=%(task_id)s AND host_id=%(host_id)s" + values = {'task_id': task['id'], 'host_id': self.id} + if not _fetchSingle(q, values): + q = 'INSERT INTO skipped_tasks (host_id, task_id) VALUES (%(host_id)s, %(task_id)s)' + _dml(q, values) + + # determine our normalized bin rank + for pos, cap in enumerate(bin_avail): + if our_avail >= cap: + break + if len(bin_avail) > 1: + rank = float(pos) / (len(bin_avail) - 1) + else: + rank = 0.0 + # so, 0.0 for highest available capacity, 1.0 for lowest + + delay = context.opts['task_avail_delay'] + delay *= rank + + q = """SELECT NOW() - seen FROM skipped_tasks + WHERE + task_id=%(task_id)s AND + host_id=%(host_id)s AND + seen >= NOW() - '%(delay)s seconds'::interval""" + values['delay'] = delay + x = _fetchSingle(q, values) + if x: + logger.debug("skipping task %i, age=%s, rank=%s" % (task['id'], x[0], rank)) + return True + else: + q = "DELETE FROM skipped_tasks WHERE host_id = %(host_id)s AND task_id = %(task_id)s" + _dml(q, values) + return False + + def getLoadData(self): - """Get load balancing data + """Get load balancing data""" + c = context.cnx.cursor() + q = "SELECT count(*) FROM task WHERE state = %(st_free)s" + c.execute(q, {'st_free': koji.TASK_STATES['FREE']}) + if c.fetchone()[0] == 0: + return None + tasks = get_active_tasks() - This data is relatively small and the necessary load analysis is - relatively complex, so we let the host machines crunch it.""" + bin_hosts = {} # hosts indexed by bin + bins = {} # bins for this host + our_avail = None hosts = get_ready_hosts() for host in hosts: + host['bins'] = [] if host['id'] == self.id: - break - else: - # this host not in ready list - return [[], []] - # host is the host making the call - tasks = get_active_tasks(host) - return [hosts, tasks] + # note: task_load reported by server might differ from what we + # sent due to precision variation + our_avail = host['capacity'] - host['task_load'] + for chan in host['channels']: + for arch in host['arches'].split() + ['noarch']: + bin = "%s:%s" % (chan, arch) + bin_hosts.setdefault(bin, []).append(host) + if host['id'] == self.id: + bins[bin] = 1 + if our_avail is None: + logger.info("Server did not report this host. Are we disabled?") + return None + elif not bins: + logger.info("No bins for this host. Missing channel/arch config?") + # Note: we may still take an assigned task below + + # sort available capacities for each of our bins + avail = {} + for bin in bins: + avail[bin] = [host['capacity'] - host['task_load'] for host in bin_hosts[bin]] + avail[bin].sort(reverse=True) + + q = """DELETE FROM skipped_tasks + WHERE + host_id = %(host_id)s AND + seen < NOW() - 10 * '%(delay)s seconds'::interval""" + _dml(q, { + 'host_id': host['id'], + 'delay': context.opts['task_avail_delay'] + }) + + for task in tasks: + # note: tasks are in priority order + logger.debug("task: %r" % task) + if task['state'] == koji.TASK_STATES['ASSIGNED']: + logger.debug("task is assigned") + if self.id == task['host_id']: + return task + elif task['state'] == koji.TASK_STATES['FREE']: + bin = "%(channel_id)s:%(arch)s" % task + logger.debug("task is free, bin=%r" % bin) + if bin not in bins: + continue + # see where our available capacity is compared to other hosts for this bin + # (note: the hosts in this bin are exactly those that could + # accept this task) + bin_avail = avail.get(bin, [0]) + if self.checkAvailDelay(task, bin_avail, our_avail): + # decline for now and give the upper half a chance + continue + return task + else: + # should not happen + logger.error("Invalid task state (task: %d, state: %d)" % task['id'], task['state']) + return None def getTask(self): """Open next available task and return it""" @@ -14107,10 +14214,10 @@ class HostExports(object): host.verify() return host.id - def updateHost(self, task_load, ready): + def updateHost(self, task_load, ready, free_resources): host = Host() host.verify() - host.updateHost(task_load, ready) + host.updateHost(task_load, ready, free_resources) def getLoadData(self): host = Host() diff --git a/koji/daemon.py b/koji/daemon.py index 41ac10c..ac0b667 100644 --- a/koji/daemon.py +++ b/koji/daemon.py @@ -51,6 +51,17 @@ from koji.util import ( ) +def _factor(unit): + if unit == 'kB': + return 1024 + elif unit == 'MB': + return 1024**2 + elif unit == 'GB': + return 1024**3 + else: + return 1 + + def incremental_upload(session, fname, fd, path, retries=5, logger=None): if not fd: return @@ -716,7 +727,6 @@ class TaskManager(object): self.options = options self.session = session self.tasks = {} - self.skipped_tasks = {} self.pids = {} self.subsessions = {} self.handlers = {} @@ -762,7 +772,29 @@ class TaskManager(object): for task_id in self.pids: self.cleanupTask(task_id) self.session.host.freeTasks(to_list(self.tasks.keys())) - self.session.host.updateHost(task_load=0.0, ready=False) + self.session.host.updateHost(0.0, False, self.free_resources()) + + def free_resources(self): + memory = 0 + with open('/proc/meminfo', 'rt') as f: + for line in f.readlines(): + if line.startswith('MemAvailable:'): + _, size, unit = line.split() + # in MB + memory = int(size) * _factor(unit) + + br_path = self.options.mockdir + if not os.path.exists(br_path): + self.logger.error("No such directory: %s" % br_path) + raise IOError("No such directory: %s" % br_path) + fs_stat = os.statvfs(br_path) + space = fs_stat.f_bavail * fs_stat.f_bsize + + return { + 'memory': memory / 1024 ** 2, + 'space': space / 1024 ** 2 + } + def updateBuildroots(self, nolocal=False): """Handle buildroot cleanup/maintenance @@ -1020,127 +1052,25 @@ class TaskManager(object): def getNextTask(self): self.ready = self.readyForTask() - self.session.host.updateHost(self.task_load, self.ready) + self.session.host.updateHost(self.task_load, self.ready, self.free_resources()) if not self.ready: self.logger.info("Not ready for task") return False - hosts, tasks = self.session.host.getLoadData() - self.logger.debug("Load Data:") - self.logger.debug(" hosts: %r" % hosts) - self.logger.debug(" tasks: %r" % tasks) - # now we organize this data into channel-arch bins - bin_hosts = {} # hosts indexed by bin - bins = {} # bins for this host - our_avail = None - for host in hosts: - host['bins'] = [] - if host['id'] == self.host_id: - # note: task_load reported by server might differ from what we - # sent due to precision variation - our_avail = host['capacity'] - host['task_load'] - for chan in host['channels']: - for arch in host['arches'].split() + ['noarch']: - bin = "%s:%s" % (chan, arch) - bin_hosts.setdefault(bin, []).append(host) - if host['id'] == self.host_id: - bins[bin] = 1 - self.logger.debug("bins: %r" % bins) - if our_avail is None: - self.logger.info("Server did not report this host. Are we disabled?") + task = self.session.host.getLoadData() + if not task: return False - elif not bins: - self.logger.info("No bins for this host. Missing channel/arch config?") - # Note: we may still take an assigned task below - # sort available capacities for each of our bins - avail = {} - for bin in bins: - avail[bin] = [host['capacity'] - host['task_load'] for host in bin_hosts[bin]] - avail[bin].sort() - avail[bin].reverse() - self.cleanDelayTimes() - for task in tasks: - # note: tasks are in priority order - self.logger.debug("task: %r" % task) - if task['method'] not in self.handlers: - self.logger.warning("Skipping task %(id)i, no handler for method %(method)s", task) - continue - if task['id'] in self.tasks: - # we were running this task, but it apparently has been - # freed or reassigned. We can't do anything with it until - # updateTasks notices this and cleans up. - self.logger.debug("Task %(id)s freed or reassigned", task) - continue - if task['state'] == koji.TASK_STATES['ASSIGNED']: - self.logger.debug("task is assigned") - if self.host_id == task['host_id']: - # assigned to us, we can take it regardless - if self.takeTask(task): - return True - elif task['state'] == koji.TASK_STATES['FREE']: - bin = "%(channel_id)s:%(arch)s" % task - self.logger.debug("task is free, bin=%r" % bin) - if bin not in bins: - continue - # see where our available capacity is compared to other hosts for this bin - # (note: the hosts in this bin are exactly those that could - # accept this task) - bin_avail = avail.get(bin, [0]) - if self.checkAvailDelay(task, bin_avail, our_avail): - # decline for now and give the upper half a chance - continue - # otherwise, we attempt to open the task - if self.takeTask(task): - return True - else: - # should not happen - raise Exception("Invalid task state reported by server") - return False - - def checkAvailDelay(self, task, bin_avail, our_avail): - """Check to see if we should still delay taking a task - - Returns True if we are still in the delay period and should skip the - task. Otherwise False (delay has expired). - """ - - now = time.time() - ts = self.skipped_tasks.get(task['id']) - if not ts: - ts = self.skipped_tasks[task['id']] = now - - # determine our normalized bin rank - for pos, cap in enumerate(bin_avail): - if our_avail >= cap: - break - if len(bin_avail) > 1: - rank = float(pos) / (len(bin_avail) - 1) - else: - rank = 0.0 - # so, 0.0 for highest available capacity, 1.0 for lowest - - delay = getattr(self.options, 'task_avail_delay', 180) - delay *= rank - - # return True if we should delay - if now - ts < delay: - self.logger.debug("skipping task %i, age=%s rank=%s" - % (task['id'], int(now - ts), rank)) - return True - # otherwise - del self.skipped_tasks[task['id']] - return False + if task['method'] not in self.handlers: + self.logger.warning("Skipping task %(id)i, no handler for method %(method)s", task) + return False + if task['id'] in self.tasks: + # we were running this task, but it apparently has been + # freed or reassigned. We can't do anything with it until + # updateTasks notices this and cleans up. + self.logger.debug("Task %(id)s freed or reassigned", task) + return False + self.takeTask(task) + return True - def cleanDelayTimes(self): - """Remove old entries from skipped_tasks""" - now = time.time() - delay = getattr(self.options, 'task_avail_delay', 180) - cutoff = now - delay * 10 - # After 10x the delay, we've had plenty of opportunity to take the - # task, so either it has already been taken or we can't take it. - for task_id in list(self.skipped_tasks): - ts = self.skipped_tasks[task_id] - if ts < cutoff: - del self.skipped_tasks[task_id] def _waitTask(self, task_id, pid=None): """Wait (nohang) on the task, return true if finished""" From 60ae6c4ea8e54b4dc3a143f0add115c77e91daaf Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 07 2022 12:20:59 +0000 Subject: [PATCH 2/5] basic task policy --- diff --git a/hub/kojihub.py b/hub/kojihub.py index ba05bcb..fb1f31e 100644 --- a/hub/kojihub.py +++ b/hub/kojihub.py @@ -14063,7 +14063,7 @@ class Host(object): rank = 0.0 # so, 0.0 for highest available capacity, 1.0 for lowest - delay = context.opts['task_avail_delay'] + delay = context.opts['TaskAvailDelay'] delay *= rank q = """SELECT NOW() - seen FROM skipped_tasks @@ -14107,6 +14107,8 @@ class Host(object): bin_hosts.setdefault(bin, []).append(host) if host['id'] == self.id: bins[bin] = 1 + free_mem = host['free_mem'] + free_space = host['free_space'] if our_avail is None: logger.info("Server did not report this host. Are we disabled?") return None @@ -14126,7 +14128,7 @@ class Host(object): seen < NOW() - 10 * '%(delay)s seconds'::interval""" _dml(q, { 'host_id': host['id'], - 'delay': context.opts['task_avail_delay'] + 'delay': context.opts['TaskAvailDelay'] }) for task in tasks: @@ -14148,10 +14150,25 @@ class Host(object): if self.checkAvailDelay(task, bin_avail, our_avail): # decline for now and give the upper half a chance continue - return task + policy_data = { + # task data + 'method': task['method'], + #'channel': task['channel'], + # host data + 'free_mem': free_mem, + 'free_space': free_space, + } + # additional data from task + policy_data.update(policy_data_from_task(task['id'])) + logger.debug("POLICY: %s", policy_data) + access, reason = check_policy('builder_resources', policy_data, strict=False) + if access: + return task + else: + logger.debug("Policy denied host %s to pick task %s", host['name'], task['id']) else: # should not happen - logger.error("Invalid task state (task: %d, state: %d)" % task['id'], task['state']) + logger.error("Invalid task state (task: %d, state: %d)", task['id'], task['state']) return None def getTask(self): diff --git a/hub/kojixmlrpc.py b/hub/kojixmlrpc.py index 2d802ff..04697cd 100644 --- a/hub/kojixmlrpc.py +++ b/hub/kojixmlrpc.py @@ -475,6 +475,8 @@ def load_config(environ): '%(asctime)s [%(levelname)s] m=%(method)s u=%(user_name)s p=%(process)s r=%(remoteaddr)s ' '%(name)s: %(message)s'], + ['TaskAvailDelay', 'int', 180], + ['MissingPolicyOk', 'boolean', True], ['EnableMaven', 'boolean', False], ['EnableWin', 'boolean', False], @@ -595,6 +597,9 @@ _default_policies = { 'volume': ''' all :: DEFAULT ''', + 'builder_resources': ''' + all :: allow + ''', 'priority': ''' all :: stay ''', diff --git a/koji/daemon.py b/koji/daemon.py index ac0b667..6c17e73 100644 --- a/koji/daemon.py +++ b/koji/daemon.py @@ -789,6 +789,7 @@ class TaskManager(object): raise IOError("No such directory: %s" % br_path) fs_stat = os.statvfs(br_path) space = fs_stat.f_bavail * fs_stat.f_bsize + # space must be tuned to what different tasks requested return { 'memory': memory / 1024 ** 2, From fba81a3f5521d0e669b1b13926f12acda7fdb87b Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 07 2022 12:20:59 +0000 Subject: [PATCH 3/5] reservations --- diff --git a/koji/daemon.py b/koji/daemon.py index 6c17e73..ad52d12 100644 --- a/koji/daemon.py +++ b/koji/daemon.py @@ -789,7 +789,12 @@ class TaskManager(object): raise IOError("No such directory: %s" % br_path) fs_stat = os.statvfs(br_path) space = fs_stat.f_bavail * fs_stat.f_bsize - # space must be tuned to what different tasks requested + + for task in self.tasks: + # can go through listBuildroots and subtract + # (space_reserved - space_consumed_on_disk) + # what about memory? + pass return { 'memory': memory / 1024 ** 2, From 8490febd033404a08365c383b9af035df5a7a5b9 Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 07 2022 12:20:59 +0000 Subject: [PATCH 4/5] subtract used/reserved resources from free_resources --- diff --git a/koji/daemon.py b/koji/daemon.py index ad52d12..bbbf55d 100644 --- a/koji/daemon.py +++ b/koji/daemon.py @@ -61,6 +61,18 @@ def _factor(unit): else: return 1 +def _get_memory_usage(pid): + mem = 0 + r = subprocess.check_output(["ps", "-o", "pid,ppid,pgid,rss", "-u", "tkopecek"]).decode('utf-8') + for line in r.split('\n'): + if not line: + continue + pid, ppid, pgid, rss = line.split() + if pid == 'PID': + continue + if PID in (pid, ppid, pgid): + mem += int(rss) + return mem def incremental_upload(session, fname, fd, path, retries=5, logger=None): if not fd: @@ -729,6 +741,7 @@ class TaskManager(object): self.tasks = {} self.pids = {} self.subsessions = {} + self.reservations = {} self.handlers = {} self.status = '' self.restart_pending = False @@ -790,17 +803,38 @@ class TaskManager(object): fs_stat = os.statvfs(br_path) space = fs_stat.f_bavail * fs_stat.f_bsize - for task in self.tasks: - # can go through listBuildroots and subtract - # (space_reserved - space_consumed_on_disk) - # what about memory? - pass + # subtract used/reserved memory + for task_id in tasks.keys(): + used_mem = 0 + if task_id in self.pids: + used_mem = _get_memory_usage(self.pids[task_id]) + reserved_mem = 0 + if task_id in self.reservations: + reserved_mem = self.reservations[task_id].get('memory', 0) + memory -= max(reserved_mem, used_mem) + + # underestimate space - checking it correctly would be too slow (running "du -s") + for r in self.reservations.values(): + space -= r.get('space', 0) return { - 'memory': memory / 1024 ** 2, - 'space': space / 1024 ** 2 + 'memory': memory, + 'space': space, } + def reserve_resources(self, task_id, requested): + available = self.free_resources() + mem = requested.get('memory', 0) + if available['memory'] < mem: + self.logger.warning("Insufficient memory %s (requested %s)", available['memory'], mem) + return False + space = requested.get('space', 0) + if available['space'] < space: + self.logger.warning("Insufficient space %s (requested %s)", available['space'], space) + return False + + self.reservations[task_id] = requested + return True def updateBuildroots(self, nolocal=False): """Handle buildroot cleanup/maintenance @@ -1034,6 +1068,8 @@ class TaskManager(object): del self.pids[id] if id in self.tasks: del self.tasks[id] + if id in self.reservations: + del self.reservations[id] for id, pid in list(self.pids.items()): if id not in tasks: # expected to happen when: @@ -1049,10 +1085,14 @@ class TaskManager(object): self.logger.info("Killing canceled task %r (pid %r)" % (id, pid)) if self.cleanupTask(id): del self.pids[id] + if id in self.reservations: + del self.reservations[id] elif tinfo['host_id'] != self.host_id: self.logger.info("Killing reassigned task %r (pid %r)" % (id, pid)) if self.cleanupTask(id): del self.pids[id] + if id in self.reservations: + del self.reservations[id] else: self.logger.info("Lingering task %r (pid %r)" % (id, pid)) @@ -1345,6 +1385,12 @@ class TaskManager(object): self.logger.warning("Task '%s' has no request" % task['id']) return False params = task_info['request'] + + if not self.reserve_resources(task_info['id'], task_info.get('required_resources')): + self.logger.info("Skipping task %s (%s) due to insufficient resources", + task['id'], task['method']) + return False + handler = handlerClass(task_info['id'], method, params, self.session, self.options) if hasattr(handler, 'checkHost'): try: From a93cb1f5b3b7f2dd923dfd252f328fb339047915 Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: Dec 07 2022 12:20:59 +0000 Subject: [PATCH 5/5] remove skipped_tasks table and resources management --- diff --git a/docs/schema.sql b/docs/schema.sql index 6a9502f..39fdbf7 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -199,6 +199,7 @@ CREATE TABLE host_channels ( UNIQUE (host_id, channel_id, active) ) WITHOUT OIDS; + -- tasks are pretty general and may refer to all sorts of jobs, not -- just package builds. -- tasks may spawn subtasks (hence the parent field) @@ -243,14 +244,6 @@ CREATE INDEX task_by_host ON task (host_id); CREATE INDEX task_by_no_parent_state_method ON task(parent, state, method) WHERE parent IS NULL; --- helper table for scheduler, see getLoadData -CREATE TABLE skipped_tasks ( - host_id INTEGER NOT NULL REFERENCES host (id), - task_id INTEGER NOT NULL REFERENCES task (id), - seen TIMESTAMPTZ NOT NULL DEFAULT NOW() -) WITHOUT OIDS; - - -- by package, we mean srpm -- we mean the package in general, not an individual build CREATE TABLE package ( diff --git a/hub/kojihub.py b/hub/kojihub.py index fb1f31e..039ccf4 100644 --- a/hub/kojihub.py +++ b/hub/kojihub.py @@ -14022,74 +14022,18 @@ class Host(object): task['alert'] = True return tasks - def updateHost(self, task_load, ready, free_resources): + def updateHost(self, task_load, ready): host_data = get_host(self.id) task_load = float(task_load) if task_load != host_data['task_load'] or ready != host_data['ready']: update = UpdateProcessor('host', clauses=['id=%(id)i'], values={'id': self.id}, data={'task_load': task_load, 'ready': ready}) update.execute() - free_mem = free_resources.get('memory') - q = """UPDATE host SET - task_load = %(task_load)f, - ready = %(ready)s, - free_space=%(free_space)s, - free_mem = %(free_mem)s - WHERE id=%(id)s""" context.commit_pending = True - def checkAvailDelay(self, task, bin_avail, our_avail): - """Check to see if we should still delay taking a task - - Returns True if we are still in the delay period and should skip the - task. Otherwise False (delay has expired). - """ - - logger.debug("checkAvailDelay: host: %s, task: %s, bin_avail: %s, our_avail: %s" % ( - self.id, task['id'], bin_avail, our_avail)) - q = "SELECT seen FROM skipped_tasks WHERE task_id=%(task_id)s AND host_id=%(host_id)s" - values = {'task_id': task['id'], 'host_id': self.id} - if not _fetchSingle(q, values): - q = 'INSERT INTO skipped_tasks (host_id, task_id) VALUES (%(host_id)s, %(task_id)s)' - _dml(q, values) - - # determine our normalized bin rank - for pos, cap in enumerate(bin_avail): - if our_avail >= cap: - break - if len(bin_avail) > 1: - rank = float(pos) / (len(bin_avail) - 1) - else: - rank = 0.0 - # so, 0.0 for highest available capacity, 1.0 for lowest - - delay = context.opts['TaskAvailDelay'] - delay *= rank - - q = """SELECT NOW() - seen FROM skipped_tasks - WHERE - task_id=%(task_id)s AND - host_id=%(host_id)s AND - seen >= NOW() - '%(delay)s seconds'::interval""" - values['delay'] = delay - x = _fetchSingle(q, values) - if x: - logger.debug("skipping task %i, age=%s, rank=%s" % (task['id'], x[0], rank)) - return True - else: - q = "DELETE FROM skipped_tasks WHERE host_id = %(host_id)s AND task_id = %(task_id)s" - _dml(q, values) - return False - - - def getLoadData(self): - """Get load balancing data""" - c = context.cnx.cursor() - q = "SELECT count(*) FROM task WHERE state = %(st_free)s" - c.execute(q, {'st_free': koji.TASK_STATES['FREE']}) - if c.fetchone()[0] == 0: - return None - tasks = get_active_tasks() + def getAvailableTask(self): + """Get one available (free/assigned) task for given builder""" + logger = logging.getLogger('koji.scheduler') bin_hosts = {} # hosts indexed by bin bins = {} # bins for this host @@ -14098,8 +14042,6 @@ class Host(object): for host in hosts: host['bins'] = [] if host['id'] == self.id: - # note: task_load reported by server might differ from what we - # sent due to precision variation our_avail = host['capacity'] - host['task_load'] for chan in host['channels']: for arch in host['arches'].split() + ['noarch']: @@ -14107,13 +14049,11 @@ class Host(object): bin_hosts.setdefault(bin, []).append(host) if host['id'] == self.id: bins[bin] = 1 - free_mem = host['free_mem'] - free_space = host['free_space'] if our_avail is None: - logger.info("Server did not report this host. Are we disabled?") + logger.info(f"Server did not report this host. Are we disabled? {self.id}") return None elif not bins: - logger.info("No bins for this host. Missing channel/arch config?") + logger.info("No bins for this host. Missing channel/arch config? {self.id}") # Note: we may still take an assigned task below # sort available capacities for each of our bins @@ -14122,50 +14062,47 @@ class Host(object): avail[bin] = [host['capacity'] - host['task_load'] for host in bin_hosts[bin]] avail[bin].sort(reverse=True) - q = """DELETE FROM skipped_tasks - WHERE - host_id = %(host_id)s AND - seen < NOW() - 10 * '%(delay)s seconds'::interval""" - _dml(q, { - 'host_id': host['id'], - 'delay': context.opts['TaskAvailDelay'] - }) - - for task in tasks: - # note: tasks are in priority order - logger.debug("task: %r" % task) + for task in get_active_tasks(): + bin = "%(channel_id)s:%(arch)s" % task + # note: tasks are already in priority order + logger.debug(f"task: {task}, host: {self.id}") if task['state'] == koji.TASK_STATES['ASSIGNED']: - logger.debug("task is assigned") + logger.debug(f"task {task} is assigned") if self.id == task['host_id']: + logger.debug(f"task {task} is assigned to us: {self.id}") return task elif task['state'] == koji.TASK_STATES['FREE']: - bin = "%(channel_id)s:%(arch)s" % task - logger.debug("task is free, bin=%r" % bin) + logger.debug(f"task {task} is free, bin={bin}, host={self.id}") if bin not in bins: continue # see where our available capacity is compared to other hosts for this bin # (note: the hosts in this bin are exactly those that could # accept this task) bin_avail = avail.get(bin, [0]) - if self.checkAvailDelay(task, bin_avail, our_avail): - # decline for now and give the upper half a chance - continue - policy_data = { - # task data - 'method': task['method'], - #'channel': task['channel'], - # host data - 'free_mem': free_mem, - 'free_space': free_space, - } - # additional data from task - policy_data.update(policy_data_from_task(task['id'])) - logger.debug("POLICY: %s", policy_data) - access, reason = check_policy('builder_resources', policy_data, strict=False) - if access: - return task + + # Check to see if we should still delay taking a task + logger.debug(f"checkAvailDelay: host: {self.id}, task: {task['id']}, " + f"bin_avail: {bin_avail}, our_avail: {our_avail}") + + # determine our normalized bin rank + for pos, cap in enumerate(bin_avail): + if our_avail >= cap: + break + if len(bin_avail) > 1: + rank = float(pos) / (len(bin_avail) - 1) else: - logger.debug("Policy denied host %s to pick task %s", host['name'], task['id']) + rank = 0.0 + # so, 0.0 for highest available capacity, 1.0 for lowest + + delay = rank * context.opts['TaskAvailDelay'] + now = time.time() + if task['create_time'].timestamp() > (now + delay): + logger.debug( + f"skipping task {task['id']}, " + f"age={now - task['create_time'].timestamp()}, " + f"rank={rank}") + continue + return task else: # should not happen logger.error("Invalid task state (task: %d, state: %d)", task['id'], task['state']) @@ -14231,15 +14168,15 @@ class HostExports(object): host.verify() return host.id - def updateHost(self, task_load, ready, free_resources): + def updateHost(self, task_load, ready): host = Host() host.verify() - host.updateHost(task_load, ready, free_resources) + host.updateHost(task_load, ready) - def getLoadData(self): + def getAvailableTask(self): host = Host() host.verify() - return host.getLoadData() + return host.getAvailableTask() def getHost(self): """Return information about this host""" diff --git a/hub/kojixmlrpc.py b/hub/kojixmlrpc.py index 04697cd..1133037 100644 --- a/hub/kojixmlrpc.py +++ b/hub/kojixmlrpc.py @@ -597,9 +597,6 @@ _default_policies = { 'volume': ''' all :: DEFAULT ''', - 'builder_resources': ''' - all :: allow - ''', 'priority': ''' all :: stay ''', diff --git a/koji/daemon.py b/koji/daemon.py index bbbf55d..4efc903 100644 --- a/koji/daemon.py +++ b/koji/daemon.py @@ -51,29 +51,6 @@ from koji.util import ( ) -def _factor(unit): - if unit == 'kB': - return 1024 - elif unit == 'MB': - return 1024**2 - elif unit == 'GB': - return 1024**3 - else: - return 1 - -def _get_memory_usage(pid): - mem = 0 - r = subprocess.check_output(["ps", "-o", "pid,ppid,pgid,rss", "-u", "tkopecek"]).decode('utf-8') - for line in r.split('\n'): - if not line: - continue - pid, ppid, pgid, rss = line.split() - if pid == 'PID': - continue - if PID in (pid, ppid, pgid): - mem += int(rss) - return mem - def incremental_upload(session, fname, fd, path, retries=5, logger=None): if not fd: return @@ -741,7 +718,6 @@ class TaskManager(object): self.tasks = {} self.pids = {} self.subsessions = {} - self.reservations = {} self.handlers = {} self.status = '' self.restart_pending = False @@ -785,56 +761,7 @@ class TaskManager(object): for task_id in self.pids: self.cleanupTask(task_id) self.session.host.freeTasks(to_list(self.tasks.keys())) - self.session.host.updateHost(0.0, False, self.free_resources()) - - def free_resources(self): - memory = 0 - with open('/proc/meminfo', 'rt') as f: - for line in f.readlines(): - if line.startswith('MemAvailable:'): - _, size, unit = line.split() - # in MB - memory = int(size) * _factor(unit) - - br_path = self.options.mockdir - if not os.path.exists(br_path): - self.logger.error("No such directory: %s" % br_path) - raise IOError("No such directory: %s" % br_path) - fs_stat = os.statvfs(br_path) - space = fs_stat.f_bavail * fs_stat.f_bsize - - # subtract used/reserved memory - for task_id in tasks.keys(): - used_mem = 0 - if task_id in self.pids: - used_mem = _get_memory_usage(self.pids[task_id]) - reserved_mem = 0 - if task_id in self.reservations: - reserved_mem = self.reservations[task_id].get('memory', 0) - memory -= max(reserved_mem, used_mem) - - # underestimate space - checking it correctly would be too slow (running "du -s") - for r in self.reservations.values(): - space -= r.get('space', 0) - - return { - 'memory': memory, - 'space': space, - } - - def reserve_resources(self, task_id, requested): - available = self.free_resources() - mem = requested.get('memory', 0) - if available['memory'] < mem: - self.logger.warning("Insufficient memory %s (requested %s)", available['memory'], mem) - return False - space = requested.get('space', 0) - if available['space'] < space: - self.logger.warning("Insufficient space %s (requested %s)", available['space'], space) - return False - - self.reservations[task_id] = requested - return True + self.session.host.updateHost(task_load=0.0, ready=False) def updateBuildroots(self, nolocal=False): """Handle buildroot cleanup/maintenance @@ -1068,8 +995,6 @@ class TaskManager(object): del self.pids[id] if id in self.tasks: del self.tasks[id] - if id in self.reservations: - del self.reservations[id] for id, pid in list(self.pids.items()): if id not in tasks: # expected to happen when: @@ -1085,24 +1010,20 @@ class TaskManager(object): self.logger.info("Killing canceled task %r (pid %r)" % (id, pid)) if self.cleanupTask(id): del self.pids[id] - if id in self.reservations: - del self.reservations[id] elif tinfo['host_id'] != self.host_id: self.logger.info("Killing reassigned task %r (pid %r)" % (id, pid)) if self.cleanupTask(id): del self.pids[id] - if id in self.reservations: - del self.reservations[id] else: self.logger.info("Lingering task %r (pid %r)" % (id, pid)) def getNextTask(self): self.ready = self.readyForTask() - self.session.host.updateHost(self.task_load, self.ready, self.free_resources()) + self.session.host.updateHost(self.task_load, self.ready) if not self.ready: self.logger.info("Not ready for task") return False - task = self.session.host.getLoadData() + task = self.session.host.getAvailableTask() if not task: return False if task['method'] not in self.handlers: @@ -1117,7 +1038,6 @@ class TaskManager(object): self.takeTask(task) return True - def _waitTask(self, task_id, pid=None): """Wait (nohang) on the task, return true if finished""" if pid is None: @@ -1385,12 +1305,6 @@ class TaskManager(object): self.logger.warning("Task '%s' has no request" % task['id']) return False params = task_info['request'] - - if not self.reserve_resources(task_info['id'], task_info.get('required_resources')): - self.logger.info("Skipping task %s (%s) due to insufficient resources", - task['id'], task['method']) - return False - handler = handlerClass(task_info['id'], method, params, self.session, self.options) if hasattr(handler, 'checkHost'): try: