Add healtcheck endpoint

Provide new healtcheck endpoint to monasca-api.
Add simple check for HEAD and complex check for GET.
Complex check contains information about dependent
services like:
- kafka
- relational database (mariadb, postgresql)
- timeseries database (influxdb, cassandra)

Story: 2000974
Task: 4125

Change-Id: I863071194041a512b144262bbffce5024b97086b

devstack/files/monasca-api/python/api-config.conf

@@ -20,6 +20,7 @@ notification_methods = monasca_api.v2.reference.notifications:Notifications
 dimension_values = monasca_api.v2.reference.metrics:DimensionValues
 dimension_names = monasca_api.v2.reference.metrics:DimensionNames
 notification_method_types = monasca_api.v2.reference.notificationstype:NotificationsType
+healthchecks = monasca_api.healthchecks:HealthChecks
 
 [security]
 # The roles that are allowed full access to the API.
@@ -69,6 +70,12 @@ uri = %KAFKA_HOST%:9092
 # The topic that metrics will be published too
 metrics_topic = metrics
 
+# The topic that events will be published too
+events_topic = events
+
+# The topic that alarm state will be published too
+alarm_state_transitions_topic = alarm-state-transitions
+
 # consumer group name
 group = api
 

devstack/files/monasca-api/python/api-config.ini

@@ -8,7 +8,7 @@ pipeline = request_id auth api
 paste.app_factory = monasca_api.api.server:launch
 
 [filter:auth]
-paste.filter_factory = keystonemiddleware.auth_token:filter_factory
+paste.filter_factory = monasca_api.healthcheck.keystone_protocol:filter_factory
 
 [filter:request_id]
 paste.filter_factory = oslo_middleware.request_id:RequestId.factory

docs/monasca-api-spec.md

