| ####################################################################################### |
| # 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" |