From 961975996e4da732fdacd0aee814401829fa7bcb Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Fri, 8 Nov 2013 12:18:01 +0000
Subject: [PATCH 01/26] Make Ceph collector to pick up cluster name

Previously the prefix on admin socket names
was a config setting, had to be manually set
for one and only one cluster name.  In the
output stats, the cluster name was not used,
'ceph' was always passed through.

This change makes the collector gather stats
for all clusters it sees, and include the
name of the cluster in the metric names.
---
 src/collectors/ceph/ceph.py | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 4aea17e32..77f6e2c36 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -69,7 +69,6 @@ def get_default_config(self):
         config = super(CephCollector, self).get_default_config()
         config.update({
             'socket_path': '/var/run/ceph',
-            'socket_prefix': 'ceph-',
             'socket_ext': 'asok',
             'ceph_binary': '/usr/bin/ceph',
         })
@@ -80,18 +79,20 @@ def _get_socket_paths(self):
         with ceph daemons.
         """
         socket_pattern = os.path.join(self.config['socket_path'],
-                                      (self.config['socket_prefix'] +
-                                       '*.' + self.config['socket_ext']))
+                                      ('*.' + self.config['socket_ext']))
         return glob.glob(socket_pattern)
 
     def _get_counter_prefix_from_socket_name(self, name):
         """Given the name of a UDS socket, return the prefix
         for counters coming from that source.
+
+        This takes the form <cluster name>.<service type>.<service id>
+        for example, ceph.osd.2
+
         """
-        base = os.path.splitext(os.path.basename(name))[0]
-        if base.startswith(self.config['socket_prefix']):
-            base = base[len(self.config['socket_prefix']):]
-        return 'ceph.' + base
+        cluster_name, service_type, service_id = re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
+                                                          os.path.basename(name)).groups()
+        return "{0}.{1}.{2}".format(cluster_name, service_type, service_id)
 
     def _get_stats_from_socket(self, name):
         """Return the parsed JSON data returned when ceph is told to

From 545cfdf03aab5d2d3c0fef3a60a3ef7740420fd5 Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Fri, 8 Nov 2013 16:13:41 +0000
Subject: [PATCH 02/26] Extend Ceph collector to collect pool stats

These stats are only present in more recent versions
of Ceph (currently >=0.72, but expected to be backported).

Because these stats are for cluster-wide pools rather than
an individual host/service, they are reported with a global
name, ceph.<cluster name>.pool.<pool id>.*
---
 src/collectors/ceph/ceph.py | 146 ++++++++++++++++++++++++------------
 1 file changed, 99 insertions(+), 47 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 77f6e2c36..42722b257 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -20,6 +20,9 @@
 import glob
 import os
 import subprocess
+import re
+from distutils.version import StrictVersion
+
 
 import diamond.collector
 
@@ -46,15 +49,30 @@ def flatten_dictionary(input, sep='.', prefix=None):
             yield (fullname, value)
 
 
-class CephCollector(diamond.collector.Collector):
+class AdminSocketError(Exception):
+    def __init__(self, socket_name, command):
+        self.socket_name = socket_name
+        self.command = command
+
+    def __str__(self):
+        return "Admin socket error calling %s on socket %s" % (self.command, self.socket_name)
+
 
+class MonError(Exception):
+    def __init__(self, cluster_name, command):
+        self.cluster_name = cluster_name
+        self.command = command
+
+    def __str__(self):
+        return "Mon command error calling %s on cluster %s" % (self.command, self.cluster_name)
+
+
+class CephCollector(diamond.collector.Collector):
     def get_default_config_help(self):
         config_help = super(CephCollector, self).get_default_config_help()
         config_help.update({
             'socket_path': 'The location of the ceph monitoring sockets.'
                            ' Defaults to "/var/run/ceph"',
-            'socket_prefix': 'The first part of all socket names.'
-                             ' Defaults to "ceph-"',
             'socket_ext': 'Extension for socket filenames.'
                           ' Defaults to "asok"',
             'ceph_binary': 'Path to "ceph" executable. '
@@ -82,48 +100,13 @@ def _get_socket_paths(self):
                                       ('*.' + self.config['socket_ext']))
         return glob.glob(socket_pattern)
 
-    def _get_counter_prefix_from_socket_name(self, name):
-        """Given the name of a UDS socket, return the prefix
-        for counters coming from that source.
-
-        This takes the form <cluster name>.<service type>.<service id>
-        for example, ceph.osd.2
+    def _parse_socket_name(self, path):
+        """Parse a socket name like /var/run/ceph/foo-osd.2.asok
 
+        Return a 3 tuple of cluster name, service type, service id
         """
-        cluster_name, service_type, service_id = re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
-                                                          os.path.basename(name)).groups()
-        return "{0}.{1}.{2}".format(cluster_name, service_type, service_id)
-
-    def _get_stats_from_socket(self, name):
-        """Return the parsed JSON data returned when ceph is told to
-        dump the stats from the named socket.
-
-        In the event of an error error, the exception is logged, and
-        an empty result set is returned.
-        """
-        try:
-            json_blob = subprocess.check_output(
-                [self.config['ceph_binary'],
-                 '--admin-daemon',
-                 name,
-                 'perf',
-                 'dump',
-                 ])
-        except subprocess.CalledProcessError, err:
-            self.log.info('Could not get stats from %s: %s',
-                          name, err)
-            self.log.exception('Could not get stats from %s' % name)
-            return {}
-
-        try:
-            json_data = json.loads(json_blob)
-        except Exception, err:
-            self.log.info('Could not parse stats from %s: %s',
-                          name, err)
-            self.log.exception('Could not parse stats from %s' % name)
-            return {}
-
-        return json_data
+        return re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
+                        os.path.basename(path)).groups()
 
     def _publish_stats(self, counter_prefix, stats):
         """Given a stats dictionary from _get_stats_from_socket,
@@ -135,13 +118,82 @@ def _publish_stats(self, counter_prefix, stats):
         ):
             self.publish_gauge(stat_name, stat_value)
 
+    def _mon_command(self, cluster, command):
+        try:
+            json_blob = subprocess.check_output(
+                [self.config['ceph_binary'], '--cluster', cluster, '-f', 'json-pretty'] + command)
+        except subprocess.CalledProcessError:
+            raise MonError(cluster, command)
+
+        try:
+            return json.loads(json_blob)
+        except (ValueError, IndexError):
+            self.log.exception('Error parsing output from %s: %s' % (cluster, command))
+            raise MonError(cluster, command)
+
+    def _admin_command(self, socket_path, command):
+        try:
+            json_blob = subprocess.check_output(
+                [self.config['ceph_binary'], '--admin-daemon', socket_path] + command)
+        except subprocess.CalledProcessError:
+            self.log.exception('Error calling to %s' % socket_path)
+            raise AdminSocketError(socket_path, command)
+
+        try:
+            return json.loads(json_blob)
+        except (ValueError, IndexError):
+            self.log.exception('Error parsing output from %s' % socket_path)
+            raise AdminSocketError(socket_path, command)
+
+    def _collect_cluster_stats(self, path):
+        cluster_name, service_type, service_id = self._parse_socket_name(path)
+        if service_type != 'mon':
+            return
+
+        # Check if we are a high enough version to have pool throughput statistics
+        version_str = self._admin_command(path, ['version'])['version']
+        try:
+            version = StrictVersion(version_str)
+            # We expect to backport pool stats to the dumpling release series
+            # in the next release, and the current release at time of writing
+            # is 0.67.4.
+            if version < StrictVersion("0.67.5"):
+                return
+        except ValueError:
+            # If it doesn't parse, assume it's a git hash, therefore
+            # it should be recent and have the features we want.
+            pass
+
+        # We have a mon, see if it is the leader
+        mon_status = self._admin_command(path, ['mon_status'])
+        if mon_status['state'] != 'leader':
+            return
+
+        # We are the leader, gather cluster-wide statistics
+        for pool_data in self._mon_command(cluster_name, ['osd', 'pool', 'stats']):
+            pool_id = pool_data['pool_id']
+            del pool_data['pool_name']
+            del pool_data['pool_id']
+            self._publish_stats(
+                "{0}ceph.{1}.pool.{2}".format(diamond.collector.ABSOLUTE_PATH_MARKER, cluster_name, pool_id),
+                pool_data
+            )
+
+    def _collect_service_stats(self, path):
+        # The prefix is <cluster name>.<service type>.<service id>
+        counter_prefix = "{0}.{1}.{2}".format(*self._parse_socket_name(path))
+        stats = self._admin_command(path, ['perf', 'dump'])
+        self._publish_stats(counter_prefix, stats)
+
     def collect(self):
         """
         Collect stats
         """
         for path in self._get_socket_paths():
             self.log.debug('checking %s', path)
-            counter_prefix = self._get_counter_prefix_from_socket_name(path)
-            stats = self._get_stats_from_socket(path)
-            self._publish_stats(counter_prefix, stats)
-        return
+            # Publish statistics about this service
+            self._collect_service_stats(path)
+
+            # If this service is a mon and it is the leader of a quorum, then
+            # publish statistics about the cluster.
+            self._collect_cluster_stats(path)
\ No newline at end of file

From 023f6df5407d05e35e7ae52f5d625105315ceb40 Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Mon, 11 Nov 2013 18:16:44 +0000
Subject: [PATCH 03/26] CephCollector: Gather 'ceph df' statistics

---
 src/collectors/ceph/ceph.py | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 42722b257..d8ab987f6 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -169,6 +169,8 @@ def _collect_cluster_stats(self, path):
         if mon_status['state'] != 'leader':
             return
 
+        self.log.debug("mon leader found, gathering cluster stats for cluster '%s'" % cluster_name)
+
         # We are the leader, gather cluster-wide statistics
         for pool_data in self._mon_command(cluster_name, ['osd', 'pool', 'stats']):
             pool_id = pool_data['pool_id']
@@ -179,6 +181,14 @@ def _collect_cluster_stats(self, path):
                 pool_data
             )
 
+        df = self._mon_command(cluster_name, ['df'])
+        self._publish_stats("{0}ceph.{1}.df".format(diamond.collector.ABSOLUTE_PATH_MARKER, cluster_name), df['stats'])
+        for pool_data in df['pools']:
+            self._publish_stats(
+                "{0}ceph.{1}.pool.{2}".format(diamond.collector.ABSOLUTE_PATH_MARKER, cluster_name, pool_data['id']),
+                pool_data['stats']
+            )
+
     def _collect_service_stats(self, path):
         # The prefix is <cluster name>.<service type>.<service id>
         counter_prefix = "{0}.{1}.{2}".format(*self._parse_socket_name(path))
@@ -190,7 +200,7 @@ def collect(self):
         Collect stats
         """
         for path in self._get_socket_paths():
-            self.log.debug('checking %s', path)
+            self.log.debug('gathering service stats for %s', path)
             # Publish statistics about this service
             self._collect_service_stats(path)
 

From 59030a5859416065ea32f34692e87d6bc8e4ce93 Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Tue, 3 Dec 2013 23:32:42 +0000
Subject: [PATCH 04/26] Generate an 'all' pool which includes sum of pool
 statistics.

Optionally (off by default) report service
statistics under ceph.cluster isntead
of servers.<hostname>
---
 src/collectors/ceph/ceph.py | 195 ++++++++++++++++++++++++++----------
 1 file changed, 141 insertions(+), 54 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index d8ab987f6..85f82dd87 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -1,7 +1,7 @@
 # coding=utf-8
 
 """
-The CephCollector collects utilization info from the Ceph storage system.
+The CephCollector collects utilization info from Ceph services.
 
 Documentation for ceph perf counters:
 http://ceph.com/docs/master/dev/perf_counters/
@@ -12,22 +12,17 @@
 
 """
 
-try:
-    import json
-except ImportError:
-    import simplejson as json
-
+import json  # No need for simplejson fallback b/c ceph py modules are >=2.6
 import glob
 import os
 import subprocess
 import re
+from collections import defaultdict
 from distutils.version import StrictVersion
-
-
 import diamond.collector
 
 
-def flatten_dictionary(input, sep='.', prefix=None):
+def flatten_dictionary(input_dict, sep='.', prefix=None):
     """Produces iterator of pairs where the first value is
     the joined key names and the second value is the value
     associated with the lowest level key. For example::
