mirror of
https://github.com/ansible-collections/kubernetes.core.git
synced 2026-08-02 12:54:48 +00:00
action/k8s_info: refactoring to simplify the code base (#331)
Break down the long `run()` method to simplify the readability of the plugin.
This commit is contained in:
Gonéri Le Bouder
GitHub
@@ -8,6 +8,8 @@ __metaclass__ = type
|
||||
|
||||
import copy
|
||||
import traceback
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
from ansible.config.manager import ensure_type
|
||||
from ansible.errors import AnsibleError, AnsibleFileNotFound, AnsibleAction, AnsibleActionFail
|
||||
@@ -34,63 +36,35 @@ class ActionModule(ActionBase):
|
||||
|
||||
return result
|
||||
|
||||
def run(self, tmp=None, task_vars=None):
|
||||
''' handler for k8s options '''
|
||||
if task_vars is None:
|
||||
task_vars = dict()
|
||||
|
||||
result = super(ActionModule, self).run(tmp, task_vars)
|
||||
del tmp # tmp no longer has any effect
|
||||
|
||||
# Check current transport connection and depending upon
|
||||
# look for kubeconfig and src
|
||||
# 'local' => look files on Ansible Controller
|
||||
# Transport other than 'local' => look files on remote node
|
||||
remote_transport = self._connection.transport != 'local'
|
||||
|
||||
new_module_args = copy.deepcopy(self._task.args)
|
||||
|
||||
kubeconfig = self._task.args.get('kubeconfig', None)
|
||||
# find the kubeconfig in the expected search path
|
||||
if kubeconfig and not remote_transport:
|
||||
# kubeconfig is local
|
||||
@contextmanager
|
||||
def get_template_data(self, template_path):
|
||||
try:
|
||||
# find in expected paths
|
||||
kubeconfig = self._find_needle('files', kubeconfig)
|
||||
source = self._find_needle('templates', template_path)
|
||||
except AnsibleError as e:
|
||||
result['failed'] = True
|
||||
result['msg'] = to_text(e)
|
||||
result['exception'] = traceback.format_exc()
|
||||
return result
|
||||
raise AnsibleActionFail(to_text(e))
|
||||
|
||||
# decrypt kubeconfig found
|
||||
actual_file = self._loader.get_real_file(kubeconfig, decrypt=True)
|
||||
new_module_args['kubeconfig'] = actual_file
|
||||
|
||||
# find the file in the expected search path
|
||||
src = self._task.args.get('src', None)
|
||||
|
||||
if src:
|
||||
if remote_transport:
|
||||
# src is on remote node
|
||||
result.update(self._execute_module(module_name=self._task.action, task_vars=task_vars))
|
||||
return self._ensure_invocation(result)
|
||||
|
||||
# src is local
|
||||
# Get vault decrypted tmp file
|
||||
try:
|
||||
# find in expected paths
|
||||
src = self._find_needle('files', src)
|
||||
except AnsibleError as e:
|
||||
result['failed'] = True
|
||||
result['msg'] = to_text(e)
|
||||
result['exception'] = traceback.format_exc()
|
||||
return result
|
||||
tmp_source = self._loader.get_real_file(source)
|
||||
except AnsibleFileNotFound as e:
|
||||
raise AnsibleActionFail("could not find template=%s, %s" % (source, to_text(e)))
|
||||
b_tmp_source = to_bytes(tmp_source, errors='surrogate_or_strict')
|
||||
|
||||
if src:
|
||||
new_module_args['src'] = src
|
||||
try:
|
||||
with open(b_tmp_source, 'rb') as f:
|
||||
try:
|
||||
template_data = to_text(f.read(), errors='surrogate_or_strict')
|
||||
except UnicodeError:
|
||||
raise AnsibleActionFail("Template source files must be utf-8 encoded")
|
||||
yield template_data
|
||||
except AnsibleAction:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise AnsibleActionFail("%s: %s" % (type(e).__name__, to_text(e)))
|
||||
finally:
|
||||
self._loader.cleanup_tmp_file(b_tmp_source)
|
||||
|
||||
template = self._task.args.get('template', None)
|
||||
if template:
|
||||
def load_template(self, template, new_module_args, task_vars):
|
||||
# template is only supported by k8s module.
|
||||
if self._task.action not in ('k8s', 'community.kubernetes.k8s', 'community.okd.k8s'):
|
||||
raise AnsibleActionFail("'template' is only supported parameter for 'k8s' module.")
|
||||
@@ -153,26 +127,8 @@ class ActionModule(ActionBase):
|
||||
elif newline_sequence not in allowed_sequences:
|
||||
raise AnsibleActionFail("newline_sequence needs to be one of: \n, \r or \r\n")
|
||||
|
||||
try:
|
||||
source = self._find_needle('templates', template_path)
|
||||
except AnsibleError as e:
|
||||
raise AnsibleActionFail(to_text(e))
|
||||
|
||||
# Get vault decrypted tmp file
|
||||
try:
|
||||
tmp_source = self._loader.get_real_file(source)
|
||||
except AnsibleFileNotFound as e:
|
||||
raise AnsibleActionFail("could not find template=%s, %s" % (source, to_text(e)))
|
||||
b_tmp_source = to_bytes(tmp_source, errors='surrogate_or_strict')
|
||||
|
||||
# template the source data locally & get ready to transfer
|
||||
try:
|
||||
with open(b_tmp_source, 'rb') as f:
|
||||
try:
|
||||
template_data = to_text(f.read(), errors='surrogate_or_strict')
|
||||
except UnicodeError:
|
||||
raise AnsibleActionFail("Template source files must be utf-8 encoded")
|
||||
|
||||
with self.get_template_data(template_path) as template_data:
|
||||
# add ansible 'template' vars
|
||||
temp_vars = task_vars.copy()
|
||||
old_vars = self._templar.available_variables
|
||||
@@ -195,12 +151,65 @@ class ActionModule(ActionBase):
|
||||
if not resource_definition:
|
||||
new_module_args.pop('template')
|
||||
new_module_args['definition'] = resultant
|
||||
except AnsibleAction:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise AnsibleActionFail("%s: %s" % (type(e).__name__, to_text(e)))
|
||||
finally:
|
||||
self._loader.cleanup_tmp_file(b_tmp_source)
|
||||
|
||||
def run(self, tmp=None, task_vars=None):
|
||||
''' handler for k8s options '''
|
||||
if task_vars is None:
|
||||
task_vars = dict()
|
||||
|
||||
result = super(ActionModule, self).run(tmp, task_vars)
|
||||
del tmp # tmp no longer has any effect
|
||||
|
||||
# Check current transport connection and depending upon
|
||||
# look for kubeconfig and src
|
||||
# 'local' => look files on Ansible Controller
|
||||
# Transport other than 'local' => look files on remote node
|
||||
remote_transport = self._connection.transport != 'local'
|
||||
|
||||
new_module_args = copy.deepcopy(self._task.args)
|
||||
|
||||
kubeconfig = self._task.args.get('kubeconfig', None)
|
||||
# find the kubeconfig in the expected search path
|
||||
if kubeconfig and not remote_transport:
|
||||
# kubeconfig is local
|
||||
try:
|
||||
# find in expected paths
|
||||
kubeconfig = self._find_needle('files', kubeconfig)
|
||||
except AnsibleError as e:
|
||||
result['failed'] = True
|
||||
result['msg'] = to_text(e)
|
||||
result['exception'] = traceback.format_exc()
|
||||
return result
|
||||
|
||||
# decrypt kubeconfig found
|
||||
actual_file = self._loader.get_real_file(kubeconfig, decrypt=True)
|
||||
new_module_args['kubeconfig'] = actual_file
|
||||
|
||||
# find the file in the expected search path
|
||||
src = self._task.args.get('src', None)
|
||||
|
||||
if src:
|
||||
if remote_transport:
|
||||
# src is on remote node
|
||||
result.update(self._execute_module(module_name=self._task.action, task_vars=task_vars))
|
||||
return self._ensure_invocation(result)
|
||||
|
||||
# src is local
|
||||
try:
|
||||
# find in expected paths
|
||||
src = self._find_needle('files', src)
|
||||
except AnsibleError as e:
|
||||
result['failed'] = True
|
||||
result['msg'] = to_text(e)
|
||||
result['exception'] = traceback.format_exc()
|
||||
return result
|
||||
|
||||
if src:
|
||||
new_module_args['src'] = src
|
||||
|
||||
template = self._task.args.get('template', None)
|
||||
if template:
|
||||
self.load_template(template, new_module_args, task_vars)
|
||||
|
||||
# Execute the k8s_* module.
|
||||
module_return = self._execute_module(module_name=self._task.action, module_args=new_module_args, task_vars=task_vars)
|
||||
|
||||
Reference in New Issue
Block a user