22

✏️ Editing: equinixmetal.py

# 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.
"""
Equinix Metal Driver
"""

try:  # Try to use asyncio to perform requests in parallel across projects
    import asyncio
except ImportError:  # If not available will do things serially
    asyncio = None

import json

from libcloud.utils.py3 import httplib
from libcloud.common.base import JsonResponse, ConnectionKey
from libcloud.compute.base import Node, KeyPair, NodeSize, NodeImage, NodeDriver, NodeLocation
from libcloud.compute.types import Provider, NodeState, InvalidCredsError

EQUINIXMETAL_ENDPOINT = "api.equinix.com"

# True to use async io if available (aka running under Python 3)
USE_ASYNC_IO_IF_AVAILABLE = True

def use_asyncio():
    return asyncio is not None and USE_ASYNC_IO_IF_AVAILABLE

class EquinixMetalResponse(JsonResponse):
    valid_response_codes = [
        httplib.OK,
        httplib.ACCEPTED,
        httplib.CREATED,
        httplib.NO_CONTENT,
    ]

def parse_error(self):
        if self.status == httplib.UNAUTHORIZED:
            body = self.parse_body()
            raise InvalidCredsError(body.get("error"))
        else:
            body = self.parse_body()

if "message" in body:
                error = "{} (code: {})".format(body.get("message"), self.status)
            elif "errors" in body:
                error = body.get("errors")
            else:
                error = body
            raise Exception(error)

def success(self):
        return self.status in self.valid_response_codes

class EquinixMetalConnection(ConnectionKey):
    """
    Connection class for the Equinix Metal driver.
    """

host = EQUINIXMETAL_ENDPOINT
    responseCls = EquinixMetalResponse

def add_default_headers(self, headers):
        """
        Add headers that are necessary for every request
        """
        headers["Content-Type"] = "application/json"
        headers["X-Auth-Token"] = self.key
        headers["X-Consumer-Token"] = (
            "kcrhMn7hwG8Ceo2hAhGFa2qpxLBvVHxEjS9ue8iqmsNkeeB2iQgMq4dNc1893pYu"
        )

return headers

class EquinixMetalNodeDriver(NodeDriver):
    """
    Equinix Metal NodeDriver
    """

connectionCls = EquinixMetalConnection
    type = Provider.EQUINIXMETAL
    name = "EquinixMetal"
    website = "https://metal.equinix.com/"

NODE_STATE_MAP = {
        "queued": NodeState.PENDING,
        "provisioning": NodeState.PENDING,
        "rebuilding": NodeState.PENDING,
        "powering_on": NodeState.REBOOTING,
        "powering_off": NodeState.REBOOTING,
        "rebooting": NodeState.REBOOTING,
        "inactive": NodeState.STOPPED,
        "deleted": NodeState.TERMINATED,
        "deprovisioning": NodeState.TERMINATED,
        "failed": NodeState.ERROR,
        "active": NodeState.RUNNING,
    }

def __init__(self, key, project=None):
        """
        Initialize a NodeDriver for Equinix Metal using the API token
        and optionally the project (name or id).

If project name is specified we validate it lazily and populate
        self.project_id during the first access of self.projects variable
        """
        super().__init__(key=key)

self.project_name = project
        self.project_id = None

# Lazily populated on first access to self.project
        self._project = project

# Variable which indicates if self._projects has been populated yet and
        # has been called self._project validated
        self._projects_populated = False
        self._projects = None

@property
    def projects(self):
        """
        Lazily retrieve projects and set self.project_id variable on initial
        access to self.projects variable.
        """

if not self._projects_populated:
            # NOTE: Each EquinixMetal account needs at least one project,
            # but to be on the safe side and avoid infinite loop
            # in case there are no projects on the account, we don't use
            # a more robust way to determine
            # if project list has been populated yet
            self._projects = self.ex_list_projects()
            self._projects_populated = True

# If project name is specified, verify it's valid and populate
            # self.project_id

if self._project:
                for project_obj in self._projects:
                    if self._project in [project_obj.name, project_obj.id]:
                        self.project_id = project_obj.id

break

if not self.project_id:
                    # Invalid project name
                    self.project_name = None

return self._projects

def ex_list_projects(self):
        projects = []
        data = self.connection.request("/metal/v1/projects").object
        projects = data.get("projects")

