app-gen-tool/stx-app-generator/stx-app-generator/app_gen_tool/fluxcd.py

"""Module for generating FluxCD package."""
import os
import shutil
import subprocess
import yaml

from app_gen_tool.application import Application
from app_gen_tool.common import uppercase_name
from app_gen_tool import constants


class FluxCD(Application):
    """Class for generating FluxCD package based on Application class."""

    # Function to call all process fot the creation of the FluxCD app tarball
    # 1 - Validate input file and helm chart data
    # 2 - Create application directories
    # 3 - Generate FluxCD Manifests
    # 4 - Generate application plugins
    # 5 - Generate application metadata
    # 6 - Package helm-charts
    # 7 - Package plugins in wheel format
    # 8 - Generate checksum
    # 9 - Package entire application
    def generate(self, package_only, no_package):
        """Responsible for all the process for the creation of the FluxCD app tarball.

        Args:
            package_only (bool): Instructs the generator to only execute packaging related code.
            no_package (bool): Instrutcs the generator to only create the files without packaging.

        Returns:
            bool: False if a step in the generator fails
        """
        app_name_fixed = self._app['appName'].replace(' ', '_').replace('-', '_')
        updated_app_name = f'k8sapp_{app_name_fixed}'

        self._app['outputDir'] = self.output_folder
        self._app['outputFluxChartDir'] = os.path.join(
            self.output_folder, 'charts'
        )
        self._app['outputManifestDir'] = os.path.join(
            self.output_folder, 'fluxcd-manifests'
        )
        self._app['outputFluxBaseDir'] = os.path.join(
            self.output_folder, 'fluxcd-manifests', 'base'
        )

        self._app['outputPluginDir'] = os.path.join(
            self.output_folder, 'plugins'
        )
        self._app['outputHelmDir'] = os.path.join(
            self.output_folder, 'plugins', updated_app_name, 'helm'
        )
        self._app['outputCommonDir'] = os.path.join(
            self.output_folder, 'plugins', updated_app_name, 'common'
        )
        self._app['outputKustomizeDir'] = os.path.join(
            self.output_folder, 'plugins', updated_app_name, 'kustomize'
        )
        self._app['outputLifecycleDir'] = os.path.join(
            self.output_folder, 'plugins', updated_app_name, 'lifecycle'
        )

        # 0 - Print out helm version.
        self._print_helm_version()

        # 1 - Validate input file and helm chart data
        self.check_charts()

        if not package_only:
            # 2 - Create application directories
            self._create_flux_dir()
            self._create_plugins_dir()

            # 3 - Generate FluxCD Manifests
            ret = self._gen_fluxcd_manifest()
            if ret:
                print('FluxCD manifest generated!')
            else:
                print('FluxCD manifest generation failed!')
                return False

            # 4 - Generate application plugins
            ret = self._gen_plugins()
            if ret:
                print('FluxCD Plugins generated!')
            else:
                print('FluxCD Plugins generation failed!')
                return False

            # 5 - Generate application metadata
            ret = self._gen_metadata(self._app['outputDir'])
            if ret:
                print('FluxCD Metadata generated!')
            else:
                print('FluxCD Metadata generation failed!')
                return False

        if not no_package:
            # 6 - Package helm-charts
            for chart in self._chart:
                ret = self._gen_helm_chart_tarball(
                    chart, self._app['outputFluxChartDir']
                )
                if ret:
                    print(f'Helm chart {chart["name"]} tarball generated!')
                    print('')
                else:
                    print(f'Generating tarball for helm chart: {chart["name"]} error!')
                    return False

            # 7 - Package plugins in wheel format
            ret = self._gen_plugin_wheels()
            if ret:
                print('Plugin wheels generated!')
            else:
                print('Plugin wheels generation failed!')
                return False

            # 8 - Generate checksum &&
            # 9 - Package entire application
            ret = self._gen_checksum_and_app_tarball(self._app['outputDir'])
            if ret:
                print('Checksum generated!')
                print(
                    f'FluxCD App tarball generated at {self._app["outputDir"]}/{ret}'
                )
                print('')
            else:
                print('Checksum and App tarball generation failed!')
                return False

    def _create_flux_dir(self):
        """Sub-process of app generation. Create the folders for the FluxCD Manifest and charts."""
        if not os.path.exists(self._app['outputFluxChartDir']):
            os.makedirs(self._app['outputFluxChartDir'])
        if not os.path.exists(self._app['outputManifestDir']):
            os.makedirs(self._app['outputFluxBaseDir'])
            for idx in range(len(self._chart)):
                chart = self._chart[idx]
                self._app['outputFluxManifestDir'] = os.path.join(
                    self.output_folder, 'fluxcd-manifests', chart['name']
                )
                os.makedirs(self._app['outputFluxManifestDir'])

    def _gen_metadata(self, output_dir):
        """Sub-process of app generation. Append data in the application metadata.

        Args:
            output_dir (str): Path to the folder where the metadata file is placed.

        Returns:
            bool: True if data was successfully appended to metadata file. False otherwise.
        """
        # Gets the keys and values defined in the input yaml and writes the metadata.yaml app file.
        super()._gen_metadata(output_dir)
        metadata_file = os.path.join(output_dir, 'metadata.yaml')
        try:
            with open(metadata_file, 'a', encoding='utf-8') as f:
                f.write('\nhelm_repo: stx-platform\n')
                if self.metadata is not None:
                    yaml.safe_dump(self.metadata, f, sort_keys=False)
        except Exception:
            return False

        return True

    def _create_plugins_dir(self):
        """Sub-process of app generation. Create folders for the application plugins."""
        if not os.path.exists(self._app['outputPluginDir']):
            os.makedirs(self._app['outputPluginDir'])
        if not os.path.exists(self._app['outputHelmDir']):
            os.makedirs(self._app['outputHelmDir'])
        if not os.path.exists(self._app['outputCommonDir']):
            os.makedirs(self._app['outputCommonDir'])
        if not os.path.exists(self._app['outputKustomizeDir']):
            os.makedirs(self._app['outputKustomizeDir'])
        if not os.path.exists(self._app['outputLifecycleDir']):
            os.makedirs(self._app['outputLifecycleDir'])

    # Sub-process of app generation
    # generate application plugin files
    def _gen_plugins(self):
        """Sub-process of app generation. Generate application plugins files.

        Returns:
            bool: True if all files have been created with success. False otherwise.
        """
        plugin_dir = self._app['outputPluginDir']

        common_template = os.path.join(
            constants.APP_GEN_PY_PATH, constants.FLUXCD_COMMON_TEMPLATE
        )
        helm_template = os.path.join(
            constants.APP_GEN_PY_PATH, constants.FLUXCD_HELM_TEMPLATE
        )
        kustomize_template = os.path.join(
            constants.APP_GEN_PY_PATH, constants.FLUXCD_KUSTOMIZE_TEMPLATE
        )
        lifecycle_template = os.path.join(
            constants.APP_GEN_PY_PATH, constants.FLUXCD_LIFECYCLE_TEMPLATE
        )

        appname = 'k8sapp_' + self.APP_NAME_WITH_UNDERSCORE
        namespace = self._app['namespace']
        chart = self._chart
        name = self._chart[0]['name']

        # generate Common files
        try:
            with open(common_template, 'r', encoding='utf-8') as f:
                common_schema = f.read()
        except FileNotFoundError:
            print(f'File {common_template} not found')
            return False

        common_file = os.path.join(plugin_dir, appname, 'common', 'constants.py')

        with open(common_file, 'w', encoding='utf-8') as f:
            f.write(f'HELM_NS = "{namespace}"\nHELM_APP = "{appname}"\n')

        for idx in range(len(chart)):
            a_chart = chart[idx]
            name = a_chart['name']
            upper_name = uppercase_name(name)

            output = common_schema.format(upper=upper_name, name=name)
            with open(common_file, 'a', encoding='utf-8') as f:
                f.write(output)

        self.create_init_file(self._app['outputCommonDir'])

        # Generate Helm files
        try:
            with open(helm_template, 'r', encoding='utf-8') as f:
                helm_schema = f.read()
        except FileNotFoundError:
            print(f'File {helm_template} not found')
            return False

        for idx in range(len(chart)):
            a_chart = chart[idx]

            updated_chart_name = (
                a_chart['name'].replace(' ', '_').replace('-', '_') + '.py'
            )
            upper_name = uppercase_name(a_chart['name'])

            helm_file = os.path.join(plugin_dir, appname, 'helm', updated_chart_name)

            name = a_chart['name'].replace('-', ' ').title().replace(' ', '')
            namespace = a_chart['namespace']

            output = helm_schema.format(appname=appname, name=name, upper=upper_name)

            with open(helm_file, 'w', encoding='utf-8') as f:
                f.write(output)

        self.create_init_file(self._app['outputHelmDir'])

        # Generate Kustomize files
        try:
            with open(kustomize_template, 'r', encoding='utf-8') as f:
                kustomize_schema = f.read()
        except FileNotFoundError:
            print(f'File {kustomize_template} not found')
            return False

        kustomize_file_name = f'kustomie_{self.APP_NAME_WITH_UNDERSCORE}.py'
        kustomize_file = os.path.join(
            plugin_dir, appname, 'kustomize', kustomize_file_name
        )
        output = kustomize_schema.format(
            appname=appname, appnameStriped=self.APP_NAME_CAMEL_CASE
        )

        with open(kustomize_file, 'w', encoding='utf-8') as f:
            f.write(output)

        self.create_init_file(self._app['outputKustomizeDir'])

        # Generate Lifecycle files
        try:
            with open(lifecycle_template, 'r', encoding='utf-8') as f:
                lifecycle_schema = f.read()
        except FileNotFoundError:
            print(f'File {lifecycle_template} not found')
            return False

        lifecycle_file_name = f'lifecycle_{self.APP_NAME_WITH_UNDERSCORE}.py'
        lifecycle_file = os.path.join(
            plugin_dir, appname, 'lifecycle', lifecycle_file_name
        )
        output = lifecycle_schema.format(appnameStriped=self.APP_NAME_CAMEL_CASE)

        with open(lifecycle_file, 'w', encoding='utf-8') as f:
            f.write(output)

        self.create_init_file(self._app['outputLifecycleDir'])

        # Generate setup.py
        setup_py_file = plugin_dir + '/setup.py'
        setup_py_file = os.path.join(plugin_dir, 'setup.py')
        file_content = (
            'import setuptools\n\nsetuptools.setup(\n    '
            'setup_requires=["pbr>=2.0.0"],\n    pbr=True)'
        )

        with open(setup_py_file, 'w', encoding='utf-8') as f:
            f.write(file_content)
            f.close()

        # Generate setup.cfg file
        self.write_app_setup()

        self.create_init_file(plugin_dir)

        directory = os.path.join(plugin_dir, appname)
        self.create_init_file(directory)

        return True

    # Sub-process of app generation
    # generate plugin wheels
    #
    def _gen_plugin_wheels(self):
        """Sub-process of app generation. Package the plugins files in a wheels format.

        Returns:
            bool: True if packaging command executed without error. False otherwise.
        """
        dirplugins = self._app['outputPluginDir']

        store_cwd = os.getcwd()
        os.makedirs(dirplugins, exist_ok=True)
        os.chdir(dirplugins)

        command = [
            'python3',
            'setup.py',
            'bdist_wheel',
            '--universal',
            '-d',
            '.',
        ]

        try:
            subprocess.call(command, stderr=subprocess.STDOUT)
        except Exception:
            return False

        files = ['./ChangeLog', './AUTHORS']
        for file in files:
            if os.path.exists(file):
                os.remove(file)

        dirs = [
            './build/',
            f'./k8sapp_{self.APP_NAME_WITH_UNDERSCORE}.egg-info/',
        ]
        for dire in dirs:
            if os.path.exists(dire):
                shutil.rmtree(dire)

        for _, _, filenames in os.walk('./'):
            for filename in filenames:
                if filename[-4:] == ".whl":
                    os.chdir(store_cwd)
                    return True

        return False

    def _gen_fluxcd_manifest(self):  # pylint: disable=too-many-return-statements
        """Sub-process of app generation. Generate application fluxcd manifest files.

        Returns:
            bool: True if all files have been created with success. False otherwise.
        """
        # check manifest file existance
        fluxcd_dir = self._app['outputManifestDir']

        # update schema path to abspath
        kustomization_template = os.path.join(
            constants.APP_GEN_PY_PATH, constants.FLUXCD_KUSTOMIZATION_TEMPLATE
        )

        base_helmrepo_template = os.path.join(
            constants.APP_GEN_PY_PATH,
            constants.FLUXCD_BASE_TEMPLATES,
            'helmrepository.template',
        )

        base_kustom_template = os.path.join(
            constants.APP_GEN_PY_PATH,
            constants.FLUXCD_BASE_TEMPLATES,
            'kustomization.template',
        )

        base_namespace_template = os.path.join(
            constants.APP_GEN_PY_PATH,
            constants.FLUXCD_BASE_TEMPLATES,
            'namespace.template',
        )

        manifest_helmrelease_template = os.path.join(
            constants.APP_GEN_PY_PATH,
            constants.FLUXCD_MANIFEST_TEMPLATE,
            'helmrelease.template',
        )

        manifest_kustomization_template = os.path.join(
            constants.APP_GEN_PY_PATH,
            constants.FLUXCD_MANIFEST_TEMPLATE,
            'kustomization.template',
        )

        manifest = self._app
        chartgroup = self._listcharts
        chart = self._chart

        # generate kustomization file
        try:
            with open(kustomization_template, 'r', encoding='utf-8') as f:
                kustomization_schema = f.readlines()
        except FileNotFoundError:
            print(f'File {kustomization_template} not found')
            return False
        kustom_file = os.path.join(fluxcd_dir, 'kustomization.yaml')
        with open(kustom_file, 'a', encoding='utf-8') as f:
            # substitute values
            for line in kustomization_schema:
                self._write_values_and_blocks(f, line, chartgroup)

        # generate base/namespace file
        try:
            with open(base_namespace_template, 'r', encoding='utf-8') as f:
                base_namespace_schema = f.readlines()
        except FileNotFoundError:
            print(f'File {base_namespace_template} not found')
            return False
        base_namespace_file = os.path.join(fluxcd_dir, 'base', 'namespace.yaml')
        with open(base_namespace_file, 'a', encoding='utf-8') as f:
            # substitute values
            for line in base_namespace_schema:
                self._write_values_and_blocks(f, line, manifest)

        # generate base/kustomization file
        # generate base/helmrepository file
        # Both yaml files don't need to add informations from the input file
        try:
            with open(base_kustom_template, 'r', encoding='utf-8') as f:
                base_kustom_schema = f.readlines()
        except FileNotFoundError:
            print(f'File {base_kustom_template} not found')
            return False

        base_kustom_file = os.path.join(fluxcd_dir, 'base', 'kustomization.yaml')
        with open(base_kustom_file, 'a', encoding='utf-8') as f:
            for line in base_kustom_schema:
                out_line = line
                f.write(out_line)

        try:
            with open(base_helmrepo_template, 'r', encoding='utf-8') as f:
                base_helmrepo_schema = f.readlines()
        except FileNotFoundError:
            print(f'File {base_helmrepo_template} not found')
            return False

        base_helmrepo_file = os.path.join(fluxcd_dir, 'base', 'helmrepository.yaml')
        with open(base_helmrepo_file, 'a', encoding='utf-8') as f:
            for line in base_helmrepo_schema:
                out_line = line
                f.write(out_line)

        # iterate each fluxcd_chart for the generation of its fluxcd manifests
        for idx in range(len(chart)):
            a_chart = chart[idx]

            # generate manifest/helmrelease file
            try:
                with open(manifest_helmrelease_template, 'r', encoding='utf-8') as f:
                    manifest_helmrelease_schema = f.readlines()
            except FileNotFoundError:
                print(f'File {manifest_helmrelease_template} not found')
                return False

            manifest_helmrelease_file = os.path.join(
                fluxcd_dir, a_chart['name'], 'helmrelease.yaml'
            )

            with open(manifest_helmrelease_file, 'a', encoding='utf-8') as f:
                # fetch chart specific info
                for line in manifest_helmrelease_schema:
                    # substitute template values and blocks
                    self._write_values_and_blocks(f, line, a_chart)

            # generate manifest/kustomizaion file
            try:
                with open(manifest_kustomization_template, 'r', encoding='utf-8') as f:
                    manifest_kustomization_schema = f.readlines()
            except FileNotFoundError:
                print(f'File {manifest_kustomization_template} not found')
                return False

            manifest_kustomization_file = os.path.join(
                fluxcd_dir, a_chart['name'], 'kustomization.yaml'
            )

            with open(manifest_kustomization_file, 'a', encoding='utf-8') as f:
                # fetch chart specific info
                for line in manifest_kustomization_schema:
                    # substitute template values and blocks
                    self._write_values_and_blocks(f, line, a_chart)

            # file names
            system_overrides_name = f'{a_chart["name"]}-system-overrides.yaml'
            static_overrides_name = f'{a_chart["name"]}-static-overrides.yaml'

            # generate an empty manifest/system-overrides file
            system_override_file = os.path.join(
                fluxcd_dir, a_chart['name'], system_overrides_name
            )
            open(system_override_file, 'w', encoding='utf-8').close()

            # generate a manifest/static-overrides file
            static_override_file = os.path.join(
                fluxcd_dir, a_chart['name'], static_overrides_name
            )
            try:
                with open(static_override_file, mode='w', encoding="utf-8") as stream:
                    yaml.dump(a_chart['values'], stream, sort_keys=False)
            except yaml.YAMLError:
                print(f'Error while trying to write {static_override_file}')

        return True

    def create_init_file(self, app_path: str):
        """Sub-process of app generation. Create __init__.py file.

        Args:
            app_path (str): Path where the init file will be placed.
        """
        init_file = os.path.join(app_path, '__init__.py')
        open(init_file, 'w', encoding='utf-8').close()

    def _print_helm_version(self):
        """Print the available Helm version.

        Returns:
            bool: False if the command return code is different from 0.
        """
        print('Getting which version of helm is in use.')

        cmd_version = ['helm', 'version']
        subproc = subprocess.run(
            cmd_version,
            env=os.environ.copy(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if subproc.returncode == 0:
            print(str(subproc.stdout, encoding='utf-8'))
        else:
            print(str(subproc.stdout, encoding='utf-8'))
            print(str(subproc.stderr, encoding='utf-8'))
            return False

    def write_app_setup(self):
        """Sub-process of app generation. Write setup.cfg file."""
        def split_and_format_value(value) -> str:
            if isinstance(value, str):
                return ''.join([f'\t{lin}\n' for lin in value.split('\n')])
            else:
                return ''.join([f'\t{lin}\n' for lin in value])

        def expected_order(tup: tuple) -> int:
            if tup[0] == 'name':
                return 0
            elif tup[0] == 'summary':
                return 1
            return 2

        yml_data = self.plugin_setup
        yml_data['metadata']['name'] = f'k8sapp-{self.APP_NAME}'
        yml_data['metadata'][
            'summary'
        ] = f'StarlingX sysinv extensions for {self.APP_NAME}'
        yml_data['metadata'] = dict(
            sorted(yml_data['metadata'].items(), key=expected_order)
        )

        out = ''
        for label in yml_data:
            out += f'[{label}]\n'
            for key, val in yml_data[label].items():
                if label == 'metadata' and val is None:
                    raise ValueError(f'You should\'ve written a value for: {key}')
                elif not isinstance(val, list):
                    out += f'{key} = {val}\n'
                else:
                    out += f'{key} =\n'
                    out += split_and_format_value(val)
            out += '\n'
        charts_data = self._chart
        plugins_names = []
        for dic in charts_data:
            plugins_names.append(dic['name'])
        out += f'[files]\npackages =\n\tk8sapp_{self.APP_NAME_WITH_UNDERSCORE}\n\n'
        out += '[global]\nsetup-hooks =\n\tpbr.hooks.setup_hook\n\n'
        out += (
            '[entry_points]\nsystemconfig.helm_applications =\n\t'
            f'{self.APP_NAME} = systemconfig.helm_plugins.{self.APP_NAME_WITH_UNDERSCORE}\n\n'
            f'systemconfig.helm_plugins.{self.APP_NAME_WITH_UNDERSCORE} =\n'
        )

        for i, plug in enumerate(plugins_names):
            out += (
                f'\t{i + 1: 03d}_{plug} = k8sapp'
                f'_{self.APP_NAME_WITH_UNDERSCORE}.'
                f'helm.{plug.replace("-", "_")}'
            )

            out += f':{plug.replace("-", " ").title().replace(" ", "")}Helm\n'

        out += '\n'
        out += (
            'systemconfig.fluxcd.kustomize_ops =\n'
            f'\t{self.APP_NAME} = k8sapp_{self.APP_NAME_WITH_UNDERSCORE}.kustomize.kustomize_'
            f'{self.APP_NAME_WITH_UNDERSCORE}:{self.APP_NAME_CAMEL_CASE}'
            'FluxCDKustomizeOperator\n\n'
            'systemconfig.app_lifecycle =\n'
            f'\t{self.APP_NAME} = k8sapp_{self.APP_NAME_WITH_UNDERSCORE}.lifecycle.lifecycle_'
            f'{self.APP_NAME_WITH_UNDERSCORE}:{self.APP_NAME_CAMEL_CASE}AppLifecycleOperator\n\n'
        )
        out += '[bdist_wheel]\nuniversal = 1'
        with open(
            f'{self._app["outputPluginDir"]}/setup.cfg', 'w+', encoding='utf-8'
        ) as f:
            f.write(out)