SageMakerエンドポイントのIDLE状態は検知できるのか

はじめに

ご覧いただきありがとうございます。阿河です。

https://zenn.dev/megazone_japan/articles/e365cea61a7af0

以前SageMakerリソースのIDLE検知について記事を書きましたが、課金という面ではエンドポイントの消し忘れのほうがお財布に響くよなー(というか私が痛い目にあったことがある)と思っていたら、お客様からも同じような御相談を受けました。

ということで安心してSageMakerサービスを漫喫頂くために、エンドポイントのIDLE状態を検知する方法を検討しました。

概要

1. ひとまず検知
2. 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エンドポイントが2つあります。
うち一方は作成から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つのエンドポイントを検知できます。
環境に合わせて、閾値を変更下さい。

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に送る処理を入れました。

無事に通知が届きました。

おわりに

ここまで御覧いただき ありがとうございました。
誰かの参考になれば幸いです。