Repository: Azure/WALinuxAgent
Pull request #1879 (2020-04-30 22:46:33)

Fix timestamp for periodic operations in the monitor thread
#1770 reset the timestamp of the tasks in the monitor thread on each iteration.

As a result, tasks with periods greater than the minimum period (e.g. perf counters, reset periodic messages) will be executed only once, at agent startup.


Patch:
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index a0e34aa9..fe6bf1d6 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -33,6 +33,7 @@ from azurelinuxagent.common.protocol.imds import get_imds_client
 from azurelinuxagent.common.utils.restutil import IOErrorCounter
 from azurelinuxagent.common.utils.textutil import hash_strings
 from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
+from azurelinuxagent.ga.periodic_operation import PeriodicOperation
 
 
 def generate_extension_metrics_telemetry_dictionary(schema_version=1.0,
@@ -51,9 +52,11 @@ def get_monitor_handler():
 
 
 class MonitorHandler(object):
+    # telemetry
     EVENT_COLLECTION_PERIOD = datetime.timedelta(minutes=1)
+    # host health
     TELEMETRY_HEARTBEAT_PERIOD = datetime.timedelta(minutes=30)
-    # extension metrics period
+    # cgroup data period
     CGROUP_TELEMETRY_POLLING_PERIOD = datetime.timedelta(minutes=5)
     CGROUP_TELEMETRY_REPORTING_PERIOD = datetime.timedelta(minutes=30)
     # host plugin
@@ -62,7 +65,8 @@ class MonitorHandler(object):
     # imds
     IMDS_HEARTBEAT_PERIOD = datetime.timedelta(minutes=1)
     IMDS_HEALTH_PERIOD = datetime.timedelta(minutes=3)
-
+    # log network configuration
+    LOG_NETWORK_CONFIGURATION_PERIOD = datetime.timedelta(minutes=1)
     # Resetting loggers period
     RESET_LOGGERS_PERIOD = datetime.timedelta(hours=12)
 
@@ -71,13 +75,14 @@ class MonitorHandler(object):
         self.imds_client = None
 
         self.event_thread = None
-        self.last_reset_loggers_time = None
-        self.last_event_collection = None
-        self.last_telemetry_heartbeat = None
-        self.last_cgroup_polling_telemetry = None
-        self.last_cgroup_report_telemetry = None
-        self.last_host_plugin_heartbeat = None
-        self.last_imds_heartbeat = None
+        self._reset_loggers_op = PeriodicOperation("reset_loggers", self.reset_loggers, self.RESET_LOGGERS_PERIOD)
+        self._collect_and_send_events_op = PeriodicOperation("collect_and_send_events", self.collect_and_send_events, self.EVENT_COLLECTION_PERIOD)
+        self._send_telemetry_heartbeat_op = PeriodicOperation("send_telemetry_heartbeat", self.send_telemetry_heartbeat, self.TELEMETRY_HEARTBEAT_PERIOD)
+        self._poll_telemetry_metrics_op = PeriodicOperation("poll_telemetry_metrics usage", self.poll_telemetry_metrics, self.CGROUP_TELEMETRY_POLLING_PERIOD)
+        self._send_telemetry_metrics_op = PeriodicOperation("send_telemetry_metrics usage", self.send_telemetry_metrics, self.CGROUP_TELEMETRY_REPORTING_PERIOD)
+        self._send_host_plugin_heartbeat_op = PeriodicOperation("send_host_plugin_heartbeat", self.send_host_plugin_heartbeat, self.HOST_PLUGIN_HEARTBEAT_PERIOD)
+        self._send_imds_heartbeat_op = PeriodicOperation("send_imds_heartbeat", self.send_imds_heartbeat, self.IMDS_HEARTBEAT_PERIOD)
+        self._log_altered_network_configuration_op = PeriodicOperation("log_altered_network_configuration", self.log_altered_network_configuration, self.LOG_NETWORK_CONFIGURATION_PERIOD)
         self.protocol = None
         self.protocol_util = None
         self.health_service = None
@@ -95,7 +100,13 @@ class MonitorHandler(object):
     def stop(self):
         self.should_run = False
         if self.is_alive():
-            self.event_thread.join()
+            self.join()
+
+    def join(self):
+        self.event_thread.join()
+
+    def stopped(self):
+        return not self.should_run
 
     def init_protocols(self):
         # The initialization of ProtocolUtil for the Monitor thread should be done within the thread itself rather
@@ -122,22 +133,10 @@ class MonitorHandler(object):
         """
         Periodically send any events located in the events folder
         """
-        try:
-            if self.last_event_collection is None:
-                self.last_event_collection = datetime.datetime.utcnow() - MonitorHandler.EVENT_COLLECTION_PERIOD
+        event_list = collect_events()
 
-            if datetime.datetime.utcnow() >= (self.last_event_collection + MonitorHandler.EVENT_COLLECTION_PERIOD):
-                try:
-                    event_list = collect_events()
-
-                    if len(event_list.events) > 0:
-                        self.protocol.report_event(event_list)
-                except Exception as e:
-                    logger.warn("{0}", ustr(e))
-        except Exception as e:
-            logger.warn("Failed to send events: {0}", ustr(e))
-
-        self.last_event_collection = datetime.datetime.utcnow()
+        if len(event_list.events) > 0:
+            self.protocol.report_event(event_list)
 
     def daemon(self, init_data=False):
 
@@ -151,19 +150,19 @@ class MonitorHandler(object):
                         MonitorHandler.EVENT_COLLECTION_PERIOD,
                         MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD,
                         MonitorHandler.IMDS_HEARTBEAT_PERIOD).seconds
-        while self.should_run:
+        while not self.stopped():
             try:
                 self.protocol.update_host_plugin_from_goal_state()
-                self.send_telemetry_heartbeat()
-                self.poll_telemetry_metrics()
+                self._send_telemetry_heartbeat_op.run()
+                self._poll_telemetry_metrics_op.run()
                 # This will be removed in favor of poll_telemetry_metrics() and it'll directly send the perf data for
                 # each cgroup.
-                self.send_telemetry_metrics()
-                self.collect_and_send_events()
-                self.send_host_plugin_heartbeat()
-                self.send_imds_heartbeat()
-                self.log_altered_network_configuration()
-                self.reset_loggers()
+                self._send_telemetry_metrics_op.run()
+                self._collect_and_send_events_op.run()
+                self._send_host_plugin_heartbeat_op.run()
+                self._send_imds_heartbeat_op.run()
+                self._log_altered_network_configuration_op.run()
+                self._reset_loggers_op.run()
             except Exception as e:
                 logger.warn("An error occurred in the monitor thread main loop; will skip the current iteration.\n{0}", ustr(e))
             time.sleep(min_delta)
