Merge lp:~ack/landscape-client/ceph-monitor-plugin into lp:~landscape/landscape-client/trunk

[Alberto Donato]

This changes CephUsage from being a manager plugin to a monitor one, so it doesn't run as root.
It now looks for a landscape-specific ceph config and key in /var/lib/landscape/client/ceph-client, which is passed to the "ceph".

This changes CephUsage from being a manager plugin to a monitor one, so it doesn't run as root.
It now looks for a landscape-specific ceph config and key in /var/lib/landscape/client/ceph-client, which is passed to the "ceph".

The landscape-client charm at lp:~ack/charms/precise/landscape-client/ceph-client-interface creates the required ceph config for the plugin.

[Chris Glass (tribaal)]

Nice, it's much cleaner as a monitor plugin! +1
Thanks for refactoring the tests too :)

[1]
Please add some docstrings comments on the monitor plugin class to explain how to generate a usable ceph key, since some people might need/want to use the ceph feature of landscape with a legacy ceph insallation or a non-charm one.
There is already some description of how it works in the docstring for _run_ceph_command() but it would be good to have a quick refresher in the class/module docstring as well.

review: Approve

[Free Ekanayaka (free.ekanayaka)]

Looks good, +1!

#1

+ The configured keyring can be generated with:
+
+ ceph-authtool <keyring-file> --create-keyring
+ --name=client.landscape-client --add-key=<key>

Please add something like "This will be done automatically by the landscape-client charm, when deploying as a ceph node subordinate".

#2

+ output = run_command(" ".join(command))

To be picky, this should use Twisted's process protocol stuff (we have some helper for that), otherwise the reactor will block. It's important in case the ceph command hangs for some reason, for example. It can be addressed in a separate branch.

#3

Please mark the tests where you access private methods as whitebox ("test_wb_xxx").

review: Approve

[🤖 Landscape Builder (landscape-builder)]

There are additional revisions which have not been approved in review. Please seek review and approval of these new revisions.

=== removed file 'landscape/lib/command.py'
--- landscape/lib/command.py	2009-03-27 13:07:24 +0000
+++ landscape/lib/command.py	1970-01-01 00:00:00 +0000
@@ -1,38 +0,0 @@
-"""Shell commands execution."""
-import commands
-
-class CommandError(Exception):
-    """
-    Raised by L{run_command} in case of non-0 exit code.
-
-    @cvar command: The shell command that failed.
-    @cvar exit_status: Its non-zero exit status.
-    @cvar output: The command's output.
-    """
-    def __init__(self, command, exit_status, output):
-        self.command = command
-        self.exit_status = exit_status
-        self.output = output
-
-    def __str__(self):
-        return "'%s' exited with status %d (%s)" % (
-            self.command, self.exit_status, self.output)
-
-    def __repr__(self):
-        return "<CommandError command=<%s> exit_status=%d output=<%s>>" % (
-            self.command, self.exit_status, self.output)
-
-
-def run_command(command):
-    """
-    Execute a command in a shell and return the command's output.
-
-    If the command's exit status is not 0 a L{CommandError} exception
-    is raised.
-    """
-    exit_status, output = commands.getstatusoutput(command)
-    # shift down 8 bits to get shell-like exit codes
-    exit_status = exit_status >> 8
-    if exit_status != 0:
-        raise CommandError(command, exit_status, output)
-    return output

