Repository: Azure/WALinuxAgent
Pull request #1770 (2020-01-31 21:46:26)

Monitor thread errors
Added code to handle exceptions in the monitor thread main loop.

Minor refactoring of the functions called by the main loop to ensure consistency on how they handle exceptions. I decided to leave log_altered_network_configuration alone for now (it is setup as a high-priority operation).

Tested the refactoring changes using the debugger on a live VM.

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/azure/walinuxagent/1770)
<!-- Reviewable:end -->



Patch:
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index b66f315a..ae208925 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -234,11 +234,11 @@ class MonitorHandler(object):
         agent or the extensions. We don't have control over extension's events parameters, but we will override
         any values they might have set for sys_info parameters.
         """
-        if self.last_event_collection is None:
-            self.last_event_collection = datetime.datetime.utcnow() - MonitorHandler.EVENT_COLLECTION_PERIOD
+        try:
+            if self.last_event_collection is None:
+                self.last_event_collection = datetime.datetime.utcnow() - MonitorHandler.EVENT_COLLECTION_PERIOD
 
-        if datetime.datetime.utcnow() >= (self.last_event_collection + MonitorHandler.EVENT_COLLECTION_PERIOD):
-            try:
+            if datetime.datetime.utcnow() >= (self.last_event_collection + MonitorHandler.EVENT_COLLECTION_PERIOD):
                 event_list = TelemetryEventList()
                 event_dir = os.path.join(conf.get_lib_dir(), "events")
                 event_files = os.listdir(event_dir)
@@ -267,10 +267,10 @@ class MonitorHandler(object):
                     self.protocol.report_event(event_list)
                 except ProtocolError as e:
                     logger.error("{0}", ustr(e))
-            except Exception as e:
-                logger.warn("Failed to send events: {0}", ustr(e))
+        except Exception as e:
+            logger.warn("Failed to send events: {0}", ustr(e))
 
-            self.last_event_collection = datetime.datetime.utcnow()
+        self.last_event_collection = datetime.datetime.utcnow()
 
     def daemon(self, init_data=False):
 
@@ -285,15 +285,18 @@ class MonitorHandler(object):
                         MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD,
                         MonitorHandler.IMDS_HEARTBEAT_PERIOD).seconds
         while self.should_run:
-            self.protocol.update_host_plugin_from_goal_state()
-            self.send_telemetry_heartbeat()
-            self.poll_telemetry_metrics()
-            self.send_telemetry_metrics()   # This will be removed in favor of poll_telemetry_metrics() and it'll directly send the perf data for each cgroup.
-            self.collect_and_send_events()
-            self.send_host_plugin_heartbeat()
-            self.send_imds_heartbeat()
-            self.log_altered_network_configuration()
-            self.reset_loggers()
+            try:
+                self.protocol.update_host_plugin_from_goal_state()
+                self.send_telemetry_heartbeat()
+                self.poll_telemetry_metrics()
+                self.send_telemetry_metrics()   # This will be removed in favor of poll_telemetry_metrics() and it'll directly send the perf data for each cgroup.
+                self.collect_and_send_events()
+                self.send_host_plugin_heartbeat()
+                self.send_imds_heartbeat()
+                self.log_altered_network_configuration()
+                self.reset_loggers()
+            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)
 
     def reset_loggers(self):
@@ -302,16 +305,18 @@ class MonitorHandler(object):
         For reference, please check azurelinuxagent.common.logger.Logger and
         azurelinuxagent.common.event.EventLogger classes
         """
-        time_now = datetime.datetime.utcnow()
-        if not self.last_reset_loggers_time:
-            self.last_reset_loggers_time = time_now
+        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):
-            try:
+            if time_now >= (self.last_reset_loggers_time + MonitorHandler.RESET_LOGGERS_PERIOD):
                 logger.reset_periodic()
-            finally:
-                self.last_reset_loggers_time = time_now
+
+        except Exception as e:
+            logger.warn("Failed to clear periodic loggers: {0}", ustr(e))
+
+        self.last_reset_loggers_time = time_now
 
     def add_sysinfo(self, event):
         """
@@ -345,11 +350,11 @@ class MonitorHandler(object):
         successfully called and validated a response in the last IMDS_HEALTH_PERIOD.
         """
 
-        if self.last_imds_heartbeat is None:
-            self.last_imds_heartbeat = datetime.datetime.utcnow() - MonitorHandler.IMDS_HEARTBEAT_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):
-            try:
+            if datetime.datetime.utcnow() >= (self.last_imds_heartbeat + MonitorHandler.IMDS_HEARTBEAT_PERIOD):
                 is_currently_healthy, response = self.imds_client.validate()
 
                 if is_currently_healthy:
