# Code shared by various Koji daemons # Copyright (c) 2010-2014 Red Hat, Inc. # # Koji is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public # License as published by the Free Software Foundation; # version 2.1 of the License. # # This software is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this software; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA # Authors: # Mike McLean # Mike Bonnet from __future__ import absolute_import, division import errno import hashlib import logging import os import re import signal import subprocess import sys import time import traceback from fnmatch import fnmatch import six from six.moves import range, urllib import koji import koji.tasks import koji.xmlrpcplus from koji.tasks import safe_rmtree from koji.util import ( adler32_constructor, base64encode, dslice, parseStatus, to_list, joinpath, ) def incremental_upload(session, fname, fd, path, retries=5, logger=None): if not fd: return if logger is None: logger = logging.getLogger('koji.daemon') if session.opts.get('use_fast_upload'): fast_incremental_upload(session, fname, fd, path, retries, logger) return while True: offset = fd.tell() contents = fd.read(65536) size = len(contents) if size == 0: break data = base64encode(contents) digest = hashlib.sha256(contents).hexdigest() del contents tries = 0 while True: if session.uploadFile(path, fname, size, ("sha256", digest), offset, data): break if tries <= retries: tries += 1 time.sleep(10) continue else: logger.error("Error uploading file %s to %s at offset %d" % (fname, path, offset)) break def fast_incremental_upload(session, fname, fd, path, retries, logger): """Like incremental_upload, but use the fast upload mechanism""" while True: offset = fd.tell() contents = fd.read(65536) if not contents: break hexdigest = adler32_constructor(contents).hexdigest() tries = 0 while True: result = session.rawUpload(contents, offset, path, fname, overwrite=True) if result['hexdigest'] == hexdigest: break if tries <= retries: tries += 1 time.sleep(10) continue else: logger.error("Error uploading file %s to %s at offset %d" % (fname, path, offset)) break def log_output(session, path, args, outfile, uploadpath, cwd=None, logerror=0, append=0, chroot=None, env=None): """Run command with output redirected. If chroot is not None, chroot to the directory specified before running the command.""" pid = os.fork() fd = None if not pid: session._forget() try: if chroot: os.chroot(chroot) if cwd: os.chdir(cwd) flags = os.O_CREAT | os.O_WRONLY if append: flags |= os.O_APPEND fd = os.open(outfile, flags, 0o666) os.dup2(fd, 1) if logerror: os.dup2(fd, 2) # echo the command we're running into the logfile msg = '$ %s\n' % ' '.join(args) if six.PY3: msg = msg.encode('utf-8') os.write(fd, msg) environ = os.environ.copy() if env: environ.update(env) os.execvpe(path, args, environ) except BaseException: msg = ''.join(traceback.format_exception(*sys.exc_info())) if fd: try: if six.PY3: os.write(fd, msg.encode('utf-8')) else: os.write(fd, msg) os.close(fd) except Exception: pass print(msg) os._exit(1) else: if chroot: outfile = os.path.normpath(chroot + outfile) outfd = None remotename = os.path.basename(outfile) while True: status = os.waitpid(pid, os.WNOHANG) time.sleep(1) if not outfd: try: outfd = open(outfile, 'rb') except IOError: # will happen if the forked process has not created the logfile yet continue except Exception: print('Error reading log file: %s' % outfile) print(''.join(traceback.format_exception(*sys.exc_info()))) incremental_upload(session, remotename, outfd, uploadpath) if status[0] != 0: if outfd: outfd.close() return status[1] # BEGIN kojikamid dup # class SCM(object): "SCM abstraction class" types = {'CVS': ('cvs://',), 'CVS+SSH': ('cvs+ssh://',), 'GIT': ('git://', 'git+http://', 'git+https://', 'git+rsync://'), 'GIT+SSH': ('git+ssh://',), 'SVN': ('svn://', 'svn+http://', 'svn+https://'), 'SVN+SSH': ('svn+ssh://',)} @classmethod def is_scm_url(cls, url, strict=False): """ Return True if the url appears to be a valid, accessible source location, False otherwise """ schemes = [s for t in cls.types for s in cls.types[t]] for scheme in schemes: if url.startswith(scheme): return True # otherwise not valid if strict: raise koji.GenericError('Invalid scheme in scm url. Valid schemes ' 'are: %s' % ' '.join(sorted(schemes))) else: return False def __init__(self, url, allow_password=False): """ Initialize the SCM object using the specified url. The expected url format is: scheme://[user@]host/path/to/repo?path/to/module#revision_or_tag_identifier The initialized SCM object will have the following attributes: - url (the unmodified url) - scheme - user (may be null) - host - repository - module - revision - use_common (defaults to True, may be set by assert_allowed()) - source_cmd (defaults to ['make', 'sources'], may be set by assert_allowed()) - scmtype The exact format of each attribute is SCM-specific, but the structure of the url must conform to the template above, or an error will be raised. """ self.logger = logging.getLogger('koji.build.SCM') if not SCM.is_scm_url(url, strict=True): raise koji.GenericError('Invalid SCM URL: %s' % url) self.url = url scheme, user, password, host, path, query, fragment = self._parse_url( allow_password=allow_password) self.scheme = scheme if password is not None: self.user = '%s:%s' % (user, password) else: self.user = user self.host = host self.repository = path self.module = query self.revision = fragment self.use_common = True self.source_cmd = ['make', 'sources'] for scmtype, schemes in SCM.types.items(): if self.scheme in schemes: self.scmtype = scmtype break else: # pragma: no cover # should never happen raise koji.GenericError('Invalid SCM URL: %s' % url) def get_info(self, keys=None): if keys is None: keys = ["url", "scheme", "user", "host", "repository", "module", "revision", "scmtype"] return dslice(vars(self), keys) def _parse_url(self, allow_password=False): """ Parse the SCM url into usable components. Return the following tuple: (scheme, user, host, path, query, fragment) user may be None, everything else will have a value """ # get the url's scheme scheme = self.url.split('://')[0] + '://' # replace the scheme with http:// so that the urlparse works in all cases dummyurl = self.url.replace(scheme, 'http://', 1) parsed_url = urllib.parse.urlparse(dummyurl) path = parsed_url.path params = parsed_url.params query = parsed_url.query fragment = parsed_url.fragment user = None password = None userhost = parsed_url.netloc.split('@') if len(userhost) > 2: raise koji.GenericError('Invalid username@hostname specified: %s' % userhost) if not userhost: raise koji.GenericError( 'Unable to parse SCM URL: %s . Could not find the netloc element.' % self.url) if parsed_url.username: user = parsed_url.username if parsed_url.password: password = parsed_url.password if password is not None and not allow_password: raise koji.GenericError('username:password format not supported: %s:%s' % (user, password)) netloc = userhost[-1] # check for empty path before we apply normpath if not path: raise koji.GenericError( 'Unable to parse SCM URL: %s . Could not find the path element.' % self.url) path = os.path.normpath(path) # path and query should not end with / path = path.rstrip('/') query = query.rstrip('/') # normpath might not strip // at start of path if path.startswith('//'): path = '/' + path.strip('/') # path should start with / if not path.startswith('/'): # pragma: no cover # any such url should have already been caught by is_scm_url raise koji.GenericError('Invalid SCM URL. Path should begin with /: %s) ') # check for validity: params should be empty, query may be empty, everything else should be # populated if params: raise koji.GenericError( 'Unable to parse SCM URL: %s . Params element %s should be empty.' % (self.url, params)) if not scheme: # pragma: no cover # should not happen because of is_scm_url check earlier raise koji.GenericError( 'Unable to parse SCM URL: %s . Could not find the scheme element.' % self.url) if not fragment: raise koji.GenericError( 'Unable to parse SCM URL: %s . Could not find the fragment element.' % self.url) # return parsed values return (scheme, user, password, netloc, path, query, fragment) def assert_allowed(self, allowed='', session=None, by_config=True, by_policy=False, policy_data=None): """ Check whether this scm is allowed and apply options by either/both approach below: - "allowed_scms" from config file, see :func:`~koji.daemon.SCM.assert_allowed_by_config` - "build_from_scm" hub policy, see :func:`~koji.daemon.SCM.assert_allowed_by_policy` When both approaches are applied, the config one will be applied, then the policy one. :param str allowed: The allowed_scms config content which is used for by-config approach :param koji.ClientSession session: The allowed_scms config content which is used for by-policy approach :param bool by_config: Using config or not, Default: True :param bool by_policy: Using policy or not, Default: False :param dict policy_data: The policy data which will be merged with the generated scm info, then will be passed to hub call for policy assertion when using policy. :raises koji.BuildError: if the scm is denied. """ if by_config: self.assert_allowed_by_config(allowed or '') if by_policy: if session is None: raise koji.ConfigurationError( 'When allowed SCM assertion is by policy, session must be passed in.') self.assert_allowed_by_policy(session, **(policy_data or {})) def assert_allowed_by_config(self, allowed): """ Check this scm against allowed list and apply options allowed is a space-separated list of entries in one of the following forms: host:repository[:use_common[:source_cmd]] !host:repository Incorrectly-formatted entries will be skipped with a warning. The first form allows a host:repository pattern and optionally sets a few options for it. The second form explicitly blocks a host:repository pattern Both host and repository are treated as glob patterns If there is a matching entry, then the optional fields, if given, will be applied to the instance. If there is no matching entry, or if the host:repository is blocked then BuildError is raised. The use_common option defaults to on. If it is set to no, off, false or 0, it will be disabled. If the option is on, then kojid will attempt to checkout a common/ directory from the repository. The source_command is a shell command to be run before building the srpm. It defaults to "make sources". This can be overridden by the matching allowed entry. The command must be encoded with commas instead of spaces (e.g. "make,sources"). """ is_allowed = False for allowed_scm in allowed.split(): scm_tuple = allowed_scm.split(':') if len(scm_tuple) < 2: self.logger.warning('Ignoring incorrectly formatted SCM host:repository: %s' % allowed_scm) continue host_pat = scm_tuple[0] repo_pat = scm_tuple[1] invert = False if host_pat.startswith('!'): invert = True host_pat = host_pat[1:] if fnmatch(self.host, host_pat) and fnmatch(self.repository, repo_pat): # match if invert: break is_allowed = True # check if we specify a value for use_common if len(scm_tuple) >= 3: if scm_tuple[2].lower() in ('no', 'off', 'false', '0'): self.use_common = False # check if we specify a custom source_cmd if len(scm_tuple) >= 4: if scm_tuple[3]: self.source_cmd = scm_tuple[3].split(',') else: # there was nothing after the trailing :, # so they don't want to run a source_cmd at all self.source_cmd = None break if not is_allowed: raise koji.BuildError( '%s:%s is not in the list of allowed SCMs' % (self.host, self.repository)) def assert_allowed_by_policy(self, session, **extra_data): """ Check this scm against hub policy: build_from_scm and apply options The policy data is the combination of scminfo with scm_ prefix and kwargs. It should at least contain following keys: - scm_url - scm_scheme - scm_user - scm_host - scm_repository - scm_module - scm_revision - scm_type More keys could be added as kwargs(extra_data). You can pass any reasonable data which could be handled by policy tests, like: - scratch (if the task is scratch) - channel (which channel the task is assigned) - user_id (the task owner) If the key in extra_data is one of scm_* listed above, it will override the one generated from scminfo. The format of the action returned from build_from_scm could be one of following forms:: allow [use_common] [] deny [] If use_common is not set, use_common property is False. If source_cmd is none, it will be parsed as None. If it not set, the default value: ['make', 'sources'], or the value set by :func:`~koji.daemon.SCM.assert_allowed_by_config` will be set. Policy example: build_from_scm = bool scratch :: allow none match scm_host scm.example.com :: allow use_common make sources match scm_host scm2.example.com :: allow all :: deny :param koji.ClientSession session: the session object to call hub xmlrpc APIs. It should be a host session. :raises koji.BuildError: if the scm is denied. """ policy_data = {} for k, v in six.iteritems(self.get_info()): policy_data[re.sub(r'^(scm_?)?', 'scm_', k)] = v policy_data.update(extra_data) result = (session.evalPolicy('build_from_scm', policy_data) or '').split() is_allowed = result and result[0].lower() in ('yes', 'true', 'allow', 'allowed') if not is_allowed: raise koji.BuildError( 'SCM: %s:%s is not allowed, reason: %s' % (self.host, self.repository, ' '.join(result[1:]) or None)) # Apply options when it's allowed applied = result[1:] self.use_common = len(applied) != 0 and applied[0].lower() == 'use_common' idx = 1 if self.use_common else 0 self.source_cmd = applied[idx:] or self.source_cmd if self.source_cmd is not None and len(self.source_cmd) > 0 \ and self.source_cmd[0].lower() == 'none': self.source_cmd = None def checkout(self, scmdir, session=None, uploadpath=None, logfile=None): """ Checkout the module from SCM. Accepts the following parameters: - scmdir: the working directory - session: a ClientSession object - uploadpath: the path on the server the logfile should be uploaded to - logfile: the file used for logging command output session, uploadpath, and logfile are not used when run within kojikamid, but are otherwise required. Returns the directory that the module was checked-out into (a subdirectory of scmdir) """ # TODO: sanity check arguments sourcedir = '%s/%s' % (scmdir, self.module) update_checkout_cmds = None update_recovery_cmd = None update_checkout_dir = None env = None def _run(cmd, chdir=None, fatal=False, log=True, _count=[0]): if globals().get('KOJIKAMID'): # we've been inserted into kojikamid, use its run() return run(cmd, chdir=chdir, fatal=fatal, log=log) # noqa: F821 else: append = (_count[0] > 0) _count[0] += 1 if log_output(session, cmd[0], cmd, logfile, uploadpath, cwd=chdir, logerror=1, append=append, env=env): raise koji.BuildError('Error running %s command "%s", see %s for details' % (self.scmtype, ' '.join(cmd), os.path.basename(logfile))) if self.scmtype == 'CVS': pserver = ':pserver:%s@%s:%s' % ((self.user or 'anonymous'), self.host, self.repository) module_checkout_cmd = ['cvs', '-d', pserver, 'checkout', '-r', self.revision, self.module] common_checkout_cmd = ['cvs', '-d', pserver, 'checkout', 'common'] elif self.scmtype == 'CVS+SSH': if not self.user: raise koji.BuildError( 'No user specified for repository access scheme: %s' % self.scheme) cvsserver = ':ext:%s@%s:%s' % (self.user, self.host, self.repository) module_checkout_cmd = ['cvs', '-d', cvsserver, 'checkout', '-r', self.revision, self.module] common_checkout_cmd = ['cvs', '-d', cvsserver, 'checkout', 'common'] env = {'CVS_RSH': 'ssh'} elif self.scmtype == 'GIT': scheme = self.scheme if '+' in scheme: scheme = scheme.split('+')[1] if self.user: gitrepo = '%s%s@%s%s' % (scheme, self.user, self.host, self.repository) else: gitrepo = '%s%s%s' % (scheme, self.host, self.repository) commonrepo = os.path.dirname(gitrepo) + '/common' checkout_path = os.path.basename(self.repository) if self.repository.endswith('/.git'): # If we're referring to the .git subdirectory of the main module, # assume we need to do the same for the common module checkout_path = os.path.basename(self.repository[:-5]) commonrepo = os.path.dirname(gitrepo[:-5]) + '/common/.git' elif self.repository.endswith('.git'): # If we're referring to a bare repository for the main module, # assume we need to do the same for the common module checkout_path = os.path.basename(self.repository[:-4]) commonrepo = os.path.dirname(gitrepo[:-4]) + '/common.git' # git-reset happily accepted origin/x spec, fetch has it split if self.revision.startswith('origin/'): rev = self.revision[7:] else: rev = self.revision sourcedir = '%s/%s' % (scmdir, checkout_path) module_checkout_cmd = ['git', 'clone', '-n', gitrepo, sourcedir] common_checkout_cmd = ['git', 'clone', commonrepo, 'common'] update_checkout_cmds = [ ['git', 'fetch', 'origin', '%s:KOJI_FETCH_HEAD' % rev], ['git', 'reset', '--hard', 'KOJI_FETCH_HEAD'] ] update_recovery_cmd = ['git', 'reset', '--hard', self.revision] update_checkout_dir = sourcedir # self.module may be empty, in which case the specfile should be in the top-level # directory if self.module: # Treat the module as a directory inside the git repository sourcedir = '%s/%s' % (sourcedir, self.module) elif self.scmtype == 'GIT+SSH': if not self.user: raise koji.BuildError( 'No user specified for repository access scheme: %s' % self.scheme) gitrepo = 'git+ssh://%s@%s%s' % (self.user, self.host, self.repository) commonrepo = os.path.dirname(gitrepo) + '/common' checkout_path = os.path.basename(self.repository) if self.repository.endswith('/.git'): # If we're referring to the .git subdirectory of the main module, # assume we need to do the same for the common module checkout_path = os.path.basename(self.repository[:-5]) commonrepo = os.path.dirname(gitrepo[:-5]) + '/common/.git' elif self.repository.endswith('.git'): # If we're referring to a bare repository for the main module, # assume we need to do the same for the common module checkout_path = os.path.basename(self.repository[:-4]) commonrepo = os.path.dirname(gitrepo[:-4]) + '/common.git' # git-reset happily accepted origin/x spec, fetch has it split if self.revision.startswith('origin/'): rev = self.revision[7:] else: rev = self.revision sourcedir = '%s/%s' % (scmdir, checkout_path) module_checkout_cmd = ['git', 'clone', '-n', gitrepo, sourcedir] common_checkout_cmd = ['git', 'clone', commonrepo, 'common'] update_checkout_cmds = [ ['git', 'fetch', 'origin', '%s:KOJI_FETCH_HEAD' % rev], ['git', 'reset', '--hard', 'KOJI_FETCH_HEAD'] ] update_recovery_cmd = ['git', 'reset', '--hard', self.revision] update_checkout_dir = sourcedir # self.module may be empty, in which case the specfile should be in the top-level # directory if self.module: # Treat the module as a directory inside the git repository sourcedir = '%s/%s' % (sourcedir, self.module) elif self.scmtype == 'SVN': scheme = self.scheme if '+' in scheme: scheme = scheme.split('+')[1] svnserver = '%s%s%s' % (scheme, self.host, self.repository) module_checkout_cmd = ['svn', 'checkout', '-r', self.revision, '%s/%s' % (svnserver, self.module), self.module] common_checkout_cmd = ['svn', 'checkout', '%s/common' % svnserver] elif self.scmtype == 'SVN+SSH': if not self.user: raise koji.BuildError( 'No user specified for repository access scheme: %s' % self.scheme) svnserver = 'svn+ssh://%s@%s%s' % (self.user, self.host, self.repository) module_checkout_cmd = ['svn', 'checkout', '-r', self.revision, '%s/%s' % (svnserver, self.module), self.module] common_checkout_cmd = ['svn', 'checkout', '%s/common' % svnserver] else: raise koji.BuildError('Unknown SCM type: %s' % self.scmtype) # perform checkouts _run(module_checkout_cmd, chdir=scmdir, fatal=True) if update_checkout_cmds: # Currently only required for GIT checkouts # Run the command in the directory the source was checked out into if self.scmtype.startswith('GIT') and globals().get('KOJIKAMID'): _run(['git', 'config', 'core.autocrlf', 'true'], chdir=update_checkout_dir, fatal=True) _run(['git', 'config', 'core.safecrlf', 'true'], chdir=update_checkout_dir, fatal=True) try: for cmd in update_checkout_cmds: _run(cmd, chdir=update_checkout_dir, fatal=True) except Exception: # use old-style checkout, e.g. for shortened refs _run(update_recovery_cmd, chdir=update_checkout_dir, fatal=True) if self.use_common and not globals().get('KOJIKAMID'): _run(common_checkout_cmd, chdir=scmdir, fatal=True) if not os.path.exists('%s/../common' % sourcedir): # find the relative distance from sourcedir/../common to scmdir/common destdir = os.path.split(sourcedir)[0] path_comps = destdir[len(scmdir) + 1:] rel_path = '../' * len(path_comps.split('/')) os.symlink(rel_path + 'common', '%s/../common' % sourcedir) self.sourcedir = sourcedir return sourcedir def get_source(self): r = { 'url': self.url, 'source': '', } if self.scmtype.startswith('GIT'): cmd = ['git', 'rev-parse', 'HEAD'] proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, cwd=self.sourcedir,) out, _ = proc.communicate() status = proc.wait() if status != 0: raise koji.GenericError('Error getting commit hash for git') fragment = out.strip() if six.PY3: fragment = fragment.decode() scheme = self.scheme[:-3] netloc = self.host path = self.repository query = self.module r['source'] = urllib.parse.urlunsplit([scheme, netloc, path, query, fragment]) else: # just use the same url r['source'] = self.url return r # END kojikamid dup # class TaskManager(object): def __init__(self, options, session): self.options = options self.session = session self.tasks = {} self.skipped_tasks = {} self.pids = {} self.subsessions = {} self.handlers = {} self.status = '' self.restart_pending = False self.ready = False self.hostdata = {} self.task_load = 0.0 self.host_id = self.session.host.getID() self.start_ts = self.session.getSessionInfo()['start_ts'] self.logger = logging.getLogger("koji.TaskManager") def findHandlers(self, vars): """Find and index task handlers""" for v in vars.values(): self.registerHandler(v) def registerHandler(self, entry): """register and index task handler""" if isinstance(entry, type(koji.tasks.BaseTaskHandler)) and \ issubclass(entry, koji.tasks.BaseTaskHandler): for method in entry.Methods: self.handlers[method] = entry def registerCallback(self, entry): """register and index callback plugins""" if callable(entry) and getattr(entry, 'callbacks', None): for cbtype in entry.callbacks: koji.plugin.register_callback(cbtype, entry) def registerEntries(self, vars): """Register task handlers and other plugins""" for v in vars.values(): self.registerHandler(v) self.registerCallback(v) def scanPlugin(self, plugin): """Find task handlers in a plugin""" self.registerEntries(vars(plugin)) def shutdown(self): """Attempt to shut down cleanly""" 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) def updateBuildroots(self, nolocal=False): """Handle buildroot cleanup/maintenance - examine current buildroots on system - compare with db - clean up as needed - /var/lib/mock - /etc/mock/koji If nolocal is True, do not try to scan local buildroots. """ # query buildroots in db that are not expired states = [koji.BR_STATES[x] for x in ('INIT', 'WAITING', 'BUILDING')] db_br = self.session.listBuildroots(hostID=self.host_id, state=tuple(states)) # index by id db_br = dict([(row['id'], row) for row in db_br]) st_expired = koji.BR_STATES['EXPIRED'] for id, br in db_br.items(): task_id = br['task_id'] if task_id is None: # not associated with a task # this makes no sense now, but may in the future self.logger.warning("Expiring taskless buildroot: %(id)i/%(tag_name)s/%(arch)s" % br) self.session.host.setBuildRootState(id, st_expired) elif task_id not in self.tasks: # task not running - expire the buildroot # TODO - consider recycling hooks here (with strong sanity checks) self.logger.info("Expiring buildroot: %(id)i/%(tag_name)s/%(arch)s" % br) self.logger.debug( "Buildroot task: %r, Current tasks: %r" % (task_id, to_list(self.tasks.keys()))) self.session.host.setBuildRootState(id, st_expired) continue if nolocal: return local_br = self._scanLocalBuildroots() # get info on local_only buildroots (most likely expired) local_only = [id for id in local_br if id not in db_br] if local_only: missed_br = self.session.listBuildroots(buildrootID=tuple(local_only)) # get all the task info in one call tasks = [] for br in missed_br: task_id = br['task_id'] if task_id: tasks.append(task_id) # index missed_br = dict([(row['id'], row) for row in missed_br]) tasks = dict([(row['id'], row) for row in self.session.getTaskInfo(tasks)]) # go from +- oldest for id in sorted(local_only): # Cleaning options # - wait til later # - "soft" clean (leaving empty root/ dir) # - full removal data = local_br[id] br = missed_br.get(id) if not br: self.logger.warning("%(name)s: not in db" % data) continue desc = "%(id)i/%(tag_name)s/%(arch)s" % br if not br['retire_ts']: self.logger.warning("%s: no retire timestamp" % desc) continue age = time.time() - br['retire_ts'] self.logger.debug("Expired/stray buildroot: %s" % desc) if br and br['task_id']: task = tasks.get(br['task_id']) if not task: self.logger.warning("%s: invalid task %s" % (desc, br['task_id'])) continue if task['state'] == koji.TASK_STATES['FAILED'] and \ age < self.options.failed_buildroot_lifetime: # XXX - this could be smarter # keep buildroots for failed tasks around for a little while if self.checkSpace(): # we can leave it in place, otherwise delete it self.logger.debug("Keeping failed buildroot: %s" % desc) continue topdir = data['dir'] rootdir = None if topdir: rootdir = "%s/root" % topdir try: st = os.lstat(rootdir) except OSError as e: if e.errno == errno.ENOENT: rootdir = None else: self.logger.warning("%s: %s" % (desc, e)) continue else: age = min(age, time.time() - st.st_mtime) topdir_bootstrap = "%s-bootstrap" % topdir if not os.path.exists(topdir_bootstrap): topdir_bootstrap = None # note: https://bugzilla.redhat.com/bugzilla/show_bug.cgi?id=192153) # If rpmlib is installing in this chroot, removing it entirely # can lead to a world of hurt. # We remove the rootdir contents but leave the rootdir unless it # is really old if age > self.options.buildroot_final_cleanup_delay: # dir untouched for a day self.logger.info("Removing buildroot: %s" % desc) if ((topdir and safe_rmtree(topdir, unmount=True, strict=False) != 0) or (topdir_bootstrap and safe_rmtree(topdir_bootstrap, unmount=True, strict=False) != 0)): continue # also remove the config try: os.unlink(data['cfg']) except OSError as e: self.logger.warning("%s: can't remove config: %s" % (desc, e)) elif age > self.options.buildroot_basic_cleanup_delay and rootdir: for d in (topdir, topdir_bootstrap): if not d: continue if d == topdir_bootstrap: desc2 = "%s [bootstrap]" % desc else: desc2 = desc rootdir = joinpath(d, 'root') try: flist = os.listdir(rootdir) except OSError as e: self.logger.warning("%s: can't list rootdir: %s" % (desc2, e)) continue if flist: self.logger.info("%s: clearing rootdir" % desc2) for fn in flist: safe_rmtree("%s/%s" % (rootdir, fn), unmount=True, strict=False) # bootstrap's resultdir is 'results', so we try the best to remove both # 'result(s)' dirs for r in ('result', 'results'): resultdir = "%s/%s" % (d, r) if os.path.isdir(resultdir): self.logger.info("%s: clearing resultdir: %s" % (desc2, resultdir)) safe_rmtree(resultdir, unmount=True, strict=False) else: self.logger.debug("Recent buildroot: %s: %i seconds" % (desc, age)) self.logger.debug("Local buildroots: %d" % len(local_br)) self.logger.debug("Active buildroots: %d" % len(db_br)) self.logger.debug("Expired/stray buildroots: %d" % len(local_only)) def _scanLocalBuildroots(self): # XXX configdir = '/etc/mock/koji' buildroots = {} for f in os.listdir(configdir): if not f.endswith('.cfg'): continue fn = "%s/%s" % (configdir, f) if not os.path.isfile(fn): continue fo = koji._open_text_file(fn) id = None name = None for n in range(10): # data should be in first few lines line = fo.readline() if line.startswith('# Koji buildroot id:'): try: id = int(line.split(':')[1]) except (ValueError, IndexError): continue if line.startswith('# Koji buildroot name:'): try: name = line.split(':')[1].strip() except (ValueError, IndexError): continue if id is None or name is None: continue # see if there's a dir for the buildroot vardir = os.path.join(self.options.mockdir, name) buildroots[id] = {} buildroots[id]['name'] = name buildroots[id]['cfg'] = fn buildroots[id]['dir'] = None if os.path.isdir(vardir): buildroots[id]['dir'] = vardir return buildroots def updateTasks(self): """Read and process task statuses from server The processing we do is: 1) clean up after tasks that are not longer active: * kill off processes * retire buildroots * remove buildroots - with some possible exceptions 2) wake waiting tasks if appropriate """ tasks = {} stale = [] task_load = 0.0 if self.pids: self.logger.info("pids: %r" % self.pids) for task in self.session.host.getHostTasks(): self.logger.info("open task: %r" % task) # the tasks returned are those that are open and locked # by this host. id = task['id'] if id not in self.pids: # We don't have a process for this # Expected to happen after a restart, otherwise this is an error stale.append(id) continue tasks[id] = task if task.get('alert', False): # wake up the process self.logger.info("Waking up task: %r" % task) os.kill(self.pids[id], signal.SIGUSR2) if not task['waiting']: task_load += task['weight'] self.logger.debug("Task Load: %s" % task_load) self.task_load = task_load self.tasks = tasks self.logger.debug("Current tasks: %r" % self.tasks) if len(stale) > 0: # A stale task is one which is opened to us, but we know nothing # about). This will happen after a daemon restart, for example. self.logger.info("freeing stale tasks: %r" % stale) self.session.host.freeTasks(stale) for id, pid in list(self.pids.items()): if self._waitTask(id, pid): # the subprocess handles most everything, we just need to clear things out if self.cleanupTask(id, wait=False): del self.pids[id] if id in self.tasks: del self.tasks[id] for id, pid in list(self.pids.items()): if id not in tasks: # expected to happen when: # - we are in the narrow gap between the time the task # records its result and the time the process actually # exits. # - task is canceled # - task is forcibly reassigned/unassigned tinfo = self.session.getTaskInfo(id) if tinfo is None: raise koji.GenericError("Invalid task %r (pid %r)" % (id, pid)) elif tinfo['state'] == koji.TASK_STATES['CANCELED']: self.logger.info("Killing canceled task %r (pid %r)" % (id, pid)) if self.cleanupTask(id): del self.pids[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] else: self.logger.info("Lingering task %r (pid %r)" % (id, pid)) def _get_host_data(self): data = { 'methods': list(self.handlers.keys()), 'maxjobs': self.options.maxjobs, # TODO: now it would be duplicated by updateHost # 'ready': self.ready, # 'task_load': self.task_load, # cpu_load, free_mem, free_disk, ... } return data def getNextTask(self): """Task the next task :returns: True if a task was taken, False otherwise """ self.ready = self.readyForTask() self.session.host.updateHost(self.task_load, self.ready, data=self._get_host_data()) if not self.ready: self.logger.info("Not ready for task") return False # get our assigned tasks tasks = self.session.host.getTasks() for task in tasks: self.logger.debug("task: %r" % task) if task['id'] in self.tasks: # we were running this task, but it apparently has been # reassigned. We can't do anything with it until # updateTasks notices this and cleans up. self.logger.info("Task %(id)s reassigned", task) continue if task['state'] != koji.TASK_STATES['ASSIGNED']: # shouldn't happen self.logger.error("Recieved task %(id)s is not assigned, state=%(state)s", task) continue if task['host_id'] != self.host_id: # shouldn't happen self.logger.error("Recieved task %(id)s is not ours, host=%(host_id)s", task) continue # otherwise attempt to take it if self.takeTask(task): return True return False def _waitTask(self, task_id, pid=None): """Wait (nohang) on the task, return true if finished""" if pid is None: pid = self.pids.get(task_id) if not pid: raise koji.GenericError("No pid for task %i" % task_id) prefix = "Task %i (pid %i)" % (task_id, pid) try: (childpid, status) = os.waitpid(pid, os.WNOHANG) except OSError as e: # check errno if e.errno != errno.ECHILD: # should not happen raise # otherwise assume the process is gone self.logger.info("%s: %s" % (prefix, e)) return True if childpid != 0: self.logger.info(parseStatus(status, prefix)) return True return False def _doKill(self, task_id, pid, cmd, sig, timeout, pause): """ Kill the process with the given process ID. Return True if the process is successfully killed in the given timeout, False otherwise. """ self.logger.info('Checking "%s" (pid %i, taskID %i)...' % (cmd, pid, task_id)) execname = cmd.split()[0] signaled = False t = 0.0 while True: status = self._getStat(pid) if status and status[1] == cmd and status[2] != 'Z': self.logger.info('%s (pid %i, taskID %i) is running' % (execname, pid, task_id)) else: if signaled: self.logger.info( '%s (pid %i, taskID %i) was killed by signal %i' % (execname, pid, task_id, sig)) else: self.logger.info('%s (pid %i, taskID %i) exited' % (execname, pid, task_id)) return True if t >= timeout: self.logger.warning('Failed to kill %s (pid %i, taskID %i) with signal %i' % (execname, pid, task_id, sig)) return False try: os.kill(pid, sig) except OSError as e: # process probably went away, we'll find out on the next iteration self.logger.info('Error sending signal %i to %s (pid %i, taskID %i): %s' % (sig, execname, pid, task_id, e)) else: signaled = True self.logger.info('Sent signal %i to %s (pid %i, taskID %i)' % (sig, execname, pid, task_id)) time.sleep(pause) t += pause def _getStat(self, pid): """ Get the stat info for the given pid. Return a list of all the fields in /proc//stat. The second entry will contain the full command-line instead of just the command name. If the process does not exist, return None. """ try: proc_path = '/proc/%i/stat' % pid if not os.path.isfile(proc_path): return None proc_file = koji._open_text_file(proc_path) procstats = [not field.isdigit() and field or int(field) for field in proc_file.read().split()] proc_file.close() cmd_path = '/proc/%i/cmdline' % pid if not os.path.isfile(cmd_path): return None cmd_file = koji._open_text_file(cmd_path) procstats[1] = cmd_file.read().replace('\0', ' ').strip() cmd_file.close() if not procstats[1]: return None return procstats except IOError: # process may have already gone away return None def _childPIDs(self, pid): """Recursively get the children of the process with the given ID. Return a list containing the process IDs of the children in breadth-first order, without duplicates.""" statsByPPID = {} pidcmd = None for procdir in os.listdir('/proc'): if not procdir.isdigit(): continue procid = int(procdir) procstats = self._getStat(procid) if not procstats: continue statsByPPID.setdefault(procstats[3], []).append(procstats) if procid == pid: pidcmd = procstats[1] pids = [] if pidcmd: # only append the pid if it still exists pids.append((pid, pidcmd)) parents = [pid] while parents: for ppid in parents[:]: for procstats in statsByPPID.get(ppid, []): # get the /proc entries with ppid as their parent, and append their pid to the # list, then recheck for their children pid is the 0th field, ppid is the 3rd # field pids.append((procstats[0], procstats[1])) parents.append(procstats[0]) parents.remove(ppid) return pids def _killChildren(self, task_id, children, sig=signal.SIGTERM, timeout=2.0, pause=1.0): """ Kill child processes of the given task, as specified in the children list, by sending sig. Retry every pause seconds, within timeout. Remove successfully killed processes from the "children" list. """ for childpid, cmd in children[::-1]: # iterate in reverse order so processes whose children are killed might have # a chance to cleanup before they're killed if self._doKill(task_id, childpid, cmd, sig, timeout, pause): children.remove((childpid, cmd)) def cleanupTask(self, task_id, wait=True): """Clean up after task - kill children - expire session Return True if all children were successfully killed, False otherwise. """ pid = self.pids.get(task_id) if not pid: raise koji.GenericError("No pid for task %i" % task_id) children = self._childPIDs(pid) if children: # send SIGINT once to let mock mock try to clean up self._killChildren(task_id, children, sig=signal.SIGINT, pause=3.0) if children: self._killChildren(task_id, children) if children: self._killChildren(task_id, children, sig=signal.SIGKILL, timeout=3.0) # expire the task's subsession session_id = self.subsessions.get(task_id) if session_id: self.logger.info("Expiring subsession %i (task %i)" % (session_id, task_id)) try: self.session.logoutChild(session_id) del self.subsessions[task_id] except Exception: # not much we can do about it pass if wait: return self._waitTask(task_id, pid) else: # task has already been waited on, and we've cleaned # up as much as we can return True def checkSpace(self): """See if we have enough space to accept another job""" 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) available = fs_stat.f_bavail * fs_stat.f_bsize availableMB = available // 1024 // 1024 self.logger.debug("disk space available in '%s': %i MB", br_path, availableMB) if availableMB < self.options.minspace: self.status = "Insufficient disk space at %s: %i MB, %i MB required" % \ (br_path, availableMB, self.options.minspace) self.logger.warning(self.status) return False return True def readyForTask(self): """Determine if the system is ready to accept a new task. This function measures the system load and tries to determine if there is room to accept a new task.""" # key resources to track: # disk_space # df -P path # df -iP path ? # memory (meminfo/vmstat) # vmstat fields 3-6 (also 7-8 for swap) # https://www.redhat.com/advice/tips/meminfo.html # cpu cycles (vmstat?) # vmstat fields 13-16 (and others?) # others?: # io (iostat/vmstat) # network (netstat?) if self.restart_pending: if self.tasks: self.status = "Pending restart" self.logger.info(self.status) return False else: raise koji.tasks.ServerRestart self.hostdata = self.session.host.getHost() self.logger.debug('hostdata: %r' % self.hostdata) if not self.hostdata['enabled']: self.status = "Host is disabled" self.logger.info(self.status) return False if self.task_load > self.hostdata['capacity']: self.status = "Over capacity" self.logger.info( "Task load (%.2f) exceeds capacity (%.2f)" % (self.task_load, self.hostdata['capacity'])) return False if len(self.tasks) >= self.options.maxjobs: # This serves as a backup to the capacity check and prevents # a tremendous number of low weight jobs from piling up self.status = "Full queue" self.logger.info(self.status) return False if not self.checkSpace(): # checkSpace() does its own logging return False loadavgs = os.getloadavg() # this likely treats HT processors the same as real ones # but that's fine, it's a conservative test maxload = 4.0 * os.sysconf('SC_NPROCESSORS_ONLN') if loadavgs[0] > maxload: self.status = "Load average %.2f > %.2f" % (loadavgs[0], maxload) self.logger.info(self.status) return False # XXX - add more checks return True def takeTask(self, task): """Attempt to open the specified task Returns True if successful, False otherwise """ self.logger.info("Attempting to take task %s" % task['id']) method = task['method'] if method in self.handlers: handlerClass = self.handlers[method] else: self.logger.warning("Refusing task %(id)s, no handler for %(method)s", task) self.session.host.refuseTask(task['id'], soft=False, msg="no handler for method") return False task_info = self.session.getTaskInfo(task['id'], request=True) if task_info.get('request') is None: self.logger.warning("Task '%s' has no request" % task['id']) return False params = task_info['request'] handler = handlerClass(task_info['id'], method, params, self.session, self.options) if hasattr(handler, 'checkHost'): try: valid_host = handler.checkHost(self.hostdata) except (SystemExit, KeyboardInterrupt): raise except Exception: valid_host = False self.logger.warning('Error during host check') self.logger.warning(''.join(traceback.format_exception(*sys.exc_info()))) if not valid_host: self.logger.info( 'Skipping task %s (%s) due to host check', task['id'], task['method']) if task['state'] == koji.TASK_STATES['ASSIGNED']: self.session.host.refuseTask(task['id'], soft=False, msg='failed host check') return False data = self.session.host.openTask(task['id']) if data is None: self.logger.warning("Could not open") return False task_id = data['id'] self.tasks[task_id] = data # set weight try: self.session.host.setTaskWeight(task_id, handler.weight()) except koji.ActionNotAllowed: info2 = self.session.getTaskInfo(task['id']) if info2['host_id'] != self.host_id: self.logger.warning("Task %i was reassigned", task_id) return False state = koji.TASK_STATES[info2['state']] if state != 'OPEN': self.logger.warning("Task %i changed is %s", task_id, state) return False # otherwise... raise if handler.Foreground: self.logger.info("running task in foreground") handler.setManager(self) self.runTask(handler) else: pid, session_id = self.forkTask(handler) self.pids[task_id] = pid self.subsessions[task_id] = session_id return True def forkTask(self, handler): # get the subsession before we fork newhub = self.session.subsession() session_id = newhub.sinfo['session-id'] pid = os.fork() if pid: newhub._forget() return pid, session_id # in no circumstance should we return after the fork # nor should any exceptions propagate past here try: self.session._forget() # set process group os.setpgrp() # use the subsession self.session = newhub handler.session = self.session # set a do-nothing handler for sigusr2 signal.signal(signal.SIGUSR2, lambda *args: None) self.runTask(handler) finally: # diediedie try: self.session.logout() finally: os._exit(0) def runTask(self, handler): try: response = (handler.run(),) # note that we wrap response in a singleton tuple response = koji.xmlrpcplus.dumps(response, methodresponse=1, allow_none=1) self.logger.info("RESPONSE: %r" % response) self.session.host.closeTask(handler.id, response) return except koji.xmlrpcplus.Fault as fault: response = koji.xmlrpcplus.dumps(fault) tb = ''.join(traceback.format_exception(*sys.exc_info())).replace(r"\n", "\n") self.logger.warning("FAULT:\n%s" % tb) except (SystemExit, koji.tasks.ServerExit, KeyboardInterrupt): # we do not trap these raise except koji.tasks.RefuseTask as refuse: self.session.host.refuseTask(handler.id, msg=str(refuse)) return except koji.tasks.ServerRestart: # freeing this task will allow the pending restart to take effect self.session.host.freeTasks([handler.id]) return except Exception: tb = ''.join(traceback.format_exception(*sys.exc_info())) self.logger.warning("TRACEBACK: %s" % tb) # report exception back to server e_class, e = sys.exc_info()[:2] faultCode = getattr(e_class, 'faultCode', 1) if issubclass(e_class, koji.GenericError): # just pass it through tb = str(e) response = koji.xmlrpcplus.dumps(koji.xmlrpcplus.Fault(faultCode, tb)) # if we get here, then we're handling an exception, so fail the task self.session.host.failTask(handler.id, response)