add tuple 4 elements in performancemonitor

Author: marcomelloni

performance/performance_analysis.py

@@ -6,7 +6,7 @@
 import pandas as pd
 
 
-DEFAULT_INPUT  = Path("../performance_log/performance_metrics.csv")
+DEFAULT_INPUT  = Path("../performance_log/performance_metrics_test.csv")
 DEFAULT_OUTPUT = Path("overhead_summary.csv")
 AVG_INPUT_OVERHEAD  = "Avg Input Overhead"
 AVG_OUTPUT_OVERHEAD = "Avg Output Overhead"

simulation_bridge/src/core/bridge_core.py

@@ -142,11 +142,15 @@ def handle_input_message(self, sender, **kwargs):  # pylint: disable=unused-argu
         # Initialize performance monitor
         performance_monitor = PerformanceMonitor()
         message_dict = kwargs.get('message', {})
+        producer = kwargs.get('producer', 'unknown')
+        simulation_type = message_dict.get(
+            'simulation', {}).get('type', 'unknown')
         protocol = kwargs.get('protocol', 'unknown')
         operation_id = message_dict.get(
             'simulation', {}).get(
             'request_id', 'unknown')
-        performance_monitor.record_core_received_input(operation_id, protocol)
+        performance_monitor.record_core_received_input(
+            operation_id, protocol, producer, simulation_type)
         try:
             message = MessageModel.model_validate(message_dict)
         except Exception as e:  # pylint: disable=broad-exception-caught
@@ -177,6 +181,9 @@ def handle_result_rabbitmq_message(self, sender, **kwargs):  # pylint: disable=u
         # Initialize performance monitor
         performance_monitor = PerformanceMonitor()
         message = kwargs.get('message', {})
+        destinations = message.get('destinations', [])
+        destination = destinations[0] if destinations else 'unknown'
+        simulation_type = message.get('simulation', {}).get('type', 'unknown')
         producer = message.get('source', 'unknown')
         consumer = "result"
         operation_id = message.get('request_id', 'unknown')
@@ -188,9 +195,11 @@ def handle_result_rabbitmq_message(self, sender, **kwargs):  # pylint: disable=u
             protocol='rabbitmq',
             operation_id=operation_id)
         status = message.get('status', 'unknown')
-        performance_monitor.record_result_sent(operation_id, 'rabbitmq')
+        performance_monitor.record_result_sent(
+            operation_id, 'rabbitmq', destination, simulation_type)
         if status == 'completed':
