openstack
/
tripleo-validations
Code Issues Proposed changes
Browse Source

Remove callback plugin from tripleo-validation 

validations-common now owns the callback plugin for
validation, so those files need to be removed

Change-Id: I83b26e81eaf18cc2f94bf990a8455f2de1aacf4a
changes/70/753270/6
Mathieu Bultel 1 year ago
parent
acefed99f0
commit
3cae7749e8
4 changed files with 0 additions and 436 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 0
      callback_plugins/.keep
  2. 31
      callback_plugins/fail_if_no_hosts.py
  3. 200
      callback_plugins/validation_json.py
  4. 205
      callback_plugins/validation_output.py

0
callback_plugins/.keep
View File

31
callback_plugins/fail_if_no_hosts.py
View File

@ -1,31 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 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 sys
from ansible.plugins.callback import CallbackBase
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_NAME = 'fail_if_no_hosts'
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
def v2_playbook_on_stats(self, stats):
if len(stats.processed.keys()) == 0:
sys.exit(10)

200
callback_plugins/validation_json.py
View File

@ -1,200 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 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.
__metaclass__ = type
import datetime
import json
import time
import os
from functools import partial
from ansible.module_utils.six.moves import reduce
from ansible.parsing.ajson import AnsibleJSONEncoder
from ansible.plugins.callback import CallbackBase
DOCUMENTATION = '''
callback: json
short_description: Ansible screen output as JSON file
version_added: "1.0"
description:
- This callback converts all events into a JSON file
stored in /var/log/validations
type: stdout
requirements: None
'''
VALIDATIONS_LOG_DIR = "/var/log/validations"
def current_time():
return '%sZ' % datetime.datetime.utcnow().isoformat()
def secondsToStr(t):
def rediv(ll, b):
return list(divmod(ll[0], b)) + ll[1:]
return "%d:%02d:%02d.%03d" % tuple(
reduce(rediv, [[
t * 1000,
], 1000, 60, 60]))
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'stdout'
CALLBACK_NAME = 'validation_json'
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
self.results = []
self.simple_results = []
self.env = {}
self.t0 = None
self.current_time = current_time()
def _new_play(self, play):
return {
'play': {
'host': play.get_name(),
'validation_id': self.env['playbook_name'],
'validation_path': self.env['playbook_path'],
'id': (os.getenv('ANSIBLE_UUID') if os.getenv('ANSIBLE_UUID')
else str(play._uuid)),
'duration': {
'start': current_time()
}
},
'tasks': []
}
def _new_task(self, task):
return {
'task': {
'name': task.get_name(),
'id': str(task._uuid),
'duration': {
'start': current_time()
}
},
'hosts': {}
}
def _val_task(self, task_name):
return {
'task': {
'name': task_name,
'hosts': {}
}
}
def _val_task_host(self, task_name):
return {
'task': {
'name': task_name,
'hosts': {}
}
}
def v2_playbook_on_start(self, playbook):
self.t0 = time.time()
pl = playbook._file_name
validation_id = os.path.splitext(os.path.basename(pl))[0]
self.env = {
"playbook_name": validation_id,
"playbook_path": playbook._basedir
}
def v2_playbook_on_play_start(self, play):
self.results.append(self._new_play(play))
def v2_playbook_on_task_start(self, task, is_conditional):
self.results[-1]['tasks'].append(self._new_task(task))
def v2_playbook_on_handler_task_start(self, task):
self.results[-1]['tasks'].append(self._new_task(task))
def v2_playbook_on_stats(self, stats):
"""Display info about playbook statistics"""
hosts = sorted(stats.processed.keys())
summary = {}
for h in hosts:
s = stats.summarize(h)
summary[h] = s
output = {
'plays': self.results,
'stats': summary,
'validation_output': self.simple_results
}
log_file = "{}/{}_{}_{}.json".format(
VALIDATIONS_LOG_DIR,
(os.getenv('ANSIBLE_UUID') if os.getenv('ANSIBLE_UUID') else
self.results[0].get('play').get('id')),
self.env['playbook_name'],
self.current_time)
with open(log_file, 'w') as js:
js.write(json.dumps(output,
cls=AnsibleJSONEncoder,
indent=4,
sort_keys=True))
def _record_task_result(self, on_info, result, **kwargs):
"""This function is used as a partial to add
failed/skipped info in a single method
"""
host = result._host
task = result._task
task_result = result._result.copy()
task_result.update(on_info)
task_result['action'] = task.action
self.results[-1]['tasks'][-1]['hosts'][host.name] = task_result
if 'failed' in task_result.keys():
self.simple_results.append(self._val_task(task.name))
self.simple_results[-1]['task']['status'] = "FAILED"
self.simple_results[-1]['task']['hosts'][host.name] = task_result
if 'warnings' in task_result.keys():
self.simple_results.append(self._val_task(task.name))
self.simple_results[-1]['task']['status'] = "WARNING"
self.simple_results[-1]['task']['hosts'][host.name] = task_result
end_time = current_time()
time_elapsed = secondsToStr(time.time() - self.t0)
self.results[-1]['tasks'][-1]['task']['duration']['end'] = end_time
self.results[-1]['play']['duration']['end'] = end_time
self.results[-1]['play']['duration']['time_elapsed'] = time_elapsed
def __getattribute__(self, name):
"""Return ``_record_task_result`` partial with a dict
containing skipped/failed if necessary
"""
if name not in ('v2_runner_on_ok', 'v2_runner_on_failed',
'v2_runner_on_unreachable', 'v2_runner_on_skipped'):
return object.__getattribute__(self, name)
on = name.rsplit('_', 1)[1]
on_info = {}
if on in ('failed', 'skipped'):
on_info[on] = True
return partial(self._record_task_result, on_info)

