[EventHubs] Stress test using azure monitor service for metrics/traces/loggings (Azure#18837)

Addressing issue: Azure#18792

Author: yunhaoling

sdk/eventhub/azure-eventhub/stress/app_insights_metric.py

@@ -0,0 +1,92 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+
+from opencensus.ext.azure import metrics_exporter
+from opencensus.stats import aggregation as aggregation_module
+from opencensus.stats import measure as measure_module
+from opencensus.stats import stats as stats_module
+from opencensus.stats import view as view_module
+
+from logger import get_azure_logger
+
+
+class AzureMonitorMetric:
+    def __init__(self, test_name, test_description=None):
+        # oc will automatically search for the ENV VAR 'APPLICATIONINSIGHTS_CONNECTION_STRING'
+        self.exporter = metrics_exporter.new_metrics_exporter()
+        self.stats = stats_module.stats
+        self.view_manager = self.stats.view_manager
+        self.stats_recorder = self.stats.stats_recorder
+        self.azure_logger = get_azure_logger(test_name)
+        self.name = test_name
+        self.desc = test_description
+
+        events_measure_name = "The number of events handled by " + self.name
+        events_measure_desc = "The number of events handled by " + self.desc if self.desc else None
+        memory_measure_name = "memory usage percentage for " + self.name
+        memory_measure_desc = "memory usage percentage for " + self.desc if self.desc else None
+        cpu_measure_name = "cpu usage percentage for " + self.name
+        cpu_measure_desc = "cpu usage percentage for " + self.desc if self.desc else None
+        error_measure_name = "error count for " + self.name
+        error_measure_desc = "The number of errors happened while running the test for " + self.desc if self.desc else None
+
+        self.events_measure = measure_module.MeasureInt(events_measure_name, events_measure_desc, "events")
+        self.memory_measure = measure_module.MeasureFloat(memory_measure_name, memory_measure_desc)
+        self.cpu_measure = measure_module.MeasureFloat(cpu_measure_name, cpu_measure_desc)
+        self.error_measure = measure_module.MeasureInt(error_measure_name, error_measure_desc)
+
+        self.events_measure_view = view_module.View(
+            events_measure_name,
+            events_measure_desc,
+            [],
+            self.events_measure,
+            aggregation_module.SumAggregation()
+        )
+
+        self.memory_measure_view = view_module.View(
+            memory_measure_name,
+            memory_measure_desc,
+            [],
+            self.memory_measure,
+            aggregation_module.LastValueAggregation()
+        )
+
+        self.cpu_measure_view = view_module.View(
+            cpu_measure_name,
+            cpu_measure_desc,
+            [],
+            self.cpu_measure,
+            aggregation_module.LastValueAggregation()
+        )
+
+        self.error_measure_view = view_module.View(
+            error_measure_name,
+            error_measure_desc,
+            [],
+            self.error_measure,
+            aggregation_module.CountAggregation()
+        )
+
+        self.view_manager.register_view(self.events_measure_view)
+        self.view_manager.register_view(self.memory_measure_view)
+        self.view_manager.register_view(self.cpu_measure_view)
+        self.view_manager.register_view(self.error_measure_view)
+
+        self.mmap = self.stats_recorder.new_measurement_map()
+
+    def record_events_cpu_memory(self, number_of_events, cpu_usage, memory_usage):
+        self.mmap.measure_int_put(self.events_measure, number_of_events)
+        self.mmap.measure_float_put(self.memory_measure, memory_usage)
+        self.mmap.measure_float_put(self.cpu_measure, cpu_usage)
+        self.mmap.record()
+
+    def record_error(self, error, extra=None):
+        self.mmap.measure_int_put(self.error_measure, 1)
+        self.mmap.record()
+        self.azure_logger.exception(
+            "Error happened when running {}: {}. Extra info: {}".format(self.name, repr(error), extra)
+        )

sdk/eventhub/azure-eventhub/stress/azure_eventhub_consumer_stress_async.py

@@ -10,6 +10,7 @@
 import os
 import logging
 from collections import defaultdict
+from functools import partial
 
 from azure.identity.aio import ClientSecretCredential
 from azure.eventhub.aio import EventHubConsumerClient
@@ -19,6 +20,7 @@
 
 from logger import get_logger
 from process_monitor import ProcessMonitor
+from app_insights_metric import AzureMonitorMetric
 
 
 def parse_starting_position(args):
@@ -65,7 +67,7 @@ def parse_starting_position(args):
     type=int,
     default=0
 )
-parser.add_argument("--parallel_recv_cnt", help="Number of receive clients doing parallel receiving", type=int)
+parser.add_argument("--parallel_recv_cnt", help="Number of receive clients doing parallel receiving", type=int, default=1)
 parser.add_argument("--proxy_hostname", type=str)
 parser.add_argument("--proxy_port", type=str)
 parser.add_argument("--proxy_username", type=str)
