Pushing cloud-init log to the KVP (#529)

Push the cloud-init.log file (Up to 500KB at once) to the KVP before reporting ready to the Azure platform.
Based on the analysis done on a large sample of cloud-init.log files, Here's the statistics collected on the log file size:
P50 	P90 	P95 	P99 	P99.9 	P99.99
137K 	423K 	537K 	3.5MB 	6MB 	16MB

This change limits the size of cloud-init.log file data that gets dumped to KVP to 500KB. So for ~95% of the cases, the whole log file will be dumped and for the remaining ~5%, we will get the last 500KB of the cloud-init.log file.
To asses the performance of the 500KB limit, 250 VM were deployed with a 500KB cloud-init.log file and the time taken to compress, encode and dump the entries to KVP was measured. Here's the time in milliseconds percentiles:
P50 	P99 	P999
75.705 	232.701 	1169.636

Another 250 VMs were deployed with this logic dumping their normal cloud-init.log file to KVP, the same timing was measured as above. Here's the time in milliseconds percentiles:
P50 	P99 	P999
1.88 	5.277 	6.992

Added excluded_handlers to the report_event function to be able to opt-out from reporting the events of the compressed cloud-init.log file to the cloud-init.log file.

The KVP break_down logic had a bug, where it will reuse the same key for all the split chunks of KVP which results in overwriting the split KVPs by the last one when consumed by Hyper-V. I added the split chunk index as a differentiator to the KVP key.

The Hyper-V consumes the KVPs from the KVP file as chunks whose key is 512KB and value is 2048KB but the Azure platform expects the value to be 1024KB, thus I introduced the Azure value limit.

1 files changed, 71 insertions, 3 deletions

diff --git a/tests/unittests/test_reporting_hyperv.py b/tests/unittests/test_reporting_hyperv.py
index bacf5da9..47ede670 100644
--- a/tests/unittests/test_reporting_hyperv.py
+++ b/tests/unittests/test_reporting_hyperv.py
@@ -1,7 +1,9 @@
 # This file is part of cloud-init. See LICENSE file for license information.
+import base64
+import zlib
 
-from cloudinit.reporting import events
-from cloudinit.reporting.handlers import HyperVKvpReportingHandler
+from cloudinit.reporting import events, instantiated_handler_registry
+from cloudinit.reporting.handlers import HyperVKvpReportingHandler, LogHandler
 
 import json
 import os
@@ -72,7 +74,7 @@ class TextKvpReporter(CiTestCase):
     def test_event_very_long(self):
         reporter = HyperVKvpReportingHandler(
             kvp_file_path=self.tmp_file_path)
-        description = 'ab' * reporter.HV_KVP_EXCHANGE_MAX_VALUE_SIZE
+        description = 'ab' * reporter.HV_KVP_AZURE_MAX_VALUE_SIZE
         long_event = events.FinishReportingEvent(
             'event_name',
             description,
@@ -199,6 +201,72 @@ class TextKvpReporter(CiTestCase):
         if "test_diagnostic" not in evt_msg:
             raise AssertionError("missing expected diagnostic message")
 
+    def test_report_compressed_event(self):
+        reporter = HyperVKvpReportingHandler(kvp_file_path=self.tmp_file_path)
+        try:
+            instantiated_handler_registry.register_item("telemetry", reporter)
+            event_desc = b'test_compressed'
+            azure.report_compressed_event(
+                "compressed event", event_desc)
+
+            self.validate_compressed_kvps(reporter, 1, [event_desc])
+        finally:
+            instantiated_handler_registry.unregister_item("telemetry",
+                                                          force=False)
+
+    @mock.patch.object(LogHandler, 'publish_event')
+    def test_push_log_to_kvp(self, publish_event):
+        reporter = HyperVKvpReportingHandler(kvp_file_path=self.tmp_file_path)
+        try:
+            instantiated_handler_registry.register_item("telemetry", reporter)
+            log_file = self.tmp_path("cloud-init.log")
+            azure.MAX_LOG_TO_KVP_LENGTH = 100
+            azure.LOG_PUSHED_TO_KVP_MARKER_FILE = self.tmp_path(
+                'log_pushed_to_kvp')
+            with open(log_file, "w") as f:
+                log_content = "A" * 50 + "B" * 100
+                f.write(log_content)
+            azure.push_log_to_kvp(log_file)
+
+            with open(log_file, "a") as f:
+                extra_content = "C" * 10
+                f.write(extra_content)
+            azure.push_log_to_kvp(log_file)
+
+            for call_arg in publish_event.call_args_list:
+                event = call_arg[0][0]
+                self.assertNotEqual(
+                    event.event_type, azure.COMPRESSED_EVENT_TYPE)
+            self.validate_compressed_kvps(
+                reporter, 1,
+                [log_content[-azure.MAX_LOG_TO_KVP_LENGTH:].encode()])
+        finally:
+            instantiated_handler_registry.unregister_item("telemetry",
+                                                          force=False)
+
+    def validate_compressed_kvps(self, reporter, count, values):
+        reporter.q.join()
+        kvps = list(reporter._iterate_kvps(0))
+        compressed_count = 0
+        for i in range(len(kvps)):
+            kvp = kvps[i]
+            kvp_value = kvp['value']
+            kvp_value_json = json.loads(kvp_value)
+            evt_msg = kvp_value_json["msg"]
+            evt_type = kvp_value_json["type"]
+            if evt_type != azure.COMPRESSED_EVENT_TYPE:
+                continue
+            evt_msg_json = json.loads(evt_msg)
+            evt_encoding = evt_msg_json["encoding"]
+            evt_data = zlib.decompress(
+                base64.decodebytes(evt_msg_json["data"].encode("ascii")))
+
+            self.assertLess(compressed_count, len(values))
+            self.assertEqual(evt_data, values[compressed_count])
+            self.assertEqual(evt_encoding, "gz+b64")
+            compressed_count += 1
+        self.assertEqual(compressed_count, count)
+
     def test_unique_kvp_key(self):
         reporter = HyperVKvpReportingHandler(kvp_file_path=self.tmp_file_path)
         evt1 = events.ReportingEvent(