@@ -40,7 +35,7 @@ def flatten_dictionary(input, sep='.', prefix=None):
 
       [('a.b', 10), ('c', 20)]
     """
-    for name, value in sorted(input.items()):
+    for name, value in sorted(input_dict.items()):
         fullname = sep.join(filter(None, [prefix, name]))
         if isinstance(value, dict):
             for result in flatten_dictionary(value, sep, fullname):
@@ -67,6 +62,10 @@ def __str__(self):
         return "Mon command error calling %s on cluster %s" % (self.command, self.cluster_name)
 
 
+class GlobalName(str):
+    pass
+
+
 class CephCollector(diamond.collector.Collector):
     def get_default_config_help(self):
         config_help = super(CephCollector, self).get_default_config_help()
@@ -77,6 +76,13 @@ def get_default_config_help(self):
                           ' Defaults to "asok"',
             'ceph_binary': 'Path to "ceph" executable. '
                            'Defaults to /usr/bin/ceph.',
+            'short_names': "If true, use cluster names instead of UUIDs"
+                           "in metric paths.  Defaults to true.",
+            'cluster_prefix': "Prefix for per-cluster metrics.  Defaults"
+                           "to 'ceph.cluster'.",
+            'service_stats_global': "If true, stats from osds and mons are"
+                                    "stored under the cluster prefix (not by host).  If false, these"
+                                    "stats are stored in per-host paths."
         })
         return config_help
 
@@ -89,9 +95,23 @@ def get_default_config(self):
             'socket_path': '/var/run/ceph',
             'socket_ext': 'asok',
             'ceph_binary': '/usr/bin/ceph',
+            'short_names': True,
+            'cluster_prefix': 'ceph.cluster',
+            'service_stats_global': False
         })
         return config
 
+    def get_metric_path(self, name, instance=None):
+        """
+        This collector returns some cluster-wide statistics rather than
+        server-specific statistics, so we override this to
+        avoid diamond prefixing the hostname to our metrics.
+        """
+        if isinstance(name, GlobalName):
+            return ".".join([self.config['cluster_prefix'], name])
+        else:
+            return super(CephCollector, self).get_metric_path(name, instance)
+
     def _get_socket_paths(self):
         """Return a sequence of paths to sockets for communicating
         with ceph daemons.
@@ -108,7 +128,7 @@ def _parse_socket_name(self, path):
         return re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
                         os.path.basename(path)).groups()
 
-    def _publish_stats(self, counter_prefix, stats):
+    def _publish_stats(self, counter_prefix, stats, global_name=False):
         """Given a stats dictionary from _get_stats_from_socket,
         publish the individual values.
         """
@@ -116,20 +136,21 @@ def _publish_stats(self, counter_prefix, stats):
             stats,
             prefix=counter_prefix,
         ):
-            self.publish_gauge(stat_name, stat_value)
+            self.publish_gauge(GlobalName(stat_name) if global_name else stat_name, stat_value)
 
-    def _mon_command(self, cluster, command):
-        try:
-            json_blob = subprocess.check_output(
-                [self.config['ceph_binary'], '--cluster', cluster, '-f', 'json-pretty'] + command)
-        except subprocess.CalledProcessError:
-            raise MonError(cluster, command)
+    def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats):
+        """
+        Given a stats dictionary, publish under the cluster path (respecting
+        short_names and cluster_prefix
+        """
+        # We'll either use the cluster name (human friendly but may not be unique)
+        # or the UUID (robust but obscure)
+        if self.config['short_names']:
+            cluster_id_prefix = cluster_name
+        else:
+            cluster_id_prefix = fsid
 
-        try:
-            return json.loads(json_blob)
-        except (ValueError, IndexError):
-            self.log.exception('Error parsing output from %s: %s' % (cluster, command))
-            raise MonError(cluster, command)
+        self._publish_stats("{0}.{1}".format(cluster_id_prefix, prefix), stats, global_name=True)
 
     def _admin_command(self, socket_path, command):
         try:
@@ -145,7 +166,24 @@ def _admin_command(self, socket_path, command):
             self.log.exception('Error parsing output from %s' % socket_path)
             raise AdminSocketError(socket_path, command)
 
+    def _mon_command(self, cluster, command):
+        try:
+            json_blob = subprocess.check_output(
+                [self.config['ceph_binary'], '--cluster', cluster, '-f', 'json-pretty'] + command)
+        except subprocess.CalledProcessError:
+            raise MonError(cluster, command)
+
+        try:
+            return json.loads(json_blob)
+        except (ValueError, IndexError):
+            self.log.exception('Error parsing output from %s: %s' % (cluster, command))
+            raise MonError(cluster, command)
+
     def _collect_cluster_stats(self, path):
+        """
+        If this service is a mon and it is the leader of a quorum, then
+        publish statistics about the cluster.
+        """
         cluster_name, service_type, service_id = self._parse_socket_name(path)
         if service_type != 'mon':
             return
@@ -154,46 +192,98 @@ def _collect_cluster_stats(self, path):
         version_str = self._admin_command(path, ['version'])['version']
         try:
             version = StrictVersion(version_str)
-            # We expect to backport pool stats to the dumpling release series
-            # in the next release, and the current release at time of writing
-            # is 0.67.4.
-            if version < StrictVersion("0.67.5"):
-                return
         except ValueError:
-            # If it doesn't parse, assume it's a git hash, therefore
-            # it should be recent and have the features we want.
-            pass
+            # In case it's a git hash or something like that
+            version = None
 
         # We have a mon, see if it is the leader
         mon_status = self._admin_command(path, ['mon_status'])
         if mon_status['state'] != 'leader':
             return
-
-        self.log.debug("mon leader found, gathering cluster stats for cluster '%s'" % cluster_name)
+        fsid = mon_status['monmap']['fsid']
 
         # We are the leader, gather cluster-wide statistics
-        for pool_data in self._mon_command(cluster_name, ['osd', 'pool', 'stats']):
-            pool_id = pool_data['pool_id']
-            del pool_data['pool_name']
-            del pool_data['pool_id']
-            self._publish_stats(
-                "{0}ceph.{1}.pool.{2}".format(diamond.collector.ABSOLUTE_PATH_MARKER, cluster_name, pool_id),
-                pool_data
-            )
+        self.log.debug("mon leader found, gathering cluster stats for cluster '%s'" % cluster_name)
 
+        # Recent Ceph versions have some per-pool statistics for us
+        if version is None or version >= StrictVersion("0.67.5"):
+            # Not everything in the pool stats makes sense to sum (e.g. ratios), so
+            # we have an explicit list of which items to put into the 'all' pool.
+            aggregates = {
+                'client_io_rate': {
+                    'op_per_sec': 0,
+                    'write_bytes_sec': 0,
+                    'read_bytes_sec': 0
+                },
+                'recovery': {
+                    'degraded_objects': 0,
+                    'degraded_total': 0,
+                },
+                'recovery_rate': {
+                    'recovering_objects_per_sec': 0,
+                    'recovering_keys_per_sec': 0,
+                    'recovering_bytes_per_sec': 0,
+                }
+            }
+            for pool_data in self._mon_command(cluster_name, ['osd', 'pool', 'stats']):
+                pool_id = pool_data['pool_id']
+                del pool_data['pool_name']
+                del pool_data['pool_id']
+
+                for k, v in aggregates.items():
+                    if k in pool_data:
+                        pool_data_k = pool_data[k]
+                        for k2, v2 in v.items():
+                            if k2 in pool_data_k:
+                                v[k2] += pool_data_k[k2]
+
+                self._publish_cluster_stats(cluster_name, fsid,
+                                            "pool.{0}".format(pool_id),
+                                            pool_data)
+            self._publish_cluster_stats(cluster_name, fsid,
+                                        "pool.all",
+                                        aggregates)
+
+        # Older Ceph versions only give us some global throughput stats
+        if version <= StrictVersion("0.67.4"):
+            summary = self._mon_command(cluster_name, ['pg', 'dump', 'summary'])
+            pg_stats_delta = summary['pg_stats_delta']['stat_sum']
+
+            # We will synthesize the 'client_io_rate.op_per_sec' statistic that
+            # would otherwise come from 'osd pool stats'
+            tick_period = 5.0  # We assume this has been left as the default
+            op_per_sec = (pg_stats_delta['num_write'] + pg_stats_delta['num_read']) / tick_period
+            self._publish_cluster_stats(cluster_name, fsid,
+                                        "pool.all",
+                                        {'client_io_rate': {"op_per_sec": op_per_sec}})
+
+        # Gather "ceph df" and file the stats by pool
         df = self._mon_command(cluster_name, ['df'])
-        self._publish_stats("{0}ceph.{1}.df".format(diamond.collector.ABSOLUTE_PATH_MARKER, cluster_name), df['stats'])
+        self._publish_cluster_stats(cluster_name, fsid, "df", df['stats'])
+        all_pools_df = defaultdict(int)
         for pool_data in df['pools']:
-            self._publish_stats(
-                "{0}ceph.{1}.pool.{2}".format(diamond.collector.ABSOLUTE_PATH_MARKER, cluster_name, pool_data['id']),
-                pool_data['stats']
-            )
+            self._publish_cluster_stats(cluster_name, fsid,
+                                        "pool.{0}".format(pool_data['id']),
+                                        pool_data['stats'])
+
+            for k, v in pool_data['stats'].items():
+                all_pools_df[k] += v
+        self._publish_cluster_stats(cluster_name, fsid,
+                                    "pool.all",
+                                    all_pools_df)
 
     def _collect_service_stats(self, path):
-        # The prefix is <cluster name>.<service type>.<service id>
-        counter_prefix = "{0}.{1}.{2}".format(*self._parse_socket_name(path))
+        cluster_name, service_type, service_id = self._parse_socket_name(path)
+        fsid = self._admin_command(path, ['config', 'get', 'fsid'])['fsid']
+
         stats = self._admin_command(path, ['perf', 'dump'])
-        self._publish_stats(counter_prefix, stats)
+        if self.config['service_stats_global']:
+            counter_prefix = "{0}.{1}".format(service_type, service_id)
+            self._publish_cluster_stats(cluster_name, fsid, counter_prefix, stats)
+        else:
+            # The prefix is <cluster name>.<service type>.<service id>
+            counter_prefix = "{0}.{1}.{2}".format(*self._parse_socket_name(path))
+            self._publish_stats(counter_prefix, stats)
 
     def collect(self):
         """
@@ -201,9 +291,6 @@ def collect(self):
         """
         for path in self._get_socket_paths():
             self.log.debug('gathering service stats for %s', path)
-            # Publish statistics about this service
-            self._collect_service_stats(path)
 
-            # If this service is a mon and it is the leader of a quorum, then
-            # publish statistics about the cluster.
-            self._collect_cluster_stats(path)
\ No newline at end of file
+            self._collect_service_stats(path)
+            self._collect_cluster_stats(path)

From a72319c1c63737faa1e92b7f73a885d7bbbfc10f Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Fri, 6 Dec 2013 18:30:58 +0000
Subject: [PATCH 05/26] Use counters instead of deltas for throughput stats

---
 src/collectors/ceph/ceph.py | 86 ++++++++++---------------------------
 1 file changed, 23 insertions(+), 63 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 85f82dd87..7f764df0c 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -128,7 +128,7 @@ def _parse_socket_name(self, path):
         return re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
                         os.path.basename(path)).groups()
 
-    def _publish_stats(self, counter_prefix, stats, global_name=False):
+    def _publish_stats(self, counter_prefix, stats, global_name=False, counter=False):
         """Given a stats dictionary from _get_stats_from_socket,
         publish the individual values.
         """
@@ -136,9 +136,13 @@ def _publish_stats(self, counter_prefix, stats, global_name=False):
             stats,
             prefix=counter_prefix,
         ):
-            self.publish_gauge(GlobalName(stat_name) if global_name else stat_name, stat_value)
+            name = GlobalName(stat_name) if global_name else stat_name
+            if counter:
+                self.publish_counter(name, stat_value)
+            else:
+                self.publish_gauge(name, stat_value)
 