-            performance_monitor.finalize_operation(operation_id, 'rabbitmq')
+            performance_monitor.finalize_operation(
+                operation_id, 'rabbitmq', destination, simulation_type)
 
     def handle_result_unknown_message(self, sender, **kwargs):  # pylint: disable=unused-argument
         """
@@ -222,7 +231,7 @@ def _publish_message(self, producer, consumer, message,  # pylint: disable=too-m
 
         # Initialize performance monitor
         performance_monitor = PerformanceMonitor()
-
+        simulation_type = message.get('simulation', {}).get('type', 'unknown')
         routing_key = f"{producer}.{consumer}"
         message['simulation']['bridge_meta'] = {
             'protocol': protocol
@@ -242,7 +251,7 @@ def _publish_message(self, producer, consumer, message,  # pylint: disable=too-m
             # Record sent input time in performance monitor
             if exchange == 'ex.bridge.output':
                 performance_monitor.record_core_sent_input(
-                    operation_id, protocol)
+                    operation_id, protocol, producer, simulation_type)
         except (pika.exceptions.AMQPConnectionError,
                 pika.exceptions.AMQPChannelError) as e:
             logger.error("RabbitMQ connection error: %s", e)

simulation_bridge/src/protocol_adapters/mqtt/mqtt_adapter.py

@@ -138,13 +138,12 @@ def on_message(self, client, userdata, msg):
             producer = simulation.get('client_id', 'unknown')
             consumer = simulation.get('simulator', 'unknown')
             operation_id = simulation.get('request_id', 'unknown')
-
+            simulation_type = simulation.get('type', 'unknown')
             # Process message directly - no need for queuing
             logger.debug(
                 "MQTT - Processing message %s, from producer: %s, simulator: %s",
                 message, producer, consumer)
 
-            simulation_type = simulation.get('type', 'unknown')
             performance_monitor.start_operation(
                 operation_id,
                 client_id=producer,
@@ -228,6 +227,11 @@ def send_result(self, message):
             # Initialize performance monitor
             performance_monitor = PerformanceMonitor()
             operation_id = message.get('request_id', 'unknown')
+            destinations = message.get('destinations', [])
+            producer = destinations[0] if destinations else 'unknown'
+            simulation_type = message.get(
+                'simulation', {}).get(
+                'type', 'unknown')
             output_topic = self.mqtt_config['output_topic']
             self.mqtt_client.publish(
                 topic=output_topic,
@@ -237,9 +241,11 @@ def send_result(self, message):
             logger.debug(
                 "Message published to MQTT topic '%s': %s", output_topic, message)
             status = message.get('status', 'unknown')
-            performance_monitor.record_result_sent(operation_id, 'mqtt')
+            performance_monitor.record_result_sent(
+                operation_id, 'mqtt', producer, simulation_type)
             if status == 'completed':
-                performance_monitor.finalize_operation(operation_id, 'mqtt')
+                performance_monitor.finalize_operation(
+                    operation_id, 'mqtt', producer, simulation_type)
         except (ConnectionError, TimeoutError) as e:
             logger.error("Error publishing MQTT message: %s", e)
 

simulation_bridge/src/protocol_adapters/rabbitmq/rabbitmq_adapter.py

@@ -154,6 +154,11 @@ def _process_message(self, ch, method, properties, body, queue_name):
                 )
             elif queue_name == 'Q.bridge.result':
                 operation_id = message.get('request_id', 'unknown')
+                destinations = message.get('destinations', [])
+                producer = destinations[0] if destinations else 'unknown'
+                simulation_type = message.get(
+                    'simulation', {}).get(
+                    'type', 'unknown')
                 bridge_meta = message.get('bridge_meta', {})
                 if isinstance(bridge_meta, str):
                     if bridge_meta.strip().startswith('{'):
@@ -169,9 +174,7 @@ def _process_message(self, ch, method, properties, body, queue_name):
                         bridge_meta = {}
                 protocol = bridge_meta.get('protocol', 'unknown')
                 performance_monitor.record_core_received_result(
-                    operation_id, protocol)
-                destinations = message.get('destinations', [])
-                producer = destinations[0] if destinations else 'unknown'
+                    operation_id, protocol, producer, simulation_type)
                 consumer = message.get('source', 'unknown')
                 kwargs = {
                     "message": message,

simulation_bridge/src/protocol_adapters/rest/rest_adapter.py

@@ -225,12 +225,17 @@ def publish_result_message_rest(self, sender, **kwargs):
             performance_monitor = PerformanceMonitor()
             message = kwargs.get('message', {})
             operation_id = message.get('request_id', 'unknown')
+            simulation_type = message.get(
+                'simulation', {}).get(
+                'type', 'unknown')
             destination = message.get('destinations', [])[0]
             self.send_result_sync(destination, message)
             status = message.get('status', 'unknown')
-            performance_monitor.record_result_sent(operation_id, 'rest')
+            performance_monitor.record_result_sent(
+                operation_id, 'rest', destination, simulation_type)
             if status == 'completed':
-                performance_monitor.finalize_operation(operation_id, 'rest')
+                performance_monitor.finalize_operation(
+                    operation_id, 'rest', destination, simulation_type)
             logger.debug(
                 "Successfully scheduled result message for REST client: %s",
                 destination)