Adding healthcheck

Healthcheck allows to verify if:
 - API is up and running
 - Kafka, that monasca-log-api sends data to, is up and running
 and an expected topic can be found there.

Other:
- added documentation entries

Change-Id: I316c1d9518cfed37119f11c326c071bfbfc7658e

documentation/monasca_log_api.healthcheck.rst

@@ -0,0 +1,21 @@
+monasca_log_api.healthcheck package
+===================================
+
+Submodules
+----------
+
+monasca_log_api.healthcheck.kafka_check module
+----------------------------------------------
+
+.. automodule:: monasca_log_api.healthcheck.kafka_check
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+monasca_log_api.healthcheck.keystone_protocol module
+----------------------------------------------------
+
+.. automodule:: monasca_log_api.healthcheck.keystone_protocol
+    :members:
+    :undoc-members:
+    :show-inheritance:

documentation/monasca_log_api.rst

@@ -9,6 +9,7 @@ Subpackages
     monasca_log_api.api
     monasca_log_api.v2
     monasca_log_api.middleware
+    monasca_log_api.healthcheck
 
 Submodules
 ----------

etc/monasca/log-api-config.conf

@@ -9,6 +9,7 @@ debug=True
 [dispatcher]
 logs = monasca_log_api.v2.reference.logs:Logs
 versions = monasca_log_api.v2.reference.versions:Versions
+healthchecks = monasca_log_api.v2.reference.healthchecks:HealthChecks
 
 [service]
 max_log_size = 1048576
@@ -29,7 +30,11 @@ certfile =
 keyfile =
 insecure = false
 
+[kafka_healthcheck]
+kafka_url = localhost:8900
+kafka_topics = log
+
 [roles_middleware]
 path = /v2.0/log
 default_roles = monasca-user
-agent_roles = monasca-log-agent
+agent_roles = monasca-log-agent

etc/monasca/log-api-config.ini

@@ -8,7 +8,7 @@ pipeline = auth roles api
 paste.app_factory = monasca_log_api.server:launch
 
 [filter:auth]
-paste.filter_factory = keystonemiddleware.auth_token:filter_factory
+paste.filter_factory = monasca_log_api.healthcheck.keystone_protocol:filter_factory
 
 [filter:roles]
 paste.filter_factory = monasca_log_api.middleware.role_middleware:RoleMiddleware.factory

monasca_log_api/api/healthcheck_api.py

@@ -0,0 +1,60 @@
+# Copyright 2016 FUJITSU 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.
+
+import collections
+
+import falcon
+from oslo_log import log
+
+LOG = log.getLogger(__name__)
+
+HealthCheckResult = collections.namedtuple('HealthCheckResult',
+                                           ['status', 'details'])
+
+
+# TODO(feature) monasca-common candidate
+class HealthChecksApi(object):
+    """HealthChecks Api
+
+    HealthChecksApi server information regarding health of the API.
+
+    """
+
+    def __init__(self):
+        super(HealthChecksApi, self).__init__()
+        LOG.info('Initializing HealthChecksApi!')
+
+    def on_get(self, req, res):
+        """Complex healthcheck report on GET.
+
+        Returns complex report regarding API well being
+        and all dependent services.
+
+        :param falcon.Request req: current request
+        :param falcon.Response res: current response
+        """
+        res.status = falcon.HTTP_501
+
+    def on_head(self, req, res):
+        """Simple healthcheck report on HEAD.
+
+        In opposite to :py:meth:`.HealthChecksApi.on_get`, this
+        method is supposed to execute ASAP to inform user that
+        API is up and running.
+
+        :param falcon.Request req: current request
+        :param falcon.Response res: current response
+
+        """
+        res.status = falcon.HTTP_501

monasca_log_api/healthcheck/__init__.py

@@ -0,0 +1 @@
+"""Base package for monasca-log-api healthcheck"""

monasca_log_api/healthcheck/kafka_check.py

