Merge pull request #29 from aznashwan/licensing

Add support for interactions with the licensing server.

coriolis/conductor/rpc/server.py

@@ -14,6 +14,7 @@ from coriolis.db import api as db_api
 from coriolis.db.sqlalchemy import models
 from coriolis import exception
 from coriolis import keystone
+from coriolis.licensing import client as licensing_client
 from coriolis.replica_cron.rpc import client as rpc_cron_client
 from coriolis import schemas
 from coriolis import utils
@@ -86,9 +87,61 @@ def tasks_execution_synchronized(func):
 
 class ConductorServerEndpoint(object):
     def __init__(self):
+        self._licensing_client = licensing_client.LicensingClient.from_env()
         self._rpc_worker_client = rpc_worker_client.WorkerClient()
         self._replica_cron_client = rpc_cron_client.ReplicaCronClient()
 
+    def _check_delete_reservation_for_transfer(self, transfer_action):
+        action_id = transfer_action.base_id
+        if not self._licensing_client:
+            LOG.warn(
+                "Licensing client not instantiated. Skipping deletion of "
+                "reservation for transfer action '%s'", action_id)
+            return
+
+        reservation_id = transfer_action.reservation_id
+        if reservation_id:
+            try:
+                self._licensing_client.delete_reservation(reservation_id)
+            except (Exception, KeyboardInterrupt):
+                LOG.warn(
+                    "Failed to delete reservation with ID '%s' for transfer "
+                    "action with ID '%s'. Skipping. Exception\n%s",
+                    reservation_id, action_id, utils.get_exception_details())
+
+    def _check_create_reservation_for_transfer(
+            self, transfer_action, transfer_type):
+        action_id = transfer_action.base_id
+        if not self._licensing_client:
+            LOG.warn(
+                "Licensing client not instantiated. Skipping creation of "
+                "reservation for transfer action '%s'", action_id)
+            return
+
+        ninstances = len(transfer_action.instances)
+        LOG.debug(
+            "Attempting to create '%s' reservation for %d instances for "
+            "transfer action with ID '%s'.",
+            transfer_type, ninstances, action_id)
+        reservation = self._licensing_client.add_reservation(
+            transfer_type, ninstances)
+        transfer_action.reservation_id = reservation['id']
+
+    def _check_reservation_for_transfer(self, transfer_action):
+        action_id = transfer_action.base_id
+        if not self._licensing_client:
+            LOG.warn(
+                "Licensing client not instantiated. Skipping checking of "
+                "reservation for transfer action '%s'", action_id)
+            return
+
+        reservation_id = transfer_action.reservation_id
+        if reservation_id:
+            LOG.debug(
+                "Attempting to check reservation with ID '%s' for transfer "
+                "action '%s'", reservation_id, action_id)
+        self._licensing_client.check_reservation(reservation_id)
+
     def create_endpoint(self, ctxt, name, endpoint_type, description,
                         connection_info):
         endpoint = models.Endpoint()
@@ -244,6 +297,7 @@ class ConductorServerEndpoint(object):
     @replica_synchronized
     def execute_replica_tasks(self, ctxt, replica_id, shutdown_instances):
         replica = self._get_replica(ctxt, replica_id)
+        self._check_reservation_for_transfer(replica)
         self._check_replica_running_executions(ctxt, replica)
         execution = models.TasksExecution()
         execution.id = str(uuid.uuid4())
@@ -352,6 +406,7 @@ class ConductorServerEndpoint(object):
     def delete_replica(self, ctxt, replica_id):
         replica = self._get_replica(ctxt, replica_id)
         self._check_replica_running_executions(ctxt, replica)
+        self._check_delete_reservation_for_transfer(replica)
         db_api.delete_replica(ctxt, replica_id)
 
     @replica_synchronized
@@ -413,6 +468,9 @@ class ConductorServerEndpoint(object):
         replica.network_map = network_map
         replica.storage_mappings = storage_mappings
 
+        self._check_create_reservation_for_transfer(
+            replica, licensing_client.RESERVATION_TYPE_REPLICA)
+
         db_api.add_replica(ctxt, replica)
         LOG.info("Replica created: %s", replica.id)
         return self.get_replica(ctxt, replica.id)
@@ -481,6 +539,7 @@ class ConductorServerEndpoint(object):
     def deploy_replica_instances(self, ctxt, replica_id, clone_disks, force,
                                  skip_os_morphing=False):
         replica = self._get_replica(ctxt, replica_id)
+        self._check_reservation_for_transfer(replica)
         self._check_replica_running_executions(ctxt, replica)
         self._check_valid_replica_tasks_execution(replica, force)
 
@@ -621,6 +680,9 @@ class ConductorServerEndpoint(object):
         migration.info = {}
         migration.notes = notes
 
