Merge lp:~tribaal/landscape-client/ceph-usage-report into lp:~landscape/landscape-client/trunk

[Chris Glass]

Add a Ceph storage monitor plugin that reports Ceph disk usage to the server

This branch adds a Ceph storage monitor plugin that reports ceph usage to the server if usage statistics are available from the host (namely, it tries running "ceph status" and "ceph quorum_status" to gather ring information).

This approach was chosen because the current state of the python library does not allow to extract this information easily. It is however pretty simple to change that in the future.

[Jerry Seutter (jseutter)]

+1 looks good

Just some lintian errors:
landscape/message_schemas.py:116:14: E203 whitespace before ':'
landscape/monitor/ceph.py:12:1: E302 expected 2 blank lines, found 1
landscape/monitor/tests/test_ceph.py:5:80: E501 line too long (80 characters)
landscape/monitor/tests/test_ceph.py:223:9: E301 expected 1 blank line, found 0

review: Approve

[Jerry Seutter (jseutter)]

+1 looks good

Just some lintian errors:
landscape/message_schemas.py:116:14: E203 whitespace before ':'
landscape/monitor/ceph.py:12:1: E302 expected 2 blank lines, found 1
landscape/monitor/tests/test_ceph.py:5:80: E501 line too long (80 characters)
landscape/monitor/tests/test_ceph.py:223:9: E301 expected 1 blank line, found 0

review: Approve

[lp:~tribaal/landscape-client/ceph-usage-report]

610. By Chris Glass on 2013-01-15

lint fixes

[Chris Glass (tribaal)]

All errors should be fixed.

[Alberto Donato (ack)]

Looks good! +1

A few nitpicks:

#1:
+ ring_id = self._ceph_ring_id
+ if ring_id is None:
+ ring_id = self._get_ceph_ring_id()
+ self._ceph_ring_id = ring_id

I think you can write this as

+ if self._ceph_ring_id is None:
+ self._ceph_ring_id = self._get_ceph_ring_id()

#2:
+ self.assertNotEqual([], plugin._ceph_usage_points)
+ self.assertEqual([(300, 1.0), (600, 1.0)], plugin._ceph_usage_points)

The first assert is redundant.

#3:
+ if len(message["ceph-usages"]) and message["ring-id"] is not None:

len() can be dropped.

review: Approve

[lp:~tribaal/landscape-client/ceph-usage-report]

611. By Chris Glass on 2013-01-15

Fixes all comments.

=== modified file 'landscape/message_schemas.py'
--- landscape/message_schemas.py	2013-01-09 13:20:10 +0000
+++ landscape/message_schemas.py	2013-01-15 10:26:22 +0000
@@ -111,6 +111,11 @@
     "cpu-usages": List(Tuple(Int(), Float())),
     })
 
+CEPH_USAGE = Message("ceph-usage", {
+    "ceph-usages": List(Tuple(Int(), Float())),
+    "ring-id": utf8,
+    })
+
 MEMORY_INFO = Message("memory-info", {
     "memory-info": List(Tuple(Float(), Int(), Int())),
     })
@@ -420,5 +425,6 @@
                ADD_PACKAGES, PACKAGE_REPORTER_RESULT, TEXT_MESSAGE, TEST,
                CUSTOM_GRAPH, REBOOT_REQUIRED, APT_PREFERENCES, EUCALYPTUS_INFO,
                EUCALYPTUS_INFO_ERROR, NETWORK_DEVICE, NETWORK_ACTIVITY,
-               REBOOT_REQUIRED_INFO, UPDATE_MANAGER_INFO, CPU_USAGE]:
+               REBOOT_REQUIRED_INFO, UPDATE_MANAGER_INFO, CPU_USAGE,
+               CEPH_USAGE]:
     message_schemas[schema.type] = schema