@@ -0,0 +1,109 @@
+# Copyright 2015 FUJITSU 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.
+
+import collections
+
+import kafka.client as client
+from oslo_config import cfg
+from oslo_log import log
+
+LOG = log.getLogger(__name__)
+CONF = cfg.CONF
+
+kafka_check_opts = [
+    cfg.StrOpt('kafka_url',
+               required=True,
+               help='Url to kafka server'),
+    cfg.ListOpt('kafka_topics',
+                required=True,
+                default=['logs'],
+                help='Verify existence of configured topics')
+]
+kafka_check_group = cfg.OptGroup(name='kafka_healthcheck',
+                                 title='kafka_healthcheck')
+
+cfg.CONF.register_group(kafka_check_group)
+cfg.CONF.register_opts(kafka_check_opts, kafka_check_group)
+
+
+CheckResult = collections.namedtuple('CheckResult', ['healthy', 'message'])
+"""Result from the healthcheck, contains healthy(boolean) and message"""
+
+
+# TODO(feature) monasca-common candidate
+class KafkaHealthCheck(object):
+    """Evaluates kafka health
+
+    Healthcheck verifies if:
+
+    * kafka server is up and running
+    * there is a configured topic in kafka
+
+    If following conditions are met healthcheck returns healthy status.
+    Otherwise unhealthy status is returned with explanation.
+
+     Example of middleware configuration:
+
+    .. code-block:: ini
+
+      [kafka_healthcheck]
+      kafka_url = localhost:8900
+      kafka_topics = log
+
+    Note:
+        It is possible to specify multiple topics if necessary.
+        Just separate them with ,
+
+    """
+
+    def healthcheck(self):
+        url = CONF.kafka_healthcheck.kafka_url
+
+        try:
+            kafka_client = client.KafkaClient(hosts=url)
+        except client.KafkaUnavailableError as ex:
+            LOG.error(repr(ex))
+            error_str = 'Could not connect to kafka at %s' % url
+            return CheckResult(healthy=False, message=error_str)
+
+        result = self._verify_topics(kafka_client)
+        self._disconnect_gracefully(kafka_client)
+
+        return result
+
+    # noinspection PyMethodMayBeStatic
+    def _verify_topics(self, kafka_client):
+        topics = CONF.kafka_healthcheck.kafka_topics
+
+        for t in topics:
+            # kafka client loads metadata for topics as fast
+            # as possible (happens in __init__), therefore this
+            # topic_partitions is sure to be filled
+            for_topic = t in kafka_client.topic_partitions
+            if not for_topic:
+                error_str = 'Kafka: Topic %s not found' % t
+                LOG.error(error_str)
+                return CheckResult(healthy=False, message=error_str)
+
+        return CheckResult(healthy=True, message='OK')
+
+    # noinspection PyMethodMayBeStatic
+    def _disconnect_gracefully(self, kafka_client):
+        # at this point, client is connected so it must be closed
+        # regardless of topic existence
+        try:
+            kafka_client.close()
+        except Exception as ex:
+            # log that something went wrong and move on
+            LOG.error(repr(ex))

monasca_log_api/healthcheck/keystone_protocol.py

@@ -0,0 +1,60 @@
+# Copyright 2015 FUJITSU 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 keystonemiddleware import auth_token
+from oslo_log import log
+
+LOG = log.getLogger(__name__)
+
+
+class SkippingAuthProtocol(auth_token.AuthProtocol):
+    """SkippingAuthProtocol to reach healthcheck endpoint
+
+    Because healthcheck endpoints exists as endpoint, it
+    is hidden behind keystone filter thus a request
+    needs to authenticated before it is reached.
+
+    Note:
+        SkippingAuthProtocol is lean customization
+        of :py:class:`keystonemiddleware.auth_token.AuthProtocol`
+        that disables keystone communication if request
+        is meant to reach healthcheck
+
+    """
+
+    def process_request(self, request):
+        path = request.path
+        if path == '/healthcheck':
+            LOG.debug(('Request path is %s and it does not require keystone '
+                       'communication'), path)
+            return None  # return NONE to reach actual logic
+
+        return super(SkippingAuthProtocol, self).process_request(request)
+
+
+def filter_factory(global_conf, **local_conf):  # pragma: no cover
+    """Return factory function for :py:class:`.SkippingAuthProtocol`
+
+    :param global_conf: global configuration
+    :param local_conf: local configuration
+    :return: factory function
+    :rtype: function
+    """
+    conf = global_conf.copy()
+    conf.update(local_conf)
+
+    def auth_filter(app):
+        return SkippingAuthProtocol(app, conf)
+
+    return auth_filter

monasca_log_api/server.py

@@ -28,10 +28,16 @@ CONF = cfg.CONF
 dispatcher_opts = [
     cfg.StrOpt('versions',
                default=None,
-               help='Versions'),
+               required=True,
+               help='Versions endpoint'),
     cfg.StrOpt('logs',
                default=None,
-               help='Logs')
+               required=True,
+               help='Logs endpoint'),
+    cfg.StrOpt('healthchecks',
+               default=None,
+               required=True,
+               help='Healthchecks endpoint')
 ]
 dispatcher_group = cfg.OptGroup(name='dispatcher', title='dispatcher')
 CONF.register_group(dispatcher_group)