+        self._check_create_reservation_for_transfer(
+            migration, licensing_client.RESERVATION_TYPE_MIGRATION)
+
         for instance in instances:
             task_validate = self._create_task(
                 instance, constants.TASK_TYPE_VALIDATE_MIGRATION_INPUTS,
@@ -714,6 +776,7 @@ class ConductorServerEndpoint(object):
                 "The migration is not running")
         execution = migration.executions[0]
         self._cancel_tasks_execution(ctxt, execution, force)
+        self._check_delete_reservation_for_transfer(migration)
 
     def _cancel_tasks_execution(self, ctxt, execution, force=False):
         has_running_tasks = False
@@ -936,9 +999,16 @@ class ConductorServerEndpoint(object):
         task = db_api.get_task(ctxt, task_id)
         execution = db_api.get_tasks_execution(ctxt, task.execution_id)
 
-        with lockutils.lock(execution.action_id):
+        action_id = execution.action_id
+        with lockutils.lock(action_id):
             self._cancel_tasks_execution(ctxt, execution)
 
+        # NOTE: if this is a migration, make sure to delete
+        # its associated reservation.
+        action = db_api.get_action(ctxt, action_id)
+        if action.type == "migration":
+            self._check_delete_reservation_for_transfer(action)
+
     @task_synchronized
     def task_event(self, ctxt, task_id, level, message):
         LOG.info("Task event: %s", task_id)

coriolis/db/sqlalchemy/migrate_repo/versions/010_adds_reservation_id.py

@@ -0,0 +1,17 @@
+# Copyright 2018 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sqlalchemy
+
+
+def upgrade(migrate_engine):
+    meta = sqlalchemy.MetaData()
+    meta.bind = migrate_engine
+
+    # add 'reservation_id' column to 'base_transfer_action':
+    base_transfer_action = sqlalchemy.Table(
+        'base_transfer_action', meta, autoload=True)
+
+    reservation_id = sqlalchemy.Column(
+        "reservation_id", sqlalchemy.String(36), nullable=True)
+    base_transfer_action.create_column(reservation_id)

coriolis/db/sqlalchemy/models.py

@@ -103,6 +103,7 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
                                   "base_id==TasksExecution.action_id, "
                                   "TasksExecution.deleted=='0')")
     instances = sqlalchemy.Column(types.List, nullable=False)