-    def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats):
+    def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=False):
         """
         Given a stats dictionary, publish under the cluster path (respecting
         short_names and cluster_prefix
@@ -150,7 +154,7 @@ def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats):
         else:
             cluster_id_prefix = fsid
 
-        self._publish_stats("{0}.{1}".format(cluster_id_prefix, prefix), stats, global_name=True)
+        self._publish_stats("{0}.{1}".format(cluster_id_prefix, prefix), stats, global_name=True, counter=counter)
 
     def _admin_command(self, socket_path, command):
         try:
@@ -188,14 +192,6 @@ def _collect_cluster_stats(self, path):
         if service_type != 'mon':
             return
 
-        # Check if we are a high enough version to have pool throughput statistics
-        version_str = self._admin_command(path, ['version'])['version']
-        try:
-            version = StrictVersion(version_str)
-        except ValueError:
-            # In case it's a git hash or something like that
-            version = None
-
         # We have a mon, see if it is the leader
         mon_status = self._admin_command(path, ['mon_status'])
         if mon_status['state'] != 'leader':
@@ -205,57 +201,20 @@ def _collect_cluster_stats(self, path):
         # We are the leader, gather cluster-wide statistics
         self.log.debug("mon leader found, gathering cluster stats for cluster '%s'" % cluster_name)
 
-        # Recent Ceph versions have some per-pool statistics for us
-        if version is None or version >= StrictVersion("0.67.5"):
-            # Not everything in the pool stats makes sense to sum (e.g. ratios), so
-            # we have an explicit list of which items to put into the 'all' pool.
-            aggregates = {
-                'client_io_rate': {
-                    'op_per_sec': 0,
-                    'write_bytes_sec': 0,
-                    'read_bytes_sec': 0
-                },
-                'recovery': {
-                    'degraded_objects': 0,
-                    'degraded_total': 0,
-                },
-                'recovery_rate': {
-                    'recovering_objects_per_sec': 0,
-                    'recovering_keys_per_sec': 0,
-                    'recovering_bytes_per_sec': 0,
-                }
-            }
-            for pool_data in self._mon_command(cluster_name, ['osd', 'pool', 'stats']):
-                pool_id = pool_data['pool_id']
-                del pool_data['pool_name']
-                del pool_data['pool_id']
-
-                for k, v in aggregates.items():
-                    if k in pool_data:
-                        pool_data_k = pool_data[k]
-                        for k2, v2 in v.items():
-                            if k2 in pool_data_k:
-                                v[k2] += pool_data_k[k2]
-
-                self._publish_cluster_stats(cluster_name, fsid,
-                                            "pool.{0}".format(pool_id),
-                                            pool_data)
-            self._publish_cluster_stats(cluster_name, fsid,
-                                        "pool.all",
-                                        aggregates)
-
-        # Older Ceph versions only give us some global throughput stats
-        if version <= StrictVersion("0.67.4"):
-            summary = self._mon_command(cluster_name, ['pg', 'dump', 'summary'])
-            pg_stats_delta = summary['pg_stats_delta']['stat_sum']
-
-            # We will synthesize the 'client_io_rate.op_per_sec' statistic that
-            # would otherwise come from 'osd pool stats'
-            tick_period = 5.0  # We assume this has been left as the default
-            op_per_sec = (pg_stats_delta['num_write'] + pg_stats_delta['num_read']) / tick_period
-            self._publish_cluster_stats(cluster_name, fsid,
-                                        "pool.all",
-                                        {'client_io_rate': {"op_per_sec": op_per_sec}})
+        def publish_pool_stats(pool_id, stats):
+            # Some of these guys we treat as counters, some as gauges
+            delta_fields = ['num_read', 'num_read_kb', 'num_write', 'num_write_kb', 'num_objects_recovered',
+                            'num_bytes_recovered', 'num_keys_recovered']
+            for k, v in stats.items():
+                self._publish_cluster_stats(cluster_name, fsid, "pool.{0}".format(pool_id), {k: v},
+                                            counter=k in delta_fields)
+
+        # Gather "ceph pg dump pools" and file the stats by pool
+        for pool in self._mon_command(cluster_name, ['pg', 'dump', 'pools']):
+            publish_pool_stats(pool['poolid'], pool['stat_sum'])
+
+        all_pools_stats = self._mon_command(cluster_name, ['pg', 'dump', 'summary'])['pg_stats_sum']['stat_sum']
+        publish_pool_stats('all', all_pools_stats)
 
         # Gather "ceph df" and file the stats by pool
         df = self._mon_command(cluster_name, ['df'])
@@ -268,6 +227,7 @@ def _collect_cluster_stats(self, path):
 
             for k, v in pool_data['stats'].items():
                 all_pools_df[k] += v
+
         self._publish_cluster_stats(cluster_name, fsid,
                                     "pool.all",
                                     all_pools_df)

From 9c23cf2499d3a6e2ad59c613ec36b854f2efe4ff Mon Sep 17 00:00:00 2001
From: John Spray <jcspray@gmail.com>
Date: Thu, 12 Dec 2013 15:09:14 -0800
Subject: [PATCH 06/26] Fix boolean config option handling

---
 src/collectors/ceph/ceph.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 7f764df0c..4c5f5551f 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -20,6 +20,7 @@
 from collections import defaultdict
 from distutils.version import StrictVersion
 import diamond.collector
+from diamond.collector import str_to_bool
 
 
 def flatten_dictionary(input_dict, sep='.', prefix=None):
@@ -67,6 +68,11 @@ class GlobalName(str):
 
 
 class CephCollector(diamond.collector.Collector):
+    def __init__(self, config, handlers):
+        super(CephCollector, self).__init__(config, handlers)
+        self.config['short_names'] = str_to_bool(self.config['short_names'])
+        self.config['service_stats_global'] = str_to_bool(self.config['service_stats_global'])
+
     def get_default_config_help(self):
         config_help = super(CephCollector, self).get_default_config_help()
         config_help.update({
@@ -242,7 +248,7 @@ def _collect_service_stats(self, path):
             self._publish_cluster_stats(cluster_name, fsid, counter_prefix, stats)
         else:
             # The prefix is <cluster name>.<service type>.<service id>
-            counter_prefix = "{0}.{1}.{2}".format(*self._parse_socket_name(path))
+            counter_prefix = "{0}.{1}.{2}".format(cluster_name, service_type, service_id)
             self._publish_stats(counter_prefix, stats)
 
     def collect(self):

From 63bcde400ee0f9beddc9edb23641fbd20aa0540d Mon Sep 17 00:00:00 2001
From: John Spray <jcspray@gmail.com>
Date: Mon, 16 Dec 2013 09:52:50 -0800
Subject: [PATCH 07/26] BUG #7005 Remove redundant space stats

---
 src/collectors/ceph/ceph.py | 14 +-------------
 1 file changed, 1 insertion(+), 13 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 4c5f5551f..5de46dbdc 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -222,21 +222,9 @@ def publish_pool_stats(pool_id, stats):
         all_pools_stats = self._mon_command(cluster_name, ['pg', 'dump', 'summary'])['pg_stats_sum']['stat_sum']
         publish_pool_stats('all', all_pools_stats)
 
-        # Gather "ceph df" and file the stats by pool
+        # Gather "ceph df"
         df = self._mon_command(cluster_name, ['df'])
         self._publish_cluster_stats(cluster_name, fsid, "df", df['stats'])
-        all_pools_df = defaultdict(int)
-        for pool_data in df['pools']:
-            self._publish_cluster_stats(cluster_name, fsid,
-                                        "pool.{0}".format(pool_data['id']),
-                                        pool_data['stats'])
-
-            for k, v in pool_data['stats'].items():
-                all_pools_df[k] += v
-
-        self._publish_cluster_stats(cluster_name, fsid,
-                                    "pool.all",
-                                    all_pools_df)
 
     def _collect_service_stats(self, path):
         cluster_name, service_type, service_id = self._parse_socket_name(path)

From 8610f1e8403c68b2a5f8310e5256a6df8725d63f Mon Sep 17 00:00:00 2001
From: Noah Watkins <noahwatkins@gmail.com>
Date: Wed, 6 Nov 2013 13:24:45 -0800
Subject: [PATCH 08/26] ceph: fix ceph tests and temporarily disable mocks

This fixes some of the Ceph tests, and disables the tests that use
mocking.

Signed-off-by: Noah Watkins <noahwatkins@gmail.com>
---
 src/collectors/ceph/test/testceph.py | 208 +++++++++++++--------------
 1 file changed, 97 insertions(+), 111 deletions(-)

diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index 8095f7e17..dfcb166ef 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -40,35 +40,21 @@ def test_empty(self):
     @run_only_if_assertSequenceEqual_is_available
     def test_simple(self):
         data = {'a': 1, 'b': 2}
-        expected = [('a', 1), ('b', 2)]
+        expected = [(['a'], 1), (['b'], 2)]
         actual = list(ceph.flatten_dictionary(data))
         self.assertSequenceEqual(actual, expected)
 
-    @run_only_if_assertSequenceEqual_is_available
-    def test_prefix(self):
-        data = {'a': 1, 'b': 2}
-        expected = [('Z.a', 1), ('Z.b', 2)]
-        actual = list(ceph.flatten_dictionary(data, prefix='Z'))
-        self.assertSequenceEqual(actual, expected)
-
-    @run_only_if_assertSequenceEqual_is_available
-    def test_sep(self):
-        data = {'a': 1, 'b': 2}
-        expected = [('Z:a', 1), ('Z:b', 2)]
-        actual = list(ceph.flatten_dictionary(data, prefix='Z', sep=':'))
-        self.assertSequenceEqual(actual, expected)
-
     @run_only_if_assertSequenceEqual_is_available
     def test_nested(self):
         data = {'a': 1, 'b': 2, 'c': {'d': 3}}
-        expected = [('a', 1), ('b', 2), ('c.d', 3)]
+        expected = [(['a'], 1), (['b'], 2), (['c','d'], 3)]
         actual = list(ceph.flatten_dictionary(data))
         self.assertSequenceEqual(actual, expected)
 
     @run_only_if_assertSequenceEqual_is_available
     def test_doubly_nested(self):
         data = {'a': 1, 'b': 2, 'c': {'d': 3}, 'e': {'f': {'g': 1}}}
-        expected = [('a', 1), ('b', 2), ('c.d', 3), ('e.f.g', 1)]
+        expected = [(['a'], 1), (['b'], 2), (['c', 'd'], 3), (['e', 'f', 'g'], 1)]
         actual = list(ceph.flatten_dictionary(data))
         self.assertSequenceEqual(actual, expected)
 
@@ -81,11 +67,11 @@ def test_complex(self):
                          "sum": 0},
                 }
         expected = [
-            ('get', 60910),
-            ('max', 524288000),
-            ('val', 0),
-            ('wait.avgcount', 0),
-            ('wait.sum', 0),
+            (['get'], 60910),
+            (['max'], 524288000),
+            (['val'], 0),
+            (['wait', 'avgcount'], 0),
+            (['wait', 'sum'], 0),
         ]
         actual = list(ceph.flatten_dictionary(data))
         self.assertSequenceEqual(actual, expected)
@@ -100,13 +86,13 @@ def setUp(self):
         self.collector = ceph.CephCollector(config, None)
 
     def test_counter_default_prefix(self):
-        expected = 'ceph.osd.325'
+        expected = 'ceph.osd-325'
         sock = '/var/run/ceph/ceph-osd.325.asok'
         actual = self.collector._get_counter_prefix_from_socket_name(sock)
         self.assertEquals(actual, expected)
 
     def test_counter_alternate_prefix(self):
-        expected = 'ceph.keep-osd.325'
+        expected = 'ceph.keep-osd-325'
         sock = '/var/run/ceph/keep-osd.325.asok'
         actual = self.collector._get_counter_prefix_from_socket_name(sock)
         self.assertEquals(actual, expected)
@@ -124,93 +110,93 @@ def test_get_socket_paths(self, glob_mock):
         glob_mock.assert_called_with('/path/prefix-*.ext')
 
 
-class TestCephCollectorGettingStats(CollectorTestCase):
-
-    def setUp(self):
-        config = get_collector_config('CephCollector', {
-            'interval': 10,
-        })
-        self.collector = ceph.CephCollector(config, None)
-
-    def test_import(self):
-        self.assertTrue(ceph.CephCollector)
-
-    @run_only_if_subprocess_check_output_is_available
-    @patch('subprocess.check_output')
-    def test_load_works(self, check_output):
-        expected = {'a': 1,
-                    'b': 2,
-                    }
-        check_output.return_value = json.dumps(expected)
-        actual = self.collector._get_stats_from_socket('a_socket_name')
-        check_output.assert_called_with(['/usr/bin/ceph',
-                                         '--admin-daemon',
-                                         'a_socket_name',
-                                         'perf',
-                                         'dump',
-                                         ])
-        self.assertEqual(actual, expected)
-
-    @run_only_if_subprocess_check_output_is_available
-    @patch('subprocess.check_output')
-    def test_ceph_command_fails(self, check_output):
-        check_output.side_effect = subprocess.CalledProcessError(
-            255, ['/usr/bin/ceph'], 'error!',
-        )
-        actual = self.collector._get_stats_from_socket('a_socket_name')
-        check_output.assert_called_with(['/usr/bin/ceph',
-                                         '--admin-daemon',
-                                         'a_socket_name',
-                                         'perf',
-                                         'dump',
-                                         ])
-        self.assertEqual(actual, {})
-
-    @run_only_if_subprocess_check_output_is_available
-    @patch('json.loads')
-    @patch('subprocess.check_output')
-    def test_json_decode_fails(self, check_output, loads):
-        input = {'a': 1,
-                 'b': 2,
-                 }
-        check_output.return_value = json.dumps(input)
-        loads.side_effect = ValueError('bad data')
-        actual = self.collector._get_stats_from_socket('a_socket_name')
-        check_output.assert_called_with(['/usr/bin/ceph',
-                                         '--admin-daemon',
-                                         'a_socket_name',
-                                         'perf',
-                                         'dump',
-                                         ])
-        loads.assert_called_with(json.dumps(input))
-        self.assertEqual(actual, {})
-
-
-class TestCephCollectorPublish(CollectorTestCase):
-
-    def setUp(self):
-        config = get_collector_config('CephCollector', {
-            'interval': 10,
-        })
-        self.collector = ceph.CephCollector(config, None)
-
-    @patch.object(Collector, 'publish')
-    def test_simple(self, publish_mock):
-        self.collector._publish_stats('prefix', {'a': 1})
-        publish_mock.assert_called_with('prefix.a', 1,
-                                        metric_type='GAUGE', instance=None,
-                                        precision=0)
-
-    @patch.object(Collector, 'publish')
-    def test_multiple(self, publish_mock):
-        self.collector._publish_stats('prefix', {'a': 1, 'b': 2})
-        publish_mock.assert_has_calls([call('prefix.a', 1,
-                                            metric_type='GAUGE', instance=None,
-                                            precision=0),
-                                       call('prefix.b', 2,
-                                            metric_type='GAUGE', instance=None,
-                                            precision=0),
-                                       ])
+#class TestCephCollectorGettingStats(CollectorTestCase):
+#
+#    def setUp(self):
+#        config = get_collector_config('CephCollector', {
+#            'interval': 10,
+#        })
+#        self.collector = ceph.CephCollector(config, None)
+#
+#    def test_import(self):
+#        self.assertTrue(ceph.CephCollector)
+#
+#    @run_only_if_subprocess_check_output_is_available
+#    @patch('subprocess.check_output')
+#    def test_load_works(self, check_output):
+#        expected = {'a': 1,
+#                    'b': 2,
+#                    }
+#        check_output.return_value = json.dumps(expected)
+#        actual = self.collector._get_stats_from_socket('a_socket_name')
+#        check_output.assert_called_with(['/usr/bin/ceph',
+#                                         '--admin-daemon',
+#                                         'a_socket_name',
+#                                         'perf',
+#                                         'dump',
+#                                         ])
+#        self.assertEqual(actual, expected)
+#
+#    @run_only_if_subprocess_check_output_is_available
+#    @patch('subprocess.check_output')
+#    def test_ceph_command_fails(self, check_output):
+#        check_output.side_effect = subprocess.CalledProcessError(
+#            255, ['/usr/bin/ceph'], 'error!',
+#        )
+#        actual = self.collector._get_stats_from_socket('a_socket_name')
+#        check_output.assert_called_with(['/usr/bin/ceph',
+#                                         '--admin-daemon',
+#                                         'a_socket_name',
+#                                         'perf',
+#                                         'dump',
+#                                         ])
+#        self.assertEqual(actual, {})
+#
+#    @run_only_if_subprocess_check_output_is_available
+#    @patch('json.loads')
+#    @patch('subprocess.check_output')
+#    def test_json_decode_fails(self, check_output, loads):
+#        input = {'a': 1,
+#                 'b': 2,
+#                 }
+#        check_output.return_value = json.dumps(input)
+#        loads.side_effect = ValueError('bad data')
+#        actual = self.collector._get_stats_from_socket('a_socket_name')
+#        check_output.assert_called_with(['/usr/bin/ceph',
+#                                         '--admin-daemon',
+#                                         'a_socket_name',
+#                                         'perf',
+#                                         'dump',
+#                                         ])
+#        loads.assert_called_with(json.dumps(input))
+#        self.assertEqual(actual, {})
+#
+#
+#class TestCephCollectorPublish(CollectorTestCase):
+#
+#    def setUp(self):
+#        config = get_collector_config('CephCollector', {
+#            'interval': 10,
+#        })
+#        self.collector = ceph.CephCollector(config, None)
+#
+#    @patch.object(Collector, 'publish')
+#    def test_simple(self, publish_mock):
+#        self.collector._publish_stats('prefix', {'a': 1})
+#        publish_mock.assert_called_with('prefix.a', 1,
+#                                        metric_type='GAUGE', instance=None,
+#                                        precision=0)
+#
+#    @patch.object(Collector, 'publish')
+#    def test_multiple(self, publish_mock):
+#        self.collector._publish_stats('prefix', {'a': 1, 'b': 2})
+#        publish_mock.assert_has_calls([call('prefix.a', 1,
+#                                            metric_type='GAUGE', instance=None,
+#                                            precision=0),
+#                                       call('prefix.b', 2,
+#                                            metric_type='GAUGE', instance=None,
+#                                            precision=0),
+#                                       ])
 
 if __name__ == "__main__":
     unittest.main()

From 0b66026dc954f57bb364c16c01c3f40eb252c343 Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Mon, 30 Dec 2013 16:45:50 +0000
Subject: [PATCH 09/26] Reconcile branches wip-ceph vs cluster-stats

- Removing use of 2.7-only subprocess.check_call
- Using 'perf schema' for perf counter output
---
 src/collectors/ceph/ceph.py          | 263 +++++++++++++++++++++++----
 src/collectors/ceph/test/testceph.py |  88 ++++-----
 2 files changed, 273 insertions(+), 78 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 5de46dbdc..10e8cffd5 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -17,16 +17,27 @@
 import os
 import subprocess
 import re
-from collections import defaultdict
-from distutils.version import StrictVersion
 import diamond.collector
+import diamond.convertor
 from diamond.collector import str_to_bool
 
 
-def flatten_dictionary(input_dict, sep='.', prefix=None):
-    """Produces iterator of pairs where the first value is
-    the joined key names and the second value is the value
-    associated with the lowest level key. For example::
+# Metric name/path separator
+_PATH_SEP = "."
+
+_NSEC_PER_SEC = 1000000000
+
+# Performance metric data types
+_PERFCOUNTER_NONE = 0
+_PERFCOUNTER_TIME = 0x1
+_PERFCOUNTER_U64 = 0x2
+_PERFCOUNTER_LONGRUNAVG = 0x4
+_PERFCOUNTER_COUNTER = 0x8
+
+
+def flatten_dictionary(input_dict, path=list()):
+    """Produces iterator of pairs where the first value is the key path and
+    the second value is the value associated with the key. For example::
 
       {'a': {'b': 10},
        'c': 20,
@@ -34,15 +45,49 @@ def flatten_dictionary(input_dict, sep='.', prefix=None):
 
     produces::
 
-      [('a.b', 10), ('c', 20)]
+      [([a,b], 10), ([c], 20)]
     """
     for name, value in sorted(input_dict.items()):