@@ -362,29 +367,29 @@ class MonitorHandler(object):
 
                 self.health_service.report_imds_status(is_healthy, response)
 
-            except Exception as e:
-                msg = "Exception sending imds heartbeat: {0}".format(ustr(e))
-                add_event(
-                    name=AGENT_NAME,
-                    version=CURRENT_VERSION,
-                    op=WALAEventOperation.ImdsHeartbeat,
-                    is_success=False,
-                    message=msg,
-                    log_event=False)
+        except Exception as e:
+            msg = "Exception sending imds heartbeat: {0}".format(ustr(e))
+            add_event(
+                name=AGENT_NAME,
+                version=CURRENT_VERSION,
+                op=WALAEventOperation.ImdsHeartbeat,
+                is_success=False,
+                message=msg,
+                log_event=False)
 
-            self.last_imds_heartbeat = datetime.datetime.utcnow()
+        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.
         """
-        if self.last_host_plugin_heartbeat is None:
-            self.last_host_plugin_heartbeat = datetime.datetime.utcnow() - MonitorHandler.HOST_PLUGIN_HEARTBEAT_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() >= (
+            if datetime.datetime.utcnow() >= (
                 self.last_host_plugin_heartbeat + MonitorHandler.HOST_PLUGIN_HEARTBEAT_PERIOD):
-            try:
                 host_plugin = self.protocol.client.get_host_plugin()
                 host_plugin.ensure_initialized()
                 is_currently_healthy = host_plugin.get_health()
@@ -408,25 +413,24 @@ class MonitorHandler(object):
                         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))
-                add_event(
-                    name=AGENT_NAME,
-                    version=CURRENT_VERSION,
-                    op=WALAEventOperation.HostPluginHeartbeat,
-                    is_success=False,
-                    message=msg,
-                    log_event=False)
+        except Exception as e:
+            msg = "Exception sending host plugin heartbeat: {0}".format(ustr(e))
+            add_event(
+                name=AGENT_NAME,
+                version=CURRENT_VERSION,
+                op=WALAEventOperation.HostPluginHeartbeat,
+                is_success=False,
+                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 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):
-            try:
+            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")
@@ -442,10 +446,10 @@ class MonitorHandler(object):
                         is_success=True,
                         message=msg,
                         log_event=False)
-            except Exception as e:
-                logger.warn("Failed to send heartbeat: {0}", ustr(e))
+        except Exception as e:
+            logger.warn("Failed to send heartbeat: {0}", ustr(e))
 
-            self.last_telemetry_heartbeat = datetime.datetime.utcnow()
+        self.last_telemetry_heartbeat = datetime.datetime.utcnow()
 
     def poll_telemetry_metrics(self):
         """
@@ -461,7 +465,6 @@ class MonitorHandler(object):
             if time_now >= (self.last_cgroup_polling_telemetry +
                             MonitorHandler.CGROUP_TELEMETRY_POLLING_PERIOD):
                 metrics = CGroupsTelemetry.poll_all_tracked()
-                self.last_cgroup_polling_telemetry = time_now
 
                 if metrics:
                     for metric in metrics:
@@ -469,21 +472,22 @@ class MonitorHandler(object):
         except Exception as e:
             logger.warn("Could not poll all the tracked telemetry due to {0}", ustr(e))
 
+        self.last_cgroup_polling_telemetry = datetime.datetime.utcnow()
+
     def send_telemetry_metrics(self):
         """
         The send_telemetry_metrics would soon be removed in favor of sending performance metrics directly.
 
         :return:
         """
-        time_now = datetime.datetime.utcnow()
-
         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()
