Repository: Azure/WALinuxAgent
Pull request #1937 (2020-07-07 19:28:53)

Fix telemetry unicode errors
<!-- DO NOT DELETE THIS TEMPLATE -->

## Description
This PR mainly deals with ensuring that the telemetry events sent by the agent are encoded properly in UTF-8 for both Py2 and Py3. This is to support special characters in telemetry.

Issue # <!-- if any -->
<!--
Please add an informative description that covers that changes made by the pull request. 
This checklist is used to make sure that common issues in a pull request are addressed.
This will expedite the process of getting your pull request merged and avoid extra work on your part to fix issues discovered during the review process.
-->

---

### PR information
- [ ] The title of the PR is clear and informative.
- [ ] There are a small number of commits, each of which has an informative message. This means that previously merged commits do not appear in the history of the PR. For information on cleaning up the commits in your pull request, [see this page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).
- [ ] Except for special cases involving multiple contributors, the PR is started from a fork of the main repository, not a branch.
- [ ] If applicable, the PR references the bug/issue that it fixes in the description.
- [ ] New Unit tests were added for the changes made and Travis.CI is passing.

### Quality of Code and Contribution Guidelines
- [ ] I have read the [contribution guidelines](https://github.com/Azure/WALinuxAgent/blob/master/.github/CONTRIBUTING.md).


Patch:
diff --git a/azurelinuxagent/common/event.py b/azurelinuxagent/common/event.py
index eca62556..07a69c29 100644
--- a/azurelinuxagent/common/event.py
+++ b/azurelinuxagent/common/event.py
@@ -76,6 +76,8 @@ class WALAEventOperation:
     CGroupsInfo = "CGroupsInfo"
     CGroupsInitialize = "CGroupsInitialize"
     CGroupsLimitsCrossed = "CGroupsLimitsCrossed"
+    CollectEventErrors = "CollectEventErrors"
+    CollectEventUnicodeErrors = "CollectEventUnicodeErrors"
     ConfigurationChange = "ConfigurationChange"
     CustomData = "CustomData"
     Deploy = "Deploy"
@@ -105,6 +107,8 @@ class WALAEventOperation:
     Provision = "Provision"
     ProvisionGuestAgent = "ProvisionGuestAgent"
     RemoteAccessHandling = "RemoteAccessHandling"
+    ReportEventErrors = "ReportEventErrors"
+    ReportEventUnicodeErrors = "ReportEventUnicodeErrors"
     ReportStatus = "ReportStatus"
     ReportStatusExtended = "ReportStatusExtended"
     Restart = "Restart"
@@ -574,14 +578,25 @@ class EventLogger(object):
 
         event.parameters = trimmed_params
 
+    @staticmethod
+    def report_dropped_events_error(count, errors, op, max_errors_to_report):
+        err_msg_format = "DroppedEventsCount: {0}\nReasons (first {1} errors): {2}"
+        if count > 0:
+            add_event(op=op,
+                      message=err_msg_format.format(count, max_errors_to_report, ', '.join(errors)),
+                      is_success=False)
+
     def collect_events(self):
         """
         Retuns a list of events that need to be sent to the telemetry pipeline and deletes the corresponding files
         from the events directory.
         """
+        max_collect_errors_to_report = 5
         event_list = TelemetryEventList()
         event_directory_full_path = os.path.join(conf.get_lib_dir(), EVENTS_DIRECTORY)
         event_files = os.listdir(event_directory_full_path)
+        unicode_error_count, unicode_errors = 0, []
+        collect_event_error_count, collect_event_errors = 0, []
 
         for event_file in event_files:
             try:
@@ -617,8 +632,20 @@ class EventLogger(object):
                     event_list.events.append(event)
                 finally:
                     os.remove(event_file_path)
+            except UnicodeError as e:
+                unicode_error_count += 1
+                if len(unicode_errors) < max_collect_errors_to_report:
+                    unicode_errors.append(ustr(e))
             except Exception as e:
-                logger.warn("Failed to process event file {0}: {1}", event_file, ustr(e))
+                collect_event_error_count += 1
+                if len(collect_event_errors) < max_collect_errors_to_report:
+                    collect_event_errors.append(ustr(e))
+
+        EventLogger.report_dropped_events_error(collect_event_error_count, collect_event_errors,
+                                                WALAEventOperation.CollectEventErrors, max_collect_errors_to_report)
+        EventLogger.report_dropped_events_error(unicode_error_count, unicode_errors,
+                                                WALAEventOperation.CollectEventUnicodeErrors,
+                                                max_collect_errors_to_report)
 
         return event_list
 
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index b373ea2a..a28de91d 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -29,7 +29,7 @@ import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.textutil as textutil
 from azurelinuxagent.common.datacontract import validate_param
-from azurelinuxagent.common.event import add_event, add_periodic, WALAEventOperation, EVENTS_DIRECTORY
+from azurelinuxagent.common.event import add_event, add_periodic, WALAEventOperation, EVENTS_DIRECTORY, EventLogger
 from azurelinuxagent.common.exception import ProtocolNotFoundError, \
     ResourceGoneError, ExtensionDownloadError, InvalidContainerError, ProtocolError, HttpError
 from azurelinuxagent.common.future import httpclient, bytebuffer, ustr
@@ -488,7 +488,7 @@ class StatusBlob(object):
 
 
 def event_param_to_v1(param):
-    param_format = '<Param Name="{0}" Value={1} T="{2}" />'
+    param_format = ustr('<Param Name="{0}" Value={1} T="{2}" />')
     param_type = type(param.value)
     attr_type = ""
     if param_type is int:
@@ -506,14 +506,12 @@ def event_param_to_v1(param):
                                attr_type)
 
 
