diff options
| -rw-r--r-- | setup.py | 2 | ||||
| -rw-r--r-- | tox.ini | 6 | ||||
| -rwxr-xr-x | zuul-server | 6 | ||||
| -rw-r--r-- | zuul/launcher/jenkins.py | 60 | ||||
| -rw-r--r-- | zuul/lib/gerrit.py | 14 | ||||
| -rw-r--r-- | zuul/model.py | 33 | ||||
| -rw-r--r-- | zuul/scheduler.py | 84 | ||||
| -rw-r--r-- | zuul/trigger/gerrit.py | 6 |
8 files changed, 120 insertions, 91 deletions
@@ -14,9 +14,7 @@ # under the License. from setuptools import find_packages -from setuptools.command.sdist import sdist from setuptools import setup -import subprocess setup(name='zuul', version='1.0', diff --git a/tox.ini b/tox.ini new file mode 100644 index 000000000..2ddaee115 --- /dev/null +++ b/tox.ini @@ -0,0 +1,6 @@ +[tox] +envlist = pep8 + +[testenv:pep8] +deps = pep8 +commands = pep8 --repeat --show-source zuul zuul-server setup.py diff --git a/zuul-server b/zuul-server index f63ae6ad3..bb7a1e6ab 100755 --- a/zuul-server +++ b/zuul-server @@ -23,14 +23,16 @@ import zuul.trigger.gerrit import logging.config + def parse_arguments(): parser = argparse.ArgumentParser(description='Project gating system.') parser.add_argument('-c', dest='config', help='specify the config file') return parser.parse_args() + def read_config(args): - config=ConfigParser.ConfigParser() + config = ConfigParser.ConfigParser() if args.config: locations = [args.config] else: @@ -42,6 +44,7 @@ def read_config(args): return config raise Exception("Unable to locate config file in %s" % locations) + def setup_logging(config): if config.has_option('zuul', 'log_config'): fp = os.path.expanduser(config.get('zuul', 'log_config')) @@ -51,6 +54,7 @@ def setup_logging(config): else: logging.basicConfig(level=logging.DEBUG) + def main(config): sched = zuul.scheduler.Scheduler(config) diff --git a/zuul/launcher/jenkins.py b/zuul/launcher/jenkins.py index 0cda714a7..19012a8a7 100644 --- a/zuul/launcher/jenkins.py +++ b/zuul/launcher/jenkins.py @@ -28,6 +28,7 @@ import pprint from zuul.model import Build + class JenkinsCallback(threading.Thread): log = logging.getLogger("zuul.JenkinsCallback") @@ -61,7 +62,8 @@ class JenkinsCallback(threading.Thread): uuid = params.get('UUID') if (status and url and uuid and phase and phase == 'COMPLETED'): - self.jenkins.onBuildCompleted(uuid, status, url, number) + self.jenkins.onBuildCompleted(uuid, status, url, + number) if (phase and phase == 'STARTED'): self.jenkins.onBuildStarted(uuid, url, number) @@ -70,10 +72,11 @@ STOP_BUILD = 'job/%(name)s/%(number)s/stop' CANCEL_QUEUE = 'queue/item/%(number)s/cancelQueue' BUILD_INFO = 'job/%(name)s/%(number)s/api/json?depth=0' + class ExtendedJenkins(jenkins.Jenkins): def jenkins_open(self, req): ''' - Utility routine for opening an HTTP request to a Jenkins server. + Utility routine for opening an HTTP request to a Jenkins server. ''' try: if self.auth: @@ -93,7 +96,7 @@ class ExtendedJenkins(jenkins.Jenkins): @param number: Jenkins build number for the job @type number: int ''' - self.jenkins_open(urllib2.Request(self.server + STOP_BUILD%locals())) + self.jenkins_open(urllib2.Request(self.server + STOP_BUILD % locals())) def cancel_queue(self, number): ''' @@ -104,10 +107,10 @@ class ExtendedJenkins(jenkins.Jenkins): ''' # Jenkins returns a 302 from this URL, unless Referer is not set, # then you get a 404. - self.jenkins_open(urllib2.Request(self.server + CANCEL_QUEUE%locals(), + self.jenkins_open(urllib2.Request(self.server + + CANCEL_QUEUE % locals(), headers={'Referer': self.server})) - def get_build_info(self, name, number): ''' Get information for a build. @@ -118,7 +121,9 @@ class ExtendedJenkins(jenkins.Jenkins): @type number: int @return: dictionary ''' - return json.loads(self.jenkins_open(urllib2.Request(self.server + BUILD_INFO%locals()))) + return json.loads(self.jenkins_open(urllib2.Request( + self.server + BUILD_INFO % locals()))) + class Jenkins(object): log = logging.getLogger("zuul.Jenkins") @@ -132,27 +137,29 @@ class Jenkins(object): self.jenkins = ExtendedJenkins(server, user, apikey) self.callback_thread = JenkinsCallback(self) self.callback_thread.start() - - def launch(self, job, change, dependent_changes = []): - self.log.info("Launch job %s for change %s with dependent changes %s" % ( - job, change, dependent_changes)) + + def launch(self, job, change, dependent_changes=[]): + self.log.info("Launch job %s for change %s with dependent changes %s" % + (job, change, dependent_changes)) uuid = str(uuid1()) changes_str = '^'.join( - ['%s:%s:%s' % (c.project.name, c.branch, c.refspec) - for c in dependent_changes+[change]]) + ['%s:%s:%s' % (c.project.name, c.branch, c.refspec) + for c in dependent_changes + [change]]) params = dict(UUID=uuid, GERRIT_PROJECT=change.project.name, GERRIT_BRANCH=change.branch, GERRIT_CHANGES=changes_str) build = Build(job, uuid) self.builds[uuid] = build - # We can get the started notification on another thread before this is done - # so we add the build even before we trigger the job on Jenkins. We should - # be careful to clean it up if it doesn't actually kick off. + # We can get the started notification on another thread before + # this is done so we add the build even before we trigger the + # job on Jenkins. We should be careful to clean it up if it + # doesn't actually kick off. try: self.jenkins.build_job(job.name, parameters=params) except: - self.log.exception("Exception launching build %s for job %s for change %s:" % ( + self.log.exception( + "Exception launching build %s for job %s for change %s:" % ( build, job, change)) # Whoops. Remove that build we added. del self.builds[uuid] @@ -171,33 +178,36 @@ class Jenkins(object): self.log.debug("Looking for build %s in queue" % build) for item in self.jenkins.get_queue_info(): - if not item.has_key('actions'): + if 'actions' not in item: continue for action in item['actions']: - if not action.has_key('parameters'): + if 'parameters' not in action: continue parameters = action['parameters'] for param in parameters: - if (param['name'] == 'UUID' and build.uuid == param['value']): + if (param['name'] == 'UUID' and + build.uuid == param['value']): self.log.debug("Found queue item %s for build %s" % ( item['id'], build)) try: self.jenkins.cancel_queue(item['id']) - self.log.debug("Canceled queue item %s for build %s" % ( + self.log.debug( + "Canceled queue item %s for build %s" % ( item['id'], build)) return except: - self.log.exception("Exception canceling queue item %s for build %s" % ( - item['id'], build)) - + self.log.exception("Exception canceling queue \ +item %s for build %s" % (item['id'], build)) self.log.debug("Still unable to find build %s to cancel" % build) if build.number: self.log.debug("Build %s has just started" % build) self.jenkins.stop_build(build.job.name, build.number) self.log.debug("Canceled just running build %s" % build) else: - self.log.error("Build %s has not started but was not found in queue" % build) - + self.log.error( + "Build %s has not started but was not found in queue" % + build) + def onBuildCompleted(self, uuid, status, url, number): self.log.info("Build %s #%s complete, status %s" % ( uuid, number, status)) diff --git a/zuul/lib/gerrit.py b/zuul/lib/gerrit.py index 15c7c0583..a8838dd43 100644 --- a/zuul/lib/gerrit.py +++ b/zuul/lib/gerrit.py @@ -1,3 +1,4 @@ +# Copyright 2011 OpenStack, LLC. # Copyright 2012 Hewlett-Packard Development Company, L.P. # # Licensed under the Apache License, Version 2.0 (the "License"); you may @@ -24,6 +25,7 @@ import pprint # TODO: switch this to paramiko? + class GerritWatcher(threading.Thread): log = logging.getLogger("gerrit.GerritWatcher") @@ -45,7 +47,7 @@ class GerritWatcher(threading.Thread): cmd += ['-l', self.username, self.server, 'gerrit', 'stream-events'] self.proc = subprocess.Popen(cmd, - bufsize=1, + bufsize=1, stdin=None, stdout=subprocess.PIPE, stderr=None, @@ -67,7 +69,8 @@ class GerritWatcher(threading.Thread): def _read(self): l = self.proc.stdout.readline() data = json.loads(l) - self.log.debug("Received data from Gerrit event stream: \n%s" % pprint.pformat(data)) + self.log.debug("Received data from Gerrit event stream: \n%s" % + pprint.pformat(data)) self.gerrit.addEvent(data) def _listen(self): @@ -79,7 +82,7 @@ class GerritWatcher(threading.Thread): self._read() else: raise Exception("event on ssh connection") - + def _run(self): try: if not self.proc: @@ -94,6 +97,7 @@ class GerritWatcher(threading.Thread): while True: self._run() + class Gerrit(object): log = logging.getLogger("gerrit.Gerrit") @@ -122,7 +126,7 @@ class Gerrit(object): def review(self, project, change, message, action={}): cmd = 'gerrit review --project %s --message "%s"' % ( project, message) - for k,v in action.items(): + for k, v in action.items(): if v is True: cmd += ' --%s' % k else: @@ -151,7 +155,7 @@ class Gerrit(object): client = paramiko.SSHClient() client.load_system_host_keys() client.set_missing_host_key_policy(paramiko.WarningPolicy()) - client.connect(self.hostname, + client.connect(self.hostname, username=self.username, port=29418) diff --git a/zuul/model.py b/zuul/model.py index aafcdd643..2b7be6bd6 100644 --- a/zuul/model.py +++ b/zuul/model.py @@ -14,6 +14,7 @@ import re + class ChangeQueue(object): def __init__(self, queue_name): self.name = '' @@ -51,6 +52,7 @@ class ChangeQueue(object): for project in other.projects: self.addProject(project) + class Job(object): def __init__(self, name): self.name = name @@ -64,6 +66,7 @@ class Job(object): def __repr__(self): return '<Job %s>' % (self.name) + class Build(object): def __init__(self, job, uuid): self.job = job @@ -75,6 +78,7 @@ class Build(object): def __repr__(self): return '<Build %s of %s>' % (self.uuid, self.job.name) + class JobTree(object): """ A JobTree represents an instance of one Job, and holds JobTrees whose jobs should be run if that Job succeeds. A root node of a @@ -106,6 +110,7 @@ class JobTree(object): return ret return None + class Project(object): def __init__(self, name): self.name = name @@ -122,7 +127,7 @@ class Project(object): return self.job_trees[name] def hasQueue(self, name): - if self.job_trees.has_key(name): + if name in self.job_trees: return True return False @@ -135,6 +140,7 @@ class Project(object): return [] return tree.getJobs() + class Change(object): def __init__(self, queue_name, project, branch, number, patchset, refspec): self.queue_name = queue_name @@ -153,18 +159,18 @@ class Change(object): return '<Change 0x%x %s,%s>' % (id(self), self.number, self.patchset) def formatStatus(self, indent=0): - indent_str = ' '*indent + indent_str = ' ' * indent ret = '' ret += '%sProject %s change %s,%s\n' % (indent_str, self.project.name, - self.number, + self.number, self.patchset) for job in self.project.getJobs(self.queue_name): result = self.jobs.get(job.name) ret += '%s %s: %s\n' % (indent_str, job.name, result) if self.change_ahead: ret += '%sWaiting on:\n' % (indent_str) - ret += self.change_ahead.formatStatus(indent+2) + ret += self.change_ahead.formatStatus(indent + 2) return ret def formatReport(self): @@ -173,7 +179,7 @@ class Change(object): ret += 'Build successful\n\n' else: ret += 'Build failed\n\n' - + for job in self.project.getJobs(self.queue_name): result = self.jobs.get(job.name) url = self.job_urls.get(job.name, job.name) @@ -222,7 +228,7 @@ class Change(object): def areAllJobsComplete(self): tree = self.project.getJobTreeForQueue(self.queue_name) for job in tree.getJobs(): - if not self.jobs.has_key(job.name): + if not job.name in self.jobs: return False return True @@ -236,6 +242,7 @@ class Change(object): if self.change_behind: self.change_behind.change_ahead = None + class TriggerEvent(object): def __init__(self): self.data = None @@ -250,17 +257,19 @@ class TriggerEvent(object): def __str__(self): ret = '<TriggerEvent %s %s' % (self.type, self.project_name) - + if self.branch: ret += " %s" % self.branch if self.change_number: ret += " %s,%s" % (self.change_number, self.patch_number) if self.approvals: - ret += ' '+', '.join(['%s:%s' % (a['type'], a['value']) for a in self.approvals]) + ret += ' ' + ', '.join( + ['%s:%s' % (a['type'], a['value']) for a in self.approvals]) ret += '>' return ret + class EventFilter(object): def __init__(self, types=[], branches=[], refs=[], approvals=[]): self._types = types @@ -273,7 +282,7 @@ class EventFilter(object): def __str__(self): ret = '<EventFilter' - + if self._types: ret += ' types: %s' % ', '.join(self._types) if self._branches: @@ -281,7 +290,8 @@ class EventFilter(object): if self._refs: ret += ' refs: %s' % ', '.join(self._refs) if self.approvals: - ret += ' approvals: %s' % ', '.join(['%s:%s' % a for a in self.approvals.items()]) + ret += ' approvals: %s' % ', '.join( + ['%s:%s' % a for a in self.approvals.items()]) ret += '>' return ret @@ -322,5 +332,6 @@ class EventFilter(object): if (normalizeCategory(eapproval['description']) == category and int(eapproval['value']) == int(value)): matches_approval = True - if not matches_approval: return False + if not matches_approval: + return False return True diff --git a/zuul/scheduler.py b/zuul/scheduler.py index 0760c2868..4084bb555 100644 --- a/zuul/scheduler.py +++ b/zuul/scheduler.py @@ -18,9 +18,9 @@ import threading import logging import yaml - from model import Job, Change, Project, ChangeQueue, EventFilter + class Scheduler(object): log = logging.getLogger("zuul.Scheduler") @@ -39,7 +39,8 @@ class Scheduler(object): def _parseConfig(self, fp): def toList(item): - if not item: return [] + if not item: + return [] if type(item) == type([]): return item return [item] @@ -50,11 +51,10 @@ class Scheduler(object): raise Exception("Unable to read layout config file at %s" % fp) fp = open(fp) data = yaml.load(fp) - import pprint - pprint.pprint(data) - + for config_queue in data['queue']: - manager = globals()[config_queue['manager']](self, config_queue['name']) + manager = globals()[config_queue['manager']](self, + config_queue['name']) self.queue_managers[config_queue['name']] = manager manager.success_action = config_queue.get('success') manager.failure_action = config_queue.get('failure') @@ -62,7 +62,7 @@ class Scheduler(object): approvals = {} for approval_dict in toList(trigger.get('approval')): for k, v in approval_dict.items(): - approvals[k]=v + approvals[k] = v f = EventFilter(types=toList(trigger['event']), branches=toList(trigger.get('branch')), refs=toList(trigger.get('ref')), @@ -97,12 +97,10 @@ class Scheduler(object): project = Project(config_project['name']) self.projects[config_project['name']] = project for qname in self.queue_managers.keys(): - if config_project.has_key(qname): - print project, qname + if qname in config_project: job_tree = project.addQueue(qname) config_jobs = config_project[qname] add_jobs(job_tree, config_jobs) - # TODO(jeblair): check that we don't end up with jobs like # "foo - bar" because a ':' is missing in the yaml for a dependent job @@ -110,7 +108,7 @@ class Scheduler(object): manager._postConfig() def getJob(self, name): - if self.jobs.has_key(name): + if name in self.jobs: return self.jobs[name] job = Job(name) self.jobs[name] = job @@ -166,7 +164,7 @@ class Scheduler(object): self.log.info("Adding %s, %s to to %s" % ( project, change, manager)) manager.addChange(change) - + def process_result_queue(self): self.log.debug("Fetching result event") build = self.result_event_queue.get() @@ -175,6 +173,7 @@ class Scheduler(object): if manager.onBuildCompleted(build): return + class BaseQueueManager(object): log = logging.getLogger("zuul.BaseQueueManager") @@ -195,17 +194,19 @@ class BaseQueueManager(object): for e in self.event_filters: self.log.info(" %s" % e) self.log.info(" Projects:") + def log_jobs(tree, indent=0): - istr = ' '+' '*indent + istr = ' ' + ' ' * indent if tree.job: efilters = '' for e in tree.job.event_filters: efilters += str(e) if efilters: - efilters = ' '+efilters + efilters = ' ' + efilters self.log.info("%s%s%s" % (istr, repr(tree.job), efilters)) for x in tree.job_trees: - log_jobs(x, indent+2) + log_jobs(x, indent + 2) + for p in self.sched.projects.values(): if p.hasQueue(self.name): self.log.info(" %s" % p) @@ -217,15 +218,12 @@ class BaseQueueManager(object): self.log.info(" On failure:") self.log.info(" %s" % self.failure_action) - def eventMatches(self, event): for ef in self.event_filters: - print ef if ef.matches(event): return True return False - def addChange(self, change): self.log.debug("Adding change %s" % change) self.launchJobs(change) @@ -241,12 +239,12 @@ class BaseQueueManager(object): build, job, change)) change.addBuild(build) except: - self.log.exception("Exception while launching job %s for change %s:" % ( - job, change)) + self.log.exception("Exception while launching job %s \ +for change %s:" % (job, change)) def onBuildCompleted(self, build): self.log.debug("Build %s completed" % build) - if not self.building_jobs.has_key(build): + if build not in self.building_jobs: self.log.warning("Build %s not found (may have been canceled)" % ( build)) # Or triggered externally, or triggered before zuul started, @@ -291,15 +289,16 @@ class BaseQueueManager(object): if ret: self.log.error("Reporting change %s received: %s" % ( change, ret)) - print ret except: self.log.exception("Exception while reporting:") return ret + class IndependentQueueManager(BaseQueueManager): log = logging.getLogger("zuul.IndependentQueueManager") pass + class DependentQueueManager(BaseQueueManager): log = logging.getLogger("zuul.DependentQueueManager") @@ -335,7 +334,7 @@ class DependentQueueManager(BaseQueueManager): if not merged_a: self.log.debug("Keeping queue %s" % (a)) new_change_queues.append(a) - + self.change_queues = new_change_queues self.log.info(" Shared change queues:") for x in self.change_queues: @@ -346,7 +345,7 @@ class DependentQueueManager(BaseQueueManager): if project in queue.projects: return queue self.log.error("Unable to find change queue for project %s" % project) - + def addChange(self, change): self.log.debug("Adding change %s" % change) change_queue = self.getQueue(change.project) @@ -368,15 +367,15 @@ class DependentQueueManager(BaseQueueManager): for job in change.findJobsToRun(): self.log.debug("Found job %s for change %s" % (job, change)) try: - build = self.sched.launcher.launch(job, change, + build = self.sched.launcher.launch(job, change, dependent_changes) self.building_jobs[build] = change self.log.debug("Adding build %s of job %s to change %s" % ( build, job, change)) change.addBuild(build) except: - self.log.exception("Exception while launching job %s for change %s:" % ( - job, change)) + self.log.exception("Exception while launching job %s \ +for change %s:" % (job, change)) if change.change_behind: self.log.debug("Launching jobs for change %s, behind change %s" % ( change.change_behind, change)) @@ -393,22 +392,22 @@ class DependentQueueManager(BaseQueueManager): try: self.sched.launcher.cancel(build) except: - self.log.exception("Exception while canceling build %s for change %s" % ( - build, change)) + self.log.exception("Exception while canceling build %s \ +for change %s" % (build, change)) to_remove.append(build) for build in to_remove: self.log.debug("Removing build %s from running builds" % build) del self.building_jobs[build] if change.change_behind: - self.log.debug("Canceling jobs for change %s, behind change %s" % ( - change.change_behind, change)) + self.log.debug("Canceling jobs for change %s, \ +behind change %s" % (change.change_behind, change)) self.cancelJobs(change.change_behind) def possiblyReportChange(self, change): self.log.debug("Possibly reporting change %s" % change) if not change.change_ahead: - self.log.debug("Change %s is at the front of the queue, reporting" % ( - change)) + self.log.debug("Change %s is at the front of the queue, \ +reporting" % (change)) ret = self.reportChange(change) self.log.debug("Removing reported change %s from queue" % change) change.delete() @@ -417,21 +416,22 @@ class DependentQueueManager(BaseQueueManager): if merged: merged = self.sched.trigger.isMerged(change) succeeded = change.didAllJobsSucceed() - self.log.info("Reported change %s status: all-succeeded: %s, merged: %s" % ( - change, succeeded, merged)) - + self.log.info("Reported change %s status: all-succeeded: %s, \ +merged: %s" % (change, succeeded, merged)) + if not (succeeded and merged): - self.log.debug("Reported change %s failed tests or failed to merge" % ( - change)) + self.log.debug("Reported change %s failed tests or failed \ +to merge" % (change)) # The merge or test failed, re-run all jobs behind this one if change.change_behind: - self.log.info("Canceling/relaunching jobs for change %s behind failed change %s" % ( + self.log.info("Canceling/relaunching jobs for change %s \ +behind failed change %s" % ( change.change_behind, change)) self.cancelJobs(change.change_behind) self.launchJobs(change.change_behind) # If the change behind this is ready, notify - if (change.change_behind and + if (change.change_behind and change.change_behind.areAllJobsComplete()): - self.log.info("Change %s behind change %s is ready, possibly reporting" % ( - change.change_behind, change)) + self.log.info("Change %s behind change %s is ready, \ +possibly reporting" % (change.change_behind, change)) self.possiblyReportChange(change.change_behind) diff --git a/zuul/trigger/gerrit.py b/zuul/trigger/gerrit.py index 36b1618cd..cd3ff81c4 100644 --- a/zuul/trigger/gerrit.py +++ b/zuul/trigger/gerrit.py @@ -17,6 +17,7 @@ import logging from zuul.lib import gerrit from zuul.model import TriggerEvent + class GerritEventConnector(threading.Thread): """Move events from Gerrit to the scheduler.""" @@ -27,7 +28,6 @@ class GerritEventConnector(threading.Thread): self.gerrit = gerrit self.sched = sched - def _handleEvent(self): data = self.gerrit.getEvent() event = TriggerEvent() @@ -44,7 +44,6 @@ class GerritEventConnector(threading.Thread): event.approvals = data.get('approvals') self.sched.addEvent(event) - def run(self): while True: try: @@ -70,7 +69,6 @@ class Gerrit(object): self.gerrit, sched) self.gerrit_connector.start() - def report(self, change, message, action): self.log.debug("Report change %s, action %s, message: %s" % (change, action, message)) @@ -78,7 +76,6 @@ class Gerrit(object): return self.gerrit.review(change.project.name, changeid, message, action) - def isMerged(self, change): self.log.debug("Checking if change %s is merged", change) data = self.gerrit.query(change.number) @@ -90,4 +87,3 @@ class Gerrit(object): self.log.debug("Change %s status: %s" % (change, status)) if status == 'MERGED' or status == 'SUBMITTED': return True - |