@@ -89,6 +91,8 @@ def parse_starting_position(args):
 recv_cnt_iteration_map = defaultdict(int)
 recv_time_map = dict()
 
+azure_metric_monitor = AzureMonitorMetric("Async EventHubConsumerClient")
+
 
 class EventHubConsumerClientTest(EventHubConsumerClient):
     async def get_partition_ids(self):
@@ -98,7 +102,7 @@ async def get_partition_ids(self):
             return await super(EventHubConsumerClientTest, self).get_partition_ids()
 
 
-async def on_event_received(partition_context, event):
+async def on_event_received(process_monitor, partition_context, event):
     recv_cnt_map[partition_context.partition_id] += 1 if event else 0
     if recv_cnt_map[partition_context.partition_id] % LOG_PER_COUNT == 0:
         total_time_elapsed = time.perf_counter() - start_time
@@ -113,10 +117,15 @@ async def on_event_received(partition_context, event):
                     recv_cnt_map[partition_context.partition_id] / total_time_elapsed,
                     LOG_PER_COUNT / (partition_current_time - partition_previous_time) if partition_previous_time else None
                     )
+        azure_metric_monitor.record_events_cpu_memory(
+            LOG_PER_COUNT,
+            process_monitor.cpu_usage_percent,
+            process_monitor.memory_usage_percent
+        )
         await partition_context.update_checkpoint(event)
 
 
-async def on_event_batch_received(partition_context, event_batch):
+async def on_event_batch_received(process_monitor, partition_context, event_batch):
     recv_cnt_map[partition_context.partition_id] += len(event_batch)
     recv_cnt_iteration_map[partition_context.partition_id] += len(event_batch)
     if recv_cnt_iteration_map[partition_context.partition_id] > LOG_PER_COUNT:
@@ -133,9 +142,18 @@ async def on_event_batch_received(partition_context, event_batch):
                     recv_cnt_iteration_map[partition_context.partition_id] / (partition_current_time - partition_previous_time) if partition_previous_time else None
                     )
         recv_cnt_iteration_map[partition_context.partition_id] = 0
+        azure_metric_monitor.record_events_cpu_memory(
+            LOG_PER_COUNT,
+            process_monitor.cpu_usage_percent,
+            process_monitor.memory_usage_percent
+        )
         await partition_context.update_checkpoint()
 
 
+async def on_error(partition_context, exception):
+    azure_metric_monitor.record_error(exception, extra="partition: {}".format(partition_context.partition_id))
+
+
 def create_client(args):
 
     if args.storage_conn_str:
@@ -198,26 +216,31 @@ def create_client(args):
 
 async def run(args):
 
-    with ProcessMonitor("monitor_{}".format(args.log_filename), "consumer_stress_async", print_console=args.print_console):
+    with ProcessMonitor("monitor_{}".format(args.log_filename), "consumer_stress_async", print_console=args.print_console) as process_monitor:
         kwargs_dict = {
             "prefetch": args.link_credit,
             "partition_id": str(args.recv_partition_id) if args.recv_partition_id else None,
             "track_last_enqueued_event_properties": args.track_last_enqueued_event_properties,
-            "starting_position": starting_position
+            "starting_position": starting_position,
+            "on_error": on_error
         }
         if args.max_batch_size:
             kwargs_dict["max_batch_size"] = args.max_batch_size
         if args.max_wait_time:
             kwargs_dict["max_wait_time"] = args.max_wait_time
+
+        on_event_received_with_process_monitor = partial(on_event_received, process_monitor)
+        on_event_batch_received_with_process_monitor = partial(on_event_batch_received, process_monitor)
+
         if args.parallel_recv_cnt and args.parallel_recv_cnt > 1:
             clients = [create_client(args) for _ in range(args.parallel_recv_cnt)]
             tasks = [
                 asyncio.ensure_future(
                     clients[i].receive_batch(
-                        on_event_batch_received,
+                        on_event_batch_received_with_process_monitor,
                         **kwargs_dict
                     ) if args.max_batch_size else clients[i].receive(
-                        on_event_received,
+                        on_event_received_with_process_monitor,
                         **kwargs_dict
                     )
                 ) for i in range(args.parallel_recv_cnt)
@@ -226,10 +249,10 @@ async def run(args):
             clients = [create_client(args)]
             tasks = [asyncio.ensure_future(
                 clients[0].receive_batch(
-                    on_event_batch_received,
+                    on_event_batch_received_with_process_monitor,
                     **kwargs_dict
                 ) if args.max_batch_size else clients[0].receive(
-                    on_event_received,
+                    on_event_received_with_process_monitor,
                     **kwargs_dict
                 )
             )]

sdk/eventhub/azure-eventhub/stress/azure_eventhub_consumer_stress_sync.py

@@ -10,13 +10,15 @@
 import os
 import logging
 from collections import defaultdict
+from functools import partial
 
 from azure.identity import ClientSecretCredential
 from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
 from azure.eventhub import EventHubConsumerClient, TransportType, EventHubSharedKeyCredential
 
 from logger import get_logger
 from process_monitor import ProcessMonitor
+from app_insights_metric import AzureMonitorMetric
 
 
 def parse_starting_position(args):
@@ -88,6 +90,8 @@ def parse_starting_position(args):
 recv_cnt_iteration_map = defaultdict(int)
 recv_time_map = dict()
 
+azure_metric_monitor = AzureMonitorMetric("Sync EventHubConsumerClient")
+
 
 class EventHubConsumerClientTest(EventHubConsumerClient):
     def get_partition_ids(self):
@@ -97,7 +101,7 @@ def get_partition_ids(self):
             return super(EventHubConsumerClientTest, self).get_partition_ids()
 
 
-def on_event_received(partition_context, event):
+def on_event_received(process_monitor, partition_context, event):
     recv_cnt_map[partition_context.partition_id] += 1 if event else 0
     if recv_cnt_map[partition_context.partition_id] % LOG_PER_COUNT == 0:
         total_time_elapsed = time.perf_counter() - start_time
@@ -112,10 +116,15 @@ def on_event_received(partition_context, event):
                     recv_cnt_map[partition_context.partition_id] / total_time_elapsed,
                     LOG_PER_COUNT / (partition_current_time - partition_previous_time) if partition_previous_time else None
                     )
