Repository: Azure/WALinuxAgent
Pull request #3133 (2024-05-30 16:29:58)

Telemetry pipeline improvements (iteration 1)
<!-- DO NOT DELETE THIS TEMPLATE -->

## Description

This pr address below issues to improve data reliability

1. _Sending important events_: Implemented as sending this event directly to wireserver and bypass the collecton_and_send channel. Added new parameter immediate_flush to add_event. Callers need to set to true if this is important event and needs to flush immediately. As a fallback, if we fail to send, we will save to disk for collector to pick up as like a normal event.
2. _Throttling issues/retries on telemetrydata API_:  Today we have aggressive retries on throttling errors. This fix will reduce the number of retries with better delay.
3. _Retry on reading event files_: Collector may pick up and read the file while file is modified. 

I added necessary comments, still questions/concerns happy to go over a call. 

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).
- [ ] If applicable, the PR references the bug/issue that it fixes in the description.
- [ ] New Unit tests were added for the changes made

### 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 e5f980a7..6b9521ca 100644
--- a/azurelinuxagent/common/event.py
+++ b/azurelinuxagent/common/event.py
@@ -213,13 +213,9 @@ def parse_json_event(data_str):
 
 def parse_event(data_str):
     try:
-        try:
-            return parse_json_event(data_str)
-        except ValueError:
-            return parse_xml_event(data_str)
-    except Exception as e:
-        raise EventError("Error parsing event: {0}".format(ustr(e)))
-
+        return parse_json_event(data_str)
+    except ValueError:
+        return parse_xml_event(data_str)
 
 def parse_xml_param(param_node):
     name = getattrib(param_node, "Name")
@@ -344,11 +340,15 @@ class CollectOrReportEventDebugInfo(object):
     def update_op_error(self, op_err):
         self.__op_error_count = self._update_errors_and_get_count(self.__op_error_count, self.__op_errors, op_err)
 
+    def get_error_count(self):
+        return self.__op_error_count + self.__unicode_error_count
+
 
 class EventLogger(object):
     def __init__(self):
         self.event_dir = None
         self.periodic_events = {}
+        self.protocol = None
 
         #
         # All events should have these parameters.
@@ -494,7 +494,10 @@ class EventLogger(object):
             self.periodic_events[h] = datetime.now()
 
     def add_event(self, name, op=WALAEventOperation.Unknown, is_success=True, duration=0, version=str(CURRENT_VERSION),
-                  message="", log_event=True):
+                  message="", log_event=True, flush=False):
+        """
+        :param flush: Flush the event immediately to the wire server
+        """
 
         if (not is_success) and log_event:
             _log_event(name, op, message, duration, is_success=is_success)
@@ -508,12 +511,7 @@ class EventLogger(object):
         event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Duration, int(duration)))
         self.add_common_event_parameters(event, datetime.utcnow())
 
-        data = get_properties(event)
-
-        try:
-            self.save_event(json.dumps(data))
-        except EventError as e:
-            logger.periodic_error(logger.EVERY_FIFTEEN_MINUTES, "[PERIODIC] {0}".format(ustr(e)))
+        self.report_or_save_event(event, flush)
 
     def add_log_event(self, level, message):
         event = TelemetryEvent(TELEMETRY_LOG_EVENT_ID, TELEMETRY_LOG_PROVIDER_ID)
@@ -524,11 +522,7 @@ class EventLogger(object):
         event.parameters.append(TelemetryEventParam(GuestAgentGenericLogsSchema.Context3, ''))
         self.add_common_event_parameters(event, datetime.utcnow())
 
-        data = get_properties(event)
-        try:
-            self.save_event(json.dumps(data))
-        except EventError:
-            pass
+        self.report_or_save_event(event)
 
     def add_metric(self, category, counter, instance, value, log_event=False):
         """
@@ -551,11 +545,25 @@ class EventLogger(object):
         event.parameters.append(TelemetryEventParam(GuestAgentPerfCounterEventsSchema.Value, float(value)))
         self.add_common_event_parameters(event, datetime.utcnow())
 
-        data = get_properties(event)
-        try:
-            self.save_event(json.dumps(data))
-        except EventError as e:
-            logger.periodic_error(logger.EVERY_FIFTEEN_MINUTES, "[PERIODIC] {0}".format(ustr(e)))
+        self.report_or_save_event(event)
+
+    def report_or_save_event(self, event, flush=False):
+        """
+        Flush the event to wireserver if flush to set to true, else
+        save it disk if we fail to send or not required to flush immediately.
+        TODO: pickup as many events as possible and send them in one go.
+        """
+        report_success = False
+        if flush and self.protocol is not None:
+            report_success = self.protocol.report_event([event], flush)
+
+        if not report_success:
+            try:
+                data = get_properties(event)
+                self.save_event(json.dumps(data))
+            except EventError as e:
+                logger.periodic_error(logger.EVERY_FIFTEEN_MINUTES, "[PERIODIC] {0}".format(ustr(e)))
+
 
     @staticmethod
     def _clean_up_message(message):
@@ -636,13 +644,16 @@ def elapsed_milliseconds(utc_start):
                     (d.microseconds / 1000.0))
 
 
-def report_event(op, is_success=True, message='', log_event=True):
+def report_event(op, is_success=True, message='', log_event=True, flush=False):
+    """
+    :param flush: if true, flush the event immediately to the wire server
+    """
     add_event(AGENT_NAME,
               version=str(CURRENT_VERSION),
               is_success=is_success,
               message=message,
               op=op,
-              log_event=log_event)
+              log_event=log_event, flush=flush)
 
 
 def report_periodic(delta, op, is_success=True, message=''):
@@ -675,12 +686,17 @@ def report_metric(category, counter, instance, value, log_event=False, reporter=
                                                      "{0}/{1} [{2}] = {3}".format(category, counter, instance, value))
 
 
-def initialize_event_logger_vminfo_common_parameters(protocol, reporter=__event_logger__):
+def initialize_event_logger_vminfo_common_parameters_and_protocol(protocol, reporter=__event_logger__):
+    # Initialize protocal for event logger to directly send events to wireserver
+    reporter.protocol = protocol
     reporter.initialize_vminfo_common_parameters(protocol)
 
 
 def add_event(name=AGENT_NAME, op=WALAEventOperation.Unknown, is_success=True, duration=0, version=str(CURRENT_VERSION),
-              message="", log_event=True, reporter=__event_logger__):
+              message="", log_event=True, flush=False, reporter=__event_logger__):
+    """
+    :param flush: if true, flush the event immediately to the wire server
+    """
     if reporter.event_dir is None:
         logger.warn("Cannot add event -- Event reporter is not initialized.")
         _log_event(name, op, message, duration, is_success=is_success)
@@ -690,7 +706,7 @@ def add_event(name=AGENT_NAME, op=WALAEventOperation.Unknown, is_success=True, d
         mark_event_status(name, version, op, is_success)
         reporter.add_event(name, op=op, is_success=is_success, duration=duration, version=str(version),
                            message=message,
-                           log_event=log_event)
+                           log_event=log_event, flush=flush)
 
 
 def info(op, fmt, *args):
@@ -723,7 +739,7 @@ def add_log_event(level, message, forced=False, reporter=__event_logger__):
     :param message: Message
     :param forced: Force write the event even if send_logs_to_telemetry() is disabled
         (NOTE: Remove this flag once send_logs_to_telemetry() is enabled for all events)
-    :param reporter:
+    :param reporter: The EventLogger instance to which metric events should be sent
     :return:
     """
     if reporter.event_dir is None:
