diff --git a/landscape/client/manager/config.py b/landscape/client/manager/config.py
index f89ebbc99..40fabce3d 100644
--- a/landscape/client/manager/config.py
+++ b/landscape/client/manager/config.py
@@ -15,6 +15,8 @@
     "SnapManager",
     "SnapServicesManager",
     "UbuntuProInfo",
+    "LivePatch",
+    "UbuntuProRebootRequired",
 ]
 
 
diff --git a/landscape/client/monitor/livepatch.py b/landscape/client/manager/livepatch.py
similarity index 92%
rename from landscape/client/monitor/livepatch.py
rename to landscape/client/manager/livepatch.py
index 61735087e..40abb9513 100644
--- a/landscape/client/monitor/livepatch.py
+++ b/landscape/client/manager/livepatch.py
@@ -1,11 +1,14 @@
 import json
+import logging
 import subprocess
+import traceback
 import yaml
 
-from landscape.client.monitor.plugin import DataWatcher
 
+from landscape.client.manager.plugin import DataWatcherManager
 
-class LivePatch(DataWatcher):
+
+class LivePatch(DataWatcherManager):
     """
     Plugin that captures and reports Livepatch status information
     information.
@@ -46,6 +49,7 @@ def get_livepatch_status(format_type):
         data['return_code'] = -2
         data['error'] = str(exc)
         data['output'] = ""
+        logging.error(traceback.format_exc())
     else:
         output = completed_process.stdout.strip()
         try:
diff --git a/landscape/client/manager/plugin.py b/landscape/client/manager/plugin.py
index ea20b8c74..f4ab16218 100644
--- a/landscape/client/manager/plugin.py
+++ b/landscape/client/manager/plugin.py
@@ -1,8 +1,12 @@
+import logging
+from pathlib import Path
+
 from twisted.internet.defer import maybeDeferred
 
 from landscape.client.broker.client import BrokerClientPlugin
 from landscape.lib.format import format_object
 from landscape.lib.log import log_failure
+from landscape.lib.persist import Persist
 
 # Protocol messages! Same constants are defined in the server.
 FAILED = 5
@@ -66,3 +70,71 @@ def send(args):
         deferred.addCallback(send)
 
         return deferred
+
+
+class DataWatcherManager(ManagerPlugin):
+    """
+    A utility for plugins which send data to the Landscape server
+    which does not constantly change. New messages will only be sent
+    when the result of get_data() has changed since the last time it
+    was called. Note this is the same as the DataWatcher plugin but
+    for Manager plugins instead of Monitor.Subclasses should provide
+    a get_data method
+    """
+
+    message_type = None
+
+    def __init__(self):
+        super().__init__()
+        self._persist = None
+
+    def register(self, registry):
+        super().register(registry)
+        self._persist_filename = Path(
+            self.registry.config.data_path,
+            self.message_type + '.manager.bpkl',
+        )
+        self._persist = Persist(filename=self._persist_filename)
+        self.call_on_accepted(self.message_type, self.send_message)
+
+    def run(self):
+        return self.registry.broker.call_if_accepted(
+            self.message_type,
+            self.send_message,
+        )
+
+    def send_message(self):
+        """Send a message to the broker if the data has changed since the last
+        call"""
+        result = self.get_new_data()
+        if not result:
+            logging.debug("{} unchanged so not sending".format(
+                          self.message_type))
+            return
+        logging.debug("Sending new {} data!".format(self.message_type))
+        message = {"type": self.message_type, self.message_type: result}
+        return self.registry.broker.send_message(message, self._session_id)
+
+    def get_new_data(self):
+        """Returns the data only if it has changed"""
+        data = self.get_data()
+        if self._persist is None:  # Persist not initialized yet
+            return data
+        elif self._persist.get("data") != data:
+            self._persist.set("data", data)
+            return data
+        else:  # Data not changed
+            return None
+
+    def get_data(self):
+        """
+        The result of this will be cached and subclasses must implement this
+        and return the correct return type defined in the server bound message
+        schema
+        """
+        raise NotImplementedError("Subclasses must implement get_data()")
+
+    def _reset(self):
+        """Reset the persist."""
+        if self._persist:
+            self._persist.remove("data")
diff --git a/landscape/client/manager/tests/test_config.py b/landscape/client/manager/tests/test_config.py
index 085bd09fe..bb322a578 100644
--- a/landscape/client/manager/tests/test_config.py
+++ b/landscape/client/manager/tests/test_config.py
@@ -23,6 +23,8 @@ def test_plugin_factories(self):
                 "SnapManager",
                 "SnapServicesManager",
                 "UbuntuProInfo",
+                "LivePatch",
+                "UbuntuProRebootRequired"
             ],
             ALL_PLUGINS,
         )
diff --git a/landscape/client/monitor/tests/test_livepatch.py b/landscape/client/manager/tests/test_livepatch.py
similarity index 86%
rename from landscape/client/monitor/tests/test_livepatch.py
rename to landscape/client/manager/tests/test_livepatch.py
index 85cbc772c..a95f6d0a8 100644
--- a/landscape/client/monitor/tests/test_livepatch.py
+++ b/landscape/client/manager/tests/test_livepatch.py
@@ -2,8 +2,8 @@
 import yaml
 from unittest import mock
 
-from landscape.client.monitor.livepatch import LivePatch
-from landscape.client.tests.helpers import LandscapeTest, MonitorHelper
+from landscape.client.manager.livepatch import LivePatch
+from landscape.client.tests.helpers import LandscapeTest, ManagerHelper
 
 
 def subprocess_livepatch_mock(*args, **kwargs):
@@ -19,20 +19,21 @@ def subprocess_livepatch_mock(*args, **kwargs):
 class LivePatchTest(LandscapeTest):
     """Livepatch status plugin tests."""
 
-    helpers = [MonitorHelper]
+    helpers = [ManagerHelper]
 
     def setUp(self):
         super(LivePatchTest, self).setUp()
+        self.mstore = self.broker_service.message_store
         self.mstore.set_accepted_types(["livepatch"])
 
     def test_livepatch(self):
         """Tests calling livepatch status."""
         plugin = LivePatch()
-        self.monitor.add(plugin)
 
         with mock.patch("subprocess.run") as run_mock:
             run_mock.side_effect = subprocess_livepatch_mock
-            plugin.exchange()
+            self.manager.add(plugin)
+            plugin.run()
 
         messages = self.mstore.get_pending_messages()
         self.assertTrue(len(messages) > 0)
@@ -47,11 +48,11 @@ def test_livepatch(self):
     def test_livepatch_when_not_installed(self):
         """Tests calling livepatch when it is not installed."""
         plugin = LivePatch()
-        self.monitor.add(plugin)
 
         with mock.patch("subprocess.run") as run_mock:
             run_mock.side_effect = FileNotFoundError("Not found!")
-            plugin.exchange()
+            self.manager.add(plugin)
+            plugin.run()
 
         messages = self.mstore.get_pending_messages()
         message = json.loads(messages[0]["livepatch"])
@@ -61,14 +62,17 @@ def test_livepatch_when_not_installed(self):
         self.assertEqual(message["json"]["return_code"], -1)
         self.assertEqual(message["humane"]["return_code"], -1)
 
-    def test_undefined_exception(self):
+    @mock.patch('landscape.client.manager.livepatch.logging.error')
+    def test_undefined_exception(self, logger_mock):
         """Tests calling livepatch when random exception occurs"""
         plugin = LivePatch()
-        self.monitor.add(plugin)
 
         with mock.patch("subprocess.run") as run_mock:
             run_mock.side_effect = ValueError("Not found!")
-            plugin.exchange()
+            self.manager.add(plugin)
+            plugin.run()
+
+        logger_mock.assert_called()
 
         messages = self.mstore.get_pending_messages()
         message = json.loads(messages[0]["livepatch"])
@@ -83,13 +87,13 @@ def test_yaml_json_parse_error(self):
         If json or yaml parsing error than show exception and unparsed data
         """
         plugin = LivePatch()