@@ -174,41 +173,25 @@ class MonitorHandler(object):
         For reference, please check azurelinuxagent.common.logger.Logger and
         azurelinuxagent.common.event.EventLogger classes
         """
-        try:
-            time_now = datetime.datetime.utcnow()
-            if not self.last_reset_loggers_time:
-                self.last_reset_loggers_time = time_now
-
-            if time_now >= (self.last_reset_loggers_time + MonitorHandler.RESET_LOGGERS_PERIOD):
-                logger.reset_periodic()
-
-        except Exception as e:
-            logger.warn("Failed to clear periodic loggers: {0}", ustr(e))
-
-        self.last_reset_loggers_time = time_now
+        logger.reset_periodic()
 
     def send_imds_heartbeat(self):
         """
         Send a health signal every IMDS_HEARTBEAT_PERIOD. The signal is 'Healthy' when we have
         successfully called and validated a response in the last IMDS_HEALTH_PERIOD.
         """
-
         try:
-            if self.last_imds_heartbeat is None:
-                self.last_imds_heartbeat = datetime.datetime.utcnow() - MonitorHandler.IMDS_HEARTBEAT_PERIOD
-
-            if datetime.datetime.utcnow() >= (self.last_imds_heartbeat + MonitorHandler.IMDS_HEARTBEAT_PERIOD):
-                is_currently_healthy, response = self.imds_client.validate()
+            is_currently_healthy, response = self.imds_client.validate()
 
-                if is_currently_healthy:
-                    self.imds_errorstate.reset()
-                else:
-                    self.imds_errorstate.incr()
+            if is_currently_healthy:
+                self.imds_errorstate.reset()
+            else:
+                self.imds_errorstate.incr()
 
-                is_healthy = self.imds_errorstate.is_triggered() is False
-                logger.verbose("IMDS health: {0} [{1}]", is_healthy, response)
+            is_healthy = self.imds_errorstate.is_triggered() is False
+            logger.verbose("IMDS health: {0} [{1}]", is_healthy, response)
 
-                self.health_service.report_imds_status(is_healthy, response)
+            self.health_service.report_imds_status(is_healthy, response)
 
         except Exception as e:
             msg = "Exception sending imds heartbeat: {0}".format(ustr(e))
@@ -220,41 +203,34 @@ class MonitorHandler(object):
                 message=msg,
                 log_event=False)
 
-        self.last_imds_heartbeat = datetime.datetime.utcnow()
-
     def send_host_plugin_heartbeat(self):
         """
         Send a health signal every HOST_PLUGIN_HEARTBEAT_PERIOD. The signal is 'Healthy' when we have been able to
         communicate with HostGAPlugin at least once in the last HOST_PLUGIN_HEALTH_PERIOD.
         """
         try:
-            if self.last_host_plugin_heartbeat is None:
-                self.last_host_plugin_heartbeat = datetime.datetime.utcnow() - MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD
-
-            if datetime.datetime.utcnow() >= (
-                self.last_host_plugin_heartbeat + MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD):
-                host_plugin = self.protocol.client.get_host_plugin()
-                host_plugin.ensure_initialized()
-                is_currently_healthy = host_plugin.get_health()
-
-                if is_currently_healthy:
-                    self.host_plugin_errorstate.reset()
-                else:
-                    self.host_plugin_errorstate.incr()
-
-                is_healthy = self.host_plugin_errorstate.is_triggered() is False
-                logger.verbose("HostGAPlugin health: {0}", is_healthy)
-
-                self.health_service.report_host_plugin_heartbeat(is_healthy)
-
-                if not is_healthy:
-                    add_event(
-                        name=AGENT_NAME,
-                        version=CURRENT_VERSION,
-                        op=WALAEventOperation.HostPluginHeartbeatExtended,
-                        is_success=False,
-                        message='{0} since successful heartbeat'.format(self.host_plugin_errorstate.fail_time),
-                        log_event=False)
+            host_plugin = self.protocol.client.get_host_plugin()
+            host_plugin.ensure_initialized()
+            is_currently_healthy = host_plugin.get_health()
+
+            if is_currently_healthy:
+                self.host_plugin_errorstate.reset()
+            else:
+                self.host_plugin_errorstate.incr()
+
+            is_healthy = self.host_plugin_errorstate.is_triggered() is False
+            logger.verbose("HostGAPlugin health: {0}", is_healthy)
+
+            self.health_service.report_host_plugin_heartbeat(is_healthy)
+
+            if not is_healthy:
+                add_event(
+                    name=AGENT_NAME,
+                    version=CURRENT_VERSION,
+                    op=WALAEventOperation.HostPluginHeartbeatExtended,
+                    is_success=False,
+                    message='{0} since successful heartbeat'.format(self.host_plugin_errorstate.fail_time),
+                    log_event=False)
 
         except Exception as e:
             msg = "Exception sending host plugin heartbeat: {0}".format(ustr(e))
@@ -266,33 +242,22 @@ class MonitorHandler(object):
                 message=msg,
                 log_event=False)
 
-            self.last_host_plugin_heartbeat = datetime.datetime.utcnow()
-
     def send_telemetry_heartbeat(self):
-        try:
-            if self.last_telemetry_heartbeat is None:
-                self.last_telemetry_heartbeat = datetime.datetime.utcnow() - MonitorHandler.TELEMETRY_HEARTBEAT_PERIOD
-
-            if datetime.datetime.utcnow() >= (self.last_telemetry_heartbeat + MonitorHandler.TELEMETRY_HEARTBEAT_PERIOD):
-                io_errors = IOErrorCounter.get_and_reset()
-                hostplugin_errors = io_errors.get("hostplugin")
-                protocol_errors = io_errors.get("protocol")
-                other_errors = io_errors.get("other")
-
-                if hostplugin_errors > 0 or protocol_errors > 0 or other_errors > 0:
-                    msg = "hostplugin:{0};protocol:{1};other:{2}".format(hostplugin_errors, protocol_errors,
-                                                                         other_errors)
-                    add_event(
-                        name=AGENT_NAME,
-                        version=CURRENT_VERSION,
-                        op=WALAEventOperation.HttpErrors,
-                        is_success=True,
-                        message=msg,
-                        log_event=False)
-        except Exception as e:
-            logger.warn("Failed to send heartbeat: {0}", ustr(e))
-
-        self.last_telemetry_heartbeat = datetime.datetime.utcnow()
+        io_errors = IOErrorCounter.get_and_reset()
+        hostplugin_errors = io_errors.get("hostplugin")
+        protocol_errors = io_errors.get("protocol")
+        other_errors = io_errors.get("other")
+
+        if hostplugin_errors > 0 or protocol_errors > 0 or other_errors > 0:
+            msg = "hostplugin:{0};protocol:{1};other:{2}".format(hostplugin_errors, protocol_errors,
+                                                                 other_errors)
+            add_event(
+                name=AGENT_NAME,
+                version=CURRENT_VERSION,
+                op=WALAEventOperation.HttpErrors,
+                is_success=True,
+                message=msg,
+                log_event=False)
 
     def poll_telemetry_metrics(self):
         """
