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