diff --git a/apps/ops/models/task.py b/apps/ops/models/task.py index 979d36479..9dd3c5b24 100644 --- a/apps/ops/models/task.py +++ b/apps/ops/models/task.py @@ -10,49 +10,23 @@ from ops.models import TaskRecord from django.db import models from django.utils.translation import ugettext_lazy as _ -__all__ = ["Task", "SubTask"] - - -logger = logging.getLogger(__name__) +__all__ = ["Task"] class Task(models.Model): - record = models.OneToOneField(TaskRecord) - name = models.CharField(max_length=128, blank=True, verbose_name=_('Name')) - is_gather_facts = models.BooleanField(default=False, verbose_name=_('Is Gather Ansible Facts')) - assets = models.ManyToManyField(Asset, related_name='tasks') + """ + Ansible 的Task + """ + name = models.CharField(max_length=128, verbose_name=_('Task name')) + module_name = models.CharField(max_length=128, verbose_name=_('Task module')) + module_args = models.CharField(max_length=512, blank=True, verbose_name=_("Module args")) def __unicode__(self): return "%s" % self.name - @property - def ansible_assets(self): - return [] - def run(self): - from ops.utils.ansible_api import ADHocRunner, Options - conf = Options() - gather_facts = "yes" if self.is_gather_facts else "no" - play_source = { - "name": "Ansible Play", - "hosts": "default", - "gather_facts": gather_facts, - "tasks": [ - dict(action=dict(module='ping')), - ] - } - hoc = ADHocRunner(conf, play_source, *self.ansible_assets) - uuid = "tasker-" + uuid4().hex - ext_code, result = hoc.run("test_task", uuid) - print(ext_code) - print(result) +class Play(models.Model): + """ + Playbook 模板, 定义好Template后生成 Playbook + """ - -class SubTask(models.Model): - task = models.ForeignKey(Task, related_name='sub_tasks', verbose_name=_('Ansible Task')) - module_name = models.CharField(max_length=128, verbose_name=_('Ansible Module Name')) - module_args = models.CharField(max_length=512, blank=True, verbose_name=_("Ansible Module Args")) - register = models.CharField(max_length=128, blank=True, verbose_name=_('Ansible Task Register')) - - def __unicode__(self): - return "%s %s" % (self.module_name, self.module_args) diff --git a/apps/ops/utils/ansible_api.py b/apps/ops/utils/ansible_api.py deleted file mode 100644 index af3cafb98..000000000 --- a/apps/ops/utils/ansible_api.py +++ /dev/null @@ -1,404 +0,0 @@ -# ~*~ coding: utf-8 ~*~ -# from __future__ import unicode_literals, print_function - -import os -import logging -from collections import namedtuple - -from ansible.executor.task_queue_manager import TaskQueueManager -from ansible.vars import VariableManager -from ansible.parsing.dataloader import DataLoader -from ansible.executor.playbook_executor import PlaybookExecutor -from ansible.playbook.play import Play -from ansible.plugins.callback import CallbackBase -import ansible.constants as C -from ansible.utils.vars import load_extra_vars -from ansible.utils.vars import load_options_vars - -from .inventory import JMSInventory -from common.utils import get_logger - - -__all__ = ["ADHocRunner", "Options"] - -C.HOST_KEY_CHECKING = False - -logger = get_logger(__name__) - - -class AnsibleError(StandardError): - pass - - -class AdHocResultCallback(CallbackBase): - """ - Custom Callback - """ - def __init__(self, display=None): - self.result_q = dict(contacted={}, dark={}) - super(AdHocResultCallback, self).__init__(display) - - def gather_result(self, n, res): - self.result_q[n].update({res._host.name: res._result}) - - def v2_runner_on_ok(self, result): - self.gather_result("contacted", result) - - def v2_runner_on_failed(self, result, ignore_errors=False): - self.gather_result("dark", result) - - def v2_runner_on_unreachable(self, result): - self.gather_result("dark", result) - - def v2_runner_on_skipped(self, result): - self.gather_result("dark", result) - - def v2_playbook_on_task_start(self, task, is_conditional): - pass - - def v2_playbook_on_play_start(self, play): - pass - - -class PlaybookResultCallBack(CallbackBase): - """ - Custom callback model for handlering the output data of - execute playbook file, - Base on the build-in callback plugins of ansible which named `json`. - """ - - CALLBACK_VERSION = 2.0 - CALLBACK_TYPE = 'stdout' - CALLBACK_NAME = 'Dict' - - def __init__(self, display=None): - super(PlaybookResultCallBack, self).__init__(display) - self.results = [] - self.output = "" - self.item_results = {} # {"host": []} - - def _new_play(self, play): - return { - 'play': { - 'name': play.name, - 'id': str(play._uuid) - }, - 'tasks': [] - } - - def _new_task(self, task): - return { - 'task': { - 'name': task.get_name(), - }, - 'hosts': {} - } - - def v2_playbook_on_no_hosts_matched(self): - self.output = "skipping: No match hosts." - - def v2_playbook_on_no_hosts_remaining(self): - pass - - def v2_playbook_on_task_start(self, task, is_conditional): - self.results[-1]['tasks'].append(self._new_task(task)) - - def v2_playbook_on_play_start(self, play): - self.results.append(self._new_play(play)) - - def v2_playbook_on_stats(self, stats): - hosts = sorted(stats.processed.keys()) - summary = {} - for h in hosts: - s = stats.summarize(h) - summary[h] = s - - if self.output: - pass - else: - self.output = { - 'plays': self.results, - 'stats': summary - } - - def gather_result(self, res): - if res._task.loop and "results" in res._result and res._host.name in self.item_results: - res._result.update({"results": self.item_results[res._host.name]}) - del self.item_results[res._host.name] - - self.results[-1]['tasks'][-1]['hosts'][res._host.name] = res._result - - def v2_runner_on_ok(self, res, **kwargs): - if "ansible_facts" in res._result: - del res._result["ansible_facts"] - - self.gather_result(res) - - def v2_runner_on_failed(self, res, **kwargs): - self.gather_result(res) - - def v2_runner_on_unreachable(self, res, **kwargs): - self.gather_result(res) - - def v2_runner_on_skipped(self, res, **kwargs): - self.gather_result(res) - - def gather_item_result(self, res): - self.item_results.setdefault(res._host.name, []).append(res._result) - - def v2_runner_item_on_ok(self, res): - self.gather_item_result(res) - - def v2_runner_item_on_failed(self, res): - self.gather_item_result(res) - - def v2_runner_item_on_skipped(self, res): - self.gather_item_result(res) - - -class PlayBookRunner(object): - """ - 用于执行AnsiblePlaybook的接口.简化Playbook对象的使用. - """ - Options = namedtuple('Options', [ - 'listtags', 'listtasks', 'listhosts', 'syntax', 'connection', - 'module_path', 'forks', 'remote_user', 'private_key_file', 'timeout', - 'ssh_common_args', 'ssh_extra_args', 'sftp_extra_args', - 'scp_extra_args', 'become', 'become_method', 'become_user', - 'verbosity', 'check', 'extra_vars']) - - def __init__(self, - hosts=None, - playbook_path=None, - forks=C.DEFAULT_FORKS, - listtags=False, - listtasks=False, - listhosts=False, - syntax=False, - module_path=None, - remote_user='root', - timeout=C.DEFAULT_TIMEOUT, - ssh_common_args=None, - ssh_extra_args=None, - sftp_extra_args=None, - scp_extra_args=None, - become=True, - become_method=None, - become_user="root", - verbosity=None, - extra_vars=None, - connection_type="ssh", - passwords=None, - private_key_file=None, - check=False): - - C.RETRY_FILES_ENABLED = False - self.callbackmodule = PlaybookResultCallBack() - if playbook_path is None or not os.path.exists(playbook_path): - raise AnsibleError( - "Not Found the playbook file: %s." % playbook_path) - self.playbook_path = playbook_path - self.loader = DataLoader() - self.variable_manager = VariableManager() - self.passwords = passwords or {} - self.inventory = JMSInventory(hosts) - - self.options = self.Options( - listtags=listtags, - listtasks=listtasks, - listhosts=listhosts, - syntax=syntax, - timeout=timeout, - connection=connection_type, - module_path=module_path, - forks=forks, - remote_user=remote_user, - private_key_file=private_key_file, - ssh_common_args=ssh_common_args or "", - ssh_extra_args=ssh_extra_args or "", - sftp_extra_args=sftp_extra_args, - scp_extra_args=scp_extra_args, - become=become, - become_method=become_method, - become_user=become_user, - verbosity=verbosity, - extra_vars=extra_vars or [], - check=check - ) - - self.variable_manager.extra_vars = load_extra_vars(loader=self.loader, - options=self.options) - self.variable_manager.options_vars = load_options_vars(self.options) - - self.variable_manager.set_inventory(self.inventory) - - # 初始化playbook的executor - self.runner = PlaybookExecutor( - playbooks=[self.playbook_path], - inventory=self.inventory, - variable_manager=self.variable_manager, - loader=self.loader, - options=self.options, - passwords=self.passwords) - - if self.runner._tqm: - self.runner._tqm._stdout_callback = self.callbackmodule - - def run(self): - if not self.inventory.list_hosts('all'): - raise AnsibleError('Inventory is empty') - self.runner.run() - self.runner._tqm.cleanup() - return self.callbackmodule.output - - -class ADHocRunner(object): - """ - ADHoc接口 - """ - Options = namedtuple("Options", [ - 'connection', 'module_path', 'private_key_file', "remote_user", - 'timeout', 'forks', 'become', 'become_method', 'become_user', - 'check', 'extra_vars', - ] - ) - - def __init__(self, - hosts=C.DEFAULT_HOST_LIST, - task_name='Ansible Ad-hoc', - module_name=C.DEFAULT_MODULE_NAME, # * command - module_args=C.DEFAULT_MODULE_ARGS, # * 'cmd args' - forks=C.DEFAULT_FORKS, # 5 - timeout=C.DEFAULT_TIMEOUT, # SSH timeout = 10s - pattern="all", # all - remote_user=C.DEFAULT_REMOTE_USER, # root - module_path=None, # dirs of custome modules - connection_type="smart", - become=None, - become_method=None, - become_user=None, - check=False, - passwords=None, - extra_vars=None, - private_key_file=None, - gather_facts='no'): - - self.pattern = pattern - self.variable_manager = VariableManager() - self.loader = DataLoader() - self.module_name = module_name - self.module_args = module_args - self.check_module_args() - self.gather_facts = gather_facts - self.results_callback = AdHocResultCallback() - self.options = self.Options( - connection=connection_type, - timeout=timeout, - module_path=module_path, - forks=forks, - become=become, - become_method=become_method, - become_user=become_user, - check=check, - remote_user=remote_user, - extra_vars=extra_vars or [], - private_key_file=private_key_file, - ) - - self.variable_manager.extra_vars = load_extra_vars(self.loader, options=self.options) - self.variable_manager.options_vars = load_options_vars(self.options) - self.passwords = passwords or {} - self.inventory = JMSInventory(hosts) - self.variable_manager.set_inventory(self.inventory) - - self.play_source = dict( - name=task_name, - hosts=self.pattern, - gather_facts=self.gather_facts, - tasks=[ - dict(action=dict( - module=self.module_name, - args=self.module_args, - ) - ) - ] - ) - - self.play = Play().load( - self.play_source, - variable_manager=self.variable_manager, - loader=self.loader, - ) - - self.runner = TaskQueueManager( - inventory=self.inventory, - variable_manager=self.variable_manager, - loader=self.loader, - options=self.options, - passwords=self.passwords, - stdout_callback=self.results_callback, - ) - - def check_module_args(self): - if self.module_name in C.MODULE_REQUIRE_ARGS and not self.module_args: - err = "No argument passed to '%s' module." % self.module_name - raise AnsibleError(err) - - def run(self): - if not self.inventory.list_hosts("all"): - raise AnsibleError("Inventory is empty.") - - if not self.inventory.list_hosts(self.pattern): - raise AnsibleError( - "pattern: %s dose not match any hosts." % self.pattern) - - try: - self.runner.run(self.play) - except Exception as e: - logger.warning(e) - else: - logger.debug(self.results_callback.result_q) - return self.results_callback.result_q - finally: - if self.runner: - self.runner.cleanup() - if self.loader: - self.loader.cleanup_all_tmp_files() - - def clean_result(self): - failed = self.results_callback.result_q['dark'].keys() - success = self.results_callback.result_q['contacted'].keys() - return {'failed': failed, 'success': success} - - -def test_run(): - assets = [ - { - "hostname": "192.168.152.129", - "ip": "192.168.152.129", - "port": 22, - "username": "root", - "password": "redhat", - }, - ] - hoc = ADHocRunner(module_name='shell', module_args='ls', hosts=assets) - ret = hoc.run() - print(ret) - - play = PlayBookRunner(assets, playbook_path='/tmp/some.yml') - """ - # /tmp/some.yml - --- - - name: Test the plabybook API. - hosts: all - remote_user: root - gather_facts: yes - tasks: - - name: exec uptime - shell: uptime - """ - play.run() - - -if __name__ == "__main__": - test_run() diff --git a/apps/ops/utils/callback.py b/apps/ops/utils/callback.py index 8448c4254..6e0f29cb6 100644 --- a/apps/ops/utils/callback.py +++ b/apps/ops/utils/callback.py @@ -12,7 +12,10 @@ class AdHocResultCallback(CallbackBase): super(AdHocResultCallback, self).__init__(display) def gather_result(self, n, res): - self.result_q[n].update({res._host.name: res._result}) + if res._host.name in self.result_q[n]: + self.result_q[n][res._host.name].append(res._result) + else: + self.result_q[n][res._host.name] = [res._result] def v2_runner_on_ok(self, result): self.gather_result("contacted", result) diff --git a/apps/ops/utils/runner.py b/apps/ops/utils/runner.py index a1603d9e4..1f84ad434 100644 --- a/apps/ops/utils/runner.py +++ b/apps/ops/utils/runner.py @@ -1,17 +1,14 @@ # ~*~ coding: utf-8 ~*~ -# ~*~ coding: utf-8 ~*~ -# from __future__ import unicode_literals, print_function - import os -from collections import namedtuple +import sys +from collections import namedtuple, defaultdict from ansible.executor.task_queue_manager import TaskQueueManager from ansible.vars import VariableManager from ansible.parsing.dataloader import DataLoader from ansible.executor.playbook_executor import PlaybookExecutor from ansible.playbook.play import Play -from ansible.plugins.callback import CallbackBase import ansible.constants as C from ansible.utils.vars import load_extra_vars from ansible.utils.vars import load_options_vars @@ -128,7 +125,7 @@ class PlayBookRunner(object): return self.callbackmodule.output -class ADHocRunner(object): +class AdHocRunner(object): """ ADHoc接口 """ @@ -141,12 +138,8 @@ class ADHocRunner(object): def __init__(self, hosts=C.DEFAULT_HOST_LIST, - task_name='Ansible Ad-hoc', - module_name=C.DEFAULT_MODULE_NAME, # * command - module_args=C.DEFAULT_MODULE_ARGS, # * 'cmd args' forks=C.DEFAULT_FORKS, # 5 timeout=C.DEFAULT_TIMEOUT, # SSH timeout = 10s - pattern="all", # all remote_user=C.DEFAULT_REMOTE_USER, # root module_path=None, # dirs of custome modules connection_type="smart", @@ -159,12 +152,9 @@ class ADHocRunner(object): private_key_file=None, gather_facts='no'): - self.pattern = pattern + self.pattern = '' self.variable_manager = VariableManager() self.loader = DataLoader() - self.module_name = module_name - self.module_args = module_args - self.check_module_args() self.gather_facts = gather_facts self.results_callback = AdHocResultCallback() self.options = self.Options( @@ -186,18 +176,41 @@ class ADHocRunner(object): self.passwords = passwords or {} self.inventory = JMSInventory(hosts) self.variable_manager.set_inventory(self.inventory) + self.tasks = [] + self.play_source = None + self.play = None + self.runner = None + + @staticmethod + def check_module_args(module_name, module_args=''): + if module_name in C.MODULE_REQUIRE_ARGS and not module_args: + err = "No argument passed to '%s' module." % module_name + print(err) + return False + return True + + def run(self, task_tuple, pattern='all', task_name='Ansible Ad-hoc'): + """ + :param task_tuple: (('shell', 'ls'), ('ping', '')) + :param pattern: + :param task_name: + :return: + """ + for module, args in task_tuple: + if not self.check_module_args(module, args): + return + self.tasks.append( + dict(action=dict( + module=module, + args=args, + )) + ) self.play_source = dict( name=task_name, - hosts=self.pattern, + hosts=pattern, gather_facts=self.gather_facts, - tasks=[ - dict(action=dict( - module=self.module_name, - args=self.module_args, - ) - ) - ] + tasks=self.tasks ) self.play = Play().load( @@ -215,12 +228,6 @@ class ADHocRunner(object): stdout_callback=self.results_callback, ) - def check_module_args(self): - if self.module_name in C.MODULE_REQUIRE_ARGS and not self.module_args: - err = "No argument passed to '%s' module." % self.module_name - raise AnsibleError(err) - - def run(self): if not self.inventory.list_hosts("all"): raise AnsibleError("Inventory is empty.") @@ -242,9 +249,13 @@ class ADHocRunner(object): self.loader.cleanup_all_tmp_files() def clean_result(self): - failed = self.results_callback.result_q['dark'].keys() - success = self.results_callback.result_q['contacted'].keys() - return {'failed': failed, 'success': success} + result = defaultdict(dict) + for host, msgs in self.results_callback.result_q['contacted'].items(): + result[host]['success'] = len(msgs) + + for host, msgs in self.results_callback.result_q['dark'].items(): + result[host]['failed'] = len(msgs) + return result def test_run(): @@ -257,8 +268,9 @@ def test_run(): "password": "redhat", }, ] - hoc = ADHocRunner(module_name='shell', module_args='ls', hosts=assets) - ret = hoc.run() + task_tuple = (('shell', 'ls'), ('ping', '')) + hoc = AdHocRunner(hosts=assets) + ret = hoc.run(task_tuple) print(ret) play = PlayBookRunner(assets, playbook_path='/tmp/some.yml')