@@ -300,51 +265,27 @@ class MonitorHandler(object):
 
         :return: List of Metrics (which would be sent to PerfCounterMetrics directly.
         """
-        try:  # If there is an issue in reporting, it should not take down whole monitor thread.
-            time_now = datetime.datetime.utcnow()
-            if not self.last_cgroup_polling_telemetry:
-                self.last_cgroup_polling_telemetry = time_now
-
-            if time_now >= (self.last_cgroup_polling_telemetry +
-                            MonitorHandler.CGROUP_TELEMETRY_POLLING_PERIOD):
-                metrics = CGroupsTelemetry.poll_all_tracked()
-
-                if metrics:
-                    for metric in metrics:
-                        report_metric(metric.category, metric.counter, metric.instance, metric.value)
-        except Exception as e:
-            logger.warn("Could not poll all the tracked telemetry due to {0}", ustr(e))
+        metrics = CGroupsTelemetry.poll_all_tracked()
 
-        self.last_cgroup_polling_telemetry = datetime.datetime.utcnow()
+        if metrics:
+            for metric in metrics:
+                report_metric(metric.category, metric.counter, metric.instance, metric.value)
 
     def send_telemetry_metrics(self):
         """
         The send_telemetry_metrics would soon be removed in favor of sending performance metrics directly.
-
-        :return:
         """
-        try:  # If there is an issue in reporting, it should not take down whole monitor thread.
-            time_now = datetime.datetime.utcnow()
-
-            if not self.last_cgroup_report_telemetry:
-                self.last_cgroup_report_telemetry = time_now
-
-            if time_now >= (self.last_cgroup_report_telemetry + MonitorHandler.CGROUP_TELEMETRY_REPORTING_PERIOD):
-                performance_metrics = CGroupsTelemetry.report_all_tracked()
-
-                if performance_metrics:
-                    message = generate_extension_metrics_telemetry_dictionary(schema_version=1.0,
-                                                                              performance_metrics=performance_metrics)
-                    add_event(name=AGENT_NAME,
-                              version=CURRENT_VERSION,
-                              op=WALAEventOperation.ExtensionMetricsData,
-                              is_success=True,
-                              message=ustr(message),
-                              log_event=False)
-        except Exception as e:
-            logger.warn("Could not report all the tracked telemetry due to {0}", ustr(e))
+        performance_metrics = CGroupsTelemetry.report_all_tracked()
 
-        self.last_cgroup_report_telemetry = datetime.datetime.utcnow()
+        if performance_metrics:
+            message = generate_extension_metrics_telemetry_dictionary(schema_version=1.0,
+                                                                      performance_metrics=performance_metrics)
+            add_event(name=AGENT_NAME,
+                      version=CURRENT_VERSION,
+                      op=WALAEventOperation.ExtensionMetricsData,
+                      is_success=True,
+                      message=ustr(message),
+                      log_event=False)
 
     def log_altered_network_configuration(self):
         """
diff --git a/azurelinuxagent/ga/periodic_operation.py b/azurelinuxagent/ga/periodic_operation.py
new file mode 100644
index 00000000..1135586e
--- /dev/null
+++ b/azurelinuxagent/ga/periodic_operation.py
@@ -0,0 +1,56 @@
+# Copyright 2020 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+#
+
+import datetime
+
+from azurelinuxagent.common import logger
+from azurelinuxagent.common.future import ustr
+
+
+class PeriodicOperation(object):
+    '''
+    Instances of PeriodicOperation are tasks that are executed only after the given
+    time period has elapsed.
+
+    NOTE: the run() method catches any exceptions raised by the operation and logs them as warnings.
+    '''
+
+    # To prevent flooding the log with error messages we report failures at most every hour
+    _LOG_WARNING_PERIOD = datetime.timedelta(minutes=60)
+
+    def __init__(self, name, operation, period):
+        self._name = name
+        self._operation = operation
+        self._period = period
+        self._last_run = None
+        self._last_warning = None
+        self._last_warning_time = None
+
+    def run(self):
+        try:
+            if self._last_run is None or datetime.datetime.utcnow() >= self._last_run + self._period:
+                try:
+                    self._operation()
+                finally:
+                    self._last_run = datetime.datetime.utcnow()
+        except Exception as e:
+            warning = "Failed to {0}: {1} --- [NOTE: Will not log the same error for the next hour]".format(self._name, ustr(e))
+            if warning != self._last_warning or self._last_warning_time is None or datetime.datetime.utcnow() >= self._last_warning_time + self._LOG_WARNING_PERIOD:
+                logger.warn(warning)
+                self._last_warning_time = datetime.datetime.utcnow()
+                self._last_warning = warning
+


Test patch:
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index 51b8c194..de10424d 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -15,6 +15,7 @@
 # Requires Python 2.6+ and Openssl 1.0+
 #
 import datetime
+import contextlib
 import json
 import os
 import platform
@@ -24,14 +25,13 @@ import tempfile
 import time
 from datetime import timedelta
 
-from azurelinuxagent.common.protocol.util import ProtocolUtil, get_protocol_util
-from nose.plugins.attrib import attr
+from azurelinuxagent.common.protocol.util import ProtocolUtil
 
 from azurelinuxagent.common import event, logger
 from azurelinuxagent.common.cgroup import CGroup, CpuCgroup, MemoryCgroup
 from azurelinuxagent.common.cgroupstelemetry import CGroupsTelemetry, MetricValue
 from azurelinuxagent.common.datacontract import get_properties
-from azurelinuxagent.common.event import WALAEventOperation, EVENTS_DIRECTORY
+from azurelinuxagent.common.event import add_event, WALAEventOperation, EVENTS_DIRECTORY
 from azurelinuxagent.common.exception import HttpError
 from azurelinuxagent.common.logger import Logger
 from azurelinuxagent.common.osutil import get_osutil
@@ -39,34 +39,53 @@ from azurelinuxagent.common.protocol.wire import WireProtocol
 from azurelinuxagent.common.telemetryevent import TelemetryEvent, TelemetryEventParam
 from azurelinuxagent.common.utils import fileutil, restutil
 from azurelinuxagent.common.version import AGENT_VERSION, CURRENT_VERSION, CURRENT_AGENT, DISTRO_NAME, DISTRO_VERSION, DISTRO_CODE_NAME
-from azurelinuxagent.ga.monitor import generate_extension_metrics_telemetry_dictionary, get_monitor_handler, MonitorHandler
+from azurelinuxagent.ga.monitor import generate_extension_metrics_telemetry_dictionary, get_monitor_handler, MonitorHandler, PeriodicOperation
 from tests.protocol.mockwiredata import DATA_FILE
-from tests.protocol.mocks import mock_wire_protocol
+from tests.protocol.mocks import mock_wire_protocol, HttpRequestPredicates, MockHttpResponse
 from tests.tools import Mock, MagicMock, patch, AgentTestCase, clear_singleton_instances, PropertyMock
 from tests.utils.event_logger_tools import EventLoggerTools
 
 
-class ResponseMock(Mock):
-    def __init__(self, status=restutil.httpclient.OK, response=None, reason=None):
-        Mock.__init__(self)
-        self.status = status
-        self.reason = reason
-        self.response = response
+def random_generator(size=6, chars=string.ascii_uppercase + string.digits + string.ascii_lowercase):
+    return ''.join(random.choice(chars) for x in range(size))
 
-    def read(self):
-        return self.response
+@contextlib.contextmanager
+def _create_monitor_handler(enabled_operations=[], iterations=1):
+    """
+    Creates an instance of MonitorHandler that
+        * Uses a mock_wire_protocol for network requests,
+        * Executes only the operations given in the 'enabled_operations' parameter,
+        * Runs its main loop only the number of times given in the 'iterations' parameter, and
+        * Does not sleep at the end of each iteration
+
+    The returned MonitorHandler is augmented with 2 methods:
+        * get_mock_wire_protocol() - returns the mock protocol
+        * run_and_wait() - invokes run() and wait() on the MonitorHandler
+
+    """
+    def run(self):
+        if len(enabled_operations) == 0 or self._name in enabled_operations:
+            run.original_definition(self)
+    run.original_definition = PeriodicOperation.run
+
+    with mock_wire_protocol(DATA_FILE) as protocol:
+        protocol_util = MagicMock()
+        protocol_util.get_protocol = Mock(return_value=protocol)
+        with patch("azurelinuxagent.ga.monitor.get_protocol_util", return_value=protocol_util):
+            with patch.object(PeriodicOperation, "run", side_effect=run, autospec=True):
+                with patch("azurelinuxagent.ga.monitor.MonitorHandler.stopped", side_effect=[False] * iterations + [True]):
+                    with patch("time.sleep"):
+                        def run_and_wait():
+                            monitor_handler.run()
+                            monitor_handler.join()
 
+                        monitor_handler = get_monitor_handler()
+                        monitor_handler.get_mock_wire_protocol = lambda: protocol
+                        monitor_handler.run_and_wait = run_and_wait
+                        yield monitor_handler
 
-def random_generator(size=6, chars=string.ascii_uppercase + string.digits + string.ascii_lowercase):
-    return ''.join(random.choice(chars) for x in range(size))
 
-@patch('azurelinuxagent.common.event.EventLogger.add_event')
-@patch('azurelinuxagent.common.osutil.get_osutil')
-@patch('azurelinuxagent.common.protocol.util.get_protocol_util')
-@patch('azurelinuxagent.common.protocol.util.ProtocolUtil.get_protocol')
-@patch("azurelinuxagent.common.protocol.healthservice.HealthService._report")
-@patch("azurelinuxagent.common.utils.restutil.http_get")
-class TestMonitor(AgentTestCase):
+class TestMonitor(AgentTestCase, HttpRequestPredicates):
     def setUp(self):
         AgentTestCase.setUp(self)
         prefix = "UnitTest"
@@ -79,203 +98,72 @@ class TestMonitor(AgentTestCase):
     def tearDown(self):
         AgentTestCase.tearDown(self)
 
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_heartbeat")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.collect_and_send_events")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_host_plugin_heartbeat")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.poll_telemetry_metrics")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_metrics")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_imds_heartbeat")
-    def test_heartbeats(self,
-                        patch_imds_heartbeat,
-                        patch_send_telemetry_metrics,
-                        patch_poll_telemetry_metrics,
-                        patch_hostplugin_heartbeat,
-                        patch_send_events,
-                        patch_telemetry_heartbeat,
-                        *args):
-        monitor_handler = get_monitor_handler()
-
-        MonitorHandler.TELEMETRY_HEARTBEAT_PERIOD = timedelta(milliseconds=100)
-        MonitorHandler.EVENT_COLLECTION_PERIOD = timedelta(milliseconds=100)
-        MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD = timedelta(milliseconds=100)
-        MonitorHandler.IMDS_HEARTBEAT_PERIOD = timedelta(milliseconds=100)
-
-        self.assertEqual(0, patch_hostplugin_heartbeat.call_count)
-        self.assertEqual(0, patch_send_events.call_count)
-        self.assertEqual(0, patch_telemetry_heartbeat.call_count)
-        self.assertEqual(0, patch_imds_heartbeat.call_count)
-        self.assertEqual(0, patch_send_telemetry_metrics.call_count)
-        self.assertEqual(0, patch_poll_telemetry_metrics.call_count)
-
-        with patch.object(monitor_handler, 'protocol'):
-            monitor_handler.start()
-            time.sleep(1)
-            self.assertTrue(monitor_handler.is_alive())
-
-            self.assertNotEqual(0, patch_hostplugin_heartbeat.call_count)
-            self.assertNotEqual(0, patch_send_events.call_count)
-            self.assertNotEqual(0, patch_telemetry_heartbeat.call_count)
-            self.assertNotEqual(0, patch_imds_heartbeat.call_count)
-            self.assertNotEqual(0, patch_send_telemetry_metrics.call_count)
-            self.assertNotEqual(0, patch_poll_telemetry_metrics.call_count)
-
-            monitor_handler.stop()
-
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_metrics")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.poll_telemetry_metrics")
-    def test_heartbeat_timings_updates_after_window(self, *args):
-        monitor_handler = get_monitor_handler()
-
-        MonitorHandler.TELEMETRY_HEARTBEAT_PERIOD = timedelta(milliseconds=100)
-        MonitorHandler.EVENT_COLLECTION_PERIOD = timedelta(milliseconds=100)
-        MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD = timedelta(milliseconds=100)
-        MonitorHandler.IMDS_HEARTBEAT_PERIOD = timedelta(milliseconds=100)
+    def test_it_should_invoke_all_periodic_operations(self):
+        invoked_operations = []
 