-def event_to_v1(event):
+def event_to_v1_encoded(event, encoding='utf-8'):
     params = ""
     for param in event.parameters:
         params += event_param_to_v1(param)
-    event_str = ('<Event id="{0}">'
-                 '<![CDATA[{1}]]>'
-                 '</Event>').format(event.eventId, params)
-    return event_str
+    event_str = ustr('<Event id="{0}"><![CDATA[{1}]]></Event>').format(event.eventId, params)
+    return event_str.encode(encoding)
 
 
 class WireClient(object):
@@ -1070,14 +1068,13 @@ class WireClient(object):
                                  u",{0}: {1}").format(resp.status,
                                                       resp.read()))
 
-    def send_event(self, provider_id, event_str):
+    def send_encoded_event(self, provider_id, event_str, encoding='utf-8'):
         uri = TELEMETRY_URI.format(self.get_endpoint())
-        data_format = ('<?xml version="1.0"?>'
-                       '<TelemetryData version="1.0">'
-                       '<Provider id="{0}">{1}'
-                       '</Provider>'
-                       '</TelemetryData>')
-        data = data_format.format(provider_id, event_str)
+        data_format_header = ustr('<?xml version="1.0"?><TelemetryData version="1.0"><Provider id="{0}">').format(
+            provider_id).encode(encoding)
+        data_format_footer = ustr('</Provider></TelemetryData>').encode(encoding)
+        # Event string should already be encoded by the time it gets here, to avoid double encoding, dividing it into parts.
+        data = data_format_header + event_str + data_format_footer
         try:
             header = self.get_header_for_xml_content()
             # NOTE: The call to wireserver requests utf-8 encoding in the headers, but the body should not
@@ -1092,30 +1089,54 @@ class WireClient(object):
                 "Failed to send events:{0}".format(resp.status))
 
     def report_event(self, event_list):
+        max_send_errors_to_report = 5
         buf = {}
+        events_per_request = 0
+        unicode_error_count, unicode_errors = 0, []
+        event_report_error_count, event_report_errors = 0, []
+
         # Group events by providerId
         for event in event_list.events:
