qinling/qinling/rpc.py

# Copyright 2017 Catalyst IT Limited
#
#    Licensed 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.

from oslo_config import cfg
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_messaging.rpc import client

from qinling import context as ctx
from qinling import exceptions as exc

LOG = logging.getLogger(__name__)

_TRANSPORT = None
_ENGINE_CLIENT = None


def cleanup():
    """Intended to be used by tests to recreate all RPC related objects."""

    global _TRANSPORT
    global _ENGINE_CLIENT

    _TRANSPORT = None
    _ENGINE_CLIENT = None


def get_transport():
    global _TRANSPORT

    if not _TRANSPORT:
        _TRANSPORT = messaging.get_transport(cfg.CONF)

    return _TRANSPORT


def get_engine_client():
    global _ENGINE_CLIENT

    if not _ENGINE_CLIENT:
        _ENGINE_CLIENT = EngineClient(get_transport())

    return _ENGINE_CLIENT


def _wrap_exception_and_reraise(exception):
    message = "%s: %s" % (exception.__class__.__name__, exception.args[0])

    raise exc.QinlingException(message)


def wrap_messaging_exception(method):
    """This decorator unwrap remote error in one of QinlingException.

    oslo.messaging has different behavior on raising exceptions
    when fake or rabbit transports are used. In case of rabbit
    transport it raises wrapped RemoteError which forwards directly
    to API. Wrapped RemoteError contains one of QinlingException raised
    remotely on Engine and for correct exception interpretation we
    need to unwrap and raise given exception and manually send it to
    API layer.
    """
    def decorator(*args, **kwargs):
        try:
            return method(*args, **kwargs)
        except exc.QinlingException:
            raise
        except (client.RemoteError, Exception) as e:
            if hasattr(e, 'exc_type') and hasattr(exc, e.exc_type):
                exc_cls = getattr(exc, e.exc_type)
                raise exc_cls(e.value)

            _wrap_exception_and_reraise(e)

    return decorator


class ContextSerializer(messaging.Serializer):
    def __init__(self, base):
        self._base = base

    def serialize_entity(self, context, entity):
        if not self._base:
            return entity
        return self._base.serialize_entity(context, entity)

    def deserialize_entity(self, context, entity):
        if not self._base:
            return entity
        return self._base.deserialize_entity(context, entity)

    def serialize_context(self, context):
        return context.convert_to_dict()

    def deserialize_context(self, context):
        qinling_ctx = ctx.Context.from_dict(context)
        ctx.set_ctx(qinling_ctx)

        return qinling_ctx


class EngineClient(object):
    """RPC Engine client."""

    def __init__(self, transport):
        """Constructs an RPC client for engine.

        :param transport: Messaging transport.
        """
        serializer = ContextSerializer(
            messaging.serializer.JsonPayloadSerializer())

        self.topic = cfg.CONF.engine.topic

        self._client = messaging.RPCClient(
            transport,
            messaging.Target(topic=self.topic),
            serializer=serializer
        )

    @wrap_messaging_exception
    def create_runtime(self, id):
        return self._client.prepare(topic=self.topic, server=None).cast(
            ctx.get_ctx(),
            'create_runtime',
            runtime_id=id
        )

    @wrap_messaging_exception
    def delete_runtime(self, id):
        return self._client.prepare(topic=self.topic, server=None).cast(
            ctx.get_ctx(),
            'delete_runtime',
            runtime_id=id
        )

    @wrap_messaging_exception
    def update_runtime(self, id, image=None, pre_image=None):
        return self._client.prepare(topic=self.topic, server=None).cast(
            ctx.get_ctx(),
            'update_runtime',
            runtime_id=id,
            image=image,
            pre_image=pre_image
        )

    @wrap_messaging_exception
    def create_execution(self, execution_id, function_id, runtime_id,
                         input=None, is_sync=True):
        method_client = self._client.prepare(topic=self.topic, server=None)

        if is_sync:
            return method_client.call(
                ctx.get_ctx(),
                'create_execution',
                execution_id=execution_id,
                function_id=function_id,
                runtime_id=runtime_id,
                input=input
            )
        else:
            method_client.cast(
                ctx.get_ctx(),
                'create_execution',
                execution_id=execution_id,
                function_id=function_id,
                runtime_id=runtime_id,
                input=input
            )

    @wrap_messaging_exception
    def delete_function(self, id):
        return self._client.prepare(topic=self.topic, server=None).cast(
            ctx.get_ctx(),
            'delete_function',
            function_id=id
        )

    @wrap_messaging_exception
    def scaleup_function(self, id, runtime_id, count=1):
        return self._client.prepare(topic=self.topic, server=None).cast(
            ctx.get_ctx(),
            'scaleup_function',
            function_id=id,
            runtime_id=runtime_id,
            count=count
        )

    @wrap_messaging_exception
    def scaledown_function(self, id, count=1):
        return self._client.prepare(topic=self.topic, server=None).cast(
            ctx.get_ctx(),
            'scaledown_function',
            function_id=id,
            count=count
        )