| # Copyright 2012 Hewlett-Packard Development Company, L.P. |
| # Copyright 2013 OpenStack Foundation |
| # Copyright 2013 Antoine "hashar" Musso |
| # Copyright 2013 Wikimedia Foundation Inc. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); you may |
| # not use this file except in compliance with the License. You may obtain |
| # a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| # License for the specific language governing permissions and limitations |
| # under the License. |
| |
| import extras |
| import json |
| import logging |
| import os |
| import pickle |
| import Queue |
| import re |
| import threading |
| import time |
| import yaml |
| |
| import layoutvalidator |
| import model |
| from model import ActionReporter, Pipeline, Project, ChangeQueue, EventFilter |
| import merger |
| |
| statsd = extras.try_import('statsd.statsd') |
| |
| |
| def deep_format(obj, paramdict): |
| """Apply the paramdict via str.format() to all string objects found within |
| the supplied obj. Lists and dicts are traversed recursively. |
| |
| Borrowed from Jenkins Job Builder project""" |
| if isinstance(obj, str): |
| ret = obj.format(**paramdict) |
| elif isinstance(obj, list): |
| ret = [] |
| for item in obj: |
| ret.append(deep_format(item, paramdict)) |
| elif isinstance(obj, dict): |
| ret = {} |
| for item in obj: |
| exp_item = item.format(**paramdict) |
| |
| ret[exp_item] = deep_format(obj[item], paramdict) |
| else: |
| ret = obj |
| return ret |
| |
| |
| class MergeFailure(Exception): |
| pass |
| |
| |
| class Scheduler(threading.Thread): |
| log = logging.getLogger("zuul.Scheduler") |
| |
| def __init__(self): |
| threading.Thread.__init__(self) |
| self.daemon = True |
| self.wake_event = threading.Event() |
| self.layout_lock = threading.Lock() |
| self.reconfigure_complete_event = threading.Event() |
| self._pause = False |
| self._reconfigure = False |
| self._exit = False |
| self._stopped = False |
| self.launcher = None |
| self.triggers = dict() |
| self.reporters = dict() |
| self.config = None |
| self._maintain_trigger_cache = False |
| |
| self.trigger_event_queue = Queue.Queue() |
| self.result_event_queue = Queue.Queue() |
| self.layout = model.Layout() |
| |
| def stop(self): |
| self._stopped = True |
| self.wake_event.set() |
| |
| def testConfig(self, config_path): |
| return self._parseConfig(config_path) |
| |
| def _parseConfig(self, config_path): |
| layout = model.Layout() |
| project_templates = {} |
| |
| def toList(item): |
| if not item: |
| return [] |
| if isinstance(item, list): |
| return item |
| return [item] |
| |
| if config_path: |
| config_path = os.path.expanduser(config_path) |
| if not os.path.exists(config_path): |
| raise Exception("Unable to read layout config file at %s" % |
| config_path) |
| config_file = open(config_path) |
| data = yaml.load(config_file) |
| |
| validator = layoutvalidator.LayoutValidator() |
| validator.validate(data) |
| |
| config_env = {} |
| for include in data.get('includes', []): |
| if 'python-file' in include: |
| fn = include['python-file'] |
| if not os.path.isabs(fn): |
| base = os.path.dirname(config_path) |
| fn = os.path.join(base, fn) |
| fn = os.path.expanduser(fn) |
| execfile(fn, config_env) |
| |
| for conf_pipeline in data.get('pipelines', []): |
| pipeline = Pipeline(conf_pipeline['name']) |
| pipeline.description = conf_pipeline.get('description') |
| precedence = model.PRECEDENCE_MAP[conf_pipeline.get('precedence')] |
| pipeline.precedence = precedence |
| pipeline.failure_message = conf_pipeline.get('failure-message', |
| "Build failed.") |
| pipeline.success_message = conf_pipeline.get('success-message', |
| "Build succeeded.") |
| pipeline.dequeue_on_new_patchset = conf_pipeline.get( |
| 'dequeue-on-new-patchset', True) |
| |
| action_reporters = {} |
| for action in ['start', 'success', 'failure']: |
| action_reporters[action] = [] |
| if conf_pipeline.get(action): |
| for reporter_name, params \ |
| in conf_pipeline.get(action).items(): |
| if reporter_name in self.reporters.keys(): |
| action_reporters[action].append(ActionReporter( |
| self.reporters[reporter_name], params)) |
| else: |
| self.log.error('Invalid reporter name %s' % |
| reporter_name) |
| pipeline.start_actions = action_reporters['start'] |
| pipeline.success_actions = action_reporters['success'] |
| pipeline.failure_actions = action_reporters['failure'] |
| |
| manager = globals()[conf_pipeline['manager']](self, pipeline) |
| pipeline.setManager(manager) |
| layout.pipelines[conf_pipeline['name']] = pipeline |
| |
| # TODO: move this into triggers (may require pluggable |
| # configuration) |
| if 'gerrit' in conf_pipeline['trigger']: |
| pipeline.trigger = self.triggers['gerrit'] |
| for trigger in toList(conf_pipeline['trigger']['gerrit']): |
| approvals = {} |
| for approval_dict in toList(trigger.get('approval')): |
| for k, v in approval_dict.items(): |
| approvals[k] = v |
| f = EventFilter(types=toList(trigger['event']), |
| branches=toList(trigger.get('branch')), |
| refs=toList(trigger.get('ref')), |
| approvals=approvals, |
| comment_filters= |
| toList(trigger.get('comment_filter')), |
| email_filters= |
| toList(trigger.get('email_filter'))) |
| manager.event_filters.append(f) |
| elif 'timer' in conf_pipeline['trigger']: |
| pipeline.trigger = self.triggers['timer'] |
| for trigger in toList(conf_pipeline['trigger']['timer']): |
| f = EventFilter(types=['timer'], |
| timespecs=toList(trigger['time'])) |
| manager.event_filters.append(f) |
| |
| for project_template in data.get('project-templates', []): |
| # Make sure the template only contains valid pipelines |
| tpl = dict( |
| (pipe_name, project_template.get(pipe_name)) |
| for pipe_name in layout.pipelines.keys() |
| if pipe_name in project_template |
| ) |
| project_templates[project_template.get('name')] = tpl |
| |
| for config_job in data.get('jobs', []): |
| job = layout.getJob(config_job['name']) |
| # Be careful to only set attributes explicitly present on |
| # this job, to avoid squashing attributes set by a meta-job. |
| m = config_job.get('failure-message', None) |
| if m: |
| job.failure_message = m |
| m = config_job.get('success-message', None) |
| if m: |
| job.success_message = m |
| m = config_job.get('failure-pattern', None) |
| if m: |
| job.failure_pattern = m |
| m = config_job.get('success-pattern', None) |
| if m: |
| job.success_pattern = m |
| m = config_job.get('hold-following-changes', False) |
| if m: |
| job.hold_following_changes = True |
| m = config_job.get('voting', None) |
| if m is not None: |
| job.voting = m |
| fname = config_job.get('parameter-function', None) |
| if fname: |
| func = config_env.get(fname, None) |
| if not func: |
| raise Exception("Unable to find function %s" % fname) |
| job.parameter_function = func |
| branches = toList(config_job.get('branch')) |
| if branches: |
| job._branches = branches |
| job.branches = [re.compile(x) for x in branches] |
| files = toList(config_job.get('files')) |
| if files: |
| job._files = files |
| job.files = [re.compile(x) for x in files] |
| |
| def add_jobs(job_tree, config_jobs): |
| for job in config_jobs: |
| if isinstance(job, list): |
| for x in job: |
| add_jobs(job_tree, x) |
| if isinstance(job, dict): |
| for parent, children in job.items(): |
| parent_tree = job_tree.addJob(layout.getJob(parent)) |
| add_jobs(parent_tree, children) |
| if isinstance(job, str): |
| job_tree.addJob(layout.getJob(job)) |
| |
| for config_project in data.get('projects', []): |
| project = Project(config_project['name']) |
| |
| for requested_template in config_project.get('template', []): |
| # Fetch the template from 'project-templates' |
| tpl = project_templates.get( |
| requested_template.get('name')) |
| # Expand it with the project context |
| expanded = deep_format(tpl, requested_template) |
| # Finally merge the expansion with whatever has been already |
| # defined for this project |
| config_project.update(expanded) |
| |
| layout.projects[config_project['name']] = project |
| mode = config_project.get('merge-mode', 'merge-resolve') |
| project.merge_mode = model.MERGER_MAP[mode] |
| for pipeline in layout.pipelines.values(): |
| if pipeline.name in config_project: |
| job_tree = pipeline.addProject(project) |
| config_jobs = config_project[pipeline.name] |
| add_jobs(job_tree, config_jobs) |
| |
| # All jobs should be defined at this point, get rid of |
| # metajobs so that getJob isn't doing anything weird. |
| layout.metajobs = [] |
| |
| for pipeline in layout.pipelines.values(): |
| pipeline.manager._postConfig(layout) |
| |
| return layout |
| |
| def _setupMerger(self): |
| if self.config.has_option('zuul', 'git_dir'): |
| merge_root = self.config.get('zuul', 'git_dir') |
| else: |
| merge_root = '/var/lib/zuul/git' |
| |
| if self.config.has_option('zuul', 'git_user_email'): |
| merge_email = self.config.get('zuul', 'git_user_email') |
| else: |
| merge_email = None |
| |
| if self.config.has_option('zuul', 'git_user_name'): |
| merge_name = self.config.get('zuul', 'git_user_name') |
| else: |
| merge_name = None |
| |
| if self.config.has_option('zuul', 'push_change_refs'): |
| push_refs = self.config.getboolean('zuul', 'push_change_refs') |
| else: |
| push_refs = False |
| |
| if self.config.has_option('gerrit', 'sshkey'): |
| sshkey = self.config.get('gerrit', 'sshkey') |
| else: |
| sshkey = None |
| |
| # TODO: The merger should have an upstream repo independent of |
| # triggers, and then each trigger should provide a fetch |
| # location. |
| self.merger = merger.Merger(self.triggers['gerrit'], |
| merge_root, push_refs, |
| sshkey, merge_email, merge_name) |
| for project in self.layout.projects.values(): |
| url = self.triggers['gerrit'].getGitUrl(project) |
| self.merger.addProject(project, url) |
| |
| def setLauncher(self, launcher): |
| self.launcher = launcher |
| |
| def registerTrigger(self, trigger, name=None): |
| if name is None: |
| name = trigger.name |
| self.triggers[name] = trigger |
| |
| def registerReporter(self, reporter, name=None): |
| if name is None: |
| name = reporter.name |
| self.reporters[name] = reporter |
| |
| def getProject(self, name): |
| self.layout_lock.acquire() |
| p = None |
| try: |
| p = self.layout.projects.get(name) |
| finally: |
| self.layout_lock.release() |
| return p |
| |
| def addEvent(self, event): |
| self.log.debug("Adding trigger event: %s" % event) |
| try: |
| if statsd: |
| statsd.incr('gerrit.event.%s' % event.type) |
| except: |
| self.log.exception("Exception reporting event stats") |
| self.trigger_event_queue.put(event) |
| self.wake_event.set() |
| self.log.debug("Done adding trigger event: %s" % event) |
| |
| def onBuildStarted(self, build): |
| self.log.debug("Adding start event for build: %s" % build) |
| build.start_time = time.time() |
| self.result_event_queue.put(('started', build)) |
| self.wake_event.set() |
| self.log.debug("Done adding start event for build: %s" % build) |
| |
| def onBuildCompleted(self, build): |
| self.log.debug("Adding complete event for build: %s" % build) |
| build.end_time = time.time() |
| try: |
| if statsd and build.pipeline: |
| jobname = build.job.name.replace('.', '_') |
| key = 'zuul.pipeline.%s.job.%s.%s' % (build.pipeline.name, |
| jobname, build.result) |
| if build.result in ['SUCCESS', 'FAILURE'] and build.start_time: |
| dt = int((build.end_time - build.start_time) * 1000) |
| statsd.timing(key, dt) |
| statsd.incr(key) |
| key = 'zuul.pipeline.%s.all_jobs' % build.pipeline.name |
| statsd.incr(key) |
| except: |
| self.log.exception("Exception reporting runtime stats") |
| self.result_event_queue.put(('completed', build)) |
| self.wake_event.set() |
| self.log.debug("Done adding complete event for build: %s" % build) |
| |
| def reconfigure(self, config): |
| self.log.debug("Prepare to reconfigure") |
| self.config = config |
| self._reconfigure = True |
| self.wake_event.set() |
| self.log.debug("Waiting for reconfiguration") |
| self.reconfigure_complete_event.wait() |
| self.reconfigure_complete_event.clear() |
| self.log.debug("Reconfiguration complete") |
| |
| def exit(self): |
| self.log.debug("Prepare to exit") |
| self._pause = True |
| self._exit = True |
| self.wake_event.set() |
| self.log.debug("Waiting for exit") |
| |
| def _get_queue_pickle_file(self): |
| if self.config.has_option('zuul', 'state_dir'): |
| state_dir = os.path.expanduser(self.config.get('zuul', |
| 'state_dir')) |
| else: |
| state_dir = '/var/lib/zuul' |
| return os.path.join(state_dir, 'queue.pickle') |
| |
| def _save_queue(self): |
| pickle_file = self._get_queue_pickle_file() |
| events = [] |
| while not self.trigger_event_queue.empty(): |
| events.append(self.trigger_event_queue.get()) |
| self.log.debug("Queue length is %s" % len(events)) |
| if events: |
| self.log.debug("Saving queue") |
| pickle.dump(events, open(pickle_file, 'wb')) |
| |
| def _load_queue(self): |
| pickle_file = self._get_queue_pickle_file() |
| if os.path.exists(pickle_file): |
| self.log.debug("Loading queue") |
| events = pickle.load(open(pickle_file, 'rb')) |
| self.log.debug("Queue length is %s" % len(events)) |
| for event in events: |
| self.trigger_event_queue.put(event) |
| else: |
| self.log.debug("No queue file found") |
| |
| def _delete_queue(self): |
| pickle_file = self._get_queue_pickle_file() |
| if os.path.exists(pickle_file): |
| self.log.debug("Deleting saved queue") |
| os.unlink(pickle_file) |
| |
| def resume(self): |
| try: |
| self._load_queue() |
| except: |
| self.log.exception("Unable to load queue") |
| try: |
| self._delete_queue() |
| except: |
| self.log.exception("Unable to delete saved queue") |
| self.log.debug("Resuming queue processing") |
| self.wake_event.set() |
| |
| def _doPauseEvent(self): |
| if self._exit: |
| self.log.debug("Exiting") |
| self._save_queue() |
| os._exit(0) |
| |
| def _doReconfigureEvent(self): |
| # This is called in the scheduler loop after another thread sets |
| # the reconfigure flag |
| self.layout_lock.acquire() |
| try: |
| self.log.debug("Performing reconfiguration") |
| layout = self._parseConfig( |
| self.config.get('zuul', 'layout_config')) |
| for name, new_pipeline in layout.pipelines.items(): |
| old_pipeline = self.layout.pipelines.get(name) |
| if not old_pipeline: |
| if self.layout.pipelines: |
| # Don't emit this warning on startup |
| self.log.warning("No old pipeline matching %s found " |
| "when reconfiguring" % name) |
| continue |
| self.log.debug("Re-enqueueing changes for pipeline %s" % |
| name) |
| items_to_remove = [] |
| for shared_queue in old_pipeline.queues: |
| for item in shared_queue.queue: |
| item.item_ahead = None |
| item.items_behind = [] |
| item.pipeline = None |
| project = layout.projects.get(item.change.project.name) |
| if not project: |
| self.log.warning("Unable to find project for " |
| "change %s while reenqueueing" % |
| item.change) |
| item.change.project = None |
| items_to_remove.append(item) |
| continue |
| item.change.project = project |
| if not new_pipeline.manager.reEnqueueItem(item): |
| items_to_remove.append(item) |
| builds_to_remove = [] |
| for build, item in old_pipeline.manager.building_jobs.items(): |
| if item in items_to_remove: |
| builds_to_remove.append(build) |
| self.log.warning("Deleting running build %s for " |
| "change %s while reenqueueing" % ( |
| build, item.change)) |
| for build in builds_to_remove: |
| del old_pipeline.manager.building_jobs[build] |
| new_pipeline.manager.building_jobs = \ |
| old_pipeline.manager.building_jobs |
| self.layout = layout |
| self._setupMerger() |
| for trigger in self.triggers.values(): |
| trigger.postConfig() |
| if statsd: |
| try: |
| for pipeline in self.layout.pipelines.values(): |
| items = len(pipeline.getAllItems()) |
| # stats.gauges.zuul.pipeline.NAME.current_changes |
| key = 'zuul.pipeline.%s' % pipeline.name |
| statsd.gauge(key + '.current_changes', items) |
| except Exception: |
| self.log.exception("Exception reporting initial " |
| "pipeline stats:") |
| self._reconfigure = False |
| self.reconfigure_complete_event.set() |
| finally: |
| self.layout_lock.release() |
| |
| def _areAllBuildsComplete(self): |
| self.log.debug("Checking if all builds are complete") |
| waiting = False |
| for pipeline in self.layout.pipelines.values(): |
| for build in pipeline.manager.building_jobs.keys(): |
| self.log.debug("%s waiting on %s" % (pipeline.manager, build)) |
| waiting = True |
| if not waiting: |
| self.log.debug("All builds are complete") |
| return True |
| self.log.debug("All builds are not complete") |
| return False |
| |
| def run(self): |
| if statsd: |
| self.log.debug("Statsd enabled") |
| else: |
| self.log.debug("Statsd disabled because python statsd " |
| "package not found") |
| while True: |
| self.log.debug("Run handler sleeping") |
| self.wake_event.wait() |
| self.wake_event.clear() |
| if self._stopped: |
| return |
| self.log.debug("Run handler awake") |
| try: |
| if self._reconfigure: |
| self._doReconfigureEvent() |
| |
| # Give result events priority -- they let us stop builds, |
| # whereas trigger evensts cause us to launch builds. |
| if not self.result_event_queue.empty(): |
| self.process_result_queue() |
| elif not self._pause: |
| if not self.trigger_event_queue.empty(): |
| self.process_event_queue() |
| |
| if self._pause and self._areAllBuildsComplete(): |
| self._doPauseEvent() |
| |
| if not self._pause: |
| if not (self.trigger_event_queue.empty() and |
| self.result_event_queue.empty()): |
| self.wake_event.set() |
| else: |
| if not self.result_event_queue.empty(): |
| self.wake_event.set() |
| |
| if self._maintain_trigger_cache: |
| self.maintainTriggerCache() |
| self._maintain_trigger_cache = False |
| |
| except: |
| self.log.exception("Exception in run handler:") |
| |
| def maintainTriggerCache(self): |
| relevant = set() |
| for pipeline in self.layout.pipelines.values(): |
| self.log.debug("Start maintain trigger cache for: %s" % pipeline) |
| for item in pipeline.getAllItems(): |
| relevant.add(item.change) |
| relevant.update(item.change.getRelatedChanges()) |
| self.log.debug("End maintain trigger cache for: %s" % pipeline) |
| self.log.debug("Trigger cache size: %s" % len(relevant)) |
| for trigger in self.triggers.values(): |
| trigger.maintainCache(relevant) |
| |
| def process_event_queue(self): |
| self.log.debug("Fetching trigger event") |
| event = self.trigger_event_queue.get() |
| self.log.debug("Processing trigger event %s" % event) |
| project = self.layout.projects.get(event.project_name) |
| if not project: |
| self.log.warning("Project %s not found" % event.project_name) |
| self.trigger_event_queue.task_done() |
| return |
| |
| # Preprocessing for ref-update events |
| if event.ref: |
| # Make sure the local git repo is up-to-date with the remote one. |
| # We better have the new ref before enqueuing the changes. |
| # This is done before enqueuing the changes to avoid calling an |
| # update per pipeline accepting the change. |
| self.log.info("Fetching references for %s" % project) |
| self.merger.updateRepo(project) |
| |
| for pipeline in self.layout.pipelines.values(): |
| change = event.getChange(project, |
| self.triggers.get(event.trigger_name)) |
| if event.type == 'patchset-created': |
| pipeline.manager.removeOldVersionsOfChange(change) |
| if pipeline.manager.eventMatches(event): |
| self.log.info("Adding %s, %s to %s" % |
| (project, change, pipeline)) |
| pipeline.manager.addChange(change) |
| while pipeline.manager.processQueue(): |
| pass |
| |
| self.trigger_event_queue.task_done() |
| |
| def process_result_queue(self): |
| self.log.debug("Fetching result event") |
| event_type, build = self.result_event_queue.get() |
| self.log.debug("Processing result event %s" % build) |
| for pipeline in self.layout.pipelines.values(): |
| if event_type == 'started': |
| if pipeline.manager.onBuildStarted(build): |
| self.result_event_queue.task_done() |
| return |
| elif event_type == 'completed': |
| if pipeline.manager.onBuildCompleted(build): |
| self.result_event_queue.task_done() |
| return |
| self.log.warning("Build %s not found by any queue manager" % (build)) |
| self.result_event_queue.task_done() |
| |
| def formatStatusHTML(self): |
| ret = '<html><pre>' |
| if self._pause: |
| ret += '<p><b>Queue only mode:</b> preparing to ' |
| if self._exit: |
| ret += 'exit' |
| ret += ', queue length: %s' % self.trigger_event_queue.qsize() |
| ret += '</p>' |
| |
| keys = self.layout.pipelines.keys() |
| for key in keys: |
| pipeline = self.layout.pipelines[key] |
| s = 'Pipeline: %s' % pipeline.name |
| ret += s + '\n' |
| ret += '-' * len(s) + '\n' |
| ret += pipeline.formatStatusHTML() |
| ret += '\n' |
| ret += '</pre></html>' |
| return ret |
| |
| def formatStatusJSON(self): |
| data = {} |
| if self._pause: |
| ret = '<p><b>Queue only mode:</b> preparing to ' |
| if self._exit: |
| ret += 'exit' |
| ret += ', queue length: %s' % self.trigger_event_queue.qsize() |
| ret += '</p>' |
| data['message'] = ret |
| |
| data['trigger_event_queue'] = {} |
| data['trigger_event_queue']['length'] = \ |
| self.trigger_event_queue.qsize() |
| data['result_event_queue'] = {} |
| data['result_event_queue']['length'] = \ |
| self.result_event_queue.qsize() |
| |
| pipelines = [] |
| data['pipelines'] = pipelines |
| keys = self.layout.pipelines.keys() |
| for key in keys: |
| pipeline = self.layout.pipelines[key] |
| pipelines.append(pipeline.formatStatusJSON()) |
| return json.dumps(data) |
| |
| |
| class BasePipelineManager(object): |
| log = logging.getLogger("zuul.BasePipelineManager") |
| |
| def __init__(self, sched, pipeline): |
| self.sched = sched |
| self.pipeline = pipeline |
| self.building_jobs = {} |
| self.event_filters = [] |
| if self.sched.config and self.sched.config.has_option( |
| 'zuul', 'report_times'): |
| self.report_times = self.sched.config.getboolean( |
| 'zuul', 'report_times') |
| else: |
| self.report_times = True |
| |
| def __str__(self): |
| return "<%s %s>" % (self.__class__.__name__, self.pipeline.name) |
| |
| def _postConfig(self, layout): |
| self.log.info("Configured Pipeline Manager %s" % self.pipeline.name) |
| self.log.info(" Events:") |
| for e in self.event_filters: |
| self.log.info(" %s" % e) |
| self.log.info(" Projects:") |
| |
| def log_jobs(tree, indent=0): |
| istr = ' ' + ' ' * indent |
| if tree.job: |
| efilters = '' |
| for b in tree.job._branches: |
| efilters += str(b) |
| for f in tree.job._files: |
| efilters += str(f) |
| if efilters: |
| efilters = ' ' + efilters |
| hold = '' |
| if tree.job.hold_following_changes: |
| hold = ' [hold]' |
| voting = '' |
| if not tree.job.voting: |
| voting = ' [nonvoting]' |
| self.log.info("%s%s%s%s%s" % (istr, repr(tree.job), |
| efilters, hold, voting)) |
| for x in tree.job_trees: |
| log_jobs(x, indent + 2) |
| |
| for p in layout.projects.values(): |
| tree = self.pipeline.getJobTree(p) |
| if tree: |
| self.log.info(" %s" % p) |
| log_jobs(tree) |
| self.log.info(" On start:") |
| self.log.info(" %s" % self.pipeline.start_actions) |
| self.log.info(" On success:") |
| self.log.info(" %s" % self.pipeline.success_actions) |
| self.log.info(" On failure:") |
| self.log.info(" %s" % self.pipeline.failure_actions) |
| |
| def getSubmitAllowNeeds(self): |
| # Get a list of code review labels that are allowed to be |
| # "needed" in the submit records for a change, with respect |
| # to this queue. In other words, the list of review labels |
| # this queue itself is likely to set before submitting. |
| allow_needs = set() |
| for action_reporter in self.pipeline.success_actions: |
| allow_needs.update(action_reporter.getSubmitAllowNeeds()) |
| return allow_needs |
| |
| def eventMatches(self, event): |
| for ef in self.event_filters: |
| if ef.matches(event): |
| return True |
| return False |
| |
| def isChangeAlreadyInQueue(self, change): |
| for c in self.pipeline.getChangesInQueue(): |
| if change.equals(c): |
| return True |
| return False |
| |
| def reportStart(self, change): |
| try: |
| self.log.info("Reporting start, action %s change %s" % |
| (self.pipeline.start_actions, change)) |
| msg = "Starting %s jobs." % self.pipeline.name |
| if self.sched.config.has_option('zuul', 'status_url'): |
| msg += "\n" + self.sched.config.get('zuul', 'status_url') |
| ret = self.sendReport(self.pipeline.start_actions, |
| change, msg) |
| if ret: |
| self.log.error("Reporting change start %s received: %s" % |
| (change, ret)) |
| except: |
| self.log.exception("Exception while reporting start:") |
| |
| def sendReport(self, action_reporters, change, message): |
| """Sends the built message off to configured reporters. |
| |
| Takes the action_reporters, change, message and extra options and |
| sends them to the pluggable reporters. |
| """ |
| report_errors = [] |
| if len(action_reporters) > 0: |
| if not change.number: |
| self.log.debug("Not reporting change %s: No number present." |
| % change) |
| return |
| for action_reporter in action_reporters: |
| ret = action_reporter.report(change, message) |
| if ret: |
| report_errors.append(ret) |
| if len(report_errors) == 0: |
| return |
| return report_errors |
| |
| def isChangeReadyToBeEnqueued(self, change): |
| return True |
| |
| def enqueueChangesAhead(self, change): |
| return True |
| |
| def enqueueChangesBehind(self, change): |
| return True |
| |
| def checkForChangesNeededBy(self, change): |
| return True |
| |
| def getFailingDependentItem(self, item): |
| return None |
| |
| def getDependentItems(self, item): |
| orig_item = item |
| items = [] |
| while item.item_ahead: |
| items.append(item.item_ahead) |
| item = item.item_ahead |
| self.log.info("Change %s depends on changes %s" % |
| (orig_item.change, |
| [x.change for x in items])) |
| return items |
| |
| def getItemForChange(self, change): |
| for item in self.pipeline.getAllItems(): |
| if item.change.equals(change): |
| return item |
| return None |
| |
| def findOldVersionOfChangeAlreadyInQueue(self, change): |
| for c in self.pipeline.getChangesInQueue(): |
| if change.isUpdateOf(c): |
| return c |
| return None |
| |
| def removeOldVersionsOfChange(self, change): |
| if not self.pipeline.dequeue_on_new_patchset: |
| return |
| old_change = self.findOldVersionOfChangeAlreadyInQueue(change) |
| if old_change: |
| self.log.debug("Change %s is a new version of %s, removing %s" % |
| (change, old_change, old_change)) |
| self.removeChange(old_change) |
| |
| def reEnqueueItem(self, item): |
| change_queue = self.pipeline.getQueue(item.change.project) |
| if change_queue: |
| self.log.debug("Re-enqueing change %s in queue %s" % |
| (item.change, change_queue)) |
| change_queue.enqueueItem(item) |
| self.reportStats(item) |
| return True |
| else: |
| self.log.error("Unable to find change queue for project %s" % |
| item.change.project) |
| return False |
| |
| def addChange(self, change): |
| self.log.debug("Considering adding change %s" % change) |
| if self.isChangeAlreadyInQueue(change): |
| self.log.debug("Change %s is already in queue, ignoring" % change) |
| return True |
| |
| if not self.isChangeReadyToBeEnqueued(change): |
| self.log.debug("Change %s is not ready to be enqueued, ignoring" % |
| change) |
| return False |
| |
| if not self.enqueueChangesAhead(change): |
| self.log.debug("Failed to enqueue changes ahead of %s" % change) |
| return False |
| |
| if self.isChangeAlreadyInQueue(change): |
| self.log.debug("Change %s is already in queue, ignoring" % change) |
| return True |
| |
| change_queue = self.pipeline.getQueue(change.project) |
| if change_queue: |
| self.log.debug("Adding change %s to queue %s" % |
| (change, change_queue)) |
| if len(self.pipeline.start_actions) > 0: |
| self.reportStart(change) |
| item = change_queue.enqueueChange(change) |
| self.reportStats(item) |
| self.enqueueChangesBehind(change) |
| else: |
| self.log.error("Unable to find change queue for project %s" % |
| change.project) |
| return False |
| |
| def dequeueItem(self, item): |
| self.log.debug("Removing change %s from queue" % item.change) |
| change_queue = self.pipeline.getQueue(item.change.project) |
| change_queue.dequeueItem(item) |
| self.sched._maintain_trigger_cache = True |
| |
| def removeChange(self, change): |
| # Remove a change from the queue, probably because it has been |
| # superceded by another change. |
| for item in self.pipeline.getAllItems(): |
| if item.change == change: |
| self.log.debug("Canceling builds behind change: %s " |
| "because it is being removed." % item.change) |
| self.cancelJobs(item) |
| self.dequeueItem(item) |
| self.reportStats(item) |
| |
| def prepareRef(self, item): |
| # Returns False on success. |
| # Returns True if we were unable to prepare the ref. |
| ref = item.current_build_set.ref |
| if hasattr(item.change, 'refspec') and not ref: |
| self.log.debug("Preparing ref for: %s" % item.change) |
| item.current_build_set.setConfiguration() |
| ref = item.current_build_set.ref |
| dependent_items = self.getDependentItems(item) |
| dependent_items.reverse() |
| all_items = dependent_items + [item] |
| commit = self.sched.merger.mergeChanges(all_items, ref) |
| item.current_build_set.commit = commit |
| if not commit: |
| self.log.info("Unable to merge change %s" % item.change) |
| msg = ("This change was unable to be automatically merged " |
| "with the current state of the repository. Please " |
| "rebase your change and upload a new patchset.") |
| self.pipeline.setUnableToMerge(item, msg) |
| return True |
| return False |
| |
| def _launchJobs(self, item, jobs): |
| self.log.debug("Launching jobs for change %s" % item.change) |
| dependent_items = self.getDependentItems(item) |
| for job in jobs: |
| self.log.debug("Found job %s for change %s" % (job, item.change)) |
| try: |
| build = self.sched.launcher.launch(job, item, |
| self.pipeline, |
| dependent_items) |
| self.building_jobs[build] = item |
| self.log.debug("Adding build %s of job %s to item %s" % |
| (build, job, item)) |
| item.addBuild(build) |
| except: |
| self.log.exception("Exception while launching job %s " |
| "for change %s:" % (job, item.change)) |
| |
| def launchJobs(self, item): |
| jobs = self.pipeline.findJobsToRun(item) |
| if jobs: |
| self._launchJobs(item, jobs) |
| |
| def cancelJobs(self, item, prime=True): |
| self.log.debug("Cancel jobs for change %s" % item.change) |
| canceled = False |
| to_remove = [] |
| if prime and item.current_build_set.builds: |
| item.resetAllBuilds() |
| for build, build_item in self.building_jobs.items(): |
| if build_item == item: |
| self.log.debug("Found build %s for change %s to cancel" % |
| (build, item.change)) |
| try: |
| self.sched.launcher.cancel(build) |
| except: |
| self.log.exception("Exception while canceling build %s " |
| "for change %s" % (build, item.change)) |
| to_remove.append(build) |
| canceled = True |
| for build in to_remove: |
| self.log.debug("Removing build %s from running builds" % build) |
| build.result = 'CANCELED' |
| del self.building_jobs[build] |
| for item_behind in item.items_behind: |
| self.log.debug("Canceling jobs for change %s, behind change %s" % |
| (item_behind.change, item.change)) |
| if self.cancelJobs(item_behind, prime=prime): |
| canceled = True |
| return canceled |
| |
| def _processOneItem(self, item, nnfi): |
| changed = False |
| item_ahead = item.item_ahead |
| change_queue = self.pipeline.getQueue(item.change.project) |
| failing_reasons = [] # Reasons this item is failing |
| |
| if self.checkForChangesNeededBy(item.change) is not True: |
| # It's not okay to enqueue this change, we should remove it. |
| self.log.info("Dequeuing change %s because " |
| "it can no longer merge" % item.change) |
| self.cancelJobs(item) |
| self.dequeueItem(item) |
| self.pipeline.setDequeuedNeedingChange(item) |
| try: |
| self.reportItem(item) |
| except MergeFailure: |
| pass |
| return (True, nnfi) |
| dep_item = self.getFailingDependentItem(item) |
| if dep_item: |
| failing_reasons.append('a needed change is failing') |
| self.cancelJobs(item, prime=False) |
| else: |
| if (item_ahead and item_ahead != nnfi and |
| not item_ahead.change.is_merged): |
| # Our current base is different than what we expected, |
| # and it's not because our current base merged. Something |
| # ahead must have failed. |
| self.log.info("Resetting builds for change %s because the " |
| "item ahead, %s, is not the nearest non-failing " |
| "item, %s" % (item.change, item_ahead, nnfi)) |
| change_queue.moveItem(item, nnfi) |
| changed = True |
| self.cancelJobs(item) |
| self.prepareRef(item) |
| if item.current_build_set.unable_to_merge: |
| failing_reasons.append("merge conflict") |
| if self.launchJobs(item): |
| changed = True |
| if self.pipeline.didAnyJobFail(item): |
| failing_reasons.append("at least one job failed") |
| if (not item_ahead) and self.pipeline.areAllJobsComplete(item): |
| try: |
| self.reportItem(item) |
| except MergeFailure: |
| failing_reasons.append("did not merge") |
| for item_behind in item.items_behind: |
| self.log.info("Resetting builds for change %s because the " |
| "item ahead, %s, failed to merge" % |
| (item_behind.change, item)) |
| self.cancelJobs(item_behind) |
| self.dequeueItem(item) |
| changed = True |
| elif not failing_reasons: |
| nnfi = item |
| item.current_build_set.failing_reasons = failing_reasons |
| if failing_reasons: |
| self.log.debug("%s is a failing item because %s" % |
| (item, failing_reasons)) |
| return (changed, nnfi) |
| |
| def processQueue(self): |
| # Do whatever needs to be done for each change in the queue |
| self.log.debug("Starting queue processor: %s" % self.pipeline.name) |
| changed = False |
| for queue in self.pipeline.queues: |
| queue_changed = False |
| nnfi = None # Nearest non-failing item |
| for item in queue.queue[:]: |
| item_changed, nnfi = self._processOneItem(item, nnfi) |
| if item_changed: |
| queue_changed = True |
| self.reportStats(item) |
| if queue_changed: |
| changed = True |
| status = '' |
| for item in queue.queue: |
| status += self.pipeline.formatStatus(item) |
| if status: |
| self.log.debug("Queue %s status is now:\n %s" % |
| (queue.name, status)) |
| self.log.debug("Finished queue processor: %s (changed: %s)" % |
| (self.pipeline.name, changed)) |
| return changed |
| |
| def updateBuildDescriptions(self, build_set): |
| for build in build_set.getBuilds(): |
| desc = self.formatDescription(build) |
| self.sched.launcher.setBuildDescription(build, desc) |
| |
| if build_set.previous_build_set: |
| for build in build_set.previous_build_set.getBuilds(): |
| desc = self.formatDescription(build) |
| self.sched.launcher.setBuildDescription(build, desc) |
| |
| def onBuildStarted(self, build): |
| if build not in self.building_jobs: |
| # Or triggered externally, or triggered before zuul started, |
| # or restarted |
| return False |
| |
| self.log.debug("Build %s started" % build) |
| self.updateBuildDescriptions(build.build_set) |
| while self.processQueue(): |
| pass |
| return True |
| |
| def onBuildCompleted(self, build): |
| if build not in self.building_jobs: |
| # Or triggered externally, or triggered before zuul started, |
| # or restarted |
| return False |
| |
| self.log.debug("Build %s completed" % build) |
| change = self.building_jobs[build] |
| self.log.debug("Found change %s which triggered completed build %s" % |
| (change, build)) |
| |
| del self.building_jobs[build] |
| |
| self.pipeline.setResult(change, build) |
| self.log.debug("Change %s status is now:\n %s" % |
| (change, self.pipeline.formatStatus(change))) |
| self.updateBuildDescriptions(build.build_set) |
| while self.processQueue(): |
| pass |
| return True |
| |
| def reportItem(self, item): |
| if item.change.is_reportable and item.reported: |
| raise Exception("Already reported change %s" % item.change) |
| ret = self._reportItem(item) |
| if self.changes_merge: |
| succeeded = self.pipeline.didAllJobsSucceed(item) |
| merged = (not ret) |
| if merged: |
| merged = self.pipeline.trigger.isMerged(item.change, |
| item.change.branch) |
| self.log.info("Reported change %s status: all-succeeded: %s, " |
| "merged: %s" % (item.change, succeeded, merged)) |
| if not (succeeded and merged): |
| self.log.debug("Reported change %s failed tests or failed " |
| "to merge" % (item.change)) |
| raise MergeFailure("Change %s failed to merge" % item.change) |
| |
| def _reportItem(self, item): |
| if not item.change.is_reportable: |
| return False |
| if item.change.is_reportable and item.reported: |
| return 0 |
| self.log.debug("Reporting change %s" % item.change) |
| ret = True # Means error as returned by trigger.report |
| if self.pipeline.didAllJobsSucceed(item): |
| self.log.debug("success %s %s" % (self.pipeline.success_actions, |
| self.pipeline.failure_actions)) |
| actions = self.pipeline.success_actions |
| item.setReportedResult('SUCCESS') |
| else: |
| actions = self.pipeline.failure_actions |
| item.setReportedResult('FAILURE') |
| report = self.formatReport(item) |
| item.reported = True |
| try: |
| self.log.info("Reporting change %s, actions: %s" % |
| (item.change, actions)) |
| ret = self.sendReport(actions, item.change, report) |
| if ret: |
| self.log.error("Reporting change %s received: %s" % |
| (item.change, ret)) |
| except: |
| self.log.exception("Exception while reporting:") |
| item.setReportedResult('ERROR') |
| self.updateBuildDescriptions(item.current_build_set) |
| return ret |
| |
| def formatReport(self, item): |
| ret = '' |
| if self.pipeline.didAllJobsSucceed(item): |
| ret += self.pipeline.success_message + '\n\n' |
| else: |
| ret += self.pipeline.failure_message + '\n\n' |
| |
| if item.dequeued_needing_change: |
| ret += "This change depends on a change that failed to merge." |
| elif item.current_build_set.unable_to_merge_message: |
| ret += item.current_build_set.unable_to_merge_message |
| else: |
| if self.sched.config.has_option('zuul', 'url_pattern'): |
| url_pattern = self.sched.config.get('zuul', 'url_pattern') |
| else: |
| url_pattern = None |
| for job in self.pipeline.getJobs(item.change): |
| build = item.current_build_set.getBuild(job.name) |
| result = build.result |
| pattern = url_pattern |
| if result == 'SUCCESS': |
| if job.success_message: |
| result = job.success_message |
| if job.success_pattern: |
| pattern = job.success_pattern |
| elif result == 'FAILURE': |
| if job.failure_message: |
| result = job.failure_message |
| if job.failure_pattern: |
| pattern = job.failure_pattern |
| if pattern: |
| url = pattern.format(change=item.change, |
| pipeline=self.pipeline, |
| job=job, |
| build=build) |
| else: |
| url = build.url or job.name |
| if not job.voting: |
| voting = ' (non-voting)' |
| else: |
| voting = '' |
| if self.report_times and build.end_time and build.start_time: |
| dt = int(build.end_time - build.start_time) |
| m, s = divmod(dt, 60) |
| h, m = divmod(m, 60) |
| if h: |
| elapsed = ' in %dh %02dm %02ds' % (h, m, s) |
| elif m: |
| elapsed = ' in %dm %02ds' % (m, s) |
| else: |
| elapsed = ' in %ds' % (s) |
| else: |
| elapsed = '' |
| name = '' |
| if self.sched.config.has_option('zuul', 'job_name_in_report'): |
| if self.sched.config.getboolean('zuul', |
| 'job_name_in_report'): |
| name = job.name + ' ' |
| ret += '- %s%s : %s%s%s\n' % (name, url, result, elapsed, |
| voting) |
| return ret |
| |
| def formatDescription(self, build): |
| concurrent_changes = '' |
| concurrent_builds = '' |
| other_builds = '' |
| |
| for change in build.build_set.other_changes: |
| concurrent_changes += '<li><a href="{change.url}">\ |
| {change.number},{change.patchset}</a></li>'.format( |
| change=change) |
| |
| change = build.build_set.item.change |
| |
| for build in build.build_set.getBuilds(): |
| if build.url: |
| concurrent_builds += """\ |
| <li> |
| <a href="{build.url}"> |
| {build.job.name} #{build.number}</a>: {build.result} |
| </li> |
| """.format(build=build) |
| else: |
| concurrent_builds += """\ |
| <li> |
| {build.job.name}: {build.result} |
| </li>""".format(build=build) |
| |
| if build.build_set.previous_build_set: |
| other_build = build.build_set.previous_build_set.getBuild( |
| build.job.name) |
| if other_build: |
| other_builds += """\ |
| <li> |
| Preceded by: <a href="{build.url}"> |
| {build.job.name} #{build.number}</a> |
| </li> |
| """.format(build=other_build) |
| |
| if build.build_set.next_build_set: |
| other_build = build.build_set.next_build_set.getBuild( |
| build.job.name) |
| if other_build: |
| other_builds += """\ |
| <li> |
| Succeeded by: <a href="{build.url}"> |
| {build.job.name} #{build.number}</a> |
| </li> |
| """.format(build=other_build) |
| |
| result = build.build_set.result |
| |
| if hasattr(change, 'number'): |
| ret = """\ |
| <p> |
| Triggered by change: |
| <a href="{change.url}">{change.number},{change.patchset}</a><br/> |
| Branch: <b>{change.branch}</b><br/> |
| Pipeline: <b>{self.pipeline.name}</b> |
| </p>""" |
| elif hasattr(change, 'ref'): |
| ret = """\ |
| <p> |
| Triggered by reference: |
| {change.ref}</a><br/> |
| Old revision: <b>{change.oldrev}</b><br/> |
| New revision: <b>{change.newrev}</b><br/> |
| Pipeline: <b>{self.pipeline.name}</b> |
| </p>""" |
| else: |
| ret = "" |
| |
| if concurrent_changes: |
| ret += """\ |
| <p> |
| Other changes tested concurrently with this change: |
| <ul>{concurrent_changes}</ul> |
| </p> |
| """ |
| if concurrent_builds: |
| ret += """\ |
| <p> |
| All builds for this change set: |
| <ul>{concurrent_builds}</ul> |
| </p> |
| """ |
| |
| if other_builds: |
| ret += """\ |
| <p> |
| Other build sets for this change: |
| <ul>{other_builds}</ul> |
| </p> |
| """ |
| if result: |
| ret += """\ |
| <p> |
| Reported result: <b>{result}</b> |
| </p> |
| """ |
| |
| ret = ret.format(**locals()) |
| return ret |
| |
| def reportStats(self, item): |
| if not statsd: |
| return |
| try: |
| # Update the gauge on enqueue and dequeue, but timers only |
| # when dequeing. |
| if item.dequeue_time: |
| dt = int((item.dequeue_time - item.enqueue_time) * 1000) |
| else: |
| dt = None |
| items = len(self.pipeline.getAllItems()) |
| |
| # stats.timers.zuul.pipeline.NAME.resident_time |
| # stats_counts.zuul.pipeline.NAME.total_changes |
| # stats.gauges.zuul.pipeline.NAME.current_changes |
| key = 'zuul.pipeline.%s' % self.pipeline.name |
| statsd.gauge(key + '.current_changes', items) |
| if dt: |
| statsd.timing(key + '.resident_time', dt) |
| statsd.incr(key + '.total_changes') |
| |
| # stats.timers.zuul.pipeline.NAME.ORG.PROJECT.resident_time |
| # stats_counts.zuul.pipeline.NAME.ORG.PROJECT.total_changes |
| project_name = item.change.project.name.replace('/', '.') |
| key += '.%s' % project_name |
| if dt: |
| statsd.timing(key + '.resident_time', dt) |
| statsd.incr(key + '.total_changes') |
| except: |
| self.log.exception("Exception reporting pipeline stats") |
| |
| |
| class IndependentPipelineManager(BasePipelineManager): |
| log = logging.getLogger("zuul.IndependentPipelineManager") |
| changes_merge = False |
| |
| def _postConfig(self, layout): |
| super(IndependentPipelineManager, self)._postConfig(layout) |
| |
| change_queue = ChangeQueue(self.pipeline, dependent=False) |
| for project in self.pipeline.getProjects(): |
| change_queue.addProject(project) |
| |
| self.pipeline.addQueue(change_queue) |
| |
| |
| class DependentPipelineManager(BasePipelineManager): |
| log = logging.getLogger("zuul.DependentPipelineManager") |
| changes_merge = True |
| |
| def __init__(self, *args, **kwargs): |
| super(DependentPipelineManager, self).__init__(*args, **kwargs) |
| |
| def _postConfig(self, layout): |
| super(DependentPipelineManager, self)._postConfig(layout) |
| self.buildChangeQueues() |
| |
| def buildChangeQueues(self): |
| self.log.debug("Building shared change queues") |
| change_queues = [] |
| |
| for project in self.pipeline.getProjects(): |
| change_queue = ChangeQueue(self.pipeline) |
| change_queue.addProject(project) |
| change_queues.append(change_queue) |
| self.log.debug("Created queue: %s" % change_queue) |
| |
| self.log.debug("Combining shared queues") |
| new_change_queues = [] |
| for a in change_queues: |
| merged_a = False |
| for b in new_change_queues: |
| if not a.getJobs().isdisjoint(b.getJobs()): |
| self.log.debug("Merging queue %s into %s" % (a, b)) |
| b.mergeChangeQueue(a) |
| merged_a = True |
| break # this breaks out of 'for b' and continues 'for a' |
| if not merged_a: |
| self.log.debug("Keeping queue %s" % (a)) |
| new_change_queues.append(a) |
| |
| self.log.info(" Shared change queues:") |
| for queue in new_change_queues: |
| self.pipeline.addQueue(queue) |
| self.log.info(" %s" % queue) |
| |
| def isChangeReadyToBeEnqueued(self, change): |
| if not self.pipeline.trigger.canMerge(change, |
| self.getSubmitAllowNeeds()): |
| self.log.debug("Change %s can not merge, ignoring" % change) |
| return False |
| return True |
| |
| def enqueueChangesBehind(self, change): |
| to_enqueue = [] |
| self.log.debug("Checking for changes needing %s:" % change) |
| if not hasattr(change, 'needed_by_changes'): |
| self.log.debug(" Changeish does not support dependencies") |
| return |
| for needs in change.needed_by_changes: |
| if self.pipeline.trigger.canMerge(needs, |
| self.getSubmitAllowNeeds()): |
| self.log.debug(" Change %s needs %s and is ready to merge" % |
| (needs, change)) |
| to_enqueue.append(needs) |
| if not to_enqueue: |
| self.log.debug(" No changes need %s" % change) |
| |
| for other_change in to_enqueue: |
| self.addChange(other_change) |
| |
| def enqueueChangesAhead(self, change): |
| ret = self.checkForChangesNeededBy(change) |
| if ret in [True, False]: |
| return ret |
| self.log.debug(" Change %s must be merged ahead of %s" % |
| (ret, change)) |
| return self.addChange(ret) |
| |
| def checkForChangesNeededBy(self, change): |
| self.log.debug("Checking for changes needed by %s:" % change) |
| # Return true if okay to proceed enqueing this change, |
| # false if the change should not be enqueued. |
| if not hasattr(change, 'needs_change'): |
| self.log.debug(" Changeish does not support dependencies") |
| return True |
| if not change.needs_change: |
| self.log.debug(" No changes needed") |
| return True |
| if change.needs_change.is_merged: |
| self.log.debug(" Needed change is merged") |
| return True |
| if not change.needs_change.is_current_patchset: |
| self.log.debug(" Needed change is not the current patchset") |
| return False |
| if self.isChangeAlreadyInQueue(change.needs_change): |
| self.log.debug(" Needed change is already ahead in the queue") |
| return True |
| if self.pipeline.trigger.canMerge(change.needs_change, |
| self.getSubmitAllowNeeds()): |
| self.log.debug(" Change %s is needed" % |
| change.needs_change) |
| return change.needs_change |
| # The needed change can't be merged. |
| self.log.debug(" Change %s is needed but can not be merged" % |
| change.needs_change) |
| return False |
| |
| def getFailingDependentItem(self, item): |
| if not hasattr(item.change, 'needs_change'): |
| return None |
| if not item.change.needs_change: |
| return None |
| needs_item = self.getItemForChange(item.change.needs_change) |
| if not needs_item: |
| return None |
| if needs_item.current_build_set.failing_reasons: |
| return needs_item |
| return None |