Add incoming/indexer/storage information to V1 API's status endpoint

Author: leo-naeka

gnocchi/common/ceph.py

@@ -12,7 +12,9 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from collections import OrderedDict
 import errno
+import six
 
 import daiquiri
 
@@ -97,3 +99,22 @@ def errno_to_exception(ret):
             raise rados.Error("Unhandled error '%s'" % ret)
         else:
             raise getattr(rados, name)
+
+
+def get_ceph_health_status(driver):
+    """Return ceph status.
+
+    Include ceph stats.
+    """
+    response = OrderedDict([
+        ('name', driver.__class__.__name__)
+    ])
+    try:
+        stats = driver.rados.get_cluster_stats()
+    except Exception as e:
+        response['is_available'] = False
+        response['error'] = six.text_type(e)
+    else:
+        response['is_available'] = True
+        response['stats'] = stats
+    return response

gnocchi/common/file.py

@@ -0,0 +1,24 @@
+# -*- coding:Utf-8 -*-
+#
+# 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 __future__ import unicode_literals
+
+from collections import OrderedDict
+
+
+def get_file_health_status(driver):
+    """Return file status."""
+    return OrderedDict([
+        ('name', driver.__class__.__name__),
+        ('is_available', True)
+    ])

gnocchi/common/redis.py

@@ -16,6 +16,7 @@
 
 from __future__ import absolute_import
 
+from collections import OrderedDict
 from oslo_config import cfg
 import six
 from six.moves.urllib import parse
@@ -167,3 +168,22 @@ def get_client(conf, scripts=None):
         }
 
     return client, scripts
+
+
+def get_redis_health_status(driver):
+    """Return redis status.
+
+    Include redis info.
+    """
+    response = OrderedDict([
+        ('name', driver.__class__.__name__)
+    ])
+    try:
+        info = driver._client.info()
+    except Exception as e:
+        response['is_available'] = False
+        response['error'] = six.text_type(e)
+    else:
+        response['is_available'] = True
+        response['info'] = info
+    return response

gnocchi/common/s3.py

@@ -13,6 +13,10 @@
 # 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 collections import OrderedDict
+import six
+
 import daiquiri
 
 import tenacity
@@ -82,3 +86,18 @@ def bulk_delete(conn, bucket, objects):
         deleted += len(response['Deleted'])
     LOG.debug('%s objects deleted, %s objects skipped',
               deleted, len(objects) - deleted)
+
+
+def get_s3_health_status(driver):
+    """Return s3 status."""
+    response = OrderedDict([
+        ('name', driver.__class__.__name__)
+    ])
+    try:
+        driver.s3.head_bucket(Bucket=driver._bucket_name)
+    except Exception as e:
+        response['is_available'] = False
+        response['error'] = six.text_type(e)
+    else:
+        response['is_available'] = True
+    return response

gnocchi/common/sqlalchemy.py

@@ -0,0 +1,33 @@
+# -*- coding:Utf-8 -*-
+#
+# 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 __future__ import unicode_literals
+
+from collections import OrderedDict
+import six
+
+
+def get_sqlalchemy_health_status(driver):
+    """Return sqlalchemy status."""
+    response = OrderedDict([
+        ('name', driver.__class__.__name__)
+    ])
+    try:
+        with driver.facade.independent_reader() as session:
+            session.execute('SELECT 1')
+    except Exception as e:
+        response['is_available'] = False
+        response['error'] = six.text_type(e)
+    else:
+        response['is_available'] = True
+    return response

gnocchi/common/swift.py

@@ -11,9 +11,13 @@
 # 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 daiquiri
+
+from collections import OrderedDict
+import six
 from six.moves.urllib.parse import quote
 
+import daiquiri
+
 try:
     from swiftclient import client as swclient
     from swiftclient import utils as swift_utils
@@ -71,3 +75,22 @@ def bulk_delete(conn, container, objects):
     resp = swift_utils.parse_api_response(headers, body)
     LOG.debug('# of objects deleted: %s, # of objects skipped: %s',
               resp['Number Deleted'], resp['Number Not Found'])
