From 2b720ca61cb8c72ef527f58c4e8fd6939ce7225a Mon Sep 17 00:00:00 2001 From: Martin Krizek Date: May 04 2017 16:25:47 +0000 Subject: [PATCH 1/2] *WIP* Hacky support for Ansible Tasks This patch removes libtaskotron formula format for writing tasks and introduces support for running tasks in form of an ansible playbook. The following parts are removed: * task formula handling code * remote execution (Paramiko Wrapper) * concept of overlord and minion * the exitcode directive (haven't been really used) All of this is subject to change and needs a lot of polishing. There are still parts of codebase that will be removed or re-written. The code expects 'force_imageurl: True" and "imageurl" filled in taskotron.yaml and image to have python2 and whatever ansible requires to run. When running locally, user is asked for sudo password since tasks are ran as root. To test run: $ git clone https://github.com/stefwalter/gzip-dist-git $ runtask -d -i gzip -t koji_build -a x86_64 gzip-dist-git/tests/test_rpm.yml $ runtask -d -i gzip -t koji_build -a x86_64 --ssh $IP_ADDRESS --ssh-privkey conf/id_taskotron gzip-dist-git/tests/test_rpm.yml $ runtask -d -i gzip -t koji_build -a x86_64 --libvirt --ssh-privkey conf/id_taskotron gzip-dist-git/tests/test_rpm.yml https://fedoraproject.org/wiki/Changes/InvokingTestsAnsible --- diff --git a/libtaskotron.spec b/libtaskotron.spec index a02a233..3d7729f 100644 --- a/libtaskotron.spec +++ b/libtaskotron.spec @@ -84,9 +84,7 @@ Module for libtaskotron which contains all functionality which is fedora-only Summary: disposable client module for libtaskotron Requires: libtaskotron-core = %{version}-%{release} -Requires: python-paramiko >= 1.15.1 Requires: testcloud >= 0.1.10 -BuildRequires: python-paramiko >= 1.15.1 BuildRequires: testcloud >= 0.1.10 %description -n libtaskotron-disposable diff --git a/libtaskotron/executor.py b/libtaskotron/executor.py index 7087fe0..81ff2e6 100644 --- a/libtaskotron/executor.py +++ b/libtaskotron/executor.py @@ -4,211 +4,111 @@ # See the LICENSE file for more details on Licensing from __future__ import absolute_import -import tempfile + +import os import os.path -import imp -import copy -import collections -import pipes - -from libtaskotron import taskformula -from libtaskotron import logger -from libtaskotron import python_utils +import subprocess + from libtaskotron import config from libtaskotron.logger import log -import libtaskotron.exceptions as exc -from libtaskotron.directives import exitcode_directive +from libtaskotron import file_utils +from libtaskotron import os_utils +from libtaskotron import exceptions as exc -try: - from libtaskotron.ext.fedora import rpm_utils -except ImportError, e: - raise exc.TaskotronImportError(e) +from libtaskotron.ext.disposable import vm +# TODO exc class Executor(object): - '''This class serves the purpose of actual task execution. It is instantiated by - :class:`.Overlord`, :class:`.PersistentMinion` or :class:`.DisposableMinion`. - - :cvar str default_namespace: namespace to be used when the formula doesn't contain a namespace - definition - :ivar dict formula: parsed task formula - :ivar dict arg_data: processed cli arguments with some extra runtime variables - :ivar str workdir: path to working directory; a temporary one is created during execution if - ``None`` - :ivar dict working_data: variables exported from formula directives, to be injected when - referenced using ``${variable}`` - :ivar int exitcode: exit code set by ``exitcode`` directive; it is ``None`` if ``exitcode`` - directive was not used - ''' - - default_namespace = 'scratch' - - def __init__(self, formula, arg_data, workdir=None): - self.formula = formula + def __init__(self, arg_data): self.arg_data = arg_data - self.working_data = {} - self.directives = {} - self.workdir = workdir - self.exitcode = None - - def execute(self): - '''Execute the task (consists of preparing the task and running it).''' - - filelog_path = os.path.join(self.arg_data['artifactsdir'], 'taskotron.log') - logger.add_filehandler(filelog_path=filelog_path, remove_mem_handler=True) - - self._prepare_task() - self._run() - - def _prepare_task(self): - '''Prepare the environment for running a task. If there are packages, requirements or - groups specified in the task formula, this method tries to install them (in case of - production profile) or checks whether they are installed (in case of developer profile). - ''' - rpms = self.formula.get('environment', {}).get('rpm', []) - profile = config.get_config().profile - - if not rpm_utils.is_installed(rpms): - if profile == config.ProfileName.PRODUCTION or self.arg_data['local']: - rpm_utils.install(rpms) - else: - raise exc.TaskotronError('Some packages are not installed. Please run ' - '"dnf install %s" to install all required packages.' - % " ".join([pipes.quote(rpm) for rpm in rpms])) - - def _run(self): - self._validate_input() - - if not self.workdir: # create temporary workdir if needed - self.workdir = tempfile.mkdtemp(prefix="task-", - dir=config.get_config().tmpdir) - log.debug("Current workdir: %s", self.workdir) - self.arg_data['workdir'] = self.workdir - self.arg_data['checkname'] = self.formula['name'] - self.arg_data['namespace'] = self.formula.get('namespace') or self.default_namespace - - # override variable values - for var, val in self.arg_data['override'].items(): - log.debug("Overriding variable %s, new value: %s", var, val) - self.arg_data[var] = eval(val, {}, {}) - - self._do_actions() - - def _validate_input(self): - if 'input' not in self.formula: - return - - if not isinstance(self.formula['input'], collections.Mapping): - raise exc.TaskotronYamlError("Input yaml should contain correct 'input'" - "section (a mapping). Yours was: %s" % - type(self.formula['input'])) - - required_args = self.formula['input'].get('args', None) - - if not python_utils.iterable(required_args): - raise exc.TaskotronYamlError("Input yaml should contain correct 'args' " - "section (an iterable). Yours was: %s" % - type(required_args)) - - for arg in required_args: - if arg not in self.arg_data: - raise exc.TaskotronYamlError("Required input arg '%s' " - "was not defined" % arg) - - def _do_actions(self): - '''Sequentially run all actions for a task. An 'action' is a single step - under the ``actions:`` key. An example action looks like:: - - - name: download rpms from koji - koji: - action: download - koji_build: $koji_build - arch: $arch - ''' - if 'actions' not in self.formula or not self.formula['actions']: - raise exc.TaskotronYamlError("At least one task should be specified" - " in input formula") - - for action in self.formula['actions']: - self._do_single_action(action) - - def _do_single_action(self, action): - '''Execute a single action from the task. See :meth:`do_actions` to see - how an action looks like. - - :param dict action: An action specification parsed from the task formula - ''' - directive_name = self._extract_directive_from_action(action) - - rendered_action = self._render_action(action) - - log.debug("Executing directive: %s", directive_name) - self._load_directive(directive_name) - - directive_object = self.directives[directive_name] - directive_classname = directive_object.directive_class - - directive_callable = getattr(directive_object, directive_classname)() - - output = directive_callable.process(rendered_action[directive_name], - self.arg_data) - - if 'export' in action: - self.working_data[action['export']] = output - log.debug("Variable ${%s} was exported with value:\n%s" % - (action['export'], output)) - - # unable to use isinstance() due to dynamically loaded directives - if directive_classname == exitcode_directive.directive_class: - # set exitcode if was None or overwrite previous SUCCESS with FAILURE (not vice versa) - if not self.exitcode or output != exitcode_directive.SUCCESS: - log.info("Setting exitcode to %s" % output) - self.exitcode = output - - def _extract_directive_from_action(self, action): - for key in action: - if key not in ['name', 'export']: - return key - raise exc.TaskotronYamlError('no directive found in action %s' % str(action)) - - def _render_action(self, action): - '''Take an action and replace all included variables with actual values - from :attr:`arg_data` and :attr:`working_data`. See :meth:`do_actions` - to see how an action looks like. - - :param dict action: An action specification parsed from the task formula - :return: a rendered action - :rtype: dict - ''' - # copy the input so that we don't disrupt what we're processing - rendered_action = copy.deepcopy(action) - - variables = copy.deepcopy(self.arg_data) - variables.update(self.working_data) - taskformula.replace_vars_in_action(rendered_action, variables) - - return rendered_action - - def _load_directive(self, directive_name, directive_dir=None): - # look in default path if nothing is specified - if not directive_dir: - directive_dir = os.path.join(os.path.dirname(__file__), - 'directives') - - real_name = "%s_directive" % directive_name - directive_file = os.path.join(directive_dir, '%s.py' % real_name) - - if not os.path.exists(directive_file): - raise exc.TaskotronDirectiveError("Directive %s not found in directory %s" % - (directive_name, directive_dir)) + self.task_vm = None + self.run_remotely = False - try: - loaded_directive = imp.load_source(real_name, directive_file) - except ImportError, e: - raise exc.TaskotronImportError(e) + def _spawn_vm(self): + self.task_vm = vm.TestCloudMachine(self.arg_data['uuid']) + self.task_vm.prepare() + + log.info('Running task over SSH on machine %s', self.task_vm.instancename) + + self.task_vm.wait_for_port(22) + + return self.task_vm.ipaddr + + def _prepare(self): + # when running remotely, run directly over ssh, instead of using libvirt + persistent = False - self.directives[directive_name] = loaded_directive + runtask_mode = config.get_config().runtask_mode + if runtask_mode == config.RuntaskModeName.LOCAL: + self.run_remotely = False + elif runtask_mode == config.RuntaskModeName.LIBVIRT: + self.run_remotely = True + else: + assert False, 'This should never occur' - def _validate_env(self): - # TODO: implement this - raise NotImplementedError("Environment validation is not yet implemented") + if self.arg_data['local']: + log.debug("Forcing local execution (option --local)") + self.run_remotely = False + + elif self.arg_data['libvirt']: + log.debug("Forcing remote execution (option --libvirt)") + self.run_remotely = True + persistent = False + + elif self.arg_data['ssh']: + log.debug('Forcing remote execution (option --ssh)') + self.run_remotely = True + persistent = True + + log.debug('Execution mode: %s', 'remote' if self.run_remotely else 'local') + + ipaddr = '127.0.0.1' + if self.run_remotely: + if persistent: + ipaddr = self.arg_data['machine'] + else: + ipaddr = self._spawn_vm() + + with open(os.path.join('inventory'), 'w') as f: + f.write('[localhost]\n%s' % ipaddr) + + def _execute(self): + cmd = [ + 'ansible-playbook', self.arg_data['task'], + '-b', + '-u root', + '-i', 'inventory', + '-e', 'artifacts=%s' % self.arg_data['artifactsdir'], + '-e', 'subjects=%s' % self.arg_data['item'], + ] + + if self.run_remotely: + cmd.extend(['--private-key', self.arg_data['ssh_privkey']]) + else: + cmd.extend(['-K', '-c', 'local']) + + log.debug('Running ansible playbook %s', ' '.join(cmd)) + try: + output, _ = os_utils.popen_rt(cmd, stderr=subprocess.STDOUT) + with open(os.path.join(self.arg_data['artifactsdir'], 'ansible.log'), 'w') as f: + f.write(output) + except subprocess.CalledProcessError, e: + log.error('ansible-playbook ended with %d return code', e.returncode) + log.debug(e.output) + + def start(self): + try: + file_utils.makedirs(self.arg_data['artifactsdir']) + log.info("Task artifacts will be saved in: %s", self.arg_data['artifactsdir']) + except OSError: + log.error("Can't create artifacts directory %s", self.arg_data['artifactsdir']) + raise + + self._prepare() + self._execute() + + # clean up + if self.task_vm is not None: + self.task_vm.teardown() + os.remove('inventory') diff --git a/libtaskotron/main.py b/libtaskotron/main.py index de983c7..1ceb346 100644 --- a/libtaskotron/main.py +++ b/libtaskotron/main.py @@ -8,7 +8,6 @@ import logging import os.path import argparse import datetime -import sys import copy import libtaskotron @@ -16,7 +15,7 @@ from libtaskotron import logger from libtaskotron import config from libtaskotron import check from libtaskotron.logger import log -from libtaskotron.overlord import Overlord +from libtaskotron.executor import Executor def get_argparser(): @@ -24,11 +23,9 @@ def get_argparser(): :rtype: :class:`argparse.ArgumentParser` ''' - # IMPORTANT: When you add a new argument, look into `minion.BaseMinion.arg_data_exclude` and - # update it if needed! parser = argparse.ArgumentParser() - parser.add_argument("task", help="task formula to run") + parser.add_argument("task", help="task playbook to run") parser.add_argument("-a", "--arch", choices=["i386", "x86_64", "armhfp", "noarch"], default='noarch', help="architecture specifying the item to be checked. 'noarch' value " @@ -47,13 +44,6 @@ def get_argparser(): help="Unique job identifier for the execution" "status tracking purposes. If unset, defaults to" "current datetime in UTC") - parser.add_argument("--override", action="append", default=[], metavar='VAR=VALUE', - help="override internal variable values used in runner " - "and the task formula. Value itself is evaluated " - "by eval(). This option can be used multiple times. " - "Example: --override \"workdir='/some/dir/'\"") - parser.add_argument("-p", "--patch", help="patch to apply to remote system" - "before execution of task") parser.add_argument("--local", action="store_true", help="make the task run locally on this very machine (the default " "behavior for development profile). This also approves any required " @@ -107,12 +97,6 @@ def process_args(raw_args): if args['type'] in check.ReportType.list(): args[args['type']] = args['item'] - override = {} - for var in args['override']: - name, value = var.split('=', 1) - override[name] = value - args['override'] = override - # parse ssh if args['ssh']: args['user'], machine = args['ssh'].split('@') @@ -152,12 +136,10 @@ def main(): logger.init(level_stream=level_stream) # start execution - overlord = Overlord(arg_data) - overlord.start() + executor = Executor(arg_data) + executor.start() # finalize log.info('Execution finished at: %s. Task artifacts were saved in: %s', datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC'), arg_data['artifactsdir']) - log.info("Exiting with exit status %d" % overlord.exitcode) - sys.exit(overlord.exitcode) diff --git a/libtaskotron/minion.py b/libtaskotron/minion.py deleted file mode 100644 index 057b1b2..0000000 --- a/libtaskotron/minion.py +++ /dev/null @@ -1,257 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2009-2015, Red Hat, Inc. -# License: GPL-2.0+ -# See the LICENSE file for more details on Licensing - -from __future__ import absolute_import -import os.path -import yaml -import pipes - -from libtaskotron import logger -from libtaskotron import config -from libtaskotron import remote_exec -from libtaskotron import python_utils -from libtaskotron.logger import log -from libtaskotron import taskformula -import libtaskotron.exceptions as exc - -try: - from libtaskotron.ext.disposable import vm -except ImportError, e: - raise exc.TaskotronImportError(e) - - -class BaseMinion(object): - '''Base Minion class that shouldn't be used on its own, it solely initiates inner attributes - and environment. It also provides method _run() that delegates the task execution to Executor - over SSH. - - :cvar tuple arg_data_exclude: a tuple of cmdline options/arguments which should not be - forwarded from the host to the minion when execution the task. Use the same names as the - stored variables in ``arg_data`` use (i.e. dashes converted to underscores, etc). - :ivar dict formula: parsed task formula - :ivar dict arg_data: processed cli arguments with some extra runtime variables - :ivar str artifactsdir: path to ``artifactsdir`` for storing logs - :ivar int exitcode: exit code of libtaskotron runner executed on the minion - :ivar ssh: an instance of :class:`.ParamikoWrapper` which is used to execute all remote - commands on a minion. ``None`` by default, you need to set this before you start - :meth:`execute`. - ''' - - arg_data_exclude = ('task', 'libvirt', 'ssh', 'ssh_privkey', 'no_destroy', 'patch') - - def __init__(self, formula, arg_data): - self.formula = formula - self.arg_data = arg_data - self.artifactsdir = self.arg_data['artifactsdir'] - - self.user = arg_data.get('user', None) - self.machine = arg_data.get('machine', None) - self.port = arg_data.get('port', None) - self.ssh_privkey = arg_data['ssh_privkey'] or config.get_config().ssh_privkey - - self.exitcode = None - self.ssh = None - - def _prepare_task(self): - '''Prepare the environment to run the task.''' - # TODO: add sudo support - - log.info('Examining the minion and installing libtaskotron...') - self.taskdir = config.get_config().client_taskdir - - # create initial DNF cache (but only if needed) - # uses the same logic as in libtaskotron.ext.fedora.rpm_utils.dnf_cache_available() - self.ssh.cmd('dnf --cacheonly repolist || dnf makecache') - - # add additional repos to minion - repos = ['--add-repo %s' % pipes.quote(repo) for repo in config.get_config().minion_repos] - if repos: - self.ssh.cmd('dnf config-manager %s' % ' '.join(repos)) - - # create DNF cache for taskotron repos - # this hard-coding is very suboptimal, but it's hard to find a better way how to refresh - # only taskotron repos (unless we require their names listed in configuration) - excluded = ['fedora', 'updates', 'updates-testing'] - exclude_cmd = ' '.join(['--disablerepo %s' % pipes.quote(repo) for repo in excluded]) - self.ssh.cmd('dnf makecache %s' % exclude_cmd) - - # install libtaskotron - # FIXME this should only need libtaskotron-core T651. libtaskotron-fedora should be - # installed as a check dep. - # TODO: remove tar dep once libtaskotron 0.4.16 has been released for long enough - # (tar is here just for backward compatibility, see D965) - self.ssh.install_pkgs(['libtaskotron-core', 'libtaskotron-fedora', 'tar']) - - log.info('Copying task files onto the minion...') - - # Copy config files. Go through CONF_DIRS in reversed order to preserve priority. - for config_dir in reversed(config.CONF_DIRS): - if os.path.exists(config_dir): - self.ssh.put_dir(config_dir, '/etc/taskotron/') - - # create taskdir, remove the old one first - self.ssh.cmd('rm -rf {0} && mkdir -p {0}'.format(pipes.quote(self.taskdir))) - - # patch remote libtaskotron if needed (for dev, shouldn't be used in production) - if self.arg_data['patch'] is not None: - self.ssh.put_file(os.path.abspath(self.arg_data['patch']), '%s/%s' % ( - self.taskdir, self.arg_data['patch'])) - self.ssh.install_pkgs(['patch']) - self.ssh.cmd('patch -d /usr/lib/python2.7/site-packages/ -p1 -i %s/%s' % ( - pipes.quote(self.taskdir), pipes.quote(self.arg_data['patch']))) - - # put files needed for execution (task, input files, etc) - self.ssh.put_dir(os.path.dirname(os.path.abspath(self.arg_data['task'])), self.taskdir) - - # need to have default_flow_style false to get valid yaml w/ nested dicts - self.ssh.write_file(os.path.join(self.taskdir, - os.path.basename(self.arg_data['task'])), - yaml.dump(self.formula, default_flow_style=False)) - - def _run(self): - '''The main method that runs the task''' - - task = os.path.basename(self.arg_data['task']) - cmdline = python_utils.reverse_argparse(self.arg_data['_orig_args'], - ignore=self.arg_data_exclude) - cmdline.append('--local') - cmdline.append(task) - - # be paranoid and escape everything - cmdline = [pipes.quote(elem) for elem in cmdline] - - # execute task - log.info('Executing the task on the minion...') - task_cmd = ['cd', pipes.quote(self.taskdir), '&&', 'runtask'] + cmdline - - self.exitcode = self.ssh.cmd(' '.join(task_cmd)) - log.info('Task execution on the minion is complete') - - def _get_output(self): - '''copy artifacts from vm''' - log.info('Copying task artifacts and logs from the minion...') - try: - self.ssh.get_dir(self.artifactsdir, self.artifactsdir) - except exc.TaskotronRemoteError, e: - log.error(e) - - def execute(self): - '''This method has to be implemented by classes that inherits from :class:`.BaseMinion`. - It should contain environment initialization (namely set :attr:`ssh` to properly - instantiated :class:`.ParamikoWrapper`) and cleanup (if necessary). Moreover, the order of - private methods has to be: :meth:`_prepare_task`, :meth:`_run` and :meth:`_get_output`. - ''' - raise exc.TaskotronNotImplementedError() - - -class PersistentMinion(BaseMinion): - '''Minion class that connects to alredy running machine using SSH and runs the task there. - ''' - - def execute(self): - '''Init environment and connect to remote machine for task execution.''' - - filelog_path = os.path.join(self.artifactsdir, 'taskotron-overlord.log') - logger.add_filehandler(filelog_path=filelog_path, remove_mem_handler=True) - - stdio_filename = os.path.join(self.artifactsdir, 'taskotron-stdio.log') - - log.info("Running task over SSH on machine '%s'...", self.machine) - - with remote_exec.ParamikoWrapper(self.machine, self.port, self.user, - self.ssh_privkey, stdio_filename) as self.ssh: - try: - self._prepare_task() - self._run() - except exc.TaskotronRemoteTimeoutError: - # do not try to get logs - the minion is likely dead and we would hang on - # getting the logs, see https://phab.qa.fedoraproject.org/T665 - log.error('Connection to the minion timed out') - raise - except exc.TaskotronRemoteError: - # this is the case when remote execution failed but it should be possible to - # get logs from the minion, so try it - log.error('Error occurred on the minion, task execution stopped') - self._get_output() - raise - else: - self._get_output() - - -class DisposableMinion(BaseMinion): - '''Minion class that creates a disposable client, connects to it over SSH and runs the task - there. - ''' - - def execute(self): - '''Init environment and connect to disposable client for task execution.''' - - filelog_path = os.path.join(self.artifactsdir, 'taskotron-overlord.log') - logger.add_filehandler(filelog_path=filelog_path, remove_mem_handler=True) - - stdio_filename = os.path.join(self.artifactsdir, 'taskotron-stdio.log') - - execution_ok = False - try: - log.info("Running task on a disposable client using libvirt...") - - log.info("Creating VM for task") - - env = taskformula.devise_environment(self.formula, self.arg_data) - - task_vm = vm.TestCloudMachine(self.arg_data['uuid']) - task_vm.prepare(**env) - self.machine = task_vm.ipaddr - self.port = 22 - self.user = task_vm.username - - log.info('Running task over SSH on machine %s (%s@%s:%s)', task_vm.instancename, - self.user, self.machine, self.port) - - task_vm.wait_for_port(self.port) - - with remote_exec.ParamikoWrapper(self.machine, self.port, self.user, - self.ssh_privkey, stdio_filename) as self.ssh: - try: - self._prepare_task() - self._run() - except exc.TaskotronRemoteTimeoutError: - # do not try to get logs - the minion is likely dead and we would hang on - # getting the logs, see https://phab.qa.fedoraproject.org/T665 - log.error('Connection to the minion timed out') - raise - except exc.TaskotronRemoteError: - # this is the case when remote execution failed but it should be possible to - # get logs from the minion, so try it - log.error('Error occurred on the minion, task execution stopped') - self._get_output() - raise - else: - self._get_output() - - execution_ok = True - finally: - if not execution_ok: - log.warn('Task execution was interrupted by an error, doing emergency cleanup') - - if self.arg_data['no_destroy']: - log.info('--no-destroy mode was set from command line, not tearing VM ' - 'down for task %s: %s@%s:%s', self.arg_data['uuid'], self.user, - self.machine, self.port) - else: - try: - log.info('Shutting down the minion...') - task_vm.teardown() - # Catch 'em all since we can't be sure that every exception trown - # is TaskotronRemoteError. If we omitted any exception that would - # be later thrown in the teardown, we would suppress (effectively - # make it silent) any possible exception from execution above - # which is more important. - except Exception, e: - if execution_ok: - raise e - else: - # a different exception is already being raised, don't override it - log.exception(e) diff --git a/libtaskotron/overlord.py b/libtaskotron/overlord.py deleted file mode 100644 index d29c0cd..0000000 --- a/libtaskotron/overlord.py +++ /dev/null @@ -1,100 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2009-2015, Red Hat, Inc. -# License: GPL-2.0+ -# See the LICENSE file for more details on Licensing - -from __future__ import absolute_import - -from libtaskotron import config -from libtaskotron.logger import log -from libtaskotron import executor -from libtaskotron import file_utils -from libtaskotron import exceptions as exc -from libtaskotron.directives import exitcode_directive - - -class Overlord(object): - """Overlord class encapsulates decision whether the task is run locally or remotely (in - persistent or disposable client) and orchestrates the execution. - - :ivar dict arg_data: processed cli arguments with some extra runtime variables - :ivar int exitcode: exit code of the task; if ``None``, it means :meth:`start` was not called - yet - :ivar dict formula: parsed task formula - """ - - def __init__(self, arg_data): - self.arg_data = arg_data - self.exitcode = None - - # parse task formula - self.formula = config.parse_yaml_from_file(arg_data['task']) - if not self.formula: - raise exc.TaskotronYamlError('Task formula file should not be empty: %s' % - arg_data['task']) - - log.debug('Found task: %s.%s', - self.formula.get('namespace') or executor.Executor.default_namespace, - self.formula['name']) - - def _get_runner(self): - """Decide the run mode and get appropriate runner (Executor or Minion). - - :returns: either :class:`.Executor` (local run mode) or :class:`.PersistentMinion` or - :class:`.DisposableMinion` (remote run mode) - """ - # run locally or remotely - run_remotely = False - # when running remotely, run directly over ssh, instead of using libvirt - persistent = False - - # decide whether to run locally or remotely - runtask_mode = config.get_config().runtask_mode - if runtask_mode == config.RuntaskModeName.LOCAL: - run_remotely = False - elif runtask_mode == config.RuntaskModeName.LIBVIRT: - run_remotely = True - else: - assert False, 'This should never occur' - - if self.arg_data['local']: - log.debug("Forcing local execution (option --local)") - run_remotely = False - - elif self.arg_data['libvirt']: - log.debug("Forcing remote execution (option --libvirt)") - run_remotely = True - persistent = False - - elif self.arg_data['ssh']: - log.debug('Forcing remote execution (option --ssh)') - run_remotely = True - persistent = True - - log.debug('Execution mode: %s', 'remote' if run_remotely else 'local') - - if run_remotely: - from libtaskotron import minion - if persistent: - return minion.PersistentMinion(self.formula, self.arg_data) - else: - return minion.DisposableMinion(self.formula, self.arg_data) - else: - return executor.Executor(self.formula, self.arg_data) - - def start(self): - """Start the overlord, get runner and execute the task (either locally or remotely). - """ - try: - file_utils.makedirs(self.arg_data['artifactsdir']) - log.info("Task artifacts will be saved in: %s", self.arg_data['artifactsdir']) - except OSError: - log.error("Can't create artifacts directory %s", self.arg_data['artifactsdir']) - raise - - runner = self._get_runner() - runner.execute() - - self.exitcode = 0 - if runner.exitcode == exitcode_directive.FAILURE: - self.exitcode = runner.exitcode diff --git a/libtaskotron/remote_exec.py b/libtaskotron/remote_exec.py deleted file mode 100644 index a14bac8..0000000 --- a/libtaskotron/remote_exec.py +++ /dev/null @@ -1,363 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2009-2015, Red Hat, Inc. -# License: GPL-2.0+ -# See the LICENSE file for more details on Licensing - -'''Tools for remote execution primary for disposable clients''' - -from __future__ import absolute_import -import os -import os.path -import sys -import socket -import stat -import time -import pipes -import tarfile -import tempfile - -import paramiko - -from .logger import log -from . import exceptions as exc - -from libtaskotron.directives import exitcode_directive -from libtaskotron import file_utils - - -class ParamikoWrapper(object): - '''Wrapper for SSH communication using paramiko library''' - - #: timeout for network operations in seconds (900 seconds = 15 minutes) - TIMEOUT = 900 - - def __init__(self, hostname, port, username, key_filename, stdio_filename=None): - self.ssh = None - self.sftp = None - self.hostname = hostname - self.port = port - self.username = username - self.key_filename = key_filename - self.stdio_filename = stdio_filename - self.outstream = file_utils.Tee(sys.stdout) - - def __enter__(self): - self.connect() - return self - - def __exit__(self, type, value, traceback): - self.close() - - def __str__(self): - return '<%s: %s@%s:%s>' % (self.__class__.__name__, self.username, - self.hostname, self.port) - - def connect(self): - '''Connect to a machine over ssh. Open sftp channel and, if applicable, file that - stdout/err from the machine will be saved to. - - :raise TaskotronRemoteError: when the connection does not succeed''' - - self.ssh = paramiko.SSHClient() - # accept unknown hosts - self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - - log.debug('Connecting to remote host: %s@%s:%s', self.username, self.hostname, self.port) - - try: - self.ssh.connect(self.hostname, - port=self.port, - username=self.username, - # we use empty string by default (no special key), but paramiko - # requires receiving None in that case - key_filename=self.key_filename or None) - - self.sftp = self.ssh.open_sftp() - self.sftp.get_channel().settimeout(self.TIMEOUT) - - if self.stdio_filename is not None: - try: - f = open(self.stdio_filename, 'w') - self.outstream.add(f) - except IOError, e: - log.warning('Could not open %s. Falling back to writing vm\'s output ' - 'to stdout only.', self.stdio_filename) - except paramiko.BadHostKeyException, e: - raise exc.TaskotronRemoteError('Server\'s (%s@%s:%s) hostkey could not be verified: %s' - % (self.username, self.hostname, self.port, str(e))) - except paramiko.AuthenticationException, e: - raise exc.TaskotronRemoteError('Authentication to %s@%s:%s failed: %s' % - (self.username, self.hostname, self.port, str(e))) - except (paramiko.SSHException, socket.error), e: - raise exc.TaskotronRemoteError('Could not connect to %s@%s:%s: %s' % - (self.username, self.hostname, self.port, str(e))) - except IOError as e: - # let's hope this does not occur in more situations - raise exc.TaskotronRemoteError("The private key '%s' could not be read: %s" % - (self.key_filename, str(e))) - - def close(self): - '''Close open connections and files.''' - - self.outstream.close() - self.sftp.close() - self.ssh.close() - - def cmd(self, cmd, debug=True): - '''Execute a command. - - :param str cmd: A command to be executed. Make sure you escape it properly to prevent shell - expansion, in case it is not desired. - :param bool debug: Whether print out debugging log messages about what's happening. - :returns: returncode of the command - :raise TaskotronRemoteProcessError: If the command has non-zero return code and it isn't a - code of the exitcode directive. - :raise TaskotronRemoteTimeoutError: If the remote hasn't sent any output for - :attr:`TIMEOUT`. - ''' - - if debug: - log.debug('Running command on remote host: %s', cmd) - # write the executed command also to stdio log, so that it's clear what's happening - # when reading it - self.outstream.write('$ %s\n' % cmd) - - # bufsize=1 means line buffering, which should improve the chance of receiving complete - # lines. get_pty=True allocates a pseudo-terminal, which is needed to have stdout and - # stderr interlaced properly (in the exact chronological order) - stdin, stdout, _ = self.ssh.exec_command(cmd, timeout=self.TIMEOUT, bufsize=1, - get_pty=True) - - # stdout.channel represents channel for both stdout and stderr - channel = stdout.channel - - # there seems to be no way to tell whether the remote is alive, - # so just use the counter - # https://phab.qa.fedoraproject.org/T593 - alive_counter = 0 - while not channel.exit_status_ready(): - if channel.recv_ready(): - data = channel.recv(65536) - self.outstream.write(data) - alive_counter = 0 - - time.sleep(0.1) - alive_counter += 0.1 - - if alive_counter >= self.TIMEOUT: - raise exc.TaskotronRemoteTimeoutError( - 'No output received from machine %s@%s:%s for %d seconds while running: %s' % - (self.username, self.hostname, self.port, self.TIMEOUT, cmd)) - - retcode = channel.recv_exit_status() - # reprint the rest of data that was left in the buffer - self.outstream.write(stdout.read()) - - if retcode != 0 and retcode != exitcode_directive.FAILURE: - raise exc.TaskotronRemoteProcessError('Command "%s" on %s@%s exited with code %s' % - (cmd, self.username, self.hostname, retcode)) - return retcode - - def install_pkgs(self, pkgs): - '''Install packages via dnf. - - First tries to install packages using ``dnf --cacheonly --best``. Running from cache - improves performance, while ``--best`` ensures updating to the latest version. If this - doesn't work for some reason, we try again with refreshed cache, and if that still - doesn't work, we drop ``--best`` to allow for broken deps of latest packages, if there - are older packages with working deps. - - :param list pkgs: A list of packages to be installed (supports any argument that - ``dnf install`` accepts) - :raise TaskotronRemoteError: If the command has non-zero return code or times out (see - :meth:`cmd`). - ''' - - pkgs_esc = [pipes.quote(pkg) for pkg in pkgs] - log.info('Installing %d packages on remote host...', len(pkgs)) - try: - self.cmd('dnf --assumeyes --cacheonly --best install %s' % ' '.join(pkgs_esc)) - except exc.TaskotronRemoteProcessError as e: - log.debug('Installation failed: %s', e) - log.debug('Trying again with forced metadata refresh...') - try: - self.cmd('dnf --assumeyes --refresh --best install %s' % ' '.join(pkgs_esc)) - except exc.TaskotronRemoteError as e: - log.debug('Installation failed: %s', e) - log.debug('Trying again without --best...') - self.cmd('dnf --assumeyes install %s' % ' '.join(pkgs_esc)) - - def write_file(self, remote_path, data, overwrite=True, debug=True): - '''Write data to a remote file. - - :param str remote_path: A path to the remote file - :param str data: Data to be written - :param bool overwrite: Whether to overwrite remote path. Default is True. - :param bool debug: Whether print out debugging log messages about what's happening. - :raise TaskotronRemoteError: If data could not be written - ''' - - if debug: - log.debug('Writing data to %s@%s:%s ...', self.username, self.hostname, remote_path) - - # keep the conditionals in this order to avoid unnecessary remote calls - if not overwrite and self._remote_file_exists(remote_path): - log.info('Remote path %s already exists, not overwriting.', remote_path) - return - - try: - with self.sftp.open(remote_path, 'w') as remote_file: - remote_file.write(data) - except (socket.error, IOError) as e: - raise exc.TaskotronRemoteError('Could not write data to %s@%s:%s: %s' % - (self.username, self.hostname, remote_path, e)) - - def put_file(self, local_path, remote_path, overwrite=True, debug=True): - '''Copy a file to a remote path. File permissions are not preserved. - - :param str local_path: A path to the local file - :param str remote_path: A path to the remote file - :param bool overwrite: Whether to overwrite remote path. Default is True. - :param bool debug: Whether print out debugging log messages about what's happening. - :return: :class:`paramiko.SFTPAttributes` object containing attributes about the given - file, if successful. ``None`` otherwise. - :raise TaskotronRemoteError: If the file could not be copied - ''' - if debug: - log.debug('Copying %s to %s@%s:%s ...', local_path, self.username, self.hostname, - remote_path) - - # keep the conditionals in this order to avoid unnecessary remote calls - if not overwrite and self._remote_file_exists(remote_path): - if debug: - log.debug('Remote path %s already exists, not overwriting.', remote_path) - return - - try: - return self.sftp.put(local_path, remote_path) - except (socket.error, IOError) as e: - raise exc.TaskotronRemoteError('Could not put file %s to %s@%s:%s: %s' % - (local_path, self.username, self.hostname, - remote_path, e)) - - def get_file(self, remote_path, local_path, debug=True): - '''Get a file from a remote path. File permissions are not preserved. - - :param str remote_path: A path to the remote file - :param str local_path: A path to the local file - :param bool debug: Whether print out debugging log messages about what's happening. - :raise TaskotronRemoteError: If the file could not be downloaded - ''' - - if debug: - log.debug('Copying %s@%s:%s to %s ...', self.username, self.hostname, remote_path, - local_path) - - try: - self.sftp.get(remote_path, local_path) - except (socket.error, IOError) as e: - raise exc.TaskotronRemoteError('Could not copy file %s@%s:%s to %s: %s' % - (self.username, self.hostname, remote_path, - local_path, e)) - - def _remote_file_exists(self, remote_path): - try: - path_stats = self.sftp.lstat(remote_path) - if path_stats is not None: - return True - except IOError: - pass - - return False - - def _remote_isdir(self, remote_path): - '''Return ``True`` if ``remote_path`` exists and is a directory, otherwise ``False`` (so - e.g. even if it exists, but is a file).''' - try: - return stat.S_ISDIR(self.sftp.lstat(remote_path).st_mode) - except IOError: - # when the dir doesn't exist: IOError: [Errno 2] No such file - return False - - def put_dir(self, local_path, remote_path, overwrite=True, debug=True): - '''Copy a directory to a remote path. This method creates a tarball from contents of - ``local_path``, copies the tarball to remote machine and extracts it to ``remote_path``. - File permissions and symlinks are preserved. - - :param str remote_path: A path to the remote directory. This directory will be created, if - its parent exists. Otherwise you need to create the full tree - structure manually beforehand. - :param str local_path: A path to the local directory - :param bool overwrite: Whether to overwrite remote path (merge local dir with the remote - dir). If you choose to not overwrite and ``remote_path`` exists, - this method with just immediately return. - :param bool debug: Whether print out debugging log messages about what's happening. - :raise TaskotronRemoteError: If the directory could not be copied - ''' - if debug: - log.debug('Copying %s to %s@%s:%s ...', local_path, self.username, - self.hostname, remote_path) - - # keep the conditionals in this order to avoid unnecessary remote calls - if not overwrite and self._remote_file_exists(remote_path): - if debug: - log.debug('Remote path %s already exists, not overwriting.', remote_path) - return - - try: - if not self._remote_isdir(remote_path): - self.sftp.mkdir(remote_path) - - files_to_copy = os.listdir(local_path) - tar_local_path = tempfile.mktemp() - tar_remote_path = os.path.join(remote_path, 'tarred_files.tar') - - with tarfile.open(tar_local_path, 'w') as tar: - for filename in files_to_copy: - tar.add(os.path.join(local_path, filename), arcname=filename) - - self.put_file(tar_local_path, tar_remote_path, debug=False) - self.cmd('tar xf %s -C %s' % (pipes.quote(tar_remote_path), pipes.quote(remote_path)), - debug=False) - - os.remove(tar_local_path) - self.cmd('rm -f %s' % pipes.quote(tar_remote_path), debug=False) - - except (OSError, IOError) as e: - log.exception(e) - raise exc.TaskotronRemoteError('Could not copy dir %s to %s@%s:%s: %s' % - (local_path, self.username, self.hostname, - remote_path, e)) - - def get_dir(self, remote_path, local_path, debug=True): - '''Get a directory from a remote path. File permissions are not preserved. - - :param str remote_path: A path to the remote directory - :param str local_path: A path to the local directory. This directory will be created, if - its parent exists. Otherwise you need to create the full tree - structure manually beforehand. - :param bool debug: Whether print out debugging log messages about what's happening. - :raise TaskotronRemoteError: If the directory could not be downloaded - ''' - - if debug: - log.debug('Recursively copying %s@%s:%s to %s ...', self.username, self.hostname, - remote_path, local_path) - - try: - if not os.path.isdir(local_path): - os.mkdir(local_path) - - fileattrs = self.sftp.listdir_attr(remote_path) - - for fileattr in fileattrs: - remote_file = os.path.join(remote_path, fileattr.filename) - local_file = os.path.join(local_path, fileattr.filename) - if stat.S_ISDIR(fileattr.st_mode): - self.get_dir(remote_file, local_file, debug=False) - else: - self.get_file(remote_file, local_file, debug=False) - except (OSError, IOError) as e: - raise exc.TaskotronRemoteError('Could not copy dir %s@%s:%s to %s: %s' % - (self.username, self.hostname, remote_path, - local_path, e)) diff --git a/libtaskotron/taskformula.py b/libtaskotron/taskformula.py deleted file mode 100644 index 933fd07..0000000 --- a/libtaskotron/taskformula.py +++ /dev/null @@ -1,246 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2009-2014, Red Hat, Inc. -# License: GPL-2.0+ -# See the LICENSE file for more details on Licensing - -'''Methods for operating with a task formula''' - -from __future__ import absolute_import -import collections -import string -import re - -from libtaskotron import exceptions as exc -from libtaskotron.logger import log -from libtaskotron.ext.fedora import rpm_utils - - -class DotTemplate(string.Template): - """Redefines the :class:`string.Template`'s ``idpattern`` property to - allow using ``.`` in the variable names. This is then later used for - key/property access in variable replacement. - """ - idpattern = '[_a-z](?:\.?[_a-z0-9]+)*' - - -def _get_key_or_attribute(variables, var_name): - """Allows for the ``.`` selector to work as selector for variables. - When a variable is dict-like (collections.MutableMapping), the selector looks for a key. - When a variable is not dict-like, it looks for the object's attribute. - - :param dict variables: names (keys) and values of available variables - :param str var_name: variable name, possibly containing ``.`` selector to be found - :raise KeyError: when a dict-like object (anywhere in the chain) does not contain requested key - :raise AttributeError: when an non-dic object (anywhere in the chain) - does not have the requested attribute - """ - if '.' not in var_name: - return variables[var_name] - - var_name, selector = var_name.split('.', 1) - outcome = variables[var_name] - for part in selector.split('.'): - if isinstance(outcome, collections.MutableMapping): - outcome = outcome[part] - else: - outcome = outcome.__getattribute__(part) - return outcome - -def _replace_vars(text, variables): - '''Go through ``text`` and replace all variables (in the form of what is - supported by :class:`VariableTemplate`) with their values, as provided in - ``variables``. This is used for variable expansion in the task formula. - - :param str text: input text where to search for variables - :param dict variables: names (keys) and values of variables to replace - :return: if ``text`` contains just a single variable and nothing else, the - variable value is directly returned (i.e. with matching type, not - cast to ``str``). If ``text`` contains something else as well - (other variables or text), a string is returned. - :raise TaskotronYamlError: if ``text`` contains a variable that is not - present in ``variables``, or if the variable - syntax is incorrect - ''' - - try: - # try to find the first match - match = DotTemplate.pattern.search(text) - - if not match: - return text - - # There are 4 groups in the pattern: 1 - escaped, 2 - named, 3 - braced, - # 4 - invalid. Group 0 returns the whole match. - if match.group(0) == text and (match.group(2) or match.group(3)): - # We found a single variable and nothing more. We shouldn't return - # a string, but the exact value, so that we don't lose value type. - # This makes it possible to pass lists, dicts, etc as variables. - var_name = match.group(2) or match.group(3) - return _get_key_or_attribute(variables, var_name) - - # Now it's clear there's also something else in `text` than just a - # single variable. We will replace all variables and return a string - # again. - - # To make things easy, we'll create expanded_variables dict, that contains values for - # all the replacable variables in the text. - # i.e. if variables = {"n": 1}, and text is "foo bar {$n.imag}" then - # "n.imag" key is added to variables, with the respective value. - - expanded_variables = {} - for match in DotTemplate.pattern.findall(text): - var_name = match[1] or match[2] - if var_name: - expanded_variables[var_name] = _get_key_or_attribute(variables, var_name) - - output = DotTemplate(text).substitute(expanded_variables) - return output - - except (KeyError, AttributeError) as e: - raise exc.TaskotronYamlError("The task formula includes a variable, " - "but no value has been provided for it: %s" % e) - - except ValueError as e: - raise exc.TaskotronYamlError("The task formula includes an incorrect " - "variable definition. Dollar signs must be doubled if they " - "shouldn't be considered as a variable denotation.\n" - "Error: %s\n" - "Text: %s" % (e, text)) - - -def replace_vars_in_action(action, variables): - '''Replace variables (${var} or ${var.key}) with their values in an action (YAML dictionary). - - Only basestring values are searched for variables syntax, which means variables can either be - inside leaves (in a tree sense, meaning they can't be traversed further and are of a primitive - type), or inside key names in a dictionary. - - :param dict action: An action specification parsed from the task formula. See - :meth:`.Runner.do_actions` to see what an action looks like. - :param dict variables: names (keys) and values of variables to replace - :raise TaskotronYamlError: if ``text`` contains a variable that is not present in - ``variables``; or if the variable syntax is incorrect; of if you try - to replace a dictionary key with something else than a string - ''' - - visited = [] # all visited nodes in a tree - stack = [action] # nodes waiting for inspection - - while stack: - vertex = stack.pop() - - if vertex in visited: - continue - - visited.append(vertex) - children = [] # list of tuples (index/key, child_value) - - if isinstance(vertex, collections.MutableMapping): - children = vertex.items() # list of (key, value) - elif isinstance(vertex, collections.MutableSequence): - children = list(enumerate(vertex)) # list of (index, value) - else: - log.warn("Unknown structure '%s' in YAML file, this shouldn't happen: %s", - type(vertex), vertex) - - for index, child_val in children: - if isinstance(child_val, basestring): - # leaf node and a string, replace variables - vertex[index] = _replace_vars(child_val, variables) - elif isinstance(child_val, collections.Iterable): - # traversable further down, mark for visit - stack.append(child_val) - - # for dicts, also replace vars in key names - if isinstance(index, basestring): # meaning it's a dictionary - new_key = _replace_vars(index, variables) - if not isinstance(new_key, basestring): - raise exc.TaskotronYamlError( - "Dictionary key '%s' can't be replaced with value %r, because it's not a " - "string, but %s" % (index, new_key, type(new_key))) - # only swap keys if the new key is actually different from the old one - if new_key != index: - vertex[new_key] = vertex[index] - del vertex[index] - - -def devise_environment(formula, arg_data): - '''Takes a parsed formula, and returns a required run-environment (i.e. - distro, arch, fedora release, and base-image flavor), based on item and type. - - :param dict formula: parsed formula file (or dict with equivalent structure) - :param dict arg_data: parsed command-line arguments. item, type and arch - are used in this method - :return: dict containing release, flavor, arch. Each either set, or None - :raise TaskotronValueError: when environment can't be parsed from the formula - ''' - - environment = formula.get('environment', {}) - env = {'distro': None, 'release': None, 'flavor': None, 'arch': None} - - if not isinstance(environment, dict): - raise exc.TaskotronValueError( - "Environment must be dictionary, not %s" % type(environment)) - - env['distro'] = environment.get('distro', None) - env['release'] = environment.get('release', None) - env['flavor'] = environment.get('flavor', None) - env['arch'] = environment.get('arch', None) - - item = arg_data.get('item', None) - item_type = arg_data.get('type', None) - - if not env['distro']: - if item_type == 'koji_build': - # FIXME: find a way to make this not Fedora-specific - # For `xchat-2.8.8-21.fc20` disttag is `fc20` for example - try: - distro = rpm_utils.get_dist_tag(item)[:2] - env['distro'] = {'fc': 'fedora'}.get(distro, None) - except exc.TaskotronValueError: - env['distro'] = None - - elif item_type == 'koji_tag': - if re.match(r'^f[0-9]{2}-.*', item): - env['distro'] = 'fedora' - - if env['distro']: - log.debug("Environment/distro overriden with ENVVAR from %r:%r to %r" % - (item_type, item, env['distro'])) - else: - log.debug("Environment/distro can not be inferred from %r:%r. Using default." % - (item_type, item)) - - if not env['release']: - if item_type == 'koji_build': - # FIXME: find a way to make this not Fedora-specific - # Last two characters in rpm's disttag are the Fedora release. - # For `xchat-2.8.8-21.fc20` disttag is `fc20` for example - try: - env['release'] = rpm_utils.get_dist_tag(item)[-2:] - except exc.TaskotronValueError: - env['release'] = None - - elif item_type == 'koji_tag': - if re.match(r'^f[0-9]{2}-.*', item): - env['release'] = item[1:3] - - if env['release']: - log.debug("Environment/release inferred from %r:%r to %r" % - (item_type, item, env['release'])) - else: - log.debug("Environment/release can not be inferred from %r:%r. Using default." % - (item_type, item)) - - if not env['flavor']: - log.debug("Environment/flavor not specified. Using default") - - if not env['arch']: - arch = arg_data.get('arch') - if not arch or arch == 'noarch': - log.warn("Environment/arch can not be inferred from %r. Using default." % arch) - else: - env['arch'] = arch - log.debug("Environment/arch inferred from %r to %r" % (arch, env['arch'])) - - return env diff --git a/requirements.txt b/requirements.txt index 4ea5dd8..eb879f5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -21,7 +21,6 @@ python-fedora >= 0.8.0 munch >= 2.0.2 # libtaskotron-disposable -paramiko >= 1.15.1 testcloud >= 0.1.10 # Test suite requirements From 342b3798eb190fc92ba48656d33c16961d763011 Mon Sep 17 00:00:00 2001 From: Mike Ruckman Date: Aug 07 2017 11:59:15 +0000 Subject: [PATCH 2/2] *WIP* More hacky support, but for containers Summary: This patch builds off of Martin's ansible tasks commit (2b720ca). I've added a Dockerfile in the 'libtaskotron/ext/' directory as well as a module to 'libtaskotron/ext/disposable' for dealing with Docker containers. This expects a docker daemon to be running on the host (so is akin to --libvirt), and will build the 'taskotron-worker' image if it's not already found on the system. Testing this is very similar to testing the patch Martin wrote, except you use the '--docker' flag during invocation. To test: $ git clone https://github.com/stefwalter/gzip-dist-git $ sudo runtask -d -i gzip -t koji_build -a x86_64 --docker gzip-dist-git/tests/test_rpm.yml Test Plan: - Run the above playbook Reviewers: tflink, mkrizek, #libtaskotron, jskladan, kparal, jsedlak Subscribers: tflink, kparal, jskladan Tags: #libtaskotron Differential Revision: https://phab.qa.fedoraproject.org/D1196 --- diff --git a/libtaskotron/executor.py b/libtaskotron/executor.py index 81ff2e6..0895d82 100644 --- a/libtaskotron/executor.py +++ b/libtaskotron/executor.py @@ -13,9 +13,9 @@ from libtaskotron import config from libtaskotron.logger import log from libtaskotron import file_utils from libtaskotron import os_utils -from libtaskotron import exceptions as exc from libtaskotron.ext.disposable import vm +from libtaskotron.ext.disposable import docker # TODO exc @@ -24,6 +24,7 @@ class Executor(object): self.arg_data = arg_data self.task_vm = None self.run_remotely = False + self.task_container = None def _spawn_vm(self): self.task_vm = vm.TestCloudMachine(self.arg_data['uuid']) @@ -35,15 +36,29 @@ class Executor(object): return self.task_vm.ipaddr + def _spawn_container(self): + self.client = docker.DockerClient() + self.task_container = self.client.create_container(uuid=self.arg_data['uuid'], + artifacts=self.arg_data['artifactsdir']) + + log.info('Running task on a container on port {}.'.format(self.task_container)) + + return ("127.0.0.1", self.task_container) + def _prepare(self): # when running remotely, run directly over ssh, instead of using libvirt persistent = False + # when running on docker, run directly over ssh, into the container + docker = False runtask_mode = config.get_config().runtask_mode if runtask_mode == config.RuntaskModeName.LOCAL: self.run_remotely = False elif runtask_mode == config.RuntaskModeName.LIBVIRT: self.run_remotely = True + elif runtask_mode == config.RuntaskModeName.DOCKER: + self.run_remotely = True + docker = True else: assert False, 'This should never occur' @@ -61,10 +76,16 @@ class Executor(object): self.run_remotely = True persistent = True + elif self.arg_data['docker']: + log.debug('Forcing execution on docker (option --docker)') + self.run_remotely = True + persistent = False + docker = True + log.debug('Execution mode: %s', 'remote' if self.run_remotely else 'local') ipaddr = '127.0.0.1' - if self.run_remotely: + if self.run_remotely and not docker: if persistent: ipaddr = self.arg_data['machine'] else: @@ -73,6 +94,15 @@ class Executor(object): with open(os.path.join('inventory'), 'w') as f: f.write('[localhost]\n%s' % ipaddr) + if docker: + ipaddr, port = self._spawn_container() + + with open(os.path.join('inventory'), 'w') as f: + f.write('[localhost]\n%s\t\tansible_user=root' + ' ansible_port=%s ansible_password=passw0rd ' + 'ansible_ssh_common_args="-o UserKnownHostsFile=/dev/null ' + '-o StrictHostKeyChecking=no"'% (ipaddr, port)) + def _execute(self): cmd = [ 'ansible-playbook', self.arg_data['task'], @@ -83,8 +113,10 @@ class Executor(object): '-e', 'subjects=%s' % self.arg_data['item'], ] - if self.run_remotely: + if self.run_remotely and not self.arg_data['docker']: cmd.extend(['--private-key', self.arg_data['ssh_privkey']]) + elif self.arg_data['docker']: + cmd.extend(['-K']) else: cmd.extend(['-K', '-c', 'local']) @@ -112,3 +144,7 @@ class Executor(object): if self.task_vm is not None: self.task_vm.teardown() os.remove('inventory') + + if self.task_container is not None: + self.client.stop_container('taskotron-worker-{}'.format(self.arg_data['uuid'])) + self.client.rm_container('taskotron-worker-{}'.format(self.arg_data['uuid'])) diff --git a/libtaskotron/ext/disposable/docker.py b/libtaskotron/ext/disposable/docker.py new file mode 100644 index 0000000..9393617 --- /dev/null +++ b/libtaskotron/ext/disposable/docker.py @@ -0,0 +1,186 @@ +# -*- coding: utf-8 -*- +# Copyright 2009-2015, Red Hat, Inc. +# License: GPL-2.0+ +# See the LICENSE file for more details on Licensing + +"""Interface for locally spawned docker containers used for task execution.""" + +import os +import random +import subprocess as sp + +from libtaskotron.logger import log + + +class DockerClient(object): + '''Helper class for working with the docker daemon.''' + + def __init__(self, host_ip="127.0.0.1"): + ''':param host_ip: The host IP where the docker daemon runs. We + assume for now that it's running on localhost.''' + + # FIXME: Support running commands against a remote host. + self.host = host_ip + self.images = self._get_images() + self.containers = self._get_containers() + # We need to keep a list of the containers we create, so + # we can co-exist with other users of whatever docker + # daemon we might be interacting with. + self.owned_containers = [] + self.owned_images = [] + + def _get_images(self): + '''Return a list of image dicts.''' + + template = "{{.ID}}\t{{.Repository}}" + raw_output = sp.check_output(['docker', + 'images', + '--format', + template]) + + images = raw_output.split('\n') + attrs = ['id', 'name'] + images = [image.split('\t') for image in images[:-1]] + images = [dict(zip(attrs, image)) for image in images] + + return images + + def _get_containers(self): + '''Return a list of all containers on the host.''' + + # This is a go template to clean up docker ps output + template = "{{.ID}}\t{{.Command}}\t{{.Status}}\t{{.Ports}}\t{{.Names}}" + + raw_output = sp.check_output(['docker', + 'ps', + '-a', + '--format', + template]) + + containers = raw_output.split('\n') + + # Clean up the output a bit, also remove the last, empty, entry + containers = [x.split('\t') for x in containers][:-1] + + attrs = ['id', 'command', 'status', 'ports', 'names'] + containers = [dict(zip(attrs, container)) for container in containers] + + return containers + + def _find_container(self, search_term): + '''Find a container based on either name or id.''' + + for container in self.containers: + if search_term in container.values(): + return container + else: + pass + + return "No container found by {}.".format(search_term) + + def _manipulate_container(self, identifier, action=None): + '''Start or stop a container based on either it's name or id. + + :param action: command for docker, either 'start', 'stop', or 'rm' + :param identifier: a container ID or Name''' + + if action not in ['start', 'stop', 'rm']: + raise ClientError("Available actions are: start, stop, and rm. " + "Could not perform: {}".format(action)) + + container = self._find_container(identifier) + try: + sp.check_call(['docker', + action, + container['names']]) + + except: + log.debug("Failed to {} the container '{}'.".format(action, container['names'])) + raise ClientError("Could not {} the container.".format(action)) + + def build_image(self, path_to_dockerfile_dir, imgtag='taskotron-worker'): + '''Run `docker build Dockerfile` against the specified path.''' + sp.check_call(['docker', + 'build', + '-t', + imgtag, + path_to_dockerfile_dir]) + # Update our lists + self.images = self._get_images() + self.owned_images.append(imgtag) + + def create_container(self, + image_name='taskotron-worker', + container_name='taskotron-worker', + uuid=None, + artifacts=None): + '''Method to create a container from a specified image. + :param image_name: the name of the built image you want to launch. + ''' + + # Build the image if it's not already there + image_found = False + for image in self.images: + if image_name == image['name']: + log.debug("Image already exists, skipping image build...") + image_found = True + + if not image_found: + log.debug("Image not found, building...") + self.build_image(os.path.abspath(os.curdir) + '/libtaskotron/ext/docker/') + + if uuid: + container_name += "-{}".format(uuid) + + # Generate random port for host to use (between 2200-2300) + host_port = "{}".format(random.randrange(2200, 2300)) + log.info("Container will be available on port: {}".format(host_port)) + sp.check_call(['docker', + 'run', + '-d', + '--name', container_name, + '-p', '{}:22'.format(host_port), + '-v', '{}:/artifacts'.format(artifacts), + image_name]) + + # Update our list of all containers + self.containers = self._get_containers() + + # Update the list of owned_containers + self.owned_containers.append(self._find_container(container_name)) + + return host_port + + def start_container(self, identifier): + self._manipulate_container(identifier, action='start') + + def stop_container(self, identifier): + self._manipulate_container(identifier, action='stop') + + def rm_container(self, identifier): + # Update our list of owned containers, since we're deleting it + for i in xrange(len(self.owned_containers)): + if self.owned_containers[i]['names'] or self.owned_containers[i]['id'] == identifier: + self.owned_containers.pop(i) + break + + self._manipulate_container(identifier, action='rm') + + def clear_all(self): + '''Clean out all owned containers and images.''' + + log.info("Clearing out all our containers...") + for container in self.owned_containers: + self.stop_container(container['names']) + self.rm_container(container['names']) + + log.info("Clearing out all our images...") + for image in self.owned_images: + sp.check_call(['docker', + 'rmi', + image]) + + +class ClientError(Exception): + '''General exceptions we might run into using this client.''' + pass diff --git a/libtaskotron/ext/docker/Dockerfile b/libtaskotron/ext/docker/Dockerfile new file mode 100644 index 0000000..24d90d1 --- /dev/null +++ b/libtaskotron/ext/docker/Dockerfile @@ -0,0 +1,21 @@ +#FROM registry.fedoraproject.org/fedora:25 +FROM fedora:25 + +RUN dnf install openssh-server PyYAML libtaskotron-core libtaskotron-fedora \ + libtaskotron-config python-solv python-librepo passwd -y && \ + mkdir -p /var/log/taskotron /srv/taskotron/artifacts && \ + dnf clean all + +ENV LANG C.utf8 +ENV LC_ALL C.utf8 + +ADD https://pagure.io/taskotron/libtaskotron/blob/master/f/conf/yumrepoinfo.conf.example /etc/taskotron/yumrepoinfo.conf +ADD taskotron.yaml /etc/taskotron/taskotron.yaml +ADD namespaces.yaml /etc/taskotron/namespaces.yaml + +RUN echo "passw0rd" | passwd --stdin root +RUN ssh-keygen -t rsa -f /etc/ssh/ssh_host_rsa_key -N "" + +EXPOSE 22 + +CMD ["/sbin/sshd", "-D"] diff --git a/libtaskotron/ext/docker/namespaces.yaml b/libtaskotron/ext/docker/namespaces.yaml new file mode 100644 index 0000000..0c24276 --- /dev/null +++ b/libtaskotron/ext/docker/namespaces.yaml @@ -0,0 +1,5 @@ +namespaces_safe: + - scratch + - qa + - pkg + - dist diff --git a/libtaskotron/ext/docker/taskotron.yaml b/libtaskotron/ext/docker/taskotron.yaml new file mode 100644 index 0000000..301fb58 --- /dev/null +++ b/libtaskotron/ext/docker/taskotron.yaml @@ -0,0 +1,7 @@ +profile: production +runtask_mode: local +resultsdb_server: http://resultsdb:5001/api/v2.0 +resultsdb_frontend: http://localhost:5002 +execdb_server: http://localhost:5003 +taskotron_master: http://localhost:8010 +artifacts_baseurl: http://localhost:8081 diff --git a/libtaskotron/main.py b/libtaskotron/main.py index 1ceb346..0237e75 100644 --- a/libtaskotron/main.py +++ b/libtaskotron/main.py @@ -58,6 +58,8 @@ def get_argparser(): help="path to private key for remote connections over ssh") parser.add_argument("--no-destroy", action="store_true", help="do not destroy disposable client at the end of task execution") + parser.add_argument("--docker", action="store_true", + help="Run the task inside a docker container.") return parser