-        fullname = sep.join(filter(None, [prefix, name]))
+        path.append(name)
         if isinstance(value, dict):
-            for result in flatten_dictionary(value, sep, fullname):
+            for result in flatten_dictionary(value, path):
                 yield result
         else:
-            yield (fullname, value)
+            yield (path[:], value)
+        del path[-1]
+
+
+def lookup_dict_path(d, path, extra=list()):
+    """Lookup value in dictionary based on path + extra.
+
+    For instance, [a,b,c] -> d[a][b][c]
+    """
+    element = None
+    for component in path + extra:
+        d = d[component]
+        element = d
+    return element
+
+
+class CalledProcessError(Exception):
+    pass
+
+
+def _popen_check_output(*popenargs):
+    """
+    Collect Popen output and check for errors.
+
+    This is inspired by subprocess.check_output, added in Python 2.7. This
+    method provides similar functionality but will work with Python 2.6.
+    """
+    process = subprocess.Popen(*popenargs, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    output, err = process.communicate()
+    retcode = process.poll()
+    if retcode:
+        msg = "Command '%s' exited with non-zero status %d" % \
+              (popenargs[0], retcode)
+        raise CalledProcessError(msg)
+    return output, err
 
 
 class AdminSocketError(Exception):
@@ -72,6 +117,7 @@ def __init__(self, config, handlers):
         super(CephCollector, self).__init__(config, handlers)
         self.config['short_names'] = str_to_bool(self.config['short_names'])
         self.config['service_stats_global'] = str_to_bool(self.config['service_stats_global'])
+        self.config['perf_counters_enabled'] = str_to_bool(self.config['perf_counters_enabled'])
 
     def get_default_config_help(self):
         config_help = super(CephCollector, self).get_default_config_help()
@@ -103,7 +149,8 @@ def get_default_config(self):
             'ceph_binary': '/usr/bin/ceph',
             'short_names': True,
             'cluster_prefix': 'ceph.cluster',
-            'service_stats_global': False
+            'service_stats_global': False,
+            'perf_counters_enabled': True
         })
         return config
 
@@ -134,37 +181,167 @@ def _parse_socket_name(self, path):
         return re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
                         os.path.basename(path)).groups()
 
