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

Azure Collector CloudWatch로 수집 개수 전송

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

@@ -8,7 +8,7 @@
 from load_if import load_if
 from merge_df import merge_df
 from load_price import collect_price_with_multithreading
-from upload_data import upload_timestream, update_latest, save_raw, query_selector
+from upload_data import upload_timestream, update_latest, save_raw, query_selector, upload_cloudwatch
 from compare_data import compare
 
 STORAGE_CONST = Storage()
@@ -69,6 +69,9 @@ def azure_collector(timestamp):
         update_latest(join_df, timestamp)
         save_raw(join_df, timestamp)
 
+        # upload count-log to cloudwatch
+        upload_cloudwatch(join_df, timestamp)
+
         # compare and upload changed_df to timestream
         changed_df = compare(previous_df, join_df, AZURE_CONST.DF_WORKLOAD_COLS, AZURE_CONST.DF_FEATURE_COLS)
         if not changed_df.empty:

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

@@ -139,3 +139,22 @@ def query_selector(data):
     s3 = session.resource('s3')
     object_acl = s3.ObjectAcl(STORAGE_CONST.BUCKET_NAME, AZURE_CONST.S3_QUERY_SELECTOR_SAVE_PATH)
     response = object_acl.put(ACL='public-read')
+
+
+def upload_cloudwatch(data, timestamp):
+    ondemand_count = len(data.drop(columns=['SpotPrice', 'Savings']).dropna())
+    spot_count = len(data.drop(columns=['OndemandPrice', 'Savings']).dropna())
+    if_count = len(data.drop(columns=['OndemandPrice', 'SpotPrice', 'Savings']).dropna())
+
+    cw_client = boto3.client('logs')
+
+    log_event = {
+        'timestamp': int(timestamp.timestamp()) * 1000,
+        'message': f'AZUREONDEMAND: {ondemand_count} AZURESPOT: {spot_count} AZUREIF: {if_count}'
+    }
+
+    cw_client.put_log_events(
+        logGroupName=AZURE_CONST.SPOT_DATA_COLLECTION_LOG_GROUP_NAME,
+        logStreamName=AZURE_CONST.LOG_STREAM_NAME,
+        logEvents=[log_event]
+    )

const_config.py

@@ -114,6 +114,14 @@ def FILTER_LOCATIONS():
     def MAX_SKIP():
         return 200
 
+    @constant
+    def SPOT_DATA_COLLECTION_LOG_GROUP_NAME():
+        return "Collection-Data-Count"
+
+    @constant
+    def LOG_STREAM_NAME():
+        return "Azure-Count"
+
 class GcpCollector(object):
     @constant
     def API_LINK():