eid-ops/scripts/create-sc-instance

#!/usr/bin/env python3
# pylint: disable=invalid-name
# pylint: enable=invalid-name
"""
Tool used for creating CNaaS safespring machines the way we expect them to be created

It is formatted and linted in the following way:
===
black create-cnaas-instance
isort --profile black create-cnaas-instance
mypy --python-executable /opt/homebrew/bin/python3 --strict create-cnaas-instance
PYTHONPATH=/opt/homebrew/lib/python3.10/site-packages pylint create-cnaas-instance
===
"""

# Make code work on python 3.7
from __future__ import annotations

import argparse
import base64
import ipaddress
import re
import string
import sys
import textwrap
import typing
from typing import Any, Union

try:
    from typing import TypeGuard
except ImportError:
    from typing_extensions import (  # use `typing_extensions` for Python 3.9 and below
        TypeGuard,
    )

import openstack
from openstack.compute.v2.flavor import Flavor
from openstack.compute.v2.image import Image
from openstack.compute.v2.keypair import Keypair
from openstack.compute.v2.server import Server
from openstack.compute.v2.server_group import ServerGroup

# Make mypy type checking happy, also helps with go-to-definition support in editor
from openstack.connection import Connection
from openstack.exceptions import ConflictException
from openstack.network.v2.network import Network
from openstack.network.v2.port import Port
from openstack.network.v2.security_group import SecurityGroup


def get_keypair(conn: Connection, keypair_name: str) -> Keypair:
    """Lookup already existing keypair by name and return it"""
    keypair = conn.compute.find_keypair(keypair_name)

    if not keypair:
        raise RuntimeError(f"unable to find keypair with name '{keypair_name}'")

    if not isinstance(keypair, Keypair):
        raise TypeError(f"keypair should be {Keypair} but is {type(keypair)}")

    return keypair


def get_port(  # pylint:disable=too-many-arguments,too-many-branches
    args: argparse.Namespace,
    conn: Connection,
    network: Network,
    port_name: str,
) -> Port:
    """Either create a new named port or return already existing one"""

    for port in conn.network.ports():
        if not isinstance(port, Port):
            raise TypeError(f"port should be {Port} but is {type(port)}")
        if port.name == port_name:
            print(f"named port already exists, using it: {port.name} ({port.id})")
            return port

    print(f"creating port: {port_name}")
    port = conn.network.create_port(network_id=network.id, name=port_name)
    if not isinstance(port, Port):
        raise TypeError(f"port should be {Port} but is {type(port)}")

    return port


def is_int_list(test_list: list[Any]) -> TypeGuard[list[int]]:
    """Help with typechecking a list of ints"""
    return all(isinstance(element, int) for element in test_list)


def create_server(  # pylint: disable=too-many-arguments, too-many-locals
    conn: Connection,
    server_name: str,
    port: Port,
    image: Image,
    flavor: Flavor,
    keypair: Keypair,
    disk_size: int,
) -> Server:
    """Create a new server"""

    print(f"creating server: {server_name}")

    disk_sizes = [disk_size]

    block_mappings = []

    if flavor.disk == 0:
        print(
            f"flavor '{flavor.name}' includes no disk, using block storage for all disks"
        )
        volume_type_name = "fast"
        #volume_type = conn.volume.find_type(volume_type_name)
        #if volume_type is None:
            #raise RuntimeError(f"unable to find volume type '{volume_type_name}'")
        for i, size in enumerate(disk_sizes):
            block_mapping = {
                "delete_on_termination": True,
                "destination_type": "volume",
                "volume_size": size,
                #"volume_type": volume_type.name,
            }
            # The first disk is the OS disk, base it on the selected image and bootable
            if i == 0:
                block_mapping["boot_index"] = 0
                block_mapping["source_type"] = "image"
                block_mapping["uuid"] = image.id
            else:
                # Any aditional disk is a data disk, should not be booted from
                # and has no automatic contents
                block_mapping["source_type"] = "blank"

            block_mappings.append(block_mapping)
        server = conn.compute.create_server(
            name=server_name,
            flavor_id=flavor.id,
            networks=[{"port": port.id}],
            key_name=keypair.name,
            block_device_mapping_v2=block_mappings,
            disk_config="AUTO",
        )
    else:
        print(f"flavor '{flavor.name}' includes disk, using local storage")
        server = conn.compute.create_server(
            name=server_name,
            image_id=image.id,
            flavor_id=flavor.id,
            networks=[{"port": port.id}],
            key_name=keypair.name,
        )

    print(f"waiting for server to be active: {server_name}")
    server = conn.compute.wait_for_server(server)
    print(f"server is active: {server_name}")

    if not isinstance(server, Server):
        raise TypeError(f"server should be {Server} but is {type(server)}")

    return server