-    def _publish_stats(self, counter_prefix, stats, global_name=False, counter=False):
-        """Given a stats dictionary from _get_stats_from_socket,
-        publish the individual values.
+    def _publish_longrunavg(self, counter_prefix, stats, path, stat_type):
+        """Publish a long-running average metric.
+
+        A long-running metric has two components: 'avgcount' and 'sum'. We
+        publish both the raw components, and a derived metric named
+        <metric>.last_interval_avg that is the average since the last run of
+        the collector.
+
+        For a given long-running average metric with name <metric>, we publish
+        the following derived metrics:
+
+            <metric>.sum
+            <metric>.count
+            <metric>.last_interval_avg
+
+        Args:
+            counter_prefix: string prefixed to metric names
+            stats: dictionary containing performance counters
+            path: full path of the metric name (e.g. [osd, op_rw_rlat])
+            stat_type: the metric type taken from the schema
         """
-        for stat_name, stat_value in flatten_dictionary(
-            stats,
-            prefix=counter_prefix,
-        ):
-            name = GlobalName(stat_name) if global_name else stat_name
-            if counter:
-                self.publish_counter(name, stat_value)
-            else:
-                self.publish_gauge(name, stat_value)
+        # name of <metric>
+        base_name = _PATH_SEP.join(filter(None, [counter_prefix] + path))
+        total_sum_name = "%s%s%s" % (base_name, _PATH_SEP, "sum")
+        total_count_name = "%s%s%s" % (base_name, _PATH_SEP, "count")
+        delta_sum_name = "%s%s%s" % (base_name, _PATH_SEP, "delta_sum")
+        delta_count_name = "%s%s%s" % (base_name, _PATH_SEP, "delta_count")
+        delta_avg_name = "%s%s%s" % (base_name, _PATH_SEP, "last_interval_avg")
+
+        # lookup raw metric component values
+        total_sum = lookup_dict_path(stats, path, ['sum'])
+        total_count = lookup_dict_path(stats, path, ['avgcount'])
+
+        # perform metric-specific type conversions
+        if stat_type & _PERFCOUNTER_TIME:
+            total_sum = self._ceph_time_to_seconds(total_sum)
+
+        # Calculate deltas since last time we queried admin socket. The
+        # derivitive function records from the last invocation the
+        # total_sum/total_count, and simply returns the difference.
+        delta_sum = self.derivative(delta_sum_name, total_sum, time_delta=False)
+        delta_count = self.derivative(delta_count_name, total_count, time_delta=False)
+
+        # average in the last collection interval
+        if delta_count == 0:
+            delta_avg = 0
+        else:
+            delta_avg = float(delta_sum) / float(delta_count)
 
-    def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=False):
+        # publish raw data
+        self.publish_gauge(total_sum_name, total_sum)
+        self.publish_gauge(total_count_name, total_count)
+
+        # publish averages
+        self.publish_gauge(delta_avg_name, delta_avg, 6)
+
+    def _ceph_time_to_seconds(self, val):
+        """Convert Ceph time format into seconds.  Older Ceph
+           versions output times as a string, while newer
+           versions output a float (which we pass through)
+
+        :param val: string in format "seconds.nanoseconds" or
+                    floating point number.
+
+        Returns:
+            Time in seconds as a floating point number.
         """
-        Given a stats dictionary, publish under the cluster path (respecting
-        short_names and cluster_prefix
+        if isinstance(val, basestring):
+            sec, nsec = map(lambda v: long(v), val.split("."))
+            return float(sec * _NSEC_PER_SEC + nsec) / float(_NSEC_PER_SEC)
+        else:
+            return val
+
+    def _get_byte_metrics(self, name, metric_value):
+        """Return list of metrics derived from byte units.
+
+        Args:
+            name: the name of the metric
+            metric_value: the value of the metric in bytes
+
+        Returns:
+            List of (name, value) pairs for each unit.
         """
+        assert name.endswith("bytes")
+        result = []
+        for unit in self.config['byte_unit']:
+            new_value = diamond.convertor.binary.convert(
+                value=metric_value, oldUnit='byte', newUnit=unit)
+            new_name = name.replace("bytes", unit)
+            result.append((new_name, new_value))
+        return result
+
+    def _publish_stats(self, counter_prefix, stats, schema, global_name=False):
+        """Publish a set of Ceph performance counters, including schema.
+
+        :param counter_prefix: string prefixed to metric names
+        :param stats: dictionary containing performance counters
+        :param schema: performance counter schema
+        """
+        for path, stat_type in flatten_dictionary(schema):
+            # remove 'stat_type' component to get metric name
+            assert path[-1] == 'type'
+            del path[-1]
+
+            if stat_type & _PERFCOUNTER_LONGRUNAVG:
+                self._publish_longrunavg(counter_prefix, stats, path, stat_type)
+            else:
+                name = _PATH_SEP.join(filter(None, [counter_prefix] + path))
+                if global_name:
+                    name = GlobalName(name)
+
+                value = lookup_dict_path(stats, path)
+
+                if stat_type & _PERFCOUNTER_TIME:
+                    value = self._ceph_time_to_seconds(value)
+                    self.publish_gauge(name, value, 6)
+
+                elif stat_type & _PERFCOUNTER_U64:
+                    # create a list of values to log. we'll either log a list
+                    # of derived metrics, or the single metric we began with.
+                    if name.endswith("bytes"):
+                        values = self._get_byte_metrics(name, value)
+                    else:
+                        values = [(name, value)]
+
+                    for name, value in values:
+                        if stat_type & _PERFCOUNTER_COUNTER:
+                            self.publish_counter(name, value, 2)
+                        else:
+                            self.publish_gauge(name, value, 2)
+                else:
+                    self.log.error("Unexpected metric stat_type: %s/%d", name, stat_type)
+
+    def _cluster_id_prefix(self, cluster_name, fsid):
         # We'll either use the cluster name (human friendly but may not be unique)
         # or the UUID (robust but obscure)
         if self.config['short_names']:
-            cluster_id_prefix = cluster_name
+            return cluster_name
         else:
-            cluster_id_prefix = fsid
+            return fsid
+
+    def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=False):
+        """
+        Given a stats dictionary, publish under the cluster path (respecting
+        short_names and cluster_prefix)
+        """
 
-        self._publish_stats("{0}.{1}".format(cluster_id_prefix, prefix), stats, global_name=True, counter=counter)
+
+        for stat_name, stat_value in flatten_dictionary(
+            stats,
+            path=[self._cluster_id_prefix(cluster_name, fsid), prefix]
+        ):
+            name = GlobalName(stat_name)
+            if counter:
+                self.publish_counter(name, stat_value)
+            else:
+                self.publish_gauge(name, stat_value)
 
     def _admin_command(self, socket_path, command):
         try:
-            json_blob = subprocess.check_output(
+            json_blob, err = _popen_check_output(
                 [self.config['ceph_binary'], '--admin-daemon', socket_path] + command)
         except subprocess.CalledProcessError:
             self.log.exception('Error calling to %s' % socket_path)
@@ -178,7 +355,7 @@ def _admin_command(self, socket_path, command):
 
     def _mon_command(self, cluster, command):
         try:
-            json_blob = subprocess.check_output(
+            json_blob, err = _popen_check_output(
                 [self.config['ceph_binary'], '--cluster', cluster, '-f', 'json-pretty'] + command)
         except subprocess.CalledProcessError:
             raise MonError(cluster, command)
@@ -226,18 +403,34 @@ def publish_pool_stats(pool_id, stats):
         df = self._mon_command(cluster_name, ['df'])
         self._publish_cluster_stats(cluster_name, fsid, "df", df['stats'])
 
+    def _get_perf_counters(self, name):
+        """Return perf counters and schema from admin socket.
+
+        Args:
+            name: path to admin socket
+
+        Returns:
+            Tuple (counters, schema)
+        """
+        counters = self._admin_command(name, ['perf', 'dump'])
+        schema = self._admin_command(name, ['perf', 'schema'])
+        return counters, schema
+
     def _collect_service_stats(self, path):
+        if not self.config['perf_counters_enabled']:
+            return
+
         cluster_name, service_type, service_id = self._parse_socket_name(path)
         fsid = self._admin_command(path, ['config', 'get', 'fsid'])['fsid']
 
-        stats = self._admin_command(path, ['perf', 'dump'])
+        stats, schema = self._get_perf_counters(path)
         if self.config['service_stats_global']:
-            counter_prefix = "{0}.{1}".format(service_type, service_id)
-            self._publish_cluster_stats(cluster_name, fsid, counter_prefix, stats)
+            counter_prefix = "{0}.{1}.{2}".format(self._cluster_id_prefix(cluster_name, fsid), service_type, service_id)
+            self._publish_stats(cluster_name, fsid, counter_prefix, stats, global_name=True)
         else:
             # The prefix is <cluster name>.<service type>.<service id>
             counter_prefix = "{0}.{1}.{2}".format(cluster_name, service_type, service_id)
-            self._publish_stats(counter_prefix, stats)
+            self._publish_stats(counter_prefix, stats, schema)
 
     def collect(self):
         """
diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index dfcb166ef..6a590c60c 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -23,11 +23,6 @@ def run_only_if_assertSequenceEqual_is_available(func):
     return run_only(func, pred)
 
 
-def run_only_if_subprocess_check_output_is_available(func):
-    pred = lambda: 'check_output' in dir(subprocess)
-    return run_only(func, pred)
-
-
 class TestCounterIterator(unittest.TestCase):
 
     @run_only_if_assertSequenceEqual_is_available
@@ -85,57 +80,64 @@ def setUp(self):
         })
         self.collector = ceph.CephCollector(config, None)
 
-    def test_counter_default_prefix(self):
-        expected = 'ceph.osd-325'
-        sock = '/var/run/ceph/ceph-osd.325.asok'
-        actual = self.collector._get_counter_prefix_from_socket_name(sock)
-        self.assertEquals(actual, expected)
-
-    def test_counter_alternate_prefix(self):
-        expected = 'ceph.keep-osd-325'
-        sock = '/var/run/ceph/keep-osd.325.asok'
-        actual = self.collector._get_counter_prefix_from_socket_name(sock)
+    def test_parse_socket_name(self):
+        expected = ('cephadoodle', 'osd', '325')
+        sock = '/var/run/ceph/cephadoodle-osd.325.asok'
+        actual = self.collector._parse_socket_name(sock)
         self.assertEquals(actual, expected)
 
     @patch('glob.glob')
     def test_get_socket_paths(self, glob_mock):
         config = get_collector_config('CephCollector', {
             'socket_path': '/path/',
-            'socket_prefix': 'prefix-',
             'socket_ext': 'ext',
         })
         collector = ceph.CephCollector(config, None)
 
         collector._get_socket_paths()
-        glob_mock.assert_called_with('/path/prefix-*.ext')
+        glob_mock.assert_called_with('/path/*.ext')
 
 
-#class TestCephCollectorGettingStats(CollectorTestCase):
-#
-#    def setUp(self):
-#        config = get_collector_config('CephCollector', {
-#            'interval': 10,
-#        })
-#        self.collector = ceph.CephCollector(config, None)
-#
-#    def test_import(self):
-#        self.assertTrue(ceph.CephCollector)
-#
-#    @run_only_if_subprocess_check_output_is_available
-#    @patch('subprocess.check_output')
-#    def test_load_works(self, check_output):
-#        expected = {'a': 1,
-#                    'b': 2,
-#                    }
-#        check_output.return_value = json.dumps(expected)
-#        actual = self.collector._get_stats_from_socket('a_socket_name')
-#        check_output.assert_called_with(['/usr/bin/ceph',
-#                                         '--admin-daemon',
-#                                         'a_socket_name',
-#                                         'perf',
-#                                         'dump',
-#                                         ])
-#        self.assertEqual(actual, expected)
+class TestCephCollectorGettingStats(CollectorTestCase):
+
+    def setUp(self):
+        config = get_collector_config('CephCollector', {
+            'interval': 10,
+        })
+        self.collector = ceph.CephCollector(config, None)
+
+    def test_import(self):
+        self.assertTrue(ceph.CephCollector)
+
+    @patch('ceph._popen_check_output')
+    def test_load_works(self, check_output):
+        expected = {'a': 1,
+                    'b': 2,
+        }
+        check_output.return_value = (json.dumps(expected), "")
+        actual_stats, actual_schema = self.collector._get_perf_counters('a_socket_name')
+        self.assertListEqual(check_output.mock_calls,
+                             [
+                                 call(
+                                     [
+                                         '/usr/bin/ceph',
+                                         '--admin-daemon',
+                                         'a_socket_name',
+                                         'perf',
+                                         'dump',
+                                     ]
+                                 ),
+                                 call(
+                                     [
+                                         '/usr/bin/ceph',
+                                         '--admin-daemon',
+                                         'a_socket_name',
+                                         'perf',
+                                         'schema',
+                                     ]
+                                 ),
+                             ])
+        self.assertEqual(actual_stats, expected)
 #
 #    @run_only_if_subprocess_check_output_is_available
 #    @patch('subprocess.check_output')

From 9fe17436c96aea1d42b22f43dad117b9dfa61d9b Mon Sep 17 00:00:00 2001
From: Dan Mick <dan.mick@inktank.com>
Date: Fri, 3 Jan 2014 21:41:39 -0800
Subject: [PATCH 10/26] ceph.py: join path returned from flatten_dictionary

I don't think the filter(None,...) idiom is necessary here?...

Signed-off-by: Dan Mick <dan.mick@inktank.com>
---
 src/collectors/ceph/ceph.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 10e8cffd5..8398d2a6e 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -333,6 +333,7 @@ def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=Fals
             stats,
             path=[self._cluster_id_prefix(cluster_name, fsid), prefix]
         ):
+            stat_name = _PATH_SEP.join(stat_name)
             name = GlobalName(stat_name)
             if counter:
                 self.publish_counter(name, stat_value)

From 94dc1a901b0564c46764f6e666f364b3b277acc0 Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Sun, 19 Jan 2014 14:01:06 +0000
Subject: [PATCH 11/26] Fix Calamari issue #7137 with some better error
 handling

The goal here is to have failures to talk to one service
not prevent us gathering stats from other services.  The
symptom was that when there was one stale ceph socket file
all ceph stats stopped being collected.
---
 src/collectors/ceph/ceph.py | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 8398d2a6e..732349355 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -344,8 +344,7 @@ def _admin_command(self, socket_path, command):
         try:
             json_blob, err = _popen_check_output(
                 [self.config['ceph_binary'], '--admin-daemon', socket_path] + command)
-        except subprocess.CalledProcessError:
-            self.log.exception('Error calling to %s' % socket_path)
+        except CalledProcessError:
             raise AdminSocketError(socket_path, command)
 
         try:
@@ -358,7 +357,7 @@ def _mon_command(self, cluster, command):
         try:
             json_blob, err = _popen_check_output(
                 [self.config['ceph_binary'], '--cluster', cluster, '-f', 'json-pretty'] + command)
-        except subprocess.CalledProcessError:
+        except CalledProcessError:
             raise MonError(cluster, command)
 
         try:
@@ -423,8 +422,8 @@ def _collect_service_stats(self, path):
 
         cluster_name, service_type, service_id = self._parse_socket_name(path)
         fsid = self._admin_command(path, ['config', 'get', 'fsid'])['fsid']
-
         stats, schema = self._get_perf_counters(path)
+
         if self.config['service_stats_global']:
             counter_prefix = "{0}.{1}.{2}".format(self._cluster_id_prefix(cluster_name, fsid), service_type, service_id)
             self._publish_stats(cluster_name, fsid, counter_prefix, stats, global_name=True)
@@ -440,5 +439,8 @@ def collect(self):
         for path in self._get_socket_paths():
             self.log.debug('gathering service stats for %s', path)
 
-            self._collect_service_stats(path)
-            self._collect_cluster_stats(path)
+            try:
+                self._collect_service_stats(path)
+                self._collect_cluster_stats(path)
+            except (AdminSocketError, MonError) as e:
+                self.log.warn(e.__str__())

From b6a7949f480a35525ba4e2525b17d6db2ac90e58 Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Sun, 19 Jan 2014 14:02:20 +0000
Subject: [PATCH 12/26] Fix Calamari issue #7138, bad default arg

---
 src/collectors/ceph/ceph.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 732349355..4478a0598 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -35,7 +35,7 @@
 _PERFCOUNTER_COUNTER = 0x8
 
 
-def flatten_dictionary(input_dict, path=list()):
+def flatten_dictionary(input_dict, path=None):
     """Produces iterator of pairs where the first value is the key path and
     the second value is the value associated with the key. For example::
 
@@ -47,6 +47,9 @@ def flatten_dictionary(input_dict, path=list()):
 
       [([a,b], 10), ([c], 20)]
     """
+    if path is None:
+        path = []
+
     for name, value in sorted(input_dict.items()):
         path.append(name)
         if isinstance(value, dict):

From c7a9e566f73f01749587ec22ab991b48d785b7cd Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Thu, 13 Mar 2014 16:45:21 +0000
Subject: [PATCH 13/26] CephCollector: add options for disabling stats

