kubernetes.core/plugins/module_utils/helm.py

# -*- coding: utf-8 -*-
# Copyright: (c) 2020, Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import absolute_import, division, print_function

__metaclass__ = type


from contextlib import contextmanager
import os
import tempfile
import traceback
import re

from ansible.module_utils.basic import missing_required_lib


try:
    import yaml

    HAS_YAML = True
except ImportError:
    YAML_IMP_ERR = traceback.format_exc()
    HAS_YAML = False


@contextmanager
def prepare_helm_environ_update(module):
    environ_update = {}
    file_to_cleam_up = None
    kubeconfig_path = module.params.get("kubeconfig")
    if module.params.get("context") is not None:
        environ_update["HELM_KUBECONTEXT"] = module.params.get("context")
    if module.params.get("release_namespace"):
        environ_update["HELM_NAMESPACE"] = module.params.get("release_namespace")
    if module.params.get("api_key"):
        environ_update["HELM_KUBETOKEN"] = module.params["api_key"]
    if module.params.get("host"):
        environ_update["HELM_KUBEAPISERVER"] = module.params["host"]
    if module.params.get("validate_certs") is False or module.params.get("ca_cert"):
        kubeconfig_path = write_temp_kubeconfig(
            module.params["host"],
            validate_certs=module.params["validate_certs"],
            ca_cert=module.params["ca_cert"],
        )
        file_to_cleam_up = kubeconfig_path
    if kubeconfig_path is not None:
        environ_update["KUBECONFIG"] = kubeconfig_path

    try:
        yield environ_update
    finally:
        if file_to_cleam_up:
            os.remove(file_to_cleam_up)


def run_helm(module, command, fails_on_error=True):
    if not HAS_YAML:
        module.fail_json(msg=missing_required_lib("PyYAML"), exception=YAML_IMP_ERR)

    with prepare_helm_environ_update(module) as environ_update:
        rc, out, err = module.run_command(command, environ_update=environ_update)
    if fails_on_error and rc != 0:
        module.fail_json(
            msg="Failure when executing Helm command. Exited {0}.\nstdout: {1}\nstderr: {2}".format(
                rc, out, err
            ),
            stdout=out,
            stderr=err,
            command=command,
        )
    return rc, out, err


def get_values(module, command, release_name):
    """
    Get Values from deployed release
    """
    if not HAS_YAML:
        module.fail_json(msg=missing_required_lib("PyYAML"), exception=YAML_IMP_ERR)

    get_command = command + " get values --output=yaml " + release_name

    rc, out, err = run_helm(module, get_command)
    # Helm 3 return "null" string when no values are set
    if out.rstrip("\n") == "null":
        return {}
    return yaml.safe_load(out)


def write_temp_kubeconfig(server, validate_certs=True, ca_cert=None):
    # Workaround until https://github.com/helm/helm/pull/8622 is merged
    content = {
        "apiVersion": "v1",
        "kind": "Config",
        "clusters": [{"cluster": {"server": server}, "name": "generated-cluster"}],
        "contexts": [
            {"context": {"cluster": "generated-cluster"}, "name": "generated-context"}
        ],
        "current-context": "generated-context",
    }

    if not validate_certs:
        content["clusters"][0]["cluster"]["insecure-skip-tls-verify"] = True
    if ca_cert:
        content["clusters"][0]["cluster"]["certificate-authority"] = ca_cert

    _fd, file_name = tempfile.mkstemp()
    with os.fdopen(_fd, "w") as fp:
        yaml.dump(content, fp)
    return file_name


def get_helm_plugin_list(module, helm_bin=None):
    """
    Return `helm plugin list`
    """
    if not helm_bin:
        return []
    helm_plugin_list = helm_bin + " list"
    rc, out, err = run_helm(module, helm_plugin_list)
    if rc != 0 or (out == "" and err == ""):
        module.fail_json(
            msg="Failed to get Helm plugin info",
            command=helm_plugin_list,
            stdout=out,
            stderr=err,
            rc=rc,
        )
    return (rc, out, err)


def parse_helm_plugin_list(module, output=None):
    """
    Parse `helm plugin list`, return list of plugins
    """
    ret = []
    if not output:
        return ret

    for line in output:
        if line.startswith("NAME"):
            continue
        name, version, description = line.split("\t", 3)
        name = name.strip()
        version = version.strip()
        description = description.strip()
        if name == "":
            continue
        ret.append((name, version, description))

    return ret


def get_helm_version(module, helm_bin):

    helm_version_command = helm_bin + " version"
    rc, out, err = module.run_command(helm_version_command)
    if rc == 0:
        m = re.match(r'version.BuildInfo{Version:"v([0-9\.]*)",', out)
        if m:
            return m.group(1)
    return None