=== added file 'landscape/monitor/ceph.py'
--- landscape/monitor/ceph.py	1970-01-01 00:00:00 +0000
+++ landscape/monitor/ceph.py	2013-01-15 10:26:22 +0000
@@ -0,0 +1,150 @@
+import time
+import os
+
+from landscape.accumulate import Accumulator
+from landscape.lib.monitor import CoverageMonitor
+from landscape.lib.command import run_command, CommandError
+from landscape.monitor.plugin import MonitorPlugin
+
+ACCUMULATOR_KEY = "ceph-usage-accumulator"
+CEPH_CONFIG_FILE = "/etc/ceph/ceph.conf"
+
+
+class CephUsage(MonitorPlugin):
+    """
+    Plugin that captures Ceph usage information. This only works if the client
+    runs on one of the Ceph monitor nodes, and it noops otherwise.
+    """
+    persist_name = "ceph-usage"
+    # Prevent the Plugin base-class from scheduling looping calls.
+    run_interval = None
+
+    def __init__(self, interval=30, monitor_interval=60 * 60,
+                 create_time=time.time):
+        self._interval = interval
+        self._monitor_interval = monitor_interval
+        self._ceph_usage_points = []
+        self._ceph_ring_id = None
+        self._create_time = create_time
+        self._ceph_config = CEPH_CONFIG_FILE
+
+    def register(self, registry):
+        super(CephUsage, self).register(registry)
+        self._accumulate = Accumulator(self._persist, registry.step_size)
+
+        self.registry.reactor.call_every(self._interval, self.run)
+
+        self._monitor = CoverageMonitor(self._interval, 0.8,
+                                        "Ceph usage snapshot",
+                                        create_time=self._create_time)
+        self.registry.reactor.call_every(self._monitor_interval,
+                                         self._monitor.log)
+        self.registry.reactor.call_on("stop", self._monitor.log, priority=2000)
+        self.call_on_accepted("ceph-usage", self.send_message, True)
+
+    def create_message(self):
+        ceph_points = self._ceph_usage_points
+        ring_id = self._ceph_ring_id
+        self._ceph_usage_points = []
+        return {"type": "ceph-usage", "ceph-usages": ceph_points,
+                "ring-id": ring_id}
+
+    def send_message(self, urgent=False):
+        message = self.create_message()
+        if message["ceph-usages"] and message["ring-id"] is not None:
+            self.registry.broker.send_message(message, urgent=urgent)
+
+    def exchange(self, urgent=False):
+        self.registry.broker.call_if_accepted("ceph-usage",
+                                              self.send_message, urgent)
+
+    def run(self):
+        self._monitor.ping()
+
+        config_file = self._ceph_config
+        # Check if a ceph config file is available. No need to run anything
+        # if we know that we're not on a Ceph monitor node anyway.
+        if not os.path.exists(config_file):
+            # There is no config file - it's not a ceph machine.
+            return None
+
+        # Extract the ceph ring Id and cache it.
+        if self._ceph_ring_id is None:
+            self._ceph_ring_id = self._get_ceph_ring_id()
+
+        new_timestamp = int(self._create_time())
+        new_ceph_usage = self._get_ceph_usage()
+
+        step_data = None
+        if new_ceph_usage is not None:
+            step_data = self._accumulate(new_timestamp, new_ceph_usage,
+                                        ACCUMULATOR_KEY)
+        if step_data is not None:
+            self._ceph_usage_points.append(step_data)
+
+    def _get_ceph_usage(self):
+        """
+        Grab the ceph usage data by parsing the output of the "ceph status"
+        command output.
+        """
+        output = self._get_ceph_command_output()
+
+        if output is None:
+            return None
+
+        lines = output.split("\n")
+
+        pg_line = None
+        for line in lines:
+            if "pgmap" in line:
+                pg_line = line.split()
+                break
+
+        if pg_line is None:
+            return None
+
+        total = pg_line[-3]  # Total space
+        available = pg_line[-6]  # Available for objects
+        #used = pg_line[-9]  # Used by objects
+        # Note: used + available is NOT equal to total (there is some used
+        # space for duplication and system info etc...)
+
+        filled = int(total) - int(available)
+
+        return filled / float(total)
+
+    def _get_ceph_command_output(self):
+        try:
+            output = run_command("ceph status")
+        except (OSError, CommandError):
+            # If the command line client isn't available, we assume it's not
+            # a ceph monitor machine.
+            return None
+        return output
+
+    def _get_ceph_ring_id(self):
+        output = self._get_quorum_command_output()
+        lines = output.split("\n")
+        fsid_line = None
+        for line in lines:
+            if "fsid" in line:
+                fsid_line = line.split()
+                break
+
+        if fsid_line is None:
+            return None
+
+        wrapped_id = fsid_line[-1]
+        ring_id = wrapped_id.replace('",', '')
+        ring_id = ring_id.replace('"', '')
+
+        return ring_id
+
+    def _get_quorum_command_output(self):
+        try:
+            output = run_command("ceph quorum_status")
+        except (OSError, CommandError):
+            # If the command line client isn't available, we assume it's not
+            # a ceph monitor machine.
+            return None
+        return output

