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