1+ import boto3
2+ import json
3+ import base64
4+ import re
5+ import time
6+
7+ REGION_NAME = "ap-northeast-2"
8+ BUCKET_NAME = "snorose-bucket"
9+ IMAGE_KEY = "test/12345/test_video.mp4"
10+ FUNC_CON = "CreateThumbnail"
11+ FUNC_IMG = "CreateThumbnailECR"
12+
13+ client = boto3 .client ('lambda' , region_name = REGION_NAME )
14+
15+ # 이벤트 객체
16+ payload_data = {
17+ "bucket" : BUCKET_NAME ,
18+ "key" : IMAGE_KEY
19+ }
20+ payload_bytes = json .dumps (payload_data ).encode ('utf-8' )
21+
22+ def trigger_cold_start (func_name ):
23+ """
24+ 함수의 환경 변수에 현재 시간을 업데이트하여 강제로 Cold Start 환경을 조성함
25+ """
26+ print (f"[Setup] { func_name } 환경 변수 업데이트 중 (Cold Start 유도)..." )
27+
28+ config = client .get_function_configuration (FunctionName = func_name )
29+ env_vars = config .get ('Environment' , {}).get ('Variables' , {})
30+
31+ # 임의의 변수 업데이트
32+ env_vars ['TEST_TIMESTAMP' ] = str (time .time ())
33+
34+ client .update_function_configuration (
35+ FunctionName = func_name ,
36+ Environment = {'Variables' : env_vars }
37+ )
38+
39+ print (f"{ func_name } 업데이트 진행 중..." )
40+ while True :
41+ response = client .get_function (FunctionName = func_name )
42+ status = response ['Configuration' ]['LastUpdateStatus' ]
43+ state = response ['Configuration' ]['State' ]
44+
45+ if state == 'Active' and status == 'Successful' :
46+ print ("업데이트 완료! \n " )
47+ break
48+ elif status == 'Failed' :
49+ raise Exception (f"함수 업데이트 실패: { response ['Configuration' ]['LastUpdateStatusReason' ]} " )
50+
51+ time .sleep (2 )
52+
53+ def run_test (func_name , label ):
54+ # 테스트 직전, 강제로 Cold Start 유발
55+ trigger_cold_start (func_name )
56+
57+ print (f"▶ Testing: { label } [{ func_name } ]" )
58+
59+ try :
60+ start_time = time .time ()
61+ response = client .invoke (
62+ FunctionName = func_name ,
63+ InvocationType = 'RequestResponse' ,
64+ LogType = 'Tail' ,
65+ Payload = payload_bytes
66+ )
67+ end_time = time .time ()
68+
69+ log_result = base64 .b64decode (response ['LogResult' ]).decode ('utf-8' )
70+
71+ # Init Duration 추출
72+ init_match = re .search (r"Init Duration:\s*([\d.]+)\s*ms" , log_result )
73+ duration_match = re .search (r"Duration:\s*([\d.]+)\s*ms" , log_result )
74+
75+ init_time = float (init_match .group (1 )) if init_match else 0.0
76+ exec_time = float (duration_match .group (1 )) if duration_match else 0.0
77+
78+ print (f" ▷ [Total Client Latency] : { round ((end_time - start_time ) * 1000 , 2 )} ms" )
79+ print (f" ▷ [Lambda Execution] : { exec_time } ms" )
80+
81+ # Cold Start 결과 강조
82+ if init_time > 0 :
83+ print (f" [Cold Start Init] : { init_time } ms" )
84+ else :
85+ print (f" [Cold Start Init] : 0 ms (실패: 이미 Warm 상태임)" )
86+
87+ except Exception as e :
88+ print (f"에러 발생: { e } " )
89+
90+ if __name__ == "__main__" :
91+ print ("=== Lambda 성능 비교 테스트 시작 ===\n " )
92+
93+ # Console 버전 테스트
94+ run_test (FUNC_CON , "Console Version" )
95+
96+ # Container 버전 테스트
97+ run_test (FUNC_IMG , "Container Image Version" )
98+
99+ print ("\n === 테스트 완료 ===" )
0 commit comments