Source code for libcloud.compute.drivers.voxel

# 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.

Voxel VoxCloud driver
import datetime
import hashlib

from libcloud.utils.py3 import b

from libcloud.common.base import XmlResponse, ConnectionUserAndKey
from libcloud.common.types import InvalidCredsError
from libcloud.compute.providers import Provider
from libcloud.compute.types import NodeState
from libcloud.compute.base import Node, NodeDriver
from libcloud.compute.base import NodeSize, NodeImage, NodeLocation


[docs]class VoxelResponse(XmlResponse): def __init__(self, response, connection): self.parsed = None super(VoxelResponse, self).__init__(response=response, connection=connection)
[docs] def parse_body(self): if not self.body: return None if self.parsed is None: self.parsed = super(VoxelResponse, self).parse_body() return self.parsed
[docs] def parse_error(self): err_list = [] if not self.body: return None if self.parsed is None: self.parsed = super(VoxelResponse, self).parse_body() for err in self.parsed.findall("err"): code = err.get("code") err_list.append("(%s) %s" % (code, err.get("msg"))) # From voxel docs: # 1: Invalid login or password # 9: Permission denied: user lacks access rights for this method if code == "1" or code == "9": # sucks, but only way to detect # bad authentication tokens so far raise InvalidCredsError(err_list[-1]) return "\n".join(err_list)
[docs] def success(self): if not self.parsed: self.parsed = super(VoxelResponse, self).parse_body() stat = self.parsed.get("stat") if stat != "ok": return False return True
[docs]class VoxelConnection(ConnectionUserAndKey): """ Connection class for the Voxel driver """ host = VOXEL_API_HOST responseCls = VoxelResponse
[docs] def add_default_params(self, params): params = dict([(k, v) for k, v in list(params.items()) if v is not None]) params["key"] = self.user_id params["timestamp"] = datetime.datetime.utcnow().isoformat() + "+0000" keys = list(params.keys()) keys.sort() md5 = hashlib.md5() md5.update(b(self.key)) for key in keys: if params[key]: if not params[key] is None: md5.update(b("%s%s" % (key, params[key]))) else: md5.update(b(key)) params["api_sig"] = md5.hexdigest() return params
VOXEL_INSTANCE_TYPES = {} RAM_PER_CPU = 2048 NODE_STATE_MAP = { "IN_PROGRESS": NodeState.PENDING, "QUEUED": NodeState.PENDING, "SUCCEEDED": NodeState.RUNNING, "shutting-down": NodeState.TERMINATED, "terminated": NodeState.TERMINATED, "unknown": NodeState.UNKNOWN, }
[docs]class VoxelNodeDriver(NodeDriver): """ Voxel VoxCLOUD node driver """ connectionCls = VoxelConnection type = Provider.VOXEL name = "Voxel VoxCLOUD" website = "" def _initialize_instance_types(): # pylint: disable=no-method-argument for cpus in range(1, 14): if cpus == 1: name = "Single CPU" else: name = "%d CPUs" % cpus id = "%dcpu" % cpus ram = cpus * RAM_PER_CPU VOXEL_INSTANCE_TYPES[id] = { "id": id, "name": name, "ram": ram, "disk": None, "bandwidth": None, "price": None, } features = {"create_node": [], "list_sizes": ["variable_disk"]} _initialize_instance_types() # pylint: disable=too-many-function-args
[docs] def list_nodes(self): params = {"method": "voxel.devices.list"} result = self.connection.request("/", params=params).object return self._to_nodes(result)
[docs] def list_sizes(self, location=None): return [ NodeSize(driver=self.connection.driver, **i) for i in list(VOXEL_INSTANCE_TYPES.values()) ]
[docs] def list_images(self, location=None): params = {"method": "voxel.images.list"} result = self.connection.request("/", params=params).object return self._to_images(result)
[docs] def create_node( self, name, size, image, location, ex_privateip=None, ex_publicip=None, ex_rootpass=None, ex_consolepass=None, ex_sshuser=None, ex_sshpass=None, ex_voxel_access=None, ): """Create Voxel Node :keyword name: the name to assign the node (mandatory) :type name: ``str`` :keyword image: distribution to deploy :type image: :class:`NodeImage` :keyword size: the plan size to create (mandatory) Requires size.disk (GB) to be set manually :type size: :class:`NodeSize` :keyword location: which datacenter to create the node in :type location: :class:`NodeLocation` :keyword ex_privateip: Backend IP address to assign to node; must be chosen from the customer's private VLAN assignment. :type ex_privateip: ``str`` :keyword ex_publicip: Public-facing IP address to assign to node; must be chosen from the customer's public VLAN assignment. :type ex_publicip: ``str`` :keyword ex_rootpass: Password for root access; generated if unset. :type ex_rootpass: ``str`` :keyword ex_consolepass: Password for remote console; generated if unset. :type ex_consolepass: ``str`` :keyword ex_sshuser: Username for SSH access :type ex_sshuser: ``str`` :keyword ex_sshpass: Password for SSH access; generated if unset. :type ex_sshpass: ``str`` :keyword ex_voxel_access: Allow access Voxel administrative access. Defaults to False. :type ex_voxel_access: ``bool`` :rtype: :class:`Node` or ``None`` """ # assert that disk > 0 if not size.disk: raise ValueError("size.disk must be non-zero") # convert voxel_access to string boolean if needed if ex_voxel_access is not None: ex_voxel_access = "true" if ex_voxel_access else "false" params = { "method": "voxel.voxcloud.create", "hostname": name, "disk_size": int(size.disk), "facility":, "image_id":, "processing_cores": size.ram / RAM_PER_CPU, "backend_ip": ex_privateip, "frontend_ip": ex_publicip, "admin_password": ex_rootpass, "console_password": ex_consolepass, "ssh_username": ex_sshuser, "ssh_password": ex_sshpass, "voxel_access": ex_voxel_access, } object = self.connection.request("/", params=params).object if self._getstatus(object): return Node( id=object.findtext("device/id"), name=name, state=NODE_STATE_MAP[object.findtext("device/status")], public_ips=ex_publicip, private_ips=ex_privateip, driver=self.connection.driver, ) else: return None
[docs] def reboot_node(self, node): params = { "method": "voxel.devices.power", "device_id":, "power_action": "reboot", } return self._getstatus(self.connection.request("/", params=params).object)
[docs] def destroy_node(self, node): params = {"method": "voxel.voxcloud.delete", "device_id":} return self._getstatus(self.connection.request("/", params=params).object)
[docs] def list_locations(self): params = {"method": "voxel.voxcloud.facilities.list"} result = self.connection.request("/", params=params).object nodes = self._to_locations(result) return nodes
def _getstatus(self, element): status = element.attrib["stat"] return status == "ok" def _to_locations(self, object): return [ NodeLocation( element.attrib["label"], element.findtext("description"), element.findtext("description"), self, ) for element in object.findall("facilities/facility") ] def _to_nodes(self, object): nodes = [] for element in object.findall("devices/device"): if element.findtext("type") == "Virtual Server": try: state = self.NODE_STATE_MAP[element.attrib["status"]] except KeyError: state = NodeState.UNKNOWN public_ip = private_ip = None ipassignments = element.findall("ipassignments/ipassignment") for ip in ipassignments: if ip.attrib["type"] == "frontend": public_ip = ip.text elif ip.attrib["type"] == "backend": private_ip = ip.text nodes.append( Node( id=element.attrib["id"], name=element.attrib["label"], state=state, public_ips=public_ip, private_ips=private_ip, driver=self.connection.driver, ) ) return nodes def _to_images(self, object): images = [] for element in object.findall("images/image"): images.append( NodeImage( id=element.attrib["id"], name=element.attrib["summary"], driver=self.connection.driver, ) ) return images