-        self.monitor.add(plugin)
 
         invalid_data = "'"
         with mock.patch("subprocess.run") as run_mock:
             run_mock.return_value = mock.Mock(stdout=invalid_data)
             run_mock.return_value.returncode = 0
-            plugin.exchange()
+            self.manager.add(plugin)
+            plugin.run()
 
         messages = self.mstore.get_pending_messages()
         message = json.loads(messages[0]["livepatch"])
@@ -104,14 +108,14 @@ def test_empty_string(self):
         If livepatch is disabled, stdout is empty string
         """
         plugin = LivePatch()
-        self.monitor.add(plugin)
 
         invalid_data = ""
         with mock.patch("subprocess.run") as run_mock:
             run_mock.return_value = mock.Mock(stdout=invalid_data,
                                               stderr='Error')
             run_mock.return_value.returncode = 1
-            plugin.exchange()
+            self.manager.add(plugin)
+            plugin.run()
 
         messages = self.mstore.get_pending_messages()
         message = json.loads(messages[0]["livepatch"])
@@ -126,11 +130,11 @@ def test_timestamped_fields_deleted(self):
         """This is so data doesn't keep getting sent if not changed"""
 
         plugin = LivePatch()
-        self.monitor.add(plugin)
 
         with mock.patch("subprocess.run") as run_mock:
             run_mock.side_effect = subprocess_livepatch_mock