-            if event.providerId not in buf:
-                buf[event.providerId] = ""
-            event_str = event_to_v1(event)
-            if len(event_str) >= MAX_EVENT_BUFFER_SIZE:
-                details_of_event = [ustr(x.name) + ":" + ustr(x.value) for x in event.parameters if x.name in
-                                    [GuestAgentExtensionEventsSchema.Name, GuestAgentExtensionEventsSchema.Version,
-                                     GuestAgentExtensionEventsSchema.Operation,
-                                     GuestAgentExtensionEventsSchema.OperationSuccess]]
-                logger.periodic_warn(logger.EVERY_HALF_HOUR,
-                                     "Single event too large: {0}, with the length: {1} more than the limit({2})"
-                                     .format(str(details_of_event), len(event_str), MAX_EVENT_BUFFER_SIZE))
-                continue
-            if len(buf[event.providerId] + event_str) >= MAX_EVENT_BUFFER_SIZE:
-                self.send_event(event.providerId, buf[event.providerId])
-                buf[event.providerId] = ""
-            buf[event.providerId] = buf[event.providerId] + event_str
+            try:
+                if event.providerId not in buf:
+                    buf[event.providerId] = b''
+                event_str = event_to_v1_encoded(event)
+                if len(event_str) >= MAX_EVENT_BUFFER_SIZE:
+                    details_of_event = [ustr(x.name) + ":" + ustr(x.value) for x in event.parameters if x.name in
+                                        [GuestAgentExtensionEventsSchema.Name, GuestAgentExtensionEventsSchema.Version,
+                                         GuestAgentExtensionEventsSchema.Operation,
+                                         GuestAgentExtensionEventsSchema.OperationSuccess]]
+                    logger.periodic_warn(logger.EVERY_HALF_HOUR,
+                                         "Single event too large: {0}, with the length: {1} more than the limit({2})"
+                                         .format(str(details_of_event), len(event_str), MAX_EVENT_BUFFER_SIZE))
+                    continue
+                if len(buf[event.providerId] + event_str) >= MAX_EVENT_BUFFER_SIZE:
+                    self.send_encoded_event(event.providerId, buf[event.providerId])
+                    buf[event.providerId] = b''
+                    logger.verbose("No of events this request = {0}".format(events_per_request))
+                    events_per_request = 0
+                buf[event.providerId] = buf[event.providerId] + event_str
+                events_per_request += 1
+            except UnicodeError as e:
+                unicode_error_count += 1
+                if len(unicode_errors) < max_send_errors_to_report:
+                    unicode_errors.append(ustr(e))
+            except Exception as e:
+                event_report_error_count += 1
+                if len(event_report_errors) < max_send_errors_to_report:
+                    event_report_errors.append(ustr(e))
+
+        EventLogger.report_dropped_events_error(event_report_error_count, event_report_errors,
+                                                WALAEventOperation.CollectEventErrors, max_send_errors_to_report)
+        EventLogger.report_dropped_events_error(unicode_error_count, unicode_errors,
+                                                WALAEventOperation.CollectEventUnicodeErrors,
+                                                max_send_errors_to_report)
 
         # Send out all events left in buffer.
         for provider_id in list(buf.keys()):
             if len(buf[provider_id]) > 0:
-                self.send_event(provider_id, buf[provider_id])
+                logger.verbose("No of events this request = {0}".format(events_per_request))
+                self.send_encoded_event(provider_id, buf[provider_id])
 
     def report_status_event(self, message, is_success):
         from azurelinuxagent.common.event import report_event, \
diff --git a/azurelinuxagent/common/utils/restutil.py b/azurelinuxagent/common/utils/restutil.py
index fd78b969..4f9ca317 100644
--- a/azurelinuxagent/common/utils/restutil.py
+++ b/azurelinuxagent/common/utils/restutil.py
@@ -319,10 +319,11 @@ def _http_request(method, host, rel_uri, port=None, data=None, secure=False,
                                          conn_port,
                                          timeout=10)
 