+        azure_metric_monitor.record_events_cpu_memory(
+            LOG_PER_COUNT,
+            process_monitor.cpu_usage_percent,
+            process_monitor.memory_usage_percent
+        )
         partition_context.update_checkpoint(event)
 
 
-def on_event_batch_received(partition_context, event_batch):
+def on_event_batch_received(process_monitor, partition_context, event_batch):
     recv_cnt_map[partition_context.partition_id] += len(event_batch)
     recv_cnt_iteration_map[partition_context.partition_id] += len(event_batch)
     if recv_cnt_iteration_map[partition_context.partition_id] > LOG_PER_COUNT:
@@ -131,9 +140,18 @@ def on_event_batch_received(partition_context, event_batch):
                     recv_cnt_iteration_map[partition_context.partition_id] / (partition_current_time - partition_previous_time) if partition_previous_time else None
                     )
         recv_cnt_iteration_map[partition_context.partition_id] = 0
+        azure_metric_monitor.record_events_cpu_memory(
+            LOG_PER_COUNT,
+            process_monitor.cpu_usage_percent,
+            process_monitor.memory_usage_percent
+        )
         partition_context.update_checkpoint()
 
 
+def on_error(partition_context, exception):
+    azure_metric_monitor.record_error(exception, extra="partition: {}".format(partition_context.partition_id))
+
+
 def create_client(args):
     if args.storage_conn_str:
         checkpoint_store = BlobCheckpointStore.from_connection_string(args.storage_conn_str, args.storage_container_name)
@@ -194,23 +212,29 @@ def create_client(args):
 
 
 def run(args):
-    with ProcessMonitor("monitor_{}".format(args.log_filename), "consumer_stress_sync", print_console=args.print_console):
+
+    with ProcessMonitor("monitor_{}".format(args.log_filename), "consumer_stress_sync", print_console=args.print_console) as process_monitor:
         kwargs_dict = {
             "prefetch": args.link_credit,
             "partition_id": str(args.recv_partition_id) if args.recv_partition_id else None,
             "track_last_enqueued_event_properties": args.track_last_enqueued_event_properties,
-            "starting_position": starting_position
+            "starting_position": starting_position,
+            "on_error": on_error
         }
         if args.max_batch_size:
             kwargs_dict["max_batch_size"] = args.max_batch_size
         if args.max_wait_time:
             kwargs_dict["max_wait_time"] = args.max_wait_time
+
+        on_event_received_with_process_monitor = partial(on_event_received, process_monitor)
+        on_event_batch_received_with_process_monitor = partial(on_event_batch_received, process_monitor)
+
         if args.parallel_recv_cnt and args.parallel_recv_cnt > 1:
             clients = [create_client(args) for _ in range(args.parallel_recv_cnt)]
             threads = [
                 threading.Thread(
                     target=clients[i].receive_batch if args.max_batch_size else clients[i].receive,
-                    args=(on_event_batch_received if args.max_batch_size else on_event_received,),
+                    args=(on_event_batch_received_with_process_monitor if args.max_batch_size else on_event_received_with_process_monitor,),
                     kwargs=kwargs_dict,
                     daemon=True
                 ) for i in range(args.parallel_recv_cnt)
@@ -219,7 +243,7 @@ def run(args):
             clients = [create_client(args)]
             threads = [threading.Thread(
                 target=clients[0].receive_batch if args.max_batch_size else clients[0].receive,
-                args=(on_event_batch_received if args.max_batch_size else on_event_received,),
+                args=(on_event_batch_received_with_process_monitor if args.max_batch_size else on_event_received_with_process_monitor,),
                 kwargs=kwargs_dict,
                 daemon=True
             )]