-            plugin.exchange()
+            self.manager.add(plugin)
+            plugin.run()
 
         messages = self.mstore.get_pending_messages()
         self.assertTrue(len(messages) > 0)
diff --git a/landscape/client/manager/tests/test_plugin.py b/landscape/client/manager/tests/test_plugin.py
index 1aadd0acc..2bd8d08fd 100644
--- a/landscape/client/manager/tests/test_plugin.py
+++ b/landscape/client/manager/tests/test_plugin.py
@@ -1,7 +1,7 @@
 from twisted.internet.defer import Deferred
 
 from landscape.client.manager.plugin import FAILED
-from landscape.client.manager.plugin import ManagerPlugin
+from landscape.client.manager.plugin import ManagerPlugin, DataWatcherManager
 from landscape.client.manager.plugin import SUCCEEDED
 from landscape.client.tests.helpers import LandscapeTest
 from landscape.client.tests.helpers import ManagerHelper
@@ -126,3 +126,37 @@ def assert_messages(ignored):
         result.addCallback(assert_messages)
         deferred.callback("blah")
         return result
+
+
+class StubDataWatchingPlugin(DataWatcherManager):
+
+    message_type = "wubble"
+
+    def __init__(self, data=None):
+        self.data = data
+
+    def get_data(self):
+        return self.data
+
+
+class DataWatcherManagerTest(LandscapeTest):
+
+    helpers = [ManagerHelper]
+
+    def setUp(self):
+        LandscapeTest.setUp(self)
+        self.plugin = StubDataWatchingPlugin("hello world")
+        self.plugin.register(self.manager)
+
+    def test_get_message(self):
+        self.assertEqual(
+            self.plugin.get_new_data(),
+            "hello world",
+        )
+
+    def test_get_message_unchanging(self):
+        self.assertEqual(
+            self.plugin.get_new_data(),
+            "hello world",
+        )
+        self.assertEqual(self.plugin.get_new_data(), None)
diff --git a/landscape/client/manager/tests/test_ubuntuproinfo.py b/landscape/client/manager/tests/test_ubuntuproinfo.py
index 9d5daa88a..4e3a7da20 100644
--- a/landscape/client/manager/tests/test_ubuntuproinfo.py
+++ b/landscape/client/manager/tests/test_ubuntuproinfo.py
@@ -7,16 +7,17 @@
 from landscape.client.manager.ubuntuproinfo import get_ubuntu_pro_info
 from landscape.client.manager.ubuntuproinfo import UbuntuProInfo
 from landscape.client.tests.helpers import LandscapeTest
-from landscape.client.tests.helpers import MonitorHelper
+from landscape.client.tests.helpers import ManagerHelper
 
 
 class UbuntuProInfoTest(LandscapeTest):
     """Ubuntu Pro info plugin tests."""
 
-    helpers = [MonitorHelper]
+    helpers = [ManagerHelper]
 
     def setUp(self):
         super().setUp()
