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 rollback(
 212         self, cluster_uuid: str, kdu_instance: str, revision=0, db_dict: dict = None
 213     ):
 214         """
 215         Rolls back a previous update of a KDU instance. It would implicitly use the
 216         `rollback` call. It can be used both to rollback from a Chart/Bundle version
 217         update or from a reconfiguration. This would be exposed as Day-2 primitive.
 218
 219         :param cluster_uuid: UUID of a K8s cluster known by OSM
 220         :param kdu_instance: unique name for the KDU instance
 221         :param revision: revision to which revert changes. If omitted, it will revert
 222             the last update only
 223         :param dict db_dict: where to write into database when the status changes.
 224                         It contains a dict with {collection: <str>, filter: {},
 225                         path: <str>},
 226                             e.g. {collection: "nsrs", filter:
 227                             {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 228         :return:If successful, reference to the current active revision of the KDU
 229             instance after the rollback
 230         """
 231
 232     @abc.abstractmethod
 233     async def uninstall(self, cluster_uuid: str, kdu_instance: str):
 234         """
 235         Removes an existing KDU instance. It would implicitly use the `delete` call
 236         (this call would happen after all _terminate-config-primitive_ of the VNF are
 237         invoked).
 238
 239         :param cluster_uuid: UUID of a K8s cluster known by OSM
 240         :param kdu_instance: unique name for the KDU instance to be deleted
 241         :return: True if successful
 242         """
 243
 244     @abc.abstractmethod
 245     async def exec_primitive(
 246         self,
 247         cluster_uuid: str = None,
 248         kdu_instance: str = None,
 249         primitive_name: str = None,
 250         timeout: float = 300,
 251         params: dict = None,
 252         db_dict: dict = None,
 253     ) -> str:
 254         """Exec primitive (Juju action)
 255
 256         :param cluster_uuid str: The UUID of the cluster
 257         :param kdu_instance str: The unique name of the KDU instance
 258         :param primitive_name: Name of action that will be executed
 259         :param timeout: Timeout for action execution
 260         :param params: Dictionary of all the parameters needed for the action
 261         :db_dict: Dictionary for any additional data
 262
 263         :return: Returns the output of the action
 264         """
 265
 266     @abc.abstractmethod
 267     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 268         """
 269         These calls will retrieve from the Chart/Bundle:
 270
 271             - The list of configurable values and their defaults (e.g. in Charts,
 272                 it would retrieve the contents of `values.yaml`).
 273             - If available, any embedded help file (e.g. `readme.md`) embedded in the
 274                 Chart/Bundle.
 275
 276         :param kdu_model: chart/bundle reference
 277         :param repo_url: optional, reposotory URL (None if tar.gz, URl in other cases,
 278             even stable URL)
 279         :return:
 280
 281         If successful, it will return the available parameters and their default values
 282         as provided by the backend.
 283         """
 284
 285     @abc.abstractmethod
 286     async def help_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 287         """
 288
 289         :param kdu_model: chart/bundle reference
 290         :param repo_url: optional, reposotory URL (None if tar.gz, URl in other cases,
 291             even stable URL)
 292         :return: If successful, it will return the contents of the 'readme.md'
 293         """
 294
 295     @abc.abstractmethod
 296     async def status_kdu(self, cluster_uuid: str, kdu_instance: str) -> str:
 297         """
 298         This call would retrieve tha current state of a given KDU instance. It would be
 299         would allow to retrieve the _composition_ (i.e. K8s objects) and _specific
 300         values_ of the configuration parameters applied to a given instance. This call
 301         would be based on the `status` call.
 302
 303         :param cluster_uuid: UUID of a K8s cluster known by OSM
 304         :param kdu_instance: unique name for the KDU instance
 305         :return: If successful, it will return the following vector of arguments:
 306         - K8s `namespace` in the cluster where the KDU lives
 307         - `state` of the KDU instance. It can be:
 308               - UNKNOWN
 309               - DEPLOYED
 310               - DELETED
 311               - SUPERSEDED
 312               - FAILED or
 313               - DELETING
 314         - List of `resources` (objects) that this release consists of, sorted by kind,
 315           and the status of those resources
 316         - Last `deployment_time`.
 317
 318         """
 319
 320     @abc.abstractmethod
 321     async def get_services(self,
 322                            cluster_uuid: str,
 323                            kdu_instance: str,
 324                            namespace: str) -> list:
 325         """
 326         Returns a list of services defined for the specified kdu instance.
 327
 328         :param cluster_uuid: UUID of a K8s cluster known by OSM
 329         :param kdu_instance: unique name for the KDU instance
 330         :param namespace: K8s namespace used by the KDU instance
 331         :return: If successful, it will return a list of services, Each service
 332         can have the following data:
 333         - `name` of the service
 334         - `type` type of service in the k8 cluster
 335         - `ports` List of ports offered by the service, for each port includes at least
 336         name, port, protocol
 337         - `cluster_ip` Internal ip to be used inside k8s cluster
 338         - `external_ip` List of external ips (in case they are available)
 339         """
 340
 341     @abc.abstractmethod
 342     async def get_service(self,
 343                           cluster_uuid: str,
 344                           service_name: str,
 345                           namespace: str = None) -> object:
 346         """
 347         Obtains the data of the specified service in the k8cluster.
 348
 349         :param cluster_uuid: UUID of a K8s cluster known by OSM
 350         :param service_name: name of the K8s service in the specified namespace
 351         :param namespace: K8s namespace used by the KDU instance
 352         :return: If successful, it will return a list of services, Each service can have
 353         the following data:
 354         - `name` of the service
 355         - `type` type of service in the k8 cluster
 356         - `ports` List of ports offered by the service, for each port includes at least
 357         name, port, protocol
 358         - `cluster_ip` Internal ip to be used inside k8s cluster
 359         - `external_ip` List of external ips (in case they are available)
 360         """
 361
 362     """
 363     ####################################################################################
 364     ################################### P R I V A T E ##################################
 365     ####################################################################################
 366     """
 367
 368     async def write_app_status_to_db(
 369         self, db_dict: dict, status: str, detailed_status: str, operation: str
 370     ) -> bool:
 371
 372         if not self.db:
 373             self.warning("No db => No database write")
 374             return False
 375
 376         if not db_dict:
 377             self.warning("No db_dict => No database write")
 378             return False
 379
 380         self.log.debug("status={}".format(status))
 381
 382         try:
 383
 384             the_table = db_dict["collection"]
 385             the_filter = db_dict["filter"]
 386             the_path = db_dict["path"]
 387             if not the_path[-1] == ".":
 388                 the_path = the_path + "."
 389             update_dict = {
 390                 the_path + "operation": operation,
 391                 the_path + "status": status,
 392                 the_path + "detailed-status": detailed_status,
 393                 the_path + "status-time": str(time.time()),
 394             }
 395
 396             self.db.set_one(
 397                 table=the_table,
 398                 q_filter=the_filter,
 399                 update_dict=update_dict,
 400                 fail_on_empty=True,
 401             )
 402
 403             # database callback
 404             if self.on_update_db:
 405                 if asyncio.iscoroutinefunction(self.on_update_db):
 406                     await self.on_update_db(
 407                         the_table, the_filter, the_path, update_dict
 408                     )
 409                 else:
 410                     self.on_update_db(the_table, the_filter, the_path, update_dict)
 411
 412             return True
 413
 414         except Exception as e:
 415             self.log.info("Exception writing status to database: {}".format(e))
 416             return False