=== removed file 'landscape/lib/tests/test_command.py'
--- landscape/lib/tests/test_command.py	2011-07-05 05:09:11 +0000
+++ landscape/lib/tests/test_command.py	1970-01-01 00:00:00 +0000
@@ -1,32 +0,0 @@
-from landscape.tests.helpers import LandscapeTest
-
-from landscape.lib.command import run_command, CommandError
-
-
-class CommandTest(LandscapeTest):
-
-    def setUp(self):
-        super(CommandTest, self).setUp()
-
-    def test_basic(self):
-        self.assertEqual(run_command("echo test"), "test")
-
-    def test_non_0_exit_status(self):
-        try:
-            run_command("false")
-        except CommandError, error:
-            self.assertEqual(error.command, "false")
-            self.assertEqual(error.output, "")
-            self.assertEqual(error.exit_status, 1)
-        else:
-            self.fail("CommandError not raised")
-
-    def test_error_str(self):
-        self.assertEqual(str(CommandError("test_command", 1, "test output")),
-                         "'test_command' exited with status 1 "
-                         "(test output)")
-
-    def test_error_repr(self):
-        self.assertEqual(repr(CommandError("test_command", 1, "test output")),
-                         "<CommandError command=<test_command> "
-                         "exit_status=1 output=<test output>>")

=== modified file 'landscape/manager/config.py'
--- landscape/manager/config.py	2013-04-04 11:08:12 +0000
+++ landscape/manager/config.py	2013-06-04 16:15:35 +0000
@@ -6,7 +6,7 @@
 
 ALL_PLUGINS = ["ProcessKiller", "PackageManager", "UserManager",
                "ShutdownManager", "AptSources", "HardwareInfo",
-               "CephUsage", "KeystoneToken", "HAService"]
+               "KeystoneToken", "HAService"]
 
 
 class ManagerConfiguration(Configuration):

=== modified file 'landscape/manager/tests/test_config.py'
--- landscape/manager/tests/test_config.py	2013-04-04 11:38:19 +0000
+++ landscape/manager/tests/test_config.py	2013-06-04 16:15:35 +0000
@@ -12,9 +12,8 @@
     def test_plugin_factories(self):
         """By default all plugins are enabled."""
         self.assertEqual(["ProcessKiller", "PackageManager", "UserManager",
-                          "ShutdownManager", "AptSources",
-                          "HardwareInfo", "CephUsage", "KeystoneToken",
-                          "HAService"],
+                          "ShutdownManager", "AptSources", "HardwareInfo",
+                          "KeystoneToken", "HAService"],
                          ALL_PLUGINS)
         self.assertEqual(ALL_PLUGINS, self.config.plugin_factories)
 

=== renamed file 'landscape/manager/cephusage.py' => 'landscape/monitor/cephusage.py'
--- landscape/manager/cephusage.py	2013-04-02 14:56:19 +0000
+++ landscape/monitor/cephusage.py	2013-06-04 16:15:35 +0000
@@ -4,63 +4,70 @@
 import logging
 import re
 
+from twisted.internet.defer import inlineCallbacks, returnValue
+
 from landscape.accumulate import Accumulator
 from landscape.lib.monitor import CoverageMonitor