+    # Logger requires the data to be a ustr to log properly, ensuring that the data string that we log is always ustr.
     logger.verbose("HTTP connection [{0}] [{1}] [{2}] [{3}]",
                    method,
                    redact_sas_tokens_in_urls(url),
-                   data,
+                   textutil.str_to_encoded_ustr(data),
                    headers)
 
     conn.request(method=method, url=url, body=data, headers=headers)
diff --git a/azurelinuxagent/common/utils/textutil.py b/azurelinuxagent/common/utils/textutil.py
index 902c2420..f7171c2d 100644
--- a/azurelinuxagent/common/utils/textutil.py
+++ b/azurelinuxagent/common/utils/textutil.py
@@ -27,6 +27,8 @@ import sys
 import zlib
 import xml.dom.minidom as minidom
 
+from azurelinuxagent.common.future import ustr
+
 
 def parse_doc(xml_text):
     """
@@ -396,3 +398,34 @@ def format_memory_value(unit, value):
         raise TypeError('Value must be convertible to a float')
 
     return int(value * units[unit])
+
+def str_to_encoded_ustr(s, encoding='utf-8'):
+    """
+    This function takes the string and converts it into the corresponding encoded ustr if its not already a ustr.
+    The encoding is utf-8 by default if not specified.
+    Note: ustr() is a unicode object for Py2 and a str object for Py3.
+    :param s: The string to convert to ustr
+    :param encoding: Encoding to use. Utf-8 by default
+    :return: Returns the corresponding ustr string. Returns None if input is None.
+    """
+
+    # TODO: Import at the top of the file instead of a local import (using local import here to avoid cyclic dependency)
+    from azurelinuxagent.common.version import PY_VERSION_MAJOR
+
+    if s is None or type(s) is ustr:
+        # If its already a ustr/None then return as is
+        return s
+    if PY_VERSION_MAJOR > 2:
+        try:
+            # For py3+, str() is unicode by default
+            if isinstance(s, bytes):
+                # str.encode() returns bytes which should be decoded to get the str.
+                return s.decode(encoding)
+            else:
+                # If its not encoded, just return the string
+                return ustr(s)
+        except Exception:
+            # If some issues in decoding, just return the string
+            return ustr(s)
+    # For Py2, explicitly convert the string to unicode with the specified encoding
+    return ustr(s, encoding=encoding)
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index 42e7c118..349b1788 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -19,6 +19,8 @@ import datetime
 import threading
 import uuid
 
+from azurelinuxagent.common.utils import textutil
+
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.networkutil as networkutil
 from azurelinuxagent.common.cgroupconfigurator import CGroupConfigurator
@@ -247,10 +249,14 @@ class MonitorHandler(object):
         """
         Periodically send any events located in the events folder
         """
-        event_list = collect_events()
+        try:
+            event_list = collect_events()
 
-        if len(event_list.events) > 0:
-            self.protocol.report_event(event_list)
+            if len(event_list.events) > 0:
+                self.protocol.report_event(event_list)
+        except Exception as e:
+            err_msg = "Failure in collecting/sending Agent events: {0}".format(ustr(e))
+            add_event(op=WALAEventOperation.UnhandledError, message=err_msg, is_success=False)
 
     def send_imds_heartbeat(self):
         """


Test patch:
diff --git a/tests/common/test_event.py b/tests/common/test_event.py
index 24f81516..dd732346 100644
--- a/tests/common/test_event.py
+++ b/tests/common/test_event.py
@@ -1,3 +1,4 @@
+# coding=utf-8
 #
 # Copyright 2017 Microsoft Corporation
 #
@@ -381,7 +382,7 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
         os.chmod(self._create_test_event_file("custom_script_no_read_access.tld"), 0o200)
         self._create_test_event_file("custom_script_2.tld")  # another valid event
 
-        with patch("azurelinuxagent.common.event.logger.warn") as mock_warn:
+        with patch("azurelinuxagent.common.event.add_event") as mock_add_event:
             event_list = event.collect_events()
 
             self.assertEquals(
@@ -390,17 +391,19 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
                 all(TestEvent._get_event_message(evt) == "A test telemetry message." for evt in event_list.events),
                 "The valid events were not found")
 