diff --git a/azurelinuxagent/common/protocol/wire.py b/azurelinuxagent/common/protocol/wire.py
index 751dd2af..6eafd696 100644
--- a/azurelinuxagent/common/protocol/wire.py
+++ b/azurelinuxagent/common/protocol/wire.py
@@ -44,6 +44,8 @@ from azurelinuxagent.common.protocol.restapi import DataContract, ProvisionStatu
 from azurelinuxagent.common.telemetryevent import GuestAgentExtensionEventsSchema
 from azurelinuxagent.common.utils import fileutil, restutil
 from azurelinuxagent.common.utils.cryptutil import CryptUtil
+from azurelinuxagent.common.utils.restutil import TELEMETRY_THROTTLE_DELAY_IN_SECONDS, \
+    TELEMETRY_FLUSH_THROTTLE_DELAY_IN_SECONDS, TELEMETRY_DATA
 from azurelinuxagent.common.utils.textutil import parse_doc, findall, find, \
     findtext, gettext, remove_bom, get_bytes_from_pem, parse_json
 from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
@@ -51,7 +53,7 @@ from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
 VERSION_INFO_URI = "http://{0}/?comp=versions"
 HEALTH_REPORT_URI = "http://{0}/machine?comp=health"
 ROLE_PROP_URI = "http://{0}/machine?comp=roleProperties"
-TELEMETRY_URI = "http://{0}/machine?comp=telemetrydata"
+TELEMETRY_URI = "http://{0}/machine?comp={1}"
 
 PROTOCOL_VERSION = "2012-11-30"
 ENDPOINT_FINE_NAME = "WireServer"
@@ -144,8 +146,8 @@ class WireProtocol(DataContract):
         self.client.status_blob.set_vm_status(vm_status)
         self.client.upload_status_blob()
 
-    def report_event(self, events_iterator):
-        self.client.report_event(events_iterator)
+    def report_event(self, events_iterator, flush=False):
+        return self.client.report_event(events_iterator, flush)
 
     def upload_logs(self, logs):
         self.client.upload_logs(logs)
@@ -1047,8 +1049,8 @@ class WireClient(object):
                                  u",{0}: {1}").format(resp.status,
                                                       resp.read()))
 
-    def send_encoded_event(self, provider_id, event_str, encoding='utf8'):
-        uri = TELEMETRY_URI.format(self.get_endpoint())
+    def _send_encoded_event(self, provider_id, event_str, flush, encoding='utf8'):
+        uri = TELEMETRY_URI.format(self.get_endpoint(), TELEMETRY_DATA)
         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)
@@ -1059,7 +1061,12 @@ class WireClient(object):
             header = self.get_header_for_xml_content()
             # NOTE: The call to wireserver requests utf-8 encoding in the headers, but the body should not
             #       be encoded: some nodes in the telemetry pipeline do not support utf-8 encoding.
-            resp = self.call_wireserver(restutil.http_post, uri, data, header)
+
+            # if it's important event flush, we use less throttle delay(to avoid long delay to complete this operation)) on throttling errors
+            if flush:
+                resp = self.call_wireserver(restutil.http_post, uri, data, header, max_retry=3, throttle_delay=TELEMETRY_FLUSH_THROTTLE_DELAY_IN_SECONDS)
+            else:
+                resp = self.call_wireserver(restutil.http_post, uri, data, header, max_retry=3, throttle_delay=TELEMETRY_THROTTLE_DELAY_IN_SECONDS)
         except HttpError as e:
             raise ProtocolError("Failed to send events:{0}".format(e))
 
@@ -1068,14 +1075,14 @@ class WireClient(object):
             raise ProtocolError(
                 "Failed to send events:{0}".format(resp.status))
 
-    def report_event(self, events_iterator):
+    def report_event(self, events_iterator, flush=False):
         buf = {}
         debug_info = CollectOrReportEventDebugInfo(operation=CollectOrReportEventDebugInfo.OP_REPORT)
         events_per_provider = defaultdict(int)
 
-        def _send_event(provider_id, debug_info):
+        def _send_event(provider_id, debug_info, flush):
             try:
-                self.send_encoded_event(provider_id, buf[provider_id])
+                self._send_encoded_event(provider_id, buf[provider_id], flush)
             except UnicodeError as uni_error:
                 debug_info.update_unicode_error(uni_error)
             except Exception as error:
@@ -1102,7 +1109,7 @@ class WireClient(object):
                 # If buffer is full, send out the events in buffer and reset buffer
                 if len(buf[event.providerId] + event_str) >= MAX_EVENT_BUFFER_SIZE:
                     logger.verbose("No of events this request = {0}".format(events_per_provider[event.providerId]))