@@ -53,12 +59,18 @@ def launch(conf, config_file='/etc/monasca/log-api-config.conf'):
 
     load_versions_resource(app)
     load_logs_resource(app)
+    load_healthcheck_resource(app)
 
     LOG.debug('Dispatcher drivers have been added to the routes!')
 
     return app
 
 
+def load_healthcheck_resource(app):
+    healthchecks = simport.load(CONF.dispatcher.healthchecks)()
+    app.add_route('/healthcheck', healthchecks)
+
+
 def load_logs_resource(app):
     logs = simport.load(CONF.dispatcher.logs)()
     app.add_route('/v2.0/log/single', logs)

monasca_log_api/tests/test_healthchecks.py

@@ -0,0 +1,76 @@
+# Copyright 2015 FUJITSU 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.
+
+import falcon
+from falcon import testing
+import mock
+import simplejson as json
+
+from monasca_log_api.healthcheck import kafka_check as healthcheck
+from monasca_log_api.tests import base
+from monasca_log_api.v2.reference import healthchecks
+
+ENDPOINT = '/healthcheck'
+
+
+class TestHealthChecks(testing.TestBase):
+    def before(self):
+        self.conf = base.mock_config(self)
+        self.resource = healthchecks.HealthChecks()
+        self.api.add_route(
+                ENDPOINT,
+                self.resource
+        )
+
+    def test_should_return_200_for_head(self):
+        self.simulate_request(ENDPOINT, method='HEAD')
+        self.assertEqual(falcon.HTTP_NO_CONTENT, self.srmock.status)
+
+    @mock.patch('monasca_log_api.healthcheck.kafka_check.KafkaHealthCheck')
+    def test_should_report_healthy_if_kafka_healthy(self, kafka_check):
+        kafka_check.healthcheck.return_value = healthcheck.CheckResult(True,
+                                                                       'OK')
+        self.resource._kafka_check = kafka_check
+
+        ret = self.simulate_request(ENDPOINT,
+                                    headers={
+                                        'Content-Type': 'application/json'
+                                    },
+                                    decode='utf8',
+                                    method='GET')
+        self.assertEqual(falcon.HTTP_OK, self.srmock.status)
+
+        ret = json.loads(ret)
+        self.assertIn('kafka', ret)
+        self.assertEqual('OK', ret.get('kafka'))
+
+    @mock.patch('monasca_log_api.healthcheck.kafka_check.KafkaHealthCheck')
+    def test_should_report_unhealthy_if_kafka_healthy(self, kafka_check):
+        url = 'localhost:8200'
+        err_str = 'Could not connect to kafka at %s' % url
+        kafka_check.healthcheck.return_value = healthcheck.CheckResult(False,
+                                                                       err_str)
+        self.resource._kafka_check = kafka_check
+
+        ret = self.simulate_request(ENDPOINT,
+                                    headers={
+                                        'Content-Type': 'application/json'
+                                    },
+                                    decode='utf8',
+                                    method='GET')
+        self.assertEqual(falcon.HTTP_SERVICE_UNAVAILABLE, self.srmock.status)
+
+        ret = json.loads(ret)
+        self.assertIn('kafka', ret)
+        self.assertEqual(err_str, ret.get('kafka'))

monasca_log_api/tests/test_kafka_check.py

