Extend Composite Logger log functions to have built in buffer (#213)

* Composite logger log functions can be extended to have built in buffers

* Moving logic for buffer message in TelemetryWriter

* Add unit tests for write_event_with_buffer

* Simplifed the case of Flush. the case event_level != self.last_telemetry_event_level is already handled and hence should not come in flush case.

* Add test case for code coverage

Author: GAURAVRAMRAKHYANI

src/core/src/bootstrap/Constants.py

@@ -344,3 +344,8 @@ class UbuntuProClientSettings(EnumBackport):
         MAX_OS_MAJOR_VERSION_SUPPORTED = 18
         MINIMUM_CLIENT_VERSION = "27.14.4"
 
+    class BufferMessage(EnumBackport):
+        TRUE = 0
+        FALSE = 1
+        FLUSH = 2
+

src/core/src/local_loggers/CompositeLogger.py

@@ -35,12 +35,12 @@ def __init__(self, env_layer=None, file_logger=None, current_env=None, telemetry
         self.current_env = current_env
         self.NEWLINE_REPLACE_CHAR = " "
 
-    def log(self, message, message_type=Constants.TelemetryEventLevel.Informational):
+    def log(self, message, message_type=Constants.TelemetryEventLevel.Informational, buffer_msg=Constants.BufferMessage.FALSE):
         """log output"""
         message = self.__remove_substring_from_message(message, Constants.ERROR_ADDED_TO_STATUS)
         message = message.strip()
         if self.telemetry_writer is not None and self.telemetry_writer.events_folder_path is not None and self.current_env != Constants.DEV:  # turned off for dev environment as it severely slows down execution
-            self.telemetry_writer.write_event(message, message_type)
+            self.telemetry_writer.write_event_with_buffer(message, message_type, buffer_msg)
         if self.current_env in (Constants.DEV, Constants.TEST):
             for line in message.splitlines():  # allows the extended file logger to strip unnecessary white space
                 print(line)
@@ -60,12 +60,12 @@ def log_warning(self, message):
         message = self.WARNING + (self.NEWLINE_REPLACE_CHAR.join(message.split(os.linesep))).strip()
         self.log(message, message_type=Constants.TelemetryEventLevel.Warning)
 
-    def log_debug(self, message):
+    def log_debug(self, message, buffer_msg=Constants.BufferMessage.FALSE):
         """log debug"""
         message = self.__remove_substring_from_message(message, Constants.ERROR_ADDED_TO_STATUS)
         message = message.strip()
         if self.telemetry_writer is not None and self.telemetry_writer.events_folder_path is not None and self.current_env not in (Constants.DEV, Constants.TEST):
-            self.telemetry_writer.write_event(message, Constants.TelemetryEventLevel.Verbose)
+            self.telemetry_writer.write_event_with_buffer(message, Constants.TelemetryEventLevel.Verbose, buffer_msg)
         if self.current_env in (Constants.DEV, Constants.TEST):
             self.log(self.current_env + ": " + str(self.env_layer.datetime.datetime_utcnow()) + ": " + message, Constants.TelemetryEventLevel.Verbose)  # send to standard output if dev or test env
         elif self.file_logger is not None:
@@ -102,4 +102,3 @@ def __remove_substring_from_message(message, substring=Constants.ERROR_ADDED_TO_
         if substring in message:
             message = message.replace("[{0}]".format(Constants.ERROR_ADDED_TO_STATUS), "")
         return message
-

src/core/src/service_interfaces/TelemetryWriter.py

@@ -28,6 +28,7 @@
 class TelemetryWriter(object):
     """Class for writing telemetry data to data transports"""
 
+    TELEMETRY_BUFFER_DELIMETER= "\n|\t"
 
     def __init__(self, env_layer, composite_logger, events_folder_path, telemetry_supported):
         self.env_layer = env_layer
@@ -48,6 +49,9 @@ def __init__(self, env_layer, composite_logger, events_folder_path, telemetry_su
         self.write_event('Started Linux patch core operation.', Constants.TelemetryEventLevel.Informational)
         self.machine_info = None
         self.set_and_write_machine_config_info()
+        self.telemetry_buffer_store = ""
+        self.last_telemetry_event_level = None
+
 
     def write_config_info(self, config_info, config_type='unknown'):
         # Configuration info
@@ -156,6 +160,33 @@ def __ensure_message_restriction_compliance(self, full_message):
             self.composite_logger.log_telemetry_module_error("Error occurred while formatting message for a telemetry event. [Error={0}]".format(repr(e)))
             raise
 
+    def write_event_with_buffer(self, message, event_level, buffer_msg):
+        if buffer_msg == Constants.BufferMessage.TRUE and (event_level == self.last_telemetry_event_level or self.last_telemetry_event_level is None):
+            if self.telemetry_buffer_store != "":
+                self.telemetry_buffer_store = self.telemetry_buffer_store + self.TELEMETRY_BUFFER_DELIMETER + message
+            else:
+                self.telemetry_buffer_store = message
+
+            self.last_telemetry_event_level = event_level
+
+        elif buffer_msg == Constants.BufferMessage.FALSE or event_level != self.last_telemetry_event_level:
+            if self.telemetry_buffer_store != "":
+                self.write_event(self.telemetry_buffer_store, self.last_telemetry_event_level)
+            self.write_event(message, event_level)
+
+            self.last_telemetry_event_level = None
+            self.telemetry_buffer_store = ""
+
+        elif buffer_msg == Constants.BufferMessage.FLUSH:
+            if self.telemetry_buffer_store != "":
+                self.telemetry_buffer_store = self.telemetry_buffer_store + self.TELEMETRY_BUFFER_DELIMETER + message
+                self.write_event(self.telemetry_buffer_store, self.last_telemetry_event_level)
+            else:
+                self.write_event(message, event_level)
+
+            self.last_telemetry_event_level = None
+            self.telemetry_buffer_store = ""
+
     def write_event(self, message, event_level=Constants.TelemetryEventLevel.Informational, task_name=Constants.TelemetryTaskName.UNKNOWN, is_event_file_throttling_needed=True):
         """ Creates and writes event to event file after validating none of the telemetry size restrictions are breached
         NOTE: is_event_file_throttling_needed is used to determine if event file throttling is required and as such should always be True.

src/core/tests/Test_TelemetryWriter.py

@@ -215,6 +215,82 @@ def test_events_deleted_outside_of_extension_while_extension_is_running(self):
         self.runtime.telemetry_writer.write_event("testing telemetry write to file", Constants.TelemetryEventLevel.Error, "Test Task")
         os.listdir = backup_os_listdir
 
+    def test_write_event_with_buffer_true_and_then_flush(self):
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 1", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.TRUE)
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 2", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.TRUE)
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 3", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.FLUSH)
+
+        latest_event_file = [pos_json for pos_json in os.listdir(self.runtime.telemetry_writer.events_folder_path) if
+                             re.search('^[0-9]+.json$', pos_json)][-1]
+        with open(os.path.join(self.runtime.telemetry_writer.events_folder_path, latest_event_file), 'r+') as f:
+            events = json.load(f)
+            self.assertTrue(events is not None)
+            text_found = re.search('TC=([0-9]+)', events[-1]['Message'])
+            f.close()
+            self.assertTrue(text_found.string.startswith("Message 1 | Message 2 | Message 3"))
+
+    def test_write_event_with_buffer_only_flush(self):
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 1", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.FLUSH)
+
+        latest_event_file = [pos_json for pos_json in os.listdir(self.runtime.telemetry_writer.events_folder_path) if
+                             re.search('^[0-9]+.json$', pos_json)][-1]
+        with open(os.path.join(self.runtime.telemetry_writer.events_folder_path, latest_event_file), 'r+') as f:
+            events = json.load(f)
+            self.assertTrue(events is not None)
+            text_found = re.search('TC=([0-9]+)', events[-1]['Message'])
+            f.close()
+            self.assertTrue(text_found.string.startswith("Message 1"))
+
+    def test_write_event_with_buffer_false(self):
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 1", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.FALSE)
+
+        latest_event_file = [pos_json for pos_json in os.listdir(self.runtime.telemetry_writer.events_folder_path) if
+                             re.search('^[0-9]+.json$', pos_json)][-1]
+        with open(os.path.join(self.runtime.telemetry_writer.events_folder_path, latest_event_file), 'r+') as f:
+            events = json.load(f)
+            self.assertTrue(events is not None)
+            text_found = re.search('TC=([0-9]+)', events[-1]['Message'])
+            f.close()
+            self.assertTrue(text_found.string.startswith("Message 1"))
+
+    def test_write_event_with_buffer_true_and_then_flush_but_different_telemetry_event_level(self):
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 1", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.TRUE)
+
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 2", Constants.TelemetryEventLevel.Informational,
+                                                              Constants.BufferMessage.FLUSH)
+
+        # As the messages are with different TelemetryEventLevel, they will be written separately
+        # even though flush is used.
+        latest_event_file = [pos_json for pos_json in os.listdir(self.runtime.telemetry_writer.events_folder_path) if
+                             re.search('^[0-9]+.json$', pos_json)][-1]
+        with open(os.path.join(self.runtime.telemetry_writer.events_folder_path, latest_event_file), 'r+') as f:
+            events = json.load(f)
+            self.assertTrue(events is not None)
+            text_found = re.search('TC=([0-9]+)', events[-1]['Message'])
+            f.close()
+            self.assertTrue(text_found.string.startswith("Message 2"))
+
+    def test_write_event_with_buffer_true_and_empty_string_and_then_flush_with_non_empty_string(self):
+        self.runtime.telemetry_writer.write_event_with_buffer("", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.TRUE)
+
+        self.runtime.telemetry_writer.write_event_with_buffer("Message 1", Constants.TelemetryEventLevel.Verbose,
+                                                              Constants.BufferMessage.FLUSH)
+
+        latest_event_file = [pos_json for pos_json in os.listdir(self.runtime.telemetry_writer.events_folder_path) if
+                             re.search('^[0-9]+.json$', pos_json)][-1]
+        with open(os.path.join(self.runtime.telemetry_writer.events_folder_path, latest_event_file), 'r+') as f:
+            events = json.load(f)
+            self.assertTrue(events is not None)
+            text_found = re.search('TC=([0-9]+)', events[-1]['Message'])
+            f.close()
+            self.assertTrue(text_found.string.startswith("Message 1"))
 
 if __name__ == '__main__':
     unittest.main()