1813 lines
71 KiB
Python
1813 lines
71 KiB
Python
# 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 datetime
|
|
import json
|
|
import logging
|
|
import multiprocessing
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import shlex
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from zuul.lib.yamlutil import yaml
|
|
from zuul.lib.config import get_default
|
|
|
|
try:
|
|
import ara.plugins.callbacks as ara_callbacks
|
|
except ImportError:
|
|
ara_callbacks = None
|
|
import gear
|
|
|
|
import zuul.merger.merger
|
|
import zuul.ansible.logconfig
|
|
from zuul.lib import commandsocket
|
|
|
|
BUFFER_LINES_FOR_SYNTAX = 200
|
|
COMMANDS = ['stop', 'pause', 'unpause', 'graceful', 'verbose',
|
|
'unverbose', 'keep', 'nokeep']
|
|
DEFAULT_FINGER_PORT = 79
|
|
|
|
|
|
class ExecutorError(Exception):
|
|
"""A non-transient run-time executor error
|
|
|
|
This class represents error conditions detected by the executor
|
|
when preparing to run a job which we know are consistently fatal.
|
|
Zuul should not reschedule the build in these cases.
|
|
"""
|
|
pass
|
|
|
|
|
|
class RoleNotFoundError(ExecutorError):
|
|
pass
|
|
|
|
|
|
class DiskAccountant(object):
|
|
''' A single thread to periodically run du and monitor a base directory
|
|
|
|
Whenever the accountant notices a dir over limit, it will call the
|
|
given func with an argument of the job directory. That function
|
|
should be used to remediate the problem, generally by killing the
|
|
job producing the disk bloat). The function will be called every
|
|
time the problem is noticed, so it should be handled synchronously
|
|
to avoid stacking up calls.
|
|
'''
|
|
log = logging.getLogger("zuul.ExecutorDiskAccountant")
|
|
|
|
def __init__(self, jobs_base, limit, func, cache_dir, usage_func=None):
|
|
'''
|
|
:param str jobs_base: absolute path name of dir to be monitored
|
|
:param int limit: maximum number of MB allowed to be in use in any one
|
|
subdir
|
|
:param callable func: Function to call with overlimit dirs
|
|
:param str cache_dir: absolute path name of dir to be passed as the
|
|
first argument to du. This will ensure du does
|
|
not count any hardlinks to files in this
|
|
directory against a single job.
|
|
:param callable usage_func: Optional function to call with usage
|
|
for every dir _NOT_ over limit
|
|
'''
|
|
# Don't cross the streams
|
|
if cache_dir == jobs_base:
|
|
raise Exception("Cache dir and jobs dir cannot be the same")
|
|
self.thread = threading.Thread(target=self._run,
|
|
name='executor-diskaccountant')
|
|
self.thread.daemon = True
|
|
self._running = False
|
|
self.jobs_base = jobs_base
|
|
self.limit = limit
|
|
self.func = func
|
|
self.cache_dir = cache_dir
|
|
self.usage_func = usage_func
|
|
self.stop_event = threading.Event()
|
|
|
|
def _run(self):
|
|
while self._running:
|
|
# Walk job base
|
|
before = time.time()
|
|
du = subprocess.Popen(
|
|
['du', '-m', '--max-depth=1', self.cache_dir, self.jobs_base],
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
|
|
for line in du.stdout:
|
|
(size, dirname) = line.rstrip().split()
|
|
dirname = dirname.decode('utf8')
|
|
if dirname == self.jobs_base or dirname == self.cache_dir:
|
|
continue
|
|
if os.path.dirname(dirname) == self.cache_dir:
|
|
continue
|
|
size = int(size)
|
|
if size > self.limit:
|
|
self.log.info(
|
|
"{job} is using {size}MB (limit={limit})"
|
|
.format(size=size, job=dirname, limit=self.limit))
|
|
self.func(dirname)
|
|
elif self.usage_func:
|
|
self.log.debug(
|
|
"{job} is using {size}MB (limit={limit})"
|
|
.format(size=size, job=dirname, limit=self.limit))
|
|
self.usage_func(dirname, size)
|
|
du.wait()
|
|
after = time.time()
|
|
# Sleep half as long as that took, or 1s, whichever is longer
|
|
delay_time = max((after - before) / 2, 1.0)
|
|
self.stop_event.wait(delay_time)
|
|
|
|
def start(self):
|
|
self._running = True
|
|
self.thread.start()
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
self.stop_event.set()
|
|
# We join here to avoid whitelisting the thread -- if it takes more
|
|
# than 5s to stop in tests, there's a problem.
|
|
self.thread.join(timeout=5)
|
|
|
|
|
|
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,
|
|
name='executor-watchdog')
|
|
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)
|
|
else:
|
|
# Only set timed_out to false if we aren't _running
|
|
# anymore. This means that we stopped running not because
|
|
# of a timeout but because normal execution ended.
|
|
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
|
|
|
|
|
|
class SshAgent(object):
|
|
log = logging.getLogger("zuul.ExecutorServer")
|
|
|
|
def __init__(self):
|
|
self.env = {}
|
|
self.ssh_agent = None
|
|
|
|
def start(self):
|
|
if self.ssh_agent:
|
|
return
|
|
with open('/dev/null', 'r+') as devnull:
|
|
ssh_agent = subprocess.Popen(['ssh-agent'], close_fds=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=devnull,
|
|
stdin=devnull)
|
|
(output, _) = ssh_agent.communicate()
|
|
output = output.decode('utf8')
|
|
for line in output.split("\n"):
|
|
if '=' in line:
|
|
line = line.split(";", 1)[0]
|
|
(key, value) = line.split('=')
|
|
self.env[key] = value
|
|
self.log.info('Started SSH Agent, {}'.format(self.env))
|
|
|
|
def stop(self):
|
|
if 'SSH_AGENT_PID' in self.env:
|
|
try:
|
|
os.kill(int(self.env['SSH_AGENT_PID']), signal.SIGTERM)
|
|
except OSError:
|
|
self.log.exception(
|
|
'Problem sending SIGTERM to agent {}'.format(self.env))
|
|
self.log.debug('Sent SIGTERM to SSH Agent, {}'.format(self.env))
|
|
self.env = {}
|
|
|
|
def add(self, key_path):
|
|
env = os.environ.copy()
|
|
env.update(self.env)
|
|
key_path = os.path.expanduser(key_path)
|
|
self.log.debug('Adding SSH Key {}'.format(key_path))
|
|
try:
|
|
subprocess.check_output(['ssh-add', key_path], env=env,
|
|
stderr=subprocess.PIPE)
|
|
except subprocess.CalledProcessError as e:
|
|
self.log.exception('ssh-add failed. stdout: %s, stderr: %s',
|
|
e.output, e.stderr)
|
|
raise
|
|
self.log.info('Added SSH Key {}'.format(key_path))
|
|
|
|
def remove(self, key_path):
|
|
env = os.environ.copy()
|
|
env.update(self.env)
|
|
key_path = os.path.expanduser(key_path)
|
|
self.log.debug('Removing SSH Key {}'.format(key_path))
|
|
subprocess.check_output(['ssh-add', '-d', key_path], env=env,
|
|
stderr=subprocess.PIPE)
|
|
self.log.info('Removed SSH Key {}'.format(key_path))
|
|
|
|
def list(self):
|
|
if 'SSH_AUTH_SOCK' not in self.env:
|
|
return None
|
|
env = os.environ.copy()
|
|
env.update(self.env)
|
|
result = []
|
|
for line in subprocess.Popen(['ssh-add', '-L'], env=env,
|
|
stdout=subprocess.PIPE).stdout:
|
|
line = line.decode('utf8')
|
|
if line.strip() == 'The agent has no identities.':
|
|
break
|
|
result.append(line.strip())
|
|
return result
|
|
|
|
|
|
class JobDirPlaybook(object):
|
|
def __init__(self, root):
|
|
self.root = root
|
|
self.trusted = None
|
|
self.branch = None
|
|
self.canonical_name_and_path = None
|
|
self.path = None
|
|
self.roles = []
|
|
self.roles_path = []
|
|
self.ansible_config = os.path.join(self.root, 'ansible.cfg')
|
|
self.project_link = os.path.join(self.root, 'project')
|
|
self.secrets_root = os.path.join(self.root, 'secrets')
|
|
os.makedirs(self.secrets_root)
|
|
self.secrets = os.path.join(self.secrets_root, 'secrets.yaml')
|
|
self.secrets_content = None
|
|
|
|
def addRole(self):
|
|
count = len(self.roles)
|
|
root = os.path.join(self.root, 'role_%i' % (count,))
|
|
os.makedirs(root)
|
|
self.roles.append(root)
|
|
return root
|
|
|
|
|
|
class JobDir(object):
|
|
def __init__(self, root, keep, build_uuid):
|
|
'''
|
|
:param str root: Root directory for the individual job directories.
|
|
Can be None to use the default system temp root directory.
|
|
:param bool keep: If True, do not delete the job directory.
|
|
:param str build_uuid: The unique build UUID. If supplied, this will
|
|
be used as the temp job directory name. Using this will help the
|
|
log streaming daemon find job logs.
|
|
'''
|
|
# root
|
|
# ansible (mounted in bwrap read-only)
|
|
# logging.json
|
|
# inventory.yaml
|
|
# .ansible (mounted in bwrap read-write)
|
|
# fact-cache/localhost
|
|
# playbook_0 (mounted in bwrap for each playbook read-only)
|
|
# secrets.yaml
|
|
# project -> ../trusted/project_0/...
|
|
# role_0 -> ../trusted/project_0/...
|
|
# trusted (mounted in bwrap read-only)
|
|
# project_0
|
|
# <git.example.com>
|
|
# <project>
|
|
# work (mounted in bwrap read-write)
|
|
# .ssh
|
|
# known_hosts
|
|
# src
|
|
# <git.example.com>
|
|
# <project>
|
|
# logs
|
|
# job-output.txt
|
|
# results.json
|
|
self.keep = keep
|
|
if root:
|
|
tmpdir = root
|
|
else:
|
|
tmpdir = tempfile.gettempdir()
|
|
self.root = os.path.join(tmpdir, build_uuid)
|
|
os.mkdir(self.root, 0o700)
|
|
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)
|
|
self.ansible_root = os.path.join(self.root, 'ansible')
|
|
os.makedirs(self.ansible_root)
|
|
self.trusted_root = os.path.join(self.root, 'trusted')
|
|
os.makedirs(self.trusted_root)
|
|
ssh_dir = os.path.join(self.work_root, '.ssh')
|
|
os.mkdir(ssh_dir, 0o700)
|
|
# Create ansible cache directory
|
|
self.ansible_cache_root = os.path.join(self.root, '.ansible')
|
|
self.fact_cache = os.path.join(self.ansible_cache_root, 'fact-cache')
|
|
os.makedirs(self.fact_cache)
|
|
localhost_facts = os.path.join(self.fact_cache, 'localhost')
|
|
# NOTE(pabelanger): We do not want to leak zuul-executor facts to other
|
|
# playbooks now that smart fact gathering is enabled by default. We
|
|
# can have ansible skip populating the cache with information by the
|
|
# doing the following.
|
|
with open(localhost_facts, 'w') as f:
|
|
f.write('{"module_setup": true}')
|
|
|
|
self.result_data_file = os.path.join(self.work_root, 'results.json')
|
|
with open(self.result_data_file, 'w'):
|
|
pass
|
|
self.known_hosts = os.path.join(ssh_dir, 'known_hosts')
|
|
self.inventory = os.path.join(self.ansible_root, 'inventory.yaml')
|
|
self.logging_json = os.path.join(self.ansible_root, 'logging.json')
|
|
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.job_output_file = os.path.join(self.log_root, 'job-output.txt')
|
|
self.trusted_projects = []
|
|
self.trusted_project_index = {}
|
|
|
|
# Create a JobDirPlaybook for the Ansible setup run. This
|
|
# doesn't use an actual playbook, but it lets us use the same
|
|
# methods to write an ansible.cfg as the rest of the Ansible
|
|
# runs.
|
|
setup_root = os.path.join(self.ansible_root, 'setup_playbook')
|
|
os.makedirs(setup_root)
|
|
self.setup_playbook = JobDirPlaybook(setup_root)
|
|
self.setup_playbook.trusted = True
|
|
|
|
def addTrustedProject(self, canonical_name, branch):
|
|
# Trusted projects are placed in their own directories so that
|
|
# we can support using different branches of the same project
|
|
# in different playbooks.
|
|
count = len(self.trusted_projects)
|
|
root = os.path.join(self.trusted_root, 'project_%i' % (count,))
|
|
os.makedirs(root)
|
|
self.trusted_projects.append(root)
|
|
self.trusted_project_index[(canonical_name, branch)] = root
|
|
return root
|
|
|
|
def getTrustedProject(self, canonical_name, branch):
|
|
return self.trusted_project_index.get((canonical_name, branch))
|
|
|
|
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 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, connection_name, project_name):
|
|
self.connection_name = connection_name
|
|
self.project_name = project_name
|
|
self.event = threading.Event()
|
|
|
|
def __eq__(self, other):
|
|
if (other and other.connection_name == self.connection_name and
|
|
other.project_name == self.project_name):
|
|
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()
|
|
|
|
|
|
def _copy_ansible_files(python_module, target_dir):
|
|
library_path = os.path.dirname(os.path.abspath(python_module.__file__))
|
|
for fn in os.listdir(library_path):
|
|
if fn == "__pycache__":
|
|
continue
|
|
full_path = os.path.join(library_path, fn)
|
|
if os.path.isdir(full_path):
|
|
shutil.copytree(full_path, os.path.join(target_dir, fn))
|
|
else:
|
|
shutil.copy(os.path.join(library_path, fn), target_dir)
|
|
|
|
|
|
def make_inventory_dict(nodes, groups, all_vars):
|
|
|
|
hosts = {}
|
|
for node in nodes:
|
|
hosts[node['name']] = node['host_vars']
|
|
|
|
inventory = {
|
|
'all': {
|
|
'hosts': hosts,
|
|
'vars': all_vars,
|
|
}
|
|
}
|
|
|
|
for group in groups:
|
|
group_hosts = {}
|
|
for node_name in group['nodes']:
|
|
# children is a dict with None as values because we don't have
|
|
# and per-group variables. If we did, None would be a dict
|
|
# with the per-group variables
|
|
group_hosts[node_name] = None
|
|
inventory[group['name']] = {'hosts': group_hosts}
|
|
|
|
return inventory
|
|
|
|
|
|
class ExecutorMergeWorker(gear.TextWorker):
|
|
def __init__(self, executor_server, *args, **kw):
|
|
self.zuul_executor_server = executor_server
|
|
super(ExecutorMergeWorker, self).__init__(*args, **kw)
|
|
|
|
def handleNoop(self, packet):
|
|
# Wait until the update queue is empty before responding
|
|
while self.zuul_executor_server.update_queue.qsize():
|
|
time.sleep(1)
|
|
|
|
with self.zuul_executor_server.merger_lock:
|
|
super(ExecutorMergeWorker, self).handleNoop(packet)
|
|
|
|
|
|
class ExecutorExecuteWorker(gear.TextWorker):
|
|
def __init__(self, executor_server, *args, **kw):
|
|
self.zuul_executor_server = executor_server
|
|
super(ExecutorExecuteWorker, self).__init__(*args, **kw)
|
|
|
|
def handleNoop(self, packet):
|
|
# Delay our response to running a new job based on the number
|
|
# of jobs we're currently running, in an attempt to spread
|
|
# load evenly among executors.
|
|
workers = len(self.zuul_executor_server.job_workers)
|
|
delay = (workers ** 2) / 1000.0
|
|
time.sleep(delay)
|
|
return super(ExecutorExecuteWorker, self).handleNoop(packet)
|
|
|
|
|
|
class ExecutorServer(object):
|
|
log = logging.getLogger("zuul.ExecutorServer")
|
|
|
|
def __init__(self, config, connections={}, jobdir_root=None,
|
|
keep_jobdir=False, log_streaming_port=DEFAULT_FINGER_PORT):
|
|
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.log_streaming_port = log_streaming_port
|
|
self.merger_lock = threading.Lock()
|
|
self.verbose = False
|
|
self.command_map = dict(
|
|
stop=self.stop,
|
|
pause=self.pause,
|
|
unpause=self.unpause,
|
|
graceful=self.graceful,
|
|
verbose=self.verboseOn,
|
|
unverbose=self.verboseOff,
|
|
keep=self.keep,
|
|
nokeep=self.nokeep,
|
|
)
|
|
|
|
self.merge_root = get_default(self.config, 'executor', 'git_dir',
|
|
'/var/lib/zuul/executor-git')
|
|
self.default_username = get_default(self.config, 'executor',
|
|
'default_username', 'zuul')
|
|
self.disk_limit_per_job = int(get_default(self.config, 'executor',
|
|
'disk_limit_per_job', 250))
|
|
self.merge_email = get_default(self.config, 'merger', 'git_user_email')
|
|
self.merge_name = get_default(self.config, 'merger', 'git_user_name')
|
|
execution_wrapper_name = get_default(self.config, 'executor',
|
|
'execution_wrapper', 'bubblewrap')
|
|
load_multiplier = float(get_default(self.config, 'executor',
|
|
'load_multiplier', '2.5'))
|
|
self.max_load_avg = multiprocessing.cpu_count() * load_multiplier
|
|
self.accepting_work = False
|
|
self.execution_wrapper = connections.drivers[execution_wrapper_name]
|
|
|
|
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()
|
|
|
|
state_dir = get_default(self.config, 'executor', 'state_dir',
|
|
'/var/lib/zuul', expand_user=True)
|
|
path = os.path.join(state_dir, 'executor.socket')
|
|
self.command_socket = commandsocket.CommandSocket(path)
|
|
ansible_dir = os.path.join(state_dir, 'ansible')
|
|
self.ansible_dir = ansible_dir
|
|
if os.path.exists(ansible_dir):
|
|
shutil.rmtree(ansible_dir)
|
|
|
|
zuul_dir = os.path.join(ansible_dir, 'zuul')
|
|
plugin_dir = os.path.join(zuul_dir, 'ansible')
|
|
|
|
os.makedirs(plugin_dir, mode=0o0755)
|
|
|
|
self.library_dir = os.path.join(plugin_dir, 'library')
|
|
self.action_dir = os.path.join(plugin_dir, 'action')
|
|
self.callback_dir = os.path.join(plugin_dir, 'callback')
|
|
self.lookup_dir = os.path.join(plugin_dir, 'lookup')
|
|
self.filter_dir = os.path.join(plugin_dir, 'filter')
|
|
|
|
_copy_ansible_files(zuul.ansible, plugin_dir)
|
|
|
|
# We're copying zuul.ansible.* into a directory we are going
|
|
# to add to pythonpath, so our plugins can "import
|
|
# zuul.ansible". But we're not installing all of zuul, so
|
|
# create a __init__.py file for the stub "zuul" module.
|
|
with open(os.path.join(zuul_dir, '__init__.py'), 'w'):
|
|
pass
|
|
|
|
self.job_workers = {}
|
|
self.disk_accountant = DiskAccountant(self.jobdir_root,
|
|
self.disk_limit_per_job,
|
|
self.stopJobByJobdir,
|
|
self.merge_root)
|
|
|
|
def _getMerger(self, root, logger=None):
|
|
if root != self.merge_root:
|
|
cache_root = self.merge_root
|
|
else:
|
|
cache_root = None
|
|
return zuul.merger.merger.Merger(root, self.connections,
|
|
self.merge_email, self.merge_name,
|
|
cache_root, logger)
|
|
|
|
def start(self):
|
|
self._running = True
|
|
self._command_running = True
|
|
server = self.config.get('gearman', 'server')
|
|
port = get_default(self.config, 'gearman', 'port', 4730)
|
|
ssl_key = get_default(self.config, 'gearman', 'ssl_key')
|
|
ssl_cert = get_default(self.config, 'gearman', 'ssl_cert')
|
|
ssl_ca = get_default(self.config, 'gearman', 'ssl_ca')
|
|
self.merger_worker = ExecutorMergeWorker(self, 'Zuul Executor Merger')
|
|
self.merger_worker.addServer(server, port, ssl_key, ssl_cert, ssl_ca)
|
|
self.executor_worker = ExecutorExecuteWorker(
|
|
self, 'Zuul Executor Server')
|
|
self.executor_worker.addServer(server, port, ssl_key, ssl_cert, ssl_ca)
|
|
self.log.debug("Waiting for server")
|
|
self.merger_worker.waitForServer()
|
|
self.executor_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.merger_thread = threading.Thread(target=self.run_merger)
|
|
self.merger_thread.daemon = True
|
|
self.merger_thread.start()
|
|
self.executor_thread = threading.Thread(target=self.run_executor)
|
|
self.executor_thread.daemon = True
|
|
self.executor_thread.start()
|
|
self.governor_stop_event = threading.Event()
|
|
self.governor_thread = threading.Thread(target=self.run_governor)
|
|
self.governor_thread.daemon = True
|
|
self.governor_thread.start()
|
|
self.disk_accountant.start()
|
|
|
|
def register(self):
|
|
self.register_work()
|
|
self.executor_worker.registerFunction("executor:stop:%s" %
|
|
self.hostname)
|
|
self.merger_worker.registerFunction("merger:merge")
|
|
self.merger_worker.registerFunction("merger:cat")
|
|
self.merger_worker.registerFunction("merger:refstate")
|
|
|
|
def register_work(self):
|
|
self.accepting_work = True
|
|
self.executor_worker.registerFunction("executor:execute")
|
|
|
|
def unregister_work(self):
|
|
self.accepting_work = False
|
|
self.executor_worker.unregisterFunction("executor:execute")
|
|
|
|
def stop(self):
|
|
self.log.debug("Stopping")
|
|
self.disk_accountant.stop()
|
|
self.governor_stop_event.set()
|
|
self._running = False
|
|
self._command_running = False
|
|
self.command_socket.stop()
|
|
self.update_queue.put(None)
|
|
|
|
for job_worker in list(self.job_workers.values()):
|
|
try:
|
|
job_worker.stop()
|
|
except Exception:
|
|
self.log.exception("Exception sending stop command "
|
|
"to worker:")
|
|
self.merger_worker.shutdown()
|
|
self.executor_worker.shutdown()
|
|
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):
|
|
self.verbose = True
|
|
|
|
def verboseOff(self):
|
|
self.verbose = False
|
|
|
|
def keep(self):
|
|
self.keep_jobdir = True
|
|
|
|
def nokeep(self):
|
|
self.keep_jobdir = False
|
|
|
|
def join(self):
|
|
self.update_thread.join()
|
|
self.merger_thread.join()
|
|
self.executor_thread.join()
|
|
self.governor_thread.join()
|
|
|
|
def runCommand(self):
|
|
while self._command_running:
|
|
try:
|
|
command = self.command_socket.get().decode('utf8')
|
|
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
|
|
with self.merger_lock:
|
|
self.log.info("Updating repo %s/%s" % (
|
|
task.connection_name, task.project_name))
|
|
self.merger.updateRepo(task.connection_name, task.project_name)
|
|
self.log.debug("Finished updating repo %s/%s" %
|
|
(task.connection_name, task.project_name))
|
|
task.setComplete()
|
|
|
|
def update(self, connection_name, project_name):
|
|
# Update a repository in the main merger
|
|
task = UpdateTask(connection_name, project_name)
|
|
task = self.update_queue.put(task)
|
|
return task
|
|
|
|
def run_merger(self):
|
|
self.log.debug("Starting merger listener")
|
|
while self._running:
|
|
try:
|
|
job = self.merger_worker.getJob()
|
|
try:
|
|
if 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)
|
|
elif job.name == 'merger:refstate':
|
|
self.log.debug("Got refstate job: %s" % job.unique)
|
|
self.refstate(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().encode('utf8'))
|
|
except gear.InterruptedError:
|
|
pass
|
|
except Exception:
|
|
self.log.exception("Exception while getting job")
|
|
|
|
def run_executor(self):
|
|
self.log.debug("Starting executor listener")
|
|
while self._running:
|
|
try:
|
|
job = self.executor_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)
|
|
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().encode('utf8'))
|
|
except gear.InterruptedError:
|
|
pass
|
|
except Exception:
|
|
self.log.exception("Exception while getting job")
|
|
|
|
def run_governor(self):
|
|
while not self.governor_stop_event.wait(30):
|
|
self.manageLoad()
|
|
|
|
def executeJob(self, job):
|
|
self.job_workers[job.unique] = AnsibleJob(self, job)
|
|
self.job_workers[job.unique].run()
|
|
|
|
def manageLoad(self):
|
|
''' Apply some heuristics to decide whether or not we should
|
|
be askign for more jobs '''
|
|
load_avg = os.getloadavg()[0]
|
|
if self.accepting_work:
|
|
# Don't unregister if we don't have any active jobs.
|
|
if load_avg > self.max_load_avg and self.job_workers:
|
|
self.log.info(
|
|
"Unregistering due to high system load {} > {}".format(
|
|
load_avg, self.max_load_avg))
|
|
self.unregister_work()
|
|
elif load_avg <= self.max_load_avg:
|
|
self.log.info(
|
|
"Re-registering as load is within limits {} <= {}".format(
|
|
load_avg, self.max_load_avg))
|
|
self.register_work()
|
|
|
|
def finishJob(self, unique):
|
|
del(self.job_workers[unique])
|
|
|
|
def stopJobByJobdir(self, jobdir):
|
|
unique = os.path.basename(jobdir)
|
|
self.stopJobByUnique(unique)
|
|
|
|
def stopJob(self, job):
|
|
try:
|
|
args = json.loads(job.arguments)
|
|
self.log.debug("Stop job with arguments: %s" % (args,))
|
|
unique = args['uuid']
|
|
self.stopJobByUnique(unique)
|
|
finally:
|
|
job.sendWorkComplete()
|
|
|
|
def stopJobByUnique(self, unique):
|
|
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:")
|
|
|
|
def cat(self, job):
|
|
args = json.loads(job.arguments)
|
|
task = self.update(args['connection'], args['project'])
|
|
task.wait()
|
|
with self.merger_lock:
|
|
files = self.merger.getFiles(args['connection'], args['project'],
|
|
args['branch'], args['files'],
|
|
args.get('dirs', []))
|
|
result = dict(updated=True,
|
|
files=files)
|
|
job.sendWorkComplete(json.dumps(result))
|
|
|
|
def refstate(self, job):
|
|
args = json.loads(job.arguments)
|
|
with self.merger_lock:
|
|
success, repo_state = self.merger.getRepoState(args['items'])
|
|
result = dict(updated=success,
|
|
repo_state=repo_state)
|
|
job.sendWorkComplete(json.dumps(result))
|
|
|
|
def merge(self, job):
|
|
args = json.loads(job.arguments)
|
|
with self.merger_lock:
|
|
ret = self.merger.mergeChanges(args['items'], args.get('files'),
|
|
args.get('dirs', []),
|
|
args.get('repo_state'))
|
|
result = dict(merged=(ret is not None))
|
|
if ret is None:
|
|
result['commit'] = result['files'] = result['repo_state'] = None
|
|
else:
|
|
(result['commit'], result['files'], result['repo_state'],
|
|
recent) = ret
|
|
job.sendWorkComplete(json.dumps(result))
|
|
|
|
|
|
class AnsibleJobLogAdapter(logging.LoggerAdapter):
|
|
def process(self, msg, kwargs):
|
|
msg, kwargs = super(AnsibleJobLogAdapter, self).process(msg, kwargs)
|
|
msg = '[build: %s] %s' % (kwargs['extra']['job'], msg)
|
|
return msg, kwargs
|
|
|
|
|
|
class AnsibleJob(object):
|
|
RESULT_NORMAL = 1
|
|
RESULT_TIMED_OUT = 2
|
|
RESULT_UNREACHABLE = 3
|
|
RESULT_ABORTED = 4
|
|
|
|
RESULT_MAP = {
|
|
RESULT_NORMAL: 'RESULT_NORMAL',
|
|
RESULT_TIMED_OUT: 'RESULT_TIMED_OUT',
|
|
RESULT_UNREACHABLE: 'RESULT_UNREACHABLE',
|
|
RESULT_ABORTED: 'RESULT_ABORTED',
|
|
}
|
|
|
|
def __init__(self, executor_server, job):
|
|
logger = logging.getLogger("zuul.AnsibleJob")
|
|
self.log = AnsibleJobLogAdapter(logger, {'job': job.unique})
|
|
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
|
|
self.thread = None
|
|
self.private_key_file = get_default(self.executor_server.config,
|
|
'executor', 'private_key_file',
|
|
'~/.ssh/id_rsa')
|
|
self.ssh_agent = SshAgent()
|
|
|
|
self.executor_variables_file = None
|
|
|
|
if self.executor_server.config.has_option('executor', 'variables'):
|
|
self.executor_variables_file = self.executor_server.config.get(
|
|
'executor', 'variables')
|
|
|
|
def run(self):
|
|
self.running = True
|
|
self.thread = threading.Thread(target=self.execute)
|
|
self.thread.start()
|
|
|
|
def stop(self):
|
|
self.aborted = True
|
|
self.abortRunningProc()
|
|
if self.thread:
|
|
self.thread.join()
|
|
|
|
def execute(self):
|
|
try:
|
|
self.ssh_agent.start()
|
|
self.ssh_agent.add(self.private_key_file)
|
|
self.jobdir = JobDir(self.executor_server.jobdir_root,
|
|
self.executor_server.keep_jobdir,
|
|
str(self.job.unique))
|
|
self._execute()
|
|
except ExecutorError as e:
|
|
result_data = json.dumps(dict(result='ERROR',
|
|
error_detail=e.args[0]))
|
|
self.log.debug("Sending result: %s" % (result_data,))
|
|
self.job.sendWorkComplete(result_data)
|
|
except Exception:
|
|
self.log.exception("Exception while executing job")
|
|
self.job.sendWorkException(traceback.format_exc())
|
|
finally:
|
|
self.running = False
|
|
if self.jobdir:
|
|
try:
|
|
self.jobdir.cleanup()
|
|
except Exception:
|
|
self.log.exception("Error cleaning up jobdir:")
|
|
if self.ssh_agent:
|
|
try:
|
|
self.ssh_agent.stop()
|
|
except Exception:
|
|
self.log.exception("Error stopping SSH agent:")
|
|
try:
|
|
self.executor_server.finishJob(self.job.unique)
|
|
except Exception:
|
|
self.log.exception("Error finalizing job thread:")
|
|
|
|
def _execute(self):
|
|
args = json.loads(self.job.arguments)
|
|
self.log.debug("Beginning job %s for ref %s" %
|
|
(self.job.name, args['zuul']['ref']))
|
|
self.log.debug("Job root: %s" % (self.jobdir.root,))
|
|
tasks = []
|
|
projects = set()
|
|
|
|
# Make sure all projects used by the job are updated...
|
|
for project in args['projects']:
|
|
self.log.debug("Updating project %s" % (project,))
|
|
tasks.append(self.executor_server.update(
|
|
project['connection'], project['name']))
|
|
projects.add((project['connection'], project['name']))
|
|
|
|
# ...as well as all playbook and role projects.
|
|
repos = []
|
|
playbooks = (args['pre_playbooks'] + args['playbooks'] +
|
|
args['post_playbooks'])
|
|
for playbook in playbooks:
|
|
repos.append(playbook)
|
|
repos += playbook['roles']
|
|
|
|
for repo in repos:
|
|
self.log.debug("Updating playbook or role %s" % (repo['project'],))
|
|
key = (repo['connection'], repo['project'])
|
|
if key not in projects:
|
|
tasks.append(self.executor_server.update(*key))
|
|
projects.add(key)
|
|
|
|
for task in tasks:
|
|
task.wait()
|
|
|
|
self.log.debug("Git updates complete")
|
|
merger = self.executor_server._getMerger(self.jobdir.src_root,
|
|
self.log)
|
|
repos = {}
|
|
for project in args['projects']:
|
|
self.log.debug("Cloning %s/%s" % (project['connection'],
|
|
project['name'],))
|
|
repo = merger.getRepo(project['connection'],
|
|
project['name'])
|
|
repos[project['canonical_name']] = repo
|
|
|
|
merge_items = [i for i in args['items'] if i.get('number')]
|
|
if merge_items:
|
|
if not self.doMergeChanges(merger, merge_items,
|
|
args['repo_state']):
|
|
# There was a merge conflict and we have already sent
|
|
# a work complete result, don't run any jobs
|
|
return
|
|
|
|
state_items = [i for i in args['items'] if not i.get('number')]
|
|
if state_items:
|
|
merger.setRepoState(state_items, args['repo_state'])
|
|
|
|
for project in args['projects']:
|
|
repo = repos[project['canonical_name']]
|
|
# If this project is the Zuul project and this is a ref
|
|
# rather than a change, checkout the ref.
|
|
if (project['canonical_name'] ==
|
|
args['zuul']['project']['canonical_name'] and
|
|
(not args['zuul'].get('branch')) and
|
|
args['zuul'].get('ref')):
|
|
ref = args['zuul']['ref']
|
|
else:
|
|
ref = None
|
|
self.checkoutBranch(repo,
|
|
project['name'],
|
|
ref,
|
|
args['branch'],
|
|
args['override_branch'],
|
|
project['override_branch'],
|
|
project['default_branch'])
|
|
|
|
# Delete the origin remote from each repo we set up since
|
|
# it will not be valid within the jobs.
|
|
for repo in repos.values():
|
|
repo.deleteRemote('origin')
|
|
|
|
# This prepares each playbook and the roles needed for each.
|
|
self.preparePlaybooks(args)
|
|
|
|
self.prepareAnsibleFiles(args)
|
|
self.writeLoggingConfig()
|
|
|
|
data = {
|
|
# TODO(mordred) worker_name is needed as a unique name for the
|
|
# client to use for cancelling jobs on an executor. It's defaulting
|
|
# to the hostname for now, but in the future we should allow
|
|
# setting a per-executor override so that one can run more than
|
|
# one executor on a host.
|
|
'worker_name': self.executor_server.hostname,
|
|
'worker_hostname': self.executor_server.hostname,
|
|
'worker_log_port': self.executor_server.log_streaming_port
|
|
}
|
|
if self.executor_server.log_streaming_port != DEFAULT_FINGER_PORT:
|
|
data['url'] = "finger://{hostname}:{port}/{uuid}".format(
|
|
hostname=data['worker_hostname'],
|
|
port=data['worker_log_port'],
|
|
uuid=self.job.unique)
|
|
else:
|
|
data['url'] = 'finger://{hostname}/{uuid}'.format(
|
|
hostname=data['worker_hostname'],
|
|
uuid=self.job.unique)
|
|
|
|
self.job.sendWorkData(json.dumps(data))
|
|
self.job.sendWorkStatus(0, 100)
|
|
|
|
result = self.runPlaybooks(args)
|
|
data = self.getResultData()
|
|
result_data = json.dumps(dict(result=result,
|
|
data=data))
|
|
self.log.debug("Sending result: %s" % (result_data,))
|
|
self.job.sendWorkComplete(result_data)
|
|
|
|
def getResultData(self):
|
|
data = {}
|
|
try:
|
|
with open(self.jobdir.result_data_file) as f:
|
|
file_data = f.read()
|
|
if file_data:
|
|
data = json.loads(file_data)
|
|
except Exception:
|
|
self.log.exception("Unable to load result data:")
|
|
return data
|
|
|
|
def doMergeChanges(self, merger, items, repo_state):
|
|
ret = merger.mergeChanges(items, repo_state=repo_state)
|
|
if not ret: # merge conflict
|
|
result = dict(result='MERGER_FAILURE')
|
|
self.job.sendWorkComplete(json.dumps(result))
|
|
return False
|
|
recent = ret[3]
|
|
for key, commit in recent.items():
|
|
(connection, project, branch) = key
|
|
repo = merger.getRepo(connection, project)
|
|
repo.setRef('refs/heads/' + branch, commit)
|
|
return True
|
|
|
|
def checkoutBranch(self, repo, project_name, ref, zuul_branch,
|
|
job_branch, project_override_branch,
|
|
project_default_branch):
|
|
branches = repo.getBranches()
|
|
if project_override_branch in branches:
|
|
self.log.info("Checking out %s project override branch %s",
|
|
project_name, project_override_branch)
|
|
repo.checkoutLocalBranch(project_override_branch)
|
|
elif job_branch in branches:
|
|
self.log.info("Checking out %s job branch %s",
|
|
project_name, job_branch)
|
|
repo.checkoutLocalBranch(job_branch)
|
|
elif ref and ref.startswith('refs/heads/'):
|
|
b = ref[len('refs/heads/'):]
|
|
self.log.info("Checking out %s branch ref %s",
|
|
project_name, b)
|
|
repo.checkoutLocalBranch(b)
|
|
elif ref and ref.startswith('refs/tags/'):
|
|
t = ref[len('refs/tags/'):]
|
|
self.log.info("Checking out %s tag ref %s",
|
|
project_name, t)
|
|
repo.checkout(t)
|
|
elif zuul_branch and zuul_branch in branches:
|
|
self.log.info("Checking out %s zuul branch %s",
|
|
project_name, zuul_branch)
|
|
repo.checkoutLocalBranch(zuul_branch)
|
|
elif project_default_branch in branches:
|
|
self.log.info("Checking out %s project default branch %s",
|
|
project_name, project_default_branch)
|
|
repo.checkoutLocalBranch(project_default_branch)
|
|
else:
|
|
raise ExecutorError("Project %s does not have the "
|
|
"default branch %s" %
|
|
(project_name, project_default_branch))
|
|
|
|
def runPlaybooks(self, args):
|
|
result = None
|
|
|
|
# Run the Ansible 'setup' module on all hosts in the inventory
|
|
# at the start of the job with a 60 second timeout. If we
|
|
# aren't able to connect to all the hosts and gather facts
|
|
# within that timeout, there is likely a network problem
|
|
# between here and the hosts in the inventory; return them and
|
|
# reschedule the job.
|
|
setup_status, setup_code = self.runAnsibleSetup(
|
|
self.jobdir.setup_playbook)
|
|
if setup_status != self.RESULT_NORMAL or setup_code != 0:
|
|
return result
|
|
|
|
pre_failed = False
|
|
success = False
|
|
for index, playbook in enumerate(self.jobdir.pre_playbooks):
|
|
# TODOv3(pabelanger): Implement pre-run timeout setting.
|
|
pre_status, pre_code = self.runAnsiblePlaybook(
|
|
playbook, args['timeout'], phase='pre', index=index)
|
|
if pre_status != self.RESULT_NORMAL or pre_code != 0:
|
|
# These should really never fail, so return None and have
|
|
# zuul try again
|
|
pre_failed = True
|
|
break
|
|
|
|
if not pre_failed:
|
|
job_status, job_code = self.runAnsiblePlaybook(
|
|
self.jobdir.playbook, args['timeout'], phase='run')
|
|
if job_status == self.RESULT_ABORTED:
|
|
return 'ABORTED'
|
|
elif job_status == self.RESULT_TIMED_OUT:
|
|
# Set the pre-failure flag so this doesn't get
|
|
# overridden by a post-failure.
|
|
pre_failed = True
|
|
result = 'TIMED_OUT'
|
|
elif job_status == self.RESULT_NORMAL:
|
|
success = (job_code == 0)
|
|
if success:
|
|
result = 'SUCCESS'
|
|
else:
|
|
result = 'FAILURE'
|
|
else:
|
|
# The result of the job is indeterminate. Zuul will
|
|
# run it again.
|
|
return None
|
|
|
|
for index, playbook in enumerate(self.jobdir.post_playbooks):
|
|
# TODOv3(pabelanger): Implement post-run timeout setting.
|
|
post_status, post_code = self.runAnsiblePlaybook(
|
|
playbook, args['timeout'], success, phase='post', index=index)
|
|
if post_status != self.RESULT_NORMAL or post_code != 0:
|
|
# If we encountered a pre-failure, that takes
|
|
# precedence over the post result.
|
|
if not pre_failed:
|
|
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')
|
|
port = node.get('ssh_port', 22)
|
|
host_vars = dict(
|
|
ansible_host=ip,
|
|
ansible_user=self.executor_server.default_username,
|
|
ansible_port=port,
|
|
nodepool=dict(
|
|
label=node.get('label'),
|
|
az=node.get('az'),
|
|
cloud=node.get('cloud'),
|
|
provider=node.get('provider'),
|
|
region=node.get('region'),
|
|
interface_ip=node.get('interface_ip'),
|
|
public_ipv4=node.get('public_ipv4'),
|
|
private_ipv4=node.get('private_ipv4'),
|
|
public_ipv6=node.get('public_ipv6')))
|
|
|
|
host_keys = []
|
|
for key in node.get('host_keys'):
|
|
if port != 22:
|
|
host_keys.append("[%s]:%s %s" % (ip, port, key))
|
|
else:
|
|
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 ExecutorError(
|
|
"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 ExecutorError("Unable to find playbook %s" % path)
|
|
return None
|
|
|
|
def preparePlaybooks(self, args):
|
|
self.writeAnsibleConfig(self.jobdir.setup_playbook)
|
|
|
|
for playbook in args['pre_playbooks']:
|
|
jobdir_playbook = self.jobdir.addPrePlaybook()
|
|
self.preparePlaybook(jobdir_playbook, playbook,
|
|
args, required=True)
|
|
|
|
for playbook in args['playbooks']:
|
|
jobdir_playbook = self.jobdir.addPlaybook()
|
|
self.preparePlaybook(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 ExecutorError("No valid playbook found")
|
|
|
|
for playbook in args['post_playbooks']:
|
|
jobdir_playbook = self.jobdir.addPostPlaybook()
|
|
self.preparePlaybook(jobdir_playbook, playbook,
|
|
args, required=True)
|
|
|
|
def preparePlaybook(self, jobdir_playbook, playbook, args, required):
|
|
self.log.debug("Prepare playbook repo for %s" %
|
|
(playbook['project'],))
|
|
# Check out the playbook repo if needed and set the path to
|
|
# the playbook that should be run.
|
|
source = self.executor_server.connections.getSource(
|
|
playbook['connection'])
|
|
project = source.getProject(playbook['project'])
|
|
jobdir_playbook.trusted = playbook['trusted']
|
|
jobdir_playbook.branch = playbook['branch']
|
|
jobdir_playbook.canonical_name_and_path = os.path.join(
|
|
project.canonical_name, playbook['path'])
|
|
path = None
|
|
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'] == playbook['connection'] and
|
|
i['project'] == playbook['project']):
|
|
# We already have this repo prepared
|
|
path = os.path.join(self.jobdir.src_root,
|
|
project.canonical_hostname,
|
|
project.name,
|
|
playbook['path'])
|
|
break
|
|
if not path:
|
|
# 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.
|
|
path = self.checkoutTrustedProject(project, playbook['branch'])
|
|
path = os.path.join(path, playbook['path'])
|
|
|
|
jobdir_playbook.path = self.findPlaybook(
|
|
path,
|
|
required=required,
|
|
trusted=playbook['trusted'])
|
|
|
|
# If this playbook doesn't exist, don't bother preparing
|
|
# roles.
|
|
if not jobdir_playbook.path:
|
|
return
|
|
|
|
for role in playbook['roles']:
|
|
self.prepareRole(jobdir_playbook, role, args)
|
|
|
|
secrets = playbook['secrets']
|
|
if secrets:
|
|
if 'zuul' in secrets:
|
|
# We block this in configloader, but block it here too to make
|
|
# sure that a job doesn't pass secrets named zuul.
|
|
raise Exception("Defining secrets named 'zuul' is not allowed")
|
|
jobdir_playbook.secrets_content = yaml.safe_dump(
|
|
secrets, default_flow_style=False)
|
|
|
|
self.writeAnsibleConfig(jobdir_playbook)
|
|
|
|
def checkoutTrustedProject(self, project, branch):
|
|
root = self.jobdir.getTrustedProject(project.canonical_name,
|
|
branch)
|
|
if not root:
|
|
root = self.jobdir.addTrustedProject(project.canonical_name,
|
|
branch)
|
|
merger = self.executor_server._getMerger(root, self.log)
|
|
merger.checkoutBranch(project.connection_name, project.name,
|
|
branch)
|
|
|
|
path = os.path.join(root,
|
|
project.canonical_hostname,
|
|
project.name)
|
|
return path
|
|
|
|
def prepareRole(self, jobdir_playbook, role, args):
|
|
if role['type'] == 'zuul':
|
|
root = jobdir_playbook.addRole()
|
|
self.prepareZuulRole(jobdir_playbook, role, args, 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:
|
|
self._blockPluginDirs(d)
|
|
for entry in os.listdir(d):
|
|
entry_path = os.path.join(d, entry)
|
|
if os.path.isdir(entry_path):
|
|
self._blockPluginDirs(entry_path)
|
|
return d
|
|
# It is neither a bare role, nor a collection of roles
|
|
raise RoleNotFoundError("Unable to find role in %s" % (path,))
|
|
|
|
def prepareZuulRole(self, jobdir_playbook, role, args, 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'])
|
|
name = role['target_name']
|
|
path = None
|
|
|
|
if not jobdir_playbook.trusted:
|
|
# This playbook is untrested. Use the already checked out
|
|
# version (from speculative merging) of the role if it
|
|
# exists.
|
|
|
|
for i in args['items']:
|
|
if (i['connection'] == role['connection'] and
|
|
i['project'] == role['project']):
|
|
# We already have this repo prepared; use it.
|
|
path = os.path.join(self.jobdir.src_root,
|
|
project.canonical_hostname,
|
|
project.name)
|
|
break
|
|
|
|
if not path:
|
|
# This is a trusted playbook or the role did not appear
|
|
# in the dependency chain for the change (in which case,
|
|
# there is no existing untrusted checkout of it). Check
|
|
# out the branch tip into a dedicated space.
|
|
path = self.checkoutTrustedProject(project, 'master')
|
|
|
|
# The name of the symlink is the requested name of the role
|
|
# (which may be the repo name or may be something else; this
|
|
# can come into play if this is a bare role).
|
|
link = os.path.join(root, name)
|
|
link = os.path.realpath(link)
|
|
if not link.startswith(os.path.realpath(root)):
|
|
raise ExecutorError("Invalid role name %s", name)
|
|
os.symlink(path, link)
|
|
|
|
try:
|
|
role_path = self.findRole(link, trusted=jobdir_playbook.trusted)
|
|
except RoleNotFoundError:
|
|
if role['implicit']:
|
|
self.log.info("Implicit role not found in %s", link)
|
|
return
|
|
raise
|
|
if role_path is None:
|
|
# In the case of a bare role, add the containing directory
|
|
role_path = root
|
|
self.log.debug("Adding role path %s", role_path)
|
|
jobdir_playbook.roles_path.append(role_path)
|
|
|
|
def prepareAnsibleFiles(self, args):
|
|
all_vars = args['vars'].copy()
|
|
# TODO(mordred) Hack to work around running things with python3
|
|
all_vars['ansible_python_interpreter'] = '/usr/bin/python2'
|
|
if 'zuul' in all_vars:
|
|
# We block this in configloader, but block it here too to make
|
|
# sure that a job doesn't pass variables named zuul.
|
|
raise Exception("Defining vars named 'zuul' is not allowed")
|
|
all_vars['zuul'] = args['zuul'].copy()
|
|
all_vars['zuul']['executor'] = dict(
|
|
hostname=self.executor_server.hostname,
|
|
src_root=self.jobdir.src_root,
|
|
log_root=self.jobdir.log_root,
|
|
work_root=self.jobdir.work_root,
|
|
result_data_file=self.jobdir.result_data_file)
|
|
|
|
nodes = self.getHostList(args)
|
|
inventory = make_inventory_dict(nodes, args['groups'], all_vars)
|
|
|
|
with open(self.jobdir.inventory, 'w') as inventory_yaml:
|
|
inventory_yaml.write(
|
|
yaml.safe_dump(inventory, default_flow_style=False))
|
|
|
|
with open(self.jobdir.known_hosts, 'w') as known_hosts:
|
|
for node in nodes:
|
|
for key in node['host_keys']:
|
|
known_hosts.write('%s\n' % key)
|
|
|
|
def writeLoggingConfig(self):
|
|
self.log.debug("Writing logging config for job %s %s",
|
|
self.jobdir.job_output_file,
|
|
self.jobdir.logging_json)
|
|
logging_config = zuul.ansible.logconfig.JobLoggingConfig(
|
|
job_output_file=self.jobdir.job_output_file)
|
|
logging_config.writeJson(self.jobdir.logging_json)
|
|
|
|
def writeAnsibleConfig(self, jobdir_playbook):
|
|
trusted = jobdir_playbook.trusted
|
|
|
|
# TODO(mordred) This should likely be extracted into a more generalized
|
|
# mechanism for deployers being able to add callback
|
|
# plugins.
|
|
if ara_callbacks:
|
|
callback_path = '%s:%s' % (
|
|
self.executor_server.callback_dir,
|
|
os.path.dirname(ara_callbacks.__file__))
|
|
else:
|
|
callback_path = self.executor_server.callback_dir
|
|
with open(jobdir_playbook.ansible_config, 'w') as config:
|
|
config.write('[defaults]\n')
|
|
config.write('hostfile = %s\n' % self.jobdir.inventory)
|
|
config.write('local_tmp = %s/local_tmp\n' %
|
|
self.jobdir.ansible_cache_root)
|
|
config.write('retry_files_enabled = False\n')
|
|
config.write('gathering = smart\n')
|
|
config.write('fact_caching = jsonfile\n')
|
|
config.write('fact_caching_connection = %s\n' %
|
|
self.jobdir.fact_cache)
|
|
config.write('library = %s\n'
|
|
% self.executor_server.library_dir)
|
|
config.write('command_warnings = False\n')
|
|
config.write('callback_plugins = %s\n' % callback_path)
|
|
config.write('stdout_callback = zuul_stream\n')
|
|
config.write('filter_plugins = %s\n'
|
|
% self.executor_server.filter_dir)
|
|
# 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)
|
|
|
|
if jobdir_playbook.roles_path:
|
|
config.write('roles_path = %s\n' % ':'.join(
|
|
jobdir_playbook.roles_path))
|
|
|
|
# On playbooks with secrets we want to prevent the
|
|
# printing of args since they may be passed to a task or a
|
|
# role. Otherwise, printing the args could be useful for
|
|
# debugging.
|
|
config.write('display_args_to_stdout = %s\n' %
|
|
str(not jobdir_playbook.secrets_content))
|
|
|
|
# Increase the internal poll interval of ansible.
|
|
# The default interval of 0.001s is optimized for interactive
|
|
# ui at the expense of CPU load. As we have a non-interactive
|
|
# automation use case a longer poll interval is more suitable
|
|
# and reduces CPU load of the ansible process.
|
|
config.write('internal_poll_interval = 0.01\n')
|
|
|
|
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, playbook):
|
|
config_file = playbook.ansible_config
|
|
env_copy = os.environ.copy()
|
|
env_copy.update(self.ssh_agent.env)
|
|
if ara_callbacks:
|
|
env_copy['ARA_LOG_CONFIG'] = self.jobdir.logging_json
|
|
env_copy['ZUUL_JOB_LOG_CONFIG'] = self.jobdir.logging_json
|
|
env_copy['ZUUL_JOBDIR'] = self.jobdir.root
|
|
pythonpath = env_copy.get('PYTHONPATH')
|
|
if pythonpath:
|
|
pythonpath = [pythonpath]
|
|
else:
|
|
pythonpath = []
|
|
pythonpath = [self.executor_server.ansible_dir] + pythonpath
|
|
env_copy['PYTHONPATH'] = os.path.pathsep.join(pythonpath)
|
|
|
|
if playbook.trusted:
|
|
opt_prefix = 'trusted'
|
|
else:
|
|
opt_prefix = 'untrusted'
|
|
ro_paths = get_default(self.executor_server.config, 'executor',
|
|
'%s_ro_paths' % opt_prefix)
|
|
rw_paths = get_default(self.executor_server.config, 'executor',
|
|
'%s_rw_paths' % opt_prefix)
|
|
ro_paths = ro_paths.split(":") if ro_paths else []
|
|
rw_paths = rw_paths.split(":") if rw_paths else []
|
|
|
|
ro_paths.append(self.executor_server.ansible_dir)
|
|
ro_paths.append(self.jobdir.ansible_root)
|
|
ro_paths.append(self.jobdir.trusted_root)
|
|
ro_paths.append(playbook.root)
|
|
|
|
rw_paths.append(self.jobdir.ansible_cache_root)
|
|
|
|
if self.executor_variables_file:
|
|
ro_paths.append(self.executor_variables_file)
|
|
|
|
secrets = {}
|
|
if playbook.secrets_content:
|
|
secrets[playbook.secrets] = playbook.secrets_content
|
|
|
|
context = self.executor_server.execution_wrapper.getExecutionContext(
|
|
ro_paths, rw_paths, secrets)
|
|
|
|
popen = context.getPopen(
|
|
work_dir=self.jobdir.work_root,
|
|
ssh_auth_sock=env_copy.get('SSH_AUTH_SOCK'))
|
|
|
|
env_copy['ANSIBLE_CONFIG'] = config_file
|
|
# NOTE(pabelanger): Default HOME variable to jobdir.work_root, as it is
|
|
# possible we don't bind mount current zuul user home directory.
|
|
env_copy['HOME'] = self.jobdir.work_root
|
|
|
|
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 = popen(
|
|
cmd,
|
|
cwd=self.jobdir.work_root,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
preexec_fn=os.setsid,
|
|
env=env_copy,
|
|
)
|
|
|
|
syntax_buffer = []
|
|
ret = None
|
|
if timeout:
|
|
watchdog = Watchdog(timeout, self._ansibleTimeout,
|
|
("Ansible timeout exceeded",))
|
|
watchdog.start()
|
|
try:
|
|
# Use manual idx instead of enumerate so that RESULT lines
|
|
# don't count towards BUFFER_LINES_FOR_SYNTAX
|
|
idx = 0
|
|
for line in iter(self.proc.stdout.readline, b''):
|
|
if line.startswith(b'RESULT'):
|
|
# TODO(mordred) Process result commands if sent
|
|
continue
|
|
else:
|
|
idx += 1
|
|
if idx < BUFFER_LINES_FOR_SYNTAX:
|
|
syntax_buffer.append(line)
|
|
line = line[:1024].rstrip()
|
|
self.log.debug("Ansible output: %s" % (line,))
|
|
self.log.debug("Ansible output terminated")
|
|
ret = self.proc.wait()
|
|
self.log.debug("Ansible exit code: %s" % (ret,))
|
|
finally:
|
|
if timeout:
|
|
watchdog.stop()
|
|
self.log.debug("Stopped watchdog")
|
|
self.log.debug("Stopped disk job killer")
|
|
|
|
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)
|
|
elif ret == 4:
|
|
# Ansible could not parse the yaml.
|
|
self.log.debug("Ansible parse error")
|
|
# TODO(mordred) If/when we rework use of logger in ansible-playbook
|
|
# we'll want to change how this works to use that as well. For now,
|
|
# this is what we need to do.
|
|
# TODO(mordred) We probably want to put this into the json output
|
|
# as well.
|
|
with open(self.jobdir.job_output_file, 'a') as job_output:
|
|
job_output.write("{now} | ANSIBLE PARSE ERROR\n".format(
|
|
now=datetime.datetime.now()))
|
|
for line in syntax_buffer:
|
|
job_output.write("{now} | {line}\n".format(
|
|
now=datetime.datetime.now(),
|
|
line=line.decode('utf-8').rstrip()))
|
|
elif ret == 250:
|
|
# Unexpected error from ansible
|
|
with open(self.jobdir.job_output_file, 'a') as job_output:
|
|
job_output.write("{now} | UNEXPECTED ANSIBLE ERROR\n".format(
|
|
now=datetime.datetime.now()))
|
|
found_marker = False
|
|
for line in syntax_buffer:
|
|
if line.startswith('ERROR! Unexpected Exception'):
|
|
found_marker = True
|
|
if not found_marker:
|
|
continue
|
|
job_output.write("{now} | {line}\n".format(
|
|
now=datetime.datetime.now(),
|
|
line=line.decode('utf-8').rstrip()))
|
|
|
|
return (self.RESULT_NORMAL, ret)
|
|
|
|
def runAnsibleSetup(self, playbook):
|
|
if self.executor_server.verbose:
|
|
verbose = '-vvv'
|
|
else:
|
|
verbose = '-v'
|
|
|
|
cmd = ['ansible', '*', verbose, '-m', 'setup',
|
|
'-a', 'gather_subset=!all']
|
|
|
|
result, code = self.runAnsible(
|
|
cmd=cmd, timeout=60, playbook=playbook)
|
|
self.log.debug("Ansible complete, result %s code %s" % (
|
|
self.RESULT_MAP[result], code))
|
|
return result, code
|
|
|
|
def emitPlaybookBanner(self, playbook, step, phase, result=None):
|
|
# This is used to print a header and a footer, respectively at the
|
|
# beginning and the end of each playbook execution.
|
|
# We are doing it from the executor rather than from a callback because
|
|
# the parameters are not made available to the callback until it's too
|
|
# late.
|
|
phase = phase or ''
|
|
trusted = playbook.trusted
|
|
trusted = 'trusted' if trusted else 'untrusted'
|
|
branch = playbook.branch
|
|
playbook = playbook.canonical_name_and_path
|
|
|
|
if phase and phase != 'run':
|
|
phase = '{phase}-run'.format(phase=phase)
|
|
phase = phase.upper()
|
|
|
|
if result is not None:
|
|
result = self.RESULT_MAP[result]
|
|
msg = "{phase} {step} {result}: [{trusted} : {playbook}@{branch}]"
|
|
msg = msg.format(phase=phase, step=step, result=result,
|
|
trusted=trusted, playbook=playbook, branch=branch)
|
|
else:
|
|
msg = "{phase} {step}: [{trusted} : {playbook}@{branch}]"
|
|
msg = msg.format(phase=phase, step=step, trusted=trusted,
|
|
playbook=playbook, branch=branch)
|
|
|
|
with open(self.jobdir.job_output_file, 'a') as job_output:
|
|
job_output.write("{now} | {msg}\n".format(
|
|
now=datetime.datetime.now(),
|
|
msg=msg))
|
|
|
|
def runAnsiblePlaybook(self, playbook, timeout, success=None,
|
|
phase=None, index=None):
|
|
if self.executor_server.verbose:
|
|
verbose = '-vvv'
|
|
else:
|
|
verbose = '-v'
|
|
|
|
cmd = ['ansible-playbook', verbose, playbook.path]
|
|
if playbook.secrets_content:
|
|
cmd.extend(['-e', '@' + playbook.secrets])
|
|
|
|
if success is not None:
|
|
cmd.extend(['-e', 'zuul_success=%s' % str(bool(success))])
|
|
|
|
if phase:
|
|
cmd.extend(['-e', 'zuul_execution_phase=%s' % phase])
|
|
|
|
if index is not None:
|
|
cmd.extend(['-e', 'zuul_execution_phase_index=%s' % index])
|
|
|
|
cmd.extend(['-e', 'zuul_execution_trusted=%s' % str(playbook.trusted)])
|
|
cmd.extend([
|
|
'-e',
|
|
'zuul_execution_canonical_name_and_path=%s'
|
|
% playbook.canonical_name_and_path])
|
|
cmd.extend(['-e', 'zuul_execution_branch=%s' % str(playbook.branch)])
|
|
|
|
if self.executor_variables_file is not None:
|
|
cmd.extend(['-e@%s' % self.executor_variables_file])
|
|
|
|
self.emitPlaybookBanner(playbook, 'START', phase)
|
|
|
|
result, code = self.runAnsible(
|
|
cmd=cmd, timeout=timeout, playbook=playbook)
|
|
self.log.debug("Ansible complete, result %s code %s" % (
|
|
self.RESULT_MAP[result], code))
|
|
|
|
self.emitPlaybookBanner(playbook, 'END', phase, result=result)
|
|
return result, code
|