-from landscape.lib.command import run_command, CommandError
-from landscape.lib.persist import Persist
-from landscape.manager.plugin import ManagerPlugin
-
-ACCUMULATOR_KEY = "ceph-usage-accumulator"
-CEPH_CONFIG_FILE = "/etc/ceph/ceph.conf"
-
-EXP = re.compile(".*pgmap.*data, (\d+) MB used, (\d+) MB / (\d+) MB avail.*",
-                 flags=re.S)
-
-
-class CephUsage(ManagerPlugin):
+from landscape.lib.twisted_util import spawn_process
+from landscape.monitor.plugin import MonitorPlugin
+
+
+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.
+
+    The plugin requires the 'ceph' command to be available, which is run with a
+    config file in <data_path>/ceph-client/ceph.landscape-client.conf with the
+    following config:
+
+    [global]
+    auth supported = cephx
+    keyring = <keyring-file>
+    mon host = <ip>:6789
+
+    The configured keyring can be generated with:
+
+    ceph-authtool <keyring-file> --create-keyring
+        --name=client.landscape-client --add-key=<key>
+
+    The landscape-client charm automatically provides the client configuration
+    and key when deployed as subordinate of a ceph node.
     """
+
     persist_name = "ceph-usage"
     # Prevent the Plugin base-class from scheduling looping calls.
     run_interval = None
 
-    def __init__(self, interval=30, exchange_interval=60 * 60,
+    _usage_regexp = re.compile(
+        ".*pgmap.*data, (\d+) MB used, (\d+) MB / (\d+) MB avail.*",
+        flags=re.S)
+
+    def __init__(self, interval=30, monitor_interval=60 * 60,
                  create_time=time.time):
         self._interval = interval
-        self._exchange_interval = exchange_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
+        self._ceph_config = None
 
     def register(self, registry):
         super(CephUsage, self).register(registry)
+        self._ceph_config = os.path.join(
+            self.registry.config.data_path, "ceph-client",
+            "ceph.landscape-client.conf")
+
+        self._accumulate = Accumulator(self._persist, self._interval)
+        self._monitor = CoverageMonitor(
+            self._interval, 0.8, "Ceph usage snapshot",
+            create_time=self._create_time)
+
         self.registry.reactor.call_every(self._interval, self.run)
-
-        self._persist_filename = os.path.join(self.registry.config.data_path,
-                                              "ceph.bpickle")
-        self._persist = Persist(filename=self._persist_filename)
-
-        self._accumulate = Accumulator(self._persist, self._interval)
-
-        self._monitor = CoverageMonitor(self._interval, 0.8,
-                                        "Ceph usage snapshot",
-                                        create_time=self._create_time)
-        self.registry.reactor.call_every(self._exchange_interval,
-                                         self._monitor.log)
+        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)
-        self.registry.reactor.call_on("resynchronize", self._resynchronize)
-        self.registry.reactor.call_every(self.registry.config.flush_interval,
-                                         self.flush)
-
-    def _resynchronize(self):
-        self._persist.remove(self.persist_name)
-
-    def flush(self):
-        self._persist.save(self._persist_filename)
 
     def create_message(self):
         ceph_points = self._ceph_usage_points
@@ -75,30 +82,31 @@
             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)
+        self.registry.broker.call_if_accepted(
+            "ceph-usage", self.send_message, urgent)
 
+    @inlineCallbacks
     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
+        # Check if a ceph config file is available. If it's not , it's not a
+        # ceph machine or ceph is set up yet. No need to run anything in this
+        # case.
+        if self._ceph_config is None or not os.path.exists(self._ceph_config):
+            returnValue(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()
+            self._ceph_ring_id = yield self._get_ceph_ring_id()
 
         new_timestamp = int(self._create_time())
-        new_ceph_usage = self._get_ceph_usage()
+        new_ceph_usage = yield 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)
+            step_data = self._accumulate(
+                new_timestamp, new_ceph_usage, "ceph-usage-accumulator")
+
         if step_data is not None:
             self._ceph_usage_points.append(step_data)
 
@@ -107,52 +115,59 @@
         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
-
-        result = EXP.match(output)
-
-        if not result:
-            logging.error("Could not parse command output: '%s'." % output)
-            return None
-
-        (used, available, total) = result.groups()
-        # 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 parse(output):
+            if output is None:
+                return None
+
+            result = self._usage_regexp.match(output)
+            if not result:
+                logging.error("Could not parse command output: '%s'." % output)
+                return None
+
+            (used, available, total) = result.groups()
+            # 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)
+
+        return self._get_status_command_output().addCallback(parse)
+
+    def _get_status_command_output(self):
+        return self._run_ceph_command("status")
 
     def _get_ceph_ring_id(self):
-        output = self._get_quorum_command_output()
-        if output is None:
-            return None
-        try:
-            quorum_status = json.loads(output)
-            ring_id = quorum_status["monmap"]["fsid"]
-        except:
-            logging.error(
-                "Could not get ring_id from output: '%s'." % output)
-            return None
-        return ring_id
+        """Extract ceph ring id from ceph command output."""
+
+        def parse(output):
+            if output is None:
+                return  None
+
+            try:
+                quorum_status = json.loads(output)
+                ring_id = quorum_status["monmap"]["fsid"]
+            except:
+                logging.error(
+                    "Could not get ring_id from output: '%s'." % output)
+                return None
+            return ring_id
+
+        return self._get_quorum_command_output().addCallback(parse)
 
     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
+        return self._run_ceph_command("quorum_status")
+
+    def _run_ceph_command(self, *args):
+        """
+        Run the ceph command with the specified options using landscape ceph
+        key.  The keyring is expected to contain a configuration stanza with a
+        key for the "client.landscape-client" id.
+        """
+        params = ["--conf", self._ceph_config, "--id", "landscape-client"]
+        params.extend(args)
+        deferred = spawn_process("ceph", args=params)
+        # If the command line client isn't available, we assume it's not a ceph
+        # monitor machine.
+        deferred.addCallback(
+            lambda (out, err, code): out if code == 0 else None)
+        return deferred

=== modified file 'landscape/monitor/config.py'
--- landscape/monitor/config.py	2013-05-10 11:02:24 +0000
+++ landscape/monitor/config.py	2013-06-04 16:15:35 +0000
@@ -5,7 +5,8 @@
                "LoadAverage", "MemoryInfo", "MountInfo", "ProcessorInfo",
                "Temperature", "PackageMonitor", "UserMonitor",
                "RebootRequired", "AptPreferences", "NetworkActivity",
-               "NetworkDevice", "UpdateManager", "CPUUsage", "SwiftDeviceInfo"]
+               "NetworkDevice", "UpdateManager", "CPUUsage", "SwiftDeviceInfo",
+               "CephUsage"]
 
 
 class MonitorConfiguration(Configuration):

=== renamed file 'landscape/manager/tests/test_ceph.py' => 'landscape/monitor/tests/test_cephusage.py'
--- landscape/manager/tests/test_ceph.py	2013-04-02 14:56:19 +0000
+++ landscape/monitor/tests/test_cephusage.py	2013-06-04 16:15:35 +0000
@@ -1,171 +1,147 @@
 import os
 import json
 
-from landscape.tests.helpers import LandscapeTest, ManagerHelper
-from landscape.manager.cephusage import CephUsage
-
-
-SAMPLE_OLD_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_NEW_TEMPLATE = ("health HEALTH_OK\n"
-"   monmap e2: 3 mons at {inst-007=192.168.64.139:6789/0,"
-"inst-008=192.168.64.140:6789/0,inst-009=192.168.64.141:6789/0}, "
-"election epoch 6, quorum 0,1,2 inst-007,inst-008,inst-009\n"
-"   osdmap e28: 3 osds: 3 up, 3 in\n"
-"    pgmap v193861: 208 pgs: 208 active+clean; 5514 MB data, %s MB used, "
-"%s MB / %s MB avail; 1739KB/s wr, 54op/s\n"
-"   mdsmap e1: 0/0/1 up\n")
+from twisted.internet.defer import succeed
+
+from landscape.lib.fs import touch_file
+from landscape.tests.helpers import LandscapeTest, MonitorHelper
+from landscape.monitor.cephusage import CephUsage
+
+
+SAMPLE_OLD_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_NEW_TEMPLATE = (
+    "health HEALTH_OK\n"
+    "   monmap e2: 3 mons at {inst-007=192.168.64.139:6789/0,"
+    "inst-008=192.168.64.140:6789/0,inst-009=192.168.64.141:6789/0}, "
+    "election epoch 6, quorum 0,1,2 inst-007,inst-008,inst-009\n"
+    "   osdmap e28: 3 osds: 3 up, 3 in\n"
+    "    pgmap v193861: 208 pgs: 208 active+clean; 5514 MB data, %s MB used, "
+    "%s MB / %s MB avail; 1739KB/s wr, 54op/s\n"
+    "   mdsmap e1: 0/0/1 up\n")
 
 SAMPLE_OUTPUT = SAMPLE_NEW_TEMPLATE % (4296, 53880, 61248)
 SAMPLE_OLD_OUTPUT = SAMPLE_OLD_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 = (
+    '{ "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 = [ManagerHelper]
+    helpers = [MonitorHelper]
 
     def setUp(self):
         super(CephUsagePluginTest, self).setUp()
         self.mstore = self.broker_service.message_store
+        self.plugin = CephUsage(create_time=self.reactor.time)
 
-    def test_get_ceph_usage_if_command_not_found(self):
+    def test_wb_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.manager.add(plugin)
-
-        result = plugin._get_ceph_usage()
-        self.assertIs(None, result)
-
-    def test_get_ceph_usage(self):
+        self.plugin._get_status_command_output = lambda: succeed(None)
+        self.monitor.add(self.plugin)
+
+        self.assertIs(
+            None, self.successResultOf(self.plugin._get_ceph_usage()))
+
+    def test_wb_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.manager.add(plugin)
-
-        result = plugin._get_ceph_usage()
-        self.assertEqual(0.12029780564263323, result)
-
-    def test_get_ceph_usage_old_format(self):
+        self.plugin._get_status_command_output = lambda: succeed(SAMPLE_OUTPUT)
+        self.monitor.add(self.plugin)
+
+        self.assertEqual(
+            0.12029780564263323,
+            self.successResultOf(self.plugin._get_ceph_usage()))
+
+    def test_wb_get_ceph_usage_old_format(self):
         """
         The _get_ceph_usage method understands command output in the "old"
         format (the output changed around version 0.56.1)
         """
-        plugin = CephUsage(create_time=self.reactor.time)
-
-        def return_output():
-            return SAMPLE_OLD_OUTPUT
-
-        plugin._get_ceph_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_usage()
-        self.assertEqual(0.12029780564263323, result)
-
-    def test_get_ceph_usage_empty_disk(self):
+        self.plugin._get_status_command_output = (
+            lambda: succeed(SAMPLE_OLD_OUTPUT))
+        self.monitor.add(self.plugin)
+
+        self.assertEqual(
+            0.12029780564263323,
+            self.successResultOf(self.plugin._get_ceph_usage()))
+
+    def test_wb_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_NEW_TEMPLATE % (0, 100, 100)
-
-        plugin._get_ceph_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_usage()
-        self.assertEqual(0.0, result)
-
-    def test_get_ceph_usage_full_disk(self):
+        self.plugin._get_status_command_output = (
+            lambda: succeed(SAMPLE_NEW_TEMPLATE % (0, 100, 100)))
+        self.monitor.add(self.plugin)
+
+        self.assertEqual(
+            0.0, self.successResultOf(self.plugin._get_ceph_usage()))
+
+    def test_wb_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_NEW_TEMPLATE % (100, 0, 100)
-
-        plugin._get_ceph_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_usage()
-        self.assertEqual(1.0, result)
-
-    def test_get_ceph_usage_no_information(self):
+        self.plugin._get_status_command_output = (
+            lambda: succeed(SAMPLE_NEW_TEMPLATE % (100, 0, 100)))
+
+        self.monitor.add(self.plugin)
+        self.assertEqual(
+            1.0, self.successResultOf(self.plugin._get_ceph_usage()))
+
+    def test_wb_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)
         output = "Blah\nblah"
         error = "Could not parse command output: '%s'" % output
         self.log_helper.ignore_errors(error)
 
-        def return_output():
-            return output
-        plugin._get_ceph_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_usage()
-        self.assertEqual(None, result)
+        self.plugin._get_status_command_output = lambda: succeed(output)
+
+        self.monitor.add(self.plugin)
+        self.assertIs(
+            None, self.successResultOf(self.plugin._get_ceph_usage()))
 
     def test_never_exchange_empty_messages(self):
         """