+
+
+def get_swift_health_status(driver):
+    """Return swift status.
+
+    Include swift account info.
+    """
+    response = OrderedDict([
+        ('name', driver.__class__.__name__)
+    ])
+    try:
+        info = driver.swift.head_account()
+    except Exception as e:
+        response['is_available'] = False
+        response['error'] = six.text_type(e)
+    else:
+        response['is_available'] = True
+        response['info'] = info
+    return response

gnocchi/incoming/__init__.py

@@ -257,6 +257,10 @@ def finish_sack_processing(sack):
         """Mark sack processing has finished."""
         pass
 
+    @staticmethod
+    def get_health_status():
+        raise exceptions.NotImplementedError
+
 
 @utils.retry_on_exception_and_log("Unable to initialize incoming driver")
 def get_driver(conf):

gnocchi/incoming/ceph.py

@@ -232,3 +232,5 @@ def process_measures_for_sack(self, sack):
                 self.ioctx.remove_omap_keys(op, tuple(processed_keys))
                 self.ioctx.operate_write_op(op, str(sack),
                                             flags=self.OMAP_WRITE_FLAGS)
+
+    get_health_status = ceph.get_ceph_health_status

gnocchi/incoming/file.py

@@ -24,6 +24,7 @@
 import numpy
 import six
 
+from gnocchi.common.status import get_file_health_status
 from gnocchi import incoming
 from gnocchi import utils
 
@@ -206,3 +207,5 @@ def process_measures_for_sack(self, sack):
 
         for metric_id, files in six.iteritems(processed_files):
             self._delete_measures_files_for_metric(metric_id, files)
+
+    get_health_status = get_file_health_status

gnocchi/incoming/redis.py

@@ -193,3 +193,5 @@ def finish_sack_processing(self, sack):
         # Delete the sack key which handles no data but is used to get a SET
         # notification in iter_on_sacks_to_process
         self._client.delete(str(sack))
+
+    get_health_status = redis.get_redis_health_status

gnocchi/incoming/s3.py

@@ -42,21 +42,21 @@ def __init__(self, conf, greedy=True):
             s3.get_connection(conf)
         )
 