-        self.assertEqual(None, monitor_handler.last_host_plugin_heartbeat)
-        self.assertEqual(None, monitor_handler.last_event_collection)
-        self.assertEqual(None, monitor_handler.last_telemetry_heartbeat)
-        self.assertEqual(None, monitor_handler.last_imds_heartbeat)
+        with _create_monitor_handler() as monitor_handler:
+            def mock_run(self):
+                invoked_operations.append(self._name)
 
-        with patch.object(monitor_handler, 'protocol'):
-            monitor_handler.start()
-            time.sleep(0.2)
-            self.assertTrue(monitor_handler.is_alive())
+            with patch.object(PeriodicOperation, "run", side_effect=mock_run, spec=MonitorHandler.run):
+                monitor_handler.run_and_wait()
 
-            self.assertNotEqual(None, monitor_handler.last_host_plugin_heartbeat)
-            self.assertNotEqual(None, monitor_handler.last_event_collection)
-            self.assertNotEqual(None, monitor_handler.last_telemetry_heartbeat)
-            self.assertNotEqual(None, monitor_handler.last_imds_heartbeat)
+                expected_operations = [
+                    "reset_loggers", "collect_and_send_events", "send_telemetry_heartbeat",
+                    "poll_telemetry_metrics usage", "send_telemetry_metrics usage", "send_host_plugin_heartbeat",
+                    "send_imds_heartbeat", "log_altered_network_configuration"
+                ]
 
