Merge pull request #1 from openstack/master

sync openstack to rackerlabs fork

etc/marconi-queues.conf-sample

@@ -31,7 +31,7 @@ log_file = /var/log/marconi/queues.log
 
 # ================= Driver Options ============================
 
-[queues:drivers]
+[drivers]
 # Transport driver module (e.g., wsgi, zmq)
 transport = wsgi
 
@@ -47,7 +47,7 @@ storage = mongodb
 # Pipeline for operations on claim resources
 ;claim_pipeline =
 
-[queues:drivers:transport:wsgi]
+[drivers:transport:wsgi]
 ;bind = 0.0.0.0
 ;port = 8888
 
@@ -56,10 +56,10 @@ storage = mongodb
 ;metadata_max_length = 65536
 ;content_max_length = 262144
 
-;[queues:drivers:transport:zmq]
+;[drivers:transport:zmq]
 ;port = 9999
 
-[queues:drivers:storage:mongodb]
+[drivers:storage:mongodb]
 uri = mongodb://db1.example.net,db2.example.net:2500/?replicaSet=test&ssl=true&w=majority
 database = marconi
 
@@ -83,7 +83,7 @@ database = marconi
 # at the same instant.
 ;max_retry_jitter = 0.005
 
-[queues:limits:transport]
+[limits:transport]
 # The maximum number of queue records per page when listing queues
 ;queue_paging_uplimit = 20
 
@@ -102,7 +102,7 @@ database = marconi
 ;metadata_size_uplimit = 65536
 ;message_size_uplimit = 262144
 
-[queues:limits:storage]
+[limits:storage]
 # The default number of queue records per page when listing queues
 ;default_queue_paging = 10
 

marconi/common/api.py

@@ -0,0 +1,79 @@
+# 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.
+
+import jsonschema
+from jsonschema import validators
+
+from marconi.common import errors
+from marconi.openstack.common import log
+
+LOG = log.getLogger(__name__)
+
+
+class Api(object):
+
+    schema = {}
+    validators = {}
+
+    def get_schema(self, operation):
+        """Returns the schema for an operation
+
+        :param operation: Operation for which params need
+        to be validated.
+        :type operation: `six.text_type`
+
+        :returns: Operation's schema
+        :rtype: dict
+
+        :raises: `errors.InvalidOperation` if the operation
+        does not exist
+        """
+        try:
+            return self.schema[operation]
+        except KeyError:
+            # TODO(flaper87): gettext support
+            msg = _('{0} is not a valid operation').format(operation)
+            raise errors.InvalidOperation(msg)
+
+    def validate(self, operation, params):
+        """Validates the request data
+
+        This method relies on jsonschema and exists
+        just as a way for third-party transport to validate
+        the request. It's not recommended to validate every
+        request since they are already validated server side.
+
+        :param operation: Operation's for which params need
+        to be validated.
+        :type operation: `six.text_type`
+        :param params: Params to validate
+        :type params: dict
+
+        :returns: True if the schema is valid, False otherwise
+        :raises: `errors.InvalidOperation` if the operation
+        does not exist
+        """
+
+        if operation not in self.validators:
+            schema = self.get_schema(operation)
+            self.validators[operation] = validators.Draft4Validator(schema)
+
+        try:
+            self.validators[operation].validate(params)
+        except jsonschema.ValidationError:
+            LOG.debug('Operation is invalid.')
+            return False
+
+        return True

marconi/common/cache/_backends/memcached.py

@@ -57,7 +57,7 @@ def _get_ttl(self, ttl):
 
     def set(self, key, value, ttl=0):
         key = self._prepare_key(key)
-        self._cache.set(key, value, self._get_ttl(ttl))
+        return self._cache.set(key, value, self._get_ttl(ttl))
 
     def unset(self, key):
         self._cache.delete(self._prepare_key(key))

marconi/common/exceptions.py → marconi/common/errors.py

@@ -20,3 +20,7 @@ class InvalidDriver(Exception):
 
 class PatternNotFound(Exception):
     """A string did not match the expected pattern or regex."""
+
+
+class InvalidOperation(Exception):
+    """Raised when attempted a non existent operation."""

marconi/common/schemas/shards.py

@@ -29,9 +29,9 @@
 # NOTE(cpp-cabrera): a string valid for use in a URI
 # TODO(cpp-cabrera): perhaps validate this further using jsonschema's
 # uri validator as per rfc3987