+        self.mstore = self.broker_service.message_store
         self.mstore.set_accepted_types(["ubuntu-pro-info"])
 
     def test_ubuntu_pro_info(self):
@@ -27,7 +28,7 @@ def test_ubuntu_pro_info(self):
             run_mock.return_value = mock.Mock(
                 stdout='"This is a test"',
             )
-            self.monitor.add(plugin)
+            self.manager.add(plugin)
             plugin.run()
 
         run_mock.assert_called()
@@ -39,7 +40,7 @@ def test_ubuntu_pro_info(self):
     def test_ubuntu_pro_info_no_pro(self):
         """Tests calling `pro status` when it is not installed."""
         plugin = UbuntuProInfo()
-        self.monitor.add(plugin)
+        self.manager.add(plugin)
 
         with mock.patch("subprocess.run") as run_mock:
             run_mock.side_effect = FileNotFoundError()
@@ -78,7 +79,7 @@ def datetime_is_aware(d):
     def test_persistence_unchanged_data(self):
         """If data hasn't changed, a new message is not sent"""
         plugin = UbuntuProInfo()
-        self.monitor.add(plugin)
+        self.manager.add(plugin)
         data = '"Initial data!"'
 
         with mock.patch("subprocess.run") as run_mock:
@@ -106,7 +107,7 @@ def test_persistence_unchanged_data(self):
     def test_persistence_changed_data(self):
         """New data will be sent in a new message in the queue"""
         plugin = UbuntuProInfo()
