| # Copyright 2012-2015 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 socket |
| import sys |
| import threading |
| import time |
| |
| from zuul import configloader |
| from zuul import model |
| from zuul import exceptions |
| from zuul import version as zuul_version |
| from zuul.lib.config import get_default |
| |
| |
| class ManagementEvent(object): |
| """An event that should be processed within the main queue run loop""" |
| def __init__(self): |
| self._wait_event = threading.Event() |
| self._exc_info = None |
| |
| def exception(self, exc_info): |
| self._exc_info = exc_info |
| self._wait_event.set() |
| |
| def done(self): |
| self._wait_event.set() |
| |
| def wait(self, timeout=None): |
| self._wait_event.wait(timeout) |
| if self._exc_info: |
| # sys.exc_info returns (type, value, traceback) |
| type_, exception_instance, traceback = self._exc_info |
| raise exception_instance.with_traceback(traceback) |
| return self._wait_event.is_set() |
| |
| |
| class ReconfigureEvent(ManagementEvent): |
| """Reconfigure the scheduler. The layout will be (re-)loaded from |
| the path specified in the configuration. |
| |
| :arg ConfigParser config: the new configuration |
| """ |
| def __init__(self, config): |
| super(ReconfigureEvent, self).__init__() |
| self.config = config |
| |
| |
| class TenantReconfigureEvent(ManagementEvent): |
| """Reconfigure the given tenant. The layout will be (re-)loaded from |
| the path specified in the configuration. |
| |
| :arg Tenant tenant: the tenant to reconfigure |
| """ |
| def __init__(self, tenant): |
| super(TenantReconfigureEvent, self).__init__() |
| self.tenant = tenant |
| |
| |
| class PromoteEvent(ManagementEvent): |
| """Promote one or more changes to the head of the queue. |
| |
| :arg str tenant_name: the name of the tenant |
| :arg str pipeline_name: the name of the pipeline |
| :arg list change_ids: a list of strings of change ids in the form |
| 1234,1 |
| """ |
| |
| def __init__(self, tenant_name, pipeline_name, change_ids): |
| super(PromoteEvent, self).__init__() |
| self.tenant_name = tenant_name |
| self.pipeline_name = pipeline_name |
| self.change_ids = change_ids |
| |
| |
| class EnqueueEvent(ManagementEvent): |
| """Enqueue a change into a pipeline |
| |
| :arg TriggerEvent trigger_event: a TriggerEvent describing the |
| trigger, pipeline, and change to enqueue |
| """ |
| |
| def __init__(self, trigger_event): |
| super(EnqueueEvent, self).__init__() |
| self.trigger_event = trigger_event |
| |
| |
| class ResultEvent(object): |
| """An event that needs to modify the pipeline state due to a |
| result from an external system.""" |
| |
| pass |
| |
| |
| class BuildStartedEvent(ResultEvent): |
| """A build has started. |
| |
| :arg Build build: The build which has started. |
| """ |
| |
| def __init__(self, build): |
| self.build = build |
| |
| |
| class BuildCompletedEvent(ResultEvent): |
| """A build has completed |
| |
| :arg Build build: The build which has completed. |
| """ |
| |
| def __init__(self, build): |
| self.build = build |
| |
| |
| class MergeCompletedEvent(ResultEvent): |
| """A remote merge operation has completed |
| |
| :arg BuildSet build_set: The build_set which is ready. |
| :arg bool merged: Whether the merge succeeded (changes with refs). |
| :arg bool updated: Whether the repo was updated (changes without refs). |
| :arg str commit: The SHA of the merged commit (changes with refs). |
| :arg dict repo_state: The starting repo state before the merge. |
| """ |
| |
| def __init__(self, build_set, merged, updated, commit, |
| files, repo_state): |
| self.build_set = build_set |
| self.merged = merged |
| self.updated = updated |
| self.commit = commit |
| self.files = files |
| self.repo_state = repo_state |
| |
| |
| class NodesProvisionedEvent(ResultEvent): |
| """Nodes have been provisioned for a build_set |
| |
| :arg BuildSet build_set: The build_set which has nodes. |
| :arg list of Node objects nodes: The provisioned nodes |
| """ |
| |
| def __init__(self, request): |
| self.request = request |
| |
| |
| def toList(item): |
| if not item: |
| return [] |
| if isinstance(item, list): |
| return item |
| return [item] |
| |
| |
| class Scheduler(threading.Thread): |
| """The engine of Zuul. |
| |
| The Scheduler is reponsible for recieving events and dispatching |
| them to appropriate components (including pipeline managers, |
| mergers and executors). |
| |
| It runs a single threaded main loop which processes events |
| received one at a time and takes action as appropriate. Other |
| parts of Zuul may run in their own thread, but synchronization is |
| performed within the scheduler to reduce or eliminate the need for |
| locking in most circumstances. |
| |
| The main daemon will have one instance of the Scheduler class |
| running which will persist for the life of the process. The |
| Scheduler instance is supplied to other Zuul components so that |
| they can submit events or otherwise communicate with other |
| components. |
| |
| """ |
| |
| log = logging.getLogger("zuul.Scheduler") |
| |
| def __init__(self, config, testonly=False): |
| threading.Thread.__init__(self) |
| self.daemon = True |
| self.hostname = socket.gethostname() |
| self.wake_event = threading.Event() |
| self.layout_lock = threading.Lock() |
| self.run_handler_lock = threading.Lock() |
| self._pause = False |
| self._exit = False |
| self._stopped = False |
| self.executor = None |
| self.merger = None |
| self.connections = None |
| self.statsd = extras.try_import('statsd.statsd') |
| # TODO(jeblair): fix this |
| # Despite triggers being part of the pipeline, there is one trigger set |
| # per scheduler. The pipeline handles the trigger filters but since |
| # the events are handled by the scheduler itself it needs to handle |
| # the loading of the triggers. |
| # self.triggers['connection_name'] = triggerObject |
| self.triggers = dict() |
| self.config = config |
| |
| self.trigger_event_queue = queue.Queue() |
| self.result_event_queue = queue.Queue() |
| self.management_event_queue = queue.Queue() |
| self.abide = model.Abide() |
| |
| if not testonly: |
| time_dir = self._get_time_database_dir() |
| self.time_database = model.TimeDataBase(time_dir) |
| |
| self.zuul_version = zuul_version.version_info.release_string() |
| self.last_reconfigured = None |
| self.tenant_last_reconfigured = {} |
| self.autohold_requests = {} |
| |
| def stop(self): |
| self._stopped = True |
| self.stopConnections() |
| self.wake_event.set() |
| |
| def registerConnections(self, connections, webapp, load=True): |
| # load: whether or not to trigger the onLoad for the connection. This |
| # is useful for not doing a full load during layout validation. |
| self.connections = connections |
| self.connections.registerWebapp(webapp) |
| self.connections.registerScheduler(self, load) |
| |
| def stopConnections(self): |
| self.connections.stop() |
| |
| def setExecutor(self, executor): |
| self.executor = executor |
| |
| def setMerger(self, merger): |
| self.merger = merger |
| |
| def setNodepool(self, nodepool): |
| self.nodepool = nodepool |
| |
| def setZooKeeper(self, zk): |
| self.zk = zk |
| |
| def addEvent(self, event): |
| self.log.debug("Adding trigger event: %s" % event) |
| 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() |
| event = BuildStartedEvent(build) |
| self.result_event_queue.put(event) |
| self.wake_event.set() |
| self.log.debug("Done adding start event for build: %s" % build) |
| |
| def onBuildCompleted(self, build, result, result_data): |
| self.log.debug("Adding complete event for build: %s result: %s" % ( |
| build, result)) |
| build.end_time = time.time() |
| build.result_data = result_data |
| # Note, as soon as the result is set, other threads may act |
| # upon this, even though the event hasn't been fully |
| # processed. Ensure that any other data from the event (eg, |
| # timing) is recorded before setting the result. |
| build.result = result |
| try: |
| if self.statsd and build.pipeline: |
| jobname = build.job.name.replace('.', '_') |
| key = 'zuul.pipeline.%s.all_jobs' % build.pipeline.name |
| self.statsd.incr(key) |
| for label in build.node_labels: |
| # Jenkins includes the node name in its list of labels, so |
| # we filter it out here, since that is not statistically |
| # interesting. |
| if label == build.node_name: |
| continue |
| dt = int((build.start_time - build.execute_time) * 1000) |
| key = 'zuul.pipeline.%s.label.%s.wait_time' % ( |
| build.pipeline.name, label) |
| self.statsd.timing(key, dt) |
| 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) |
| self.statsd.timing(key, dt) |
| self.statsd.incr(key) |
| |
| key = 'zuul.pipeline.%s.job.%s.wait_time' % ( |
| build.pipeline.name, jobname) |
| dt = int((build.start_time - build.execute_time) * 1000) |
| self.statsd.timing(key, dt) |
| except: |
| self.log.exception("Exception reporting runtime stats") |
| event = BuildCompletedEvent(build) |
| self.result_event_queue.put(event) |
| self.wake_event.set() |
| self.log.debug("Done adding complete event for build: %s" % build) |
| |
| def onMergeCompleted(self, build_set, merged, updated, |
| commit, files, repo_state): |
| self.log.debug("Adding merge complete event for build set: %s" % |
| build_set) |
| event = MergeCompletedEvent(build_set, merged, |
| updated, commit, files, repo_state) |
| self.result_event_queue.put(event) |
| self.wake_event.set() |
| |
| def onNodesProvisioned(self, req): |
| self.log.debug("Adding nodes provisioned event for build set: %s" % |
| req.build_set) |
| event = NodesProvisionedEvent(req) |
| self.result_event_queue.put(event) |
| self.wake_event.set() |
| |
| def reconfigureTenant(self, tenant): |
| self.log.debug("Prepare to reconfigure") |
| event = TenantReconfigureEvent(tenant) |
| self.management_event_queue.put(event) |
| self.wake_event.set() |
| |
| def reconfigure(self, config): |
| self.log.debug("Prepare to reconfigure") |
| event = ReconfigureEvent(config) |
| self.management_event_queue.put(event) |
| self.wake_event.set() |
| self.log.debug("Waiting for reconfiguration") |
| event.wait() |
| self.log.debug("Reconfiguration complete") |
| self.last_reconfigured = int(time.time()) |
| # TODOv3(jeblair): reconfigure time should be per-tenant |
| |
| def autohold(self, tenant_name, project_name, job_name, reason, count): |
| key = (tenant_name, project_name, job_name) |
| if count == 0 and key in self.autohold_requests: |
| self.log.debug("Removing autohold for %s", key) |
| del self.autohold_requests[key] |
| else: |
| self.log.debug("Autohold requested for %s", key) |
| self.autohold_requests[key] = (count, reason) |
| |
| def promote(self, tenant_name, pipeline_name, change_ids): |
| event = PromoteEvent(tenant_name, pipeline_name, change_ids) |
| self.management_event_queue.put(event) |
| self.wake_event.set() |
| self.log.debug("Waiting for promotion") |
| event.wait() |
| self.log.debug("Promotion complete") |
| |
| def enqueue(self, trigger_event): |
| event = EnqueueEvent(trigger_event) |
| self.management_event_queue.put(event) |
| self.wake_event.set() |
| self.log.debug("Waiting for enqueue") |
| event.wait() |
| self.log.debug("Enqueue 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): |
| state_dir = get_default(self.config, 'scheduler', 'state_dir', |
| '/var/lib/zuul', expand_user=True) |
| return os.path.join(state_dir, 'queue.pickle') |
| |
| def _get_time_database_dir(self): |
| state_dir = get_default(self.config, 'scheduler', 'state_dir', |
| '/var/lib/zuul', expand_user=True) |
| d = os.path.join(state_dir, 'times') |
| if not os.path.exists(d): |
| os.mkdir(d) |
| return d |
| |
| def _get_project_key_dir(self): |
| state_dir = get_default(self.config, 'scheduler', 'state_dir', |
| '/var/lib/zuul', expand_user=True) |
| key_dir = os.path.join(state_dir, 'keys') |
| if not os.path.exists(key_dir): |
| os.mkdir(key_dir, 0o700) |
| st = os.stat(key_dir) |
| mode = st.st_mode & 0o777 |
| if mode != 0o700: |
| raise Exception("Project key directory %s must be mode 0700; " |
| "current mode is %o" % (key_dir, mode)) |
| return key_dir |
| |
| 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, event): |
| # This is called in the scheduler loop after another thread submits |
| # a request |
| self.layout_lock.acquire() |
| self.config = event.config |
| try: |
| self.log.debug("Performing reconfiguration") |
| loader = configloader.ConfigLoader() |
| abide = loader.loadConfig( |
| self.config.get('scheduler', 'tenant_config'), |
| self._get_project_key_dir(), |
| self, self.merger, self.connections) |
| for tenant in abide.tenants.values(): |
| self._reconfigureTenant(tenant) |
| self.abide = abide |
| finally: |
| self.layout_lock.release() |
| |
| def _doTenantReconfigureEvent(self, event): |
| # This is called in the scheduler loop after another thread submits |
| # a request |
| self.layout_lock.acquire() |
| try: |
| self.log.debug("Performing tenant reconfiguration") |
| loader = configloader.ConfigLoader() |
| abide = loader.reloadTenant( |
| self.config.get('scheduler', 'tenant_config'), |
| self._get_project_key_dir(), |
| self, self.merger, self.connections, |
| self.abide, event.tenant) |
| tenant = abide.tenants[event.tenant.name] |
| self._reconfigureTenant(tenant) |
| self.abide = abide |
| finally: |
| self.layout_lock.release() |
| |
| def _reenqueueGetProject(self, tenant, item): |
| project = item.change.project |
| # Attempt to get the same project as the one passed in. If |
| # the project is now found on a different connection, return |
| # the new version of the project. If it is no longer |
| # available (due to a connection being removed), return None. |
| (trusted, new_project) = tenant.getProject(project.canonical_name) |
| if new_project: |
| return new_project |
| # If this is a non-live item we may be looking at a |
| # "foreign" project, ie, one which is not defined in the |
| # config but is constructed ad-hoc to satisfy a |
| # cross-repo-dependency. Find the corresponding live item |
| # and use its source. |
| child = item |
| while child and not child.live: |
| # This assumes that the queue does not branch behind this |
| # item, which is currently true for non-live items; if |
| # that changes, this traversal will need to be more |
| # complex. |
| if child.items_behind: |
| child = child.items_behind[0] |
| else: |
| child = None |
| if child is item: |
| return None |
| if child and child.live: |
| (child_trusted, child_project) = tenant.getProject( |
| child.change.project.canonical_name) |
| if child_project: |
| source = child_project.source |
| new_project = source.getProject(project.name) |
| return new_project |
| return None |
| |
| def _reenqueueTenant(self, old_tenant, tenant): |
| for name, new_pipeline in tenant.layout.pipelines.items(): |
| old_pipeline = old_tenant.layout.pipelines.get(name) |
| if not old_pipeline: |
| 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 = [] |
| builds_to_cancel = [] |
| last_head = None |
| for shared_queue in old_pipeline.queues: |
| for item in shared_queue.queue: |
| if not item.item_ahead: |
| last_head = item |
| item.pipeline = None |
| item.queue = None |
| item.change.project = self._reenqueueGetProject( |
| tenant, item) |
| item.item_ahead = None |
| item.items_behind = [] |
| if (item.change.project and |
| new_pipeline.manager.reEnqueueItem(item, |
| last_head)): |
| for build in item.current_build_set.getBuilds(): |
| new_job = item.getJob(build.job.name) |
| if new_job: |
| build.job = new_job |
| else: |
| item.removeBuild(build) |
| builds_to_cancel.append(build) |
| else: |
| items_to_remove.append(item) |
| for item in items_to_remove: |
| self.log.warning( |
| "Removing item %s during reconfiguration" % (item,)) |
| for build in item.current_build_set.getBuilds(): |
| builds_to_cancel.append(build) |
| for build in builds_to_cancel: |
| self.log.warning( |
| "Canceling build %s during reconfiguration" % (build,)) |
| try: |
| self.executor.cancel(build) |
| except Exception: |
| self.log.exception( |
| "Exception while canceling build %s " |
| "for change %s" % (build, build.build_set.item.change)) |
| finally: |
| tenant.semaphore_handler.release( |
| build.build_set.item, build.job) |
| |
| def _reconfigureTenant(self, tenant): |
| # This is called from _doReconfigureEvent while holding the |
| # layout lock |
| old_tenant = self.abide.tenants.get(tenant.name) |
| |
| if old_tenant: |
| # Copy over semaphore handler so we don't loose the currently |
| # held semaphores. |
| tenant.semaphore_handler = old_tenant.semaphore_handler |
| |
| self._reenqueueTenant(old_tenant, tenant) |
| |
| # TODOv3(jeblair): update for tenants |
| # self.maintainConnectionCache() |
| self.connections.reconfigureDrivers(tenant) |
| |
| # TODOv3(jeblair): remove postconfig calls? |
| for pipeline in tenant.layout.pipelines.values(): |
| for trigger in pipeline.triggers: |
| trigger.postConfig(pipeline) |
| for reporter in pipeline.actions: |
| reporter.postConfig() |
| self.tenant_last_reconfigured[tenant.name] = int(time.time()) |
| if self.statsd: |
| try: |
| for pipeline in tenant.layout.pipelines.values(): |
| items = len(pipeline.getAllItems()) |
| # stats.gauges.zuul.pipeline.NAME.current_changes |
| key = 'zuul.pipeline.%s' % pipeline.name |
| self.statsd.gauge(key + '.current_changes', items) |
| except Exception: |
| self.log.exception("Exception reporting initial " |
| "pipeline stats:") |
| |
| def _doPromoteEvent(self, event): |
| tenant = self.abide.tenants.get(event.tenant_name) |
| pipeline = tenant.layout.pipelines[event.pipeline_name] |
| change_ids = [c.split(',') for c in event.change_ids] |
| items_to_enqueue = [] |
| change_queue = None |
| for shared_queue in pipeline.queues: |
| if change_queue: |
| break |
| for item in shared_queue.queue: |
| if (item.change.number == change_ids[0][0] and |
| item.change.patchset == change_ids[0][1]): |
| change_queue = shared_queue |
| break |
| if not change_queue: |
| raise Exception("Unable to find shared change queue for %s" % |
| event.change_ids[0]) |
| for number, patchset in change_ids: |
| found = False |
| for item in change_queue.queue: |
| if (item.change.number == number and |
| item.change.patchset == patchset): |
| found = True |
| items_to_enqueue.append(item) |
| break |
| if not found: |
| raise Exception("Unable to find %s,%s in queue %s" % |
| (number, patchset, change_queue)) |
| for item in change_queue.queue[:]: |
| if item not in items_to_enqueue: |
| items_to_enqueue.append(item) |
| pipeline.manager.cancelJobs(item) |
| pipeline.manager.dequeueItem(item) |
| for item in items_to_enqueue: |
| pipeline.manager.addChange( |
| item.change, |
| enqueue_time=item.enqueue_time, |
| quiet=True, |
| ignore_requirements=True) |
| |
| def _doEnqueueEvent(self, event): |
| tenant = self.abide.tenants.get(event.tenant_name) |
| (trusted, project) = tenant.getProject(event.project_name) |
| pipeline = tenant.layout.pipelines[event.forced_pipeline] |
| change = project.source.getChange(event, project) |
| self.log.debug("Event %s for change %s was directly assigned " |
| "to pipeline %s" % (event, change, self)) |
| pipeline.manager.addChange(change, ignore_requirements=True) |
| |
| def _areAllBuildsComplete(self): |
| self.log.debug("Checking if all builds are complete") |
| if self.merger.areMergesOutstanding(): |
| self.log.debug("Waiting on merger") |
| return False |
| waiting = False |
| for tenant in self.abide.tenants.values(): |
| for pipeline in tenant.layout.pipelines.values(): |
| for item in pipeline.getAllItems(): |
| for build in item.current_build_set.getBuilds(): |
| if build.result is None: |
| self.log.debug("%s waiting on %s" % |
| (pipeline.manager, build)) |
| waiting = True |
| if not waiting: |
| self.log.debug("All builds are complete") |
| return True |
| return False |
| |
| def run(self): |
| if self.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: |
| self.log.debug("Run handler stopping") |
| return |
| self.log.debug("Run handler awake") |
| self.run_handler_lock.acquire() |
| try: |
| while not self.management_event_queue.empty(): |
| self.process_management_queue() |
| |
| # Give result events priority -- they let us stop builds, |
| # whereas trigger events cause us to execute builds. |
| while not self.result_event_queue.empty(): |
| self.process_result_queue() |
| |
| if not self._pause: |
| while not self.trigger_event_queue.empty(): |
| self.process_event_queue() |
| |
| if self._pause and self._areAllBuildsComplete(): |
| self._doPauseEvent() |
| |
| for tenant in self.abide.tenants.values(): |
| for pipeline in tenant.layout.pipelines.values(): |
| while pipeline.manager.processQueue(): |
| pass |
| |
| except Exception: |
| self.log.exception("Exception in run handler:") |
| # There may still be more events to process |
| self.wake_event.set() |
| finally: |
| self.run_handler_lock.release() |
| |
| def maintainConnectionCache(self): |
| # TODOv3(jeblair): update for tenants |
| relevant = set() |
| for tenant in self.abide.tenants.values(): |
| for pipeline in tenant.layout.pipelines.values(): |
| self.log.debug("Gather relevant cache items for: %s" % |
| pipeline) |
| |
| for item in pipeline.getAllItems(): |
| relevant.add(item.change) |
| relevant.update(item.change.getRelatedChanges()) |
| for connection in self.connections.values(): |
| connection.maintainCache(relevant) |
| self.log.debug( |
| "End maintain connection cache for: %s" % connection) |
| self.log.debug("Connection cache size: %s" % len(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) |
| try: |
| full_project_name = ('/'.join([event.project_hostname, |
| event.project_name])) |
| for tenant in self.abide.tenants.values(): |
| (trusted, project) = tenant.getProject(full_project_name) |
| if project is None: |
| continue |
| try: |
| change = project.source.getChange(event) |
| except exceptions.ChangeNotFound as e: |
| self.log.debug("Unable to get change %s from " |
| "source %s", |
| e.change, project.source) |
| continue |
| if ((event.branch_updated and |
| hasattr(change, 'files') and |
| change.updatesConfig()) or |
| event.branch_created or |
| event.branch_deleted): |
| # The change that just landed updates the config |
| # or a branch was just created or deleted. Clear |
| # out cached data for this project and perform a |
| # reconfiguration. |
| change.project.unparsed_config = None |
| self.reconfigureTenant(tenant) |
| for pipeline in tenant.layout.pipelines.values(): |
| if event.isPatchsetCreated(): |
| pipeline.manager.removeOldVersionsOfChange(change) |
| elif event.isChangeAbandoned(): |
| pipeline.manager.removeAbandonedChange(change) |
| if pipeline.manager.eventMatches(event, change): |
| pipeline.manager.addChange(change) |
| finally: |
| self.trigger_event_queue.task_done() |
| |
| def process_management_queue(self): |
| self.log.debug("Fetching management event") |
| event = self.management_event_queue.get() |
| self.log.debug("Processing management event %s" % event) |
| try: |
| if isinstance(event, ReconfigureEvent): |
| self._doReconfigureEvent(event) |
| elif isinstance(event, TenantReconfigureEvent): |
| self._doTenantReconfigureEvent(event) |
| elif isinstance(event, PromoteEvent): |
| self._doPromoteEvent(event) |
| elif isinstance(event, EnqueueEvent): |
| self._doEnqueueEvent(event.trigger_event) |
| else: |
| self.log.error("Unable to handle event %s" % event) |
| event.done() |
| except Exception: |
| self.log.exception("Exception in management event:") |
| event.exception(sys.exc_info()) |
| self.management_event_queue.task_done() |
| |
| def process_result_queue(self): |
| self.log.debug("Fetching result event") |
| event = self.result_event_queue.get() |
| self.log.debug("Processing result event %s" % event) |
| try: |
| if isinstance(event, BuildStartedEvent): |
| self._doBuildStartedEvent(event) |
| elif isinstance(event, BuildCompletedEvent): |
| self._doBuildCompletedEvent(event) |
| elif isinstance(event, MergeCompletedEvent): |
| self._doMergeCompletedEvent(event) |
| elif isinstance(event, NodesProvisionedEvent): |
| self._doNodesProvisionedEvent(event) |
| else: |
| self.log.error("Unable to handle event %s" % event) |
| finally: |
| self.result_event_queue.task_done() |
| |
| def _doBuildStartedEvent(self, event): |
| build = event.build |
| if build.build_set is not build.build_set.item.current_build_set: |
| self.log.warning("Build %s is not in the current build set" % |
| (build,)) |
| return |
| pipeline = build.build_set.item.pipeline |
| if not pipeline: |
| self.log.warning("Build %s is not associated with a pipeline" % |
| (build,)) |
| return |
| try: |
| build.estimated_time = float(self.time_database.getEstimatedTime( |
| build.job.name)) |
| except Exception: |
| self.log.exception("Exception estimating build time:") |
| pipeline.manager.onBuildStarted(event.build) |
| |
| def _doBuildCompletedEvent(self, event): |
| build = event.build |
| |
| # Regardless of any other conditions which might cause us not |
| # to pass this on to the pipeline manager, make sure we return |
| # the nodes to nodepool. |
| try: |
| nodeset = build.build_set.getJobNodeSet(build.job.name) |
| autohold_key = (build.pipeline.layout.tenant.name, |
| build.build_set.item.change.project.canonical_name, |
| build.job.name) |
| |
| try: |
| self.nodepool.holdNodeSet(nodeset, autohold_key) |
| except Exception: |
| self.log.exception("Unable to process autohold for %s", |
| autohold_key) |
| |
| self.nodepool.returnNodeSet(nodeset) |
| except Exception: |
| self.log.exception("Unable to return nodeset %s" % (nodeset,)) |
| |
| if build.build_set is not build.build_set.item.current_build_set: |
| self.log.debug("Build %s is not in the current build set" % |
| (build,)) |
| return |
| pipeline = build.build_set.item.pipeline |
| if not pipeline: |
| self.log.warning("Build %s is not associated with a pipeline" % |
| (build,)) |
| return |
| if build.end_time and build.start_time and build.result: |
| duration = build.end_time - build.start_time |
| try: |
| self.time_database.update( |
| build.job.name, duration, build.result) |
| except Exception: |
| self.log.exception("Exception recording build time:") |
| pipeline.manager.onBuildCompleted(event.build) |
| |
| def _doMergeCompletedEvent(self, event): |
| build_set = event.build_set |
| if build_set is not build_set.item.current_build_set: |
| self.log.warning("Build set %s is not current" % (build_set,)) |
| return |
| pipeline = build_set.item.pipeline |
| if not pipeline: |
| self.log.warning("Build set %s is not associated with a pipeline" % |
| (build_set,)) |
| return |
| pipeline.manager.onMergeCompleted(event) |
| |
| def _doNodesProvisionedEvent(self, event): |
| request = event.request |
| build_set = request.build_set |
| |
| self.nodepool.acceptNodes(request) |
| if request.canceled: |
| return |
| |
| if build_set is not build_set.item.current_build_set: |
| self.log.warning("Build set %s is not current" % (build_set,)) |
| if request.fulfilled: |
| self.nodepool.returnNodeSet(request.nodeset) |
| return |
| pipeline = build_set.item.pipeline |
| if not pipeline: |
| self.log.warning("Build set %s is not associated with a pipeline" % |
| (build_set,)) |
| if request.fulfilled: |
| self.nodepool.returnNodeSet(request.nodeset) |
| return |
| pipeline.manager.onNodesProvisioned(event) |
| |
| def formatStatusJSON(self, tenant_name): |
| # TODOv3(jeblair): use tenants |
| data = {} |
| |
| data['zuul_version'] = self.zuul_version |
| websocket_url = get_default(self.config, 'web', 'websocket_url', None) |
| |
| 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() |
| |
| if self.last_reconfigured: |
| data['last_reconfigured'] = self.last_reconfigured * 1000 |
| |
| pipelines = [] |
| data['pipelines'] = pipelines |
| tenant = self.abide.tenants.get(tenant_name) |
| for pipeline in tenant.layout.pipelines.values(): |
| pipelines.append(pipeline.formatStatusJSON(websocket_url)) |
| return json.dumps(data) |