wally/openstack.py

import os.path
import socket
import logging
from typing import Dict, Any, List, Tuple, cast, Optional

from cephlib.common import to_ip
from cephlib.node import NodeInfo
from cephlib.ssh import ConnCreds

from .config import ConfigBlock, Config
from .openstack_api import (os_connect, find_vms, OSConnection,
                            OSCreds, get_openstack_credentials_from_env, prepare_os, launch_vms, clear_nodes)
from .test_run_class import TestRun
from .stage import Stage, StepOrder
from .utils import LogError, StopTestError, get_creds_openrc


logger = logging.getLogger("wally")


def get_floating_ip(vm: Any) -> str:
    """Get VM floating IP address"""

    for net_name, ifaces in vm.addresses.items():
        for iface in ifaces:
            if iface.get('OS-EXT-IPS:type') == "floating":
                return iface['addr']

    raise ValueError("VM {} has no floating ip".format(vm))


def ensure_connected_to_openstack(ctx: TestRun) -> None:
    if not ctx.os_connection is None:
        if ctx.os_creds is None:
            ctx.os_creds = get_OS_credentials(ctx)
        ctx.os_connection = os_connect(ctx.os_creds)


def get_OS_credentials(ctx: TestRun) -> OSCreds:
    stored = ctx.storage.get("openstack_openrc", None)
    if stored is not None:
        return OSCreds(*cast(List, stored))

    creds: OSCreds = None  # type: ignore
    os_creds: OSCreds = None  # type: ignore
    force_insecure = False
    cfg = ctx.config

    if 'openstack' in cfg.clouds:
        os_cfg = cfg.clouds['openstack']
        if 'OPENRC' in os_cfg:
            sett = os_cfg['OPENRC']
            if isinstance(sett, str):
                if 'ENV' == sett:
                    logger.info("Using OS credentials from shell environment")
                    os_creds = get_openstack_credentials_from_env()
                else:
                    logger.info("Using OS credentials from " + os_cfg['OPENRC'])
                    creds_tuple = get_creds_openrc(sett)
                    os_creds = OSCreds(*creds_tuple)
            else:
                logger.info("Using predefined credentials")
                os_creds = OSCreds(sett['OS_USERNAME'].strip(),
                                   sett['OS_PASSWORD'].strip(),
                                   sett['OS_TENANT_NAME'].strip(),
                                   sett['OS_AUTH_URL'].strip(),
                                   sett.get('OS_INSECURE', False))

        if 'insecure' in os_cfg:
            force_insecure = os_cfg.get('insecure', False)

    if os_creds is None:
        logger.error("Can't found OS credentials")
        raise StopTestError("Can't found OS credentials", None)

    if creds is None:
        creds = os_creds

    if force_insecure and not creds.insecure:
        creds = OSCreds(creds.name, creds.passwd, creds.tenant, creds.auth_url, True)

    logger.debug(("OS_CREDS: user={0.name} tenant={0.tenant} " +
                  "auth_url={0.auth_url} insecure={0.insecure}").format(creds))

    ctx.storage.put(list(creds), "openstack_openrc")
    return creds


def get_vm_keypair_path(cfg: Config) -> Tuple[str, str]:
    key_name = cfg.vm_configs['keypair_name']
    private_path = os.path.join(cfg.settings_dir, key_name + "_private.pem")
    public_path = os.path.join(cfg.settings_dir, key_name + "_public.pub")
    return (private_path, public_path)