if projects:
            projects = [Project(project) for project in projects]

return projects

def list_nodes(self, ex_project_id=None):
        if ex_project_id:
            return self.ex_list_nodes_for_project(ex_project_id=ex_project_id)

# if project has been specified during driver initialization, then
        # return nodes for this project only

if self.project_id:
            return self.ex_list_nodes_for_project(ex_project_id=self.project_id)

# In case of Python2 perform requests serially

if not use_asyncio():
            nodes = []

for project in self.projects:
                nodes.extend(self.ex_list_nodes_for_project(ex_project_id=project.id))

return nodes
        # In case of Python3 use asyncio to perform requests in parallel

return self.list_resources_async("nodes")

def list_resources_async(self, resource_type):
        # The _list_nodes function is defined dynamically using exec in
        # order to prevent a SyntaxError in Python2 due to "yield from".
        # This cruft can be removed once Python2 support is no longer
        # required.
        assert resource_type in ["nodes"]
        glob = globals()
        loc = locals()
        exec(
            """
import asyncio
@asyncio.coroutine
def _list_async(driver):
    projects = [project.id for project in driver.projects]
    loop = asyncio.get_event_loop()
    futures = [
        loop.run_in_executor(None, driver.ex_list_%s_for_project, p)
        for p in projects
    ]
    retval = []
    for future in futures:
        result = yield from future
        retval.extend(result)

return retval""" % resource_type,
            glob,
            loc,
        )
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            asyncio.set_event_loop(asyncio.new_event_loop())
            loop = asyncio.get_event_loop()

return loop.run_until_complete(loc["_list_async"](loc["self"]))

def ex_list_nodes_for_project(self, ex_project_id, include="plan", page=1, per_page=1000):
        params = {"include": include, "page": page, "per_page": per_page}
        data = self.connection.request(
            "/metal/v1/projects/%s/devices" % (ex_project_id), params=params
        ).object["devices"]

return list(map(self._to_node, data))

def list_locations(self):
        data = self.connection.request("/metal/v1/locations/metros").object["metros"]

return list(map(self._to_location, data))

def list_images(self):
        data = self.connection.request("/metal/v1/operating-systems").object["operating_systems"]

return list(map(self._to_image, data))

def list_sizes(self, ex_project_id=None):
        project_id = (
            ex_project_id or self.project_id or (len(self.projects) and self.projects[0].id)
        )

if project_id:
            data = self.connection.request("/metal/v1/projects/%s/plans" % project_id).object[
                "plans"
            ]
        else:  # This only works with personal tokens
            data = self.connection.request("/metal/v1/plans").object["plans"]

return [self._to_size(size) for size in data if size.get("line") == "baremetal"]

def create_node(
        self,
        name,
        size,
        image,
        location,
        ex_project_id=None,
        ip_addresses=None,
        cloud_init=None,
        **kwargs,
    ):
        """
        Create a node.

:return: The newly created node.
        :rtype: :class:`Node`
        """
        # if project has been specified on initialization of driver, then
        # create on this project

if self.project_id:
            ex_project_id = self.project_id
        else:
            if not ex_project_id:
                raise Exception("ex_project_id needs to be specified")

location_code = location.extra["code"]

if not self._valid_location:
            raise ValueError(
                "Failed to create node: valid parameter metro [code] is required in the input"
            )

params = {
            "hostname": name,
            "plan": size.id,
            "operating_system": image.id,
            "metro": location_code,
            "include": "plan",
            "billing_cycle": "hourly",
        }

if ip_addresses:
            params["ip_addresses"] = ip_addresses
        params.update(kwargs)

if cloud_init:
            params["userdata"] = cloud_init
        data = self.connection.request(
            "/metal/v1/projects/%s/devices" % (ex_project_id),
            data=json.dumps(params),
            method="POST",
        )

status = data.object.get("status", "OK")

if status == "ERROR":
            message = data.object.get("message", None)
            error_message = data.object.get("error_message", message)
            raise ValueError("Failed to create node: %s" % (error_message))
        node = self._to_node(data=data.object)

return node

def reboot_node(self, node):
        params = {"type": "reboot"}
        res = self.connection.request(
            "/metal/v1/devices/%s/actions" % (node.id), params=params, method="POST"
        )