def main() -> (
    None
):  # pylint: disable=too-many-locals,too-many-statements,too-many-branches
    """Starting point of the program"""

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--cloud",
        help="name of the openstack cloud in your clouds.yaml, e.g: openstack",
        required=True,
    )
    parser.add_argument(
        "--flavor",
        help="name of the instance flavor to use, e.g: b2.c1r2",
        required=False,
    )
    parser.add_argument(
        "--image",
        help="name of the instance image to use, e.g: ubuntu-20.04",
        required=False,
    )
    parser.add_argument(
        "--fqdn",
        help="full name of server to create, e.g: example-ni1.cnaas.sunet.se",
        required=True,
    )
    parser.add_argument(
        "--key-name",
        help="name of existing SSH keypair to deploy server with, e.g: uid-yk1",
        required=True,
    )
    parser.add_argument(
        "--disk-size",
        help="provide disk size in GB",
        required=True,
    )
    parser.add_argument(
        "--network-name",
        help="The network name in safespring, ex 'public'",
        required=True,
    )
    parser.add_argument(
        "--verbose",
        help="include additional informational messages about what is being done",
        required=False,
        action="store_true",
    )
    parser.add_argument(
        "--debug",
        help="enable debug logging for openstack calls "
        "(lots of output, potentially sensitive information)",
        required=False,
        action="store_true",
    )
    args = parser.parse_args()

    if not args.flavor:
        print(
            "you must supply --flavor",
        )
        sys.exit(1)

    flavor_name = args.flavor

    if not args.image:
        print(
            "you must supply --image",
        )
        sys.exit(1)

    image_name = args.image

    # Configure logging. Keep in mind that debug logging will print sensitive
    # information like passwords to the screen.
    openstack.enable_logging(debug=args.debug)  # type: ignore

    # Initialize connection
    conn = openstack.connect(cloud=args.cloud)

    keypair = get_keypair(conn, args.key_name)

    network_name = args.network_name

    server = conn.compute.find_server(args.fqdn)
    if server is not None:
        print(f"server '{args.fqdn}' already exists")
        sys.exit(1)

    image = conn.image.find_image(image_name)
    if image is None:
        print(f"image '{image_name}' not found")
        sys.exit(1)

    flavor = conn.compute.find_flavor(flavor_name)
    if flavor is None:
        print(f"flavor '{flavor_name}' not found")
        sys.exit(1)

    network = conn.network.find_network(network_name)
    if network is None:
        print(f"network '{network_name}' not found")
        sys.exit(1)

    port = get_port(
        args,
        conn,
        network,
        args.fqdn,
    )

    create_server(
        conn,
        args.fqdn,
        port,
        image,
        flavor,
        keypair,
        args.disk_size,
    )

    print("DNS records (needs to be created manually):")
    for fixed_ip in port.fixed_ips:
        address = ipaddress.ip_address(fixed_ip["ip_address"])
        if isinstance(address, ipaddress.IPv4Address):
            print(f"{args.fqdn}.	A	{address}")
        if isinstance(address, ipaddress.IPv6Address):
            print(f"{args.fqdn}.	AAAA	{address}")


if __name__ == "__main__":
    main()