Merge pull request #592 from ddps-lab/azure-collector-fix20250916

Azure SPS 수집 모델, subscription-location 순회 로직 개선(2차)

Author: krtaiyang

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

@@ -58,7 +58,7 @@ def get_price(skip_num):
         if response.status_code == 200:
             break
         else:
-            time.sleep(1)
+            time.sleep(2)
             response = requests.get(get_link)
 
     if response.status_code != 200:

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

@@ -114,6 +114,7 @@ def collect_spot_placement_score_first_time(desired_counts):
         print(f"Prepare the request pool. time: {minutes}min {seconds}sec")
 
         return sps_res_availability_zones_true_df
+    return None
 
 
 @log_execution_time
@@ -459,6 +460,7 @@ def save_tmp_files_to_s3():
         f"{base_path}/{az_str}/{AZURE_CONST.S3_INVALID_REGIONS_JSON_FILENAME}": SS_Resources.invalid_regions_tmp,
         f"{base_path}/{az_str}/{AZURE_CONST.S3_INVALID_INSTANCE_TYPES_JSON_FILENAME}": SS_Resources.invalid_instance_types_tmp,
         f"{base_path}/{az_str}/{AZURE_CONST.S3_LOCATIONS_CALL_HISTORY_JSON_FILENAME}": SS_Resources.locations_call_history_tmp,
+        f"{base_path}/{az_str}/{AZURE_CONST.S3_LAST_SUBSCRIPTION_ID_AND_LOCATION_JSON_FILENAME}": SS_Resources.last_subscription_id_and_location_tmp,
         f"{base_path}/{az_str}/{AZURE_CONST.S3_LOCATIONS_OVER_LIMIT_JSON_FILENAME}": SS_Resources.locations_over_limit_tmp
     }
 
@@ -475,13 +477,14 @@ def get_variable_from_s3():
         instance_types_data = S3.read_file(f"{base_path}/{az_str}/{AZURE_CONST.S3_INVALID_INSTANCE_TYPES_JSON_FILENAME}", 'json')
         call_history_data = S3.read_file(f"{base_path}/{az_str}/{AZURE_CONST.S3_LOCATIONS_CALL_HISTORY_JSON_FILENAME}", 'json')
         over_limit_data = S3.read_file(f"{base_path}/{az_str}/{AZURE_CONST.S3_LOCATIONS_OVER_LIMIT_JSON_FILENAME}", 'json')
-        last_location_index_data = S3.read_file(f"{base_path}/{az_str}/{AZURE_CONST.S3_LAST_SUBSCRIPTION_ID_AND_LOCATION_JSON_FILENAME}", 'json')
+        last_subscription_id_and_location = S3.read_file(f"{base_path}/{az_str}/{AZURE_CONST.S3_LAST_SUBSCRIPTION_ID_AND_LOCATION_JSON_FILENAME}", 'json')
         region_map_and_instance_map = S3.read_file(f"{base_path}/{az_str}/{AZURE_CONST.S3_REGION_MAP_AND_INSTANCE_MAP_JSON_FILENAME}", 'json')
 
         SS_Resources.invalid_regions_tmp = invalid_regions_data
         SS_Resources.invalid_instance_types_tmp = instance_types_data
         SS_Resources.locations_call_history_tmp = call_history_data
         SS_Resources.locations_over_limit_tmp = over_limit_data
+        SS_Resources.last_subscription_id_and_location_tmp = last_subscription_id_and_location
         SS_Resources.region_map_and_instance_map_tmp = {
             "region_map": region_map_and_instance_map.get('region_map'),
             "instance_map": region_map_and_instance_map.get('instance_map')
@@ -492,6 +495,7 @@ def get_variable_from_s3():
             SS_Resources.invalid_instance_types_tmp,
             SS_Resources.locations_call_history_tmp,
             SS_Resources.locations_over_limit_tmp,
+            SS_Resources.last_subscription_id_and_location_tmp,
             SS_Resources.region_map_and_instance_map_tmp
         ]):
             print("[S3]: Successfully prepared variable from s3.")

collector/spot-dataset/azure/lambda/current_collector/sps_module/sps_location_manager.py

@@ -1,7 +1,6 @@
 import re
 import requests
 import traceback
-import random
 from sps_module import sps_shared_resources
 from datetime import datetime, timedelta, timezone
 from utils.pub_service import S3, AZURE_CONST
@@ -47,6 +46,7 @@ def check_and_add_available_locations(az):
             return True
         else:
             print("No new available locations found. locations_call_history_tmp or locations_call_history_tmp unchanged.")
+            return True
 
     except Exception as e:
         print(f"Error in check_and_add_available_locations: {e}")
@@ -69,46 +69,63 @@ def validation_can_call(subscription_id, location):
 
 
 def get_next_available_location():
-    """
-     이 메서드는 사용 가능한 다음 위치를 리턴합니다.
-     호출 이력과 초과 요청 데이터를 기반으로 적절한 위치를 선택합니다.
-     호출 시 이용하는 location은 구독 내에 지난 호출의 location을 이용 안 해야 하는 로직이 들어갑니다.
-
-     이유:
-     단 기간에 같은 location으로 호출 시, timeout율이 놉습니다.
-     """
     try:
         if SS_Resources.locations_call_history_tmp is None or SS_Resources.locations_over_limit_tmp is None:
             return None
 
-        # Clean expired data
         clean_expired_over_limit_locations()
         clean_expired_over_call_history_locations()
 
-        start_subscription_index = random.randint(0, len(SS_Resources.subscriptions) - 1)
+        subs = SS_Resources.subscriptions
+        locs = SS_Resources.available_locations
+        if not subs or not locs:
+            return None
 
-        for i in range(len(SS_Resources.subscriptions)):
-            subscription_index = (start_subscription_index + i) % len(SS_Resources.subscriptions)
-            subscription_id = SS_Resources.subscriptions[subscription_index]
+        n, m = len(subs), len(locs)
 
-            start_location_index = random.randint(0, len(SS_Resources.available_locations) - 1)
-            for j in range(len(SS_Resources.available_locations)):
-                location_index = (start_location_index + j) % len(SS_Resources.available_locations)
-                location = SS_Resources.available_locations[location_index]
+        last_pair = getattr(SS_Resources, "last_subscription_id_and_location_tmp", None) or {}
+        last_sub_id = last_pair.get("last_subscription_id")
+        last_loc = last_pair.get("last_location")
 
-                if validation_can_call(subscription_id, location):
-                    SS_Resources.succeed_to_get_next_available_location_count += 1
-                    SS_Resources.succeed_to_get_next_available_location_count_all += 1
-                    SS_Resources.locations_call_history_tmp[subscription_id][location].append(datetime.now(timezone.utc).replace(tzinfo=None).isoformat())
-                    return subscription_id, location
+        if last_sub_id in subs and last_loc in locs:
+            s_idx = subs.index(last_sub_id)
+            l_idx = locs.index(last_loc)
+            l_idx = (l_idx + 1) % m
+            if l_idx == 0:
+                s_idx = (s_idx + 1) % n
+        else:
+            s_idx, l_idx = 0, 0
 
-        return None
+        attempts = 0
+        while attempts < n * m:
+            sub_id = subs[s_idx]
+            loc = locs[l_idx]
+
+            if validation_can_call(sub_id, loc):
+                SS_Resources.succeed_to_get_next_available_location_count += 1
+                SS_Resources.succeed_to_get_next_available_location_count_all += 1
+
+                SS_Resources.locations_call_history_tmp[sub_id][loc].append(
+                    datetime.now(timezone.utc).replace(tzinfo=None).isoformat())
+
+                SS_Resources.last_subscription_id_and_location = {
+                    "last_subscription_id": sub_id,
+                    "last_location": loc,
+                }
+                return sub_id, loc
+
+            l_idx = (l_idx + 1) % m
+            if l_idx == 0:
+                s_idx = (s_idx + 1) % n
+
+            attempts += 1
 
     except Exception as e:
         print("\n[ERROR] Exception occurred in get_next_available_location:")
         print(traceback.format_exc())
         print(f"\n[ERROR] Failed to get_next_available_location: {e}")
         return None
+    return None
 
 
 def collect_available_locations():
@@ -144,6 +161,7 @@ def collect_available_locations():
     except Exception as e:
         print(f"Failed to collect_available_locations, Error: {e}")
         return None
+    return None
 
 def clean_expired_over_limit_locations():
     '''

collector/spot-dataset/azure/lambda/current_collector/sps_module/sps_shared_resources.py

@@ -18,6 +18,7 @@
 invalid_instance_types_tmp = None
 locations_call_history_tmp = None
 locations_over_limit_tmp = None
+last_subscription_id_and_location_tmp = None
 region_map_and_instance_map_tmp = None
 subscriptions = None
 available_locations = None