Merge "feat(health): make health API storage-aware"

marconi/common/transport/wsgi/health.py

@@ -19,8 +19,14 @@ import falcon
 
 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/queues/storage/base.py

@@ -55,6 +55,11 @@ class DataDriverBase(object):
         self.conf.register_opts(_LIMITS_OPTIONS, group=_LIMITS_GROUP)
         self.limits_conf = self.conf[_LIMITS_GROUP]
 
+    @abc.abstractmethod
+    def is_alive(self):
+        """Check whether the storage is ready."""
+        raise NotImplementedError
+
     @abc.abstractproperty
     def queue_controller(self):
         """Returns the driver's queue controller."""

marconi/queues/storage/mongodb/driver.py

@@ -56,6 +56,14 @@ class DataDriver(storage.DataDriverBase):
                                 group=options.MONGODB_GROUP)
         self.mongodb_conf = self.conf[options.MONGODB_GROUP]
 
+    def is_alive(self):
+        try:
+            # NOTE(zyuan): Requires admin access to mongodb
+            return 'ok' in self.connection.admin.command('ping')
+
+        except pymongo.errors.PyMongoError:
+            return False
+
     @decorators.lazy_property(write=False)
     def queues_database(self):
         """Database dedicated to the "queues" collection.

marconi/queues/storage/pipeline.py

@@ -93,6 +93,9 @@ class DataDriver(base.DataDriverBase):
         super(DataDriver, self).__init__(conf)
         self._storage = storage
 
+    def is_alive(self):
+        return self._storage.is_alive()
+
     @decorators.lazy_property(write=False)
     def queue_controller(self):
         stages = _get_storage_pipeline('queue', self.conf)

marconi/queues/storage/sharding.py

@@ -46,6 +46,11 @@ class DataDriver(storage.DataDriverBase):
         super(DataDriver, self).__init__(conf)
         self._shard_catalog = Catalog(conf, control)
 
+    def is_alive(self):
+        return all(self._shard_catalog.get_driver(shard['name']).is_alive()
+                   for shard in
+                   self._shard_catalog._shards_ctrl.list(limit=0))
+
     @decorators.lazy_property(write=False)
     def queue_controller(self):
         return QueueController(self._shard_catalog)

marconi/queues/storage/sqlite/driver.py

@@ -185,6 +185,9 @@ class DataDriver(storage.DataDriverBase):
             self.__conn.rollback()
             raise
 
+    def is_alive(self):
+        return True
+
     @decorators.lazy_property(write=False)
     def queue_controller(self):
         return controllers.QueueController(self)

marconi/queues/transport/wsgi/admin/driver.py

@@ -30,5 +30,5 @@ class Driver(public_driver.Driver):
             ('/shards/{shard}',
              shards.Resource(shards_controller)),
             ('/health',
-             health.Resource())
+             health.Resource(self._storage))
         ]

marconi/queues/transport/wsgi/public/driver.py

@@ -68,5 +68,5 @@ class Driver(driver.DriverBase):
 
             # Health
             ('/health',
-             health.Resource())
+             health.Resource(self._storage))
         ]

marconi/tests/faulty_storage.py

@@ -24,6 +24,9 @@ class DataDriver(storage.DataDriverBase):
     def default_options(self):
         return {}
 
+    def is_alive(self):
+        raise NotImplementedError()
+
     @property
     def queue_controller(self):
         return QueueController(self)

tests/unit/queues/storage/test_shard_queues.py

@@ -37,7 +37,8 @@ class TestShardQueues(base.TestBase):
 
         control = utils.load_storage_driver(conf, control_mode=True)
         self.shards_ctrl = control.shards_controller
-        self.controller = sharding.DataDriver(conf, control).queue_controller
+        self.driver = sharding.DataDriver(conf, control)
+        self.controller = self.driver.queue_controller
 
         # fake two shards
         for _ in xrange(2):
@@ -47,6 +48,10 @@ class TestShardQueues(base.TestBase):
         self.shards_ctrl.drop_all()
         super(TestShardQueues, self).tearDown()
 
+    def test_health(self):
+        health = self.driver.is_alive()
+        self.assertTrue(health)
+
     def test_listing(self):
         project = "I.G"