For larger systems, the stat collection can be
overly verbose and cause excessive load on low
powered graphite servers.  We introduce 'osd_stats_enabled'
and 'long_running_detail' to provide a flexible way
to tone down the verbosity.
---
 src/collectors/ceph/ceph.py | 21 ++++++++++++++++++---
 1 file changed, 18 insertions(+), 3 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 4478a0598..c48718913 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -121,6 +121,8 @@ def __init__(self, config, handlers):
         self.config['short_names'] = str_to_bool(self.config['short_names'])
         self.config['service_stats_global'] = str_to_bool(self.config['service_stats_global'])
         self.config['perf_counters_enabled'] = str_to_bool(self.config['perf_counters_enabled'])
+        self.config['osd_stats_enabled'] = str_to_bool(self.config['osd_stats_enabled'])
+        self.config['long_running_detail'] = str_to_bool(self.config['long_running_detail'])
 
     def get_default_config_help(self):
         config_help = super(CephCollector, self).get_default_config_help()
@@ -135,9 +137,15 @@ def get_default_config_help(self):
                            "in metric paths.  Defaults to true.",
             'cluster_prefix': "Prefix for per-cluster metrics.  Defaults"
                            "to 'ceph.cluster'.",
+            'osd_stats_enabled': "Whether to enable OSD service stats.  These are the most numerous and"
+                           "may overload underpowered graphite instances when there are 100s of OSDs. "
+                           "Defaults to true",
             'service_stats_global': "If true, stats from osds and mons are"
                                     "stored under the cluster prefix (not by host).  If false, these"
-                                    "stats are stored in per-host paths."
+                                    "stats are stored in per-host paths.",
+            'long_running_detail': "Whether to break down long running averages into sum/count/average (true), or"
+                                   "only output the average from the last measurement interval (false).  Defaults"
+                                   "to false."
         })
         return config_help
 
@@ -153,6 +161,8 @@ def get_default_config(self):
             'short_names': True,
             'cluster_prefix': 'ceph.cluster',
             'service_stats_global': False,
+            'osd_stats_enabled': True,
+            'long_running_detail': False,
             'perf_counters_enabled': True
         })
         return config
@@ -234,8 +244,9 @@ def _publish_longrunavg(self, counter_prefix, stats, path, stat_type):
             delta_avg = float(delta_sum) / float(delta_count)
 
         # publish raw data
-        self.publish_gauge(total_sum_name, total_sum)
-        self.publish_gauge(total_count_name, total_count)
+        if self.config['long_running_detail']:
+            self.publish_gauge(total_sum_name, total_sum)
+            self.publish_gauge(total_count_name, total_count)
 
         # publish averages
         self.publish_gauge(delta_avg_name, delta_avg, 6)
@@ -424,6 +435,10 @@ def _collect_service_stats(self, path):
             return
 
         cluster_name, service_type, service_id = self._parse_socket_name(path)
+
+        if service_type == 'osd' and not self.config['osd_stats_enabled']:
+            return
+
         fsid = self._admin_command(path, ['config', 'get', 'fsid'])['fsid']
         stats, schema = self._get_perf_counters(path)
 

From 58f2b63b5f7ddbe34cb5ff03e1cd4d95bb0c507b Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Thu, 3 Apr 2014 14:06:47 +0100
Subject: [PATCH 14/26] CephCollector: fix service_stats_global mode

Signed-off-by: John Spray <john.spray@inktank.com>
---
 src/collectors/ceph/ceph.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index c48718913..a662de43f 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -444,7 +444,7 @@ def _collect_service_stats(self, path):
 
         if self.config['service_stats_global']:
             counter_prefix = "{0}.{1}.{2}".format(self._cluster_id_prefix(cluster_name, fsid), service_type, service_id)
-            self._publish_stats(cluster_name, fsid, counter_prefix, stats, global_name=True)
+            self._publish_stats(counter_prefix, stats, schema, global_name=True)
         else:
             # The prefix is <cluster name>.<service type>.<service id>
             counter_prefix = "{0}.{1}.{2}".format(cluster_name, service_type, service_id)

From 5ed28f27bb690950ce8526fba9ad080996c7f16f Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Thu, 3 Apr 2014 14:08:08 +0100
Subject: [PATCH 15/26] CephCollector: remove stray whitespace

---
 src/collectors/ceph/ceph.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index a662de43f..3cce71e56 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -342,7 +342,6 @@ def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=Fals
         short_names and cluster_prefix)
         """
 
-
         for stat_name, stat_value in flatten_dictionary(
             stats,
             path=[self._cluster_id_prefix(cluster_name, fsid), prefix]

From 1bc37037554ff60e772a9918a0fa19669a645bdb Mon Sep 17 00:00:00 2001
From: John Spray <john.spray@inktank.com>
Date: Fri, 4 Apr 2014 17:37:44 +0100
Subject: [PATCH 16/26] CephCollector: fix OSD stats (service_stats_global)

Longrunning avg and _bytes statistics were losing
the GlobalName type and getting emitted as stats within
a server's path.  Introduce LocalName and require explicit
local/global type for all stats names to make issues like
this more obvious.
---
 src/collectors/ceph/ceph.py | 34 +++++++++++++++++++---------------
 1 file changed, 19 insertions(+), 15 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 3cce71e56..b1e38f875 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -114,6 +114,9 @@ def __str__(self):
 class GlobalName(str):
     pass
 
+class LocalName(str):
+    pass
+
 
 class CephCollector(diamond.collector.Collector):
     def __init__(self, config, handlers):
@@ -175,8 +178,11 @@ def get_metric_path(self, name, instance=None):
         """
         if isinstance(name, GlobalName):
             return ".".join([self.config['cluster_prefix'], name])
-        else:
+        elif isinstance(name, LocalName):
             return super(CephCollector, self).get_metric_path(name, instance)
+        else:
+            # Require explicit local or global indication to catch bugs more easily
+            raise RuntimeError("Name '{0}' not LocalName or GlobalName".format(name))
 
     def _get_socket_paths(self):
         """Return a sequence of paths to sockets for communicating
@@ -194,7 +200,7 @@ def _parse_socket_name(self, path):
         return re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
                         os.path.basename(path)).groups()
 
-    def _publish_longrunavg(self, counter_prefix, stats, path, stat_type):
+    def _publish_longrunavg(self, counter_prefix, stats, path, stat_type, name_class):
         """Publish a long-running average metric.
 
         A long-running metric has two components: 'avgcount' and 'sum'. We
@@ -217,11 +223,11 @@ def _publish_longrunavg(self, counter_prefix, stats, path, stat_type):
         """
         # name of <metric>
         base_name = _PATH_SEP.join(filter(None, [counter_prefix] + path))
-        total_sum_name = "%s%s%s" % (base_name, _PATH_SEP, "sum")
-        total_count_name = "%s%s%s" % (base_name, _PATH_SEP, "count")
-        delta_sum_name = "%s%s%s" % (base_name, _PATH_SEP, "delta_sum")
-        delta_count_name = "%s%s%s" % (base_name, _PATH_SEP, "delta_count")
-        delta_avg_name = "%s%s%s" % (base_name, _PATH_SEP, "last_interval_avg")
+        total_sum_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "sum"))
+        total_count_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "count"))
+        delta_sum_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "delta_sum"))
+        delta_count_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "delta_count"))
+        delta_avg_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "last_interval_avg"))
 
         # lookup raw metric component values
         total_sum = lookup_dict_path(stats, path, ['sum'])
@@ -283,11 +289,11 @@ def _get_byte_metrics(self, name, metric_value):
         for unit in self.config['byte_unit']:
             new_value = diamond.convertor.binary.convert(
                 value=metric_value, oldUnit='byte', newUnit=unit)
-            new_name = name.replace("bytes", unit)
+            new_name = name.__class__(name.replace("bytes", unit))
             result.append((new_name, new_value))
         return result
 
-    def _publish_stats(self, counter_prefix, stats, schema, global_name=False):
+    def _publish_stats(self, counter_prefix, stats, schema, name_class):
         """Publish a set of Ceph performance counters, including schema.
 
         :param counter_prefix: string prefixed to metric names
@@ -300,11 +306,9 @@ def _publish_stats(self, counter_prefix, stats, schema, global_name=False):
             del path[-1]
 
             if stat_type & _PERFCOUNTER_LONGRUNAVG:
-                self._publish_longrunavg(counter_prefix, stats, path, stat_type)
+                self._publish_longrunavg(counter_prefix, stats, path, stat_type, name_class)
             else:
-                name = _PATH_SEP.join(filter(None, [counter_prefix] + path))
-                if global_name:
-                    name = GlobalName(name)
+                name = name_class(_PATH_SEP.join(filter(None, [counter_prefix] + path)))
 
                 value = lookup_dict_path(stats, path)
 
@@ -443,11 +447,11 @@ def _collect_service_stats(self, path):
 
         if self.config['service_stats_global']:
             counter_prefix = "{0}.{1}.{2}".format(self._cluster_id_prefix(cluster_name, fsid), service_type, service_id)
-            self._publish_stats(counter_prefix, stats, schema, global_name=True)
+            self._publish_stats(counter_prefix, stats, schema, GlobalName)
         else:
             # The prefix is <cluster name>.<service type>.<service id>
             counter_prefix = "{0}.{1}.{2}".format(cluster_name, service_type, service_id)