class DiscoverOSStage(Stage):
    """Discover openstack nodes and VMS"""

    config_block = 'openstack'

    priority = StepOrder.DISCOVER

    @classmethod
    def validate(cls, conf: ConfigBlock) -> None:
        pass

    def run(self, ctx: TestRun) -> None:
        if 'openstack' not in ctx.config.discover:
            logger.debug("Skip openstack discovery due to settings")
            return

        if 'all_nodes' in ctx.storage:
            logger.debug("Skip openstack discovery, use previously discovered nodes")
            return

        ensure_connected_to_openstack(ctx)

        os_conn: OSConnection = ctx.os_connection  # type: ignore  # remove Optional[]

        cfg = ctx.config.openstack
        os_nodes_auth = cfg.auth  # type: str
        if os_nodes_auth.count(":") == 2:
            user, password, key_file = os_nodes_auth.split(":")  # type: str, Optional[str], Optional[str]
            if not password:
                password = None
        else:
            user, password = os_nodes_auth.split(":")
            key_file = None

        if 'metadata' not in ctx.config.discover:
            services: List[Any] = os_conn.nova.services.list()
            host_services_mapping: Dict[str, List[str]] = {}

            for service in services:
                ip = cast(str, socket.gethostbyname(service.host))
                host_services_mapping.get(ip, []).append(service.binary)

            logger.debug("Found %s openstack service nodes" % len(host_services_mapping))

            for host, services in host_services_mapping.items():
                host_ip = to_ip(host)
                if host != host_ip:
                    logger.info("Will use ip_addr %r instead of hostname %r", host_ip, host)
                creds = ConnCreds(host=host_ip, user=user, passwd=password, key_file=key_file)
                ctx.merge_node(creds, set(services))
            # TODO: log OS nodes discovery results
        else:
            logger.info("Skip OS cluster discovery due to 'discovery' setting value")

        private_key_path = get_vm_keypair_path(ctx.config)[0]

        for vm_creds in cfg.get("vms", []):
            user_name, vm_name_pattern = vm_creds.split("@", 1)  # type: ignore
            msg = "Vm like {} lookup failed".format(vm_name_pattern)

            with LogError(msg):
                msg = "Looking for vm with name like {0}".format(vm_name_pattern)
                logger.debug(msg)

                ensure_connected_to_openstack(ctx)

                for ip, vm_id in find_vms(os_conn, vm_name_pattern):  # type: ignore
                    creds = ConnCreds(host=to_ip(ip), user=user_name, key_file=private_key_path)
                    info = NodeInfo(creds, {'testnode'})
                    info.os_vm_id = vm_id
                    nid = info.node_id
                    if nid in ctx.nodes_info:
                        logger.error("Test VM node has the same id(%s), as existing node %s", nid, ctx.nodes_info[nid])
                        raise StopTestError()
                    ctx.nodes_info[nid] = info


class CreateOSVMSStage(Stage):
    "Spawn new VM's in Openstack cluster"

    priority = StepOrder.SPAWN  # type: int
    config_block = 'spawn_os_vms'  # type: str

    def run(self, ctx: TestRun) -> None:
        if 'all_nodes' in ctx.storage:
            ctx.os_spawned_nodes_ids = ctx.storage.get('os_spawned_nodes_ids')
            logger.info("Skipping OS VMS discovery/spawn as all data found in storage")
            return

        if 'os_spawned_nodes_ids' in ctx.storage:
            logger.error("spawned_os_nodes_ids is found in storage, but no nodes_info is stored." +
                         "Fix this before continue")
            raise StopTestError()

        vm_spawn_config = ctx.config.spawn_os_vms
        vm_image_config = ctx.config.vm_configs[vm_spawn_config.cfg_name]

        ensure_connected_to_openstack(ctx)
        os_conn: OSConnection = ctx.os_connection  # type: ignore  # remove Optional[]

        params = vm_image_config.copy()
        params.update(vm_spawn_config)
        params.update(get_vm_keypair_path(ctx.config))
        params['group_name'] = ctx.config.run_uuid
        params['keypair_name'] = ctx.config.vm_configs['keypair_name']

        if not ctx.config.openstack.get("skip_preparation", False):
            logger.info("Preparing openstack")
            prepare_os(os_conn, params)
        else:
            logger.info("Scip openstack preparation as 'skip_preparation' is set")

        ctx.os_spawned_nodes_ids = []
        with ctx.get_pool() as pool:
            for info in launch_vms(os_conn, params, pool):
                info.roles.add('testnode')
                nid = info.node_id
                if nid in ctx.nodes_info:
                    logger.error("Test VM node has the same id(%s), as existing node %s", nid, ctx.nodes_info[nid])
                    raise StopTestError()
                ctx.nodes_info[nid] = info
                ctx.os_spawned_nodes_ids.append(info.os_vm_id)

        ctx.storage.put(ctx.os_spawned_nodes_ids, 'os_spawned_nodes_ids')

    def cleanup(self, ctx: TestRun) -> None:
        # keep nodes in case of error for future test restart
        if not ctx.config.keep_vm and ctx.os_spawned_nodes_ids:
            logger.info("Removing nodes")

            clear_nodes(ctx.os_connection, ctx.os_spawned_nodes_ids)  # type: ignore
            ctx.storage.rm('spawned_os_nodes')

            logger.info("OS spawned nodes has been successfully removed")