ITing - [Tech blog] SageMaker 엔드포인트의 IDLE 상태를 감지하는 방법

들어가며

이전 게시글을 통해 SageMaker 리소스 IDLE 감지에 대한 기사를 기재했는데요, 과금면에서 엔드포인트 삭제를 잊는 것이 더 돈이 들지 않을까 하지 않던 차에 (사실 저도 이런 쓰린 경험이 있었습니다) 고객으로부터도 같은 상담을 받았습니다.

이러한 이유로 안심하고 SageMaker 서비스를 만끽하기 위해 엔드 포인트의 IDLE 상태를 감지하는 방법을 검토해 보았습니다.

개요

일단 감지
Slack에 매일 경고 날리기

1. 일단 감지

https://docs.aws.amazon.com/cli/latest/reference/sagemaker/list-endpoints.html

SageMaker의 list-endpoints에서 ‘–creation-time-before’옵션이 있습니다. 예를 들어 24시간 전에 생성된 엔드포인트를 임계값으로 필터링할 수 있다면 요구 사항을 만족 할 수 있을 것 입니다.

※Lambda 함수 설정

런타임: Python3.12
아키텍처: x86_64

※Lambda 코드

import json
import os
import boto3
import urllib3
http = urllib3.PoolManager()
from datetime import datetime, timedelta, timezone

def set_threshold_value():
    
    nowadays = datetime.now()
    yesterday = nowadays - timedelta(1)
    year = yesterday.year
    month = yesterday.month
    day = yesterday.day
    hour = yesterday.hour
    minute = yesterday.minute
    
    date = datetime(year, month, day, hour, minute, tzinfo=timezone.utc)
    print(date)

    return date
    

def list_target_instances(sagemaker, date):
    
    response = sagemaker.list_endpoints(
        SortBy='Name',
        StatusEquals='InService',
        CreationTimeBefore=date
        )
    return response['Endpoints']
    
def endpoint_info(sagemaker, target):
    
    dict = {}
    jst = timezone(timedelta(hours=+9))
    
    for i in target:
        endpoint_name = i['EndpointName']
        creation_time = i['CreationTime']
        modified_time = i['LastModifiedTime']
        
        creation_time_jst = creation_time.astimezone(jst).strftime('%Y-%m-%d %H:%M')
        modified_time_jst = modified_time.astimezone(jst).strftime('%Y-%m-%d %H:%M')
        
        dict[endpoint_name] = [creation_time_jst, modified_time_jst]
    
    return dict

def lambda_handler(event, context):
    
    sagemaker = boto3.client('sagemaker')
    
    date = set_threshold_value()
    endpoints_list = list_target_instances(sagemaker, date)
    endpoints_info = endpoint_info(sagemaker, endpoints_list)
    
    message_text = "하기SageMaker Endpoint에 IDLE상태 의심 사항이 있습니다. 매니지먼트 콘솔에서 확인을 부탁 드립니다..\n"
    
    for endpoint, times in endpoints_info.items():
        creation_time, modified_time = times
        message_text += f"EndpointName: {endpoint}, CreationTime: {creation_time}, ModifiedTime: {modified_time}\n"
    
    print(message_text)

현재 SageMaker 엔드포인트가 두 개 있습니다.

그 중 하나는 생성에서 24시간 이상 경과하고 있는 것이고 다른 하나는 생성한 후 10분 정도 경과하였습니다.

하기 SageMaker EndpointにIDLE상태 의심 사항이 있습니다. 매니지먼트 콘솔에서 확인을 부탁 드립니다.
EndpointName: xxxx-endpoint, CreationTime: 2024-03-05 04:51, ModifiedTime: 2024-03-05 04:54

24시간이 경과한 엔드포인트만 감지할 수 있었습니다.

def set_threshold_value():
    
    nowadays = datetime.now()
    yesterday = nowadays - timedelta(1)
    year = nowadays.year
    month = nowadays.month
    day = nowadays.day
    hour = nowadays.hour
    minute = nowadays.minute
    
    date = datetime(year, month, day, hour, minute, tzinfo=timezone.utc)
    print(date)

    return date

하기 SageMaker Endpoint에 IDLE상태 의심 사항이 있습니다. 매니지먼트 콘솔에서 확인을 부탁 드립니다.
EndpointName: xxxx-test-endpoint-1, CreationTime: 2024-03-08 19:21, ModifiedTime: 2024-03-08 19:24
EndpointName: xxxx-endpoint, CreationTime: 2024-03-05 04:51, ModifiedTime: 2024-03-05 04:54

참고로 시간 임계 값을 변경하면 두 개의 엔드 포인트를 감지 할 수 있습니다. 환경에 맞게 임계값을 변경 하시길 바랍니다.

2. Slack에 매일 경보를 날리기

메시지를 Slack에 알릴 수 있습니다.

Lambda에 환경 변수를 설정합니다.

import json
import os
import boto3
import urllib3
http = urllib3.PoolManager()
from datetime import datetime, timedelta, timezone

url = os.environ['URL']

def set_threshold_value():
    
    nowadays = datetime.now()
    yesterday = nowadays - timedelta(1)
    year = yesterday.year
    month = yesterday.month
    day = yesterday.day
    hour = yesterday.hour
    minute = yesterday.minute
    
    date = datetime(year, month, day, hour, minute, tzinfo=timezone.utc)
    print(date)

    return date
    

def list_target_instances(sagemaker, date):
    
    response = sagemaker.list_endpoints(
        SortBy='Name',
        StatusEquals='InService',
        CreationTimeBefore=date
        )
    return response['Endpoints']
    
def endpoint_info(sagemaker, target):
    
    dict = {}
    jst = timezone(timedelta(hours=+9))
    
    for i in target:
        endpoint_name = i['EndpointName']
        creation_time = i['CreationTime']
        modified_time = i['LastModifiedTime']
        
        creation_time_jst = creation_time.astimezone(jst).strftime('%Y-%m-%d %H:%M')
        modified_time_jst = modified_time.astimezone(jst).strftime('%Y-%m-%d %H:%M')
        
        dict[endpoint_name] = [creation_time_jst, modified_time_jst]
    
    return dict

    
def send_slack_notification(webhook_url, message_text):
    
    message = {"text": message_text}
    encoded_message = json.dumps(message).encode('utf-8')
    response = http.request(
        'POST',
        webhook_url,
        body=encoded_message,
        headers={'Content-Type': 'application/json'}
    )

    return response
    

def lambda_handler(event, context):
    
    sagemaker = boto3.client('sagemaker')
    
    date = set_threshold_value()
    endpoints_list = list_target_instances(sagemaker, date)
    endpoints_info = endpoint_info(sagemaker, endpoints_list)
    
    message_text = "하기 SageMaker Endpoint에 IDLE상태 의심 사항이 있습니다. 매니지먼트 콘솔에서 확인을 부탁 드립니다.\n"
    
    for endpoint, times in endpoints_info.items():
        creation_time, modified_time = times
        message_text += f"EndpointName: {endpoint}, CreationTime: {creation_time}, ModifiedTime: {modified_time}\n"
    
    send_slack_notification(url, message_text)

Slack에 보내는 처리를 넣었습니다.