=== added file 'landscape/monitor/tests/test_ceph.py'
--- landscape/monitor/tests/test_ceph.py	1970-01-01 00:00:00 +0000
+++ landscape/monitor/tests/test_ceph.py	2013-01-15 10:26:22 +0000
@@ -0,0 +1,278 @@
+from landscape.tests.helpers import LandscapeTest, MonitorHelper
+from landscape.monitor.ceph import CephUsage
+
+
+SAMPLE_TEMPLATE = ("   health HEALTH_WARN 6 pgs degraded; 6 pgs stuck "
+"unclean\n"
+"monmap e2: 3 mons at {server-269703f4-5217-495a-b7f2-b3b3473c1719="
+"10.55.60.238:6789/0,server-3f370698-f3b0-4cbe-8db9-a18e304c952b="
+"10.55.60.141:6789/0,server-f635fa07-e36f-453c-b3d5-b4ce86fbc6ff="
+"10.55.60.241:6789/0}, election epoch 8, quorum 0,1,2 "
+"server-269703f4-5217-495a-b7f2-b3b3473c1719,"
+"server-3f370698-f3b0-4cbe-8db9-a18e304c952b,"
+"server-f635fa07-e36f-453c-b3d5-b4ce86fbc6ff\n   "
+"osdmap e9: 3 osds: 3 up, 3 in\n    "
+"pgmap v114: 192 pgs: 186 active+clean, 6 active+degraded; "
+"0 bytes data, %s MB used, %s MB / %s MB avail\n   "
+"mdsmap e1: 0/0/1 up\n\n")
+
+SAMPLE_OUTPUT = SAMPLE_TEMPLATE % (4296, 53880, 61248)
+
+SAMPLE_QUORUM = (''
+'{ "election_epoch": 8,\n'
+'  "quorum": [\n'
+'        0,\n'
+'        1,\n'
+'        2],\n'
+'  "monmap": { "epoch": 2,\n'
+'      "fsid": "%s",\n'
+'      "modified": "2013-01-13 16:58:00.141737",\n'
+'      "created": "0.000000",\n'
+'      "mons": [\n'
+'            { "rank": 0,\n'
+'              "name": "server-1be72d64-0ff2-4ac1-ad13-1c06c8201011",\n'
+'              "addr": "10.55.60.188:6789\/0"},\n'
+'            { "rank": 1,\n'
+'              "name": "server-e847f147-ed13-46c2-8e6d-768aa32657ab",\n'
+'              "addr": "10.55.60.202:6789\/0"},\n'
+'            { "rank": 2,\n'
+'              "name": "server-3c831a0b-51d5-43a9-95d5-63644f0965cc",\n'
+'              "addr": "10.55.60.205:6789\/0"}]}}\n'
+)
+
+SAMPLE_QUORUM_OUTPUT = SAMPLE_QUORUM % "ecbb8960-0e21-11e2-b495-83a88f44db01"
+
+
+class CephUsagePluginTest(LandscapeTest):
+    helpers = [MonitorHelper]
+
+    def test_get_ceph_usage_if_command_not_found(self):
+        """
+        When the ceph command cannot be found or accessed, the
+        C{_get_ceph_usage} method returns None.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        def return_none():
+            return None
+
+        plugin._get_ceph_command_output = return_none
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_usage()
+        self.assertIs(None, result)
+
+    def test_get_ceph_usage(self):
+        """
+        When the ceph command call returns output, the _get_ceph_usage method
+        returns the percentage of used space.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        def return_output():
+            return SAMPLE_OUTPUT
+
+        plugin._get_ceph_command_output = return_output
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_usage()
+        self.assertEqual(0.12029780564263323, result)
+
+    def test_get_ceph_usage_empty_disk(self):
+        """
+        When the ceph command call returns output for empty disks, the
+        _get_ceph_usage method returns 0.0 .
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        def return_output():
+            return SAMPLE_TEMPLATE % (0, 100, 100)
+
+        plugin._get_ceph_command_output = return_output
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_usage()
+        self.assertEqual(0.0, result)
+
+    def test_get_ceph_usage_full_disk(self):
+        """
+        When the ceph command call returns output for empty disks, the
+        _get_ceph_usage method returns 1.0 .
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        def return_output():
+            return SAMPLE_TEMPLATE % (100, 0, 100)
+
+        plugin._get_ceph_command_output = return_output
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_usage()
+        self.assertEqual(1.0, result)
+
+    def test_get_ceph_usage_no_information(self):
+        """
+        When the ceph command outputs something that does not contain the
+        disk usage information, the _get_ceph_usage method returns None.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        def return_output():
+            return "Blah\nblah"
+
+        plugin._get_ceph_command_output = return_output
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_usage()
+        self.assertEqual(None, result)
+
+    def test_never_exchange_empty_messages(self):
+        """
+        The plugin will create a message with an empty
+        C{ceph-usages} list when no previous data is available.  If an empty
+        message is created during exchange, it should not be queued.
+        """
+        self.mstore.set_accepted_types(["ceph-usage"])
+
+        plugin = CephUsage(create_time=self.reactor.time)
+        self.monitor.add(plugin)
+
+        self.monitor.exchange()
+        self.assertEqual(len(self.mstore.get_pending_messages()), 0)
+
+    def test_exchange_messages(self):
+        """
+        The Ceph usage plugin queues message when manager.exchange()
+        is called.
+        """
+        ring_id = "whatever"
+        self.mstore.set_accepted_types(["ceph-usage"])
+
+        plugin = CephUsage(create_time=self.reactor.time)
+        plugin._ceph_usage_points = [(60, 1.0)]
+        plugin._ceph_ring_id = ring_id
+        self.monitor.add(plugin)
+
+        self.monitor.exchange()
+
+        self.assertMessages(self.mstore.get_pending_messages(),
+                            [{"type": "ceph-usage",
+                              "ceph-usages": [(60, 1.0)],
+                              "ring-id": ring_id}])
+
+    def test_create_message(self):
+        """
+        Calling create_message returns an expected message.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+        self.monitor.add(plugin)
+
+        ring_id = "blah"
+        plugin._ceph_usage_points = []
+        plugin._ceph_ring_id = ring_id
+        message = plugin.create_message()
+
+        self.assertIn("type", message)
+        self.assertEqual(message["type"], "ceph-usage")
+        self.assertIn("ceph-usages", message)
+        self.assertEqual(ring_id, message["ring-id"])
+        ceph_usages = message["ceph-usages"]
+        self.assertEqual(len(ceph_usages), 0)
+
+        point = (60, 1.0)
+        plugin._ceph_usage_points = [point]
+        message = plugin.create_message()
+        self.assertIn("type", message)
+        self.assertEqual(message["type"], "ceph-usage")
+        self.assertIn("ceph-usages", message)
+        self.assertEqual(ring_id, message["ring-id"])
+        ceph_usages = message["ceph-usages"]
+        self.assertEqual(len(ceph_usages), 1)
+        self.assertEqual([point], ceph_usages)
+
+    def test_no_message_if_not_accepted(self):
+        """
+        Don't add any messages at all if the broker isn't currently
+        accepting their type.
+        """
+        interval = 30
+
+        plugin = CephUsage(create_time=self.reactor.time,
+                          interval=interval)
+
+        self.monitor.add(plugin)
+
+        self.reactor.advance(self.monitor.step_size * 2)
+        self.monitor.exchange()
+
+        self.mstore.set_accepted_types(["ceph-usage"])
+        self.assertMessages(list(self.mstore.get_pending_messages()), [])
+
+    def test_get_ceph_ring_id(self):
+        """
+        When given a well formatted command output, the _get_ceph_ring_id()
+        method returns the correct ring_id.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        uuid = "i-am-a-uuid"
+
+        def return_output():
+            return SAMPLE_QUORUM % uuid
+
+        plugin._get_quorum_command_output = return_output
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_ring_id()
+        self.assertEqual(uuid, result)
+
+    def test_get_ceph_ring_id_no_information(self):
+        """
+        When the _get_quorum_command_output method returns something without
+        the ring uuid information present, the _get-ceph_ring_id method returns
+        None.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+
+        def return_output():
+            return "Blah\nblah"
+
+        plugin._get_quorum_command_output = return_output
+
+        self.monitor.add(plugin)
+
+        result = plugin._get_ceph_ring_id()
+        self.assertEqual(None, result)
+
+    def test_plugin_run(self):
+        """
+        The plugin's run() method fills the _ceph_usage_points with
+        accumulated samples after each C{monitor.step_size} period.
+        The _ceph_ring_id member of the plugin is also filled with the output
+        of the _get_ceph_ring_id method.
+        """
+        plugin = CephUsage(create_time=self.reactor.time)
+        uuid = "i-am-a-unique-snowflake"
+
+        def return_quorum():
+            return SAMPLE_QUORUM % uuid
+
+        def return_full_disk():
+            return SAMPLE_TEMPLATE % (100, 0, 100)
+
+        plugin._ceph_config = "/etc/hosts"
+        plugin._get_quorum_command_output = return_quorum
+        plugin._get_ceph_command_output = return_full_disk
+
+        self.monitor.add(plugin)
+
+        self.reactor.advance(self.monitor.step_size * 2)
+
+        self.assertEqual([(300, 1.0), (600, 1.0)], plugin._ceph_usage_points)
+        self.assertEqual(uuid, plugin._ceph_ring_id)