-        self._bucket_name_measures = (
+        self._bucket_name = (
             self._bucket_prefix + "-" + self.MEASURE_PREFIX
         )
 
     def __str__(self):
-        return "%s: %s" % (self.__class__.__name__, self._bucket_name_measures)
+        return "%s: %s" % (self.__class__.__name__, self._bucket_name)
 
     def _get_storage_sacks(self):
-        response = self.s3.get_object(Bucket=self._bucket_name_measures,
+        response = self.s3.get_object(Bucket=self._bucket_name,
                                       Key=self.CFG_PREFIX)
         return json.loads(response['Body'].read().decode())[self.CFG_SACKS]
 
     def set_storage_settings(self, num_sacks):
         data = {self.CFG_SACKS: num_sacks}
-        self.s3.put_object(Bucket=self._bucket_name_measures,
+        self.s3.put_object(Bucket=self._bucket_name,
                            Key=self.CFG_PREFIX,
                            Body=json.dumps(data).encode())
 
@@ -67,7 +67,7 @@ def remove_sacks(num_sacks):
 
     def upgrade(self, num_sacks):
         try:
-            s3.create_bucket(self.s3, self._bucket_name_measures,
+            s3.create_bucket(self.s3, self._bucket_name,
                              self._region_name)
         except botocore.exceptions.ClientError as e:
             if e.response['Error'].get('Code') not in (
@@ -80,7 +80,7 @@ def upgrade(self, num_sacks):
     def _store_new_measures(self, metric_id, data):
         now = datetime.datetime.utcnow().strftime("_%Y%m%d_%H:%M:%S")
         self.s3.put_object(
-            Bucket=self._bucket_name_measures,
+            Bucket=self._bucket_name,
             Key="/".join((str(self.sack_for_metric(metric_id)),
                           str(metric_id),
                           str(uuid.uuid4()) + now)),
@@ -97,7 +97,7 @@ def _build_report(self, details):
             else:
                 kwargs = {}
             response = self.s3.list_objects_v2(
-                Bucket=self._bucket_name_measures,
+                Bucket=self._bucket_name,
                 **kwargs)
             for c in response.get('Contents', ()):
                 if c['Key'] != self.CFG_PREFIX:
@@ -118,7 +118,7 @@ def _list_files(self, path_items, **kwargs):
                 except KeyError:
                     pass
             response = self.s3.list_objects_v2(
-                Bucket=self._bucket_name_measures,
+                Bucket=self._bucket_name,
                 Prefix="/".join(path_items) + "/",
                 **kwargs)
             yield response
@@ -136,7 +136,7 @@ def _list_measure_files_for_metric(self, sack, metric_id):
     def delete_unprocessed_measures_for_metric(self, metric_id):
         sack = self.sack_for_metric(metric_id)
         files = self._list_measure_files_for_metric(sack, metric_id)
-        s3.bulk_delete(self.s3, self._bucket_name_measures, files)
+        s3.bulk_delete(self.s3, self._bucket_name, files)
 
     def has_unprocessed(self, metric_id):
         sack = self.sack_for_metric(metric_id)
@@ -152,7 +152,7 @@ def process_measure_for_metrics(self, metric_ids):
             all_files.extend(files)
             for f in files:
                 response = self.s3.get_object(
-                    Bucket=self._bucket_name_measures,
+                    Bucket=self._bucket_name,
                     Key=f)
                 measures[metric_id] = numpy.concatenate((
                     measures[metric_id],
@@ -162,7 +162,7 @@ def process_measure_for_metrics(self, metric_ids):
         yield measures
 
         # Now clean objects
-        s3.bulk_delete(self.s3, self._bucket_name_measures, all_files)
+        s3.bulk_delete(self.s3, self._bucket_name, all_files)
 
     @contextlib.contextmanager
     def process_measures_for_sack(self, sack):
@@ -177,7 +177,7 @@ def process_measures_for_sack(self, sack):
                 continue
 
             response = self.s3.get_object(
-                Bucket=self._bucket_name_measures,
+                Bucket=self._bucket_name,
                 Key=f)
             measures[metric_id] = numpy.concatenate((
                 measures[metric_id],
@@ -187,4 +187,6 @@ def process_measures_for_sack(self, sack):
         yield measures
 
         # Now clean objects
-        s3.bulk_delete(self.s3, self._bucket_name_measures, files)
+        s3.bulk_delete(self.s3, self._bucket_name, files)
+
+    get_health_status = s3.get_s3_health_status

gnocchi/incoming/swift.py

@@ -144,3 +144,5 @@ def process_measures_for_sack(self, sack):
         yield measures
 
         swift.bulk_delete(self.swift, sack_name, files)
+
+    get_health_status = swift.get_swift_health_status

gnocchi/indexer/__init__.py

@@ -477,3 +477,7 @@ def get_resource_attributes_schemas():
     @staticmethod
     def get_resource_type_schema():
         raise exceptions.NotImplementedError
+
+    @staticmethod
+    def get_health_status():
+        raise exceptions.NotImplementedError

gnocchi/indexer/sqlalchemy.py

@@ -45,6 +45,7 @@
 from sqlalchemy import types as sa_types
 import sqlalchemy_utils
 
+from gnocchi.common.sqlalchemy import get_sqlalchemy_health_status
 from gnocchi import exceptions
 from gnocchi import indexer
 from gnocchi.indexer import sqlalchemy_base as base
@@ -1176,6 +1177,8 @@ def _build_sort_keys(sorts, unique_keys):
 
         return sort_keys, sort_dirs
 
+    get_health_status = get_sqlalchemy_health_status
+
 
 def _operator_in(field_name, value):
     # Do not generate empty IN comparison