server.py

from time import sleep
from typing import TYPE_CHECKING, Any

from upcloud_api.firewall import FirewallRule
from upcloud_api.ip_address import IPAddress
from upcloud_api.server_group import ServerGroup
from upcloud_api.storage import STORAGE_OSES_WHICH_REQUIRE_METADATA, Storage
from upcloud_api.upcloud_resource import UpCloudResource
from upcloud_api.utils import try_it_n_times

if TYPE_CHECKING:
    from upcloud_api import CloudManager


def login_user_block(username, ssh_keys, create_password=False):
    """
    Helper function for creating Server.login_user blocks.

    (see: https://www.upcloud.com/api/8-servers/#create-server)
    """
    block = {
        'create_password': 'yes' if create_password else 'no',
        'ssh_keys': {'ssh_key': ssh_keys},
    }

    if username:
        block['username'] = username

    return block


class ServerNetworkInterface(UpCloudResource):
    """
    Class representation of server network interface
    """

    ATTRIBUTES = {
        'ip_addresses': [],
        'type': 'public',
        'network': None,
        'source_ip_filtering': 'yes',
    }

    def __init__(self, raw_dict, **kwargs):
        """
        Initialize network interface and set sane defaults
        """
        super().__init__(**kwargs)
        for k, v in raw_dict.items():
            if k in ServerNetworkInterface.ATTRIBUTES:
                setattr(self, k, v)

        if not raw_dict.get('ip_addresses'):
            self.ip_addresses = [{'family': 'IPv4'}]

    def to_dict(self):
        """
        Returns a dict implementation of a network interface to support server creation.
        """
        body = {
            'type': self.type,
            'ip_addresses': {
                'ip_address': self.ip_addresses,
            },
        }

        if hasattr(self, 'network'):
            body['network'] = self.network

        return body


