osm_lcm/odu_libs/cluster_mgmt.py

#######################################################################################
# Copyright ETSI Contributors and Others.
#
# 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.
#######################################################################################


from pyrage import x25519
import yaml
import base64


def gather_age_key(cluster):
    pubkey = cluster.get("age_pubkey")
    privkey = cluster.get("age_privkey")
    # return both public and private key
    return pubkey, privkey


def generate_age_key():
    ident = x25519.Identity.generate()
    # gets the public key
    pubkey = ident.to_public()
    # gets the private key
    privkey = str(ident)
    # return both public and private key
    return pubkey, privkey


async def create_cluster(self, op_id, op_params, content, bootstrap_only=False):
    self.logger.info("Create cluster workflow Enter")
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")

    db_cluster = content["cluster"]
    db_vim_account = content["vim_account"]

    # workflow_template = "launcher-create-aks-cluster-and-bootstrap.j2"
    workflow_template = "launcher-create-crossplane-cluster-and-bootstrap.j2"
    workflow_name = f"create-cluster-{db_cluster['_id']}"
    # cluster_name = db_cluster["name"].lower()
    cluster_name = db_cluster["git_name"].lower()

    # Generate age key
    # public_key_new_cluster, private_key_new_cluster = generate_age_key()
    # Get age key
    public_key_new_cluster, private_key_new_cluster = gather_age_key(db_cluster)
    self.logger.debug(f"public_key_new_cluster={public_key_new_cluster}")
    self.logger.debug(f"private_key_new_cluster={private_key_new_cluster}")

    # Test kubectl connection
    self.logger.debug(self._kubectl._get_kubectl_version())

    # Create secret with agekey
    secret_name = f"secret-age-{cluster_name}"
    secret_namespace = "osm-workflows"
    secret_key = "agekey"
    secret_value = private_key_new_cluster
    await self.create_secret(
        secret_name,
        secret_namespace,
        secret_key,
        secret_value,
    )

    # Additional params for the workflow
    cluster_kustomization_name = cluster_name
    osm_project_name = "osm_admin"  # TODO: get project name from content
    if bootstrap_only:
        cluster_type = ""
        providerconfig_name = ""
    else:
        vim_account_id = db_cluster["vim_account"]
        providerconfig_name = f"{vim_account_id}-config"
        vim_type = db_vim_account["vim_type"]
        if vim_type == "azure":
            cluster_type = "aks"
        elif vim_type == "aws":
            cluster_type = "eks"
        elif vim_type == "gcp":
            cluster_type = "gke"
        else:
            raise Exception("Not suitable VIM account to register cluster")

    # Render workflow
    # workflow_kwargs = {
    #     "git_fleet_url": f"{self._repo_base_url}/{self._repo_user}/fleet-osm.git",
    #     "git_sw_catalogs_url": f"{self._repo_base_url}/{self._repo_user}/sw-catalogs-osm.git",
    # }
    # manifest = self.render_jinja_template(
    #     workflow_template,
    #     output_file=None,
    #     **workflow_kwargs
    # )
    manifest = self.render_jinja_template(
        workflow_template,
        output_file=None,
        workflow_name=workflow_name,
        git_fleet_url=f"{self._repo_base_url}/{self._repo_user}/fleet-osm.git",
        git_sw_catalogs_url=f"{self._repo_base_url}/{self._repo_user}/sw-catalogs-osm.git",
        cluster_name=cluster_name,
        cluster_type=cluster_type,
        cluster_kustomization_name=cluster_kustomization_name,
        providerconfig_name=providerconfig_name,
        public_key_mgmt=self._pubkey,
        public_key_new_cluster=public_key_new_cluster,
        secret_name_private_key_new_cluster=secret_name,
        vm_size=db_cluster["node_size"],
        node_count=db_cluster["node_count"],
        k8s_version=db_cluster["k8s_version"],
        cluster_location=db_cluster["region_name"],
        osm_project_name=osm_project_name,
        rg_name=db_cluster["resource_group"],
        workflow_debug=self._workflow_debug,
        workflow_dry_run=self._workflow_dry_run,
    )
    self.logger.debug(f"Workflow manifest: {manifest}")

    # Submit workflow
    self._kubectl.create_generic_object(
        namespace="osm-workflows",
        manifest_dict=yaml.safe_load(manifest),
        api_group="argoproj.io",
        api_plural="workflows",
        api_version="v1alpha1",
    )
    return workflow_name

    # self.logger.info(f"Deleting secret {secret_name} in namespace {secret_namespace} ...")
    # self._kubectl.delete_secret(name=secret_name, namespace=secret_namespace)
    # self.logger.info("DONE")

    # self.logger.info(f"Listing secrets in namespace {secret_namespace} ...")
    # secret_list = self._kubectl.get_secrets(secret_namespace)
    # # print(secret_list)
    # for item in secret_list:
    #     print(item.metadata.name)
    # self.logger.info("DONE")

    # self.logger.info(f"Deleting secrets in namespace {secret_namespace} ...")
    # for item in secret_list:
    #     print(f"Deleting {item.metadata.name} ...")
    #     self._kubectl.delete_secret(
    #         name=item.metadata.name,
    #         namespace=secret_namespace,
    #     )
    #     self.logger.info("DELETED")
    # self.logger.info("DONE")