-            invalid_events = {}
-            for args in mock_warn.call_args_list:
-                if re.search('Failed to process event file', args[0][0]) is not None:
-                    invalid_events[args[0][1]] = args[0][1]
+            invalid_events = []
+            total_dropped_count = 0
+            for args, kwargs in mock_add_event.call_args_list:
+                match = re.search("DroppedEventsCount: (\d+)", kwargs['message'])
+                if match is not None:
+                    invalid_events.append(kwargs['op'])
+                    total_dropped_count += int(match.groups()[0])
 
-            def assert_invalid_file_was_reported(file):
-                self.assertIn(file, invalid_events, '{0} was not reported as an invalid event file'.format(file))
-
-            assert_invalid_file_was_reported("custom_script_utf-16.tld")
-            assert_invalid_file_was_reported("custom_script_invalid_json.tld")
-            assert_invalid_file_was_reported("custom_script_no_read_access.tld")
+            self.assertEquals(3, total_dropped_count, "Total dropped events dont match")
+            self.assertIn(WALAEventOperation.CollectEventErrors, invalid_events,
+                          "{0} errors not reported".format(WALAEventOperation.CollectEventErrors))
+            self.assertIn(WALAEventOperation.CollectEventUnicodeErrors, invalid_events,
+                          "{0} errors not reported".format(WALAEventOperation.CollectEventUnicodeErrors))
 
     def test_save_event_rollover(self):
         # We keep 1000 events only, and the older ones are removed.
@@ -716,8 +719,10 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
 
             raise ValueError('Could not find the Message for the telemetry event in {0}'.format(event_file))
 
-        def get_event_message_from_http_request_body(http_request_body):
+        def get_event_message_from_http_request_body(event_body):
             # The XML for the event is sent over as a CDATA element ("Event") in the request's body
+            http_request_body = event_body if (
+                        event_body is None or type(event_body) is ustr) else textutil.str_to_encoded_ustr(event_body)
             request_body_xml_doc = textutil.parse_doc(http_request_body)
 
             event_node = textutil.find(request_body_xml_doc, "Event")
@@ -764,6 +769,43 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
 
             self.assertEquals(event_message, expected_message, "The Message in the HTTP request does not match the Message in the event's *.tld file")
 
+    def test_report_event_should_encode_events_correctly(self):
+
+        def http_post_handler(url, body, **__):
+            if self.is_telemetry_request(url):
+                http_post_handler.request_body = body
+                return MockHttpResponse(status=200)
+            return None
+        http_post_handler.request_body = None
+
+        with mock_wire_protocol(mockwiredata.DATA_FILE, http_post_handler=http_post_handler) as protocol:
+            test_messages = [
+                'Non-English message -  此文字不是英文的',
+                "Ξεσκεπάζω τὴν ψυχοφθόρα βδελυγμία",
+                "The quick brown fox jumps over the lazy dog",
+                "El pingüino Wenceslao hizo kilómetros bajo exhaustiva lluvia y frío, añoraba a su querido cachorro.",
+                "Portez ce vieux whisky au juge blond qui fume sur son île intérieure, à côté de l'alcôve ovoïde, où les bûches",
+                "se consument dans l'âtre, ce qui lui permet de penser à la cænogenèse de l'être dont il est question",
+                "dans la cause ambiguë entendue à Moÿ, dans un capharnaüm qui, pense-t-il, diminue çà et là la qualité de son œuvre.",
+                "D'fhuascail Íosa, Úrmhac na hÓighe Beannaithe, pór Éava agus Ádhaimh",
+                "Árvíztűrő tükörfúrógép",
+                "Kæmi ný öxi hér ykist þjófum nú bæði víl og ádrepa",
+                "Sævör grét áðan því úlpan var ónýt",
+                "いろはにほへとちりぬるを わかよたれそつねならむ うゐのおくやまけふこえて あさきゆめみしゑひもせす",
+                "? דג סקרן שט בים מאוכזב ולפתע מצא לו חברה איך הקליטה"
+                "Pchnąć w tę łódź jeża lub ośm skrzyń fig",
+                "Normal string event"
+            ]
+            for msg in test_messages:
+                add_event('TestEventEncoding', message=msg)
+                event_list = event.collect_events()
+                protocol.client.report_event(event_list)
+                # In Py2, encode() produces a str and in py3 it produces a bytes string.
+                # type(bytes) == type(str) for Py2 so this check is mainly for Py3 to ensure that the event is encoded properly.
+                self.assertIsInstance(http_post_handler.request_body, bytes, "The Event request body should be encoded")
+                self.assertIn(textutil.str_to_encoded_ustr(msg).encode('utf-8'), http_post_handler.request_body,
+                              "Encoded message not found in body")
+
 
 class TestMetrics(AgentTestCase):
     @patch('azurelinuxagent.common.event.EventLogger.save_event')
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index 294db288..14f25dd7 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -25,6 +25,7 @@ import re
 import string
 import tempfile
 import time