+    reservation_id = sqlalchemy.Column(sqlalchemy.String(36), nullable=True)
     info = sqlalchemy.Column(types.Bson, nullable=False)
     notes = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
     origin_endpoint_id = sqlalchemy.Column(

coriolis/exception.py

@@ -124,7 +124,7 @@ class NotAuthorized(CoriolisException):
 
 
 class PolicyNotAuthorized(CoriolisException):
-    message = _("Not authorized via policy.")
+    message = _("Policy doesn't allow %(action)s to be performed.")
     code = 403
     safe = True
 
@@ -139,10 +139,6 @@ class AdminRequired(NotAuthorized):
     message = _("User does not have admin privileges")
 
 
-class PolicyNotAuthorized(NotAuthorized):
-    message = _("Policy doesn't allow %(action)s to be performed.")
-
-
 class Invalid(CoriolisException):
     message = _("Unacceptable parameters.")
     code = 400

coriolis/licensing/client.py

@@ -0,0 +1,182 @@
+# Copyright 2019 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import json
+import os
+import requests
+
+from coriolis import exception
+from coriolis import utils
+from oslo_log import log as logging
+
+
+LOG = logging.getLogger(__name__)
+
+RESERVATION_TYPE_REPLICA = "replica"
+RESERVATION_TYPE_MIGRATION = "migration"
+
+
+class LicensingClient(object):
+    """ Class for accessing the Coriolis licensing server API. """
+
+    def __init__(self, base_url, allow_untrusted=False):
+        """ :param base_url: URL for the API service, including scheme """
+        self._base_url = base_url.rstrip('/')
+        self._verify = not allow_untrusted
+
+    @classmethod
+    def from_env(cls):
+        """ Retuns a `LicensingClient` object instatiated using the
+        following env vars:
+        LICENSING_SERVER_BASE_URL="https://10.7.2.3:37667/v1"
+        LICENSING_SERVER_ALLOW_UNTRUSTED="<set to anything>"
+        Returns None if 'LICENSING_SERVER_BASE_URL' is not defined.
+        """
+        base_url = os.environ.get("LICENSING_SERVER_BASE_URL")
+        if not base_url:
+            LOG.warn(
+                "No 'LICENSING_SERVER_BASE_URL' env var present. Cannot "
+                "instantiate licensing client.")
+            return None
+        allow_untrusted = os.environ.get(
+            "LICENSING_SERVER_ALLOW_UNTRUSTED", False)
+
+        # try out client:
+        client = cls(base_url, allow_untrusted=allow_untrusted)
+        client.get_licence_status()
+
+        return client
+
+    def _get_url_for_resource(self, resource):
+        """ Provides full URL for subresource.
+        Ex: "licences" -> "http://$host:$port/v1/licences"
+        """
+        return "%s/%s" % (self._base_url, resource.strip('/'))
+
+    @utils.retry_on_error()
+    def _do_req(
+            self, method_name, resource, body=None,
+            response_key=None, raw_response=False):
+        method = getattr(requests, method_name.lower(), None)
+        if not method:
+            raise ValueError("No such HTTP method '%s'" % method_name)
+
+        url = self._get_url_for_resource(resource)
+
+        kwargs = {"verify": self._verify}
+        if body:
+            if not isinstance(body, (str, bytes)):
+                body = json.dumps(body)
+            kwargs["data"] = body
+
+        LOG.debug(
+            "Making '%s' call to licensing server at '%s' with body: %s",
+            method_name, url, kwargs.get('data'))
+        resp = method(url, **kwargs)
+
+        if raw_response:
+            return resp
+
+        if not resp.ok:
+            # try to extract error from licensing server:
+            error = None
+            try:
+                error = resp.json().get('error', {})
+            except (Exception, KeyboardInterrupt):
+                LOG.debug(
+                    "Exception occured during error extraction from licensing "
+                    "response: '%s'\nException:\n%s",
+                    resp.text, utils.get_exception_details())
+            if error and all([x in error for x in ['code', 'message']]):
+                raise exception.Conflict(
+                    message=error['message'],
+                    code=int(error['code']))
+            else:
+                resp.raise_for_status()
+
+        resp_data = resp.json()
+        if response_key:
+            if response_key not in resp_data:
+                raise ValueError(
+                    "No response key '%s' in response body: %s" % (
+                        response_key, resp_data))
+            resp_data = resp_data[response_key]
+
+        return resp_data
+
+    def _get(self, resource, response_key=None):
+        return self._do_req("GET", resource, response_key=response_key)
+
+    def _post(self, resource, body, response_key=None):
+        return self._do_req(
+            "POST", resource, body=body, response_key=response_key)
+
+    def _put(self, resource, body, response_key=None):
+        return self._do_req(
+            "PUT", resource, body=body, response_key=response_key)
+
+    def _delete(self, resource, body, response_key=None):
+        return self._do_req(
+            "DELETE", resource, body=body, response_key=response_key)
+
+    def get_licence_status(self):
+        """ Gets licence status for appliance. """
+        return self._get("/licence-status", "licence_status")
+
+    def get_licences(self):
+        """ Lists all installed licences. """
+        return self._get("/licences", response_key="licences")
+
+    def add_licence(self, licence_data):
+        """ Sends request to add licence (in .PEM format). """
+        return self._post("/licences", licence_data)
+
+    def add_reservation(self, reservation_type, num_vms):
+        """ Creates a reservation of the given type. """
+        allowed_values = [
+            RESERVATION_TYPE_MIGRATION, RESERVATION_TYPE_REPLICA]
+        if reservation_type not in allowed_values:
+            raise ValueError(
+                "Reservation type must be one of %s" % allowed_values)
+        return self._post(
+            "/reservations", {
+                "type": reservation_type, "count": num_vms},
+            response_key="reservation")
+
+    def add_migrations_reservation(self, num_vms):
+        """ Creates a reservation for the given number of VM Migrations. """
+        return self.add_reservation(RESERVATION_TYPE_MIGRATION, num_vms)
+
+    def add_replicas_reservation(self, num_vms):
+        """ Creates a reservation for the given number of VM Replicas. """
+        return self.add_reservation(RESERVATION_TYPE_REPLICA, num_vms)
+
+    def get_reservations(self):
+        """ Lists all existing reservations. """
+        return self._get("/reservations", response_key="reservations")
+
+    def get_reservation(self, reservation_id):
+        """ Gets a reservation with the given ID.  """
+        return self._get(
+            "/reservations/%s" % reservation_id, response_key="reservation")
+
+    def check_reservation(self, reservation_id):
+        """ Checks the reservation with the given ID.  """
+        return self._post(
+            "/reservations/%s/check" % reservation_id, None,
+            response_key="reservation")
+
+    def delete_reservation(self, reservation_id, raise_on_404=False):
+        """ Deletes a reservation by its ID.
+        Unless `raise_on_404` is set, ignores not found reservations.
+        """
+        resp = self._do_req(
+            "delete", "/reservations/%s" % reservation_id, raw_response=True)
+        if not resp.ok:
+            if resp.status_code == 404:
+                if raise_on_404:
+                    resp.raise_for_status()
+                LOG.warn(
+                    "Got 404 when deleting reservation '%s'", reservation_id)
+            else:
+                resp.raise_for_status()