n2vc/k8s_conn.py

   1 ##
   2 # Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
   3 # This file is part of OSM
   4 # All Rights Reserved.
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #    http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  15 # implied.
  16 # See the License for the specific language governing permissions and
  17 # limitations under the License.
  18 #
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact with: nfvlabs@tid.es
  21 ##
  22
  23 import abc
  24 import asyncio
  25 import time
  26
  27 from n2vc.loggable import Loggable
  28
  29
  30 class K8sConnector(abc.ABC, Loggable):
  31     """
  32     ####################################################################################
  33     ################################### P U B L I C ####################################
  34     ####################################################################################
  35     """
  36
  37     @staticmethod
  38     def generate_kdu_instance_name(**kwargs):
  39         raise NotImplementedError("Method not implemented")
  40
  41     def __init__(self, db: object, log: object = None, on_update_db=None):
  42         """
  43
  44         :param db: database object to write current operation status
  45         :param log: logger for tracing
  46         :param on_update_db: callback called when k8s connector updates database
  47         """
  48
  49         # parent class
  50         Loggable.__init__(self, log=log, log_to_console=True, prefix="\nK8S")
  51
  52         # self.log.info('Initializing generic K8S connector')
  53
  54         # the database and update callback
  55         self.db = db
  56         self.on_update_db = on_update_db
  57
  58         # self.log.info('K8S generic connector initialized')
  59
  60     @abc.abstractmethod
  61     async def init_env(
  62         self, k8s_creds: str, namespace: str = "kube-system", reuse_cluster_uuid=None
  63     ) -> (str, bool):
  64         """
  65         It prepares a given K8s cluster environment to run Charts or juju Bundles on
  66         both sides:
  67             client (OSM)
  68             server (Tiller/Charm)
  69
  70         :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
  71         '.kube/config'
  72         :param namespace: optional namespace to be used for the K8s engine (helm
  73         tiller, juju). By default, 'kube-system' will be used
  74         :param reuse_cluster_uuid: existing cluster uuid for reuse
  75         :return: uuid of the K8s cluster and True if connector has installed some
  76         software in the cluster (on error, an exception will be raised)
  77         """
  78
  79     @abc.abstractmethod
  80     async def repo_add(
  81         self,
  82         cluster_uuid: str,
  83         name: str,
  84         url: str,
  85         repo_type: str = "chart",
  86         cert: str = None,
  87         user: str = None,
  88         password: str = None,
  89     ):
  90         """
  91         Add a new repository to OSM database
  92
  93         :param cluster_uuid: the cluster
  94         :param name: name for the repo in OSM
  95         :param url: URL of the repo
  96         :param repo_type: either "chart" or "bundle"
  97         :return: True if successful
  98         """
  99
 100     @abc.abstractmethod
 101     async def repo_list(self, cluster_uuid: str):
 102         """
 103         Get the list of registered repositories
 104
 105         :param cluster_uuid: the cluster
 106         :return: list of registered repositories: [ (name, url) .... ]
 107         """
 108
 109     @abc.abstractmethod
 110     async def repo_remove(self, cluster_uuid: str, name: str):
 111         """
 112         Remove a repository from OSM
 113
 114         :param name: repo name in OSM
 115         :param cluster_uuid: the cluster
 116         :return: True if successful
 117         """
 118
 119     @abc.abstractmethod
 120     async def synchronize_repos(self, cluster_uuid: str, name: str):
 121         """
 122         Synchronizes the list of repositories created in the cluster with
 123         the repositories added by the NBI
 124
 125         :param cluster_uuid: the cluster
 126         :return: List of repositories deleted from the cluster and dictionary with
 127         repos added
 128         """
 129
 130     @abc.abstractmethod
 131     async def reset(
 132         self, cluster_uuid: str, force: bool = False, uninstall_sw: bool = False
 133     ) -> bool:
 134         """
 135         Uninstalls Tiller/Charm from a known K8s cluster and removes it from the list
 136         of known K8s clusters. Intended to be used e.g. when the NS instance is deleted.
 137
 138         :param cluster_uuid: UUID of a K8s cluster known by OSM.
 139         :param force: force deletion, even in case there are deployed releases
 140         :param uninstall_sw: flag to indicate that sw uninstallation from software is
 141         needed
 142         :return: str: kdu_instance generated by helm
 143         """
 144
 145     @abc.abstractmethod
 146     async def install(
 147         self,
 148         cluster_uuid: str,
 149         kdu_model: str,
 150         kdu_instance: str,
 151         atomic: bool = True,
 152         timeout: float = 300,
 153         params: dict = None,
 154         db_dict: dict = None,
 155         kdu_name: str = None,
 156         namespace: str = None,
 157     ):
 158         """
 159         Deploys of a new KDU instance. It would implicitly rely on the `install` call
 160         to deploy the Chart/Bundle properly parametrized (in practice, this call would
 161         happen before any _initial-config-primitive_of the VNF is called).
 162
 163         :param cluster_uuid: UUID of a K8s cluster known by OSM
 164         :param kdu_model: chart/bundle:version reference (string), which can be either
 165             of these options:
 166             - a name of chart/bundle available via the repos known by OSM
 167             - a path to a packaged chart/bundle
 168             - a path to an unpacked chart/bundle directory or a URL
 169         :param kdu_instance: Kdu instance name
 170         :param atomic: If set, installation process purges chart/bundle on fail, also
 171             will wait until all the K8s objects are active
 172         :param timeout: Time in seconds to wait for the install of the chart/bundle
 173             (defaults to Helm default timeout: 300s)
 174         :param params: dictionary of key-value pairs for instantiation parameters
 175             (overriding default values)
 176         :param dict db_dict: where to write into database when the status changes.
 177                         It contains a dict with {collection: <str>, filter: {},
 178                         path: <str>},
 179                             e.g. {collection: "nsrs", filter:
 180                             {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 181         :param kdu_name: Name of the KDU instance to be installed
 182         :param namespace: K8s namespace to use for the KDU instance
 183         :return: True if successful
 184         """
 185
 186     @abc.abstractmethod
 187     async def upgrade(
 188         self,
 189         cluster_uuid: str,
 190         kdu_instance: str,
 191         kdu_model: str = None,
 192         atomic: bool = True,
 193         timeout: float = 300,
 194         params: dict = None,
 195         db_dict: dict = None,
 196     ):
 197         """
 198         Upgrades an existing KDU instance. It would implicitly use the `upgrade` call
 199         over an existing Chart/Bundle. It can be used both to upgrade the chart or to
 200         reconfigure it. This would be exposed as Day-2 primitive.
 201
 202         :param cluster_uuid: UUID of a K8s cluster known by OSM
 203         :param kdu_instance: unique name for the KDU instance to be updated
 204         :param kdu_model: new chart/bundle:version reference
 205         :param atomic: rollback in case of fail and wait for pods and services are
 206             available
 207         :param timeout: Time in seconds to wait for the install of the chart/bundle
 208             (defaults to Helm default timeout: 300s)
 209         :param params: new dictionary of key-value pairs for instantiation parameters
 210         :param dict db_dict: where to write into database when the status changes.
 211                         It contains a dict with {collection: <str>, filter: {},
 212                         path: <str>},
 213                             e.g. {collection: "nsrs", filter:
 214                             {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 215         :return: reference to the new revision number of the KDU instance
 216         """
 217
 218     @abc.abstractmethod
 219     async def scale(
 220         self,
 221         kdu_instance: str,
 222         scale: int,
 223         resource_name: str,
 224         total_timeout: float = 1800,
 225         cluster_uuid: str = None,
 226         kdu_model: str = None,
 227         atomic: bool = True,
 228         db_dict: dict = None,
 229         **kwargs,
 230     ) -> bool:
 231         """Scale a resource in a KDU instance.
 232
 233         Args:
 234             kdu_instance: KDU instance name
 235             scale: Scale to which to set the resource
 236             resource_name: Resource name
 237             total_timeout: The time, in seconds, to wait for the install
 238                 to finish
 239             cluster_uuid: The UUID of the cluster
 240             kdu_model: The chart/bundle reference
 241             atomic: if set, upgrade process rolls back changes made in case of failed upgrade.
 242                 The --wait flag will be set automatically if --atomic is used
 243             db_dict: Dictionary for any additional data
 244             kwargs: Additional parameters
 245                 vca_id (str): VCA ID
 246
 247         Returns:
 248             True if successful, False otherwise
 249         """
 250
 251     @abc.abstractmethod
 252     async def get_scale_count(
 253         self,
 254         resource_name: str,
 255         kdu_instance: str,
 256         cluster_uuid: str,
 257         kdu_model: str,
 258         timeout: float = 300,
 259         **kwargs,
 260     ) -> int:
 261         """Get a resource scale count in a KDU instance.
 262
 263         Args:
 264             resource_name: Resource name
 265             kdu_instance: KDU instance name
 266             cluster_uuid: The UUID of the cluster
 267             kdu_model:    chart/bundle reference
 268             timeout:  The time, in seconds, to wait
 269             kwargs: Additional parameters
 270
 271         Returns:
 272             Resource instance count
 273         """
 274
 275     @abc.abstractmethod
 276     async def rollback(
 277         self, cluster_uuid: str, kdu_instance: str, revision=0, db_dict: dict = None
 278     ):
 279         """
 280         Rolls back a previous update of a KDU instance. It would implicitly use the
 281         `rollback` call. It can be used both to rollback from a Chart/Bundle version
 282         update or from a reconfiguration. This would be exposed as Day-2 primitive.
 283
 284         :param cluster_uuid: UUID of a K8s cluster known by OSM
 285         :param kdu_instance: unique name for the KDU instance
 286         :param revision: revision to which revert changes. If omitted, it will revert
 287             the last update only
 288         :param dict db_dict: where to write into database when the status changes.
 289                         It contains a dict with {collection: <str>, filter: {},
 290                         path: <str>},
 291                             e.g. {collection: "nsrs", filter:
 292                             {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 293         :return:If successful, reference to the current active revision of the KDU
 294             instance after the rollback
 295         """
 296
 297     @abc.abstractmethod
 298     async def uninstall(self, cluster_uuid: str, kdu_instance: str):
 299         """
 300         Removes an existing KDU instance. It would implicitly use the `delete` call
 301         (this call would happen after all _terminate-config-primitive_ of the VNF are
 302         invoked).
 303
 304         :param cluster_uuid: UUID of a K8s cluster known by OSM
 305         :param kdu_instance: unique name for the KDU instance to be deleted
 306         :return: True if successful
 307         """
 308
 309     @abc.abstractmethod
 310     async def exec_primitive(
 311         self,
 312         cluster_uuid: str = None,
 313         kdu_instance: str = None,
 314         primitive_name: str = None,
 315         timeout: float = 300,
 316         params: dict = None,
 317         db_dict: dict = None,
 318     ) -> str:
 319         """Exec primitive (Juju action)
 320
 321         :param cluster_uuid str: The UUID of the cluster
 322         :param kdu_instance str: The unique name of the KDU instance
 323         :param primitive_name: Name of action that will be executed
 324         :param timeout: Timeout for action execution
 325         :param params: Dictionary of all the parameters needed for the action
 326         :db_dict: Dictionary for any additional data
 327
 328         :return: Returns the output of the action
 329         """
 330
 331     @abc.abstractmethod
 332     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 333         """
 334         These calls will retrieve from the Chart/Bundle:
 335
 336             - The list of configurable values and their defaults (e.g. in Charts,
 337                 it would retrieve the contents of `values.yaml`).
 338             - If available, any embedded help file (e.g. `readme.md`) embedded in the
 339                 Chart/Bundle.
 340
 341         :param kdu_model: chart/bundle reference
 342         :param repo_url: optional, reposotory URL (None if tar.gz, URl in other cases,
 343             even stable URL)
 344         :return:
 345
 346         If successful, it will return the available parameters and their default values
 347         as provided by the backend.
 348         """
 349
 350     @abc.abstractmethod
 351     async def help_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 352         """
 353
 354         :param kdu_model: chart/bundle reference
 355         :param repo_url: optional, reposotory URL (None if tar.gz, URl in other cases,
 356             even stable URL)
 357         :return: If successful, it will return the contents of the 'readme.md'
 358         """
 359
 360     @abc.abstractmethod
 361     async def status_kdu(self, cluster_uuid: str, kdu_instance: str) -> str:
 362         """
 363         This call would retrieve tha current state of a given KDU instance. It would be
 364         would allow to retrieve the _composition_ (i.e. K8s objects) and _specific
 365         values_ of the configuration parameters applied to a given instance. This call
 366         would be based on the `status` call.
 367
 368         :param cluster_uuid: UUID of a K8s cluster known by OSM
 369         :param kdu_instance: unique name for the KDU instance
 370         :return: If successful, it will return the following vector of arguments:
 371         - K8s `namespace` in the cluster where the KDU lives
 372         - `state` of the KDU instance. It can be:
 373               - UNKNOWN
 374               - DEPLOYED
 375               - DELETED
 376               - SUPERSEDED
 377               - FAILED or
 378               - DELETING
 379         - List of `resources` (objects) that this release consists of, sorted by kind,
 380           and the status of those resources
 381         - Last `deployment_time`.
 382
 383         """
 384
 385     @abc.abstractmethod
 386     async def get_services(
 387         self, cluster_uuid: str, kdu_instance: str, namespace: str
 388     ) -> list:
 389         """
 390         Returns a list of services defined for the specified kdu instance.
 391
 392         :param cluster_uuid: UUID of a K8s cluster known by OSM
 393         :param kdu_instance: unique name for the KDU instance
 394         :param namespace: K8s namespace used by the KDU instance
 395         :return: If successful, it will return a list of services, Each service
 396         can have the following data:
 397         - `name` of the service
 398         - `type` type of service in the k8 cluster
 399         - `ports` List of ports offered by the service, for each port includes at least
 400         name, port, protocol
 401         - `cluster_ip` Internal ip to be used inside k8s cluster
 402         - `external_ip` List of external ips (in case they are available)
 403         """
 404
 405     @abc.abstractmethod
 406     async def get_service(
 407         self, cluster_uuid: str, service_name: str, namespace: str = None
 408     ) -> object:
 409         """
 410         Obtains the data of the specified service in the k8cluster.
 411
 412         :param cluster_uuid: UUID of a K8s cluster known by OSM
 413         :param service_name: name of the K8s service in the specified namespace
 414         :param namespace: K8s namespace used by the KDU instance
 415         :return: If successful, it will return a list of services, Each service can have
 416         the following data:
 417         - `name` of the service
 418         - `type` type of service in the k8 cluster
 419         - `ports` List of ports offered by the service, for each port includes at least
 420         name, port, protocol
 421         - `cluster_ip` Internal ip to be used inside k8s cluster
 422         - `external_ip` List of external ips (in case they are available)
 423         """
 424
 425     """
 426     ####################################################################################
 427     ################################### P R I V A T E ##################################
 428     ####################################################################################
 429     """
 430
 431     async def write_app_status_to_db(
 432         self, db_dict: dict, status: str, detailed_status: str, operation: str
 433     ) -> bool:
 434
 435         if not self.db:
 436             self.warning("No db => No database write")
 437             return False
 438
 439         if not db_dict:
 440             self.warning("No db_dict => No database write")
 441             return False
 442
 443         self.log.debug("status={}".format(status))
 444
 445         try:
 446
 447             the_table = db_dict["collection"]
 448             the_filter = db_dict["filter"]
 449             the_path = db_dict["path"]
 450             if not the_path[-1] == ".":
 451                 the_path = the_path + "."
 452             update_dict = {
 453                 the_path + "operation": operation,
 454                 the_path + "status": status,
 455                 the_path + "detailed-status": detailed_status,
 456                 the_path + "status-time": str(time.time()),
 457             }
 458
 459             self.db.set_one(
 460                 table=the_table,
 461                 q_filter=the_filter,
 462                 update_dict=update_dict,
 463                 fail_on_empty=True,
 464             )
 465
 466             # database callback
 467             if self.on_update_db:
 468                 if asyncio.iscoroutinefunction(self.on_update_db):
 469                     await self.on_update_db(
 470                         the_table, the_filter, the_path, update_dict
 471                     )
 472                 else:
 473                     self.on_update_db(the_table, the_filter, the_path, update_dict)
 474
 475             return True
 476
 477         except Exception as e:
 478             self.log.info("Exception writing status to database: {}".format(e))
 479             return False