async def update_cluster(self, op_id, op_params, content):
    self.logger.info("Update cluster eks workflow Enter")
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")

    db_cluster = content["cluster"]
    db_vim_account = content["vim_account"]

    workflow_template = "launcher-update-crossplane-cluster.j2"
    workflow_name = f"delete-cluster-{db_cluster['_id']}"
    # cluster_name = db_cluster["name"].lower()
    cluster_name = db_cluster["git_name"].lower()

    # Get age key
    public_key_cluster, private_key_cluster = gather_age_key(db_cluster)
    self.logger.debug(f"public_key_new_cluster={public_key_cluster}")
    self.logger.debug(f"private_key_new_cluster={private_key_cluster}")

    # Create secret with agekey
    secret_name = f"secret-age-{cluster_name}"
    secret_namespace = "osm-workflows"
    secret_key = "agekey"
    secret_value = private_key_cluster
    await self.create_secret(
        secret_name,
        secret_namespace,
        secret_key,
        secret_value,
    )

    # Additional params for the workflow
    cluster_kustomization_name = cluster_name
    osm_project_name = "osm_admin"  # TODO: get project name from db_cluster
    vim_account_id = db_cluster["vim_account"]
    providerconfig_name = f"{vim_account_id}-config"
    vim_type = db_vim_account["vim_type"]
    if vim_type == "azure":
        cluster_type = "aks"
    elif vim_type == "aws":
        cluster_type = "eks"
    elif vim_type == "gcp":
        cluster_type = "gke"
    else:
        raise Exception("Not suitable VIM account to update cluster")

    # Render workflow
    manifest = self.render_jinja_template(
        workflow_template,
        output_file=None,
        workflow_name=workflow_name,
        git_fleet_url=f"{self._repo_base_url}/{self._repo_user}/fleet-osm.git",
        git_sw_catalogs_url=f"{self._repo_base_url}/{self._repo_user}/sw-catalogs-osm.git",
        cluster_name=cluster_name,
        cluster_type=cluster_type,
        cluster_kustomization_name=cluster_kustomization_name,
        providerconfig_name=providerconfig_name,
        public_key_mgmt=self._pubkey,
        public_key_new_cluster=public_key_cluster,
        secret_name_private_key_new_cluster=secret_name,
        vm_size=db_cluster["node_size"],
        node_count=db_cluster["node_count"],
        k8s_version=db_cluster["k8s_version"],
        cluster_location=db_cluster["region_name"],
        osm_project_name=osm_project_name,
        workflow_debug=self._workflow_debug,
        workflow_dry_run=self._workflow_dry_run,
    )
    self.logger.info(manifest)

    # Submit workflow
    self._kubectl.create_generic_object(
        namespace="osm-workflows",
        manifest_dict=yaml.safe_load(manifest),
        api_group="argoproj.io",
        api_plural="workflows",
        api_version="v1alpha1",
    )
    return workflow_name