+import uuid
 from datetime import timedelta
 
 from azurelinuxagent.common.protocol.util import ProtocolUtil
@@ -210,8 +211,17 @@ class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
         data = get_properties(event)
         return json.dumps(data)
 
+    def _assert_error_event_reported(self, mock_add_event, expected_msg):
+        found_msg = False
+        for call_args in mock_add_event.call_args_list:
+            _, kwargs = call_args
+            if expected_msg in kwargs['message']:
+                found_msg = True
+                break
+        self.assertTrue(found_msg, "Error event not reported")
+
     @patch("azurelinuxagent.common.event.TELEMETRY_EVENT_PROVIDER_ID", _TEST_EVENT_PROVIDER_ID)
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_collect_and_send_events(self, mock_lib_dir, patch_send_event, *_):
         mock_lib_dir.return_value = self.lib_dir
@@ -233,7 +243,7 @@ class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
 
             # Validating the crafted message by the collect_and_send_events call.
             self.assertEqual(1, patch_send_event.call_count)
-            send_event_call_args = monitor_handler.get_mock_wire_protocol().client.send_event.call_args[0]
+            send_event_call_args = monitor_handler.get_mock_wire_protocol().client.send_encoded_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()
@@ -272,9 +282,9 @@ class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
                                                   osutil.get_processor_cores())
 
             self.maxDiff = None
-            self.assertEqual(sample_message, send_event_call_args[1])
+            self.assertEqual(sample_message.encode('utf-8'), send_event_call_args[1])
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_collect_and_send_events_with_small_events(self, mock_lib_dir, patch_send_event, *_):
         mock_lib_dir.return_value = self.lib_dir
@@ -293,7 +303,7 @@ class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
 
             self.assertEqual(4, patch_send_event.call_count)
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_collect_and_send_events_with_large_events(self, mock_lib_dir, patch_send_event, *_):
         mock_lib_dir.return_value = self.lib_dir
@@ -334,12 +344,14 @@ class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
                 size = 2 ** power * 1024
                 self._create_extension_event(size)
 
-            with patch("azurelinuxagent.common.logger.warn") as mock_warn:
+            with patch("azurelinuxagent.ga.monitor.add_event") as mock_add_event:
                 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(1, mock_add_event.call_count)
                 self.assertEqual(0, len(os.listdir(self.event_dir)))
+                expected_msg = "[ProtocolError] [Wireserver Exception] [ProtocolError] [Wireserver Failed] URI http://{0}/machine?comp=telemetrydata  [HTTP Failed] Status Code 503".format(
+                    protocol.get_endpoint())
+                self._assert_error_event_reported(mock_add_event, expected_msg)
+
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
     def test_collect_and_send_with_send_event_generating_exception(self, mock_lib_dir, *args):
@@ -354,34 +366,37 @@ class TestEventMonitoring(AgentTestCase, HttpRequestPredicates):
                 self._create_extension_event(size)
 
             # 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()
