Skip to content

Instantly share code, notes, and snippets.

@Faheetah
Created Oct 18, 2016
Embed
What would you like to do?
Ansible stdout callback module with better output, used by output=method in task name
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import sys
from ansible import constants as C
from ansible.plugins.callback.default import CallbackModule as CallbackModule_default
from ansible.utils.color import colorize, hostcolor
def _color(play):
if play._result.get('failed'):
return C.COLOR_ERROR
if play._result.get('changed'):
return C.COLOR_CHANGED
if play._result.get('skipped'):
return C.COLOR_SKIP
return 'green'
def _status(play):
if play._result.get('failed'):
return 'failed'
if play._result.get('changed'):
return 'changed'
if play._result.get('skipped'):
return 'skipped'
return 'ok'
def _parse_flags(play):
flags = {}
try:
name = play._task.name.split(' ')
except AttributeError as e:
return flags
for s in name:
f = s.split('=')
if len(f) == 2:
flags[f[0]] = f[1]
return flags
class CallbackModule(CallbackModule_default):
''' when pretty_output is in a task name, output both the stdout and the stderr to the console '''
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'stdout'
CALLBACK_NAME = 'improved'
CALLBACK_NEEDS_WHITELIST = False
def __init__(self):
self._play = None
self._last_task_banner = None
super(CallbackModule, self).__init__()
def _pretty(self, play):
self._display.display('{0}: [{1}]'.format(_status(play), play._host.name), color=_color(play))
stdout = play._result.get('stdout', '\n'.join(play._result.get('stdout_lines', [])))
if stdout:
self._display.display('--- stdout ---', color='blue')
self._display.display(stdout, color='blue')
stderr = play._result.get('stderr', '')
if stderr:
self._display.display('--- stderr ---', color='red')
self._display.display(stderr, color='red')
def _skip(self, play, **kwargs):
if play._result.get('skipped'):
return
task = sys._getframe(2).f_code.co_name
getattr(super(CallbackModule, self), task)(play, **kwargs)
def _output(self, play, **kwargs):
flags = _parse_flags(play)
if 'output' in flags and not self._display.verbosity:
try:
getattr(self, '_' + flags['output'])(play)
except AttributeError as e:
self._display.display(e, color='magenta')
task = sys._getframe(1).f_code.co_name
getattr(super(CallbackModule, self), task)(play, **kwargs)
else:
task = sys._getframe(1).f_code.co_name
getattr(super(CallbackModule, self), task)(play, **kwargs)
def v2_runner_on_failed(self, play, **kwargs):
self._output(play, **kwargs)
def v2_runner_on_ok(self, play, **kwargs):
self._output(play, **kwargs)
def v2_runner_on_skipped(self, play, **kwargs):
self._output(play, **kwargs)
def v2_runner_item_on_skipped(self, play, **kwargs):
self._output(play, **kwargs)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment