n2vc/k8s_juju_conn.py

   1 # Copyright 2019 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import asyncio
  16 import os
  17 import uuid
  18 import yaml
  19 import tempfile
  20 import binascii
  21 import base64
  22
  23 from n2vc.config import ModelConfig
  24 from n2vc.exceptions import K8sException, N2VCBadArgumentsException
  25 from n2vc.k8s_conn import K8sConnector
  26 from n2vc.kubectl import Kubectl, CORE_CLIENT, RBAC_CLIENT
  27 from .exceptions import MethodNotImplemented
  28 from n2vc.utils import base64_to_cacert
  29 from n2vc.libjuju import Libjuju
  30
  31 from kubernetes.client.models import (
  32     V1ClusterRole,
  33     V1ObjectMeta,
  34     V1PolicyRule,
  35     V1ServiceAccount,
  36     V1ClusterRoleBinding,
  37     V1RoleRef,
  38     V1Subject,
  39 )
  40
  41 from typing import Dict
  42
  43 SERVICE_ACCOUNT_TOKEN_KEY = "token"
  44 SERVICE_ACCOUNT_ROOT_CA_KEY = "ca.crt"
  45 RBAC_LABEL_KEY_NAME = "rbac-id"
  46
  47 ADMIN_NAMESPACE = "kube-system"
  48 RBAC_STACK_PREFIX = "juju-credential"
  49
  50
  51 def generate_rbac_id():
  52     return binascii.hexlify(os.urandom(4)).decode()
  53
  54
  55 class K8sJujuConnector(K8sConnector):
  56     def __init__(
  57         self,
  58         fs: object,
  59         db: object,
  60         kubectl_command: str = "/usr/bin/kubectl",
  61         juju_command: str = "/usr/bin/juju",
  62         log: object = None,
  63         loop: object = None,
  64         on_update_db=None,
  65         vca_config: dict = None,
  66     ):
  67         """
  68         :param fs: file system for kubernetes and helm configuration
  69         :param db: Database object
  70         :param kubectl_command: path to kubectl executable
  71         :param helm_command: path to helm executable
  72         :param log: logger
  73         :param: loop: Asyncio loop
  74         """
  75
  76         # parent class
  77         K8sConnector.__init__(
  78             self,
  79             db,
  80             log=log,
  81             on_update_db=on_update_db,
  82         )
  83
  84         self.fs = fs
  85         self.loop = loop or asyncio.get_event_loop()
  86         self.log.debug("Initializing K8S Juju connector")
  87
  88         required_vca_config = [
  89             "host",
  90             "user",
  91             "secret",
  92             "ca_cert",
  93         ]
  94         if not vca_config or not all(k in vca_config for k in required_vca_config):
  95             raise N2VCBadArgumentsException(
  96                 message="Missing arguments in vca_config: {}".format(vca_config),
  97                 bad_args=required_vca_config,
  98             )
  99         port = vca_config["port"] if "port" in vca_config else 17070
 100         url = "{}:{}".format(vca_config["host"], port)
 101         model_config = ModelConfig(vca_config)
 102         username = vca_config["user"]
 103         secret = vca_config["secret"]
 104         ca_cert = base64_to_cacert(vca_config["ca_cert"])
 105
 106         self.libjuju = Libjuju(
 107             endpoint=url,
 108             api_proxy=None,  # Not needed for k8s charms
 109             model_config=model_config,
 110             username=username,
 111             password=secret,
 112             cacert=ca_cert,
 113             loop=self.loop,
 114             log=self.log,
 115             db=self.db,
 116         )
 117         self.log.debug("K8S Juju connector initialized")
 118         # TODO: Remove these commented lines:
 119         # self.authenticated = False
 120         # self.models = {}
 121         # self.juju_secret = ""
 122
 123     """Initialization"""
 124
 125     async def init_env(
 126         self,
 127         k8s_creds: str,
 128         namespace: str = "kube-system",
 129         reuse_cluster_uuid: str = None,
 130     ) -> (str, bool):
 131         """
 132         It prepares a given K8s cluster environment to run Juju bundles.
 133
 134         :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
 135             '.kube/config'
 136         :param namespace: optional namespace to be used for juju. By default,
 137             'kube-system' will be used
 138         :param reuse_cluster_uuid: existing cluster uuid for reuse
 139         :return: uuid of the K8s cluster and True if connector has installed some
 140             software in the cluster
 141             (on error, an exception will be raised)
 142         """
 143
 144         cluster_uuid = reuse_cluster_uuid or str(uuid.uuid4())
 145
 146         kubecfg = tempfile.NamedTemporaryFile()
 147         with open(kubecfg.name, "w") as kubecfg_file:
 148             kubecfg_file.write(k8s_creds)
 149         kubectl = Kubectl(config_file=kubecfg.name)
 150
 151         # CREATING RESOURCES IN K8S
 152         rbac_id = generate_rbac_id()
 153         metadata_name = "{}-{}".format(RBAC_STACK_PREFIX, rbac_id)
 154         labels = {RBAC_STACK_PREFIX: rbac_id}
 155
 156         # Create cleanup dictionary to clean up created resources
 157         # if it fails in the middle of the process
 158         cleanup_data = []
 159         try:
 160             self._create_cluster_role(
 161                 kubectl,
 162                 name=metadata_name,
 163                 labels=labels,
 164             )
 165             cleanup_data.append(
 166                 {
 167                     "delete": self._delete_cluster_role,
 168                     "args": (kubectl, metadata_name),
 169                 }
 170             )
 171
 172             self._create_service_account(
 173                 kubectl,
 174                 name=metadata_name,
 175                 labels=labels,
 176             )
 177             cleanup_data.append(
 178                 {
 179                     "delete": self._delete_service_account,
 180                     "args": (kubectl, metadata_name),
 181                 }
 182             )
 183
 184             self._create_cluster_role_binding(
 185                 kubectl,
 186                 name=metadata_name,
 187                 labels=labels,
 188             )
 189             cleanup_data.append(
 190                 {
 191                     "delete": self._delete_service_account,
 192                     "args": (kubectl, metadata_name),
 193                 }
 194             )
 195             token, client_cert_data = await self._get_secret_data(
 196                 kubectl,
 197                 metadata_name,
 198             )
 199
 200             default_storage_class = kubectl.get_default_storage_class()
 201             await self.libjuju.add_k8s(
 202                 name=cluster_uuid,
 203                 rbac_id=rbac_id,
 204                 token=token,
 205                 client_cert_data=client_cert_data,
 206                 configuration=kubectl.configuration,
 207                 storage_class=default_storage_class,
 208                 credential_name=self._get_credential_name(cluster_uuid),
 209             )
 210             return cluster_uuid, True
 211         except Exception as e:
 212             self.log.error("Error initializing k8scluster: {}".format(e))
 213             if len(cleanup_data) > 0:
 214                 self.log.debug("Cleaning up created resources in k8s cluster...")
 215                 for item in cleanup_data:
 216                     delete_function = item["delete"]
 217                     delete_args = item["args"]
 218                     delete_function(*delete_args)
 219                 self.log.debug("Cleanup finished")
 220             raise e
 221
 222     """Repo Management"""
 223
 224     async def repo_add(
 225         self,
 226         name: str,
 227         url: str,
 228         _type: str = "charm",
 229     ):
 230         raise MethodNotImplemented()
 231
 232     async def repo_list(self):
 233         raise MethodNotImplemented()
 234
 235     async def repo_remove(
 236         self,
 237         name: str,
 238     ):
 239         raise MethodNotImplemented()
 240
 241     async def synchronize_repos(self, cluster_uuid: str, name: str):
 242         """
 243         Returns None as currently add_repo is not implemented
 244         """
 245         return None
 246
 247     """Reset"""
 248
 249     async def reset(
 250         self, cluster_uuid: str, force: bool = False, uninstall_sw: bool = False
 251     ) -> bool:
 252         """Reset a cluster
 253
 254         Resets the Kubernetes cluster by removing the model that represents it.
 255
 256         :param cluster_uuid str: The UUID of the cluster to reset
 257         :return: Returns True if successful or raises an exception.
 258         """
 259
 260         try:
 261             self.log.debug("[reset] Removing k8s cloud")
 262
 263             cloud_creds = await self.libjuju.get_cloud_credentials(
 264                 cluster_uuid,
 265                 self._get_credential_name(cluster_uuid),
 266             )
 267
 268             await self.libjuju.remove_cloud(cluster_uuid)
 269
 270             kubecfg = self.get_credentials(cluster_uuid=cluster_uuid)
 271
 272             kubecfg_file = tempfile.NamedTemporaryFile()
 273             with open(kubecfg_file.name, "w") as f:
 274                 f.write(kubecfg)
 275             kubectl = Kubectl(config_file=kubecfg_file.name)
 276
 277             delete_functions = [
 278                 self._delete_cluster_role_binding,
 279                 self._delete_service_account,
 280                 self._delete_cluster_role,
 281             ]
 282
 283             credential_attrs = cloud_creds[0].result["attrs"]
 284             if RBAC_LABEL_KEY_NAME in credential_attrs:
 285                 rbac_id = credential_attrs[RBAC_LABEL_KEY_NAME]
 286                 metadata_name = "{}-{}".format(RBAC_STACK_PREFIX, rbac_id)
 287                 delete_args = (kubectl, metadata_name)
 288                 for delete_func in delete_functions:
 289                     try:
 290                         delete_func(*delete_args)
 291                     except Exception as e:
 292                         self.log.warning("Cannot remove resource in K8s {}".format(e))
 293
 294         except Exception as e:
 295             self.log.debug("Caught exception during reset: {}".format(e))
 296             raise e
 297         return True
 298
 299     """Deployment"""
 300
 301     async def install(
 302         self,
 303         cluster_uuid: str,
 304         kdu_model: str,
 305         kdu_instance: str,
 306         atomic: bool = True,
 307         timeout: float = 1800,
 308         params: dict = None,
 309         db_dict: dict = None,
 310         kdu_name: str = None,
 311         namespace: str = None,
 312     ) -> bool:
 313         """Install a bundle
 314
 315         :param cluster_uuid str: The UUID of the cluster to install to
 316         :param kdu_model str: The name or path of a bundle to install
 317         :param kdu_instance: Kdu instance name
 318         :param atomic bool: If set, waits until the model is active and resets
 319                             the cluster on failure.
 320         :param timeout int: The time, in seconds, to wait for the install
 321                             to finish
 322         :param params dict: Key-value pairs of instantiation parameters
 323         :param kdu_name: Name of the KDU instance to be installed
 324         :param namespace: K8s namespace to use for the KDU instance
 325
 326         :return: If successful, returns ?
 327         """
 328         bundle = kdu_model
 329
 330         if not db_dict:
 331             raise K8sException("db_dict must be set")
 332         if not bundle:
 333             raise K8sException("bundle must be set")
 334
 335         if bundle.startswith("cs:"):
 336             pass
 337         elif bundle.startswith("http"):
 338             # Download the file
 339             pass
 340         else:
 341             new_workdir = kdu_model.strip(kdu_model.split("/")[-1])
 342             os.chdir(new_workdir)
 343             bundle = "local:{}".format(kdu_model)
 344
 345         self.log.debug("Checking for model named {}".format(kdu_instance))
 346
 347         # Create the new model
 348         self.log.debug("Adding model: {}".format(kdu_instance))
 349         await self.libjuju.add_model(
 350             model_name=kdu_instance,
 351             cloud_name=cluster_uuid,
 352             credential_name=self._get_credential_name(cluster_uuid),
 353         )
 354
 355         # if model:
 356         # TODO: Instantiation parameters
 357
 358         """
 359         "Juju bundle that models the KDU, in any of the following ways:
 360             - <juju-repo>/<juju-bundle>
 361             - <juju-bundle folder under k8s_models folder in the package>
 362             - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
 363                 in the package>
 364             - <URL_where_to_fetch_juju_bundle>
 365         """
 366         try:
 367             previous_workdir = os.getcwd()
 368         except FileNotFoundError:
 369             previous_workdir = "/app/storage"
 370
 371         self.log.debug("[install] deploying {}".format(bundle))
 372         await self.libjuju.deploy(
 373             bundle, model_name=kdu_instance, wait=atomic, timeout=timeout
 374         )
 375         os.chdir(previous_workdir)
 376         return True
 377
 378     async def instances_list(self, cluster_uuid: str) -> list:
 379         """
 380         returns a list of deployed releases in a cluster
 381
 382         :param cluster_uuid: the cluster
 383         :return:
 384         """
 385         return []
 386
 387     async def upgrade(
 388         self,
 389         cluster_uuid: str,
 390         kdu_instance: str,
 391         kdu_model: str = None,
 392         params: dict = None,
 393     ) -> str:
 394         """Upgrade a model
 395
 396         :param cluster_uuid str: The UUID of the cluster to upgrade
 397         :param kdu_instance str: The unique name of the KDU instance
 398         :param kdu_model str: The name or path of the bundle to upgrade to
 399         :param params dict: Key-value pairs of instantiation parameters
 400
 401         :return: If successful, reference to the new revision number of the
 402                  KDU instance.
 403         """
 404
 405         # TODO: Loop through the bundle and upgrade each charm individually
 406
 407         """
 408         The API doesn't have a concept of bundle upgrades, because there are
 409         many possible changes: charm revision, disk, number of units, etc.
 410
 411         As such, we are only supporting a limited subset of upgrades. We'll
 412         upgrade the charm revision but leave storage and scale untouched.
 413
 414         Scale changes should happen through OSM constructs, and changes to
 415         storage would require a redeployment of the service, at least in this
 416         initial release.
 417         """
 418         raise MethodNotImplemented()
 419
 420     """Rollback"""
 421
 422     async def rollback(
 423         self,
 424         cluster_uuid: str,
 425         kdu_instance: str,
 426         revision: int = 0,
 427     ) -> str:
 428         """Rollback a model
 429
 430         :param cluster_uuid str: The UUID of the cluster to rollback
 431         :param kdu_instance str: The unique name of the KDU instance
 432         :param revision int: The revision to revert to. If omitted, rolls back
 433                              the previous upgrade.
 434
 435         :return: If successful, returns the revision of active KDU instance,
 436                  or raises an exception
 437         """
 438         raise MethodNotImplemented()
 439
 440     """Deletion"""
 441
 442     async def uninstall(self, cluster_uuid: str, kdu_instance: str) -> bool:
 443         """Uninstall a KDU instance
 444
 445         :param cluster_uuid str: The UUID of the cluster
 446         :param kdu_instance str: The unique name of the KDU instance
 447
 448         :return: Returns True if successful, or raises an exception
 449         """
 450
 451         self.log.debug("[uninstall] Destroying model")
 452
 453         await self.libjuju.destroy_model(kdu_instance, total_timeout=3600)
 454
 455         # self.log.debug("[uninstall] Model destroyed and disconnecting")
 456         # await controller.disconnect()
 457
 458         return True
 459         # TODO: Remove these commented lines
 460         # if not self.authenticated:
 461         #     self.log.debug("[uninstall] Connecting to controller")
 462         #     await self.login(cluster_uuid)
 463
 464     async def exec_primitive(
 465         self,
 466         cluster_uuid: str = None,
 467         kdu_instance: str = None,
 468         primitive_name: str = None,
 469         timeout: float = 300,
 470         params: dict = None,
 471         db_dict: dict = None,
 472     ) -> str:
 473         """Exec primitive (Juju action)
 474
 475         :param cluster_uuid str: The UUID of the cluster
 476         :param kdu_instance str: The unique name of the KDU instance
 477         :param primitive_name: Name of action that will be executed
 478         :param timeout: Timeout for action execution
 479         :param params: Dictionary of all the parameters needed for the action
 480         :db_dict: Dictionary for any additional data
 481
 482         :return: Returns the output of the action
 483         """
 484
 485         if not params or "application-name" not in params:
 486             raise K8sException(
 487                 "Missing application-name argument, \
 488                                 argument needed for K8s actions"
 489             )
 490         try:
 491             self.log.debug(
 492                 "[exec_primitive] Getting model "
 493                 "kdu_instance: {}".format(kdu_instance)
 494             )
 495             application_name = params["application-name"]
 496             actions = await self.libjuju.get_actions(application_name, kdu_instance)
 497             if primitive_name not in actions:
 498                 raise K8sException("Primitive {} not found".format(primitive_name))
 499             output, status = await self.libjuju.execute_action(
 500                 application_name, kdu_instance, primitive_name, **params
 501             )
 502
 503             if status != "completed":
 504                 raise K8sException(
 505                     "status is not completed: {} output: {}".format(status, output)
 506                 )
 507
 508             return output
 509
 510         except Exception as e:
 511             error_msg = "Error executing primitive {}: {}".format(primitive_name, e)
 512             self.log.error(error_msg)
 513             raise K8sException(message=error_msg)
 514
 515     """Introspection"""
 516
 517     async def inspect_kdu(
 518         self,
 519         kdu_model: str,
 520     ) -> dict:
 521         """Inspect a KDU
 522
 523         Inspects a bundle and returns a dictionary of config parameters and
 524         their default values.
 525
 526         :param kdu_model str: The name or path of the bundle to inspect.
 527
 528         :return: If successful, returns a dictionary of available parameters
 529                  and their default values.
 530         """
 531
 532         kdu = {}
 533         if not os.path.exists(kdu_model):
 534             raise K8sException("file {} not found".format(kdu_model))
 535
 536         with open(kdu_model, "r") as f:
 537             bundle = yaml.safe_load(f.read())
 538
 539             """
 540             {
 541                 'description': 'Test bundle',
 542                 'bundle': 'kubernetes',
 543                 'applications': {
 544                     'mariadb-k8s': {
 545                         'charm': 'cs:~charmed-osm/mariadb-k8s-20',
 546                         'scale': 1,
 547                         'options': {
 548                             'password': 'manopw',
 549                             'root_password': 'osm4u',
 550                             'user': 'mano'
 551                         },
 552                         'series': 'kubernetes'
 553                     }
 554                 }
 555             }
 556             """
 557             # TODO: This should be returned in an agreed-upon format
 558             kdu = bundle["applications"]
 559
 560         return kdu
 561
 562     async def help_kdu(
 563         self,
 564         kdu_model: str,
 565     ) -> str:
 566         """View the README
 567
 568         If available, returns the README of the bundle.
 569
 570         :param kdu_model str: The name or path of a bundle
 571
 572         :return: If found, returns the contents of the README.
 573         """
 574         readme = None
 575
 576         files = ["README", "README.txt", "README.md"]
 577         path = os.path.dirname(kdu_model)
 578         for file in os.listdir(path):
 579             if file in files:
 580                 with open(file, "r") as f:
 581                     readme = f.read()
 582                     break
 583
 584         return readme
 585
 586     async def status_kdu(
 587         self,
 588         cluster_uuid: str,
 589         kdu_instance: str,
 590     ) -> dict:
 591         """Get the status of the KDU
 592
 593         Get the current status of the KDU instance.
 594
 595         :param cluster_uuid str: The UUID of the cluster
 596         :param kdu_instance str: The unique id of the KDU instance
 597
 598         :return: Returns a dictionary containing namespace, state, resources,
 599                  and deployment_time.
 600         """
 601         status = {}
 602         model_status = await self.libjuju.get_model_status(kdu_instance)
 603         for name in model_status.applications:
 604             application = model_status.applications[name]
 605             status[name] = {"status": application["status"]["status"]}
 606
 607         return status
 608
 609     async def get_services(
 610         self, cluster_uuid: str, kdu_instance: str, namespace: str
 611     ) -> list:
 612         """Return a list of services of a kdu_instance"""
 613
 614         credentials = self.get_credentials(cluster_uuid=cluster_uuid)
 615
 616         kubecfg = tempfile.NamedTemporaryFile()
 617         with open(kubecfg.name, "w") as kubecfg_file:
 618             kubecfg_file.write(credentials)
 619         kubectl = Kubectl(config_file=kubecfg.name)
 620
 621         return kubectl.get_services(
 622             field_selector="metadata.namespace={}".format(kdu_instance)
 623         )
 624
 625     async def get_service(
 626         self, cluster_uuid: str, service_name: str, namespace: str
 627     ) -> object:
 628         """Return data for a specific service inside a namespace"""
 629
 630         credentials = self.get_credentials(cluster_uuid=cluster_uuid)
 631
 632         kubecfg = tempfile.NamedTemporaryFile()
 633         with open(kubecfg.name, "w") as kubecfg_file:
 634             kubecfg_file.write(credentials)
 635         kubectl = Kubectl(config_file=kubecfg.name)
 636
 637         return kubectl.get_services(
 638             field_selector="metadata.name={},metadata.namespace={}".format(
 639                 service_name, namespace
 640             )
 641         )[0]
 642
 643     def get_credentials(self, cluster_uuid: str) -> str:
 644         """
 645         Get Cluster Kubeconfig
 646         """
 647         k8scluster = self.db.get_one(
 648             "k8sclusters", q_filter={"_id": cluster_uuid}, fail_on_empty=False
 649         )
 650
 651         self.db.encrypt_decrypt_fields(
 652             k8scluster.get("credentials"),
 653             "decrypt",
 654             ["password", "secret"],
 655             schema_version=k8scluster["schema_version"],
 656             salt=k8scluster["_id"],
 657         )
 658
 659         return yaml.safe_dump(k8scluster.get("credentials"))
 660
 661     def _get_credential_name(self, cluster_uuid: str) -> str:
 662         """
 663         Get credential name for a k8s cloud
 664
 665         We cannot use the cluster_uuid for the credential name directly,
 666         because it cannot start with a number, it must start with a letter.
 667         Therefore, the k8s cloud credential name will be "cred-" followed
 668         by the cluster uuid.
 669
 670         :param: cluster_uuid:   Cluster UUID of the kubernetes cloud (=cloud_name)
 671
 672         :return:                Name to use for the credential name.
 673         """
 674         return "cred-{}".format(cluster_uuid)
 675
 676     def get_namespace(
 677         self,
 678         cluster_uuid: str,
 679     ) -> str:
 680         """Get the namespace UUID
 681         Gets the namespace's unique name
 682
 683         :param cluster_uuid str: The UUID of the cluster
 684         :returns: The namespace UUID, or raises an exception
 685         """
 686         pass
 687
 688     def _create_cluster_role(
 689         self,
 690         kubectl: Kubectl,
 691         name: str,
 692         labels: Dict[str, str],
 693     ):
 694         cluster_roles = kubectl.clients[RBAC_CLIENT].list_cluster_role(
 695             field_selector="metadata.name={}".format(name)
 696         )
 697
 698         if len(cluster_roles.items) > 0:
 699             raise Exception(
 700                 "Cluster role with metadata.name={} already exists".format(name)
 701             )
 702
 703         metadata = V1ObjectMeta(name=name, labels=labels, namespace=ADMIN_NAMESPACE)
 704         # Cluster role
 705         cluster_role = V1ClusterRole(
 706             metadata=metadata,
 707             rules=[
 708                 V1PolicyRule(api_groups=["*"], resources=["*"], verbs=["*"]),
 709                 V1PolicyRule(non_resource_ur_ls=["*"], verbs=["*"]),
 710             ],
 711         )
 712
 713         kubectl.clients[RBAC_CLIENT].create_cluster_role(cluster_role)
 714
 715     def _delete_cluster_role(self, kubectl: Kubectl, name: str):
 716         kubectl.clients[RBAC_CLIENT].delete_cluster_role(name)
 717
 718     def _create_service_account(
 719         self,
 720         kubectl: Kubectl,
 721         name: str,
 722         labels: Dict[str, str],
 723     ):
 724         service_accounts = kubectl.clients[CORE_CLIENT].list_namespaced_service_account(
 725             ADMIN_NAMESPACE, field_selector="metadata.name={}".format(name)
 726         )
 727         if len(service_accounts.items) > 0:
 728             raise Exception(
 729                 "Service account with metadata.name={} already exists".format(name)
 730             )
 731
 732         metadata = V1ObjectMeta(name=name, labels=labels, namespace=ADMIN_NAMESPACE)
 733         service_account = V1ServiceAccount(metadata=metadata)
 734
 735         kubectl.clients[CORE_CLIENT].create_namespaced_service_account(
 736             ADMIN_NAMESPACE, service_account
 737         )
 738
 739     def _delete_service_account(self, kubectl: Kubectl, name: str):
 740         kubectl.clients[CORE_CLIENT].delete_namespaced_service_account(
 741             name, ADMIN_NAMESPACE
 742         )
 743
 744     def _create_cluster_role_binding(
 745         self,
 746         kubectl: Kubectl,
 747         name: str,
 748         labels: Dict[str, str],
 749     ):
 750         role_bindings = kubectl.clients[RBAC_CLIENT].list_cluster_role_binding(
 751             field_selector="metadata.name={}".format(name)
 752         )
 753         if len(role_bindings.items) > 0:
 754             raise Exception("Generated rbac id already exists")
 755
 756         role_binding = V1ClusterRoleBinding(
 757             metadata=V1ObjectMeta(name=name, labels=labels),
 758             role_ref=V1RoleRef(kind="ClusterRole", name=name, api_group=""),
 759             subjects=[
 760                 V1Subject(kind="ServiceAccount", name=name, namespace=ADMIN_NAMESPACE)
 761             ],
 762         )
 763         kubectl.clients[RBAC_CLIENT].create_cluster_role_binding(role_binding)
 764
 765     def _delete_cluster_role_binding(self, kubectl: Kubectl, name: str):
 766         kubectl.clients[RBAC_CLIENT].delete_cluster_role_binding(name)
 767
 768     async def _get_secret_data(self, kubectl: Kubectl, name: str) -> (str, str):
 769         v1_core = kubectl.clients[CORE_CLIENT]
 770
 771         retries_limit = 10
 772         secret_name = None
 773         while True:
 774             retries_limit -= 1
 775             service_accounts = v1_core.list_namespaced_service_account(
 776                 ADMIN_NAMESPACE, field_selector="metadata.name={}".format(name)
 777             )
 778             if len(service_accounts.items) == 0:
 779                 raise Exception(
 780                     "Service account not found with metadata.name={}".format(name)
 781                 )
 782             service_account = service_accounts.items[0]
 783             if service_account.secrets and len(service_account.secrets) > 0:
 784                 secret_name = service_account.secrets[0].name
 785             if secret_name is not None or not retries_limit:
 786                 break
 787         if not secret_name:
 788             raise Exception(
 789                 "Failed getting the secret from service account {}".format(name)
 790             )
 791         secret = v1_core.list_namespaced_secret(
 792             ADMIN_NAMESPACE,
 793             field_selector="metadata.name={}".format(secret_name),
 794         ).items[0]
 795
 796         token = secret.data[SERVICE_ACCOUNT_TOKEN_KEY]
 797         client_certificate_data = secret.data[SERVICE_ACCOUNT_ROOT_CA_KEY]
 798
 799         return (
 800             base64.b64decode(token).decode("utf-8"),
 801             base64.b64decode(client_certificate_data).decode("utf-8"),
 802         )
 803
 804     @staticmethod
 805     def generate_kdu_instance_name(**kwargs):
 806         db_dict = kwargs.get("db_dict")
 807         kdu_name = kwargs.get("kdu_name", None)
 808         if kdu_name:
 809             kdu_instance = "{}-{}".format(kdu_name, db_dict["filter"]["_id"])
 810         else:
 811             kdu_instance = db_dict["filter"]["_id"]
 812         return kdu_instance