Source code for libcloud.compute.drivers.vpsnet

# 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
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
VPS.net driver
"""
import base64

from libcloud.utils.py3 import b
from libcloud.common.base import JsonResponse, ConnectionUserAndKey
from libcloud.common.types import InvalidCredsError, MalformedResponseError
from libcloud.compute.base import Node, NodeSize, NodeImage, NodeDriver, NodeLocation
from libcloud.compute.types import NodeState
from libcloud.compute.providers import Provider

try:
    import simplejson as json
except ImportError:
    import json


API_HOST = "api.vps.net"
API_VERSION = "api10json"

RAM_PER_NODE = 256
DISK_PER_NODE = 10
BANDWIDTH_PER_NODE = 250


[docs]class VPSNetResponse(JsonResponse):
[docs] def parse_body(self): try: return super().parse_body() except MalformedResponseError: return self.body
[docs] def success(self): # vps.net wrongly uses 406 for invalid auth creds if self.status == 406 or self.status == 403: raise InvalidCredsError() return True
[docs] def parse_error(self): try: errors = super().parse_body()["errors"][0] except MalformedResponseError: return self.body else: return "\n".join(errors)
[docs]class VPSNetConnection(ConnectionUserAndKey): """ Connection class for the VPS.net driver """ host = API_HOST responseCls = VPSNetResponse allow_insecure = False
[docs] def add_default_headers(self, headers): user_b64 = base64.b64encode(b("{}:{}".format(self.user_id, self.key))) headers["Authorization"] = "Basic %s" % (user_b64.decode("utf-8")) return headers
[docs]class VPSNetNodeDriver(NodeDriver): """ VPS.net node driver """ type = Provider.VPSNET api_name = "vps_net" name = "vps.net" website = "http://vps.net/" connectionCls = VPSNetConnection def _to_node(self, vm): if vm["running"]: state = NodeState.RUNNING else: state = NodeState.PENDING n = Node( id=vm["id"], name=vm["label"], state=state, public_ips=[vm.get("primary_ip_address", None)], private_ips=[], extra={"slices_count": vm["slices_count"]}, # Number of nodes consumed by VM driver=self.connection.driver, ) return n def _to_image(self, image, cloud): image = NodeImage( id=image["id"], name="{}: {}".format(cloud, image["label"]), driver=self.connection.driver, ) return image def _to_size(self, num): size = NodeSize( id=num, name="%d Node" % (num,), ram=RAM_PER_NODE * num, disk=DISK_PER_NODE, bandwidth=BANDWIDTH_PER_NODE * num, price=self._get_price_per_node(num) * num, driver=self.connection.driver, ) return size def _get_price_per_node(self, num): single_node_price = self._get_size_price(size_id="1") return num * single_node_price
[docs] def create_node(self, name, image, size, ex_backups_enabled=False, ex_fqdn=None): """Create a new VPS.net node @inherits: :class:`NodeDriver.create_node` :keyword ex_backups_enabled: Enable automatic backups :type ex_backups_enabled: ``bool`` :keyword ex_fqdn: Fully Qualified domain of the node :type ex_fqdn: ``str`` """ ex_backups_enabled = 1 if ex_backups_enabled else 0 headers = {"Content-Type": "application/json"} request = { "virtual_machine": { "label": name, "fqdn": ex_fqdn or "", "system_template_id": image.id, "backups_enabled": ex_backups_enabled, "slices_required": size.id, } } res = self.connection.request( "/virtual_machines.{}".format(API_VERSION), data=json.dumps(request), headers=headers, method="POST", ) node = self._to_node(res.object["virtual_machine"]) return node
[docs] def reboot_node(self, node): res = self.connection.request( "/virtual_machines/{}/{}.{}".format(node.id, "reboot", API_VERSION), method="POST", ) node = self._to_node(res.object["virtual_machine"]) return True
[docs] def list_sizes(self, location=None): res = self.connection.request("/nodes.{}".format(API_VERSION)) available_nodes = len([size for size in res.object if size["slice"]["virtual_machine_id"]]) sizes = [self._to_size(i) for i in range(1, available_nodes + 1)] return sizes
[docs] def destroy_node(self, node): res = self.connection.request( "/virtual_machines/{}.{}".format(node.id, API_VERSION), method="DELETE" ) return res.status == 200
[docs] def list_nodes(self): res = self.connection.request("/virtual_machines.{}".format(API_VERSION)) return [self._to_node(i["virtual_machine"]) for i in res.object]
[docs] def list_images(self, location=None): res = self.connection.request("/available_clouds.{}".format(API_VERSION)) images = [] for cloud in res.object: label = cloud["cloud"]["label"] templates = cloud["cloud"]["system_templates"] images.extend([self._to_image(image, label) for image in templates]) return images
[docs] def list_locations(self): return [NodeLocation(0, "VPS.net Western US", "US", self)]