@@ -174,12 +150,10 @@
         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.manager.add(plugin)
-
-        self.manager.exchange()
-        self.assertEqual(len(self.mstore.get_pending_messages()), 0)
+        self.monitor.add(self.plugin)
+
+        self.monitor.exchange()
+        self.assertEqual(0, len(self.mstore.get_pending_messages()))
 
     def test_exchange_messages(self):
         """
@@ -189,13 +163,11 @@
         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.manager.add(plugin)
-
-        self.manager.exchange()
-
+        self.plugin._ceph_usage_points = [(60, 1.0)]
+        self.plugin._ceph_ring_id = ring_id
+        self.monitor.add(self.plugin)
+
+        self.monitor.exchange()
         self.assertMessages(self.mstore.get_pending_messages(),
                             [{"type": "ceph-usage",
                               "ceph-usages": [(60, 1.0)],
@@ -205,13 +177,10 @@
         """
         Calling create_message returns an expected message.
         """
-        plugin = CephUsage(create_time=self.reactor.time)
-        self.manager.add(plugin)
-
         ring_id = "blah"
-        plugin._ceph_usage_points = []
-        plugin._ceph_ring_id = ring_id
-        message = plugin.create_message()
+        self.plugin._ceph_usage_points = []
+        self.plugin._ceph_ring_id = ring_id
+        message = self.plugin.create_message()
 
         self.assertIn("type", message)
         self.assertEqual(message["type"], "ceph-usage")
@@ -221,8 +190,8 @@
         self.assertEqual(len(ceph_usages), 0)
 
         point = (60, 1.0)
-        plugin._ceph_usage_points = [point]
-        message = plugin.create_message()
+        self.plugin._ceph_usage_points = [point]
+        message = self.plugin.create_message()
         self.assertIn("type", message)
         self.assertEqual(message["type"], "ceph-usage")
         self.assertIn("ceph-usages", message)
@@ -233,101 +202,75 @@
 
     def test_no_message_if_not_accepted(self):
         """
