[AWS]19. DynamoDB TTL과 CloudWatch Event 비교 (feat, 타임어택)

프로젝트 제작 중 타임 어택 기능을 만들어야 하는 부분이 있었는데 시간이 모두 소요되면 사용자에게 디스코드 봇을 통한 알림을 보내게끔 해야 하는 기능이다.

 

처음에 가장 먼저 떠오른것은 DynamoDB의 TTL 기능을 사용해 보는 것이었다.

TTL은 특정 컬럼에 대해 항목별 타임스탬프를 정의하여 항목이 더 이상 필요하지 않은 시점을 결정할 수 있다.

 

보통은 사용자 세션에 대해 서버에서 관리할 경우 많이 사용하는 기능인데 Stream 기능을 통해 간단히 구현해 볼 수 있을 것 같아서 사용하게 되었다.

 

원리는 대충 이렇다.

클라이언트에서 타임 어택 폼을 띄우고 시간이 모두 소요됐을 시 API URL을 통해 Lambda 함수 호출(여기에서는 사용자 이름, TTL(TTL), value 컬럼에 각각 데이터를 집어넣는 로직이 포함되어 있다.)

 

여기서 TTL 컬럼에 집어넣는 값이 이제 만료시간이 되는 거다.

ttl = int((datetime.utcnow() + timedelta(hours=3)).timestamp())

이런 식으로 현재시간으로부터 + 3시간 한 timestamp 값을 저장할 수 있다.

 

하지만 한 가지 간과하고 있는 게 있었는데 DynamoDB의 TTL은 값이 설정된 후 정확히 그 시간에 삭제되는 것이 아니라, DynamoDB의 백그라운드 프로세스릍 통해 주기적으로 만료된 아이템을 삭제하는 것이다.

이 때문에 TTL로 설정한 시간과 실제 삭제 시간 사이에 차이가 발생한다.

 

즉 이렇게 타임어택 기능에 있어서 원하는 때에 사용자에게 바로 데이터를 제공해줘야 하는 부분에 TTL 기능을 사용하는 것은 적합하지 않다.

 

더군다나 Stream 기능을 통한 모든 삭제 데이터가 트리거 되기 때문에 TTL이 모두 소요돼서 삭제되는 데이터와, 타임 어택 도중 임무를 모두 수행해서 수동으로 삭제하는 TTL 데이터를 구분하기 위해 BOOL 속성의 컬럼을 추가해서 true, false 여부에 따라 사용자에게 데이터를 제공했었는데 결국 뻘짓이었다... 

 

그래서 이번엔 CloudWatch Event를 사용하기로 했다.

똑같이 API URL을 통해 Lambda 함수를 호출하여 데이터를 집어넣는 로직과 추가로 CloudWatch Events 규칙을 생성하는 로직이 추가되었다.

 

(규칙 생성)

schedule_expression = f'cron({cron_minute} {cron_hour} * * ? *)'
    print(f"Creating rule with schedule: {schedule_expression}")
    response = events.put_rule(
        Name=rule_name,
        ScheduleExpression=schedule_expression,
        State='ENABLED'
    )
    print(f"PutRule response: {response}")

 

(타겟 설정)
    target_arn = f'arn:aws:lambda:{context.invoked_function_arn.split(":")[3]}:{context.invoked_function_arn.split(":")[4]}:function:ttlsave2'
    response = events.put_targets(
        Rule=rule_name,
        Targets=[
            {
                'Id': target_id,
                'Arn': target_arn,
                'Input': json.dumps({'userid': userid})
            }
        ]
    )

이런 식으로 규칙 이름, 만료 시간(cron), 상태 등을 설정할 수 있고 만료 후 동작되는 대상 또한 타깃으로 설정할 수 있다.

위에서는 ttlsave2라는 lambda 함수를 타겟으로 설정되어 있는데 DynamoDB에 들어간 데이터를 삭제하는 로직이 포함되어 있다.

 

(ttlsave)

import json
import boto3
from datetime import datetime, timedelta
import pytz

dynamodb = boto3.client('dynamodb')
events = boto3.client('events')
table_name = 'failedsavettl'
rule_name = 'DeleteItemRule'
target_id = 'DeleteItemTarget'