-patch_location = {
+patch_uri = {
     'type': 'object', 'properties': {
-        'location': {
+        'uri': {
             'type': 'string'
         },
         'additionalProperties': False
@@ -50,11 +50,11 @@
 create = {
     'type': 'object', 'properties': {
         'weight': patch_weight['properties']['weight'],
-        'location': patch_location['properties']['location'],
+        'uri': patch_uri['properties']['uri'],
         'options': patch_options['properties']['options']
     },
     # NOTE(cpp-cabrera): options need not be present. Storage drivers
     # must provide reasonable defaults.
-    'required': ['location', 'weight'],
+    'required': ['uri', 'weight'],
     'additionalProperties': False
 }

marconi/common/storage/select.py

@@ -0,0 +1,58 @@
+# Copyright (c) 2013 Rackspace Hosting, 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.
+
+"""select: a collection of algorithms for choosing an entry from a
+collection."""
+
+import random
+
+
+def weighted(objs, key='weight', generator=random.randint):
+    """Perform a weighted select given a list of objects.
+
+    :param objs: a list of objects containing at least the field `key`
+    :type objs: [dict]
+    :param key: the field in each obj that corresponds to weight
+    :type key: six.text_type
+    :param generator: a number generator taking two ints
+    :type generator: function(int, int) -> int
+    :return: an object
+    :rtype: dict
+    """
+    acc = 0
+    lookup = []
+
+    # construct weighted spectrum
+    for o in objs:
+        # NOTE(cpp-cabrera): skip objs with 0 weight
+        if o[key] <= 0:
+            continue
+        acc += o[key]
+        lookup.append((o, acc))
+
+    # no objects were found
+    if not lookup:
+        return None
+
+    # NOTE(cpp-cabrera): select an object from the lookup table. If
+    # the selector lands in the interval [lower, upper), then choose
+    # it.
+    gen = generator
+    selector = gen(0, acc - 1)
+    lower = 0
+    for obj, upper in lookup:
+        if lower <= selector < upper:
+            return obj
+        lower = upper

marconi/common/transport/version.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""version: version information for the proxy transport API."""
+"""version: version information for the transport API."""
 
 
 def info():

marconi/common/transport/wsgi/health.py

@@ -19,8 +19,14 @@
 
 class Resource(object):
 
+    __slots__ = ('driver',)
+
+    def __init__(self, driver):
+        self.driver = driver
+
     def on_get(self, req, resp, **kwargs):
-        resp.status = falcon.HTTP_204
+        resp.status = (falcon.HTTP_204 if self.driver.is_alive()
+                       else falcon.HTTP_503)
 
     def on_head(self, req, resp, **kwargs):
         resp.status = falcon.HTTP_204

marconi/common/transport/wsgi/helpers.py

@@ -16,7 +16,6 @@
 """wsgi transport helpers."""
 
 import falcon
-import six
 
 import marconi.openstack.common.log as logging
 from marconi.queues.transport import validation
@@ -46,11 +45,11 @@ def extract_project_id(req, resp, params):
 and retry.'''))
 
 
-def validate_queue_name(validate, req, resp, params):
-    """Hook for validating the queue name specified in a request.
+def validate_queue_identification(validate, req, resp, params):
+    """Hook for validating the queue name and project id in requests.
 
-    Validation is short-circuited if 'queue_name' does not
-    exist in `params`.
+    The queue name validation is short-circuited if 'queue_name' does
+    not exist in `params`.
 
     This hook depends on the `get_project` hook, which must be
     installed upstream.
@@ -67,22 +66,22 @@ def validate_queue_name(validate, req, resp, params):
     """
 
     try:
-        validate(params['queue_name'])
+        validate(params['queue_name'],
+                 params['project_id'])
     except KeyError:
         # NOTE(kgriffs): queue_name not in params, so nothing to do
         pass
-    except validation.ValidationFailed as ex:
+    except validation.ValidationFailed:
         project = params['project_id']
         queue = params['queue_name'].decode('utf-8', 'replace')
 
         LOG.warn(_(u'Invalid queue name "%(queue)s" submitted for '
                    u'project: %(project)s'),
                  {'queue': queue, 'project': project})
 
-        info = six.text_type(ex) or _(u'The format of the submitted '
-                                      u'queue name is not valid.')
-
-        raise falcon.HTTPBadRequest(_(u'Invalid queue name'), info)
+        raise falcon.HTTPBadRequest(_(u'Invalid queue identification'),
+                                    _(u'The format of the submitted queue '
+                                      u'name or project id is not valid.'))
 
 
 def require_accepts_json(req, resp, params):

marconi/common/transport/wsgi/utils.py

@@ -19,7 +19,7 @@
 
 from marconi.openstack.common import log
 from marconi.queues.transport import utils as json_utils
-from marconi.queues.transport.wsgi import exceptions as wsgi_errors
+from marconi.queues.transport.wsgi import errors as wsgi_errors
 
 LOG = log.getLogger(__name__)