-        Don't add any messages at all if the broker isn't currently
-        accepting their type.
+        Don't add any messages at all if the broker isn't currently accepting
+        their type.
         """
         interval = 30
-        exchange_interval = 300
-
-        plugin = CephUsage(create_time=self.reactor.time,
-                          interval=interval, exchange_interval=300)
-
-        self.manager.add(plugin)
-
-        self.reactor.advance(exchange_interval * 2)
-        self.manager.exchange()
+        monitor_interval = 300
+
+        plugin = CephUsage(
+            interval=interval, monitor_interval=monitor_interval,
+            create_time=self.reactor.time)
+
+        self.monitor.add(plugin)
+
+        self.reactor.advance(monitor_interval * 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):
+    def test_wb_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.manager.add(plugin)
-
-        result = plugin._get_ceph_ring_id()
-        self.assertEqual(uuid, result)
-
-    def test_get_ceph_ring_id_valid_json_no_information(self):
+        self.plugin._get_quorum_command_output = (
+            lambda: succeed(SAMPLE_QUORUM % uuid))
+        self.assertEqual(
+            uuid, self.successResultOf(self.plugin._get_ceph_ring_id()))
+
+    def test_wb_get_ceph_ring_id_valid_json_no_information(self):
         """
         When the _get_quorum_command_output method returns something without
         the ring uuid information present but that is valid JSON, the
         _get_ceph_ring_id method returns None.
         """
-        plugin = CephUsage(create_time=self.reactor.time)
         error = "Could not get ring_id from output: '{\"election_epoch\": 8}'."
         self.log_helper.ignore_errors(error)
 
         def return_output():
             # Valid JSON - just without the info we're looking for.
             data = {"election_epoch": 8}
-            return json.dumps(data)
-
-        plugin._get_quorum_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_ring_id()
-        self.assertEqual(None, result)
-
-    def test_get_ceph_ring_id_no_information(self):
+            return succeed(json.dumps(data))
+
+        self.plugin._get_quorum_command_output = return_output
+        self.assertIs(
+            None, self.successResultOf(self.plugin._get_ceph_ring_id()))
+
+    def test_wb_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)
         error = "Could not get ring_id from output: 'Blah\nblah'."
         self.log_helper.ignore_errors(error)
 
-        def return_output():
-            return "Blah\nblah"
-
-        plugin._get_quorum_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_ring_id()
-        self.assertEqual(None, result)
-
-    def test_get_ceph_ring_id_command_exception(self):
+        self.plugin._get_quorum_command_output = lambda: succeed("Blah\nblah")
+        self.assertIs(
+            None, self.successResultOf(self.plugin._get_ceph_ring_id()))
+
+    def test_wb_get_ceph_ring_id_command_exception(self):
         """
         When the _get_quorum_command_output method returns None (if an
         exception happened for example), the _get_ceph_ring_id method
         returns None and logs no error.
         """
-        plugin = CephUsage(create_time=self.reactor.time)
-
-        def return_output():
-            return None
-
-        plugin._get_quorum_command_output = return_output
-
-        self.manager.add(plugin)
-
-        result = plugin._get_ceph_ring_id()
-        self.assertEqual(None, result)
+        self.plugin._get_quorum_command_output = lambda: succeed(None)
+        self.assertIs(
+            None, self.successResultOf(self.plugin._get_ceph_ring_id()))
 
     def test_plugin_run(self):
         """