-            heartbeat_hostplugin = monitor_handler.last_host_plugin_heartbeat
-            heartbeat_imds = monitor_handler.last_imds_heartbeat
-            heartbeat_telemetry = monitor_handler.last_telemetry_heartbeat
-            events_collection = monitor_handler.last_event_collection
+                self.assertEqual(invoked_operations.sort(), expected_operations.sort(), "The monitor thread did not invoke the expected operations")
 
-            time.sleep(0.5)
+    def test_it_should_report_host_ga_health(self):
+        with _create_monitor_handler(enabled_operations=["send_host_plugin_heartbeat"]) as monitor_handler:
+            def http_post_handler(url, _, **__):
+                if self.is_health_service_request(url):
+                    http_post_handler.health_service_posted = True
+                    return MockHttpResponse(status=200)
+                return None
+            http_post_handler.health_service_posted = False
 
-            self.assertNotEqual(heartbeat_imds, monitor_handler.last_imds_heartbeat)
-            self.assertNotEqual(heartbeat_hostplugin, monitor_handler.last_host_plugin_heartbeat)
-            self.assertNotEqual(events_collection, monitor_handler.last_event_collection)
-            self.assertNotEqual(heartbeat_telemetry, monitor_handler.last_telemetry_heartbeat)
+            monitor_handler.get_mock_wire_protocol().set_http_handlers(http_post_handler=http_post_handler)
 
-            monitor_handler.stop()
-
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.send_telemetry_metrics")
-    @patch("azurelinuxagent.ga.monitor.MonitorHandler.poll_telemetry_metrics")
-    def test_heartbeat_timings_no_updates_within_window(self, *args):
-        monitor_handler = get_monitor_handler()
-
-        MonitorHandler.TELEMETRY_HEARTBEAT_PERIOD = timedelta(seconds=1)
-        MonitorHandler.EVENT_COLLECTION_PERIOD = timedelta(seconds=1)
-        MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD = timedelta(seconds=1)
-        MonitorHandler.IMDS_HEARTBEAT_PERIOD = timedelta(seconds=1)
-
-        self.assertEqual(None, monitor_handler.last_host_plugin_heartbeat)
-        self.assertEqual(None, monitor_handler.last_event_collection)
-        self.assertEqual(None, monitor_handler.last_telemetry_heartbeat)
-        self.assertEqual(None, monitor_handler.last_imds_heartbeat)
+            monitor_handler.run_and_wait()
 
-        with patch.object(monitor_handler, 'protocol'):
-            monitor_handler.start()
-            time.sleep(0.2)
-            self.assertTrue(monitor_handler.is_alive())
+            self.assertTrue(http_post_handler.health_service_posted, "The monitor thread did not report host ga plugin health")
 
-            self.assertNotEqual(None, monitor_handler.last_host_plugin_heartbeat)
-            self.assertNotEqual(None, monitor_handler.last_event_collection)
-            self.assertNotEqual(None, monitor_handler.last_telemetry_heartbeat)
-            self.assertNotEqual(None, monitor_handler.last_imds_heartbeat)
+    def test_it_should_report_a_telemetry_event_when_host_plugin_is_not_healthy(self):
+        with _create_monitor_handler(enabled_operations=["send_host_plugin_heartbeat"]) as monitor_handler:
+            def http_get_handler(url, *_, **__):
+                if self.is_host_plugin_health_request(url):
+                    return MockHttpResponse(status=503)
+                return None
 
-            heartbeat_hostplugin = monitor_handler.last_host_plugin_heartbeat
-            heartbeat_imds = monitor_handler.last_imds_heartbeat
-            heartbeat_telemetry = monitor_handler.last_telemetry_heartbeat
-            events_collection = monitor_handler.last_event_collection
+            monitor_handler.get_mock_wire_protocol().set_http_handlers(http_get_handler=http_get_handler)
 
-            time.sleep(0.5)
+            # the error triggers only after ERROR_STATE_DELTA_DEFAULT
+            with patch('azurelinuxagent.common.errorstate.ErrorState.is_triggered', return_value=True):
+                with patch('azurelinuxagent.common.event.EventLogger.add_event') as add_event_patcher:
+                    monitor_handler.run_and_wait()
 
-            self.assertEqual(heartbeat_hostplugin, monitor_handler.last_host_plugin_heartbeat)
-            self.assertEqual(heartbeat_imds, monitor_handler.last_imds_heartbeat)
-            self.assertEqual(events_collection, monitor_handler.last_event_collection)
-            self.assertEqual(heartbeat_telemetry, monitor_handler.last_telemetry_heartbeat)
-
-            monitor_handler.stop()
-
-    @patch("azurelinuxagent.common.protocol.healthservice.HealthService.report_host_plugin_heartbeat")
-    def test_heartbeat_creates_signal(self, patch_report_heartbeat, *args):
-        monitor_handler = get_monitor_handler()
-        monitor_handler.init_protocols()
-        monitor_handler.last_host_plugin_heartbeat = datetime.datetime.utcnow() - timedelta(hours=1)
-        monitor_handler.send_host_plugin_heartbeat()
-        self.assertEqual(1, patch_report_heartbeat.call_count)
-        self.assertEqual(0, args[5].call_count)
-        monitor_handler.stop()
-
-    @patch('azurelinuxagent.common.errorstate.ErrorState.is_triggered', return_value=True)
-    @patch("azurelinuxagent.common.protocol.healthservice.HealthService.report_host_plugin_heartbeat")
-    def test_failed_heartbeat_creates_telemetry(self, patch_report_heartbeat, _, *args):
-        monitor_handler = get_monitor_handler()
-        monitor_handler.init_protocols()
-        monitor_handler.last_host_plugin_heartbeat = datetime.datetime.utcnow() - timedelta(hours=1)
-        monitor_handler.send_host_plugin_heartbeat()
-        self.assertEqual(1, patch_report_heartbeat.call_count)
-        self.assertEqual(1, args[5].call_count)
-        self.assertEqual('HostPluginHeartbeatExtended', args[5].call_args[1]['op'])
-        self.assertEqual(False, args[5].call_args[1]['is_success'])
-        monitor_handler.stop()
+                    heartbeat_events = [kwargs for _, kwargs in add_event_patcher.call_args_list if kwargs['op'] == 'HostPluginHeartbeatExtended']
+                    self.assertTrue(len(heartbeat_events) == 1, "The monitor thread should have reported exactly 1 telemetry event for an unhealthy host ga plugin")
+                    self.assertFalse(heartbeat_events[0]['is_success'], 'The reported event should indicate failure')
 