205
callback_plugins/validation_output.py
View File

@ -1,205 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 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 pprint
from ansible import constants as C
from ansible.plugins.callback import CallbackBase
FAILURE_TEMPLATE = """\
Task '{}' failed:
Host: {}
Message: {}
"""
WARNING_TEMPLATE = """\
Task '{}' succeeded, but had some warnings:
Host: {}
Warnings: {}
"""
DEBUG_TEMPLATE = """\
Task: Debug
Host: {}
{}
"""
def indent(text):
'''Indent the given text by four spaces.'''
return ''.join(' {}\n'.format(line) for line in text.splitlines())
# TODO(shadower): test with async settings
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'stdout'
CALLBACK_NAME = 'validation_output'
def __init__(self, display=None):
super(CallbackModule, self).__init__(display)
def print_failure_message(self, host_name, task_name, results,
abridged_result):
'''Print a human-readable error info from Ansible result dictionary.'''
def is_script(results):
return ('rc' in results and 'invocation' in results
and 'script' in results._task_fields['action']
and '_raw_params' in results._task_fields['args'])
display_full_results = False
if 'rc' in results and 'cmd' in results:
command = results['cmd']
# The command can be either a list or a string.
# Concat if it's a list:
if type(command) == list:
command = " ".join(results['cmd'])
message = "Command `{}` exited with code: {}".format(
command, results['rc'])
# There may be an optional message attached to the command.
# Display it:
if 'msg' in results:
message = message + ": " + results['msg']
elif is_script(results):
script_name = results['invocation']['module_args']['_raw_params']
message = "Script `{}` exited with code: {}".format(
script_name, results['rc'])
elif 'msg' in results:
message = results['msg']
else:
message = "Unknown error"
display_full_results = True
self._display.display(
FAILURE_TEMPLATE.format(task_name, host_name, message),
color=C.COLOR_ERROR)
stdout = results.get('module_stdout', results.get('stdout', ''))
if stdout:
print('stdout:')
self._display.display(indent(stdout), color=C.COLOR_ERROR)
stderr = results.get('module_stderr', results.get('stderr', ''))
if stderr:
print('stderr:')
self._display.display(indent(stderr), color=C.COLOR_ERROR)
if display_full_results:
print(
"Could not get an error message. Here is the Ansible output:")
pprint.pprint(abridged_result, indent=4)
warnings = results.get('warnings', [])
if warnings:
print("Warnings:")
for warning in warnings:
self._display.display("* %s " % warning, color=C.COLOR_WARN)
print("")
def v2_playbook_on_play_start(self, play):
pass # No need to notify that a play started
def v2_playbook_on_task_start(self, task, is_conditional):
pass # No need to notify that a task started
def v2_runner_on_ok(self, result, **kwargs):
host_name = result._host
task_name = result._task.get_name()
task_fields = result._task_fields
results = result._result # A dict of the module name etc.
self._dump_results(results)
warnings = results.get('warnings', [])
# Print only tasks that produced some warnings:
if warnings:
for warning in warnings:
warn_msg = "{}\n".format(warning)
self._display.display(WARNING_TEMPLATE.format(task_name,
host_name,
warn_msg),
color=C.COLOR_WARN)
if 'debug' in task_fields['action']:
output = ""
if 'var' in task_fields['args']:
variable = task_fields['args']['var']
value = results[variable]
output = "{}: {}".format(variable, str(value))
elif 'msg' in task_fields['args']:
output = "Message: {}".format(
task_fields['args']['msg'])
self._display.display(DEBUG_TEMPLATE.format(host_name, output),
color=C.COLOR_OK)
def v2_runner_on_failed(self, result, **kwargs):
host_name = result._host
task_name = result._task.get_name()
result_dict = result._result # A dict of the module name etc.
abridged_result = self._dump_results(result_dict)
if 'results' in result_dict:
# The task is a list of items under `results`
for item in result_dict['results']:
if item.get('failed', False):
self.print_failure_message(host_name, task_name,
item, item)
else:
# The task is a "normal" module invocation
self.print_failure_message(host_name, task_name, result_dict,
abridged_result)
def v2_runner_on_skipped(self, result, **kwargs):
pass # No need to print skipped tasks
def v2_runner_on_unreachable(self, result, **kwargs):
host_name = result._host
task_name = result._task.get_name()
results = {'msg': 'The host is unreachable.'}
self.print_failure_message(host_name, task_name, results, results)
def v2_playbook_on_stats(self, stats):
def failed(host):
_failures = stats.summarize(host).get('failures', 0) > 0
_unreachable = stats.summarize(host).get('unreachable', 0) > 0
return (_failures or _unreachable)
hosts = sorted(stats.processed.keys())
failed_hosts = [host for host in hosts if failed(host)]
if hosts:
if failed_hosts:
if len(failed_hosts) == len(hosts):
print("Failure! The validation failed for all hosts:")
for failed_host in failed_hosts:
self._display.display("* %s" % failed_host,
color=C.COLOR_ERROR)
else:
print("Failure! The validation failed for hosts:")
for failed_host in failed_hosts:
self._display.display("* %s" % failed_host,
color=C.COLOR_ERROR)
print("and passed for hosts:")
for host in [h for h in hosts if h not in failed_hosts]:
self._display.display("* %s" % host,
color=C.COLOR_OK)
else:
print("Success! The validation passed for all hosts:")
for host in hosts:
self._display.display("* %s" % host,
color=C.COLOR_OK)
else:
print("Warning! The validation did not run on any host.")
Write Preview
Loading…
Cancel
Save