Merge pull request #604 from ddps-lab/azure-collector

Add Azure T2/T3 Logic

Author: chris0765

collector/spot-dataset/azure/lambda/current_collector/lambda_function_sps.py

@@ -6,7 +6,7 @@
 from sps_module import sps_shared_resources
 from utils.merge_df import merge_if_saving_price_sps_df
 from utils.upload_data import update_latest, save_raw, upload_timestream, query_selector, upload_cloudwatch
-from utils.compare_data import compare_sps
+from utils.compare_data import compare_sps, compare_max_instance
 from utils.pub_service import send_slack_message, Logger, S3, AZURE_CONST
 from utils.azure_auth import get_sps_token_and_subscriptions
 
@@ -164,13 +164,17 @@ def handle_res_df_for_spotlake(price_saving_if_df, sps_df, time_datetime, desire
             f"{AZURE_CONST.S3_LATEST_ALL_DATA_AVAILABILITY_ZONE_TRUE_PKL_GZIP_SAVE_PATH}", 'pkl.gz')
 
         workload_cols = ['InstanceTier', 'InstanceType', 'Region', 'AvailabilityZone', 'DesiredCount']
-        feature_cols = ['OndemandPrice', 'SpotPrice', 'IF', 'Score', 'SPS_Update_Time']
+        feature_cols = ['OndemandPrice', 'SpotPrice', 'IF', 'Score', 'SPS_Update_Time', 'T2', 'T3']
 
         query_success = timestream_success = cloudwatch_success = \
             update_latest_success = save_raw_success = False
 
         if prev_availability_zone_true_all_data_df is not None and not prev_availability_zone_true_all_data_df.empty:
-            prev_availability_zone_true_all_data_df.drop(columns=['id'], inplace=True)
+            prev_availability_zone_true_all_data_df.drop(columns=['id'], inplace=True, errors='ignore')
+            
+            # Apply T2/T3 Aggregation Logic
+            sps_merged_df = compare_max_instance(prev_availability_zone_true_all_data_df, sps_merged_df, current_desired_count)
+            
             changed_df = compare_sps(prev_availability_zone_true_all_data_df, sps_merged_df, workload_cols, feature_cols)
 
             query_success = query_selector(changed_df)

collector/spot-dataset/azure/lambda/current_collector/load_sps.py

@@ -213,7 +213,9 @@ def execute_spot_placement_score_task_by_parameter_pool_df(api_calls_df, desired
                                     score.get("sku", ""), {}).get("InstanceTier"),
                                 "InstanceType": SS_Resources.region_map_and_instance_map_tmp['instance_map'].get(
                                     score.get("sku", ""), {}).get("InstanceTypeOld"),
-                                "Score": score.get("score")
+                                "Score": score.get("score"),
+                                "T3": desired_count if score.get("score") == 3 else 0,
+                                "T2": desired_count if score.get("score") == 2 else 0
                             }
                             if availability_zones is True:
                                 score_data["AvailabilityZone"] = score.get("availabilityZone", "Single")

collector/spot-dataset/azure/lambda/current_collector/utils/compare_data.py

@@ -50,4 +50,68 @@ def compare_sps(previous_df, current_df, workload_cols, feature_cols):
 
     changed_df = changed_df[current_df.columns]
 