-    @patch('azurelinuxagent.common.logger.Logger.info')
-    def test_reset_loggers(self, mock_info, *args):
+    def test_it_should_clear_periodic_log_messages(self):
         # Adding 100 different messages
         for i in range(100):
-            event_message = "Test {0}".format(i)
-            logger.periodic_info(logger.EVERY_DAY, event_message)
+            logger.periodic_info(logger.EVERY_DAY, "Test {0}".format(i))
 
-            self.assertIn(hash(event_message), logger.DEFAULT_LOGGER.periodic_messages)
-            self.assertEqual(i + 1, mock_info.call_count)  # range starts from 0.
+        if len(logger.DEFAULT_LOGGER.periodic_messages) != 100:
+            raise Exception('Test setup error: the periodic messages were not added')
 
-        self.assertEqual(100, len(logger.DEFAULT_LOGGER.periodic_messages))
+        with _create_monitor_handler(enabled_operations=["reset_loggers"]) as monitor_handler:
+            monitor_handler.run_and_wait()
 
-        # Adding 1 message 100 times, but the same message. Mock Info should be called only once.
-        for i in range(100):
-            logger.periodic_info(logger.EVERY_DAY, "Test-Message")
-
-        self.assertIn(hash("Test-Message"), logger.DEFAULT_LOGGER.periodic_messages)
-        self.assertEqual(101, mock_info.call_count)  # 100 calls from the previous section. Adding only 1.
-        self.assertEqual(101, len(logger.DEFAULT_LOGGER.periodic_messages))  # One new message in the hash map.
-
-        # Resetting the logger time states.
-        monitor_handler = get_monitor_handler()
-        monitor_handler.last_reset_loggers_time = datetime.datetime.utcnow() - timedelta(hours=1)
-        MonitorHandler.RESET_LOGGERS_PERIOD = timedelta(milliseconds=100)
-
-        monitor_handler.reset_loggers()
+            self.assertEqual(len(logger.DEFAULT_LOGGER.periodic_messages), 0, "The monitor thread did not reset the periodic log messages")
 
-        # The hash map got cleaned up by the reset_loggers method
-        self.assertEqual(0, len(logger.DEFAULT_LOGGER.periodic_messages))
 
-        monitor_handler.stop()
-
-    @patch("azurelinuxagent.common.logger.reset_periodic", side_effect=Exception())
-    def test_reset_loggers_ensuring_timestamp_gets_updated(self, *args):
-        # Resetting the logger time states.
-        monitor_handler = get_monitor_handler()
-        initial_time = datetime.datetime.utcnow() - timedelta(hours=1)
-        monitor_handler.last_reset_loggers_time = initial_time
-        MonitorHandler.RESET_LOGGERS_PERIOD = timedelta(milliseconds=100)
-
-        # noinspection PyBroadException
-        try:
-            monitor_handler.reset_loggers()
-        except:
-            pass
-
-        # The hash map got cleaned up by the reset_loggers method
-        self.assertGreater(monitor_handler.last_reset_loggers_time, initial_time)
-        monitor_handler.stop()
-
-
-@patch('azurelinuxagent.common.osutil.get_osutil')
-@patch("azurelinuxagent.common.protocol.healthservice.HealthService._report")
-class TestEventMonitoring(AgentTestCase):
+class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
     def setUp(self):
         AgentTestCase.setUp(self)
         self.lib_dir = tempfile.mkdtemp()
@@ -286,15 +174,6 @@ class TestEventMonitoring(AgentTestCase):
     def tearDown(self):
         fileutil.rm_dirs(self.lib_dir)
 