-            self._publish_stats(counter_prefix, stats, schema)
+            self._publish_stats(counter_prefix, stats, schema, LocalName)
 
     def collect(self):
         """

From 948d07d2144396b52a141bcb5e556f4b21d3b0ac Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Wed, 28 Oct 2015 16:31:22 +0000
Subject: [PATCH 17/26] cephcollector: fix signature of __init__

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/ceph.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index b1e38f875..c1e73ad63 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -119,8 +119,8 @@ class LocalName(str):
 
 
 class CephCollector(diamond.collector.Collector):
-    def __init__(self, config, handlers):
-        super(CephCollector, self).__init__(config, handlers)
+    def __init__(self, config=None, handlers=[], name=None, configfile=None):
+        super(CephCollector, self).__init__(config, handlers, name, configfile)
         self.config['short_names'] = str_to_bool(self.config['short_names'])
         self.config['service_stats_global'] = str_to_bool(self.config['service_stats_global'])
         self.config['perf_counters_enabled'] = str_to_bool(self.config['perf_counters_enabled'])

From 74e41f4d4c21e5d0c31c4ce4cc75dcf4223901cb Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Wed, 28 Oct 2015 16:32:30 +0000
Subject: [PATCH 18/26] cephcollector tests: fix tests to work with new ceph
 internal interfaces

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/test/testceph.py | 125 ++++++++++++++-------------
 1 file changed, 64 insertions(+), 61 deletions(-)

diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index 6a590c60c..89d2363e9 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -138,67 +138,70 @@ def test_load_works(self, check_output):
                                  ),
                              ])
         self.assertEqual(actual_stats, expected)
-#
-#    @run_only_if_subprocess_check_output_is_available
-#    @patch('subprocess.check_output')
-#    def test_ceph_command_fails(self, check_output):
-#        check_output.side_effect = subprocess.CalledProcessError(
-#            255, ['/usr/bin/ceph'], 'error!',
-#        )
-#        actual = self.collector._get_stats_from_socket('a_socket_name')
-#        check_output.assert_called_with(['/usr/bin/ceph',
-#                                         '--admin-daemon',
-#                                         'a_socket_name',
-#                                         'perf',
-#                                         'dump',
-#                                         ])
-#        self.assertEqual(actual, {})
-#
-#    @run_only_if_subprocess_check_output_is_available
-#    @patch('json.loads')
-#    @patch('subprocess.check_output')
-#    def test_json_decode_fails(self, check_output, loads):
-#        input = {'a': 1,
-#                 'b': 2,
-#                 }
-#        check_output.return_value = json.dumps(input)
-#        loads.side_effect = ValueError('bad data')
-#        actual = self.collector._get_stats_from_socket('a_socket_name')
-#        check_output.assert_called_with(['/usr/bin/ceph',
-#                                         '--admin-daemon',
-#                                         'a_socket_name',
-#                                         'perf',
-#                                         'dump',
-#                                         ])
-#        loads.assert_called_with(json.dumps(input))
-#        self.assertEqual(actual, {})
-#
-#
-#class TestCephCollectorPublish(CollectorTestCase):
-#
-#    def setUp(self):
-#        config = get_collector_config('CephCollector', {
-#            'interval': 10,
-#        })
-#        self.collector = ceph.CephCollector(config, None)
-#
-#    @patch.object(Collector, 'publish')
-#    def test_simple(self, publish_mock):
-#        self.collector._publish_stats('prefix', {'a': 1})
-#        publish_mock.assert_called_with('prefix.a', 1,
-#                                        metric_type='GAUGE', instance=None,
-#                                        precision=0)
-#
-#    @patch.object(Collector, 'publish')
-#    def test_multiple(self, publish_mock):
-#        self.collector._publish_stats('prefix', {'a': 1, 'b': 2})
-#        publish_mock.assert_has_calls([call('prefix.a', 1,
-#                                            metric_type='GAUGE', instance=None,
-#                                            precision=0),
-#                                       call('prefix.b', 2,
-#                                            metric_type='GAUGE', instance=None,
-#                                            precision=0),
-#                                       ])
+
+    @patch('ceph._popen_check_output')
+    def test_ceph_command_fails(self, check_output):
+        # this test check very little since the exceptionhandling ws moved into collect
+        # We've checked elsewhere that check_output is getting called with specific params
+        # TODO delete or sub in collect for _get_perf_counters
+        check_output.side_effect = subprocess.CalledProcessError(
+            255, ['/usr/bin/ceph'], 'error!',
+        )
+        with self.assertRaises(ceph.AdminSocketError):
+            actual_stats, actual_schema = self.collector._get_perf_counters('a_socket_name')
+
+        check_output.assert_called_with(['/usr/bin/ceph',
+                                         '--admin-daemon',
+                                         'a_socket_name',
+                                         'perf',
+                                         'dump',
+                                         ])
+
+    @patch('json.loads')
+    @patch('ceph._popen_check_output')
+    def test_json_decode_fails(self, check_output, loads):
+        input = {'a': 1,
+                 'b': 2,
+                 }
+        check_output.return_value = (json.dumps(input), '')
+        loads.side_effect = ValueError('bad data')
+        with self.assertRaises(ceph.AdminSocketError):
+            actual_stats, actual_schema = self.collector._get_perf_counters('a_socket_name')
+
+        check_output.assert_called_with(['/usr/bin/ceph',
+                                         '--admin-daemon',
+                                         'a_socket_name',
+                                         'perf',
+                                         'dump',
+                                         ])
+        loads.assert_called_with(json.dumps(input))
+
+
+class TestCephCollectorPublish(CollectorTestCase):
+
+    def setUp(self):
+        config = get_collector_config('CephCollector', {
+            'interval': 10,
+        })
+        self.collector = ceph.CephCollector(config, None)
+
+    @patch.object(Collector, 'publish')
+    def test_simple(self, publish_mock):
+        self.collector._publish_stats('prefix', {'a': 1})
+        publish_mock.assert_called_with('prefix.a', 1,
+                                        metric_type='GAUGE', instance=None,
+                                        precision=0)
+
+    @patch.object(Collector, 'publish')
+    def test_multiple(self, publish_mock):
+        self.collector._publish_stats('prefix', {'a': 1, 'b': 2})
+        publish_mock.assert_has_calls([call('prefix.a', 1,
+                                            metric_type='GAUGE', instance=None,
+                                            precision=0),
+                                       call('prefix.b', 2,
+                                            metric_type='GAUGE', instance=None,
+                                            precision=0),
+                                       ])
 
 if __name__ == "__main__":
     unittest.main()

From 747ad4e55324dde4c8c043c5bd19c1d400a8af6f Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Fri, 30 Oct 2015 18:22:10 +0000
Subject: [PATCH 19/26] tests: wrapping near 80 lines

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/test/testceph.py | 20 +++++++++++++-------
 1 file changed, 13 insertions(+), 7 deletions(-)

diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index 89d2363e9..1a0f09aa2 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -48,8 +48,14 @@ def test_nested(self):
 
     @run_only_if_assertSequenceEqual_is_available
     def test_doubly_nested(self):
-        data = {'a': 1, 'b': 2, 'c': {'d': 3}, 'e': {'f': {'g': 1}}}
-        expected = [(['a'], 1), (['b'], 2), (['c', 'd'], 3), (['e', 'f', 'g'], 1)]
+        data = {'a': 1,
+                'b': 2,
+                'c': {'d': 3},
+                'e': {'f': {'g': 1}}}
+        expected = [(['a'], 1),
+                    (['b'], 2),
+                    (['c', 'd'], 3),
+                    (['e', 'f', 'g'], 1)]
         actual = list(ceph.flatten_dictionary(data))
         self.assertSequenceEqual(actual, expected)
 
@@ -113,9 +119,9 @@ def test_import(self):
     def test_load_works(self, check_output):
         expected = {'a': 1,
                     'b': 2,
-        }
+                    }
         check_output.return_value = (json.dumps(expected), "")
-        actual_stats, actual_schema = self.collector._get_perf_counters('a_socket_name')
+        stats, schema = self.collector._get_perf_counters('a_socket_name')
         self.assertListEqual(check_output.mock_calls,
                              [
                                  call(
@@ -137,7 +143,7 @@ def test_load_works(self, check_output):
                                      ]
                                  ),
                              ])
-        self.assertEqual(actual_stats, expected)
+        self.assertEqual(stats, expected)
 
     @patch('ceph._popen_check_output')
     def test_ceph_command_fails(self, check_output):
@@ -148,7 +154,7 @@ def test_ceph_command_fails(self, check_output):
             255, ['/usr/bin/ceph'], 'error!',
         )
         with self.assertRaises(ceph.AdminSocketError):
-            actual_stats, actual_schema = self.collector._get_perf_counters('a_socket_name')
+            stats, schema = self.collector._get_perf_counters('a_socket_name')
 
         check_output.assert_called_with(['/usr/bin/ceph',
                                          '--admin-daemon',
@@ -166,7 +172,7 @@ def test_json_decode_fails(self, check_output, loads):
         check_output.return_value = (json.dumps(input), '')
         loads.side_effect = ValueError('bad data')
         with self.assertRaises(ceph.AdminSocketError):
-            actual_stats, actual_schema = self.collector._get_perf_counters('a_socket_name')
+            stats, schema = self.collector._get_perf_counters('a_socket_name')
 
         check_output.assert_called_with(['/usr/bin/ceph',
                                          '--admin-daemon',

From ed7d35a1b0b55a89b5919b85ef59883d7b8f988f Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Mon, 2 Nov 2015 22:39:14 +0000
Subject: [PATCH 20/26] ceph collector: deal with schema that has more than
 type

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/ceph.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index c1e73ad63..e43cd9f84 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -302,7 +302,8 @@ def _publish_stats(self, counter_prefix, stats, schema, name_class):
         """
         for path, stat_type in flatten_dictionary(schema):
             # remove 'stat_type' component to get metric name
-            assert path[-1] == 'type'
+            if path[-1] != 'type':
+                continue
             del path[-1]
 
             if stat_type & _PERFCOUNTER_LONGRUNAVG:

From 3911783ad0bc26516b9e2006a66691bfb0dcca35 Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Mon, 2 Nov 2015 22:40:18 +0000
Subject: [PATCH 21/26] tests: fix ceph collector publish tests

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/test/testceph.py | 40 +++++++++++++++++++++++-----
 1 file changed, 34 insertions(+), 6 deletions(-)

diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index 1a0f09aa2..56c1f8214 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -193,20 +193,48 @@ def setUp(self):
 
     @patch.object(Collector, 'publish')
     def test_simple(self, publish_mock):
-        self.collector._publish_stats('prefix', {'a': 1})
-        publish_mock.assert_called_with('prefix.a', 1,
+        schema = {u'cluster': {u'a': {u'description': u'a version',
+                                      u'nick': u'',
+                                      u'type': 2}}}
+
+        self.collector._publish_stats('prefix',
+                                      {'cluster': {'a': 1}},
+                                      schema,
+                                      ceph.GlobalName)
+        publish_mock.assert_called_with('prefix.cluster.a', 1,
                                         metric_type='GAUGE', instance=None,
-                                        precision=0)
+                                        precision=2)
 
     @patch.object(Collector, 'publish')
     def test_multiple(self, publish_mock):
-        self.collector._publish_stats('prefix', {'a': 1, 'b': 2})
+        schema = {u'a': {u'description': u'a version',
+                         u'nick': u'',
+                         u'type': 2},
+                  u'b': {u'description': u'a version',
+                         u'nick': u'',
+                         u'type': 2}}
+
+        self.collector._publish_stats('prefix', {'a': 1, 'b': 2},
+                                      schema, ceph.GlobalName)
         publish_mock.assert_has_calls([call('prefix.a', 1,
                                             metric_type='GAUGE', instance=None,
-                                            precision=0),
+                                            precision=2),
                                        call('prefix.b', 2,
                                             metric_type='GAUGE', instance=None,
-                                            precision=0),
+                                            precision=2),
+                                       ])
+
+    @patch.object(Collector, 'publish')
+    def test_multiple_obeys_schema(self, publish_mock):
+        schema = {u'a': {u'description': u'a version',
+                         u'nick': u'',
+                         u'type': 2}}
+
+        self.collector._publish_stats('prefix', {'a': 1, 'b': 2},
+                                      schema, ceph.GlobalName)
+        publish_mock.assert_has_calls([call('prefix.a', 1,
+                                            metric_type='GAUGE', instance=None,
+                                            precision=2),
                                        ])
 
 if __name__ == "__main__":

From b7c938cda26fa5c3551d3c2551e5414ece47c575 Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Tue, 3 Nov 2015 19:10:27 +0000
Subject: [PATCH 22/26] ceph collector: wrap long lines

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/ceph.py | 153 +++++++++++++++++++++++++-----------
 1 file changed, 105 insertions(+), 48 deletions(-)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index e43cd9f84..3017c6e97 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -83,7 +83,9 @@ def _popen_check_output(*popenargs):
     This is inspired by subprocess.check_output, added in Python 2.7. This
     method provides similar functionality but will work with Python 2.6.
     """
-    process = subprocess.Popen(*popenargs, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    process = subprocess.Popen(*popenargs,
+                               stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE)
     output, err = process.communicate()
     retcode = process.poll()
     if retcode:
@@ -99,7 +101,8 @@ def __init__(self, socket_name, command):
         self.command = command
 
     def __str__(self):
-        return "Admin socket error calling %s on socket %s" % (self.command, self.socket_name)
+        message = "Admin socket error calling %s on socket %s"
+        return message % (self.command, self.socket_name)
 
 
 class MonError(Exception):
@@ -108,12 +111,14 @@ def __init__(self, cluster_name, command):
         self.command = command
 
     def __str__(self):
-        return "Mon command error calling %s on cluster %s" % (self.command, self.cluster_name)
+        message = "Mon command error calling %s on cluster %s"
+        return message % (self.command, self.cluster_name)
 
 
 class GlobalName(str):
     pass
 
+
 class LocalName(str):
     pass
 
@@ -121,11 +126,12 @@ class LocalName(str):
 class CephCollector(diamond.collector.Collector):
     def __init__(self, config=None, handlers=[], name=None, configfile=None):
         super(CephCollector, self).__init__(config, handlers, name, configfile)
-        self.config['short_names'] = str_to_bool(self.config['short_names'])
-        self.config['service_stats_global'] = str_to_bool(self.config['service_stats_global'])
-        self.config['perf_counters_enabled'] = str_to_bool(self.config['perf_counters_enabled'])
-        self.config['osd_stats_enabled'] = str_to_bool(self.config['osd_stats_enabled'])
-        self.config['long_running_detail'] = str_to_bool(self.config['long_running_detail'])
+        for key in ('short_names',
+                    'service_stats_global',
+                    'perf_counters_enabled',
+                    'osd_stats_enabled',
+                    'long_running_detail'):
+            self.config[key] = str_to_bool(self.config[key])
 
     def get_default_config_help(self):
         config_help = super(CephCollector, self).get_default_config_help()
@@ -140,14 +146,18 @@ def get_default_config_help(self):
                            "in metric paths.  Defaults to true.",
             'cluster_prefix': "Prefix for per-cluster metrics.  Defaults"
                            "to 'ceph.cluster'.",
-            'osd_stats_enabled': "Whether to enable OSD service stats.  These are the most numerous and"
-                           "may overload underpowered graphite instances when there are 100s of OSDs. "
-                           "Defaults to true",
+            'osd_stats_enabled': "Whether to enable OSD service stats.  These"
+                           "are the most numerous and may overload"
+                           "underpowered graphite instances when there are "
+                           " 100s of OSDs. Defaults to true",
             'service_stats_global': "If true, stats from osds and mons are"
-                                    "stored under the cluster prefix (not by host).  If false, these"
+                                    "stored under the cluster prefix (not by"
+                                    "host).  If false, these"
                                     "stats are stored in per-host paths.",
-            'long_running_detail': "Whether to break down long running averages into sum/count/average (true), or"
-                                   "only output the average from the last measurement interval (false).  Defaults"
+            'long_running_detail': "Whether to break down long running"
+                                   "averages into sum/count/average (true), or"
+                                   "only output the average from the last"
+                                   "measurement interval (false).  Defaults"
                                    "to false."
         })
         return config_help
@@ -181,8 +191,9 @@ def get_metric_path(self, name, instance=None):
         elif isinstance(name, LocalName):
             return super(CephCollector, self).get_metric_path(name, instance)
         else:
-            # Require explicit local or global indication to catch bugs more easily
-            raise RuntimeError("Name '{0}' not LocalName or GlobalName".format(name))
+            # explicit local or global indication to catch bugs more easily
+            message = "Name '{0}' not LocalName or GlobalName".format(name)
+            raise RuntimeError(message)
 
     def _get_socket_paths(self):
         """Return a sequence of paths to sockets for communicating