-    return changed_df if not changed_df.empty else None
+    return changed_df if not changed_df.empty else None
+
+def compare_max_instance(previous_df, new_df, target_capacity):
+    fallback_dict = {50:45, 45:40, 40:35, 35:30, 30:25, 25:20, 20:15, 15:10, 10:5, 5:1, 1:0}
+    fallback_val = fallback_dict.get(target_capacity, 0)
+    
+    merged_df = pd.merge(
+        new_df,
+        previous_df[["InstanceType", "Region", "AvailabilityZone", "DesiredCount", "Score", "T3", "T2"]],
+        on=["InstanceType", "Region", "AvailabilityZone", "DesiredCount"],
+        how="left",
+        suffixes=("", "_prev")
+    )
+
+    # Fix SPS when single node SPS
+    if target_capacity == 1:
+        merged_df["Score"] = merged_df["Score"].combine_first(merged_df["Score_prev"])
+    
+    # Merge single node SPS with multi node SPS if (multi node SPS) > (single node SPS)
+    # Note: Score strings "3", "2", "1" are comparable.
+    # But need to handle N/A or types. Assuming Score is int or convertible.
+    # Azure Score is int from load_sps.
+    # previous_df Score might be string if read from file? feature_cols convert to str in compare_sps but here we read raw df.
+    
+    # Ensure Score types are compatible (float/int)
+    merged_df["Score"] = pd.to_numeric(merged_df["Score"], errors='coerce')
+    merged_df["Score_prev"] = pd.to_numeric(merged_df["Score_prev"], errors='coerce')
+    merged_df["T3"] = pd.to_numeric(merged_df["T3"], errors='coerce').fillna(0)
+    merged_df["T3_prev"] = pd.to_numeric(merged_df["T3_prev"], errors='coerce').fillna(0)
+    merged_df["T2"] = pd.to_numeric(merged_df["T2"], errors='coerce').fillna(0)
+    merged_df["T2_prev"] = pd.to_numeric(merged_df["T2_prev"], errors='coerce').fillna(0)
+
+    merged_df.loc[(merged_df["Score"] > merged_df["Score_prev"]), "Score_prev"] = merged_df["Score"]
+
+    # Calculate T3
+    # Use numpy where.
+    merged_df["T3"] = np.where(
+        merged_df["Score"] >= 3,
+        np.maximum(merged_df["T3"], merged_df["T3_prev"]),
+        np.minimum(fallback_val, merged_df["T3_prev"])
+    )
+
+    # Calculate T2
+    merged_df["T2"] = np.where(
+        merged_df["Score"] >= 2,
+        np.maximum(merged_df["T2"], merged_df["T2_prev"]),
+        np.minimum(fallback_val, merged_df["T2_prev"])
+    )
+    
+    if target_capacity == 1:
+        merged_df.loc[merged_df["Score"] <= 2, "T3"] = 0
+        merged_df.loc[merged_df["Score"] < 2, "T2"] = 0
+    else:
+        merged_df.loc[merged_df["Score_prev"] <= 2, "T3"] = 0
+        merged_df.loc[merged_df["Score_prev"] < 2, "T2"] = 0
+        # Fix SPS to Single node SPS ? mimic AWS "merged_df["SPS"] = merged_df["SPS_prev"]"
+        # AWS comment: "Fix SPS to Single node SPS" - this logic seems specific to assuming single node fallback.
+        # But let's copy logic:
+        merged_df["Score"] = merged_df["Score_prev"]
+
+    # Convert to standard types if needed
+    # Drop unnecessary columns
+    merged_df.drop(columns=["T3_prev", "T2_prev", "Score_prev"], errors='ignore', inplace=True)
+
+    return merged_df

collector/spot-dataset/azure/lambda/current_collector/utils/upload_data.py

@@ -137,7 +137,7 @@ def upload_timestream(data, time_datetime):
     try:
         data = data.copy()
         data = data[["InstanceTier", "InstanceType", "Region", "OndemandPrice", "SpotPrice", "Savings", "IF",
-            "DesiredCount", "AvailabilityZone", "Score", "SPS_Update_Time"]]
+            "DesiredCount", "AvailabilityZone", "Score", "SPS_Update_Time", "T2", "T3"]]
 
         fill_values = {
             "InstanceTier": 'N/A',
@@ -150,7 +150,9 @@ def upload_timestream(data, time_datetime):
             'DesiredCount': -1,
             'AvailabilityZone': 'N/A',
             'Score': 'N/A',
-            'SPS_Update_Time': 'N/A'
+            'SPS_Update_Time': 'N/A',
+            'T2': 0,
+            'T3': 0
         }
         data = data.fillna(fill_values)
 
@@ -177,7 +179,9 @@ def upload_timestream(data, time_datetime):
                 ('SpotPrice', 'DOUBLE'),
                 ('IF', 'DOUBLE'),
                 ('Score', 'VARCHAR'),
-                ('SPS_Update_Time', 'VARCHAR')
+                ('SPS_Update_Time', 'VARCHAR'),
+                ('T2', 'DOUBLE'),
+                ('T3', 'DOUBLE')
             ]
 
             for column, value_type in measure_columns: