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

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

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

@@ -112,9 +112,8 @@ def collect_spot_placement_score_first_time(desired_counts):
         elapsed = end_time - start_time
         minutes, seconds = divmod(int(elapsed), 60)
         print(f"Prepare the request pool. time: {minutes}min {seconds}sec")
-
         return sps_res_availability_zones_true_df
-
+    return None
 
 @log_execution_time
 def collect_spot_placement_score(desired_counts, instance_types=None):
@@ -459,6 +458,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,
         f"{base_path}/{az_str}/{AZURE_CONST.S3_LOCATIONS_OVER_LIMIT_JSON_FILENAME}": SS_Resources.locations_over_limit_tmp
     }
 
@@ -475,13 +475,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')
+        location_index_data = 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_location_index_data = location_index_data
         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 +493,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,
             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

@@ -47,6 +47,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 False
 
     except Exception as e:
         print(f"Error in check_and_add_available_locations: {e}")
@@ -70,7 +71,7 @@ def validation_can_call(subscription_id, location):
 
 def get_next_available_location():
     """
-     이 메서드는 사용 가능한 다음 위치를 리턴합니다.
+     이 메서드는 사용 가능한 다음 location를 리턴합니다.
      호출 이력과 초과 요청 데이터를 기반으로 적절한 위치를 선택합니다.
      호출 시 이용하는 location은 구독 내에 지난 호출의 location을 이용 안 해야 하는 로직이 들어갑니다.
 
@@ -81,34 +82,59 @@ def get_next_available_location():
         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
+
+        n, m = len(subs), len(locs)
 
-        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]
+        last_pair = getattr(SS_Resources, "last_subscription_id_and_location", None) or {}
+        last_sub_id = last_pair.get("last_subscription_id")
+        last_loc = last_pair.get("last_location")
 
-            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]
+        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
 
-                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
+        attempts = 0
+        while attempts < n * m:
+            sub_id = subs[s_idx]
+            loc = locs[l_idx]
 
-        return None
+            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():

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 = None
 region_map_and_instance_map_tmp = None
 subscriptions = None
 available_locations = None