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.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.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 for helm tiller. 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 software in the cluster
  77         (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     ):
  88         """
  89         Add a new repository to OSM database
  90
  91         :param cluster_uuid: the cluster
  92         :param name: name for the repo in OSM
  93         :param url: URL of the repo
  94         :param repo_type: either "chart" or "bundle"
  95         :return: True if successful
  96         """
  97
  98     @abc.abstractmethod
  99     async def repo_list(
 100             self,
 101             cluster_uuid: str
 102     ):
 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(
 112             self,
 113             cluster_uuid: str,
 114             name: str
 115     ):
 116         """
 117         Remove a repository from OSM
 118
 119         :param name: repo name in OSM
 120         :param cluster_uuid: the cluster
 121         :return: True if successful
 122         """
 123
 124     @abc.abstractmethod
 125     async def reset(
 126             self,
 127             cluster_uuid: str,
 128             force: bool = False,
 129             uninstall_sw: bool = False
 130     ) -> bool:
 131         """
 132         Uninstalls Tiller/Charm from a known K8s cluster and removes it from the list of known K8s clusters.
 133         Intended to be used e.g. when the NS instance is deleted.
 134
 135         :param cluster_uuid: UUID of a K8s cluster known by OSM.
 136         :param force: force deletion, even in case there are deployed releases
 137         :param uninstall_sw: flag to indicate that sw uninstallation from software is needed
 138         :return: str: kdu_instance generated by helm
 139         """
 140
 141     @abc.abstractmethod
 142     async def install(
 143             self,
 144             cluster_uuid: str,
 145             kdu_model: str,
 146             atomic: bool = True,
 147             timeout: float = 300,
 148             params: dict = None,
 149             db_dict: dict = None
 150     ):
 151         """
 152         Deploys of a new KDU instance. It would implicitly rely on the `install` call to deploy the Chart/Bundle
 153         properly parametrized (in practice, this call would happen before any _initial-config-primitive_
 154         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 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 atomic: If set, installation process purges chart/bundle on fail, also will wait until
 162             all the K8s objects are active
 163         :param timeout: Time in seconds to wait for the install of the chart/bundle (defaults to
 164             Helm default timeout: 300s)
 165         :param params: dictionary of key-value pairs for instantiation parameters (overriding default values)
 166         :param dict db_dict: where to write into database when the status changes.
 167                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 168                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 169         :return: True if successful
 170         """
 171
 172     @abc.abstractmethod
 173     async def upgrade(
 174             self,
 175             cluster_uuid: str,
 176             kdu_instance: str,
 177             kdu_model: str = None,
 178             atomic: bool = True,
 179             timeout: float = 300,
 180             params: dict = None,
 181             db_dict: dict = None
 182     ):
 183         """
 184         Upgrades an existing KDU instance. It would implicitly use the `upgrade` call over an existing Chart/Bundle.
 185         It can be used both to upgrade the chart or to reconfigure it. This would be exposed as Day-2 primitive.
 186
 187         :param cluster_uuid: UUID of a K8s cluster known by OSM
 188         :param kdu_instance: unique name for the KDU instance to be updated
 189         :param kdu_model: new chart/bundle:version reference
 190         :param atomic: rollback in case of fail and wait for pods and services are available
 191         :param timeout: Time in seconds to wait for the install of the chart/bundle (defaults to
 192             Helm default timeout: 300s)
 193         :param params: new dictionary of key-value pairs for instantiation parameters
 194         :param dict db_dict: where to write into database when the status changes.
 195                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 196                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 197         :return: reference to the new revision number of the KDU instance
 198         """
 199
 200     @abc.abstractmethod
 201     async def rollback(
 202             self,
 203             cluster_uuid: str,
 204             kdu_instance: str,
 205             revision=0,
 206             db_dict: dict = None
 207     ):
 208         """
 209         Rolls back a previous update of a KDU instance. It would implicitly use the `rollback` call.
 210         It can be used both to rollback from a Chart/Bundle version update or from a reconfiguration.
 211         This would be exposed as Day-2 primitive.
 212
 213         :param cluster_uuid: UUID of a K8s cluster known by OSM
 214         :param kdu_instance: unique name for the KDU instance
 215         :param revision: revision to which revert changes. If omitted, it will revert the last update only
 216         :param dict db_dict: where to write into database when the status changes.
 217                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 218                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 219         :return:If successful, reference to the current active revision of the KDU instance after the rollback
 220         """
 221
 222     @abc.abstractmethod
 223     async def uninstall(
 224             self,
 225             cluster_uuid: str,
 226             kdu_instance: str
 227     ):
 228         """
 229         Removes an existing KDU instance. It would implicitly use the `delete` call (this call would happen
 230         after all _terminate-config-primitive_ of the VNF are invoked).
 231
 232         :param cluster_uuid: UUID of a K8s cluster known by OSM
 233         :param kdu_instance: unique name for the KDU instance to be deleted
 234         :return: True if successful
 235         """
 236
 237     @abc.abstractmethod
 238     async def inspect_kdu(
 239             self,
 240             kdu_model: str
 241     ) -> str:
 242         """
 243         These calls will retrieve from the Charm/Bundle:
 244
 245             - The list of configurable values and their defaults (e.g. in Charts, it would retrieve
 246                 the contents of `values.yaml`).
 247             - If available, any embedded help file (e.g. `readme.md`) embedded in the Chart/Bundle.
 248
 249         :param cluster_uuid: the cluster to get the information
 250         :param kdu_model: chart/bundle reference
 251         :return: If successful, it will return a dictionary containing the list of available parameters
 252             and their default values
 253         """
 254
 255     @abc.abstractmethod
 256     async def help_kdu(
 257             self,
 258             kdu_model: str
 259     ) -> str:
 260         """
 261
 262         :param cluster_uuid: the cluster to get the information
 263         :param kdu_model: chart/bundle reference
 264         :return: If successful, it will return the contents of the 'readme.md'
 265         """
 266
 267     @abc.abstractmethod
 268     async def status_kdu(
 269             self,
 270             cluster_uuid: str,
 271             kdu_instance: str
 272     ) -> str:
 273         """
 274         This call would retrieve tha current state of a given KDU instance. It would be would allow to retrieve
 275         the _composition_ (i.e. K8s objects) and _specific values_ of the configuration parameters applied
 276         to a given instance. This call would be based on the `status` call.
 277
 278         :param cluster_uuid: UUID of a K8s cluster known by OSM
 279         :param kdu_instance: unique name for the KDU instance
 280         :return: If successful, it will return the following vector of arguments:
 281         - K8s `namespace` in the cluster where the KDU lives
 282         - `state` of the KDU instance. It can be:
 283               - UNKNOWN
 284               - DEPLOYED
 285               - DELETED
 286               - SUPERSEDED
 287               - FAILED or
 288               - DELETING
 289         - List of `resources` (objects) that this release consists of, sorted by kind, and the status of those resources
 290         - Last `deployment_time`.
 291
 292         """
 293
 294     """
 295     ##################################################################################################
 296     ########################################## P R I V A T E #########################################
 297     ##################################################################################################
 298     """
 299
 300     async def write_app_status_to_db(
 301             self,
 302             db_dict: dict,
 303             status: str,
 304             detailed_status: str,
 305             operation: str
 306     ) -> bool:
 307
 308         if not self.db:
 309             self.warning('No db => No database write')
 310             return False
 311
 312         if not db_dict:
 313             self.warning('No db_dict => No database write')
 314             return False
 315
 316         self.debug('status={}'.format(status))
 317
 318         try:
 319
 320             the_table = db_dict['collection']
 321             the_filter = db_dict['filter']
 322             the_path = db_dict['path']
 323             if not the_path[-1] == '.':
 324                 the_path = the_path + '.'
 325             update_dict = {
 326                 the_path + 'operation': operation,
 327                 the_path + 'status': status,
 328                 the_path + 'detailed-status': detailed_status,
 329                 the_path + 'status-time': str(time.time()),
 330             }
 331
 332             self.db.set_one(
 333                 table=the_table,
 334                 q_filter=the_filter,
 335                 update_dict=update_dict,
 336                 fail_on_empty=True
 337             )
 338
 339             # database callback
 340             if self.on_update_db:
 341                 if asyncio.iscoroutinefunction(self.on_update_db):
 342                     await self.on_update_db(the_table, the_filter, the_path, update_dict)
 343                 else:
 344                     self.on_update_db(the_table, the_filter, the_path, update_dict)
 345
 346             return True
 347
 348         except Exception as e:
 349             self.info('Exception writing status to database: {}'.format(e))
 350             return False