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