Source code for libcloud.compute.drivers.gridscale

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import time

from libcloud.compute.base import (
from libcloud.utils.iso8601 import parse_date
from libcloud.common.gridscale import GridscaleBaseDriver, GridscaleConnection
from libcloud.compute.providers import Provider

[docs]class GridscaleIp: """ Ip Object :param id: uuid :type id: ``str`` :param family: family of ip (v4 or v6) :type family: ``str`` :param prefix: prefix of ip :type prefix: ``str`` :param ip_address: Ip address :type ip_address: ``str`` :param create_time: Time ip was created :type create_time: ``str`` """ def __init__(self, id, family, prefix, create_time, address, extra=None): = id = family self.prefix = prefix self.create_time = create_time self.ip_address = address self.extra = extra or {} def __repr__(self): return "Ip: id={}, family={}, prefix={}, create_time={}, " "ip_address={}".format(,, self.prefix, self.create_time, self.ip_address )
[docs]class GridscaleNetwork: """ Network Object :param id: uuid :type id: ``str`` :param name: Name of Network :type name: ``str`` :param status: Network status :type status: ``str`` :param relations: object related to network :type relations: ``object`` :param create_time: Time Network was created :type create_time: ``str`` """ def __init__(self, id, name, status, create_time, relations): = id = name self.status = status self.create_time = create_time self.relations = relations def __repr__(self): return "Network: id={}, name={}, status={}, create_time={}, " "relations={}".format(,, self.status, self.create_time, self.relations )
[docs]class GridscaleNodeDriver(GridscaleBaseDriver, NodeDriver): """ create and entry in libcloud/compute/providers for gridscale """ connectionCls = GridscaleConnection type = Provider.GRIDSCALE name = "Gridscale" api_name = "gridscale" website = "" features = {"create_node": ["ssh_key"]} def __init__(self, user_id, key, **kwargs): super().__init__(user_id, key, **kwargs)
[docs] def list_nodes(self): """ List all nodes. :return: List of node objects :rtype: ``list`` of :class:`.Node` """ result = self._sync_request(data=None, endpoint="objects/servers/") nodes = [] for key, value in self._get_response_dict(result).items(): node = self._to_node(value) nodes.append(node) return sorted(nodes, key=lambda sort: sort.created_at)
[docs] def list_locations(self): """ List all available data centers. :return: List of node location objects :rtype: ``list`` of :class:`.NodeLocation` """ locations = [] result = self._sync_request(endpoint="objects/locations/") for key, value in self._get_response_dict(result).items(): location = self._to_location(value) locations.append(location) return sorted(locations, key=lambda nod:
[docs] def list_volumes(self): """ List all volumes. :return: List of StorageVolume object :rtype: ``list`` of :class:`.StorageVolume` """ volumes = [] result = self._sync_request(endpoint="objects/storages/") for key, value in self._get_response_dict(result).items(): volume = self._to_volume(value) volumes.append(volume) return sorted(volumes, key=lambda sort: sort.extra["create_time"])
[docs] def ex_list_networks(self): """ List all networks. :return: List of objects. :rtype: ``list`` of :class:`.GridscaleNetwork` """ networks = [] result = self._sync_request(endpoint="objects/networks/") for key, value in self._get_response_dict(result).items(): network = self._to_network(value) networks.append(network) return sorted(networks, key=lambda sort: sort.create_time)
[docs] def list_volume_snapshots(self, volume): """ Lists all snapshots for storage volume. :param volume: storage the snapshot is attached to :type volume: :class:`.StorageVolume` :return: Snapshots :rtype: ``list`` of :class:`.VolumeSnapshot` """ snapshots = [] result = self._sync_request(endpoint="objects/storages/" "{}/snapshots".format( for key, value in self._get_response_dict(result).items(): snapshot = self._to_volume_snapshot(value) snapshots.append(snapshot) return sorted(snapshots, key=lambda snapshot: snapshot.created)
[docs] def ex_list_ips(self): """ Lists all IPs available. :return: List of IP objects. :rtype: ``list`` of :class:`.GridscaleIp` """ ips = [] result = self._sync_request(endpoint="objects/ips/") for key, value in self._get_response_dict(result).items(): ip = self._to_ip(value) ips.append(ip) return ips
[docs] def list_images(self): """ List images. :return: List of node image objects :rtype: ``list`` of :class:`.NodeImage` """ templates = [] result = self._sync_request(endpoint="objects/templates") for key, value in self._get_response_dict(result).items(): template = self._to_node_image(value) templates.append(template) return sorted(templates, key=lambda sort:
[docs] def create_node(self, name, size, image, location, ex_ssh_key_ids=None, **kwargs): """ Create a simple node with a name, cores, memory at the designated location. :param name: Name of the server. :type name: ``str`` :param size: Nodesize object. :type size: :class:`.NodeSize` :param image: OS image to attach to the storage. :type image: :class:`.GridscaleTemplate` :param location: The data center to create a node in. :type location: :class:`.NodeLocation` :keyword ex_ssh_key_ids: List of SSH key IDs to add to the server. :type ex_ssh_key_ids: ``list`` of ``str`` :return: The newly created Node. :rtype: :class:`.Node` """ if size.ram % 1024 != 0: raise Exception("Value not accepted. Use a multiple of 1024 e.g." "1024, 2048, 3072...") data = { "name": name, "cores": size.extra["cores"], "memory": int(size.ram / 1024), "location_uuid":, } self.connection.async_request("objects/servers/", data=data, method="POST") node = self._to_node( self._get_resource( "servers", self.connection.poll_response_initial.object["object_uuid"] ) ) volume = self._create_volume_from_template( name=image.extra["ostype"], size=size.disk, location=location, template={"template_uuid":, "sshkeys": ex_ssh_key_ids}, ) ip = self.ex_create_ip(4, location, name + "_ip") self.attach_volume(node, volume) self.ex_link_ip_to_node(node, ip) self.ex_link_network_to_node(node, self.ex_list_networks()[0]) self.ex_start_node(node) return self._to_node(self._get_resource("servers",
[docs] def ex_create_ip(self, family, location, name): """ Create either an ip_v4 ip or a ip_v6. :param family: Defines if the ip is v4 or v6 with int 4 or int 6. :type family: ``int`` :param location: Defines which datacenter the created ip responds with. :type location: :class:`.NodeLocation` :param name: Name of your Ip. :type name: ``str`` :return: Ip :rtype: :class:`.GridscaleIp` """ self.connection.async_request( "objects/ips/", data={"name": name, "family": family, "location_uuid":}, method="POST", ) return self._to_ip( self._get_resource("ips", self.connection.poll_response_initial.object["object_uuid"]) )
[docs] def ex_create_networks(self, name, location): """ Create a network at the data center location. :param name: Name of the network. :type name: ``str`` :param location: Location. :type location: :class:`.NodeLocation` :return: Network. :rtype: :class:`.GridscaleNetwork` """ self.connection.async_request( "objects/networks", data={"name": name, "location_uuid":}, method="POST", ) return self._to_network( self._get_resource( "network", self.connection.poll_response_initial.object["object_uuid"] ) )
[docs] def create_volume(self, size, name, location=None, snapshot=None): """ Create a new volume. :param size: Integer in GB. :type size: ``int`` :param name: Name of the volume. :type name: ``str`` :param location: The server location. :type location: :class:`.NodeLocation` :param snapshot: Snapshot from which to create the new volume. (optional) :type snapshot: :class:`.VolumeSnapshot` :return: Newly created StorageVolume. :rtype: :class:`.StorageVolume` """ return self._create_volume_from_template(size, name, location)
def _create_volume_from_template(self, size, name, location=None, template=None): """ create Storage :param name: name of your Storage unit :type name: ``str`` :param size: Integer in GB. :type size: ``int`` :param location: your server location :type location: :class:`.NodeLocation` :param template: template to shape the storage capacity to :type template: ``dict`` :return: newly created StorageVolume :rtype: :class:`.GridscaleVolumeStorage` """ template = template self.connection.async_request( "objects/storages/", data={ "name": name, "capacity": size, "location_uuid":, "template": template, }, method="POST", ) return self._to_volume( self._get_resource( "storages", self.connection.poll_response_initial.object["object_uuid"] ) )
[docs] def create_volume_snapshot(self, volume, name): """ Creates a snapshot of the current state of your volume, you can rollback to. :param volume: Volume you want to create a snapshot of. :type volume: :class:`.StorageVolume` :param name: Name of the snapshot. :type name: ``str`` :return: VolumeSnapshot. :rtype: :class:`.VolumeSnapshot` """ self.connection.async_request( "objects/storages/{}/snapshots".format(, data={"name": name}, method="POST", ) return self._to_volume_snapshot( self._get_resource( "storages/{}/snapshots".format(, self.connection.poll_response_initial.object["object_uuid"], ) )
[docs] def create_image(self, node, name): """ Creates an image from a node object. :param node: Node to run the task on. :type node: :class:`.Node` :param name: Name for new image. :type name: ``str`` :return: NodeImage. :rtype: :class:`.NodeImage` """ storage_dict = node.extra["relations"]["storages"][0] snapshot_uuid = "" if storage_dict["bootdevice"] is True: self.connection.async_request( "objects/storages/{}/snapshots/".format(storage_dict["object_uuid"]), data={"name": name + "_snapshot"}, method="POST", ) snapshot_uuid = self.connection.poll_response_initial.object["object_uuid"] self.connection.async_request( "objects/templates/", data={"name": name, "snapshot_uuid": snapshot_uuid}, method="POST", ) snapshot_dict = self._get_response_dict( self._sync_request( endpoint="objects/storages/{}/snapshots/{}".format( storage_dict["object_uuid"], snapshot_uuid ) ) ) self.destroy_volume_snapshot(self._to_volume_snapshot(snapshot_dict)) return self._to_node_image( self._get_resource( "templates", self.connection.poll_response_initial.object["object_uuid"] ) )
[docs] def destroy_node(self, node, ex_destroy_associated_resources=False): """ Destroy node. :param node: Node object. :type node: :class:`.Node` :param ex_destroy_associated_resources: True to destroy associated resources such as storage volumes and IPs. :type ex_destroy_associated_resources: ``bool`` :return: True if the destroy was successful, otherwise False. :rtype: ``bool`` """ if ex_destroy_associated_resources: associated_volumes = self.ex_list_volumes_for_node(node=node) associated_ips = self.ex_list_ips_for_node(node=node) # 1. Delete the server itself result = self._sync_request(endpoint="objects/servers/{}".format(, method="DELETE") # 2. Destroy associated resources (if requested) if ex_destroy_associated_resources: for volume in associated_volumes: self.destroy_volume(volume=volume) for ip in associated_ips: self.ex_destroy_ip(ip=ip) return result.status == 204
[docs] def destroy_volume(self, volume): """ Delete volume. :param volume: Volume to be destroyed. :type volume: :class:`.StorageVolume` :return: True if the destroy was successful, otherwise False. :rtype: ``bool`` """ result = self._sync_request( endpoint="objects/storages/{}".format(, method="DELETE" ) return result.status == 204
[docs] def ex_destroy_ip(self, ip): """ Delete an ip. :param ip: IP object. :type ip: :class:`.GridscaleIp` :return: ``True`` if delete_image was successful, ``False`` otherwise. :rtype: ``bool`` """ result = self._sync_request(endpoint="objects/ips/{}".format(, method="DELETE") return result.status == 204
[docs] def destroy_volume_snapshot(self, snapshot): """ Destroy a snapshot. :param snapshot: The snapshot to delete. :type snapshot: :class:'.VolumeSnapshot` :return: True if the destroy was successful, otherwise False. :rtype: ``bool`` """ result = self._sync_request( endpoint="objects/storages/" "{}/snapshots/{}/".format(snapshot.extra["parent_uuid"],, method="DELETE", ) return result.status == 204
[docs] def ex_destroy_network(self, network): """ Delete network. :param network: Network object. :type network: :class:`.GridscaleNetwork` :return: ``True`` if destroyed successfully, otherwise ``False`` :rtype: ``bool`` """ result = self._sync_request( endpoint="objects/networks/{}".format(, method="DELETE" ) return result.status == 204
[docs] def delete_image(self, node_image): """ Destroy an image. :param node_image: Node image object. :type node_image: :class:`.NodeImage` :return: True if the destroy was successful, otherwise False :rtype: ``bool`` """ result = self._sync_request( endpoint="objects/templates/{}".format(, method="DELETE" ) return result.status == 204
[docs] def ex_rename_node(self, node, name): """ Modify node name. :param name: New node name. :type name: ``str`` :param node: Node :type node: :class:`.Node` :return: ``True`` or ``False`` :rtype: ``bool`` """ result = self._sync_request( data={"name": name}, endpoint="objects/servers/{}".format(, method="PATCH", ) return result.status == 204
[docs] def ex_rename_volume(self, volume, name): """ Modify storage volume name :param volume: Storage. :type volume: :class:.`StorageVolume` :param name: New storage name. :type name: ``str`` :return: ``True`` or ``False`` :rtype: ``bool`` """ result = self._sync_request( data={"name": name}, endpoint="objects/storages/{}".format(, method="PATCH", ) return result.status == 204
[docs] def ex_rename_network(self, network, name): """ Modify networks name. :param network: Network. :type network: :class:`.GridscaleNetwork` :param name: New network name. :type name: ``str`` :return: ``True`` or ``False`` :rtype: ``bool`` """ result = self._sync_request( data={"name": name}, endpoint="objects/networks/{}".format(, method="PATCH", ) return result.status == 204
[docs] def reboot_node(self, node, ex_sleep_interval=3): """ Reboot a node. :param node: Node object. :type node: :class:`.Node` :return: True if the reboot was successful, otherwise False. :rtype: ``bool`` :keyword ex_sleep_interval: time to let the shutdown process finish :type ex_sleep_interval: ``int`` """ if node.extra["power"] is True: data = dict({"power": False}) self._sync_request( data=data, endpoint="objects/servers/{}/power".format(, method="PATCH", ) time.sleep(ex_sleep_interval) data = dict({"power": True}) self._sync_request( data=data, endpoint="objects/servers/{}/power".format(, method="PATCH", ) return True else: return False
[docs] def import_key_pair_from_string(self, name, key_material): data = {"name": name, "sshkey": key_material} result = self._sync_request(endpoint="objects/sshkeys/", method="POST", data=data) key = self._to_key(result.object, name=name, sshkey=key_material) return key
[docs] def list_key_pairs(self): """ List all the available key pair objects. :rtype: ``list``of :class:`.KeyPair` objects """ keys = [] result = self._sync_request(endpoint="objects/sshkeys/") for key, value in self._get_response_dict(result).items(): key = self._to_key(value) keys.append(key) return keys
[docs] def get_image(self, image_id): """ Get an image based on an image_id. :param image_id: Image identifier. :type image_id: ``str`` :return: A NodeImage object. :rtype: :class:`.NodeImage` """ response_dict = self._get_response_dict( self._sync_request(endpoint="/objects/templates/{}".format(image_id)) ) return self._to_node_image(response_dict)
[docs] def start_node(self, node): result = self._sync_request( data={"power": True}, endpoint="objects/servers/{}/power".format(, method="PATCH", ) return result.status == 204
[docs] def ex_start_node(self, node): # NOTE: This method is here for backward compatibility reasons after # this method was promoted to be part of the standard compute API in # Libcloud v2.7.0 return self.start_node(node=node)
[docs] def attach_volume(self, node, volume): """ Attaches volume to node. :param node: Node to attach volume to. :type node: :class:`.Node` :param volume: Volume to attach. :type volume: :class:`.StorageVolume` :rytpe: ``bool`` """ result = self._sync_request( data={"object_uuid":}, endpoint="objects/servers/{}/storages/".format(, method="POST", ) return result.status == 204
[docs] def detach_volume(self, volume): """ Detaches a volume from a node. :param volume: Volume to be detached :type volume: :class:`.StorageVolume` :rtype: ``bool`` """ node = volume.extra["relations"]["servers"][0] result = self._sync_request( endpoint="objects/servers/{}/storages/{}".format(node["object_uuid"],, method="DELETE", ) return result.status == 204
[docs] def ex_storage_rollback(self, volume, snapshot, rollback): """ initiate a rollback on your storage :param volume: storage uuid :type volume: ``string`` :param snapshot: snapshot uuid :type snapshot: ``string`` :param rollback: variable :type rollback: ``bool`` :return: RequestID :rtype: ``str`` """ result = self._sync_request( data={"rollback": rollback}, endpoint="objects/storages/{}/snapshots/" "{}/rollback".format(,, method="PATCH", ) return result
[docs] def ex_list_volumes_for_node(self, node): """ Return a list of associated volumes for the provided node. :rtype: ``list`` of :class:`StorageVolume` """ volumes = self.list_volumes() result = [] for volume in volumes: related_servers = volume.extra.get("relations", {}).get("servers", []) for server in related_servers: if server["object_uuid"] == result.append(volume) return result
[docs] def ex_list_ips_for_node(self, node): """ Return a list of associated IPs for the provided node. :rype: ``list`` of :class:`GridscaleIp` """ ips = self.ex_list_ips() result = [] for ip in ips: related_servers = ip.extra.get("relations", {}).get("servers", []) for server in related_servers: # TODO: This is not consistent with volumes where key is # called "object_uuid" if server["server_uuid"] == result.append(ip) return result
def _to_node(self, data): extra_keys = ["cores", "power", "memory", "current_price", "relations"] extra = self._extract_values_to_dict(data=data, keys=extra_keys) ips = [] for diction in data["relations"]["public_ips"]: ips.append(diction["ip"]) state = "" if data["power"] is True: state = NodeState.RUNNING else: state = NodeState.STOPPED node = Node( id=data["object_uuid"], name=data["name"], state=state, public_ips=ips, created_at=parse_date(data["create_time"]), private_ips=None, driver=self.connection.driver, extra=extra, ) return node def _to_volume(self, data): extra_keys = ["create_time", "current_price", "storage_type", "relations"] extra = self._extract_values_to_dict(data=data, keys=extra_keys) storage = StorageVolume( id=data["object_uuid"], name=data["name"], size=data["capacity"], driver=self.connection.driver, extra=extra, ) return storage def _to_volume_snapshot(self, data): extra_keys = [ "labels", "status", "usage_in_minutes", "location_country", "current_price", "parent_uuid", ] extra = self._extract_values_to_dict(data=data, keys=extra_keys) volume_snapshot = VolumeSnapshot( id=data["object_uuid"], driver=self.connection.driver, size=data["capacity"], extra=extra, created=parse_date(data["create_time"]), state=StorageVolumeState.AVAILABLE, name=data["name"], ) return volume_snapshot def _to_location(self, data): location = NodeLocation( id=data["object_uuid"], name=data["name"], country=data["country"], driver=self.connection.driver, ) return location def _to_ip(self, data): extra_keys = [ "create_time", "current_price", "name", "relations", "reverse_dns", "status", ] extra = self._extract_values_to_dict(data=data, keys=extra_keys) ip = GridscaleIp( id=data["object_uuid"], family=data["family"], prefix=data["prefix"], create_time=data["create_time"], address=data["ip"], extra=extra, ) return ip def _to_network(self, data): network = GridscaleNetwork( id=data["object_uuid"], name=data["name"], create_time=data["create_time"], status=data["status"], relations=data["relations"], ) return network def _to_node_image(self, data): extra_keys = [ "capacity", "create_time", "labels", "ostype", "location_name", "private", "status", "usage_in_minutes", "version", ] extra = self._extract_values_to_dict(data=data, keys=extra_keys) template = NodeImage( id=data["object_uuid"], name=data["name"], driver=self.connection.driver, extra=extra, ) return template def _to_key(self, data, name=None, sshkey=None): extra = {"uuid": data["object_uuid"], "labels": data.get("labels", [])} name = data.get("name", name) sshkey = data.get("sshkey", sshkey) key = KeyPair( name=name, fingerprint=data["object_uuid"], public_key=sshkey, private_key=None, extra=extra, driver=self.connection.driver, ) return key def _extract_values_to_dict(self, data, keys): """ Extract extra values to dict. :param data: dict to extract values from. :type data: ``dict`` :param keys: keys to extract :type keys: ``List`` :return: dictionary containing extra values :rtype: ``dict`` """ result = {} for key in keys: if key == "memory": result[key] = data[key] * 1024 else: result[key] = data[key] return result def _get_response_dict(self, raw_response): """ Get the actual response dictionary. :param raw_response: Nested dictionary. :type raw_response: ``dict`` :return: Not-nested dictionary. :rtype: ``dict`` """ return list(raw_response.object.values())[0] def _get_resource(self, endpoint_suffix, object_uuid): """ Get specific uuid specific resource. :param endpoint_suffix: Endpoint resource e.g. servers/. :type endpoint_suffix: ``str`` :param object_uuid: Uuid of resource to be pulled. :type object_uuid: ``str`` :return: Response dictionary. :rtype: nested ``dict`` """ data = self._sync_request(endpoint="objects/{}/{}".format(endpoint_suffix, object_uuid)) data = self._get_response_dict(data) return data