[토이 프로젝트] 클라이밍 스스로 규칙 앱 개발 Feat. S3 presigned_url

클라이밍 하는 친구와 함께 관련 앱을 만들면 어떨까 싶어 개발하게 되었다.

자신의 주 암장, 오늘 깬 난이도, 개수, 커뮤니티 기능 등 관련 앱은 굉장히 많지만 뭔가 스스로가 정한 할당량을 오늘 채우지 않을 시 수행해야 하는 벌칙? 같은 규칙이 있었으면 좋겠다는 생각에 간단히 만들어보았다.

 

'나루토' 애니메이션에서 나오는 가이가 수행하는 스스로 규칙이라고 생각하면 될 것 같다..ㅋㅋ

 

먼저 클라이언트 플랫폼은 서로 다르게 만들기로 하였다.

그 이유는 서로 추구하는 기술 스택이 다르고 쌓아야 하는 포트폴리오의 방향성이 다르기 때문에 서로 데이터를 공유할 수 있는 클라우드만 공유하기로 했다. (S3)

 

먼저 필수적인 기능은 이렇다.

• 조건 설정(운동하는 날 수행해야 하는 조건 예, 보라색 난이도 5개, 못했을 시 풀업 또는 푸쉬업 몇 개 수행)
• 날짜(투데이로 자동 설정)과 오늘 수행한 난이도, 몇 개 깼는지 표시
• 위에 설정한 조건에 충족하지 못했을 시 밑에 영상 업로드란 표기(3시간 동안 시간 카운팅 표시, 로컬 스토리지 사용)
• 풀업 또는 푸쉬업을 조건대로 수행하고 영상을 찍어 S3에 업로드
• 업로드 시 친구와 정보를 공유할 수 있는 디스코드에서 봇을 활성화하여 영상을 볼 수 있는 링크 제공
• 영상은 한달 지날 시 Standard-IA로 이전
• 리스트 폼까지 제작하여 앱에서 이때까지 올린 영상 리스트를 확인할 수 있어야 함

위 기능은 무조건적으로 들어가야 하고 이후 추가적으로 기능을 확장할 수 있다.

(새로운 인프라 솔루션을 이용하는 것이 목적이기에 최소한의 디자인과 기능만을 구현)

---

인프라

Serverless Architecture

 

인프라가 이것저것 얽히고 얽히다 보면 도식 그리는 게 젤 힘든 것 같다.

어떻게 하면 좀 더 이쁘게 만들 수 있을까..

 

친구와 운동하면서 꾸준히 같이 이용해나가자는 목적으로 만들었기 때문에 비용적인 부분에 있어서 제일 효율적인 서버리스로 제작하게 되었다. (그 친구는 스프링을 통해 제작할 텐데 동적 서버가 필요하기에 제작에만 목적으로 두고 인프라 쪽을 구성하기로 했다. 서버리스 X)

 

이번 프로젝트의 핵심은 presigned_url 이다.

S3의 사전 서명된 URL을 발급받아 프라이빗 환경에서도 효율적으로 사용자들에게 콘텐츠를 제공할 수 있다.

 

원래대로라면 클라이언트에서 S3에 접근하여 콘텐츠에 액세스 하기 위해서는 자격증명을 발급받아 인증된 액세스 또는 게스트 액세스의 자격증명을 이용해 접근해야 하지만 앱 구성 자체가 서버리스이기도 하고 정적 콘텐츠에 자격증명이 노출이 되어 보안상에 굉장히 좋지 않다. 

 

물론 API Gateway를 통해 Lambda 측에서 인증하여 안전하게 액세스 할 수 있지만 오버헤드를 생각했을 때 presigned_url을 사용하는 것이 더 적은 오버헤드와 효율적인 액세스가 가능하다고 판단했다.

 

여기서 중요한 점은 presigned_url에서 제공하는 메서드가 GET 뿐만이 아니라 POST, DELETE, HEAD 메서드도 지원을 한다는 것이다.

 

간단히 코드를 통해 살펴보겠다.

 

PUT Object

 

PUT_Object

먼저 Put_Object에 대한 Lambda 함수이다.

 

s3.generate_presigned_url라는 함수를 사용하여 사전 서명된 URL을 생성할 수 있고 어떤 동작의 메서드인지 정할 수 있다. (put_object에 대한 url을 생성)

 

URL을 발급받아 HTTP 메서드를 실행하는 것이기에 API Gateway에서 GET으로 데이터를 반환하고 클라이언트 측에서

 

 

fetch 함수로 실행하여 수행 영상을 파라미터 매개변수를 통해 업로드하는 것이다.

 

업로드 실패 시

만약 제한 시간 안에 업로드를 실패했을 경우 failed log 기록을 DynamoDB에 삽입하여 리스트 폼에서 불러오는 것이다.