@@ -58,6 +58,9 @@ Document Version: v2.0
       - [Status code](#status-code-1)
       - [Response Body](#response-body-1)
       - [Response Examples](#response-examples-1)
+- [Health Check](#heltchcheck)
+  - [Complex check](#complex_check)
+  - [Simple check](#simple_check)
 - [Metrics](#metrics)
   - [Create Metric](#create-metric)
     - [POST /v2.0/metrics](#post-v20metrics)
@@ -973,6 +976,35 @@ Returns a JSON version object with details about the specified version.
 ```
 ___
 
+# Healthcheck
+The Monasca ApI comes with a built-in healthcheck mechanism. It is available in two flavours, both accessible
+under `/healthcheck` endopoint.
+
+## Complex check
+The complex check not only returns a response with success code if Monasca API is up and running by it also verifies if
+dependant components , such as __Kafka__, __Alarm database__ (MariadDB/MySQL, PostgreSQL), __Metrics database__ (Cassandra, InfluxdDB)
+are healthy too.
+
+Monasca API will respond with following codes:
+* 200 - both API and external components are healthy.
+* 503 - API is running but problems with peripheral components have been spotted.
+
+Example: `curl -XGET 192.168.10.6:8070/healthcheck`
+
+### Peripheral checks
+* __Kafka__ is considered healthy if connection to broker can be established and configured topics can be found.
+* __Alarm Database__ (MariaDB/MySQL, PostgreSQL) is considered healthy if connection to database can be established
+  and sample query can be executed.
+* __Time Series Database__ (TSDB) is considered healthy if: `InfluxDB` is set health check is verified according to the
+InfluxDB documentation ([/ping](https://docs.influxdata.com/influxdb/v1.1/tools/api/)), `Cassandra` is set health check is verified through new connection to the database.
+
+## Simple check
+The simple check only returns response only if Monasca API is up and running. It does not return any data
+because it is accessible only for `HEAD` request. If the Monasca Api is up and running the following response code:
+`204` is expected.
+
+Example: `curl -XHEAD 192.168.10.6:8070/healtcheck`
+
 # Metrics
 The metrics resource allows metrics to be created and queried. The `X-Auth-Token` is used to derive the tenant that submits metrics. Metrics are stored and scoped to the tenant that submits them, or if the `tenant_id` query parameter is specified and the tenant has the `monitoring-delegate` role, the metrics are stored using the specified tenant ID.  Note that several of the GET methods also support the tenant_id query parameter, but the `monasca-admin` role is required to get cross-tenant metrics, statistics, etc..
 

etc/api-config.conf

@@ -20,6 +20,7 @@ notification_methods = monasca_api.v2.reference.notifications:Notifications
 dimension_values = monasca_api.v2.reference.metrics:DimensionValues
 dimension_names = monasca_api.v2.reference.metrics:DimensionNames
 notification_method_types = monasca_api.v2.reference.notificationstype:NotificationsType
+healthchecks = monasca_api.healthchecks:HealthChecks
 
 [security]
 # The roles that are allowed full access to the API.
@@ -68,6 +69,12 @@ uri = 192.168.10.4:9092
 # The topic that metrics will be published too
 metrics_topic = metrics
 
+# The topic that events will be published too
+events_topic = events
+
+# The topic that alarm state will be published too
+alarm_state_transitions_topic = alarm-state-transitions
+
 # consumer group name
 group = api
 

etc/api-config.ini

@@ -8,7 +8,7 @@ pipeline = request_id auth api
 paste.app_factory = monasca_api.api.server:launch
 
 [filter:auth]
-paste.filter_factory = keystonemiddleware.auth_token:filter_factory
+paste.filter_factory = monasca_api.healthcheck.keystone_protocol:filter_factory
 
 [filter:request_id]
 paste.filter_factory = oslo_middleware.request_id:RequestId.factory

monasca_api/api/healthcheck_api.py

@@ -0,0 +1,53 @@
+# Copyright 2017 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 oslo_log import log
+
+LOG = log.getLogger(__name__)
+
+
+class HealthCheckApi(object):
+    """HealthCheck Api.
+
+    HealthCheckApi server information regarding health of the Api.
+
+    """
+
+    def __init__(self):
+        super(HealthCheckApi, self).__init__()
+        LOG.info('Initializing HealthCheckApi')
+
+    def on_get(self, req, res):
+        """Complex  healthcheck report on GET
+
+        Returns complex report regarding API health
+        and all dependent services
+
+        :param falcon.Request req: current request
+        :param falcon.Response res: current response
+        """
+        res.status = falcon.HTTP_501  # pragma: no cover
+
+    def on_head(self, req, res):
+        """Simple healthcheck report on HEAD.
+
+        In opposite to :py:meth:`.HealthCheckApi.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  # pragma: no cover

monasca_api/api/server.py

@@ -51,7 +51,9 @@ dispatcher_opts = [cfg.StrOpt('versions', default=None,
                    cfg.StrOpt('dimension_names', default=None,
                               help='Dimension names'),
                    cfg.StrOpt('notification_method_types', default=None,
-                              help='notification_method_types methods')]
+                              help='notification_method_types methods'),
+                   cfg.StrOpt('healthchecks', default=None,
+                              help='Health checks endpoint')]
 
 dispatcher_group = cfg.OptGroup(name='dispatcher', title='dispatcher')
 cfg.CONF.register_group(dispatcher_group)
@@ -129,6 +131,9 @@ def launch(conf):
         cfg.CONF.dispatcher.notification_method_types)()
     app.add_route("/v2.0/notification-methods/types", notification_method_types)
 
+    healthchecks = simport.load(cfg.CONF.dispatcher.healthchecks)()
+    app.add_route("/healthcheck", healthchecks)
+
     LOG.debug('Dispatcher drivers have been added to the routes!')
     return app
 

monasca_api/common/repositories/exceptions.py

@@ -23,5 +23,10 @@ from monasca_common.repositories.exceptions import RepositoryException
 class MultipleMetricsException(RepositoryException):
     pass
 
+
+class UnsupportedDriverException(Exception):
+    pass
+
+
 __all__ = (AlreadyExistsException, DoesNotExistException, InvalidUpdateException,
-           RepositoryException, MultipleMetricsException)
+           RepositoryException, MultipleMetricsException, UnsupportedDriverException)

monasca_api/healthcheck/alarms_db_check.py

@@ -0,0 +1,49 @@
+# Copyright 2017 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 oslo_log import log
+from sqlalchemy import text
+
+from monasca_api.common.repositories.sqla import sql_repository
+from monasca_api.healthcheck import base
+
+LOG = log.getLogger(__name__)
+
+
+class AlarmsDbHealthCheck(base.BaseHealthCheck,
+                          sql_repository.SQLRepository):
+    """Evaluates alarm db health
+
+    Healthcheck verifies if:
+    * database is up and running, it is possible to establish connection
+    * sample sql query can be executed
+
+    If following conditions are met health check return healthy status.
+    Otherwise unhealthy status is returned with explanation.
+    """
+
+    def health_check(self):
+        status = self.check_db_status()
+        return base.CheckResult(healthy=status[0],
+                                message=status[1])
+
+    def check_db_status(self):
+        try:
+            with self._db_engine.connect() as con:
+                query = text('SELECT 1')
+                con.execute(query)
+        except Exception as ex:
+            LOG.exception(str(ex))
+            return False, str(ex)
+        return True, 'OK'

monasca_api/healthcheck/base.py

@@ -0,0 +1,36 @@
+# Copyright 2017 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 abc
+import collections
+
+import six
+
+
+class CheckResult(collections.namedtuple('CheckResult', ['healthy', 'message'])):
+    """Result for the health check
+
+    healthy - boolean
+    message - string
+    """
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BaseHealthCheck(object):
+    """Abstract class implemented by the monasca-api healthcheck classes"""
+
+    @abc.abstractmethod
+    def health_check(self):
+        """Evaluate health of given service"""
+        raise NotImplementedError  # pragma: no cover

monasca_api/healthcheck/kafka_check.py

@@ -0,0 +1,76 @@
+# Copyright 2017 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 oslo_config import cfg
+from oslo_log import log
+
+from monasca_api.healthcheck import base
+from monasca_common.kafka_lib import client
+
+LOG = log.getLogger(__name__)
+CONF = cfg.CONF
+
+
+class KafkaHealthCheck(base.BaseHealthCheck):
+    """Evaluates kafka health
+
+    Healthcheck verifies if:
+
+    * kafka server is up and running
+    * there is a configured topic in kafka
+
+    If following conditions are met health check returns healthy status.
+    Otherwise unhealthy status is returned with message.
+
+    Note:
+        Healthcheck checks 3 type of topics given in configuration:
+        metrics_topic, events_topic and alarm_state_transition_topic.
+    """
+
+    def health_check(self):
+        url = CONF.kafka.uri
+
+        try:
+            kafka_client = client.KafkaClient(hosts=url)
+        except client.KafkaUnavailableError as ex:
+            LOG.error(repr(ex))
+            error_str = 'Could not connect to Kafka at {0}'.format(url)
+            return base.CheckResult(healthy=False, message=error_str)
+
+        status = self._verify_topics(kafka_client)
+        self._disconnect_gracefully(kafka_client)
+
+        return base.CheckResult(healthy=status[0],
+                                message=status[1])
+
+    @staticmethod
+    def _verify_topics(kafka_client):
+        topics = (CONF.kafka.metrics_topic,
+                  CONF.kafka.events_topic,
+                  CONF.kafka.alarm_state_transitions_topic)
+
+        for topic in topics:
+            for_topic = topic in kafka_client.topic_partitions
+            if not for_topic:
+                error_str = 'Kafka: Topic {0} not found'.format(for_topic)
+                LOG.error(error_str)
+                return False, str(error_str)
+        return True, 'OK'
+
+    @staticmethod
+    def _disconnect_gracefully(kafka_client):
+        try:
+            kafka_client.close()
+        except Exception:
+            LOG.exception('Closing Kafka Connection')

monasca_api/healthcheck/keystone_protocol.py

@@ -0,0 +1,65 @@
+# Copyright 2017 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__)
+
+_SKIP_PATH = '/version', '/healthcheck'
+"""Tuple of non-application endpoints"""
+
+
+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
+        for p in _SKIP_PATH:
+            if path.startswith(p):
+                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_api/healthcheck/metrics_db_check.py

@@ -0,0 +1,89 @@
+# Copyright 2017 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 requests
+
+from cassandra import cluster
+from oslo_config import cfg
+from oslo_log import log
+
+from monasca_api.common.repositories import exceptions
+from monasca_api.healthcheck import base
+
+LOG = log.getLogger(__name__)
+CONF = cfg.CONF
+
+
+class MetricsDbCheck(base.BaseHealthCheck):
+    """Evaluates metrics db health
+
+    Healthcheck what type of database is used (InfluxDB, Cassandra)
+    and provide health according to the given db.
+
+    Healthcheck for InfluxDB verifies if:
+    * check the db status by the /ping endpoint.
+
+    Healthcheck for the Cassandra verifies if:
+    * Cassandra is up and running (it is possible to create new connection)
+    * keyspace exists
+
+    If following conditions are met health check return healthy status.
+    Otherwise unhealthy status is returned with explanation.
+    """
+
+    def health_check(self):
+        metric_driver = CONF.repositories.metrics_driver
+        db = self._detected_database_type(metric_driver)
+
+        if db == 'influxdb':
+            status = self._check_influxdb_status()
+        else:
+            status = self._check_cassandra_status()
+
+        return base.CheckResult(healthy=status[0],
+                                message=status[1])
+
+    def _detected_database_type(self, driver):
+        if 'influxdb' in driver:
+            return 'influxdb'
+        elif 'cassandra' in driver:
+            return 'cassandra'
+        else:
+            raise exceptions.UnsupportedDriverException(
+                'Driver {0} is not supported by Healthcheck'.format(driver))
+
+    @staticmethod
+    def _check_influxdb_status():
+        uri = 'http://{0}:{1}/ping'.format(CONF.influxdb.ip_address,
+                                           CONF.influxdb.port)
+        try:
+            resp = requests.head(url=uri)
+        except Exception as ex:
+            LOG.exception(str(ex))
+            return False, str(ex)
+        return resp.ok, 'OK' if resp.ok else 'Error: {0}'.format(
+            resp.status_code)
+
+    @staticmethod
+    def _check_cassandra_status():
+        try:
+            cassandra = cluster.Cluster(
+                CONF.cassandra.cluster_ip_addresses.split(',')
+            )
+            session = cassandra.connect(CONF.cassandra.keyspace)
+            session.shutdown()
+        except Exception as ex:
+            LOG.exception(str(ex))
+            return False, str(ex)
+        return True, 'OK'

monasca_api/healthchecks.py

@@ -0,0 +1,56 @@
+# Copyright 2017 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_api.api import healthcheck_api
+from monasca_api.healthcheck import alarms_db_check
+from monasca_api.healthcheck import kafka_check
+from monasca_api.healthcheck import metrics_db_check
+from monasca_api.v2.reference import helpers
+
+
+class HealthChecks(healthcheck_api.HealthCheckApi):
+    CACHE_CONTROL = ['must-revalidate', 'no-cache', 'no-store']
+
+    HEALTHY_CODE_GET = falcon.HTTP_OK
+    HEALTHY_CODE_HEAD = falcon.HTTP_NO_CONTENT
+    NOT_HEALTHY_CODE = falcon.HTTP_SERVICE_UNAVAILABLE
+
+    def __init__(self):
+        super(HealthChecks, self).__init__()
+        self._kafka_check = kafka_check.KafkaHealthCheck()
+        self._alarm_db_check = alarms_db_check.AlarmsDbHealthCheck()
+        self._metrics_db_check = metrics_db_check.MetricsDbCheck()
+
+    def on_head(self, req, res):
+        res.status = self.HEALTHY_CODE_HEAD
+        res.cache_control = self.CACHE_CONTROL
+
+    def on_get(self, req, res):
+        kafka_result = self._kafka_check.health_check()
+        alarms_db_result = self._alarm_db_check.health_check()
+        metrics_db_result = self._metrics_db_check.health_check()
+
+        status_data = {
+            'kafka': kafka_result.message,
+            'alarms_database': alarms_db_result.message,
+            'metrics_database': metrics_db_result.message
+        }
+        health = (kafka_result.healthy and alarms_db_result.healthy and
+                  metrics_db_result.healthy)
+        res.status = (self.HEALTHY_CODE_GET
+                      if health else self.NOT_HEALTHY_CODE)
+        res.cache_control = self.CACHE_CONTROL
+        res.body = helpers.to_json(status_data)

monasca_api/tests/test_alarms_db_health_check.py

@@ -0,0 +1,61 @@
+# Copyright 2017 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 mock
+from oslo_config import fixture as oo_cfg
+from oslotest import base
+
+from monasca_api.healthcheck import alarms_db_check as rdc
+from monasca_api.v2.reference import cfg
+
+CONF = cfg.CONF
+
+
+class TestMetricsDbHealthCheckLogic(base.BaseTestCase):
+
+    db_connection = "mysql+pymysql://test:test@localhost/mon?charset=utf8mb4"
+    mocked_config = {
+        'connection': db_connection
+    }
+
+    def __init__(self, *args, **kwargs):
+        super(TestMetricsDbHealthCheckLogic, self).__init__(*args, **kwargs)
+        self._conf = None
+
+    def setUp(self):
+        super(TestMetricsDbHealthCheckLogic, self).setUp()
+        self._conf = self.useFixture(oo_cfg.Config(CONF))
+        self._conf.config(group='database', **self.mocked_config)
+
+    @mock.patch('monasca_api.healthcheck.alarms_db_check.'
+                'sql_repository.get_engine')
+    def test_should_pass_db_ok(self, _):
+
+        db_health = rdc.AlarmsDbHealthCheck()
+        db_health.check_db_status = mock.Mock(return_value=(True, 'OK'))
+        result = db_health.health_check()
+
+        self.assertTrue(result.healthy)
+        self.assertEqual('OK', result.message)
+
+    @mock.patch('monasca_api.healthcheck.alarms_db_check.'
+                'sql_repository.get_engine')
+    def test_should_fail_db_unavailable(self, _):
+
+        db_health = rdc.AlarmsDbHealthCheck()
+        db_health.check_db_status = mock.Mock(return_value=(False, 'bar'))
+        result = db_health.health_check()
+
+        self.assertFalse(result.healthy)
+        self.assertEqual('bar', result.message)

monasca_api/tests/test_healthchecks.py

@@ -0,0 +1,137 @@
+# Copyright 2017 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
+from oslo_config import fixture as oo_cfg
+
+from monasca_api.healthcheck import base
+from monasca_api import healthchecks
+from monasca_api.v2.reference import cfg
+from monasca_common.rest import utils
+
+CONF = cfg.CONF
+ENDPOINT = '/healthcheck'
+
+
+class TestHealthChecks(testing.TestBase):
+    def setUp(self):
+        super(TestHealthChecks, self).setUp()
+        self.conf = self.useFixture(oo_cfg.Config(CONF))
+
+    def set_route(self):
+        self.resources = healthchecks.HealthChecks()
+        self.api.add_route(
+            ENDPOINT,
+            self.resources
+        )
+
+    @mock.patch('monasca_api.healthcheck.alarms_db_check.sql_repository.get_engine')
+    def test_should_return_200_for_head(self, _):
+        self.set_route()
+        self.simulate_request(ENDPOINT, method='HEAD')
+        self.assertEqual(falcon.HTTP_NO_CONTENT, self.srmock.status)
+
+    @mock.patch('monasca_api.healthcheck.kafka_check.KafkaHealthCheck')
+    @mock.patch(
+        'monasca_api.healthcheck.alarms_db_check.AlarmsDbHealthCheck')
+    @mock.patch(
+        'monasca_api.healthcheck.metrics_db_check.MetricsDbCheck')
+    @mock.patch(
+        'monasca_api.healthcheck.alarms_db_check.sql_repository.SQLRepository')
+    def test_should_report_healthy_if_all_services_healthy(self, kafka_check,
+                                                           alarms_db_check,
+                                                           metrics_db_check,
+                                                           _):
+        kafka_check.health_check.return_value = base.CheckResult(True, 'OK')
+        alarms_db_check.health_check.return_value = base.CheckResult(True,
+                                                                     'OK')
+        metrics_db_check.health_check.return_value = base.CheckResult(True,
+                                                                      'OK')
+        self.set_route()
+        self.resources._kafka_check = kafka_check
+        self.resources._alarm_db_check = alarms_db_check
+        self.resources._metrics_db_check = metrics_db_check
+
+        response = self.simulate_request(ENDPOINT,
+                                         headers={
+                                             'Content-Type': 'application/json'
+                                         },
+                                         decode='utf8',
+                                         method='GET')
+        self.assertEqual(falcon.HTTP_OK, self.srmock.status)
+
+        response = utils.from_json(response)
+        self.assertIn('kafka', response)
+        self.assertIn('alarms_database', response)
+        self.assertIn('metrics_database', response)
+        self.assertEqual('OK', response.get('kafka'))
+        self.assertEqual('OK', response.get('alarms_database'))
+        self.assertEqual('OK', response.get('metrics_database'))
+
+    @mock.patch('monasca_api.healthcheck.kafka_check.KafkaHealthCheck')
+    @mock.patch(
+        'monasca_api.healthcheck.alarms_db_check.AlarmsDbHealthCheck')
+    @mock.patch(
+        'monasca_api.healthcheck.metrics_db_check.MetricsDbCheck')
+    @mock.patch(
+        'monasca_api.healthcheck.alarms_db_check.sql_repository.SQLRepository')
+    def test_should_report_not_healthy_if_one_service_not_healthy(self,
+                                                                  kafka_check,
+                                                                  alarms_db_check,
+                                                                  metrics_db_check,
+                                                                  _):
+        test_list = [
+            {'kafka': {'healthy': False, 'message': 'Unavailable'},
+             'alarms_db': {'healthy': True, 'message': 'OK'},
+             'netrics_db': {'healthy': True, 'message': 'OK'}
+             },
+            {'kafka': {'healthy': True, 'message': 'OK'},
+             'alarms_db': {'healthy': False, 'message': 'Connection Error'},
+             'netrics_db': {'healthy': True, 'message': 'OK'}
+             },
+            {'kafka': {'healthy': True, 'message': 'OK'},
+             'alarms_db': {'healthy': True, 'message': 'OK'},
+             'netrics_db': {'healthy': False, 'message': 'Error'}
+             },
+        ]
+
+        for service in test_list:
+            kafka_check.health_check.return_value = base.CheckResult(service['kafka']['healthy'],
+                                                                     service['kafka']['message'])
+            alarms_db_check.health_check.return_value = base.CheckResult(service['alarms_db']['healthy'],
+                                                                         service['alarms_db']['message'])
+            metrics_db_check.health_check.return_value = base.CheckResult(service['netrics_db']['healthy'],
+                                                                          service['netrics_db']['message'])
+            self.set_route()
+            self.resources._kafka_check = kafka_check
+            self.resources._alarm_db_check = alarms_db_check
+            self.resources._metrics_db_check = metrics_db_check
+
+            response = self.simulate_request(ENDPOINT,
+                                             headers={
+                                                 'Content-Type': 'application/json'
+                                             },
+                                             decode='utf8',
+                                             method='GET')
+            self.assertEqual(falcon.HTTP_SERVICE_UNAVAILABLE, self.srmock.status)
+
+            response = utils.from_json(response)
+            self.assertIn('kafka', response)
+            self.assertIn('alarms_database', response)
+            self.assertIn('metrics_database', response)
+            self.assertEqual(service['kafka']['message'], response.get('kafka'))
+            self.assertEqual(service['alarms_db']['message'], response.get('alarms_database'))
+            self.assertEqual(service['netrics_db']['message'], response.get('metrics_database'))

monasca_api/tests/test_kafka_health_check.py

@@ -0,0 +1,85 @@
+# Copyright 2017 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 mock
+from oslo_config import fixture as oo_cfg
+from oslotest import base
+
+from monasca_common.kafka_lib import client
+
+from monasca_api.healthcheck import kafka_check as kc
+from monasca_api.v2.reference import cfg
+
+CONF = cfg.CONF
+
+
+class TestKafkaHealthCheckLogic(base.BaseTestCase):
+
+    mock_kafka_url = 'localhost:1234'
+    mocked_topics = 'test1'
+    mocked_event_topic = 'test2'
+    mocked_alarm_state_topic = 'test3'
+    mocked_config = {
+        'uri': mock_kafka_url,
+        'metrics_topic': mocked_topics,
+        'events_topic': mocked_event_topic,
+        'alarm_state_transitions_topic': mocked_alarm_state_topic
+    }
+
+    def __init__(self, *args, **kwargs):
+        super(TestKafkaHealthCheckLogic, self).__init__(*args, **kwargs)
+        self._conf = None
+
+    def setUp(self):
+        super(TestKafkaHealthCheckLogic, self).setUp()
+        self._conf = self.useFixture(oo_cfg.Config(CONF))
+        self._conf.config(group='kafka', **self.mocked_config)
+
+    @mock.patch('monasca_api.healthcheck.kafka_check.client.KafkaClient')
+    def test_should_fail_kafka_unavailable(self, kafka_client):
+        kafka = mock.Mock()
+        kafka_client.side_effect = client.KafkaUnavailableError()
+        kafka_client.return_value = kafka
+
+        kafka_health = kc.KafkaHealthCheck()
+        result = kafka_health.health_check()
+
+        self.assertFalse(result.healthy)
+        kafka.close.assert_not_called()
+
+    @mock.patch('monasca_api.healthcheck.kafka_check.client.KafkaClient')
+    def test_should_fail_missing_topic(self, kafka_client):
+        kafka = mock.Mock()
+        kafka.topic_partitions = ['topic1']
+        kafka_client.return_value = kafka
+
+        kafka_health = kc.KafkaHealthCheck()
+        result = kafka_health.health_check()
+
+        self.assertFalse(result.healthy)
+        kafka.close.assert_called_once()
+
+    @mock.patch('monasca_api.healthcheck.kafka_check.client.KafkaClient')
+    def test_should_pass(self, kafka_client):
+        kafka = mock.Mock()
+        kafka.topic_partitions = (self.mocked_topics,
+                                  self.mocked_event_topic,
+                                  self.mocked_alarm_state_topic)
+        kafka_client.return_value = kafka
+
+        kafka_health = kc.KafkaHealthCheck()
+        result = kafka_health.health_check()
+
+        self.assertTrue(result.healthy)
+        kafka.close.assert_called_once()

monasca_api/tests/test_keystone_protocol.py

@@ -0,0 +1,44 @@
+# Copyright 2017 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 mock
+from oslotest import base
+
+from monasca_api.healthcheck import keystone_protocol
+
+_CONF = {}
+
+
+class TestKeystoneProtocol(base.BaseTestCase):
+
+    def test_should_return_none_if_healthcheck(self):
+        mocked_api = mock.Mock()
+        instance = keystone_protocol.SkippingAuthProtocol(mocked_api, _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):
+        mocked_api = mock.Mock()
+        instance = keystone_protocol.SkippingAuthProtocol(mocked_api, _CONF)
+        request = mock.Mock()
+        request.path = '/v2.0/logs/single'
+
+        instance.process_request(request)
+
+        self.assertTrue(proc_request.called)

monasca_api/tests/test_metrics_db_health_check.py

@@ -0,0 +1,187 @@
+# Copyright 2017 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 cassandra import cluster as cl
+import requests
+
+import mock
+from oslo_config import fixture as oo_cfg
+from oslotest import base
+
+from monasca_api.common.repositories import exceptions
+from monasca_api.healthcheck import metrics_db_check as tdc
+from monasca_api.v2.reference import cfg
+
+CONF = cfg.CONF
+
+
+class TestMetricsDbHealthCheck(base.BaseTestCase):
+    cassandra_conf = {
+        'cluster_ip_addresses': 'localhost',
+        'keyspace': 'test'
+    }
+
+    def __init__(self, *args, **kwargs):
+        super(TestMetricsDbHealthCheck, self).__init__(*args, **kwargs)
+        self._conf = None
+
+    def setUp(self):
+        super(TestMetricsDbHealthCheck, self).setUp()
+        self._conf = self.useFixture(oo_cfg.Config(CONF))
+        self._conf.config(group='cassandra', **self.cassandra_conf)
+
+    def test_should_detect_influxdb_db(self):
+        db_health = tdc.MetricsDbCheck()
+
+        # check if influxdb is detected
+        self.assertEqual('influxdb', db_health._detected_database_type(
+            'influxdb.metrics_repository'))
+
+    def test_should_detect_cassandra_db(self):
+        db_health = tdc.MetricsDbCheck()
+        # check if cassandra is detected
+        self.assertEqual('cassandra', db_health._detected_database_type(
+            'cassandra.metrics_repository'))
+
+    def test_should_raise_exception_during_db_detection(self):
+        db_health = tdc.MetricsDbCheck()
+        # check exception
+        db = 'postgresql.metrics_repository'
+        self.assertRaises(exceptions.UnsupportedDriverException, db_health._detected_database_type, db)
+
+    @mock.patch.object(requests, 'head')
+    def test_should_fail_influxdb_connection(self, req):
+        response_mock = mock.Mock()
+        response_mock.ok = False
+        response_mock.status_code = 500
+        req.return_value = response_mock
+
+        influxdb_conf = {
+            'ip_address': 'localhost',
+            'port': 8086
+        }
+        messaging_conf = {
+            'metrics_driver': 'influxdb.metrics_repository:MetricsRepository'
+        }
+        self._conf.config(group='repositories', **messaging_conf)
+        self._conf.config(group='influxdb', **influxdb_conf)
+
+        db_health = tdc.MetricsDbCheck()
+        result = db_health.health_check()
+
+        self.assertFalse(result.healthy)
+        self.assertEqual('Error: 500', result.message)
+
+    @mock.patch.object(requests, 'head')
+    def test_should_fail_influxdb_wrong_port_number(self, req):
+        response_mock = mock.Mock()
+        response_mock.ok = False
+        response_mock.status_code = 404
+        req.return_value = response_mock
+        influxdb_conf = {
+            'ip_address': 'localhost',
+            'port': 8099
+        }
+        messaging_conf = {
+            'metrics_driver': 'influxdb.metrics_repository:MetricsRepository'
+        }
+        self._conf.config(group='repositories', **messaging_conf)
+        self._conf.config(group='influxdb', **influxdb_conf)
+
+        db_health = tdc.MetricsDbCheck()
+        result = db_health.health_check()
+
+        self.assertFalse(result.healthy)
+        self.assertEqual('Error: 404', result.message)
+
+    @mock.patch.object(requests, 'head')
+    def test_should_fail_influxdb_service_unavailable(self, req):
+        response_mock = mock.Mock()
+        req.side_effect = requests.HTTPError()
+        req.return_value = response_mock
+        influxdb_conf = {
+            'ip_address': 'localhost',
+            'port': 8096
+        }
+        messaging_conf = {
+            'metrics_driver': 'influxdb.metrics_repository:MetricsRepository'
+        }
+        self._conf.config(group='repositories', **messaging_conf)
+        self._conf.config(group='influxdb', **influxdb_conf)
+
+        db_health = tdc.MetricsDbCheck()
+        result = db_health.health_check()
+
+        self.assertFalse(result.healthy)
+
+    @mock.patch.object(requests, 'head')
+    def test_should_pass_infuxdb_available(self, req):
+        response_mock = mock.Mock()
+        response_mock.ok = True
+        response_mock.status_code = 204
+        req.return_value = response_mock
+        influxdb_conf = {
+            'ip_address': 'localhost',
+            'port': 8086
+        }
+        messaging_conf = {
+            'metrics_driver': 'influxdb.metrics_repository:MetricsRepository'
+        }
+        self._conf.config(group='repositories', **messaging_conf)
+        self._conf.config(group='influxdb', **influxdb_conf)
+
+        db_health = tdc.MetricsDbCheck()
+        result = db_health.health_check()
+
+        self.assertTrue(result.healthy)
+        self.assertEqual('OK', result.message)
+
+    @mock.patch('monasca_api.healthcheck.metrics_db_check.cluster.Cluster')
+    def test_should_fail_cassandra_unavailable(self, cluster):
+        messaging_conf = {
+            'metrics_driver': 'cassandra.metrics_repository:MetricsRepository'
+        }
+        cassandra_conf = {
+            'cluster_ip_addresses': 'localhost',
+            'keyspace': 'test'
+        }
+        self._conf.config(group='repositories', **messaging_conf)
+        self._conf.config(group='cassandra', **cassandra_conf)
+
+        cas_mock = mock.Mock()
+        cluster.side_effect = cl.NoHostAvailable(message='Host unavailable',
+                                                 errors='Unavailable')
+        cluster.return_value = cas_mock
+
+        db_health = tdc.MetricsDbCheck()
+        result = db_health.health_check()
+
+        self.assertFalse(result.healthy)
+
+    @mock.patch('monasca_api.healthcheck.metrics_db_check.cluster.Cluster')
+    def test_should_pass_cassandra_is_available(self, _):
+        messaging_conf = {
+            'metrics_driver': 'cassandra.metrics_repository:MetricsRepository'
+        }
+        cassandra_conf = {
+            'cluster_ip_addresses': 'localhost',
+            'keyspace': 'test'
+        }
+        self._conf.config(group='repositories', **messaging_conf)
+        self._conf.config(group='cassandra', **cassandra_conf)
+
+        db_health = tdc.MetricsDbCheck()
+        result = db_health.health_check()
+
+        self.assertTrue(result.healthy)

monasca_api/v2/reference/__init__.py

@@ -94,9 +94,14 @@ cfg.CONF.register_opts(repositories_opts, repositories_group)
 kafka_opts = [cfg.StrOpt('uri', help='Address to kafka server. For example: '
                                      'uri=192.168.1.191:9092'),
               cfg.StrOpt('metrics_topic', default='metrics',
-                         help='The topic that metrics will be published too.'),
-              cfg.StrOpt('events_topic', default='raw-events',
-                         help='The topic that events will be published too.'),
+                         help='The topic that metrics will be published too.',
+                         advanced=True),
+              cfg.StrOpt('events_topic', default='events',
+                         help='The topic that events will be published too.',
+                         advanced=True),
+              cfg.StrOpt('alarm_state_transitions_topic', default='alarm-state-transitions',
+                         help='The topic that alarm state will be published too.',
+                         advanced=True),
               cfg.StrOpt('group', default='api',
                          help='The group name that this service belongs to.'),
               cfg.IntOpt('wait_time', default=1,

monasca_api/v2/reference/alarming.py

@@ -36,10 +36,10 @@ class Alarming(object):
         super(Alarming, self).__init__()
 
         self.events_message_queue = simport.load(
-            cfg.CONF.messaging.driver)('events')
+            cfg.CONF.messaging.driver)(cfg.CONF.kafka.events_topic)
 
         self.alarm_state_transitions_message_queue = simport.load(
-            cfg.CONF.messaging.driver)('alarm-state-transitions')
+            cfg.CONF.messaging.driver)(cfg.CONF.kafka.alarm_state_transitions_topic)
 
     def _send_alarm_transitioned_event(self, tenant_id, alarm_id,
                                        alarm_definition_row,