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 abc
import asyncio
from typing import Union
import time

from n2vc.loggable import Loggable


class K8sConnector(abc.ABC, Loggable):
    """
    ####################################################################################
    ################################### P U B L I C ####################################
    ####################################################################################
    """

    @staticmethod
    def generate_kdu_instance_name(**kwargs):
        raise NotImplementedError("Method not implemented")

    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",
        cert: str = None,
        user: str = None,
        password: str = None,
    ):
        """
        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,
        kdu_instance: 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 kdu_instance: Kdu instance name
        :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,
        reset_values: bool = False,
        reuse_values: bool = True,
        reset_then_reuse_values: bool = False,
        force: bool = False,
    ):
        """
        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"}
        :param reset_values: force reseting values
        :param reuse_values: force reusing values (default)
        :param reset_then_reuse_values: forces reseting values, then apply the last release's values
        :param force: force recreation of resources if necessary
        :return: reference to the new revision number of the KDU instance
        """

    @abc.abstractmethod
    async def scale(
        self,
        kdu_instance: str,
        scale: int,
        resource_name: str,
        total_timeout: float = 1800,
        cluster_uuid: str = None,
        kdu_model: str = None,
        atomic: bool = True,
        db_dict: dict = None,
        **kwargs,
    ) -> bool:
        """Scale a resource in a KDU instance.

        Args:
            kdu_instance: KDU instance name
            scale: Scale to which to set the resource
            resource_name: Resource name
            total_timeout: The time, in seconds, to wait for the install
                to finish
            cluster_uuid: The UUID of the cluster
            kdu_model: The chart/bundle reference
            atomic: if set, upgrade process rolls back changes made in case of failed upgrade.
                The --wait flag will be set automatically if --atomic is used
            db_dict: Dictionary for any additional data
            kwargs: Additional parameters
                vca_id (str): VCA ID

        Returns:
            True if successful, False otherwise
        """

    @abc.abstractmethod
    async def get_scale_count(
        self,
        resource_name: str,
        kdu_instance: str,
        cluster_uuid: str,
        kdu_model: str,
        timeout: float = 300,
        **kwargs,
    ) -> int:
        """Get a resource scale count in a KDU instance.

        Args:
            resource_name: Resource name
            kdu_instance: KDU instance name
            cluster_uuid: The UUID of the cluster
            kdu_model:    chart/bundle reference
            timeout:  The time, in seconds, to wait
            kwargs: Additional parameters

        Returns:
            Resource instance count
        """

    @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 exec_primitive(
        self,
        cluster_uuid: str = None,
        kdu_instance: str = None,
        primitive_name: str = None,
        timeout: float = 300,
        params: dict = None,
        db_dict: dict = None,
    ) -> str:
        """Exec primitive (Juju action)

        :param cluster_uuid str: The UUID of the cluster
        :param kdu_instance str: The unique name of the KDU instance
        :param primitive_name: Name of action that will be executed
        :param timeout: Timeout for action execution
        :param params: Dictionary of all the parameters needed for the action
        :db_dict: Dictionary for any additional data

        :return: Returns the output of the action
        """

    @abc.abstractmethod
    async def upgrade_charm(
        self,
        ee_id: str = None,
        path: str = None,
        charm_id: str = None,
        charm_type: str = None,
        timeout: float = None,
    ) -> str:
        """This method upgrade charms in VNFs

        Args:
            ee_id:  Execution environment id
            path:   Local path to the charm
            charm_id:   charm-id
            charm_type: Charm type can be lxc-proxy-charm, native-charm or k8s-proxy-charm
            timeout: (Float)    Timeout for the ns update operation

        Returns:
            The output of the update operation if status equals to "completed"
        """

    @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, yaml_format: str
    ) -> Union[str, dict]:
        """
        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
        :param yaml_format: if the return shall be returned as an YAML string or as a
                                dictionary
        :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`.

        """

    @abc.abstractmethod
    async def get_services(
        self, cluster_uuid: str, kdu_instance: str, namespace: str
    ) -> list:
        """
        Returns a list of services defined for the specified kdu instance.

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param kdu_instance: unique name for the KDU instance
        :param namespace: K8s namespace used by the KDU instance
        :return: If successful, it will return a list of services, Each service
        can have the following data:
        - `name` of the service
        - `type` type of service in the k8 cluster
        - `ports` List of ports offered by the service, for each port includes at least
        name, port, protocol
        - `cluster_ip` Internal ip to be used inside k8s cluster
        - `external_ip` List of external ips (in case they are available)
        """

    @abc.abstractmethod
    async def get_service(
        self, cluster_uuid: str, service_name: str, namespace: str = None
    ) -> object:
        """
        Obtains the data of the specified service in the k8cluster.

        :param cluster_uuid: UUID of a K8s cluster known by OSM
        :param service_name: name of the K8s service in the specified namespace
        :param namespace: K8s namespace used by the KDU instance
        :return: If successful, it will return a list of services, Each service can have
        the following data:
        - `name` of the service
        - `type` type of service in the k8 cluster
        - `ports` List of ports offered by the service, for each port includes at least
        name, port, protocol
        - `cluster_ip` Internal ip to be used inside k8s cluster
        - `external_ip` List of external ips (in case they are available)
        """

    """
    ####################################################################################
    ################################### 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:
        """
        This method will write the status of the application to the database.

        :param db_dict: A dictionary with the database necessary information. It shall contain the values for the keys:
            - "collection": The Mongo DB collection to write to
            - "filter": The query filter to use in the update process
            - "path": The dot separated keys which targets the object to be updated
        :param status: Status of the application
        :param detailed_status: Detailed status of the application
        :param operation: Operation that is being performed on the application
        :return: True if successful
        """

        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