n2vc/k8s_conn.py

##
# Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
# This file is part of OSM
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# For those usages not covered by the Apache License, Version 2.0 please
# contact with: nfvlabs@tid.es
##

import asyncio
from n2vc.loggable import Loggable
import abc
import time


class K8sConnector(abc.ABC, Loggable):

    """
    ##################################################################################################
    ########################################## P U B L I C ###########################################
    ##################################################################################################
    """

    def __init__(
            self,
            db: object,
            log: object = None,
            on_update_db=None
    ):
        """

        :param db: database object to write current operation status
        :param log: logger for tracing
        :param on_update_db: callback called when k8s connector updates database
        """

        # parent class
        Loggable.__init__(self, log=log, log_to_console=True, prefix='\nK8S')

        # self.log.info('Initializing generic K8S connector')

        # the database and update callback
        self.db = db
        self.on_update_db = on_update_db

        # self.log.info('K8S generic connector initialized')

    @abc.abstractmethod
    async def init_env(
            self,
            k8s_creds: str,
            namespace: str = 'kube-system',
            reuse_cluster_uuid=None
    ) -> (str, bool):
        """
        It prepares a given K8s cluster environment to run Charts or juju Bundles on both sides:
            client (OSM)
            server (Tiller/Charm)

        :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid '.kube/config'
        :param namespace: optional namespace to be used for the K8s engine (helm tiller, juju).
        By default, 'kube-system' will be used
        :param reuse_cluster_uuid: existing cluster uuid for reuse
        :return: uuid of the K8s cluster and True if connector has installed some software in the cluster
        (on error, an exception will be raised)
        """

    @abc.abstractmethod
    async def repo_add(
            self,
            cluster_uuid: str,
            name: str,
            url: str,
            repo_type: str = 'chart'
    ):
        """
        Add a new repository to OSM database

        :param cluster_uuid: the cluster
        :param name: name for the repo in OSM
        :param url: URL of the repo
        :param repo_type: either "chart" or "bundle"
        :return: True if successful
        """

    @abc.abstractmethod
    async def repo_list(
            self,
            cluster_uuid: str
    ):
        """
        Get the list of registered repositories

        :param cluster_uuid: the cluster
        :return: list of registered repositories: [ (name, url) .... ]
        """

    @abc.abstractmethod
    async def repo_remove(
            self,
            cluster_uuid: str,
            name: str
    ):
        """
        Remove a repository from OSM

        :param name: repo name in OSM
        :param cluster_uuid: the cluster
        :return: True if successful
        """

    @abc.abstractmethod
    async def synchronize_repos(
            self,
            cluster_uuid: str,
            name: str
    ):
        """
        Synchronizes the list of repositories created in the cluster with
        the repositories added by the NBI

        :param cluster_uuid: the cluster
        :return: List of repositories deleted from the cluster and dictionary with repos added
        """

    @abc.abstractmethod
    async def reset(
            self,
            cluster_uuid: str,
            force: bool = False,
            uninstall_sw: bool = False
    ) -> bool:
        """
        Uninstalls Tiller/Charm from a known K8s cluster and removes it from the list of known K8s clusters.
        Intended to be used e.g. when the NS instance is deleted.

        :param cluster_uuid: UUID of a K8s cluster known by OSM.
        :param force: force deletion, even in case there are deployed releases
        :param uninstall_sw: flag to indicate that sw uninstallation from software is needed
        :return: str: kdu_instance generated by helm
        """

    @abc.abstractmethod
    async def install(
            self,
            cluster_uuid: str,
            kdu_model: str,
            atomic: bool = True,
            timeout: float = 300,
            params: dict = None,
            db_dict: dict = None,
            kdu_name: str = None,
            namespace: str = None
    ):
        """
        Deploys of a new KDU instance. It would implicitly rely on the `install` call to deploy the Chart/Bundle
        properly parametrized (in practice, this call would happen before any _initial-config-primitive_
        of the VNF is called).

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param kdu_model: chart/bundle:version reference (string), which can be either of these options:
            - a name of chart/bundle available via the repos known by OSM
            - a path to a packaged chart/bundle
            - a path to an unpacked chart/bundle directory or a URL
        :param atomic: If set, installation process purges chart/bundle on fail, also will wait until
            all the K8s objects are active
        :param timeout: Time in seconds to wait for the install of the chart/bundle (defaults to
            Helm default timeout: 300s)
        :param params: dictionary of key-value pairs for instantiation parameters (overriding default values)
        :param dict db_dict: where to write into database when the status changes.
                        It contains a dict with {collection: <str>, filter: {},  path: <str>},
                            e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
        :param kdu_name: Name of the KDU instance to be installed
        :param namespace: K8s namespace to use for the KDU instance
        :return: True if successful
        """

    @abc.abstractmethod
    async def upgrade(
            self,
            cluster_uuid: str,
            kdu_instance: str,
            kdu_model: str = None,
            atomic: bool = True,
            timeout: float = 300,
            params: dict = None,
            db_dict: dict = None
    ):
        """
        Upgrades an existing KDU instance. It would implicitly use the `upgrade` call over an existing Chart/Bundle.
        It can be used both to upgrade the chart or to reconfigure it. This would be exposed as Day-2 primitive.

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param kdu_instance: unique name for the KDU instance to be updated
        :param kdu_model: new chart/bundle:version reference
        :param atomic: rollback in case of fail and wait for pods and services are available
        :param timeout: Time in seconds to wait for the install of the chart/bundle (defaults to
            Helm default timeout: 300s)
        :param params: new dictionary of key-value pairs for instantiation parameters
        :param dict db_dict: where to write into database when the status changes.
                        It contains a dict with {collection: <str>, filter: {},  path: <str>},
                            e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
        :return: reference to the new revision number of the KDU instance
        """

    @abc.abstractmethod
    async def rollback(
            self,
            cluster_uuid: str,
            kdu_instance: str,
            revision=0,
            db_dict: dict = None
    ):
        """
        Rolls back a previous update of a KDU instance. It would implicitly use the `rollback` call.
        It can be used both to rollback from a Chart/Bundle version update or from a reconfiguration.
        This would be exposed as Day-2 primitive.

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param kdu_instance: unique name for the KDU instance
        :param revision: revision to which revert changes. If omitted, it will revert the last update only
        :param dict db_dict: where to write into database when the status changes.
                        It contains a dict with {collection: <str>, filter: {},  path: <str>},
                            e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
        :return:If successful, reference to the current active revision of the KDU instance after the rollback
        """

    @abc.abstractmethod
    async def uninstall(
            self,
            cluster_uuid: str,
            kdu_instance: str
    ):
        """
        Removes an existing KDU instance. It would implicitly use the `delete` call (this call would happen
        after all _terminate-config-primitive_ of the VNF are invoked).

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param kdu_instance: unique name for the KDU instance to be deleted
        :return: True if successful
        """

    @abc.abstractmethod
    async def inspect_kdu(
            self,
            kdu_model: str,
            repo_url: str = None
    ) -> str:
        """
        These calls will retrieve from the Chart/Bundle:

            - The list of configurable values and their defaults (e.g. in Charts, it would retrieve
                the contents of `values.yaml`).
            - If available, any embedded help file (e.g. `readme.md`) embedded in the Chart/Bundle.

        :param kdu_model: chart/bundle reference
        :param repo_url: optional, reposotory URL (None if tar.gz, URl in other cases, even stable URL)
        :return:

        If successful, it will return the available parameters and their default values as provided by the backend.
        """

    @abc.abstractmethod
    async def help_kdu(
            self,
            kdu_model: str,
            repo_url: str = None
    ) -> str:
        """

        :param kdu_model: chart/bundle reference
        :param repo_url: optional, reposotory URL (None if tar.gz, URl in other cases, even stable URL)
        :return: If successful, it will return the contents of the 'readme.md'
        """

    @abc.abstractmethod
    async def status_kdu(
            self,
            cluster_uuid: str,
            kdu_instance: str
    ) -> str:
        """
        This call would retrieve tha current state of a given KDU instance. It would be would allow to retrieve
        the _composition_ (i.e. K8s objects) and _specific values_ of the configuration parameters applied
        to a given instance. This call would be based on the `status` call.

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param kdu_instance: unique name for the KDU instance
        :return: If successful, it will return the following vector of arguments:
        - K8s `namespace` in the cluster where the KDU lives
        - `state` of the KDU instance. It can be:
              - UNKNOWN
              - DEPLOYED
              - DELETED
              - SUPERSEDED
              - FAILED or
              - DELETING
        - List of `resources` (objects) that this release consists of, sorted by kind, and the status of those resources
        - Last `deployment_time`.

        """

    """
    ##################################################################################################
    ########################################## P R I V A T E #########################################
    ##################################################################################################
    """

    async def write_app_status_to_db(
            self,
            db_dict: dict,
            status: str,
            detailed_status: str,
            operation: str
    ) -> bool:

        if not self.db:
            self.warning('No db => No database write')
            return False

        if not db_dict:
            self.warning('No db_dict => No database write')
            return False

        self.log.debug('status={}'.format(status))

        try:

            the_table = db_dict['collection']
            the_filter = db_dict['filter']
            the_path = db_dict['path']
            if not the_path[-1] == '.':
                the_path = the_path + '.'
            update_dict = {
                the_path + 'operation': operation,
                the_path + 'status': status,
                the_path + 'detailed-status': detailed_status,
                the_path + 'status-time': str(time.time()),
            }

            self.db.set_one(
                table=the_table,
                q_filter=the_filter,
                update_dict=update_dict,
                fail_on_empty=True
            )

            # database callback
            if self.on_update_db:
                if asyncio.iscoroutinefunction(self.on_update_db):
                    await self.on_update_db(the_table, the_filter, the_path, update_dict)
                else:
                    self.on_update_db(the_table, the_filter, the_path, update_dict)

            return True

        except Exception as e:
            self.log.info('Exception writing status to database: {}'.format(e))
            return False