# TODO: should this inherit from UpcloudResource too?
class Server:
    """
    Class representation of UpCloud Server instance.

    Partially immutable class; only fields that are persisted with the `.save()` method may be set
    with the server.field=value syntax. See __setattr__ override.
    """

    cloud_manager: 'CloudManager'

    #
    # Functionality for partial immutability and repopulating the object from API.
    #

    updateable_fields = [
        'boot_order',
        'core_number',
        'firewall',
        'hostname',
        'labels',
        'memory_amount',
        'nic_model',
        'plan',
        'simple_backup',
        'title',
        'timezone',
        'video_model',
        'vnc',
        'vnc_password',
    ]

    optional_fields = [
        'avoid_host',
        'boot_order',
        'core_number',
        'firewall',
        'labels',
        'login_user',
        'memory_amount',
        'networking',
        'nic_model',
        'password_delivery',
        'plan',
        'server_group',
        'simple_backup',
        'timezone',
        'metadata',
        'user_data',
        'video_model',
        'vnc_password',
    ]

    def __init__(self, server=None, **kwargs) -> None:
        """
        Initialize Server.

        Use _reset to set attributes.
        Set title = hostname if title not given.
        """
        object.__setattr__(self, 'populated', False)
        self._reset(server, **kwargs)

        if not hasattr(self, 'title') and hasattr(self, 'hostname'):
            self.title = self.hostname

    def __setattr__(self, name: str, value: Any) -> None:
        """
        Override to prevent updating readonly fields.
        """
        if name not in self.updateable_fields:
            raise Exception(f"'{name}' is a readonly field")
        else:
            object.__setattr__(self, name, value)

    def _reset(self, server, **kwargs) -> None:
        """
        Reset the server object with new values given as params.

        - server: a dict representing the server. e.g the API response.
        - kwargs: any meta fields such as cloud_manager and populated.

        Note: storage_devices and ip_addresses may be given in server as dicts or
        in kwargs as lists containing Storage and IPAddress objects.
        """
        if server:
            # handle storage, ip_address dicts and tags if they exist
            Server._handle_server_subobjs(server, kwargs.get('cloud_manager'))

            for key in server:
                object.__setattr__(self, key, server[key])

        for key in kwargs:
            object.__setattr__(self, key, kwargs[key])

    def populate(self) -> 'Server':
        """
        Sync changes from the API to the local object.

        Note: syncs ip_addresses and storage_devices too (/server/uuid endpoint)
        """
        server, IPAddresses, storages = self.cloud_manager.get_server_data(self.uuid)
        self._reset(server, ip_addresses=IPAddresses, storage_devices=storages, populated=True)
        return self

    def __str__(self) -> str:
        return self.uuid

    #
    # Main functionality, 1:1 with UpCloud's API
    #

    def save(self) -> None:
        """
        Sync local changes in server's attributes to the API.

        Note: DOES NOT sync IPAddresses and storage_devices,
        use add_ip, add_storage, remove_ip, remove_storage instead.
        """
        # dict comprehension that also works with 2.6
        # http://stackoverflow.com/questions/21069668/alternative-to-dict-comprehension-prior-to-python-2-7
        kwargs = {
            field: getattr(self, field) for field in self.updateable_fields if hasattr(self, field)
        }

        self.cloud_manager.modify_server(self.uuid, **kwargs)
        self._reset(kwargs)

    def destroy(self, delete_storages=False):
        """
        Destroy the server.
        """
        self.cloud_manager.delete_server(self.uuid, delete_storages=delete_storages)

    def shutdown(self, hard: bool = False, timeout: int = 30) -> None:
        """
        Shutdown/stop the server. By default, issue a soft shutdown with a timeout of 30s.

        After the a timeout a hard shutdown is performed if the server has not stopped.

        Note: API responds immediately (unlike in start), with state: started.
        This client will, however, set state as 'maintenance' to signal that the server is neither
        started nor stopped.
        """
        body = dict()
        body['stop_server'] = {'stop_type': 'hard' if hard else 'soft', 'timeout': f'{timeout}'}

        path = f'/server/{self.uuid}/stop'
        self.cloud_manager.api.post_request(path, body)
        object.__setattr__(self, 'state', 'maintenance')

    def stop(self) -> None:
        """
        Alias for shutdown.
        """
        self.shutdown()

    def start(self, timeout: int = 120) -> None:
        """
        Start the server. Note: slow and blocking request.

        The API waits for confirmation from UpCloud's IaaS backend before responding.
        """
        path = f'/server/{self.uuid}/start'
        self.cloud_manager.api.post_request(path, timeout=timeout)
        object.__setattr__(self, 'state', 'started')

    def restart(self, hard: bool = False, timeout: int = 30, force: bool = True) -> None:
        """
        Restart the server. By default, issue a soft restart with a timeout of 30s
        and a hard restart after the timeout.

        After the a timeout a hard restart is performed if the server has not stopped.

        Note: API responds immediately (unlike in start), with state: started.
        This client will, however, set state as 'maintenance' to signal that the server is neither
        started nor stopped.
        """
        body = dict()
        body['restart_server'] = {
            'stop_type': 'hard' if hard else 'soft',
            'timeout': f'{timeout}',
            'timeout_action': 'destroy' if force else 'ignore',
        }

        path = f'/server/{self.uuid}/restart'
        self.cloud_manager.api.post_request(path, body)
        object.__setattr__(self, 'state', 'maintenance')

    def add_ip(self, family: str = 'IPv4') -> IPAddress:
        """
        Allocate a new (random) IP-address to the Server.
        """
        IP = self.cloud_manager.attach_ip(self.uuid, family)
        self.ip_addresses.append(IP)
        return IP

    def remove_ip(self, ip_address: IPAddress) -> None:
        """
        Release the specified IP-address from the server.
        """
        self.cloud_manager.release_ip(ip_address.address)
        self.ip_addresses.remove(ip_address)

    def add_storage(
        self,
        storage: Storage | None = None,  # TODO: this probably shouldn't be optional
        type: str = 'disk',
        address=None,
    ) -> None:
        """
        Attach the given storage to the Server.

        Default address is next available.
        """
        self.cloud_manager.attach_storage(
            server=self.uuid, storage=storage.uuid, storage_type=type, address=address
        )
        storage.address = address
        storage.type = type
        self.storage_devices.append(storage)

    def remove_storage(self, storage: Storage) -> None:
        """
        Remove Storage from a Server.

        The Storage must be a reference to an object in
        Server.storage_devices or the method will throw and Exception.

        A Storage from get_storage(uuid) will not work as it is missing the 'address' property.
        """
        if not hasattr(storage, 'address'):
            raise Exception(
                'Storage does not have an address. '
                'Access the Storage via Server.storage_devices '
                'so they include an address. '
                '(This is due how the API handles Storages)'
            )

        self.cloud_manager.detach_storage(server=self.uuid, address=storage.address)
        self.storage_devices.remove(storage)

    def add_firewall_rule(self, firewall_rule: FirewallRule) -> FirewallRule:
        """
        Add the specified FirewallRule to this server.

        Returns a FirewallRule instance that is associated with this server instance.

        Instantly calls the API, no need to call .save(). This is because firewall can not
        be configured with the same request as the rest of the Server.
        """
        return self.cloud_manager.create_firewall_rule(self, firewall_rule.to_dict())

    def remove_firewall_rule(self, firewall_rule):
        """
        Remove a firewall rule.
        """
        return firewall_rule.destroy()

    def get_firewall_rules(self):
        """
        Return all FirewallRule instances that are associated with this server instance.
        """
        return self.cloud_manager.get_firewall_rules(self)

    def add_tags(self, tags):
        """
        Add tags to a server. Accepts tags as strings or Tag objects.
        """
        if self.cloud_manager.assign_tags(self.uuid, tags):
            tags = self.tags + [str(tag) for tag in tags]
            object.__setattr__(self, 'tags', tags)

    def remove_tags(self, tags):
        """
        Add tags to a server. Accepts tags as strings or Tag objects.
        """
        if self.cloud_manager.remove_tags(self, tags):
            new_tags = [tag for tag in self.tags if tag not in tags]
            object.__setattr__(self, 'tags', new_tags)

    #
    # Helper and convenience functions.
    # May perform several API requests and contain more complex logic.
    #

    def configure_firewall(self, FirewallRules):
        """
        Helper function for automatically adding several FirewallRules in series.
        """
        firewall_rule_bodies = [FirewallRule.to_dict() for FirewallRule in FirewallRules]
        return self.cloud_manager.configure_firewall(self, firewall_rule_bodies)

    def prepare_post_body(self):
        """
        Prepare a JSON serializable dict from a Server instance with nested.

        Storage instances.
        """
        body = dict()
        # mandatory
        body['server'] = {
            'hostname': self.hostname,
            'zone': self.zone,
            'title': self.title,
            'storage_devices': {},
        }

        # optional fields

        for optional_field in self.optional_fields:
            if hasattr(self, optional_field):
                body['server'][optional_field] = getattr(self, optional_field)

        if hasattr(self, 'labels'):
            dict_labels = {'label': []}
            for label in self.labels:
                dict_labels['label'].append(label.to_dict())
            body['server']['labels'] = dict_labels

        if hasattr(self, 'metadata') and isinstance(self.metadata, bool):
            body['server']['metadata'] = "yes" if self.metadata else "no"

        # metadata service has to be "yes" for certain OSes
        for storage in self.storage_devices:
            if (
                hasattr(storage, 'os')
                and storage.os
                and storage.os in STORAGE_OSES_WHICH_REQUIRE_METADATA
            ):
                body['server']['metadata'] = "yes"
                break

        if hasattr(self, 'server_group') and isinstance(self.server_group, ServerGroup):
            body['server']['server_group'] = f"{self.server_group.uuid}"

        # set password_delivery default as 'none' to prevent API from sending
        # emails (with credentials) about each created server
        if not hasattr(self, 'password_delivery'):
            # noqa reason: no, this is not a hard-coded password
            body['server']['password_delivery'] = 'none'  # noqa: S105

        # collect storage devices and create a unique title (see: Storage.title in API doc)
        # for each of them

        body['server']['storage_devices'] = {'storage_device': []}

        if hasattr(self, 'networking') and isinstance(self.networking, list):
            interfaces = []
            for iface in self.networking:
                if isinstance(iface, ServerNetworkInterface):
                    interfaces.append(iface.to_dict())
                else:
                    interfaces.append(iface)

            body['server']['networking'] = {'interfaces': {'interface': interfaces}}

        storage_title_id = 0  # running number for unique storage titles
        for storage in self.storage_devices:
            if not hasattr(storage, 'os') or storage.os is None:
                storage_title_id += 1
            storage_body = storage.to_dict()

            # setup default titles for storages unless the user has specified
            # them at storage.title
            if not hasattr(storage, 'title') or not storage.title:
                if hasattr(storage, 'os') and storage.os:
                    storage_body['title'] = self.hostname + ' OS disk'
                else:
                    storage_body['title'] = (
                        self.hostname + ' storage disk ' + str(storage_title_id)
                    )

            # figure out the storage `action` parameter
            # public template
            if hasattr(storage, 'os') and storage.os:
                storage_body['action'] = 'clone'
                storage_body['storage'] = storage.os

            # private template
            elif hasattr(storage, 'uuid'):
                storage_body['action'] = 'clone'
                storage_body['storage'] = storage.uuid

            # create a new storage
            else:
                storage_body['action'] = 'create'

            body['server']['storage_devices']['storage_device'].append(storage_body)

        if hasattr(self, 'ip_addresses') and self.ip_addresses:
            body['server']['ip_addresses'] = {
                'ip_address': [ip.to_dict() for ip in self.ip_addresses]
            }

        return body

    def to_dict(self):
        """
        Prepare a JSON serializable dict for read-only purposes.

        Includes storages and IP-addresses.
        Use prepare_post_body for POST and .save() for PUT.
        """
        fields = dict(vars(self).items())

        if self.populated:
            fields['ip_addresses'] = []
            fields['storage_devices'] = []
            for ip in self.ip_addresses:
                fields['ip_addresses'].append(
                    {'address': ip.address, 'access': ip.access, 'family': ip.family}
                )
            fields['networking'] = []
            for iface in dict.get(dict.get(self.networking, 'interfaces'), 'interface'):
                fields['networking'].append(ServerNetworkInterface(iface).to_dict())

            for storage in self.storage_devices:
                fields['storage_devices'].append(
                    {
                        'address': storage.address,
                        'storage': storage.uuid,
                        'storage_size': storage.size,
                        'storage_title': storage.title,
                        'type': storage.type,
                    }
                )

        del fields['populated']
        del fields['cloud_manager']
        return fields

    def get_ip(self, access='public', addr_family=None):
        """
        Return the server's IP address.

        Params:
        - addr_family: IPv4, IPv6 or None. None prefers IPv4 but will
                       return IPv6 if IPv4 addr was not available.
        - access: 'public' or 'private'
        """
        if addr_family not in ['IPv4', 'IPv6', None]:
            raise Exception("`addr_family` must be 'IPv4', 'IPv6' or None")

        if access not in ['private', 'public', 'utility']:
            raise Exception("`access` must be 'public', 'utility' or 'private'")

        if not hasattr(self, 'networking'):
            raise Exception(
                "`networking` attribute is missing, server details must be fetched first"
            )

        ip_addrs = []
        for iface in self.networking['interfaces']['interface']:
            iface_ip_addrs = iface['ip_addresses']['ip_address']
            if len(iface_ip_addrs) == 0:
                continue

            for ip in iface_ip_addrs:
                if iface['type'] == access and (not addr_family or ip['family'] == addr_family):
                    ip_addrs.append(ip)

        # If IP address family has not been defined, we'll prefer v4 when it's available
        if not addr_family:
            for addr in ip_addrs:
                if addr['family'] == 'IPv4':
                    return addr['address']

        # Any remaining IP should be good
        return ip_addrs[0]['address'] if ip_addrs else None

    def get_public_ip(self, addr_family=None, *args, **kwargs):
        """Alias for get_ip('public')"""
        return self.get_ip('public', addr_family, *args, **kwargs)

    def get_utility_ip(self, addr_family=None, *args, **kwargs):
        """Alias for get_ip('utility')"""
        return self.get_ip('utility', addr_family, *args, **kwargs)

    def get_private_ip(self, addr_family=None, *args, **kwargs):
        """Alias for get_ip('private')"""
        return self.get_ip('private', addr_family, *args, **kwargs)

    def _wait_for_state_change(self, target_states, update_interval=10):
        """
        Blocking wait until target_state reached. update_interval is in seconds.

        Warning: state change must begin before calling this method.
        """
        while self.state not in target_states:
            if self.state == 'error':
                raise Exception('server is in error state')

            # update server state every 10s
            sleep(update_interval)
            self.populate()

    def ensure_started(self):
        """
        Start a server and waits (blocking wait) until it is fully started.
        """
        # server is either starting or stopping (or error)
        if self.state in ['maintenance', 'error']:
            self._wait_for_state_change(['stopped', 'started'])

        if self.state == 'stopped':
            self.start()
            self._wait_for_state_change(['started'])

        if self.state == 'started':
            return True
        else:
            # something went wrong, fail explicitly
            raise Exception('unknown server state: ' + self.state)

    def stop_and_destroy(self, sync=True):
        """
        Destroy a server and its storages. Stops the server before destroying.

        Syncs the server state from the API, use sync=False to disable.
        """

        def _self_destruct():
            """destroy the server and all storages attached to it."""

            # try_it_n_times util is used as a convenience because
            # Servers and Storages can fluctuate between "maintenance" and their
            # original state due to several different reasons especially when
            # destroying infrastructure.

            # first destroy server
            try_it_n_times(
                operation=self.destroy,
                expected_error_codes=['SERVER_STATE_ILLEGAL'],
                custom_error='destroying server failed',
            )

            # storages may be deleted instantly after server DELETE
            for storage in self.storage_devices:
                try_it_n_times(
                    operation=storage.destroy,
                    expected_error_codes=['STORAGE_STATE_ILLEGAL'],
                    custom_error='destroying storage failed',
                )

        if sync:
            self.populate()

        # server is either starting or stopping (or error)
        if self.state in ['maintenance', 'error']:
            self._wait_for_state_change(['stopped', 'started'])

        if self.state == 'started':
            try_it_n_times(
                operation=self.stop,
                expected_error_codes=['SERVER_STATE_ILLEGAL'],
                custom_error='stopping server failed',
            )

            self._wait_for_state_change(['stopped'])

        if self.state == 'stopped':
            _self_destruct()
        else:
            raise Exception('unknown server state: ' + self.state)

    @classmethod
    def _handle_server_subobjs(cls, server, cloud_manager):
        ip_data = server.pop('ip_addresses', None)
        storage_data = server.pop('storage_devices', None)
        tags = server.pop('tags', None)

        if ip_data:
            ip_addresses = IPAddress._create_ip_address_objs(ip_data, cloud_manager=cloud_manager)
            server['ip_addresses'] = ip_addresses

        if storage_data:
            storages = Storage._create_storage_objs(storage_data, cloud_manager=cloud_manager)
            server['storage_devices'] = storages

        if tags and 'tag' in tags:
            server['tags'] = tags['tag']

    @classmethod
    def _create_server_obj(cls, server, cloud_manager):
        cls._handle_server_subobjs(server, cloud_manager)

        server_dict = dict()
        server_dict.update(server)
        server_dict['cloud_manager'] = cloud_manager

        return Server(**server_dict)