[Test] Push performance data to DynamoDB table

1. Create a separate table called ParallelCluster-PerformanceTest-Metadata if not exist
2. Push OpenFoam, OSU, StarCCM performance result to the table

Author: hanwen-cluster

tests/integration-tests/framework/framework_constants.py

@@ -7,4 +7,5 @@
 #  limitations under the License.
 
 METADATA_TABLE = "ParallelCluster-IntegTest-Metadata"
+PERFORMANCE_METADATA_TABLE = "ParallelCluster-PerformanceTest-Metadata"
 METADATA_DEFAULT_REGION = "us-east-1"

tests/integration-tests/tests/performance_tests/common.py

@@ -3,16 +3,20 @@
 import os as os_lib
 import pathlib
 import shutil
+import time
+import uuid
 from math import ceil
 from os import makedirs
 
 import boto3
 from assertpy import assert_that
+from framework.framework_constants import METADATA_DEFAULT_REGION, PERFORMANCE_METADATA_TABLE
+from framework.metadata_table_manager import MetadataTableManager
 from retrying import retry
 from time_utils import seconds
 from utils import get_username_for_os
 
-from tests.common.utils import fetch_instance_slots
+from tests.common.utils import fetch_instance_slots, get_installed_parallelcluster_version
 
 # Common settings used by all test scenarios
 # You can change these variables to adapt the performance test to your needs.
@@ -249,3 +253,33 @@ def _log_output_performance_difference(node, performance_degradation, observed_v
         f"Nodes: {node}, Baseline: {baseline_value} seconds, Observed: {observed_value} seconds, "
         f"Percentage difference: {percentage_difference}%, Outcome: {outcome}"
     )
+
+
+def push_result_to_dynamodb(name, result, instance, os):
+    reporting_region = METADATA_DEFAULT_REGION
+    logging.info(f"Metadata reporting region {reporting_region}")
+    # Create the metadata table in case it doesn't exist
+    MetadataTableManager(reporting_region, PERFORMANCE_METADATA_TABLE).create_metadata_table()
+    try:
+        # Create DynamoDB resource
+        dynamodb = boto3.resource("dynamodb", region_name=reporting_region)
+        table = dynamodb.Table(PERFORMANCE_METADATA_TABLE)
+
+        # Prepare item to be inserted
+        item = {
+            "id": str(uuid.uuid4().hex),
+            "name": name,
+            "instance": instance,
+            "os": os,
+            "timestamp": int(time.time()),
+            "result": str(result),
+            "pcluster_version": f"v{get_installed_parallelcluster_version()}",
+        }
+
+        # Put item in the table
+        table.put_item(Item=item)
+        logging.info(f"Successfully pushed result to DynamoDB with id: {item['id']}")
+
+    except Exception as e:
+        logging.error(f"Failed to push result to DynamoDB: {str(e)}")
+        raise

tests/integration-tests/tests/performance_tests/test_openfoam.py

@@ -4,7 +4,7 @@
 import pytest
 from remote_command_executor import RemoteCommandExecutionError, RemoteCommandExecutor
 
-from tests.performance_tests.common import _log_output_performance_difference
+from tests.performance_tests.common import _log_output_performance_difference, push_result_to_dynamodb
 
 # timeout in seconds
 OPENFOAM_INSTALLATION_TIMEOUT = 300
@@ -91,8 +91,10 @@ def test_openfoam(
         observed_value_8 = future_8.result()
         observed_value_16 = future_16.result()
 
+    result = list(zip(number_of_nodes, [observed_value_8, observed_value_16, observed_value_32]))
+    push_result_to_dynamodb("OpenFOAM", result, instance, os)
     # Check results and log performance degradation
-    for node, observed_value in zip(number_of_nodes, [observed_value_8, observed_value_16, observed_value_32]):
+    for node, observed_value in result:
         baseline_value = BASELINE_CLUSTER_SIZE_ELAPSED_SECONDS[os][node]
         _log_output_performance_difference(node, performance_degradation, observed_value, baseline_value)
 

tests/integration-tests/tests/performance_tests/test_osu.py

@@ -25,6 +25,7 @@
     run_system_analyzer,
     write_file,
 )
+from tests.performance_tests.common import push_result_to_dynamodb
 
 # We collected OSU benchmarks results for c5n.18xlarge only.
 OSU_BENCHMARKS_INSTANCES = ["c5n.18xlarge"]
@@ -237,7 +238,9 @@ def _check_osu_benchmarks_results(test_datadir, output_dir, os, instance, mpi_ve
     metric_data = []
     metric_namespace = "ParallelCluster/test_efa"
     evaluation_output = ""
-    for packet_size, value in re.findall(r"(\d+)\s+(\d+)\.", output):
+    result = re.findall(r"(\d+)\s+(\d+)\.", output)
+    push_result_to_dynamodb(f"OSU_{benchmark_name}", result, instance, os)
+    for packet_size, value in result:
         with open(
             str(test_datadir / "osu_benchmarks" / "results" / os / instance / mpi_version / benchmark_name),
             encoding="utf-8",

tests/integration-tests/tests/performance_tests/test_starccm.py

@@ -6,7 +6,7 @@
 from remote_command_executor import RemoteCommandExecutionError, RemoteCommandExecutor
 
 from tests.common.utils import assert_no_file_handler_leak, get_compute_ip_to_num_files
-from tests.performance_tests.common import _log_output_performance_difference
+from tests.performance_tests.common import _log_output_performance_difference, push_result_to_dynamodb
 
 # timeout in seconds
 STARCCM_INSTALLATION_TIMEOUT = 1800
@@ -125,7 +125,9 @@ def test_starccm(
     )
 
     # Check results and log performance degradation
-    for node, observed_value in zip(number_of_nodes, [observed_value_8, observed_value_16, observed_value_32]):
+    result = list(zip(number_of_nodes, [observed_value_8, observed_value_16, observed_value_32]))
+    push_result_to_dynamodb("StarCCM", result, instance, os)
+    for node, observed_value in result:
         baseline_value = BASELINE_CLUSTER_SIZE_ELAPSED_SECONDS[os][node]
         _log_output_performance_difference(node, performance_degradation, observed_value, baseline_value)