async def delete_cluster(self, op_id, op_params, content):
    self.logger.info("Delete cluster workflow Enter")
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")

    db_cluster = content["cluster"]

    workflow_template = "launcher-delete-cluster.j2"
    workflow_name = f"delete-cluster-{db_cluster['_id']}"
    # cluster_name = db_cluster["name"].lower()
    cluster_name = db_cluster["git_name"].lower()

    # Additional params for the workflow
    cluster_kustomization_name = cluster_name
    osm_project_name = "osm_admin"  # TODO: get project name from DB

    # Render workflow
    manifest = self.render_jinja_template(
        workflow_template,
        output_file=None,
        workflow_name=workflow_name,
        git_fleet_url=f"{self._repo_base_url}/{self._repo_user}/fleet-osm.git",
        git_sw_catalogs_url=f"{self._repo_base_url}/{self._repo_user}/sw-catalogs-osm.git",
        cluster_name=cluster_name,
        cluster_kustomization_name=cluster_kustomization_name,
        osm_project_name=osm_project_name,
        workflow_debug=self._workflow_debug,
        workflow_dry_run=self._workflow_dry_run,
    )
    self.logger.info(manifest)

    # Submit workflow
    self._kubectl.create_generic_object(
        namespace="osm-workflows",
        manifest_dict=yaml.safe_load(manifest),
        api_group="argoproj.io",
        api_plural="workflows",
        api_version="v1alpha1",
    )
    return workflow_name


async def register_cluster(self, op_id, op_params, content):
    self.logger.info("Register cluster workflow Enter")
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")

    db_cluster = content["cluster"]
    # cluster_name = db_cluster["name"].lower()
    cluster_name = db_cluster["git_name"].lower()

    # Create secret with kubeconfig
    secret_name = f"kubeconfig-{cluster_name}"
    secret_namespace = "managed-resources"
    secret_key = "kubeconfig"
    secret_value = yaml.safe_dump(
        db_cluster["credentials"], indent=4, default_flow_style=False, sort_keys=False
    )
    await self.create_secret(
        secret_name,
        secret_namespace,
        secret_key,
        secret_value,
    )

    workflow_name = await self.create_cluster(op_id, op_params, content, True)
    return workflow_name


async def deregister_cluster(self, op_id, op_params, content):
    self.logger.info("Deregister cluster workflow Enter")
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")
    workflow_name = await self.delete_cluster(op_id, op_params, content)
    return workflow_name


async def get_cluster_credentials(self, db_cluster):
    """
    returns the kubeconfig file of a K8s cluster in a dictionary
    """
    self.logger.info("Get cluster credentials Enter")
    self.logger.info(f"Content: {db_cluster}")

    secret_name = f"kubeconfig-{db_cluster['git_name'].lower()}"
    secret_namespace = "managed-resources"
    secret_key = "kubeconfig"

    self.logger.info(f"Checking content of secret {secret_name} ...")
    try:
        returned_secret_data = await self._kubectl.get_secret_content(
            name=secret_name,
            namespace=secret_namespace,
        )
        returned_secret_value = base64.b64decode(
            returned_secret_data[secret_key]
        ).decode("utf-8")
        return True, yaml.safe_load(returned_secret_value)
    except Exception as e:
        message = f"Not possible to get the credentials of the cluster. Exception: {e}"
        self.logger.critical(message)
        return False, message


async def check_create_cluster(self, op_id, op_params, content):
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")
    return True, "OK"


async def check_update_cluster(self, op_id, op_params, content):
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")
    return True, "OK"


async def check_delete_cluster(self, op_id, op_params, content):
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")
    return True, "OK"


async def check_register_cluster(self, op_id, op_params, content):
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")
    return True, "OK"


async def check_deregister_cluster(self, op_id, op_params, content):
    self.logger.info(f"Operation {op_id}. Params: {op_params}. Content: {content}")
    return True, "OK"