Stream 기능을 활성화하여 데이터가 들어가면 디스코드 봇을 호출하여 친구에게도 나에 대한 실패 알림을 띄울 수 있다.

import requests
import json
from decimal import Decimal
import boto3

REGION = 'ap-northeast-2'

dynamodb = boto3.resource('dynamodb', region_name=REGION)
table = dynamodb.Table('')

def convert_decimal_to_str(obj):
    if isinstance(obj, Decimal):
        return str(obj)
    raise TypeError


def send_discord_message(bot_token, channel_id, message):
    url = f"https://discord.com/api/v9/channels/{channel_id}/messages"
    headers = {
        "Authorization": f"Bot {bot_token}",
        "Content-Type": "application/json"
    }
    payload = {
        "content": message
    }
    response = requests.post(url, headers=headers, data=json.dumps(payload))
    if response.status_code != 200:
        print("Failed to send message to Discord")
    else:
        print("Message sent successfully")

def lambda_handler(event, context):
    response = table.scan()
    data = response['Items']
    
    # 데이터를 날짜 기준으로 정렬합니다.
    sorted_data = sorted(data, key=lambda x: x['idx'], reverse=True)
    
    # 가장 최신 데이터를 가져옵니다.
    latest_data = sorted_data[0]
    
    # 최신 데이터를 이용하여 메시지를 생성합니다.
    message = f"```yaml\n😢나약한 자식😢\n┌{'─'*24}┐\n Date: {latest_data['idx']}\n Name: {latest_data['name']}\n Value: {latest_data['value']}\n└{'─'*24}┘```"
    
    # Discord로 메시지를 전송합니다.
    bot_token = ""
    channel_id = ""
    send_discord_message(bot_token, channel_id, message)

    return {
        'statusCode': 200,
        'body': 'Message sent to Discord!'
    }

 

GET_Object

GET_Object

 

마찬가지로 리스트 폼에서 오브젝트 리스트들을 불러올 수 있는 s3_client.list_objects_v2 함수를 사용하여 불러오고 그거에 대한 view를 볼 수 있는 사전 서명된 URL을 발급받아 사용자에게 콘텐츠를 제공할 수 있다.

(failedlog 기록 또한 DynamoDB에서 불러올 수 있다.)

 

S3 이벤트 알림

마찬가지로 S3에서도 수행 영상이 업로드되면 Lambda를 통해 디스코드 봇을 호출한다.

 

이 작업을 수행 중 몇 가지 문제점이 발생하였다.

사전 서명된 URL의 길이가 너무 길어 디스코드에서 볼 때 굉장히 지저분하게 호출이 된다.

 

이런 식으로 영상 링크가 나오게 되면 누가 올렸는지 구분도 잘 안되고 지저분해서 누르기 싫어진다..

즉 링크를 줄일 수 있는 방법이 두 가지 정도 있었는데 첫 번째는 Bitly Links를 이용하는 것이다.

 

shortened_url = shorten_url(presigned_url) 구문을 통해 줄인 URL로 반환하여 호출할 수 있지만 링크가 너무 길어서인지 제대로 반환이 안 되는 것 같았다. (짧은 링크로 테스트해봤는데 잘 된다.)

근데 또 API 대시보드에서는 생성이 잘 됐는데 람다 호출 시간문제인 건지... 정확한 문제는 아직 파악하지 못했다.

 

하지만 Bitly Links을 사용하지 않은 건 그뿐만이 아니었고 한 달 동안 API을 호출할 수 있는 제한이 있었기에 그렇게 효율적인 방법은 아니라고 생각했다.

 

 두 번째 방법은 디스코드 봇을 호출 할 때 임베드를 사용하는 방법이다.

 data = {
        "content": "",
        "embeds": [
            {
                "title": f"New Video Uploaded",
                "description": "A new video has been uploaded. You can access it using the link below.",
                "fields": [
                    {
                        "name": file_name,
                        "value": f"[Click here to view the video]({url})"
                    }
                ]
            }
        ]
    }

이런 식으로 링크를 매개변수로 집어넣어서 URL을 가리고 클릭 가능한 텍스트로 만들어서 보낼 수 있다.

 

하지만 여기에서도 문제가 발생한다.

value: 최대 1024자가 제한인데 링크는 1100자 이상이 되어버려서 중간에서 잘려서 보내진다.

 

그래서 선택한 방식이 리다이렉트 Lambda 함수를 하나 만들고 API Gateway URL을 통해 중간 프록시 역할 할 수 있게 만들었다.

 

import json
import boto3
import urllib.parse
import requests
import logging

DISCORD_BOT_TOKEN = ''
DISCORD_CHANNEL_ID = ''
API_GATEWAY_URL = ''

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3_client = boto3.client('s3')