-                self.last_cgroup_report_telemetry = time_now
 
                 if performance_metrics:
                     message = generate_extension_metrics_telemetry_dictionary(schema_version=1.0,
@@ -497,6 +501,8 @@ class MonitorHandler(object):
         except Exception as e:
             logger.warn("Could not report all the tracked telemetry due to {0}", ustr(e))
 
+        self.last_cgroup_report_telemetry = datetime.datetime.utcnow()
+
     def log_altered_network_configuration(self):
         """
         Check various pieces of network configuration and, if altered since the last check, log the new state.


Test patch:
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index ed9202f0..ad44934b 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -318,6 +318,38 @@ class TestMonitor(AgentTestCase):
         self.assertEqual(12, counter)
         os.environ.pop(CONTAINER_ID_ENV_VARIABLE)
 
+    def test_main_loop_should_handle_exceptions_retrieving_goal_state(self, *_):
+        monitor_handler = get_monitor_handler()
+
+        with patch("azurelinuxagent.common.logger.warn") as mock_warn:
+            with patch.object(monitor_handler, 'protocol') as mock_protocol:
+                mock_protocol.update_host_plugin_from_goal_state = Mock(side_effect=Exception("A TEST EXCEPTION"))
+
+                with patch("azurelinuxagent.ga.monitor.time") as mock_time:
+                    # time.sleep is called at the end of each iteration of the main loop; we use it to stop the
+                    # loop after a few iterations (this ensures that the test doesn't run for ever and is also
+                    # a check that the loop does not stop after the first error)
+                    def mock_sleep(_):
+                        mock_sleep.iterations += 1
+                        if mock_sleep.iterations >= 3:
+                            monitor_handler.stop()
+                    mock_sleep.iterations = 0
+
+                    mock_time.sleep = Mock(side_effect=mock_sleep)
+
+                    # run the main loop
+                    monitor_handler.daemon(init_data=False)
+
+                    self.assertEqual(mock_sleep.iterations, 3)
+
+                    self.assertEquals(mock_warn.call_count, 3)
+
+                    for call_args in mock_warn.call_args_list:
+                        args, kwargs = call_args
+                        self.assertEquals(len(args), 2)
+                        self.assertIn("A TEST EXCEPTION", args[1])
+
+
     @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")
diff --git a/tests/protocol/test_wire.py b/tests/protocol/test_wire.py
index 072569dd..995c1a00 100644
--- a/tests/protocol/test_wire.py
+++ b/tests/protocol/test_wire.py
@@ -213,51 +213,6 @@ class TestWireProtocol(AgentTestCase):
             self.assertEqual(goal_state.role_config_name, host_plugin.role_config_name)
             self.assertEqual(1, patch_get_goal_state.call_count)
 
-    @patch("azurelinuxagent.common.utils.restutil.http_request", side_effect=IOError)
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.get_host_plugin")
-    @patch("azurelinuxagent.common.protocol.hostplugin.HostPluginProtocol.get_artifact_request")
-    def test_download_ext_handler_pkg_fallback(self, patch_request, patch_get_host, patch_http, *args):
-        ext_uri = 'extension_uri'
-        host_uri = 'host_uri'
-        destination = 'destination'
-        patch_get_host.return_value = HostPluginProtocol(host_uri, 'container_id', 'role_config')
-        patch_request.return_value = [host_uri, {}]
-
-        WireProtocol(WIRESERVER_URL).download_ext_handler_pkg(ext_uri, destination)
-
-        self.assertEqual(patch_http.call_count, 2)
-        self.assertEqual(patch_request.call_count, 1)
-        self.assertEqual(patch_http.call_args_list[0][0][1], ext_uri)
-        self.assertEqual(patch_http.call_args_list[1][0][1], host_uri)
-
-    @skip_if_predicate_true(running_under_travis, "Travis unit tests should not have external dependencies")
-    def test_download_ext_handler_pkg_stream(self, *args):
-        ext_uri = 'https://dcrdata.blob.core.windows.net/files/packer.zip'
-        tmp = tempfile.mkdtemp()
-        destination = os.path.join(tmp, 'test_download_ext_handler_pkg_stream.zip')
-
-        success = WireProtocol(WIRESERVER_URL).download_ext_handler_pkg(ext_uri, destination)
-        self.assertTrue(success)
-        self.assertTrue(os.path.exists(destination))
-
-        # verify size
-        self.assertEqual(33193077, os.stat(destination).st_size)
-
-        # verify unzip
-        zipfile.ZipFile(destination).extractall(tmp)
-        packer = os.path.join(tmp, 'packer')
-        self.assertTrue(os.path.exists(packer))
-        fileutil.chmod(packer, os.stat(packer).st_mode | stat.S_IXUSR)
-
-        # verify unpacked size
-        self.assertEqual(105552030, os.stat(packer).st_size)
-
-        # execute, verify result
-        packer_version = '{0} --version'.format(packer)
-        rc, stdout = run_get_output(packer_version)
-        self.assertEqual(0, rc)
-        self.assertEqual('1.3.5\n', stdout)
-
     @patch("azurelinuxagent.common.protocol.wire.WireClient.update_goal_state")
     def test_upload_status_blob_default(self, *args):
         """