#!/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()