-                    _send_event(event.providerId, debug_info)
+                    _send_event(event.providerId, debug_info, flush)
                     buf[event.providerId] = b""
                     events_per_provider[event.providerId] = 0
 
@@ -1117,10 +1124,12 @@ class WireClient(object):
         for provider_id in list(buf.keys()):
             if buf[provider_id]:
                 logger.verbose("No of events this request = {0}".format(events_per_provider[provider_id]))
-                _send_event(provider_id, debug_info)
+                _send_event(provider_id, debug_info, flush)
 
         debug_info.report_debug_info()
 
+        return debug_info.get_error_count() == 0
+
     def report_status_event(self, message, is_success):
         report_event(op=WALAEventOperation.ReportStatus,
                      is_success=is_success,
diff --git a/azurelinuxagent/common/telemetryevent.py b/azurelinuxagent/common/telemetryevent.py
index 51a74938..df370bc9 100644
--- a/azurelinuxagent/common/telemetryevent.py
+++ b/azurelinuxagent/common/telemetryevent.py
@@ -111,4 +111,4 @@ class TelemetryEvent(DataContract):
         for param in self.parameters:
             if param.name == GuestAgentExtensionEventsSchema.Version:
                 return param.value
-        return None
+        return None
\ No newline at end of file
diff --git a/azurelinuxagent/common/utils/restutil.py b/azurelinuxagent/common/utils/restutil.py
index 2180a376..50410086 100644
--- a/azurelinuxagent/common/utils/restutil.py
+++ b/azurelinuxagent/common/utils/restutil.py
@@ -39,6 +39,10 @@ DELAY_IN_SECONDS = 1
 
 THROTTLE_RETRIES = 25
 THROTTLE_DELAY_IN_SECONDS = 1
+# Reducing next attempt calls when throttled since telemetrydata endpoint has a limit 15 calls per 15 secs,
+TELEMETRY_THROTTLE_DELAY_IN_SECONDS = 8
+# Considering short delay for telemetry flush imp events
+TELEMETRY_FLUSH_THROTTLE_DELAY_IN_SECONDS = 2
 
 REDACTED_TEXT = "<SAS_SIGNATURE>"
 SAS_TOKEN_RETRIEVAL_REGEX = re.compile(r'^(https?://[a-zA-Z0-9.].*sig=)([a-zA-Z0-9%-]*)(.*)$')
@@ -109,6 +113,7 @@ REQUEST_ROLE_CONFIG_FILE_NOT_FOUND = "RequestRoleConfigFileNotFound"
 KNOWN_WIRESERVER_IP = '168.63.129.16'
 HOST_PLUGIN_PORT = 32526
 
+TELEMETRY_DATA = "telemetrydata"
 
 class IOErrorCounter(object):
     _lock = threading.RLock()
@@ -163,6 +168,10 @@ def _is_retry_exception(e):
 def _is_throttle_status(status):
     return status in THROTTLE_CODES
 
+def _is_telemetry_req(url):
+    if TELEMETRY_DATA in url:
+        return True
+    return False
 
 def _parse_url(url):
     """
@@ -364,6 +373,7 @@ def http_request(method,
                  max_retry=None,
                  retry_codes=None,
                  retry_delay=DELAY_IN_SECONDS,
+                 throttle_delay=THROTTLE_DELAY_IN_SECONDS,
                  redact_data=False,
                  return_raw_response=False):
     """
@@ -427,10 +437,10 @@ def http_request(method,
             #    (with a safe, minimum number of retry attempts)
             # -- Otherwise, compute a delay that is the product of the next
             #    item in the Fibonacci series and the initial delay value
-            delay = THROTTLE_DELAY_IN_SECONDS \
-                        if was_throttled \
-                        else _compute_delay(retry_attempt=attempt,
-                                            delay=retry_delay)
+            if was_throttled:
+                delay = throttle_delay
+            else:
+                delay = _compute_delay(retry_attempt=attempt, delay=retry_delay)
 
             logger.verbose("[HTTP Retry] "
                         "Attempt {0} of {1} will delay {2} seconds: {3}", 
@@ -468,7 +478,10 @@ def http_request(method,
                     # retry attempts
                     if _is_throttle_status(resp.status):
                         was_throttled = True
-                        max_retry = max(max_retry, THROTTLE_RETRIES)
+                        # Today, THROTTLE_RETRIES is set to a large number (26) for retries, as opposed to backing off and attempting fewer retries.
+                        # However, for telemetry calls (due to throttle limit 15 calls per 15 seconds), we use max_retry set by the caller for overall retry attempts instead of THROTTLE_RETRIES.
+                        if not _is_telemetry_req(url):
+                            max_retry = max(max_retry, THROTTLE_RETRIES)
                     continue
 
             # If we got a 410 (resource gone) for any reason, raise an exception. The caller will handle it by
@@ -563,6 +576,7 @@ def http_post(url,
               max_retry=None,
               retry_codes=None,
               retry_delay=DELAY_IN_SECONDS,
+              throttle_delay=THROTTLE_DELAY_IN_SECONDS,
               timeout=10):
 
     if max_retry is None:
@@ -575,7 +589,8 @@ def http_post(url,
                         use_proxy=use_proxy,
                         max_retry=max_retry,
                         retry_codes=retry_codes,
-                        retry_delay=retry_delay)
+                        retry_delay=retry_delay,
+                        throttle_delay=throttle_delay)
 
 
 def http_put(url,
diff --git a/azurelinuxagent/daemon/main.py b/azurelinuxagent/daemon/main.py
index 3a3923a8..6c0cce40 100644
--- a/azurelinuxagent/daemon/main.py
+++ b/azurelinuxagent/daemon/main.py
@@ -25,7 +25,7 @@ import azurelinuxagent.common.conf as conf
 import azurelinuxagent.common.logger as logger
 import azurelinuxagent.common.utils.fileutil as fileutil
 
-from azurelinuxagent.common.event import add_event, WALAEventOperation, initialize_event_logger_vminfo_common_parameters
+from azurelinuxagent.common.event import add_event, WALAEventOperation, initialize_event_logger_vminfo_common_parameters_and_protocol
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.osutil import get_osutil
 from azurelinuxagent.common.protocol.goal_state import GoalState, GoalStateProperties
@@ -119,7 +119,7 @@ class DaemonHandler(object):
 
     def _initialize_telemetry(self):
         protocol = self.protocol_util.get_protocol()
-        initialize_event_logger_vminfo_common_parameters(protocol)
+        initialize_event_logger_vminfo_common_parameters_and_protocol(protocol)
 
     def daemon(self, child_args=None):
         logger.info("Run daemon")
diff --git a/azurelinuxagent/ga/collect_telemetry_events.py b/azurelinuxagent/ga/collect_telemetry_events.py
index 05f18c60..9c9a1a0f 100644
--- a/azurelinuxagent/ga/collect_telemetry_events.py
+++ b/azurelinuxagent/ga/collect_telemetry_events.py
@@ -21,6 +21,7 @@ import json
 import os
 import re
 import threading
+import time
 from collections import defaultdict
 
 import azurelinuxagent.common.logger as logger
@@ -29,8 +30,8 @@ from azurelinuxagent.common.agent_supported_feature import get_supported_feature
 from azurelinuxagent.common.event import EVENTS_DIRECTORY, TELEMETRY_LOG_EVENT_ID, \
     TELEMETRY_LOG_PROVIDER_ID, add_event, WALAEventOperation, add_log_event, get_event_logger, \
     CollectOrReportEventDebugInfo, EVENT_FILE_REGEX, parse_event
-from azurelinuxagent.common.exception import InvalidExtensionEventError, ServiceStoppedError
-from azurelinuxagent.common.future import ustr
+from azurelinuxagent.common.exception import InvalidExtensionEventError, ServiceStoppedError, EventError
+from azurelinuxagent.common.future import ustr, is_file_not_found_error
 from azurelinuxagent.ga.interfaces import ThreadHandlerInterface
 from azurelinuxagent.common.telemetryevent import TelemetryEvent, TelemetryEventParam, \
     GuestAgentGenericLogsSchema, GuestAgentExtensionEventsSchema
@@ -38,6 +39,10 @@ from azurelinuxagent.common.utils import textutil
 from azurelinuxagent.ga.exthandlers import HANDLER_NAME_PATTERN
 from azurelinuxagent.ga.periodic_operation import PeriodicOperation
 
+# Event file specific retries and delays.
+NUM_OF_EVENT_FILE_RETRIES = 3
+EVENT_FILE_RETRY_DELAY = 1  # seconds
+
 
 def get_collect_telemetry_events_handler(send_telemetry_events_handler):
     return CollectTelemetryEventsHandler(send_telemetry_events_handler)
@@ -248,17 +253,39 @@ class _ProcessExtensionEvents(PeriodicOperation):
                                      ustr(error))
                         log_err = False
 
+    @staticmethod
+    def _read_event_file(event_file_path):
+        """
+        Read the event file and return the data.
+        :param event_file_path: Full path of the event file.
+        :return: Event data in list or string format.
+        """
+        # Retry for reading the event file in case file is modified while reading
+        # We except FileNotFoundError and ValueError to handle the case where the file is deleted or modified while reading
+        error_count = 0
+        while True:
+            try:
+                # Read event file and decode it properly
+                with open(event_file_path, "rb") as event_file_descriptor:
+                    event_data = event_file_descriptor.read().decode("utf-8")
+
+                # Parse the string and get the list of events
+                return json.loads(event_data)
+            except Exception as e:
+                if is_file_not_found_error(e) or isinstance(e, ValueError):
+                    error_count += 1
+                    if error_count >= NUM_OF_EVENT_FILE_RETRIES:
+                        raise
+                else:
+                    raise
+            time.sleep(EVENT_FILE_RETRY_DELAY)
+
     def _enqueue_events_and_get_count(self, handler_name, event_file_path, captured_events_count,
                                       dropped_events_with_error_count):
 
         event_file_time = datetime.datetime.fromtimestamp(os.path.getmtime(event_file_path))
 
-        # Read event file and decode it properly
-        with open(event_file_path, "rb") as event_file_descriptor:
-            event_data = event_file_descriptor.read().decode("utf-8")
-
-        # Parse the string and get the list of events
-        events = json.loads(event_data)
+        events = self._read_event_file(event_file_path)
 
         # We allow multiple events in a file but there can be an instance where the file only has a single
         # JSON event and not a list. Handling that condition too
@@ -431,10 +458,7 @@ class _CollectAndEnqueueEvents(PeriodicOperation):
                 try:
                     logger.verbose("Processing event file: {0}", event_file_path)
 
-                    with open(event_file_path, "rb") as event_fd:
-                        event_data = event_fd.read().decode("utf-8")
-
-                    event = parse_event(event_data)
+                    event = self._read_and_parse_event_file(event_file_path)
 
                     # "legacy" events are events produced by previous versions of the agent (<= 2.2.46) and extensions;
                     # they do not include all the telemetry fields, so we add them here
@@ -458,7 +482,8 @@ class _CollectAndEnqueueEvents(PeriodicOperation):
                     # Todo: We should delete files after ensuring that we sent the data to Wireserver successfully
                     # from our end rather than deleting first and sending later. This is to ensure the data reliability
                     # of the agent telemetry pipeline.
-                    os.remove(event_file_path)
+                    if os.path.exists(event_file_path):
+                        os.remove(event_file_path)
             except ServiceStoppedError as stopped_error:
                 logger.error(
                     "Unable to enqueue events as service stopped: {0}, skipping events collection".format(
@@ -470,6 +495,30 @@ class _CollectAndEnqueueEvents(PeriodicOperation):
 
         debug_info.report_debug_info()
 
+    @staticmethod
+    def _read_and_parse_event_file(event_file_path):
+        """
+        Read the event file and parse it to a TelemetryEvent object.
+        :param event_file_path: Full path of the event file.
+        :return: TelemetryEvent object.
+        """
+        # Retry for reading the event file in case file is modified while reading
+        # We except FileNotFoundError and ValueError to handle the case where the file is deleted or modified while reading
+        error_count = 0
+        while True:
+            try:
+                with open(event_file_path, "rb") as event_fd:
+                    event_data = event_fd.read().decode("utf-8")
+                return parse_event(event_data)
+            except Exception as e:
+                if is_file_not_found_error(e) or isinstance(e, ValueError):
+                    error_count += 1
+                    if error_count >= NUM_OF_EVENT_FILE_RETRIES:
+                        raise
+                else:
+                    raise EventError("Error parsing event: {0}".format(ustr(e)))
+            time.sleep(EVENT_FILE_RETRY_DELAY)
+
     @staticmethod
     def _update_legacy_agent_event(event, event_creation_time):
         # Ensure that if an agent event is missing a field from the schema defined since 2.2.47, the missing fields
@@ -583,4 +632,4 @@ class CollectTelemetryEventsHandler(ThreadHandlerInterface):
     @staticmethod
     def add_common_params_to_telemetry_event(event, event_time):
         reporter = get_event_logger()
-        reporter.add_common_event_parameters(event, event_time)
+        reporter.add_common_event_parameters(event, event_time)
\ No newline at end of file
diff --git a/azurelinuxagent/ga/logcollector.py b/azurelinuxagent/ga/logcollector.py
index ffcc79ad..5de4f029 100644
--- a/azurelinuxagent/ga/logcollector.py
+++ b/azurelinuxagent/ga/logcollector.py
@@ -27,7 +27,7 @@ from datetime import datetime
 from heapq import heappush, heappop
 
 from azurelinuxagent.common.conf import get_lib_dir, get_ext_log_dir, get_agent_log_file
-from azurelinuxagent.common.event import initialize_event_logger_vminfo_common_parameters, add_event, WALAEventOperation
+from azurelinuxagent.common.event import initialize_event_logger_vminfo_common_parameters_and_protocol, add_event, WALAEventOperation
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.ga.logcollector_manifests import MANIFEST_NORMAL, MANIFEST_FULL
 
@@ -107,7 +107,7 @@ class LogCollector(object):
         protocol = get_protocol_util().get_protocol(init_goal_state=False)
         protocol.client.reset_goal_state(goal_state_properties=GoalStateProperties.RoleConfig | GoalStateProperties.HostingEnv)
         # Initialize the common parameters for telemetry events
-        initialize_event_logger_vminfo_common_parameters(protocol)
+        initialize_event_logger_vminfo_common_parameters_and_protocol(protocol)
 
     @staticmethod
     def _run_shell_command(command, stdout=subprocess.PIPE, log_output=False):
diff --git a/azurelinuxagent/ga/update.py b/azurelinuxagent/ga/update.py
index 34882035..b2f5d03b 100644
--- a/azurelinuxagent/ga/update.py
+++ b/azurelinuxagent/ga/update.py
@@ -36,7 +36,8 @@ from azurelinuxagent.common.utils import fileutil, textutil
 from azurelinuxagent.common.agent_supported_feature import get_supported_feature_by_name, SupportedFeatureNames, \
     get_agent_supported_features_list_for_crp
 from azurelinuxagent.ga.cgroupconfigurator import CGroupConfigurator
-from azurelinuxagent.common.event import add_event, initialize_event_logger_vminfo_common_parameters, WALAEventOperation, EVENTS_DIRECTORY
+from azurelinuxagent.common.event import add_event, initialize_event_logger_vminfo_common_parameters_and_protocol, \
+    WALAEventOperation, EVENTS_DIRECTORY
 from azurelinuxagent.common.exception import ExitException, AgentUpgradeExitException, AgentMemoryExceededException
 from azurelinuxagent.ga.firewall_manager import FirewallManager, FirewallStateError
 from azurelinuxagent.common.future import ustr
@@ -342,7 +343,7 @@ class UpdateHandler(object):
             self._initialize_goal_state(protocol)
 
             # Initialize the common parameters for telemetry events
-            initialize_event_logger_vminfo_common_parameters(protocol)
+            initialize_event_logger_vminfo_common_parameters_and_protocol(protocol)
 
             # Send telemetry for the OS-specific info.
             add_event(AGENT_NAME, op=WALAEventOperation.OSInfo, message=os_info_msg)


Test patch:
diff --git a/tests/common/protocol/test_wire.py b/tests/common/protocol/test_wire.py
index 2fbd817f..3dc04435 100644
--- a/tests/common/protocol/test_wire.py
+++ b/tests/common/protocol/test_wire.py
@@ -421,7 +421,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
 
         event_str = u'a test string'
         client = WireProtocol(WIRESERVER_URL).client
-        client.send_encoded_event("foo", event_str.encode('utf-8'))
+        client._send_encoded_event("foo", event_str.encode('utf-8'), flush=False)
 
         first_call = mock_http_request.call_args_list[0]
         args, kwargs = first_call
@@ -433,7 +433,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
         # 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_encoded_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient._send_encoded_event")
     def test_report_event_small_event(self, patch_send_event, *args):  # pylint: disable=unused-argument
         event_list = []
         client = WireProtocol(WIRESERVER_URL).client
@@ -455,7 +455,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
         # It merges the messages into one message
         self.assertEqual(patch_send_event.call_count, 1)
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient._send_encoded_event")
     def test_report_event_multiple_events_to_fill_buffer(self, patch_send_event, *args):  # pylint: disable=unused-argument
         event_list = []
         client = WireProtocol(WIRESERVER_URL).client
@@ -469,7 +469,7 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
         # It merges the messages into one message
         self.assertEqual(patch_send_event.call_count, 2)
 
-    @patch("azurelinuxagent.common.protocol.wire.WireClient.send_encoded_event")
+    @patch("azurelinuxagent.common.protocol.wire.WireClient._send_encoded_event")
     def test_report_event_large_event(self, patch_send_event, *args):  # pylint: disable=unused-argument
         event_list = []
         event_str = random_generator(2 ** 18)
@@ -478,6 +478,24 @@ class TestWireProtocol(AgentTestCase, HttpRequestPredicates):
         client.report_event(self._get_telemetry_events_generator(event_list))
 
         self.assertEqual(patch_send_event.call_count, 0)
+        
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_report_event_http_req_should_do_max_retries_on_throttling_error(self, mock_http_request, *args):  # pylint: disable=unused-argument
+        mock_http_request.return_value = MockHttpResponse(429)
+        event_list = []
+        event_str = random_generator(2 ** 15)
+        event_list.append(get_event(message=event_str))
+        client = WireProtocol(WIRESERVER_URL).client
+        with patch("azurelinuxagent.common.utils.restutil.TELEMETRY_THROTTLE_DELAY_IN_SECONDS", 0.001):
+            client.report_event(self._get_telemetry_events_generator(event_list))
+            self.assertEqual(mock_http_request.call_count, 3)
+
+        mock_http_request.reset_mock()
+        self.assertEqual(mock_http_request.call_count, 0)
+        mock_http_request.return_value = MockHttpResponse(429)
+        with patch("azurelinuxagent.common.utils.restutil.TELEMETRY_FLUSH_THROTTLE_DELAY_IN_SECONDS", 0.001):
+            client.report_event(self._get_telemetry_events_generator(event_list), flush=True)
+            self.assertEqual(mock_http_request.call_count, 3)
 
     def test_get_header_for_cert_should_use_triple_des(self, *_):
         with mock_wire_protocol(wire_protocol_data.DATA_FILE) as protocol:
diff --git a/tests/common/test_event.py b/tests/common/test_event.py
index 51eb1572..15a85ce1 100644
--- a/tests/common/test_event.py
+++ b/tests/common/test_event.py
@@ -171,7 +171,6 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
             contained_id = create_event_and_return_container_id()
             self.assertEqual(contained_id, '11111111-2222-3333-4444-555555555555', "Incorrect container ID")
 
-
     def test_add_event_should_handle_event_errors(self):
         with patch("azurelinuxagent.common.utils.fileutil.mkdir", side_effect=OSError):
             with patch('azurelinuxagent.common.logger.periodic_error') as mock_logger_periodic_error:
@@ -375,7 +374,7 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
         add_event('test', message='test event')
         mock_add_event.assert_called_once_with('test', duration=0, is_success=True, log_event=True,
                                                message='test event', op=WALAEventOperation.Unknown,
-                                               version=str(CURRENT_VERSION))
+                                               version=str(CURRENT_VERSION), flush=False)
 
     def test_collect_events_should_delete_event_files(self):
         add_event(name='Event1', op=TestEvent._Operation)
@@ -400,6 +399,39 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
             self.assertTrue(filename.endswith(AGENT_EVENT_FILE_EXTENSION),
                 'Event file does not have the correct extension ({0}): {1}'.format(AGENT_EVENT_FILE_EXTENSION, filename))
 
+    def test_add_event_flush_immediately(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(wire_protocol_data.DATA_FILE, http_post_handler=http_post_handler):
+            expected_message = 'test event'
+            add_event('test', message=expected_message, op=TestEvent._Operation, flush=True)
+
+            event_message = self._get_event_message_from_http_request_body(http_post_handler.request_body)
+
+            self.assertEqual(event_message, expected_message,
+                         "The Message in the HTTP request does not match the Message in the add_event")
+
+            # If immediate_flush is set, the event should send to wireserver directly and file should not be created
+            self.assertTrue(len(self._collect_event_files()) == 0)
+
+    def test_add_event_flush_fails(self):
+        def http_post_handler(url, **__):
+            if self.is_telemetry_request(url):
+                return MockHttpResponse(status=500)
+            return None
+
+        with mock_wire_protocol(wire_protocol_data.DATA_FILE, http_post_handler=http_post_handler):
+            expected_message = 'test event'
+            add_event('test', message=expected_message, op=TestEvent._Operation, flush=True)
+
+            # In case of failure, the event file should be created
+            self.assertTrue(len(self._collect_event_files()) == 1)
+
     @staticmethod
     def _get_event_message(evt):
         for p in evt.parameters:
@@ -424,27 +456,29 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
         self._create_test_event_file("custom_script_2.tld")  # another valid event
 
         with patch("azurelinuxagent.common.event.add_event") as mock_add_event:
-            event_list = self._collect_events()
+            # mock the max retries on parsing invalid json to avoid the test run delays
+            with patch("azurelinuxagent.ga.collect_telemetry_events.NUM_OF_EVENT_FILE_RETRIES", 1):
+                event_list = self._collect_events()
 
-            self.assertEqual(
-                len(event_list), 2)
-            self.assertTrue(
-                all(TestEvent._get_event_message(evt) == "A test telemetry message." for evt in event_list),
-                "The valid events were not found")
-
-            invalid_events = []
-            total_dropped_count = 0
-            for args, kwargs in mock_add_event.call_args_list:  # pylint: disable=unused-variable
-                match = re.search(r"DroppedEventsCount: (\d+)", kwargs['message'])
-                if match is not None:
-                    invalid_events.append(kwargs['op'])
-                    total_dropped_count += int(match.groups()[0])
-
-            self.assertEqual(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))
+                self.assertEqual(
+                    len(event_list), 2)
+                self.assertTrue(
+                    all(TestEvent._get_event_message(evt) == "A test telemetry message." for evt in event_list),
+                    "The valid events were not found")
+
+                invalid_events = []
+                total_dropped_count = 0
+                for args, kwargs in mock_add_event.call_args_list:  # pylint: disable=unused-variable
+                    match = re.search(r"DroppedEventsCount: (\d+)", kwargs['message'])
+                    if match is not None:
+                        invalid_events.append(kwargs['op'])
+                        total_dropped_count += int(match.groups()[0])
+
+                self.assertEqual(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.
@@ -638,6 +672,9 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
         shutil.copy(source_file_path, target_file_path)
         return target_file_path
 
+    def _collect_test_event_files(self, file_name):
+        return [os.path.join(self.event_dir, f) for f in os.listdir(self.event_dir) if file_name in f]
+
     @staticmethod
     def _get_file_creation_timestamp(file):  # pylint: disable=redefined-builtin
         return  TestEvent._datetime_to_event_timestamp(datetime.fromtimestamp(os.path.getmtime(file)))
@@ -733,6 +770,39 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
     def test_collect_events_should_ignore_extra_parameters_in_extension_events(self):
         self._assert_extension_event_includes_all_parameters_in_the_telemetry_schema('custom_script_extra_parameters.tld')
 
+    @staticmethod
+    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")
+        if event_node is None:
+            raise ValueError('Could not find the Event node in the XML document')
+        if len(event_node.childNodes) != 1:
+            raise ValueError('The Event node in the XML document should have exactly 1 child')
+
+        event_node_first_child = event_node.childNodes[0]
+        if event_node_first_child.nodeType != xml.dom.Node.CDATA_SECTION_NODE:
+            raise ValueError('The Event node contents should be CDATA')
+
+        event_node_cdata = event_node_first_child.nodeValue
+
+        # The CDATA will contain a sequence of "<Param Name='foo' Value='bar'/>" nodes, which
+        # correspond to the parameters of the telemetry event.  Wrap those into a "Helper" node
+        # and extract the "Message"
+        event_xml_text = '<?xml version="1.0"?><Helper>{0}</Helper>'.format(event_node_cdata)
+        event_xml_doc = textutil.parse_doc(event_xml_text)
+        helper_node = textutil.find(event_xml_doc, "Helper")
+
+        for child in helper_node.childNodes:
+            if child.getAttribute('Name') == GuestAgentExtensionEventsSchema.Message:
+                return child.getAttribute('Value')
+
+        raise ValueError(
+            'Could not find the Message for the telemetry event. Request body: {0}'.format(http_request_body))
+
     def test_report_event_should_encode_call_stack_correctly(self):
         """
         The Message in some telemetry events that include call stacks are being truncated in Kusto. While the issue doesn't seem
@@ -750,37 +820,6 @@ 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(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")
-            if event_node is None:
-                raise ValueError('Could not find the Event node in the XML document')
-            if len(event_node.childNodes) != 1:
-                raise ValueError('The Event node in the XML document should have exactly 1 child')
-
-            event_node_first_child = event_node.childNodes[0]
-            if event_node_first_child.nodeType != xml.dom.Node.CDATA_SECTION_NODE:
-                raise ValueError('The Event node contents should be CDATA')
-
-            event_node_cdata = event_node_first_child.nodeValue
-
-            # The CDATA will contain a sequence of "<Param Name='foo' Value='bar'/>" nodes, which
-            # correspond to the parameters of the telemetry event.  Wrap those into a "Helper" node
-            # and extract the "Message"
-            event_xml_text = '<?xml version="1.0"?><Helper>{0}</Helper>'.format(event_node_cdata)
-            event_xml_doc = textutil.parse_doc(event_xml_text)
-            helper_node = textutil.find(event_xml_doc, "Helper")
-
-            for child in helper_node.childNodes:
-                if child.getAttribute('Name') == GuestAgentExtensionEventsSchema.Message:
-                    return child.getAttribute('Value')
-
-            raise ValueError('Could not find the Message for the telemetry event. Request body: {0}'.format(http_request_body))
-
         def http_post_handler(url, body, **__):
             if self.is_telemetry_request(url):
                 http_post_handler.request_body = body
@@ -795,7 +834,7 @@ class TestEvent(HttpRequestPredicates, AgentTestCase):
             event_list = self._collect_events()
             self._report_events(protocol, event_list)
 
-            event_message = get_event_message_from_http_request_body(http_post_handler.request_body)
+            event_message = self._get_event_message_from_http_request_body(http_post_handler.request_body)
 
             self.assertEqual(event_message, expected_message, "The Message in the HTTP request does not match the Message in the event's *.tld file")
 
diff --git a/tests/common/utils/test_rest_util.py b/tests/common/utils/test_rest_util.py
index efcebb08..ccde2023 100644
--- a/tests/common/utils/test_rest_util.py
+++ b/tests/common/utils/test_rest_util.py
@@ -659,6 +659,23 @@ class TestHttpOperations(AgentTestCase):
             [call(1) for i in range(restutil.THROTTLE_RETRIES-1)],
             _sleep.call_args_list)
 
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_try_max_retries_when_telemetry_throttled(self, _http_request, _sleep):
+        # Ensure the code is a throttle code
+        self.assertTrue(httpclient.SERVICE_UNAVAILABLE in restutil.THROTTLE_CODES)
+        max_retries = 5
+        _http_request.side_effect = [
+                Mock(status=httpclient.SERVICE_UNAVAILABLE)
+                    for i in range(max_retries-1)  # pylint: disable=unused-variable
+            ] + [Mock(status=httpclient.OK)]
+
+        restutil.http_post("https://telemetrydata", data=None,
+                            max_retry=max_retries)
+        self.assertEqual(max_retries, _http_request.call_count)
+        self.assertEqual(max_retries-1, _sleep.call_count)
+
+
     @patch("time.sleep")
     @patch("azurelinuxagent.common.utils.restutil._http_request")
     def test_http_request_raises_for_resource_gone(self, _http_request, _sleep):
diff --git a/tests/ga/test_collect_telemetry_events.py b/tests/ga/test_collect_telemetry_events.py
index 509af2ce..a69baf10 100644
--- a/tests/ga/test_collect_telemetry_events.py
+++ b/tests/ga/test_collect_telemetry_events.py
@@ -167,14 +167,15 @@ class TestExtensionTelemetryHandler(AgentTestCase, HttpRequestPredicates):
     @contextlib.contextmanager
     def _create_extension_telemetry_processor(self, telemetry_handler=None):
 
-        event_list = []
-        if not telemetry_handler:
-            telemetry_handler = MagicMock(autospec=True)
-            telemetry_handler.stopped = MagicMock(return_value=False)
-            telemetry_handler.enqueue_event = MagicMock(wraps=event_list.append)
-        extension_telemetry_processor = _ProcessExtensionEvents(telemetry_handler)
-        extension_telemetry_processor.event_list = event_list
-        yield extension_telemetry_processor
+        with patch("azurelinuxagent.ga.collect_telemetry_events.NUM_OF_EVENT_FILE_RETRIES", 1):
+            event_list = []
+            if not telemetry_handler:
+                telemetry_handler = MagicMock(autospec=True)
+                telemetry_handler.stopped = MagicMock(return_value=False)
+                telemetry_handler.enqueue_event = MagicMock(wraps=event_list.append)
+            extension_telemetry_processor = _ProcessExtensionEvents(telemetry_handler)
+            extension_telemetry_processor.event_list = event_list
+            yield extension_telemetry_processor
 
     def _assert_handler_data_in_event_list(self, telemetry_events, ext_names_with_count, expected_count=None):
 
@@ -573,4 +574,4 @@ class TestExtensionTelemetryHandler(AgentTestCase, HttpRequestPredicates):
                 self.assertGreater(file_count, 0, "Some event files should still be there")
                 total_file_count += file_count
 
-            self.assertEqual(expected_event_file_count, total_file_count, "Expected File count doesn't match")
+            self.assertEqual(expected_event_file_count, total_file_count, "Expected File count doesn't match")
\ No newline at end of file
diff --git a/tests/ga/test_multi_config_extension.py b/tests/ga/test_multi_config_extension.py
index 450ca071..482126be 100644
--- a/tests/ga/test_multi_config_extension.py
+++ b/tests/ga/test_multi_config_extension.py
@@ -146,7 +146,6 @@ class _MultiConfigBaseTestClass(AgentTestCase):
 
     @contextlib.contextmanager
     def _setup_test_env(self, mock_manifest=False):
-
         with mock_wire_protocol(self.test_data) as protocol:
             def mock_http_put(url, *args, **_):
                 if HttpRequestPredicates.is_host_plugin_status_request(url):
diff --git a/tests/ga/test_send_telemetry_events.py b/tests/ga/test_send_telemetry_events.py
index 1d15b4ff..4ffc37c2 100644
--- a/tests/ga/test_send_telemetry_events.py
+++ b/tests/ga/test_send_telemetry_events.py
@@ -81,7 +81,6 @@ class TestSendTelemetryEventsHandler(AgentTestCase, HttpRequestPredicates):
             with patch("azurelinuxagent.ga.send_telemetry_events.SendTelemetryEventsHandler._MIN_EVENTS_TO_BATCH",
                        batching_queue_limit):
                 with patch("azurelinuxagent.ga.send_telemetry_events.SendTelemetryEventsHandler._MAX_TIMEOUT", timeout):
-
                     send_telemetry_events_handler.get_mock_wire_protocol = lambda: protocol
                     if start_thread:
                         send_telemetry_events_handler.start()
@@ -427,4 +426,4 @@ class TestSendTelemetryEventsHandler(AgentTestCase, HttpRequestPredicates):
 
     @staticmethod
     def _get_extension_events(telemetry_handler):
-        return [event_xml for _, event_xml in telemetry_handler.event_calls if TestSendTelemetryEventsHandler._TEST_EVENT_OPERATION in event_xml.decode()]
+        return [event_xml for _, event_xml in telemetry_handler.event_calls if TestSendTelemetryEventsHandler._TEST_EVENT_OPERATION in event_xml.decode()]
\ No newline at end of file
diff --git a/tests/ga/test_update.py b/tests/ga/test_update.py
index ccfc1387..2f201d21 100644
--- a/tests/ga/test_update.py
+++ b/tests/ga/test_update.py
@@ -99,7 +99,6 @@ def _get_update_handler(iterations=1, test_data=None, protocol=None, autoupdate_
     This function returns a mocked version of the UpdateHandler object to be used for testing. It will only run the
     main loop [iterations] no of times.
     """
-
     test_data = DATA_FILE if test_data is None else test_data
 
     with patch.object(HostPluginProtocol, "is_default_channel", False):
@@ -1256,7 +1255,7 @@ class UpdateHandlerRunTestCase(AgentTestCase):
                     with patch('azurelinuxagent.ga.update.get_collect_logs_handler') as mock_collect_logs:
                         with patch('azurelinuxagent.ga.update.get_send_telemetry_events_handler') as mock_telemetry_send_events:
                             with patch('azurelinuxagent.ga.update.get_collect_telemetry_events_handler') as mock_event_collector:
-                                with patch('azurelinuxagent.ga.update.initialize_event_logger_vminfo_common_parameters'):
+                                with patch('azurelinuxagent.ga.update.initialize_event_logger_vminfo_common_parameters_and_protocol'):
                                     with patch('azurelinuxagent.ga.update.is_log_collection_allowed', return_value=True):
                                         with mock_wire_protocol(DATA_FILE) as protocol:
                                             mock_exthandlers_handler = Mock()
@@ -1804,7 +1803,7 @@ class MonitorThreadTest(AgentTestCase):
                     with patch('azurelinuxagent.ga.exthandlers.get_exthandlers_handler'):
                         with patch('azurelinuxagent.ga.remoteaccess.get_remote_access_handler'):
                             with patch('azurelinuxagent.ga.agent_update_handler.get_agent_update_handler'):
-                                with patch('azurelinuxagent.ga.update.initialize_event_logger_vminfo_common_parameters'):
+                                with patch('azurelinuxagent.ga.update.initialize_event_logger_vminfo_common_parameters_and_protocol'):
                                     with patch('azurelinuxagent.ga.cgroupapi.CGroupUtil.cgroups_supported', return_value=False):  # skip all cgroup stuff
                                         with patch('azurelinuxagent.ga.update.is_log_collection_allowed', return_value=True):
                                             with patch('time.sleep'):
diff --git a/tests/lib/event_logger_tools.py b/tests/lib/event_logger_tools.py
index 5150cebd..e96e8414 100644
--- a/tests/lib/event_logger_tools.py
+++ b/tests/lib/event_logger_tools.py
@@ -55,7 +55,7 @@ class EventLoggerTools(object):
 
         with mock_wire_protocol(wire_protocol_data.DATA_FILE) as mock_protocol:
             with tools.patch("azurelinuxagent.common.event.get_imds_client", return_value=mock_imds_client):
-                event.initialize_event_logger_vminfo_common_parameters(mock_protocol)
+                event.initialize_event_logger_vminfo_common_parameters_and_protocol(mock_protocol)
 
     @staticmethod
     def get_expected_os_version():