| # Copyright 2014 OpenStack Foundation |
| # |
| # 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 collections |
| import json |
| import logging |
| import os |
| import shutil |
| import signal |
| import socket |
| import subprocess |
| import tempfile |
| import threading |
| import time |
| import traceback |
| from zuul.lib.yamlutil import yaml |
| |
| import gear |
| import git |
| from six.moves import shlex_quote |
| |
| import zuul.merger.merger |
| import zuul.ansible.action |
| import zuul.ansible.callback |
| import zuul.ansible.library |
| import zuul.ansible.lookup |
| from zuul.lib import commandsocket |
| |
| COMMANDS = ['stop', 'pause', 'unpause', 'graceful', 'verbose', |
| 'unverbose'] |
| |
| |
| class Watchdog(object): |
| def __init__(self, timeout, function, args): |
| self.timeout = timeout |
| self.function = function |
| self.args = args |
| self.thread = threading.Thread(target=self._run) |
| self.thread.daemon = True |
| self.timed_out = None |
| |
| def _run(self): |
| while self._running and time.time() < self.end: |
| time.sleep(10) |
| if self._running: |
| self.timed_out = True |
| self.function(*self.args) |
| self.timed_out = False |
| |
| def start(self): |
| self._running = True |
| self.end = time.time() + self.timeout |
| self.thread.start() |
| |
| def stop(self): |
| self._running = False |
| |
| # TODOv3(mordred): put git repos in a hierarchy that includes source |
| # hostname, eg: git.openstack.org/openstack/nova. Also, configure |
| # sources to have an alias, so that the review.openstack.org source |
| # repos end up in git.openstack.org. |
| |
| |
| class JobDirPlaybook(object): |
| def __init__(self, root): |
| self.root = root |
| self.trusted = None |
| self.path = None |
| |
| |
| class JobDir(object): |
| def __init__(self, root=None, keep=False): |
| # root |
| # ansible |
| # trusted.cfg |
| # untrusted.cfg |
| # work |
| # src |
| # logs |
| self.keep = keep |
| self.root = tempfile.mkdtemp(dir=root) |
| # Work |
| self.work_root = os.path.join(self.root, 'work') |
| os.makedirs(self.work_root) |
| self.src_root = os.path.join(self.work_root, 'src') |
| os.makedirs(self.src_root) |
| self.log_root = os.path.join(self.work_root, 'logs') |
| os.makedirs(self.log_root) |
| # Ansible |
| self.ansible_root = os.path.join(self.root, 'ansible') |
| os.makedirs(self.ansible_root) |
| self.known_hosts = os.path.join(self.ansible_root, 'known_hosts') |
| self.inventory = os.path.join(self.ansible_root, 'inventory') |
| self.vars = os.path.join(self.ansible_root, 'vars.yaml') |
| self.playbooks = [] # The list of candidate playbooks |
| self.playbook = None # A pointer to the candidate we have chosen |
| self.pre_playbooks = [] |
| self.post_playbooks = [] |
| self.roles = [] |
| self.roles_path = [] |
| self.untrusted_config = os.path.join( |
| self.ansible_root, 'untrusted.cfg') |
| self.trusted_config = os.path.join(self.ansible_root, 'trusted.cfg') |
| self.ansible_log = os.path.join(self.log_root, 'ansible_log.txt') |
| |
| def addPrePlaybook(self): |
| count = len(self.pre_playbooks) |
| root = os.path.join(self.ansible_root, 'pre_playbook_%i' % (count,)) |
| os.makedirs(root) |
| playbook = JobDirPlaybook(root) |
| self.pre_playbooks.append(playbook) |
| return playbook |
| |
| def addPostPlaybook(self): |
| count = len(self.post_playbooks) |
| root = os.path.join(self.ansible_root, 'post_playbook_%i' % (count,)) |
| os.makedirs(root) |
| playbook = JobDirPlaybook(root) |
| self.post_playbooks.append(playbook) |
| return playbook |
| |
| def addPlaybook(self): |
| count = len(self.playbooks) |
| root = os.path.join(self.ansible_root, 'playbook_%i' % (count,)) |
| os.makedirs(root) |
| playbook = JobDirPlaybook(root) |
| self.playbooks.append(playbook) |
| return playbook |
| |
| def addRole(self): |
| count = len(self.roles) |
| root = os.path.join(self.ansible_root, 'role_%i' % (count,)) |
| os.makedirs(root) |
| self.roles.append(root) |
| return root |
| |
| def cleanup(self): |
| if not self.keep: |
| shutil.rmtree(self.root) |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, etype, value, tb): |
| self.cleanup() |
| |
| |
| class UpdateTask(object): |
| def __init__(self, project, url): |
| self.project = project |
| self.url = url |
| self.event = threading.Event() |
| |
| def __eq__(self, other): |
| if other.project == self.project: |
| return True |
| return False |
| |
| def wait(self): |
| self.event.wait() |
| |
| def setComplete(self): |
| self.event.set() |
| |
| |
| class DeduplicateQueue(object): |
| def __init__(self): |
| self.queue = collections.deque() |
| self.condition = threading.Condition() |
| |
| def qsize(self): |
| return len(self.queue) |
| |
| def put(self, item): |
| # Returns the original item if added, or an equivalent item if |
| # already enqueued. |
| self.condition.acquire() |
| ret = None |
| try: |
| for x in self.queue: |
| if item == x: |
| ret = x |
| if ret is None: |
| ret = item |
| self.queue.append(item) |
| self.condition.notify() |
| finally: |
| self.condition.release() |
| return ret |
| |
| def get(self): |
| self.condition.acquire() |
| try: |
| while True: |
| try: |
| ret = self.queue.popleft() |
| return ret |
| except IndexError: |
| pass |
| self.condition.wait() |
| finally: |
| self.condition.release() |
| |
| |
| class ExecutorServer(object): |
| log = logging.getLogger("zuul.ExecutorServer") |
| |
| def __init__(self, config, connections={}, jobdir_root=None, |
| keep_jobdir=False): |
| self.config = config |
| self.keep_jobdir = keep_jobdir |
| self.jobdir_root = jobdir_root |
| # TODOv3(mordred): make the executor name more unique -- |
| # perhaps hostname+pid. |
| self.hostname = socket.gethostname() |
| self.zuul_url = config.get('merger', 'zuul_url') |
| self.command_map = dict( |
| stop=self.stop, |
| pause=self.pause, |
| unpause=self.unpause, |
| graceful=self.graceful, |
| verbose=self.verboseOn, |
| unverbose=self.verboseOff, |
| ) |
| |
| if self.config.has_option('executor', 'git_dir'): |
| self.merge_root = self.config.get('executor', 'git_dir') |
| else: |
| self.merge_root = '/var/lib/zuul/executor-git' |
| |
| if self.config.has_option('merger', 'git_user_email'): |
| self.merge_email = self.config.get('merger', 'git_user_email') |
| else: |
| self.merge_email = None |
| |
| if self.config.has_option('merger', 'git_user_name'): |
| self.merge_name = self.config.get('merger', 'git_user_name') |
| else: |
| self.merge_name = None |
| |
| self.connections = connections |
| # This merger and its git repos are used to maintain |
| # up-to-date copies of all the repos that are used by jobs, as |
| # well as to support the merger:cat functon to supply |
| # configuration information to Zuul when it starts. |
| self.merger = self._getMerger(self.merge_root) |
| self.update_queue = DeduplicateQueue() |
| |
| 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' |
| path = os.path.join(state_dir, 'executor.socket') |
| self.command_socket = commandsocket.CommandSocket(path) |
| ansible_dir = os.path.join(state_dir, 'ansible') |
| self.library_dir = os.path.join(ansible_dir, 'library') |
| if not os.path.exists(self.library_dir): |
| os.makedirs(self.library_dir) |
| self.action_dir = os.path.join(ansible_dir, 'action') |
| if not os.path.exists(self.action_dir): |
| os.makedirs(self.action_dir) |
| |
| self.callback_dir = os.path.join(ansible_dir, 'callback') |
| if not os.path.exists(self.callback_dir): |
| os.makedirs(self.callback_dir) |
| |
| self.lookup_dir = os.path.join(ansible_dir, 'lookup') |
| if not os.path.exists(self.lookup_dir): |
| os.makedirs(self.lookup_dir) |
| |
| library_path = os.path.dirname(os.path.abspath( |
| zuul.ansible.library.__file__)) |
| for fn in os.listdir(library_path): |
| shutil.copy(os.path.join(library_path, fn), self.library_dir) |
| |
| action_path = os.path.dirname(os.path.abspath( |
| zuul.ansible.action.__file__)) |
| for fn in os.listdir(action_path): |
| shutil.copy(os.path.join(action_path, fn), self.action_dir) |
| |
| callback_path = os.path.dirname(os.path.abspath( |
| zuul.ansible.callback.__file__)) |
| for fn in os.listdir(callback_path): |
| shutil.copy(os.path.join(callback_path, fn), self.callback_dir) |
| |
| lookup_path = os.path.dirname(os.path.abspath( |
| zuul.ansible.lookup.__file__)) |
| for fn in os.listdir(lookup_path): |
| shutil.copy(os.path.join(lookup_path, fn), self.lookup_dir) |
| |
| self.job_workers = {} |
| |
| def _getMerger(self, root): |
| return zuul.merger.merger.Merger(root, self.connections, |
| self.merge_email, self.merge_name) |
| |
| def start(self): |
| self._running = True |
| self._command_running = True |
| server = self.config.get('gearman', 'server') |
| if self.config.has_option('gearman', 'port'): |
| port = self.config.get('gearman', 'port') |
| else: |
| port = 4730 |
| self.worker = gear.Worker('Zuul Executor Server') |
| self.worker.addServer(server, port) |
| self.log.debug("Waiting for server") |
| self.worker.waitForServer() |
| self.log.debug("Registering") |
| self.register() |
| |
| self.log.debug("Starting command processor") |
| self.command_socket.start() |
| self.command_thread = threading.Thread(target=self.runCommand) |
| self.command_thread.daemon = True |
| self.command_thread.start() |
| |
| self.log.debug("Starting worker") |
| self.update_thread = threading.Thread(target=self._updateLoop) |
| self.update_thread.daemon = True |
| self.update_thread.start() |
| self.thread = threading.Thread(target=self.run) |
| self.thread.daemon = True |
| self.thread.start() |
| |
| def register(self): |
| self.worker.registerFunction("executor:execute") |
| self.worker.registerFunction("executor:stop:%s" % self.hostname) |
| self.worker.registerFunction("merger:merge") |
| self.worker.registerFunction("merger:cat") |
| |
| def stop(self): |
| self.log.debug("Stopping") |
| self._running = False |
| self.worker.shutdown() |
| self._command_running = False |
| self.command_socket.stop() |
| self.update_queue.put(None) |
| self.log.debug("Stopped") |
| |
| def pause(self): |
| # TODOv3: implement |
| pass |
| |
| def unpause(self): |
| # TODOv3: implement |
| pass |
| |
| def graceful(self): |
| # TODOv3: implement |
| pass |
| |
| def verboseOn(self): |
| # TODOv3: implement |
| pass |
| |
| def verboseOff(self): |
| # TODOv3: implement |
| pass |
| |
| def join(self): |
| self.update_thread.join() |
| self.thread.join() |
| |
| def runCommand(self): |
| while self._command_running: |
| try: |
| command = self.command_socket.get() |
| if command != '_stop': |
| self.command_map[command]() |
| except Exception: |
| self.log.exception("Exception while processing command") |
| |
| def _updateLoop(self): |
| while self._running: |
| try: |
| self._innerUpdateLoop() |
| except: |
| self.log.exception("Exception in update thread:") |
| |
| def _innerUpdateLoop(self): |
| # Inside of a loop that keeps the main repositories up to date |
| task = self.update_queue.get() |
| if task is None: |
| # We are asked to stop |
| return |
| self.log.info("Updating repo %s from %s" % (task.project, task.url)) |
| self.merger.updateRepo(task.project, task.url) |
| self.log.debug("Finished updating repo %s from %s" % |
| (task.project, task.url)) |
| task.setComplete() |
| |
| def update(self, project, url): |
| # Update a repository in the main merger |
| task = UpdateTask(project, url) |
| task = self.update_queue.put(task) |
| return task |
| |
| def run(self): |
| self.log.debug("Starting executor listener") |
| while self._running: |
| try: |
| job = self.worker.getJob() |
| try: |
| if job.name == 'executor:execute': |
| self.log.debug("Got execute job: %s" % job.unique) |
| self.executeJob(job) |
| elif job.name.startswith('executor:stop'): |
| self.log.debug("Got stop job: %s" % job.unique) |
| self.stopJob(job) |
| elif job.name == 'merger:cat': |
| self.log.debug("Got cat job: %s" % job.unique) |
| self.cat(job) |
| elif job.name == 'merger:merge': |
| self.log.debug("Got merge job: %s" % job.unique) |
| self.merge(job) |
| else: |
| self.log.error("Unable to handle job %s" % job.name) |
| job.sendWorkFail() |
| except Exception: |
| self.log.exception("Exception while running job") |
| job.sendWorkException(traceback.format_exc()) |
| except gear.InterruptedError: |
| pass |
| except Exception: |
| self.log.exception("Exception while getting job") |
| |
| def executeJob(self, job): |
| self.job_workers[job.unique] = AnsibleJob(self, job) |
| self.job_workers[job.unique].run() |
| |
| def finishJob(self, unique): |
| del(self.job_workers[unique]) |
| |
| def stopJob(self, job): |
| try: |
| args = json.loads(job.arguments) |
| self.log.debug("Stop job with arguments: %s" % (args,)) |
| unique = args['uuid'] |
| job_worker = self.job_workers.get(unique) |
| if not job_worker: |
| self.log.debug("Unable to find worker for job %s" % (unique,)) |
| return |
| try: |
| job_worker.stop() |
| except Exception: |
| self.log.exception("Exception sending stop command " |
| "to worker:") |
| finally: |
| job.sendWorkComplete() |
| |
| def cat(self, job): |
| args = json.loads(job.arguments) |
| task = self.update(args['project'], args['url']) |
| task.wait() |
| files = self.merger.getFiles(args['project'], args['url'], |
| args['branch'], args['files']) |
| result = dict(updated=True, |
| files=files, |
| zuul_url=self.zuul_url) |
| job.sendWorkComplete(json.dumps(result)) |
| |
| def merge(self, job): |
| args = json.loads(job.arguments) |
| ret = self.merger.mergeChanges(args['items'], args.get('files')) |
| result = dict(merged=(ret is not None), |
| zuul_url=self.zuul_url) |
| if args.get('files'): |
| if ret: |
| result['commit'], result['files'] = ret |
| else: |
| result['commit'], result['files'] = (None, None) |
| else: |
| result['commit'] = ret |
| job.sendWorkComplete(json.dumps(result)) |
| |
| |
| class AnsibleJob(object): |
| log = logging.getLogger("zuul.AnsibleJob") |
| |
| RESULT_NORMAL = 1 |
| RESULT_TIMED_OUT = 2 |
| RESULT_UNREACHABLE = 3 |
| RESULT_ABORTED = 4 |
| |
| def __init__(self, executor_server, job): |
| self.executor_server = executor_server |
| self.job = job |
| self.jobdir = None |
| self.proc = None |
| self.proc_lock = threading.Lock() |
| self.running = False |
| self.aborted = False |
| |
| if self.executor_server.config.has_option( |
| 'executor', 'private_key_file'): |
| self.private_key_file = self.executor_server.config.get( |
| 'executor', 'private_key_file') |
| else: |
| self.private_key_file = '~/.ssh/id_rsa' |
| |
| def run(self): |
| self.running = True |
| self.thread = threading.Thread(target=self.execute) |
| self.thread.start() |
| |
| def stop(self): |
| self.aborted = True |
| self.abortRunningProc() |
| self.thread.join() |
| |
| def execute(self): |
| try: |
| self.jobdir = JobDir(root=self.executor_server.jobdir_root, |
| keep=self.executor_server.keep_jobdir) |
| self._execute() |
| except Exception: |
| self.log.exception("Exception while executing job") |
| self.job.sendWorkException(traceback.format_exc()) |
| finally: |
| self.running = False |
| try: |
| self.jobdir.cleanup() |
| except Exception: |
| self.log.exception("Error cleaning up jobdir:") |
| try: |
| self.executor_server.finishJob(self.job.unique) |
| except Exception: |
| self.log.exception("Error finalizing job thread:") |
| |
| def _execute(self): |
| self.log.debug("Job %s: beginning" % (self.job.unique,)) |
| self.log.debug("Job %s: args: %s" % (self.job.unique, |
| self.job.arguments,)) |
| self.log.debug("Job %s: job root at %s" % |
| (self.job.unique, self.jobdir.root)) |
| args = json.loads(self.job.arguments) |
| tasks = [] |
| for project in args['projects']: |
| self.log.debug("Job %s: updating project %s" % |
| (self.job.unique, project['name'])) |
| tasks.append(self.executor_server.update( |
| project['name'], project['url'])) |
| for task in tasks: |
| task.wait() |
| |
| self.log.debug("Job %s: git updates complete" % (self.job.unique,)) |
| for project in args['projects']: |
| self.log.debug("Cloning %s" % (project['name'],)) |
| repo = git.Repo.clone_from( |
| os.path.join(self.executor_server.merge_root, |
| project['name']), |
| os.path.join(self.jobdir.src_root, |
| project['name'])) |
| repo.remotes.origin.config_writer.set('url', project['url']) |
| |
| merge_items = [i for i in args['items'] if i.get('refspec')] |
| if merge_items: |
| commit = self.doMergeChanges(merge_items) |
| if not commit: |
| # There was a merge conflict and we have already sent |
| # a work complete result, don't run any jobs |
| return |
| else: |
| commit = args['items'][-1]['newrev'] # noqa |
| |
| # is the playbook in a repo that we have already prepared? |
| self.preparePlaybookRepos(args) |
| |
| self.prepareRoles(args) |
| |
| # TODOv3: Ansible the ansible thing here. |
| self.prepareAnsibleFiles(args) |
| |
| data = { |
| 'manager': self.executor_server.hostname, |
| 'url': 'https://server/job/{}/0/'.format(args['job']), |
| 'worker_name': 'My Worker', |
| } |
| |
| # TODOv3: |
| # 'name': self.name, |
| # 'manager': self.executor_server.hostname, |
| # 'worker_name': 'My Worker', |
| # 'worker_hostname': 'localhost', |
| # 'worker_ips': ['127.0.0.1', '192.168.1.1'], |
| # 'worker_fqdn': 'zuul.example.org', |
| # 'worker_program': 'FakeBuilder', |
| # 'worker_version': 'v1.1', |
| # 'worker_extra': {'something': 'else'} |
| |
| self.job.sendWorkData(json.dumps(data)) |
| self.job.sendWorkStatus(0, 100) |
| |
| result = self.runPlaybooks(args) |
| |
| if result is None: |
| self.job.sendWorkFail() |
| return |
| result = dict(result=result) |
| self.job.sendWorkComplete(json.dumps(result)) |
| |
| def doMergeChanges(self, items): |
| # Get a merger in order to update the repos involved in this job. |
| merger = self.executor_server._getMerger(self.jobdir.src_root) |
| commit = merger.mergeChanges(items) # noqa |
| if not commit: # merge conflict |
| result = dict(result='MERGER_FAILURE') |
| self.job.sendWorkComplete(json.dumps(result)) |
| return commit |
| |
| def runPlaybooks(self, args): |
| result = None |
| |
| for playbook in self.jobdir.pre_playbooks: |
| # TODOv3(pabelanger): Implement pre-run timeout setting. |
| pre_status, pre_code = self.runAnsiblePlaybook( |
| playbook, args['timeout']) |
| if pre_status != self.RESULT_NORMAL or pre_code != 0: |
| # These should really never fail, so return None and have |
| # zuul try again |
| return result |
| |
| job_status, job_code = self.runAnsiblePlaybook( |
| self.jobdir.playbook, args['timeout']) |
| if job_status == self.RESULT_TIMED_OUT: |
| return 'TIMED_OUT' |
| if job_status == self.RESULT_ABORTED: |
| return 'ABORTED' |
| if job_status != self.RESULT_NORMAL: |
| # The result of the job is indeterminate. Zuul will |
| # run it again. |
| return result |
| |
| success = (job_code == 0) |
| if success: |
| result = 'SUCCESS' |
| else: |
| result = 'FAILURE' |
| |
| for playbook in self.jobdir.post_playbooks: |
| # TODOv3(pabelanger): Implement post-run timeout setting. |
| post_status, post_code = self.runAnsiblePlaybook( |
| playbook, args['timeout'], success) |
| if post_status != self.RESULT_NORMAL or post_code != 0: |
| result = 'POST_FAILURE' |
| return result |
| |
| def getHostList(self, args): |
| hosts = [] |
| for node in args['nodes']: |
| # NOTE(mordred): This assumes that the nodepool launcher |
| # and the zuul executor both have similar network |
| # characteristics, as the launcher will do a test for ipv6 |
| # viability and if so, and if the node has an ipv6 |
| # address, it will be the interface_ip. force-ipv4 can be |
| # set to True in the clouds.yaml for a cloud if this |
| # results in the wrong thing being in interface_ip |
| # TODO(jeblair): Move this notice to the docs. |
| ip = node.get('interface_ip') |
| host_vars = dict( |
| ansible_host=ip, |
| nodepool_az=node.get('az'), |
| nodepool_provider=node.get('provider'), |
| nodepool_region=node.get('region')) |
| |
| host_keys = [] |
| for key in node.get('host_keys'): |
| host_keys.append("%s %s" % (ip, key)) |
| |
| hosts.append(dict( |
| name=node['name'], |
| host_vars=host_vars, |
| host_keys=host_keys)) |
| return hosts |
| |
| def _blockPluginDirs(self, path): |
| '''Prevent execution of playbooks or roles with plugins |
| |
| Plugins are loaded from roles and also if there is a plugin |
| dir adjacent to the playbook. Throw an error if the path |
| contains a location that would cause a plugin to get loaded. |
| |
| ''' |
| for entry in os.listdir(path): |
| if os.path.isdir(entry) and entry.endswith('_plugins'): |
| raise Exception( |
| "Ansible plugin dir %s found adjacent to playbook %s in" |
| " non-trusted repo." % (entry, path)) |
| |
| def findPlaybook(self, path, required=False, trusted=False): |
| for ext in ['.yaml', '.yml']: |
| fn = path + ext |
| if os.path.exists(fn): |
| if not trusted: |
| playbook_dir = os.path.dirname(os.path.abspath(fn)) |
| self._blockPluginDirs(playbook_dir) |
| return fn |
| if required: |
| raise Exception("Unable to find playbook %s" % path) |
| return None |
| |
| def preparePlaybookRepos(self, args): |
| for playbook in args['pre_playbooks']: |
| jobdir_playbook = self.jobdir.addPrePlaybook() |
| self.preparePlaybookRepo(jobdir_playbook, playbook, |
| args, required=True) |
| |
| for playbook in args['playbooks']: |
| jobdir_playbook = self.jobdir.addPlaybook() |
| self.preparePlaybookRepo(jobdir_playbook, playbook, |
| args, required=False) |
| if jobdir_playbook.path is not None: |
| self.jobdir.playbook = jobdir_playbook |
| break |
| if self.jobdir.playbook is None: |
| raise Exception("No valid playbook found") |
| |
| for playbook in args['post_playbooks']: |
| jobdir_playbook = self.jobdir.addPostPlaybook() |
| self.preparePlaybookRepo(jobdir_playbook, playbook, |
| args, required=True) |
| |
| def preparePlaybookRepo(self, jobdir_playbook, playbook, args, required): |
| self.log.debug("Prepare playbook repo for %s" % (playbook,)) |
| # Check out the playbook repo if needed and set the path to |
| # the playbook that should be run. |
| jobdir_playbook.trusted = playbook['trusted'] |
| source = self.executor_server.connections.getSource( |
| playbook['connection']) |
| project = source.getProject(playbook['project']) |
| # TODO(jeblair): construct the url in the merger itself |
| url = source.getGitUrl(project) |
| if not playbook['trusted']: |
| # This is a project repo, so it is safe to use the already |
| # checked out version (from speculative merging) of the |
| # playbook |
| for i in args['items']: |
| if (i['connection_name'] == playbook['connection'] and |
| i['project'] == playbook['project']): |
| # We already have this repo prepared |
| path = os.path.join(self.jobdir.src_root, |
| project.name, |
| playbook['path']) |
| jobdir_playbook.path = self.findPlaybook( |
| path, |
| required=required, |
| trusted=playbook['trusted']) |
| return |
| # The playbook repo is either a config repo, or it isn't in |
| # the stack of changes we are testing, so check out the branch |
| # tip into a dedicated space. |
| |
| merger = self.executor_server._getMerger(jobdir_playbook.root) |
| merger.checkoutBranch(project.name, url, playbook['branch']) |
| |
| path = os.path.join(jobdir_playbook.root, |
| project.name, |
| playbook['path']) |
| jobdir_playbook.path = self.findPlaybook( |
| path, |
| required=required, |
| trusted=playbook['trusted']) |
| |
| def prepareRoles(self, args): |
| for role in args['roles']: |
| if role['type'] == 'zuul': |
| root = self.jobdir.addRole() |
| self.prepareZuulRole(args, role, root) |
| |
| def findRole(self, path, trusted=False): |
| d = os.path.join(path, 'tasks') |
| if os.path.isdir(d): |
| # This is a bare role |
| if not trusted: |
| self._blockPluginDirs(path) |
| # None signifies that the repo is a bare role |
| return None |
| d = os.path.join(path, 'roles') |
| if os.path.isdir(d): |
| # This repo has a collection of roles |
| if not trusted: |
| for entry in os.listdir(d): |
| self._blockPluginDirs(os.path.join(d, entry)) |
| return d |
| # We assume the repository itself is a collection of roles |
| if not trusted: |
| for entry in os.listdir(path): |
| self._blockPluginDirs(os.path.join(path, entry)) |
| return path |
| |
| def prepareZuulRole(self, args, role, root): |
| self.log.debug("Prepare zuul role for %s" % (role,)) |
| # Check out the role repo if needed |
| source = self.executor_server.connections.getSource( |
| role['connection']) |
| project = source.getProject(role['project']) |
| # TODO(jeblair): construct the url in the merger itself |
| url = source.getGitUrl(project) |
| role_repo = None |
| if not role['trusted']: |
| # This is a project repo, so it is safe to use the already |
| # checked out version (from speculative merging) of the |
| # role |
| |
| for i in args['items']: |
| if (i['connection_name'] == role['connection'] and |
| i['project'] == role['project']): |
| # We already have this repo prepared; |
| # copy it into location. |
| |
| path = os.path.join(self.jobdir.src_root, |
| project.name) |
| link = os.path.join(root, role['name']) |
| os.symlink(path, link) |
| role_repo = link |
| break |
| |
| # The role repo is either a config repo, or it isn't in |
| # the stack of changes we are testing, so check out the branch |
| # tip into a dedicated space. |
| |
| if not role_repo: |
| merger = self.executor_server._getMerger(root) |
| merger.checkoutBranch(project.name, url, 'master') |
| role_repo = os.path.join(root, project.name) |
| |
| role_path = self.findRole(role_repo, trusted=role['trusted']) |
| if role_path is None: |
| # In the case of a bare role, add the containing directory |
| role_path = root |
| self.jobdir.roles_path.append(role_path) |
| |
| def prepareAnsibleFiles(self, args): |
| keys = [] |
| with open(self.jobdir.inventory, 'w') as inventory: |
| for item in self.getHostList(args): |
| inventory.write(item['name']) |
| for k, v in item['host_vars'].items(): |
| inventory.write(' %s=%s' % (k, v)) |
| inventory.write('\n') |
| for key in item['host_keys']: |
| keys.append(key) |
| |
| with open(self.jobdir.known_hosts, 'w') as known_hosts: |
| for key in keys: |
| known_hosts.write('%s\n' % key) |
| |
| with open(self.jobdir.vars, 'w') as vars_yaml: |
| zuul_vars = dict(args['vars']) |
| zuul_vars['zuul']['executor'] = dict( |
| hostname=self.executor_server.hostname, |
| src_root=self.jobdir.src_root, |
| log_root=self.jobdir.log_root) |
| vars_yaml.write( |
| yaml.safe_dump(zuul_vars, default_flow_style=False)) |
| self.writeAnsibleConfig(self.jobdir.untrusted_config) |
| self.writeAnsibleConfig(self.jobdir.trusted_config, trusted=True) |
| |
| def writeAnsibleConfig(self, config_path, trusted=False): |
| with open(config_path, 'w') as config: |
| config.write('[defaults]\n') |
| config.write('hostfile = %s\n' % self.jobdir.inventory) |
| config.write('local_tmp = %s/.ansible/local_tmp\n' % |
| self.jobdir.root) |
| config.write('remote_tmp = %s/.ansible/remote_tmp\n' % |
| self.jobdir.root) |
| config.write('private_key_file = %s\n' % self.private_key_file) |
| config.write('retry_files_enabled = False\n') |
| config.write('log_path = %s\n' % self.jobdir.ansible_log) |
| config.write('gathering = explicit\n') |
| config.write('library = %s\n' |
| % self.executor_server.library_dir) |
| if self.jobdir.roles_path: |
| config.write('roles_path = %s\n' % |
| ':'.join(self.jobdir.roles_path)) |
| config.write('command_warnings = False\n') |
| config.write('callback_plugins = %s\n' |
| % self.executor_server.callback_dir) |
| config.write('stdout_callback = zuul_stream\n') |
| # bump the timeout because busy nodes may take more than |
| # 10s to respond |
| config.write('timeout = 30\n') |
| if not trusted: |
| config.write('action_plugins = %s\n' |
| % self.executor_server.action_dir) |
| config.write('lookup_plugins = %s\n' |
| % self.executor_server.lookup_dir) |
| |
| # On trusted jobs, we want to prevent the printing of args, |
| # since trusted jobs might have access to secrets that they may |
| # need to pass to a task or a role. On the other hand, there |
| # should be no sensitive data in untrusted jobs, and printing |
| # the args could be useful for debugging. |
| config.write('display_args_to_stdout = %s\n' % |
| str(not trusted)) |
| |
| config.write('[ssh_connection]\n') |
| # NB: when setting pipelining = True, keep_remote_files |
| # must be False (the default). Otherwise it apparently |
| # will override the pipelining option and effectively |
| # disable it. Pipelining has a side effect of running the |
| # command without a tty (ie, without the -tt argument to |
| # ssh). We require this behavior so that if a job runs a |
| # command which expects interactive input on a tty (such |
| # as sudo) it does not hang. |
| config.write('pipelining = True\n') |
| ssh_args = "-o ControlMaster=auto -o ControlPersist=60s " \ |
| "-o UserKnownHostsFile=%s" % self.jobdir.known_hosts |
| config.write('ssh_args = %s\n' % ssh_args) |
| |
| def _ansibleTimeout(self, msg): |
| self.log.warning(msg) |
| self.abortRunningProc() |
| |
| def abortRunningProc(self): |
| with self.proc_lock: |
| if not self.proc: |
| self.log.debug("Abort: no process is running") |
| return |
| self.log.debug("Abort: sending kill signal to job " |
| "process group") |
| try: |
| pgid = os.getpgid(self.proc.pid) |
| os.killpg(pgid, signal.SIGKILL) |
| except Exception: |
| self.log.exception("Exception while killing ansible process:") |
| |
| def runAnsible(self, cmd, timeout, trusted=False): |
| env_copy = os.environ.copy() |
| env_copy['LOGNAME'] = 'zuul' |
| |
| if trusted: |
| config_file = self.jobdir.trusted_config |
| else: |
| config_file = self.jobdir.untrusted_config |
| |
| env_copy['ANSIBLE_CONFIG'] = config_file |
| |
| with self.proc_lock: |
| if self.aborted: |
| return (self.RESULT_ABORTED, None) |
| self.log.debug("Ansible command: ANSIBLE_CONFIG=%s %s", |
| config_file, " ".join(shlex_quote(c) for c in cmd)) |
| self.proc = subprocess.Popen( |
| cmd, |
| cwd=self.jobdir.work_root, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| preexec_fn=os.setsid, |
| env=env_copy, |
| ) |
| |
| ret = None |
| if timeout: |
| watchdog = Watchdog(timeout, self._ansibleTimeout, |
| ("Ansible timeout exceeded",)) |
| watchdog.start() |
| try: |
| for line in iter(self.proc.stdout.readline, b''): |
| line = line[:1024].rstrip() |
| self.log.debug("Ansible output: %s" % (line,)) |
| ret = self.proc.wait() |
| finally: |
| if timeout: |
| watchdog.stop() |
| self.log.debug("Ansible exit code: %s" % (ret,)) |
| |
| with self.proc_lock: |
| self.proc = None |
| |
| if timeout and watchdog.timed_out: |
| return (self.RESULT_TIMED_OUT, None) |
| if ret == 3: |
| # AnsibleHostUnreachable: We had a network issue connecting to |
| # our zuul-worker. |
| return (self.RESULT_UNREACHABLE, None) |
| elif ret == -9: |
| # Received abort request. |
| return (self.RESULT_ABORTED, None) |
| |
| return (self.RESULT_NORMAL, ret) |
| |
| def runAnsiblePlaybook(self, playbook, timeout, success=None): |
| env_copy = os.environ.copy() |
| env_copy['LOGNAME'] = 'zuul' |
| |
| if False: # TODOv3: self.options['verbose']: |
| verbose = '-vvv' |
| else: |
| verbose = '-v' |
| |
| cmd = ['ansible-playbook', playbook.path] |
| |
| if success is not None: |
| cmd.extend(['-e', 'success=%s' % str(bool(success))]) |
| |
| cmd.extend(['-e@%s' % self.jobdir.vars, verbose]) |
| |
| return self.runAnsible( |
| cmd=cmd, timeout=timeout, trusted=playbook.trusted) |