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