@@ -0,0 +1,75 @@
+# Copyright 2015 FUJITSU 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 falcon import testing
+import kafka.client as client
+import mock
+
+from monasca_log_api.healthcheck import kafka_check as kc
+from monasca_log_api.tests import base
+
+
+class KafkaCheckLogicTest(testing.TestBase):
+    mock_kafka_url = 'localhost:1234'
+    mocked_topics = ['test_1', 'test_2']
+    mock_config = {
+        'kafka_url': mock_kafka_url,
+        'kafka_topics': mocked_topics
+    }
+
+    def __init__(self, *args, **kwargs):
+        super(KafkaCheckLogicTest, self).__init__(*args, **kwargs)
+        self._conf = None
+
+    def setUp(self):
+        super(KafkaCheckLogicTest, self).setUp()
+        self._conf = base.mock_config(self)
+        self._conf.config(group='kafka_healthcheck', **self.mock_config)
+
+    @mock.patch('monasca_log_api.healthcheck.kafka_check.client.KafkaClient')
+    def test_should_fail_kafka_unavailable(self, kafka_client):
+        kafka_client.side_effect = client.KafkaUnavailableError()
+        kafka_health = kc.KafkaHealthCheck()
+        result = kafka_health.healthcheck()
+
+        self.assertFalse(result.healthy)
+
+    @mock.patch('monasca_log_api.healthcheck.kafka_check.client.KafkaClient')
+    def test_should_fail_topic_missing(self, kafka_client):
+        kafka = mock.Mock()
+        kafka.topic_partitions = [self.mocked_topics[0]]
+        kafka_client.return_value = kafka
+
+        kafka_health = kc.KafkaHealthCheck()
+        result = kafka_health.healthcheck()
+
+        # verify result
+        self.assertFalse(result.healthy)
+
+        # ensure client was closed
+        self.assertTrue(kafka.close.called)
+
+    @mock.patch('monasca_log_api.healthcheck.kafka_check.client.KafkaClient')
+    def test_should_pass(self, kafka_client):
+        kafka = mock.Mock()
+        kafka.topic_partitions = self.mocked_topics
+        kafka_client.return_value = kafka
+
+        kafka_health = kc.KafkaHealthCheck()
+        result = kafka_health.healthcheck()
+
+        self.assertTrue(result)
+
+        # ensure client was closed
+        self.assertTrue(kafka.close.called)

monasca_log_api/tests/test_keystone_protocol.py

@@ -0,0 +1,43 @@
+# Copyright 2015 FUJITSU 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.
+
+import unittest
+
+import mock
+
+from monasca_log_api.healthcheck import keystone_protocol
+
+_APP = mock.Mock()
+_CONF = {}
+
+
+class TestKeystoneProtocol(unittest.TestCase):
+    def test_should_return_none_if_healthcheck(self):
+        instance = keystone_protocol.SkippingAuthProtocol(_APP, _CONF)
+        request = mock.Mock()
+        request.path = '/healthcheck'
+
+        ret_val = instance.process_request(request)
+
+        self.assertIsNone(ret_val)
+
+    @mock.patch('keystonemiddleware.auth_token.AuthProtocol.process_request')
+    def test_should_enter_keystone_auth_if_not_healthcheck(self, proc_request):
+        instance = keystone_protocol.SkippingAuthProtocol(_APP, _CONF)
+        request = mock.Mock()
+        request.path = '/v2.0/logs/single'
+
+        instance.process_request(request)
+
+        self.assertTrue(proc_request.called)

monasca_log_api/v2/reference/healthchecks.py

@@ -0,0 +1,59 @@
+# Copyright 2015 FUJITSU 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.
+
+import falcon
+
+from monasca_common.rest import utils as rest_utils
+
+from monasca_log_api.api import healthcheck_api
+from monasca_log_api.healthcheck import kafka_check
+
+
+class HealthChecks(healthcheck_api.HealthChecksApi):
+    # response configuration
+    CACHE_CONTROL = ['must-revalidate', 'no-cache', 'no-store']
+
+    # response codes
+    HEALTHY_CODE_GET = falcon.HTTP_OK
+    HEALTHY_CODE_HEAD = falcon.HTTP_NO_CONTENT
+    NOT_HEALTHY_CODE = falcon.HTTP_SERVICE_UNAVAILABLE
+
+    def __init__(self):
+        self._kafka_check = kafka_check.KafkaHealthCheck()
+        super(HealthChecks, self).__init__()
+
+    def on_head(self, req, res):
+        res.status = self.HEALTHY_CODE_HEAD
+        res.cache_control = self.CACHE_CONTROL
+
+    def on_get(self, req, res):
+        # at this point we know API is alive, so
+        # keep up good work and verify kafka status
+
+        kafka_result = self._kafka_check.healthcheck()
+
+        # in case it'd be unhealthy,
+        # message will contain error string
+        status_data = {
+            'kafka': kafka_result.message
+        }
+
+        # Really simple approach, ideally that should be
+        # part of monasca-common with some sort of registration of
+        # healthchecks concept
+
+        res.status = (self.HEALTHY_CODE_GET
+                      if kafka_result.healthy else self.NOT_HEALTHY_CODE)
+        res.cache_control = self.CACHE_CONTROL
+        res.body = rest_utils.as_json(status_data)

tox.ini

@@ -1,4 +1,5 @@
 [tox]
+# TODO(trebskit) Add pypy to envlist ?
 envlist = py27,py3,pep8
 skipsdist = True