-    @staticmethod
-    def _create_monitor_handler(protocol):
-        monitor_handler = get_monitor_handler()
-        protocol_util = get_protocol_util()
-        protocol_util.get_protocol = Mock(return_value=protocol)
-        monitor_handler.protocol_util = Mock(return_value=protocol_util)
-        monitor_handler.init_protocols()
-        return monitor_handler
-
     def _create_extension_event(self,
                                size=0,
                                name="DummyExtension",
@@ -331,12 +210,9 @@ class TestEventMonitoring(AgentTestCase):
     def test_collect_and_send_events(self, mock_lib_dir, patch_send_event, *_):
         mock_lib_dir.return_value = self.lib_dir
 
-        with mock_wire_protocol(DATA_FILE) as protocol:
-            monitor_handler = TestEventMonitoring._create_monitor_handler(protocol)
-
+        with _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
             self._create_extension_event(message="Message-Test")
 
-            monitor_handler.last_event_collection = None
             test_mtime = 1000  # epoch time, in ms
             test_opcodename = datetime.datetime.fromtimestamp(test_mtime).strftime(u'%Y-%m-%dT%H:%M:%S.%fZ')
             test_eventtid = 42
@@ -347,11 +223,11 @@ class TestEventMonitoring(AgentTestCase):
                 with patch('os.getpid', return_value=test_eventpid):
                     with patch("threading.Thread.ident", new_callable=PropertyMock(return_value=test_eventtid)):
                         with patch("threading.Thread.getName", return_value=test_taskname):
-                            monitor_handler.collect_and_send_events()
+                            monitor_handler.run_and_wait()
 
             # Validating the crafted message by the collect_and_send_events call.
             self.assertEqual(1, patch_send_event.call_count)
-            send_event_call_args = protocol.client.send_event.call_args[0]
+            send_event_call_args = monitor_handler.get_mock_wire_protocol().client.send_event.call_args[0]
 
             # Some of those expected values come from the mock protocol and imds client set up during test initialization
             osutil = get_osutil()
@@ -398,15 +274,15 @@ class TestEventMonitoring(AgentTestCase):
     def test_collect_and_send_events_with_small_events(self, mock_lib_dir, patch_send_event, *_):
         mock_lib_dir.return_value = self.lib_dir
 
-        with mock_wire_protocol(DATA_FILE) as protocol:
-            monitor_handler = TestEventMonitoring._create_monitor_handler(protocol)
+        with _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
 
             sizes = [15, 15, 15, 15]  # get the powers of 2 - 2**16 is the limit
 
             for power in sizes:
                 size = 2 ** power
                 self._create_extension_event(size)
-            monitor_handler.collect_and_send_events()
+
+            monitor_handler.run_and_wait()
 
             # The send_event call would be called each time, as we are filling up the buffer up to the brim for each call.
 
@@ -417,8 +293,7 @@ class TestEventMonitoring(AgentTestCase):
     def test_collect_and_send_events_with_large_events(self, mock_lib_dir, patch_send_event, *_):
         mock_lib_dir.return_value = self.lib_dir
 
-        with mock_wire_protocol(DATA_FILE) as protocol:
-            monitor_handler = TestEventMonitoring._create_monitor_handler(protocol)
+        with _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
 
             sizes = [17, 17, 17]  # get the powers of 2
 
@@ -427,19 +302,26 @@ class TestEventMonitoring(AgentTestCase):
                 self._create_extension_event(size)
 
             with patch("azurelinuxagent.common.logger.periodic_warn") as patch_periodic_warn:
-                monitor_handler.collect_and_send_events()
+                monitor_handler.run_and_wait()
+
                 self.assertEqual(3, patch_periodic_warn.call_count)
 
-            # The send_event call should never be called as the events are larger than 2**16.
-            self.assertEqual(0, patch_send_event.call_count)
+                # The send_event call should never be called as the events are larger than 2**16.
+                self.assertEqual(0, patch_send_event.call_count)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_collect_and_send_with_http_post_returning_503(self, mock_lib_dir, *_):
         mock_lib_dir.return_value = self.lib_dir
         fileutil.mkdir(self.event_dir)
 
-        with mock_wire_protocol(DATA_FILE) as protocol:
-            monitor_handler = TestEventMonitoring._create_monitor_handler(protocol)
+        with _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
+            def http_post_handler(url, _, **__):
+                if self.is_telemetry_request(url):
+                    return MockHttpResponse(restutil.httpclient.SERVICE_UNAVAILABLE)
+                return None
+
+            protocol = monitor_handler.get_mock_wire_protocol()
+            protocol.set_http_handlers(http_post_handler=http_post_handler)
 
             sizes = [1, 2, 3]  # get the powers of 2, and multiple by 1024.
 
@@ -448,37 +330,30 @@ class TestEventMonitoring(AgentTestCase):
                 self._create_extension_event(size)
 
             with patch("azurelinuxagent.common.logger.warn") as mock_warn:
-                with patch("azurelinuxagent.common.utils.restutil.http_post") as mock_http_post:
-                    mock_http_post.return_value = ResponseMock(
-                        status=restutil.httpclient.SERVICE_UNAVAILABLE,
-                        response="")
-                    monitor_handler.collect_and_send_events()
-                    self.assertEqual(1, mock_warn.call_count)
-                    self.assertEqual("[ProtocolError] [Wireserver Exception] [ProtocolError] [Wireserver Failed] "
-                                     "URI http://{0}/machine?comp=telemetrydata  [HTTP Failed] Status Code 503".format(protocol.get_endpoint()),
-                                     mock_warn.call_args[0][1])
-                    self.assertEqual(0, len(os.listdir(self.event_dir)))
+                monitor_handler.run_and_wait()
+                self.assertEqual(1, mock_warn.call_count)
+                message = "[ProtocolError] [Wireserver Exception] [ProtocolError] [Wireserver Failed] URI http://{0}/machine?comp=telemetrydata  [HTTP Failed] Status Code 503".format(protocol.get_endpoint())
+                self.assertIn(message, mock_warn.call_args[0][0])
+                self.assertEqual(0, len(os.listdir(self.event_dir)))
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_collect_and_send_with_send_event_generating_exception(self, mock_lib_dir, *args):
         mock_lib_dir.return_value = self.lib_dir
         fileutil.mkdir(self.event_dir)
 
-        with mock_wire_protocol(DATA_FILE) as protocol:
-            monitor_handler = TestEventMonitoring._create_monitor_handler(protocol)
-
+        with _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
             sizes = [1, 2, 3]  # get the powers of 2, and multiple by 1024.
 
             for power in sizes:
                 size = 2 ** power * 1024
                 self._create_extension_event(size)
 
-            monitor_handler.last_event_collection = datetime.datetime.utcnow() - timedelta(hours=1)
             # This test validates that if we hit an issue while sending an event, we never send it again.
             with patch("azurelinuxagent.common.logger.warn") as mock_warn:
                 with patch("azurelinuxagent.common.protocol.wire.WireClient.send_event") as patch_send_event:
                     patch_send_event.side_effect = Exception()
-                    monitor_handler.collect_and_send_events()
+
+                    monitor_handler.run_and_wait()
 
                     self.assertEqual(1, mock_warn.call_count)
                     self.assertEqual(0, len(os.listdir(self.event_dir)))
@@ -487,24 +362,21 @@ class TestEventMonitoring(AgentTestCase):
     def test_collect_and_send_with_call_wireserver_returns_http_error(self, mock_lib_dir, *args):
         mock_lib_dir.return_value = self.lib_dir
         fileutil.mkdir(self.event_dir)
+        add_event(name="MonitorTests", op=WALAEventOperation.HeartBeat, is_success=True, message="Test heartbeat")
 
-        with mock_wire_protocol(DATA_FILE) as protocol:
-            monitor_handler = TestEventMonitoring._create_monitor_handler(protocol)
-
-            sizes = [1, 2, 3]  # get the powers of 2, and multiple by 1024.
+        with _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
+            def http_post_handler(url, _, **__):
+                if self.is_telemetry_request(url):
+                    return HttpError("A test exception")
+                return None
 
-            for power in sizes:
-                size = 2 ** power * 1024
-                self._create_extension_event(size)
+            monitor_handler.get_mock_wire_protocol().set_http_handlers(http_post_handler=http_post_handler)
 
-            monitor_handler.last_event_collection = datetime.datetime.utcnow() - timedelta(hours=1)
             with patch("azurelinuxagent.common.logger.warn") as mock_warn:
-                with patch("azurelinuxagent.common.protocol.wire.WireClient.call_wireserver") as patch_call_wireserver:
-                    patch_call_wireserver.side_effect = HttpError
-                    monitor_handler.collect_and_send_events()
+                monitor_handler.run_and_wait()
 
-                    self.assertEqual(1, mock_warn.call_count)
-                    self.assertEqual(0, len(os.listdir(self.event_dir)))
+                self.assertEqual(1, mock_warn.call_count)
+                self.assertEqual(0, len(os.listdir(self.event_dir)))
 
 
 @patch('azurelinuxagent.common.osutil.get_osutil')
diff --git a/tests/ga/test_periodic_operation.py b/tests/ga/test_periodic_operation.py
new file mode 100644
index 00000000..80a95ba7
--- /dev/null
+++ b/tests/ga/test_periodic_operation.py
@@ -0,0 +1,100 @@
+# Copyright 2020 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.6+ and Openssl 1.0+
+#
+import datetime
+import time
+from azurelinuxagent.ga.monitor import PeriodicOperation
+from tests.tools import AgentTestCase, patch, PropertyMock
+
+
+class TestPeriodicOperation(AgentTestCase):
+    def test_it_should_be_invoked_when_run_is_called_first_time(self):
+        def operation():
+            operation.invoked = True
+        operation.invoked = False
+
+        PeriodicOperation("test_operation", operation, period=datetime.timedelta(hours=1)).run()
+
+        self.assertTrue(operation.invoked, "The operation was not invoked")
+
+    def test_it_should_not_be_invoked_if_the_period_has_not_elapsed(self):
+        def operation():
+            operation.invoked_count += 1
+        operation.invoked_count = 0
+
+        pop = PeriodicOperation("test_operation", operation, period=datetime.timedelta(hours=1))
+        for _ in range(5):
+            pop.run()
+
+        # the first run() invoked the operation, so the count is 1
+        self.assertEqual(operation.invoked_count, 1, "The operation was invoked before the period elapsed")
+
+    def test_it_should_be_invoked_if_the_period_has_elapsed(self):
+        def operation():
+            operation.invoked_count += 1
+        operation.invoked_count = 0
+
+        pop = PeriodicOperation("test_operation", operation, period=datetime.timedelta(milliseconds=1))
+        for _ in range(5):
+            pop.run()
+            time.sleep(0.001)
+
+        self.assertEqual(operation.invoked_count, 5, "The operation was not invoked after the period elapsed")
+
+    @staticmethod
+    def _operation_with_failure():
+        raise Exception("A test exception")
+
+    @staticmethod
+    def _get_number_of_warnings(warn_patcher, message="A test exception"):
+        return len([args for args, _ in warn_patcher.call_args_list if any(message in a for a in args)])
+
+    def test_it_should_log_a_warning_if_the_operation_fails(self):
+        with patch("azurelinuxagent.common.logger.warn") as warn_patcher:
+            PeriodicOperation("test_operation", self._operation_with_failure, period=datetime.timedelta(hours=1)).run()
+
+        self.assertEqual(self._get_number_of_warnings(warn_patcher), 1, "The error in the operation was should have been reported exactly once")
+
+    def test_it_should_not_log_multiple_warnings_when_the_period_has_not_elapsed(self):
+        with patch("azurelinuxagent.common.logger.warn") as warn_patcher:
+            pop = PeriodicOperation("test_operation", self._operation_with_failure, period=datetime.timedelta(hours=1))
+            for _ in range(5):
+                pop.run()
+
+        self.assertEqual(self._get_number_of_warnings(warn_patcher), 1, "The error in the operation was should have been reported exactly once")
+
+    def test_it_should_not_multiple_warnings_when_the_period_has_elapsed(self):
+        with patch("azurelinuxagent.common.logger.warn") as warn_patcher:
+            with patch("azurelinuxagent.ga.periodic_operation.PeriodicOperation._LOG_WARNING_PERIOD", new_callable=PropertyMock, return_value=datetime.timedelta(milliseconds=1)):
+                pop = PeriodicOperation("test_operation", self._operation_with_failure, period=datetime.timedelta(milliseconds=1))
+                for _ in range(5):
+                    pop.run()
+                    time.sleep(0.001)
+
+            self.assertEqual(self._get_number_of_warnings(warn_patcher), 5, "The error in the operation was not reported the expected number of times")
+
+    def test_it_should_log_warnings_if_they_are_different(self):
+        with patch("azurelinuxagent.common.logger.warn") as warn_patcher:
+            for i in range(2):
+                def operation():
+                    raise Exception("WARNING {0}".format(i))
+
+                pop = PeriodicOperation("test_operation", operation, period=datetime.timedelta(hours=1))
+                for _ in range(5):
+                    pop.run()
+
+            self.assertEqual(self._get_number_of_warnings(warn_patcher, "WARNING 0"), 1, "The first error should have been reported exactly 1 time")
+            self.assertEqual(self._get_number_of_warnings(warn_patcher, "WARNING 1"), 1, "The second error should have been reported exactly 1 time")
diff --git a/tests/protocol/mocks.py b/tests/protocol/mocks.py
index 7172ce5f..1ab39227 100644
--- a/tests/protocol/mocks.py
+++ b/tests/protocol/mocks.py
@@ -158,6 +158,10 @@ class HttpRequestPredicates(object):
     def is_telemetry_request(url):
         return url.lower() == 'http://{0}/machine?comp=telemetrydata'.format(restutil.KNOWN_WIRESERVER_IP)
 
+    @staticmethod
+    def is_health_service_request(url):
+        return url.lower() == 'http://{0}:80/healthservice'.format(restutil.KNOWN_WIRESERVER_IP)
+
     @staticmethod
     def is_in_vm_artifacts_profile_request(url):
         return re.match(r'https://.+\.blob\.core\.windows\.net/\$system/.+\.(vmSettings|settings)\?.+', url) is not None
@@ -171,6 +175,10 @@ class HttpRequestPredicates(object):
             raise ValueError('Host plugin request is missing the x-ms-artifact-location header ({0})'.format(url))
         return headers['x-ms-artifact-location']
 
+    @staticmethod
+    def is_host_plugin_health_request(url):
+        return url.lower() == 'http://{0}:{1}/health'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
+
     @staticmethod
     def is_host_plugin_extension_artifact_request(url):
         return url.lower() == 'http://{0}:{1}/extensionartifact'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
diff --git a/tests/protocol/mockwiredata.py b/tests/protocol/mockwiredata.py
index 259affcb..1b8cef3d 100644
--- a/tests/protocol/mockwiredata.py
+++ b/tests/protocol/mockwiredata.py
@@ -92,6 +92,7 @@ class WireProtocolData(object):
         self.call_counts = {
             "comp=versions": 0,
             "/versions": 0,
+            "/health": 0,
             "/HealthService": 0,
             "goalstate": 0,
             "hostingenvuri": 0,
@@ -155,6 +156,9 @@ class WireProtocolData(object):
         elif "/versions" in url:  # HostPlugin versions
             content = '["2015-09-01"]'
             self.call_counts["/versions"] += 1
+        elif url.endswith("/health"):  # HostPlugin health
+            content = ''
+            self.call_counts["/health"] += 1
         elif "goalstate" in url:
             content = self.goal_state
             self.call_counts["goalstate"] += 1