@@ -336,60 +279,44 @@
         The _ceph_ring_id member of the plugin is also filled with the output
         of the _get_ceph_ring_id method.
         """
-        exchange_interval = 300
-        interval = exchange_interval
-        plugin = CephUsage(create_time=self.reactor.time,
-                           exchange_interval=exchange_interval,
-                           interval=interval)
+        monitor_interval = 300
+        interval = monitor_interval
+        plugin = CephUsage(
+            interval=interval, monitor_interval=monitor_interval,
+            create_time=self.reactor.time)
+
         uuid = "i-am-a-unique-snowflake"
 
-        def return_quorum():
-            return SAMPLE_QUORUM % uuid
-
-        def return_full_disk():
-            return SAMPLE_NEW_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.manager.add(plugin)
-
-        self.reactor.advance(exchange_interval * 2)
+        # The config file must be present for the plugin to run.
+        ceph_client_dir = os.path.join(self.config.data_path, "ceph-client")
+        ceph_conf = os.path.join(ceph_client_dir, "ceph.landscape-client.conf")
+        os.mkdir(ceph_client_dir)
+        touch_file(ceph_conf)
+        plugin._ceph_config = ceph_conf
+
+        plugin._get_quorum_command_output = (
+            lambda: succeed(SAMPLE_QUORUM % uuid))
+        plugin._get_status_command_output = (
+            lambda: succeed(SAMPLE_NEW_TEMPLATE % (100, 0, 100)))
+        self.monitor.add(plugin)
+
+        self.reactor.advance(monitor_interval * 2)
 
         self.assertEqual([(300, 1.0), (600, 1.0)], plugin._ceph_usage_points)
         self.assertEqual(uuid, plugin._ceph_ring_id)
 
-    def test_flush_persists_data_to_disk(self):
-        """
-        The plugin's C{flush} method is called every C{flush_interval} and
-        creates the perists file.
-        """
-        flush_interval = self.config.flush_interval
-        persist_filename = os.path.join(self.config.data_path, "ceph.bpickle")
-
-        self.assertFalse(os.path.exists(persist_filename))
-        plugin = CephUsage(create_time=self.reactor.time)
-        self.manager.add(plugin)
-
-        self.reactor.advance(flush_interval)
-
-        self.assertTrue(os.path.exists(persist_filename))
-
     def test_resynchronize_message_calls_resynchronize_method(self):
         """
         If the reactor fires a "resynchronize" even the C{_resynchronize}
         method on the ceph plugin object is called.
         """
-        plugin = CephUsage(create_time=self.reactor.time)
-
         self.called = False
 
         def stub_resynchronize():
             self.called = True
-        plugin._resynchronize = stub_resynchronize
+        self.plugin._resynchronize = stub_resynchronize
 
-        self.manager.add(plugin)
+        self.monitor.add(self.plugin)
 
         self.reactor.fire("resynchronize")