-        self.monitor.add(plugin)
+        self.manager.add(plugin)
 
         with mock.patch("subprocess.run") as run_mock:
             run_mock.return_value = mock.Mock(
@@ -135,7 +136,7 @@ def test_persistence_reset(self):
         """Resetting the plugin will allow a message with identical data to
         be sent"""
         plugin = UbuntuProInfo()
-        self.monitor.add(plugin)
+        self.manager.add(plugin)
         data = '"Initial data!"'
 
         with mock.patch("subprocess.run") as run_mock:
diff --git a/landscape/client/manager/tests/test_ubuntuprorebootrequired.py b/landscape/client/manager/tests/test_ubuntuprorebootrequired.py
new file mode 100644
index 000000000..629427ad6
--- /dev/null
+++ b/landscape/client/manager/tests/test_ubuntuprorebootrequired.py
@@ -0,0 +1,57 @@
+import json
+from unittest.mock import patch
+
+from landscape.client.manager.ubuntuprorebootrequired import (
+    UbuntuProRebootRequired,
+)
+from landscape.client.tests.helpers import LandscapeTest, ManagerHelper
+
+
+class UbuntuProRebootRequiredTest(LandscapeTest):
+    """Ubuntu Pro required plugin tests."""
+
+    helpers = [ManagerHelper]
+
+    def setUp(self):
+        super().setUp()
+        self.mstore = self.broker_service.message_store
+        self.mstore.set_accepted_types(["ubuntu-pro-reboot-required"])
+
+    @patch('landscape.client.manager.ubuntuprorebootrequired.get_reboot_info')
+    def test_ubuntu_pro_reboot_required(self, mock_reboot_info):
+        """Basic test"""
+
+        mock_reboot_info.return_value = "reboot_required"
+        plugin = UbuntuProRebootRequired()
+        self.manager.add(plugin)
+        plugin.run()
+
+        messages = self.mstore.get_pending_messages()
+
+        self.assertGreater(len(messages), 0)
+        self.assertIn("ubuntu-pro-reboot-required", messages[0])
+        info = json.loads(messages[0]["ubuntu-pro-reboot-required"])
+        self.assertEqual("reboot_required", info["output"])
+        self.assertFalse(info["error"])
+
+    @patch('landscape.client.manager.ubuntuprorebootrequired.get_reboot_info')
+    @patch('landscape.client.manager.ubuntuprorebootrequired.logging.error')
+    def test_generic_error(self, mock_logger, mock_reboot_info):
+        """
+        Test we get a response and an error is logged when exception triggers
+        """
+
+        mock_reboot_info.side_effect = Exception("Error!")
+
+        plugin = UbuntuProRebootRequired()
+        self.manager.add(plugin)
+        plugin.run()
+
+        messages = self.mstore.get_pending_messages()
+
+        mock_logger.assert_called()
+        self.assertGreater(len(messages), 0)
+        self.assertIn("ubuntu-pro-reboot-required", messages[0])
+        info = json.loads(messages[0]["ubuntu-pro-reboot-required"])
+        self.assertFalse(info["output"])
+        self.assertTrue(info["error"])
diff --git a/landscape/client/manager/ubuntuproinfo.py b/landscape/client/manager/ubuntuproinfo.py
index 02348f692..d4ed04688 100644
--- a/landscape/client/manager/ubuntuproinfo.py
+++ b/landscape/client/manager/ubuntuproinfo.py
@@ -3,16 +3,14 @@
 from datetime import datetime
 from datetime import timedelta
 from datetime import timezone
-from pathlib import Path
 
 from landscape.client import IS_CORE
 from landscape.client import IS_SNAP
+from landscape.client.manager.plugin import DataWatcherManager
 from landscape.client import UA_DATA_DIR
-from landscape.client.manager.plugin import ManagerPlugin
-from landscape.lib.persist import Persist
 
 
-class UbuntuProInfo(ManagerPlugin):
+class UbuntuProInfo(DataWatcherManager):
     """
     Plugin that captures and reports Ubuntu Pro registration
     information.
@@ -26,41 +24,10 @@ class UbuntuProInfo(ManagerPlugin):
     message_type = "ubuntu-pro-info"
     run_interval = 900  # 15 minutes
 
-    def register(self, registry):
-        super().register(registry)
-        self._persist_filename = Path(
-            self.registry.config.data_path,
-            "ubuntu-pro-info.bpickle",
-        )
-        self._persist = Persist(filename=self._persist_filename)
-        self.call_on_accepted(self.message_type, self.send_message)
-
-    def run(self):
-        return self.registry.broker.call_if_accepted(
-            self.message_type,
-            self.send_message,
-        )
-
-    def send_message(self):
-        """Send a message to the broker if the data has changed since the last
-        call"""
-        result = self.get_data()
-        if not result:
-            return
-        message = {"type": self.message_type, "ubuntu-pro-info": result}
-        return self.registry.broker.send_message(message, self._session_id)
-
     def get_data(self):
-        """Persist data to avoid sending messages if result hasn't changed"""
         ubuntu_pro_info = get_ubuntu_pro_info()
-
-        if self._persist.get("data") != ubuntu_pro_info:
-            self._persist.set("data", ubuntu_pro_info)
-            return json.dumps(ubuntu_pro_info, separators=(",", ":"))
-
-    def _reset(self):
-        """Reset the persist."""
-        self._persist.remove("data")
+        return json.dumps(ubuntu_pro_info, separators=(",", ":"),
+                          sort_keys=True)
 
 
 def get_ubuntu_pro_info() -> dict:
diff --git a/landscape/client/manager/ubuntuprorebootrequired.py b/landscape/client/manager/ubuntuprorebootrequired.py
new file mode 100644
index 000000000..ed1a6a61f
--- /dev/null
+++ b/landscape/client/manager/ubuntuprorebootrequired.py
@@ -0,0 +1,44 @@
+import logging
+import json
+import traceback
+
+from landscape.client.manager.plugin import DataWatcherManager
+
+
+def get_reboot_info():
+    """
+    This code is wrapped in a function so the import or any other exceptions
+    can be caught and also so it can be mocked
+    """
+    from uaclient.api.u.pro.security.status.reboot_required.v1 import (
+        reboot_required,)
+    return reboot_required().to_dict()
+
+
+class UbuntuProRebootRequired(DataWatcherManager):
+    """
+    Plugin that captures and reports from Ubuntu Pro API if the system needs to
+    be rebooted. The `uaclient` Python API should be installed by default.
+    """
+
+    message_type = "ubuntu-pro-reboot-required"
+    scope = "ubuntu-pro"
+    run_immediately = True
+    run_interval = 900  # 15 minutes
+
+    def get_data(self):
+        """
+        Return the JSON formatted output of "reboot-required" from Ubuntu Pro
+        API.
+        """
+        data = {}
+        try:
+            info = get_reboot_info()
+            json.dumps(info)  # Make sure data is json serializable
+            data["error"] = ""
+            data["output"] = info
+        except Exception as exc:
+            data["error"] = str(exc)
+            data["output"] = {}
+            logging.error(traceback.format_exc())
+        return json.dumps(data, sort_keys=True)
diff --git a/landscape/client/monitor/config.py b/landscape/client/monitor/config.py
index 4fb351e78..f0113eaac 100644
--- a/landscape/client/monitor/config.py
+++ b/landscape/client/monitor/config.py
@@ -20,8 +20,6 @@
     "SwiftUsage",
     "CephUsage",
     "ComputerTags",
-    "LivePatch",
-    "UbuntuProRebootRequired",
     "SnapServicesMonitor",
 ]
 