def lambda_handler(event, context):
    userid = event['userid']
    value = event['value']
    delay_minutes = event.get('delayMinutes') 

    seoul_tz = pytz.timezone('Asia/Seoul')
    current_time = datetime.now(seoul_tz)
    target_time = current_time + timedelta(minutes=delay_minutes)
    target_time2 = datetime.utcnow() + timedelta(minutes=delay_minutes)
    
    cron_minute = target_time2.minute
    cron_hour = target_time2.hour

    time_str = target_time.strftime('%Y-%m-%dT%H:%M:%S')

    ttl = int(target_time.timestamp())
    dynamodb.put_item(
        TableName=table_name,
        Item={
            'userid': {'S': userid},
            'time': {'S': time_str},
            'value': {'S': value}
        }
    )

    # CloudWatch Events 규칙 생성
    schedule_expression = f'cron({cron_minute} {cron_hour} * * ? *)'
    print(f"Creating rule with schedule: {schedule_expression}")
    response = events.put_rule(
        Name=rule_name,
        ScheduleExpression=schedule_expression,
        State='ENABLED'
    )
    print(f"PutRule response: {response}")

    # CloudWatch Events 타겟 설정
    target_arn = f'arn:aws:lambda:{context.invoked_function_arn.split(":")[3]}:{context.invoked_function_arn.split(":")[4]}:function:ttlsave2'
    response = events.put_targets(
        Rule=rule_name,
        Targets=[
            {
                'Id': target_id,
                'Arn': target_arn,
                'Input': json.dumps({'userid': userid})
            }
        ]
    )
    print(f"PutTargets response: {response}")

    return {
        'statusCode': 200,
        'body': json.dumps('Item saved and delete timer set.')
    }

 

 

 

타임어택이 시작되자마다 이 로직이 동작되는 것이다.

 

그럼 위 사진과 같이 CloudWatch 콘솔의 규칙 대시보드에서 생성된 룰을 확인할 수 있다.

 

대상 탭에서 타켓으로 설정된 Lambda 함수를 확인할 수 있다.

 

(ttlsave2)

import json
import boto3

dynamodb = boto3.client('dynamodb')
table_name = 'failedsavettl'

def lambda_handler(event, context):
    userid = event['userid']

    try:
        response = dynamodb.delete_item(
            TableName=table_name,
            Key={
                'userid': {'S': userid}
            }
        )
        print(f"DeleteItem response: {response}")

        return {
            'statusCode': 200,
            'body': json.dumps('Item deleted successfully.')
        }

    except Exception as e:
        print(f"Error deleting item: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps('Error deleting item.')
        }

 

간단하게 userid를 받아와서 해당 값의 데이터를 삭제하는 로직이다.

이제 삭제가 이루어지면 이전항목에 대해서 Stream 기능을 통해 트리거하는 것이다.

 

그럼 이제 ttldisocrd라는 lambda 함수가 호출이 된다.

import json
import boto3
import requests

def lambda_handler(event, context):
    for record in event['Records']:
        if record['eventName'] == 'REMOVE':
            if 'OldImage' in record['dynamodb']:
                old_image = record['dynamodb']['OldImage']
                time = old_image['time']['S']
                user_id = old_image['userid']['S']
                value = old_image['value']['S']
                
                # Discord API 호출을 위한 데이터
                bot_token = ""
                channel_id = ""
                message = f"```yaml\n😢나약한 자식😢\n┌{'─'*24}┐\n Date: {time}\n Name: {user_id}\n Value: {value}\n└{'─'*24}┘```"
                url = f"https://discord.com/api/v9/channels/{channel_id}/messages"
                headers = {
                    "Authorization": f"Bot {bot_token}",
                    "Content-Type": "application/json"
                }
                data = {
                    "content": message
                }

                response = requests.post(url, headers=headers, data=json.dumps(data))

                if response.status_code == 200:
                    print("Message sent successfully")
                else:
                    print(f"Failed to send message: {response.status_code}, {response.text}")
            else:
                print("OldImage not found in record")

    return {
        'statusCode': 200,
        'body': json.dumps('Processing complete.')
    }

 

old_image = record['dynamodb']['OldImage'] 구조를 통해 삭제된 이전 항목을  불러올 수 있고 해당 데이터를 디스코드 봇을 호출하여 사용자에게 알림을 전송할 수 있다.