+            with patch("azurelinuxagent.ga.monitor.add_event") as mock_add_event:
+                with patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event") as patch_send_event:
+                    test_str = "Test exception, Guid: {0}".format(str(uuid.uuid4()))
+                    patch_send_event.side_effect = Exception(test_str)
 
                     monitor_handler.run_and_wait()
 
-                    self.assertEqual(1, mock_warn.call_count)
                     self.assertEqual(0, len(os.listdir(self.event_dir)))
+                    self._assert_error_event_reported(mock_add_event, test_str)
 
     @patch("azurelinuxagent.common.conf.get_lib_dir")
-    def test_collect_and_send_with_call_wireserver_returns_http_error(self, mock_lib_dir, *args):
+    def test_collect_and_send_with_call_wireserver_returns_http_error_and_reports_event(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 _create_monitor_handler(enabled_operations=["collect_and_send_events"]) as monitor_handler:
+            test_str = "A test exception, Guid: {0}".format(str(uuid.uuid4()))
+
             def http_post_handler(url, _, **__):
                 if self.is_telemetry_request(url):
-                    return HttpError("A test exception")
+                    return HttpError(test_str)
                 return None
 
             monitor_handler.get_mock_wire_protocol().set_http_handlers(http_post_handler=http_post_handler)
 
-            with patch("azurelinuxagent.common.logger.warn") as mock_warn:
+            with patch("azurelinuxagent.ga.monitor.add_event") as mock_add_event:
                 monitor_handler.run_and_wait()
 
-                self.assertEqual(1, mock_warn.call_count)
                 self.assertEqual(0, len(os.listdir(self.event_dir)))
+                self._assert_error_event_reported(mock_add_event, test_str)
 
 
 @patch('azurelinuxagent.common.osutil.get_osutil')
diff --git a/tests/protocol/test_wire.py b/tests/protocol/test_wire.py
index b98f89f0..004ccebf 100644
--- a/tests/protocol/test_wire.py
+++ b/tests/protocol/test_wire.py
@@ -357,12 +357,12 @@ class TestWireProtocol(AgentTestCase):
         self.assertEqual(json.dumps(v1_vm_status), actual.to_json())
 
     @patch("azurelinuxagent.common.utils.restutil.http_request")
-    def test_send_event(self, mock_http_request, *args):
+    def test_send_encoded_event(self, mock_http_request, *args):
         mock_http_request.return_value = MockResponse("", 200)
 
         event_str = u'a test string'
         client = WireProtocol(WIRESERVER_URL).client
-        client.send_event("foo", event_str)
+        client.send_encoded_event("foo", event_str.encode('utf-8'))
 
         first_call = mock_http_request.call_args_list[0]
         args, kwargs = first_call
@@ -371,10 +371,10 @@ class TestWireProtocol(AgentTestCase):
 
         # the headers should include utf-8 encoding...
         self.assertTrue("utf-8" in headers['Content-Type'])
-        # the body is not encoded, just check for equality
-        self.assertIn(event_str, body_received)
+        # the body is encoded, decode and check for equality
+        self.assertIn(event_str, body_received.decode('utf-8'))
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
     def test_report_event_small_event(self, patch_send_event, *args):
         event_list = TelemetryEventList()
         client = WireProtocol(WIRESERVER_URL).client
@@ -396,7 +396,7 @@ class TestWireProtocol(AgentTestCase):
         # It merges the messages into one message
         self.assertEqual(patch_send_event.call_count, 1)
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
     def test_report_event_multiple_events_to_fill_buffer(self, patch_send_event, *args):
         event_list = TelemetryEventList()
         client = WireProtocol(WIRESERVER_URL).client
@@ -410,7 +410,7 @@ class TestWireProtocol(AgentTestCase):
         # It merges the messages into one message
         self.assertEqual(patch_send_event.call_count, 2)
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
     def test_report_event_large_event(self, patch_send_event, *args):
         event_list = TelemetryEventList()
         event_str = random_generator(2 ** 18)