-
Notifications
You must be signed in to change notification settings - Fork 0
/
lambda.py
63 lines (52 loc) · 2.22 KB
/
lambda.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import boto3
from datetime import datetime
from datetime import timedelta
def lambda_handler(event, context):
idle_threshold_hr = 24 # Change this to your threshold in hours
cw = boto3.client('cloudwatch')
sm = boto3.client('sagemaker')
sns = boto3.client('sns')
try:
inservice_endpoints = sm.list_endpoints(
SortBy='CreationTime',
SortOrder='Ascending',
MaxResults=100,
# NameContains='string', # for example 'dev-'
StatusEquals='InService'
)
idle_endpoints = []
for ep in inservice_endpoints['Endpoints']:
ep_describe = sm.describe_endpoint(
EndpointName=ep['EndpointName']
)
metric_response = cw.get_metric_statistics(
Namespace='AWS/SageMaker',
MetricName='Invocations',
Dimensions=[
{
'Name': 'EndpointName',
'Value': ep['EndpointName']
},
{
'Name': 'VariantName',
'Value': ep_describe['ProductionVariants'][0]['VariantName']
}
],
StartTime=datetime.utcnow()-timedelta(hours=idle_threshold_hr),
EndTime=datetime.utcnow(),
Period=int(idle_threshold_hr*60*60),
Statistics=['Sum'],
Unit='None'
)
if len(metric_response['Datapoints'])==0:
idle_endpoints.append(ep['EndpointName'])
if len(idle_endpoints) > 0:
response_sns = sns.publish(
TopicArn='YOUR SNS TOPIC ARN HERE',
Message="The following endpoints have been idle for over {} hrs. Log on to Amazon SageMaker console to take actions.\n\n{}".format(idle_threshold_hr, '\n'.join(idle_endpoints)),
Subject='Automated Notification: Idle Endpoints Detected',
MessageStructure='string'
)
return {'Status': 'Success'}
except:
return {'Status': 'Fail'}