diff --git a/landscape/client/monitor/tests/test_ubuntuprorebootrequired.py b/landscape/client/monitor/tests/test_ubuntuprorebootrequired.py
deleted file mode 100644
index 442d08327..000000000
--- a/landscape/client/monitor/tests/test_ubuntuprorebootrequired.py
+++ /dev/null
@@ -1,29 +0,0 @@
-from landscape.client.monitor.ubuntuprorebootrequired import (
-    UbuntuProRebootRequired,
-)
-from landscape.client.tests.helpers import LandscapeTest, MonitorHelper
-
-
-class UbuntuProRebootRequiredTest(LandscapeTest):
-    """Ubuntu Pro required plugin tests."""
-
-    helpers = [MonitorHelper]
-
-    def setUp(self):
-        super().setUp()
-        self.mstore.set_accepted_types(["ubuntu-pro-reboot-required"])
-
-    def test_ubuntu_pro_reboot_required(self):
-        """Tests calling reboot required."""
-
-        plugin = UbuntuProRebootRequired()
-        self.monitor.add(plugin)
-        plugin.exchange()
-
-        messages = self.mstore.get_pending_messages()
-
-        self.assertGreater(len(messages), 0)
-        self.assertIn("ubuntu-pro-reboot-required", messages[0])
-        self.assertIn(
-            "reboot_required", messages[0]["ubuntu-pro-reboot-required"]
-        )
diff --git a/landscape/client/monitor/ubuntuprorebootrequired.py b/landscape/client/monitor/ubuntuprorebootrequired.py
deleted file mode 100644
index a401725c6..000000000
--- a/landscape/client/monitor/ubuntuprorebootrequired.py
+++ /dev/null
@@ -1,27 +0,0 @@
-from uaclient.api.u.pro.security.status.reboot_required.v1 import (
-    reboot_required,
-)
-
-from landscape.client.monitor.plugin import DataWatcher
-
-
-class UbuntuProRebootRequired(DataWatcher):
-    """
-    Plugin that captures and reports from Ubuntu Pro API if the system needs to
-    be rebooted. The `uaclient` Python API should be installed by default.
-    """
-
-    message_type = "ubuntu-pro-reboot-required"
-    message_key = message_type
-    persist_name = message_type
-    scope = "ubuntu-pro"
-    run_immediately = True
-    run_interval = 900  # 15 minutes
-
-    def get_data(self):
-        """
-        Return the JSON formatted output of "reboot-required" from Ubuntu Pro
-        API.
-        """
-
-        return reboot_required().to_json()
diff --git a/snap/hooks/default-configure b/snap/hooks/default-configure
index 11ebe7f2a..304b559db 100644
--- a/snap/hooks/default-configure
+++ b/snap/hooks/default-configure
@@ -25,7 +25,7 @@ if [ -z "$_script_users" ]; then
 fi
 
 if [ -z "$_manager_plugins" ]; then
-    _manager_plugins="ProcessKiller,UserManager,ShutdownManager,HardwareInfo,KeystoneToken,SnapManager,SnapServicesManager,ScriptExecution"
+    _manager_plugins="ProcessKiller,UserManager,ShutdownManager,HardwareInfo,KeystoneToken,SnapManager,SnapServicesManager,ScriptExecution,UbuntuProInfo"
 fi
 
 if [ -z "$_monitor_plugins" ]; then