def lambda_handler(event, context):
    try:
        # Extract bucket name and object key from the event
        bucket_name = event['Records'][0]['s3']['bucket']['name']
        object_key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'])
        file_name = object_key.split('/')[-1]  # 파일 이름 추출

        # Generate the API Gateway URL
        api_gateway_url = f"{API_GATEWAY_URL}/urlaccess_get?bucket={bucket_name}&key={object_key}"
        
        # Send the message via Discord
        send_discord_message(api_gateway_url, file_name)  # file_name 추가

        return {
            'statusCode': 200,
            'body': json.dumps('Message sent successfully!')
        }
    except Exception as e:
        logger.error("Error: " + str(e))
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def send_discord_message(url, file_name):  # file_name 매개변수 추가
    api_url = f"https://discord.com/api/v9/channels/{DISCORD_CHANNEL_ID}/messages"
    headers = {
        'Authorization': f'Bot {DISCORD_BOT_TOKEN}',
        'Content-Type': 'application/json'
    }
    data = {
        "content": "",
        "embeds": [
            {
                "title": f"New Video Uploaded",
                "description": "A new video has been uploaded. You can access it using the link below.",
                "fields": [
                    {
                        "name": file_name,
                        "value": f"[Click here to view the video]({url})"
                    }
                ]
            }
        ]
    }
    response = requests.post(api_url, headers=headers, json=data)
    if response.status_code == 200 or response.status_code == 204:
        logger.info("Message sent successfully to Discord")
    else:
        logger.error(f"Failed to send message to Discord: {response.status_code}, {response.text}")
        raise Exception(f"Failed to send message to Discord: {response.status_code}, {response.text}")

여기서 api_gateway_url = f"{API_GATEWAY_URL}/urlaccess_get?bucket={bucket_name}&key={object_key}"

파라미터를 통해 Redirect 함수로 전달한다.

 

import json
import boto3

def lambda_handler(event, context):
    s3 = boto3.client('s3')
    bucket = event['queryStringParameters']['bucket']
    key = event['queryStringParameters']['key']

    try:
        url = s3.generate_presigned_url('get_object',
                                        Params={'Bucket': bucket, 'Key': key},
                                        ExpiresIn=259200)
        return {
            'statusCode': 302,
            'headers': {
                'Location': url
            }
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps(str(e))
        }

 

전달받은 데이터를 참조하여 사전 서명된 URL을 생성하여 접속할 수 있는 원리이다.

이렇게 구성하면 길이 제한에 구애받지 않고 디스코드 임베드를 생성할 수 있다.

 

이런식으로 깔끔하게 메시지가 전송된다.

 

캐싱 자동 무효화 전략

전에 했던 학기 프로젝트에서 새롭게 알았던 구성인데 이번 토이 프로젝트에 적용해 보았다.

 

일반적인 캐싱 전략

 

간단히 설명하자면 현재는 이런 식으로 구성되어서 특정 .mp4 같은 특정 오브젝트들이 캐싱 미스가 뜨는 경우가 발생한다.

그럼 불필요한 리소스를 계속 낭비하게 된다.
이런 문제를 해결하고자 밑에처럼 구성할 수 있다.

이렇게 구성하면 S3까지 트래픽이 이동하지 않기 때문에 비용 효율적으로 운영할 수 있고 S3 버킷에 업로드될 때마다 Cloudfront에 적재시킬 수 있기 때문에 처음 페이지를 로드할 때도 캐싱 히트로 로드시킬 수 있다. 

대신 Cloudfront는 TTL 기준으로 데이터를 업데이트하기 때문에 Invalidations(무효화)를 통해 수동으로 캐싱 삭제를 해야 하지만 이 작업 또한 Lambda 함수를 호출하여 자동화시킬 수 있기 때문에 훨씬 효율적이다.

 

Lambda를 통한 자동 무효화 수행

 

이렇듯이 Invalidations을 자동 무효화를 통해 좀 더 효율적인 오버헤드를 기대해 볼 수 있는 구성이다.

 

import boto3
import time
import os

def lambda_handler(event, context):
    paths = []
    for items in event["Records"]:
        key = items["s3"]["object"]["key"]
        if key.endswith("index.html"):
            paths.append("/" + key[:-10])
        paths.append("/" + key)
    print("Invalidating " + str(paths))

    client = boto3.client('cloudfront')
    batch = {
        'Paths': {
            'Quantity': len(paths),
            'Items': paths
        },
        'CallerReference': str(time.time())
    }
    invalidation = client.create_invalidation(
        DistributionId=os.environ['CLOUDFRONT_DISTRIBUTION_ID'],
        InvalidationBatch=batch,
    )
    return batch

(CLOUDFRONT_DISTRIBUTION_ID는 환경변수 추가)

 

 

CloudWatch 로그 그룹 S3에 업로드 된 즉시 무효화가 이루어진 모습

 

---

클라이언트

클라이언트는 WebView로 제작