@@ -197,10 +208,15 @@ def _parse_socket_name(self, path):
 
         Return a 3 tuple of cluster name, service type, service id
         """
-        return re.match("^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext']),
-                        os.path.basename(path)).groups()
-
-    def _publish_longrunavg(self, counter_prefix, stats, path, stat_type, name_class):
+        pattern = "^(.*)-(.*)\.(.*).{0}$".format(self.config['socket_ext'])
+        return re.match(pattern, os.path.basename(path)).groups()
+
+    def _publish_longrunavg(self,
+                            counter_prefix,
+                            stats,
+                            path,
+                            stat_type,
+                            name_class):
         """Publish a long-running average metric.
 
         A long-running metric has two components: 'avgcount' and 'sum'. We
@@ -222,12 +238,16 @@ def _publish_longrunavg(self, counter_prefix, stats, path, stat_type, name_class
             stat_type: the metric type taken from the schema
         """
         # name of <metric>
-        base_name = _PATH_SEP.join(filter(None, [counter_prefix] + path))
-        total_sum_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "sum"))
-        total_count_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "count"))
-        delta_sum_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "delta_sum"))
-        delta_count_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "delta_count"))
-        delta_avg_name = name_class("%s%s%s" % (base_name, _PATH_SEP, "last_interval_avg"))
+        def _make_stat_name(stat_name):
+            parts = [counter_prefix] + path + [stat_name]
+            clean_parts = filter(None, parts)
+            return name_class(_PATH_SEP.join(clean_parts))
+
+        total_sum_name = _make_stat_name("sum")
+        total_count_name = _make_stat_name("count")
+        delta_sum_name = _make_stat_name("delta_sum")
+        delta_count_name = _make_stat_name("delta_count")
+        delta_avg_name = _make_stat_name("last_interval_avg")
 
         # lookup raw metric component values
         total_sum = lookup_dict_path(stats, path, ['sum'])
@@ -240,8 +260,10 @@ def _publish_longrunavg(self, counter_prefix, stats, path, stat_type, name_class
         # Calculate deltas since last time we queried admin socket. The
         # derivitive function records from the last invocation the
         # total_sum/total_count, and simply returns the difference.
-        delta_sum = self.derivative(delta_sum_name, total_sum, time_delta=False)
-        delta_count = self.derivative(delta_count_name, total_count, time_delta=False)
+        delta_sum = self.derivative(delta_sum_name, total_sum,
+                                    time_delta=False)
+        delta_count = self.derivative(delta_count_name, total_count,
+                                      time_delta=False)
 
         # average in the last collection interval
         if delta_count == 0:
@@ -293,10 +315,10 @@ def _get_byte_metrics(self, name, metric_value):
             result.append((new_name, new_value))
         return result
 
-    def _publish_stats(self, counter_prefix, stats, schema, name_class):
+    def _publish_stats(self, prefix, stats, schema, name_class):
         """Publish a set of Ceph performance counters, including schema.
 
-        :param counter_prefix: string prefixed to metric names
+        :param prefix: string prefixed to metric names
         :param stats: dictionary containing performance counters
         :param schema: performance counter schema
         """
@@ -307,9 +329,13 @@ def _publish_stats(self, counter_prefix, stats, schema, name_class):
             del path[-1]
 
             if stat_type & _PERFCOUNTER_LONGRUNAVG:
-                self._publish_longrunavg(counter_prefix, stats, path, stat_type, name_class)
+                self._publish_longrunavg(prefix,
+                                         stats,
+                                         path,
+                                         stat_type,
+                                         name_class)
             else:
-                name = name_class(_PATH_SEP.join(filter(None, [counter_prefix] + path)))
+                name = name_class(_PATH_SEP.join(filter(None, [prefix] + path)))
 
                 value = lookup_dict_path(stats, path)
 
@@ -331,17 +357,23 @@ def _publish_stats(self, counter_prefix, stats, schema, name_class):
                         else:
                             self.publish_gauge(name, value, 2)
                 else:
-                    self.log.error("Unexpected metric stat_type: %s/%d", name, stat_type)
+                    message = "Unexpected metric stat_type: %s/%d"
+                    self.log.error(message, name, stat_type)
 
     def _cluster_id_prefix(self, cluster_name, fsid):
-        # We'll either use the cluster name (human friendly but may not be unique)
-        # or the UUID (robust but obscure)
+        # We'll either use the cluster name (human friendly but may not be
+        # unique) or the UUID (robust but obscure)
         if self.config['short_names']:
             return cluster_name
         else:
             return fsid
 
-    def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=False):
+    def _publish_cluster_stats(self,
+                               cluster_name,
+                               fsid,
+                               prefix,
+                               stats,
+                               counter=False):
         """
         Given a stats dictionary, publish under the cluster path (respecting
         short_names and cluster_prefix)
@@ -360,9 +392,10 @@ def _publish_cluster_stats(self, cluster_name, fsid, prefix, stats, counter=Fals
 
     def _admin_command(self, socket_path, command):
         try:
-            json_blob, err = _popen_check_output(
-                [self.config['ceph_binary'], '--admin-daemon', socket_path] + command)
-        except CalledProcessError:
+            json_blob, err = _popen_check_output([self.config['ceph_binary'],
+                                                  '--admin-daemon',
+                                                  socket_path] + command)
+        except CalledProcessError, e:
             raise AdminSocketError(socket_path, command)
 
         try:
@@ -374,14 +407,19 @@ def _admin_command(self, socket_path, command):
     def _mon_command(self, cluster, command):
         try:
             json_blob, err = _popen_check_output(
-                [self.config['ceph_binary'], '--cluster', cluster, '-f', 'json-pretty'] + command)
+                [self.config['ceph_binary'],
+                 '--cluster',
+                 cluster,
+                 '-f',
+                 'json-pretty'] + command)
         except CalledProcessError:
             raise MonError(cluster, command)
 
         try:
             return json.loads(json_blob)
         except (ValueError, IndexError):
-            self.log.exception('Error parsing output from %s: %s' % (cluster, command))
+            message = 'Error parsing output from %s: %s' % (cluster, command)
+            self.log.exception(message)
             raise MonError(cluster, command)
 
     def _collect_cluster_stats(self, path):
@@ -400,21 +438,33 @@ def _collect_cluster_stats(self, path):
         fsid = mon_status['monmap']['fsid']
 
         # We are the leader, gather cluster-wide statistics
-        self.log.debug("mon leader found, gathering cluster stats for cluster '%s'" % cluster_name)
+        message = "mon leader found, gathering cluster stats for cluster '%s'"
+        self.log.debug(message, cluster_name)
 
         def publish_pool_stats(pool_id, stats):
             # Some of these guys we treat as counters, some as gauges
-            delta_fields = ['num_read', 'num_read_kb', 'num_write', 'num_write_kb', 'num_objects_recovered',
-                            'num_bytes_recovered', 'num_keys_recovered']
+            delta_fields = ['num_read',
+                            'num_read_kb',
+                            'num_write',
+                            'num_write_kb',
+                            'num_objects_recovered',
+                            'num_bytes_recovered',
+                            'num_keys_recovered']
             for k, v in stats.items():
-                self._publish_cluster_stats(cluster_name, fsid, "pool.{0}".format(pool_id), {k: v},
+                self._publish_cluster_stats(cluster_name,
+                                            fsid,
+                                            "pool.{0}".format(pool_id),
+                                            {k: v},
                                             counter=k in delta_fields)
 
         # Gather "ceph pg dump pools" and file the stats by pool
         for pool in self._mon_command(cluster_name, ['pg', 'dump', 'pools']):
             publish_pool_stats(pool['poolid'], pool['stat_sum'])
 
-        all_pools_stats = self._mon_command(cluster_name, ['pg', 'dump', 'summary'])['pg_stats_sum']['stat_sum']
+        all_pools_stats = self._mon_command(
+            cluster_name,
+            ['pg', 'dump', 'summary'])['pg_stats_sum']['stat_sum']
+
         publish_pool_stats('all', all_pools_stats)
 
         # Gather "ceph df"
@@ -447,11 +497,18 @@ def _collect_service_stats(self, path):
         stats, schema = self._get_perf_counters(path)
 
         if self.config['service_stats_global']:
-            counter_prefix = "{0}.{1}.{2}".format(self._cluster_id_prefix(cluster_name, fsid), service_type, service_id)
+            counter_prefix = "{0}.{1}.{2}".format(
+                self._cluster_id_prefix(cluster_name, fsid),
+                service_type,
+                service_id)
+
             self._publish_stats(counter_prefix, stats, schema, GlobalName)
         else:
             # The prefix is <cluster name>.<service type>.<service id>
-            counter_prefix = "{0}.{1}.{2}".format(cluster_name, service_type, service_id)
+            counter_prefix = "{0}.{1}.{2}".format(
+                cluster_name,
+                service_type,
+                service_id)
             self._publish_stats(counter_prefix, stats, schema, LocalName)
 
     def collect(self):

From 1216c030c55bf5b65702db5a722855855467dade Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Tue, 3 Nov 2015 19:11:23 +0000
Subject: [PATCH 23/26] ceph collector test: fix test to work with our
 CalledProcessError

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/test/testceph.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index 56c1f8214..8266323c6 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -147,10 +147,7 @@ def test_load_works(self, check_output):
 
     @patch('ceph._popen_check_output')
     def test_ceph_command_fails(self, check_output):
-        # this test check very little since the exceptionhandling ws moved into collect
-        # We've checked elsewhere that check_output is getting called with specific params
-        # TODO delete or sub in collect for _get_perf_counters
-        check_output.side_effect = subprocess.CalledProcessError(
+        check_output.side_effect = ceph.CalledProcessError(
             255, ['/usr/bin/ceph'], 'error!',
         )
         with self.assertRaises(ceph.AdminSocketError):

From c47e2c597feccc0c68fad611234f8bbe95330424 Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Wed, 11 Nov 2015 20:04:59 +0000
Subject: [PATCH 24/26] ceph collector: add docstrings to exceptions we define

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/ceph.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index 3017c6e97..c3aad7bc0 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -96,6 +96,9 @@ def _popen_check_output(*popenargs):
 
 
 class AdminSocketError(Exception):
+    """
+    Indicates an error communicating with Ceph services through their sockets
+    """
     def __init__(self, socket_name, command):
         self.socket_name = socket_name
         self.command = command
@@ -106,6 +109,9 @@ def __str__(self):
 
 
 class MonError(Exception):
+    """
+    Indicates an error in commands sent to the Ceph monitor
+    """
     def __init__(self, cluster_name, command):
         self.cluster_name = cluster_name
         self.command = command

From 8f06a94edf73ad0bc50803ab3237bdbefed764f5 Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Wed, 11 Nov 2015 20:15:49 +0000
Subject: [PATCH 25/26] ceph collector: document source of metric data types

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/ceph.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/collectors/ceph/ceph.py b/src/collectors/ceph/ceph.py
index c3aad7bc0..3e2adf8c4 100644
--- a/src/collectors/ceph/ceph.py
+++ b/src/collectors/ceph/ceph.py
@@ -28,6 +28,8 @@
 _NSEC_PER_SEC = 1000000000
 
 # Performance metric data types
+# see http://docs.ceph.com/docs/master/dev/perf_counters/#schema
+# for the source of these values
 _PERFCOUNTER_NONE = 0
 _PERFCOUNTER_TIME = 0x1
 _PERFCOUNTER_U64 = 0x2

From 02939e5d29cad18faf66fa5b9ad82e60d8e9fa00 Mon Sep 17 00:00:00 2001
From: Gregory Meno <gmeno@redhat.com>
Date: Thu, 12 Nov 2015 22:44:49 +0000
Subject: [PATCH 26/26] ceph collector test fix pep8 violation

Signed-off-by: Gregory Meno <gmeno@redhat.com>
---
 src/collectors/ceph/test/testceph.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/collectors/ceph/test/testceph.py b/src/collectors/ceph/test/testceph.py
index 8266323c6..6c8bee39e 100644
--- a/src/collectors/ceph/test/testceph.py
+++ b/src/collectors/ceph/test/testceph.py
@@ -42,7 +42,7 @@ def test_simple(self):
     @run_only_if_assertSequenceEqual_is_available
     def test_nested(self):
         data = {'a': 1, 'b': 2, 'c': {'d': 3}}
-        expected = [(['a'], 1), (['b'], 2), (['c','d'], 3)]
+        expected = [(['a'], 1), (['b'], 2), (['c', 'd'], 3)]
         actual = list(ceph.flatten_dictionary(data))
         self.assertSequenceEqual(actual, expected)