From 665ae505990f944d4e8c35ecee251ccdec3a22ad Mon Sep 17 00:00:00 2001 From: Flaper Fesp Date: Tue, 5 Mar 2013 18:52:43 +0100 Subject: [PATCH] Implements base classes for storage controllers This patch follows (or should follow) the specs written in the v1 specification: https://wiki.openstack.org/wiki/Marconi/specs/api/v1 This patch deletes driver_base.py and adds base.py in marconi/storage package. Note: Most of this implementations will be tested in further patches since there's no good way to test these base classes. Implements blueprint storage-base Change-Id: Iefa6374635fcf9dd9fc60417d8ed5c306bbc619a --- marconi/storage/__init__.py | 6 +- marconi/storage/base.py | 307 ++++++++++++++++++ .../storage/{driver_base.py => exceptions.py} | 20 +- marconi/storage/reference/driver.py | 2 +- marconi/tests/storage/__init__.py | 0 marconi/tests/storage/base.py | 116 +++++++ 6 files changed, 432 insertions(+), 19 deletions(-) create mode 100644 marconi/storage/base.py rename marconi/storage/{driver_base.py => exceptions.py} (63%) create mode 100644 marconi/tests/storage/__init__.py create mode 100644 marconi/tests/storage/base.py diff --git a/marconi/storage/__init__.py b/marconi/storage/__init__.py index 3209b011b..47ebf795b 100644 --- a/marconi/storage/__init__.py +++ b/marconi/storage/__init__.py @@ -1,3 +1,7 @@ """Marconi Storage Drivers""" -from marconi.storage.driver_base import DriverBase # NOQA +from marconi.storage.base import ClaimBase # NOQA +from marconi.storage.base import DriverBase # NOQA +from marconi.storage.base import MessageBase # NOQA +from marconi.storage.base import QueueBase # NOQA +from marconi.storage import exceptions # NOQA diff --git a/marconi/storage/base.py b/marconi/storage/base.py new file mode 100644 index 000000000..ad40b68a0 --- /dev/null +++ b/marconi/storage/base.py @@ -0,0 +1,307 @@ +# Copyright (c) 2013 Red Hat, Inc. +# +# 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. + +"""Implements the DriverBase abstract class for Marconi storage drivers.""" + +import abc + +# Seconds +MIN_TTL = 60 + +# Seconds (14 days) +MAX_TTL = 1209600 + + +class DriverBase: + __metaclass__ = abc.ABCMeta + + @abc.abstractproperty + def queue_controller(self): + """ + Returns storage's queues controller + """ + pass + + @abc.abstractproperty + def message_controller(self): + """ + Returns storage's messages controller + """ + pass + + @abc.abstractproperty + def claim_controller(self): + """ + Returns storage's claims controller + """ + pass + + +class ControllerBase(object): + """ + Top level class for controllers. + + :param driver: Instance of the driver + instantiating this controller. + """ + + def __init__(self, driver): + self.driver = driver + + +class QueueBase(ControllerBase): + """ + This class is responsible of managing + queues which means handling their CRUD + operations, monitoring and interactions. + + Storages' implementations of this class + should be capable of handling high work + loads and huge number of queues. + """ + + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def list(self, tenant=None): + """ + Base method for listing queues. + + :param tenant: Tenant id + + :returns: List of queues + """ + pass + + @abc.abstractmethod + def get(self, name, tenant=None): + """ + Base method for queue retrieval. + + :param name: The queue name + :param tenant: Tenant id + + :returns: Dictionary containing queue metadata + :raises: DoesNotExist + """ + pass + + @abc.abstractmethod + def create(self, name, tenant=None, ttl=MIN_TTL, **metadata): + """ + Base method for queue creation. + + :param name: The queue name + :param tenant: Tenant id + :param ttl: Number of seconds the server + will keep messages for this queue. + :param metadata: Arbitrary metadata + """ + pass + + @abc.abstractmethod + def update(self, name, tenant=None, **metadata): + """ + Base method for queue update. + + :param name: The queue name + :param tenant: Tenant id + :param metadata: Extra parameters defining + the new metadata for this queue. + """ + pass + + @abc.abstractmethod + def delete(self, name, tenant=None): + """ + Base method for queue deletion. + + :param name: The queue name + :param tenant: Tenant id + """ + pass + + @abc.abstractmethod + def stats(self, name, tenant=None): + """ + Base method for queue stats. + + :param name: The queue name + :param tenant: Tenant id + :returns: Dictionary with the + queue stats + """ + pass + + @abc.abstractmethod + def actions(self, name, tenant=None, marker=None, limit=10): + """ + Base method for queue actions. + + :param name: Queue name + :param tenant: Tenant id + :param marker: Tail identifier + :param limit: (Default 10) Max number + of messages to retrieve. + """ + pass + + +class MessageBase(ControllerBase): + """ + This class is responsible for managing + messages CRUD. + """ + + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def get(self, queue, tenant=None, message_id=None, + marker=None, echo=False, client_uuid=None): + """ + Base message get method + + This method is responsible for querying messages + and should be capable of retrieving a single + message based on message_id or multiple messages + based on the other parameters being passed. + + :param queue: Name of the queue to get the + message from. + :param tenant: Tenant id + :param message: Message ID + :param marker: Tail identifier + :param echo: (Default False) Boolean expressing whether + or not this client should receive its own messages. + :param client_uuid: Client's unique identifier. This param + is required when echo=False. + + :returns: List of messages + :raises: DoesNotExist + """ + pass + + @abc.abstractmethod + def post(self, queue, messages, tenant=None): + """ + Base message post method + + Implementations of this method should guarantee + and preserve the order, in the returned list, of + incoming messages. + + :param queue: Name of the queue to post message to. + :param messages: Messages to post to queue, + it can be a list of 1 or more elements. + :param tenant: Tenant id + + :returns: List of message ids + """ + pass + + @abc.abstractmethod + def delete(self, queue, message_id, tenant=None, claim=None): + """ + Base message delete method + + :param queue: Name of the queue to post + message to. + :param message_id: Message to be deleted + :param tenant: Tenant id + :param claim: Claim this message + belongs to. When specified, claim must + be valid and message_id must belong to + it. + """ + pass + + +class ClaimBase(ControllerBase): + + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def get(self, queue, claim_id, tenant=None): + """ + Base claim get method + + :param queue: Name of the queue this + claim belongs to. + :param claim_id: The claim id + :param tenant: Tenant id + + :returns: Dictionary containing claim's + metadata and claimed messages. + :raises: DoesNotExist + """ + pass + + @abc.abstractmethod + def create(self, queue, tenant=None, ttl=MIN_TTL, limit=10): + """ + Base claim create method + + :param queue: Name of the queue this + claim belongs to. + :param tenant: Tenant id + :param ttl: Number of seconds the server + will keep this claim in this queue. + :param limit: (Default 10) Max number + of messages to claim. + + :returns: Dictionary containing claim's + metadata and claimed messages. + """ + pass + + @abc.abstractmethod + def update(self, queue, claim_id, tenant=None, **metadata): + """ + Base claim update method + + :param queue: Name of the queue this + claim belongs to. + :param claim_id: Claim to be updated + :param tenant: Tenant id + :param metadata: Claim's parameters + to be updated. + """ + pass + + @abc.abstractmethod + def delete(self, queue, claim_id, tenant=None): + """ + Base claim delete method + + :param queue: Name of the queue this + claim belongs to. + :param claim_id: Claim to be deleted + :param tenant: Tenant id + """ + pass + + @abc.abstractmethod + def stats(self, queue, claim_id, tenant=None): + """ + Base method for claim stats. + + :param queue: Name of the queue this + claim belongs to. + :param claim_id: Claim to be deleted + :param tenant: Tenant id + :returns: Dictionary with the + queue stats + """ + pass diff --git a/marconi/storage/driver_base.py b/marconi/storage/exceptions.py similarity index 63% rename from marconi/storage/driver_base.py rename to marconi/storage/exceptions.py index b7265788d..19316c2e4 100644 --- a/marconi/storage/driver_base.py +++ b/marconi/storage/exceptions.py @@ -1,4 +1,4 @@ -# Copyright (c) 2013 Rackspace, Inc. +# Copyright (c) 2013 Red Hat, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -13,20 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import abc - -class DriverBase: - __metaclass__ = abc.ABCMeta - - @abc.abstractproperty - def queue_controller(self): - pass - - @abc.abstractproperty - def message_controller(self): - pass - - @abc.abstractproperty - def claim_controller(self): - pass +class DoesNotExist(Exception): + pass diff --git a/marconi/storage/reference/driver.py b/marconi/storage/reference/driver.py index 5ac0fc0b4..10710e946 100644 --- a/marconi/storage/reference/driver.py +++ b/marconi/storage/reference/driver.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -import marconi.storage as storage +from marconi import storage class Driver(storage.DriverBase): diff --git a/marconi/tests/storage/__init__.py b/marconi/tests/storage/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/marconi/tests/storage/base.py b/marconi/tests/storage/base.py new file mode 100644 index 000000000..f604596e8 --- /dev/null +++ b/marconi/tests/storage/base.py @@ -0,0 +1,116 @@ +# Copyright (c) 2013 Red Hat, Inc. +# +# 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 marconi import storage +from marconi.tests.util import suite + + +class ControllerBaseTest(suite.TestSuite): + driver_class = None + controller_class = None + controller_base_class = None + + def setUp(self): + super(ControllerBaseTest, self).setUp() + + if not self.driver_class: + self.skipTest("No driver class specified") + + if not isinstance(self.controller_class, self.controller_base_class): + self.skipTest("%s is not an instance of %s. Tests not supported" % + (self.controller_class, self.controller_base_class)) + + self.driver = self.driver_class() + self.controller = self.controller_class(self.driver) + + +class QueueControllerTest(ControllerBaseTest): + """ + Queue Controller base tests + """ + controller_base_class = storage.QueueBase + + def test_queue_lifecycle(self): + # Test Queue Creation + self.controller.create("test", ttl=10, + topic="test_queue") + + # Test Queue retrieval + queue = self.controller.get("test") + self.assertEqual(queue["name"], "test") + self.assertEqual(queue["ttl"], 10) + + # Test Queue Update + self.controller.update("tests", name="test1") + queue = self.controller.get("test1") + self.assertEqual(queue["ttl"], 10) + + # Test Queue Deletion + self.controller.delete("test1") + + # Test DoesNotExist Exception + self.assertRaises(storage.exceptions.DoesNotExist, + self.controller.get, "test1") + + +class MessageControllerTest(ControllerBaseTest): + """ + Message Controller base tests + + NOTE(flaper87): Implementations of this class should + override the tearDown method in order + to clean up storage's state. + """ + queue_name = "test_queue" + controller_base_class = storage.MessageBase + + def setUp(self): + super(MessageControllerTest, self).setUp() + + # Lets create a queue + self.queue_controller = self.driver.queue_controller() + self.queue_controller.create(self.queue_name) + + def tearDown(self): + self.queue_controller.delete(self.queue_name) + super(MessageControllerTest, self).tearDown() + + def test_message_lifecycle(self): + queue_name = self.queue_name + + messages = [ + { + "body": { + "event": "BackupStarted", + "backupId": "c378813c-3f0b-11e2-ad92-7823d2b0f3ce" + } + }, + ] + + # Test Message Creation + created = self.controller.post(queue_name, messages) + self.assertEqual(len(created), 1) + + # Test Message Get + self.controller.get(queue_name, message_id=created[0]) + + # Test Message Deletion + self.controller.delete(queue_name, created[0]) + + # Test DoesNotExist + self.assertRaises(storage.exceptions.DoesNotExist, + self.controller.get, + queue_name, created[0])