MSA 기반 미디어 업로드 고도화: 람다 함수 구조 변경으로 유지보수성 향상

안녕하세요 :) 펫프렌즈 백엔드 개발자 팀버(Timbre/박기오) 입니다 😁

2023년 9월, 커뮤니티(집사생활) 서비스는 출시 이후 끊임없이 고도화를 이어 가고 있습니다. 이번 포스팅에서는 커뮤니티 서비스의 핵심 기능 중 하나인 미디어 업로드의 고도화를 진행하게된 배경과 그 과정을 공유를 하려고합니다.

미디어 업로드 고도화의 사전 작업으로는 Presigned URL 도입이 있었습니다. Presigned URL 의 적용 과정과 그 효과에 대해서 더 자세히 알고 싶으신 분들은 관련 포스트를 참고해 보시는걸 추천합니다!

미디어 업로드 고도화 TO-BE Flow

미디어 업로드 프로세스의 결과가 위 다이어그램과 같이 최적화하기 위해 어떤 선택을 했는지 그 과정에 대해서 생생히 전달드리겠습니다. 😊

#0. 미디어 업로드 고도화 배경

펫프렌즈에서는 MSA 아키텍쳐 기반으로 서비스가 설계되어 있으며, 각 서비스는 필요로하는 각기 다른 DB와 데이터 처리방식을 사용하고 있습니다.

이를 기반으로, Lambda 함수 역시 MSA 아키텍쳐의 핵심 원칙인 서비스 독립성을 갖춘 구성이 필요했습니다. 각 Lambda가 가지고 있는 고유한 정보는 메세지를 발행하여 제공하며, 메세지가 필요한 서비스는 해당 값을 수신해 개발환경과 요구사항에 맞추어 자유롭게 데이터를 활용 할 수 있도록 변경하였습니다.

#1. 기존 람다의 문제점

1. 모든 Lambda에서 동일한 IAM 역할 사용

• 하나의 IAM 역할을 사용하다 보니, 특정 Lambda에서는 사용하지 않는 불필요한 권한도 과도하게 포함하고 있는 문제가 있었습니다.

2. 과도한 코드분기 처리

Lambda간 호출을 위해 다음과 같이 함수명을 할당하는 분기 코드가 다수 존재 했습니다.

lambda_function_name = "mediaconvert_s3_video_db_handler"
if service_mode == "test": 
    lambda_function_name = "test_mediaconvert_s3_video_db_handler"

또한, 버킷명을 기준으로 환경 변수를 설정하는 분기코드도 포함되어 있었습니다.

if event_bucket_name.startswith( "test" ): 
    service_mode = "test"

3. 불필요한 DB 업데이트 로직 존재

미디어 업로드의 실시간 진행률(percent)을 기록하기 위한 DB 업데이트 로직이 존재하였지만, 실제로 이 정보는 사용되지 않고 있었습니다. 결과적으로는 업로드 성공/실패 여부만 판단하는 상황이었습니다.

4. 하드코딩된 테스트용 더미 데이터

Lambda 테스트를 위해 아래와 같은 테스트용 데이터가 하드코딩 되어 있었습니다.

json_data = """{
            "event_bucket_name": "petfriends-video",
            "event_object_key": "file/test/devel/2020/05/14/37/aaaa.mp4",
            "event_object_key_size": 14861445,
            "service_mode": "test",
            "date_path": "2020/05/14",
            "video_id": "22",
            "video_key": "123456",
            "video_key_ext": "mp4",
            "event_type": "AfterCreateMediaConvertJob"
            ...
        }"""

이로 인해 코드의 가독성과 유지보수성이 떨어지는 문제가 있었습니다.

또한, AWS Lambda Console에서 편리한 동작 테스트 방식을 제공합니다.

5. Lambda 간 복잡한 호출 구조 (AS-IS)

Lambda 간 호출이 지나치게 많아서 업로드 과정의 흐름을 직관적으로 파악하기 어려운 문제가 있었습니다.

이러한 문제점들에 의해서 미디어 업로드 고도화를 진행하게 되었습니다.

#2. 환경별 손쉬운 배포 방식 적용

• chalice 배포
• 환경별 분리를 위한 분기코드를 모두 제거
• 람다별 별도의 IAM 권한 및 역할 생성

#.chalice/config.json

{
  "version": "2.0",
  "app_name": "video_input",
  "stages": {
    "dev": {
      ...(개발 환경에 필요한 정보)
      "environment_variables": {
        "BUCKET_NAME": "petfriends-video",
        "MEDIA_VIDEO_TOPIC": "develop.common.media-video"
      },
      "lambda_functions": {
        "lambda_handler": {
          "security_group_ids": ["..."],
          "subnet_ids": ["..."]
        }
      }
    },
    "prod": {
      ...(상용 환경에 필요한 정보)
      }
    }
  },
  "manage_iam_role": false,
  "iam_role_arn": "arn:aws:iam::(...)/video_input_lambda_execution"
}

#3. 마이그레이션 계획

RDB 에서 DynamoDB 로 전환

1) 작업 테이블

• 비디오 크기와 길이 정보, 썸네일 추출 정보, 저장경로 정보 등 업로드시 필요로하는 메타데이터 정보를 저장하는 용도의 테이블

2) 전환 목적

• AWS Lambda 와 같이 서버리스로 관리되는 서비스로, 인프라 관리에 대한 리소스 소모를 최소화 할 수 있습니다.
• 트랜잭션 처리나 테이블 간의 관계가 복잡한 데이터 모델을 다루어야 하는게 아닌, 고정된 값에대한 단순 조회 작업이기 때문에 적합하다고 판단 하였습니다.

RDB 에서 DocumentDB 로 전환

1) 작업 테이블

• 비디오가 업로드된 결과 정보를 저장하는 용도의 테이블 (예: URL 등..)
• 다른 서비스에서는 해당 테이블을 사용하기때문에, 커뮤니티에서 업로드된 비디오만 이관

2) 전환 목적

• 커뮤니티는 DocumentDB 를 사용하기 때문에 기존 RDB를 사용하던 VIDEO 테이블 정보를 이관하여 데이터를 통합 관리할 수 있게 됩니다.

작업 순서

1) 기존 로직 수정 및 선배포

• 기존 RDB에 저장하던 커뮤니티 관련 비디오 업로드 메타데이터 정보들을 DocumentDB 에도 저장하는 애플리케이션 로직 선배포

2) Presigned URL 배포 (해당 포스트 참고)

• 비디오 업로드 최적화를 위해서 Presigned URL 배포

3) RDB → DocumentDB 마이그레이션

• VIDEO 테이블에서 커뮤니티 식별자를 가진 비디오 정보 모두 조회
• DocumentDB 의 post 컬렉션에 100건씩 끊어(batch_size = 100 으로 설정) 저장

5) 모니터링 진행

• 마이그레이션 진행 상황 모니터링

6) RDB 저장 로직 제거

• RDB에 데이터를 저장하는 애플리케이션 코드를 걷어내고, DocumentDB로 완전히 전환

#4. 생성된 람다함수별 역할 (세부 작업)

mediaconvert_video_input

실제 경로로 업로드 되기전 원본파일을 temp 디렉토리에 먼저 업로드하고, 해당 S3 버킷에 트리거되어 실행되는 Lambda 함수

1. 비디오 업로드 트리거

• petfriends-video라는 S3 input bucket 에서 trigger 되어 Lambda 함수가 실행됩니다.
• input bucket 에서 넘겨받은 event 를 통해 메타데이터를 추출합니다.

2. DynamoDB에서 설정 정보 조회

• 추출된 메타데이터 정보를 활용하여 비디오 output 정보와 썸네일 저장정보 및 컨버팅 정보를 조회합니다.

3. MediaConvert 작업 실행

• 필요 정보 확인 후, MediaConvert Job 생성 및 실행하여 비디오 파일을 변환하과 썸네일 추출을 진행합니다.

4. 인코딩 결과 알림

• 컨버팅이 완료되면 카프카를 통해 성공 혹은 실패 메세지를 발행합니다.

@app.on_s3_event(bucket=os.environ.get('BUCKET_NAME'), events=['s3:ObjectCreated:*'], prefix=os.environ.get('PREFIX'))
def lambda_handler(event):
    try:
        # 1. 이벤트로 전달받은 비디오 메타데이터를 추출
        result = videoutils.get_video_info_from_s3_event_object_key(_SERVICE_PROPERTIES, event.to_dict())
        if not result['status']:
            encoding_failed(result, video_info)

        video_info = result['body']['video_info']
        result = dbutils.get_video_setting_info(video_info)

        if result['statusCode'] != 200:
            encoding_failed(result, video_info)

        video_info = result['body']['video_info']
        output_video_settings = result['body']['output_video_settings']

        # 2. 정해진 규칙에 따라 실제 경로에 비디오를 업로드 하기위해 필요한 정보를 조회 (DynamoDB)
        result = mcutils.get_mediaconvert_job(video_info, output_video_settings)
        if not result['status']:
            encoding_failed(result, video_info)

        # 3. AWS MediaConvert 를 실행시켜 서비스에 적합한 파일로 변환하고 썸네일 추출
        result = mcutils.create_mediaconvert_job(_SERVICE_PROPERTIES,
                                                 mediaconvert_job_settings, video_info)
        if not result['status']:
            encoding_failed(result, video_info)

        # 4. 비디오 업로드 성공 토픽 발행
        kafka.send_media_video_encoding_success_topic(video_info)

        ....(로직)

# 비디오 업로드 실패 토픽 발행
def encoding_failed(result, video_info):
    kafka.send_media_video_encoding_failed_topic(video_info, result)
    return error_response(result)

stepfunction_trigger

AWS Step Functions 을 실행하는 함수

1. S3 input bucket 에서 trigger 되어 실행되는 람다

• MediaConvert 에서 컨버팅 완료되어 저장된 비디오 파일, 썸네일 이미지에 trigger 되어 실행

2. step function execute

• input bucket 에서 넘겨받은 event 를 step functions 으로 전달

@app.on_s3_event(bucket=os.environ.get('BUCKET_NAME'), events=['s3:ObjectCreated:*'], prefix=os.environ.get('PREFIX'), suffix='_main.m3u8')
def lambda_handler(event):
    try:
        if '_main.m3u8' in file_key:
            # Step Functions 클라이언트 생성
            stepfunctions = boto3.client('stepfunctions')

            # 실행할 State Machine의 ARN 설정
            state_machine_arn = os.environ.get('STEPFUNCTION_ARN')
            event_data = event.to_dict()

            # Step Functions를 시작하고 실행 ID를 반환
            response = stepfunctions.start_execution(
                stateMachineArn=state_machine_arn,
                input=json.dumps(event_data)  # JSON 형식으로 입력 데이터 전달
            )

          ...(로직)

mediaconvert_ffmpeg

Step Functions 에 의해 실행되어, 업로드한 파일의 메타데이터를 추출해 전달하는 함수

1. Step Functions 에서 전달받은 event 정보를 이용하여 메타데이터 추출

• duration, ratio, codec, format, rotate, width/height 등등
• S3 Key 을 파싱해서 영상 정보가 올바른지 확인
• 영상에 대한 기본 Metadata 정보 추출

2. 추출한 데이터를 step functions 단계에 맞춰 mediaconvert_output 로 전송

• Encoding Job 을 생성해서 MediaConvert 로 전달

@app.lambda_function()
def lambda_handler(event, context):
    try:
        # 메타데이터 추출
        result = videoutils.get_video_info_from_output_s3_event_object_key(_SERVICE_PROPERTIES, event)
        video_info['metadata'] = result['body']['metadata']

        ...(로직)

    # return 하게되는 결과를 step functions의 다음 순서에 있는 람다함수가 받게 됩니다
    return {'video_info': video_info}

mediaconvert_video_output

Step Functions 마지막 순서로 실행되는 함수이며, 비디오 컨버팅 상태값을 성공으로 변경하는 메세지를 발행하는 역할을 합니다.

1. mediaconvert_ffmpeg 로 부터 전달받은 event 정보를 활용하여 dynamoDb(VIDEO_SETTINGS) 에 접근

• output 저장 정보 데이터 조회
• 썸네일 저장 정보 및 컨버팅 정보 조회

2. 위 내용을 조합하여 카프카 토픽으로 전송

@app.lambda_function()
def lambda_handler(event, context):
    try:
        # 이전 람다에서 전달받은 값을 꺼냅니다
        video_info = event['video_info']
        video_info['service_name'] = video_info['service_name'].replace('v2/', '')

        result = dbutils.get_video_setting_info(video_info)
        if result.get('statusCode') != 200:
            return error_response(result)

        # 업로드 최종 단계까지 완료되어 성공 메세지 발행
        kafka.send_media_video_topic(result['body'])

        ...(로직)

mediaconvert_error

1. MediaConvert는 작업이 생성되거나 상태가 변경될 때 이벤트를 자동으로 생성하고, 이러한 이벤트는 EventBridge에 전달합니다.

2. EventBridge 에서 아래와 같은 이벤트 패턴과 일치하는지 체크

{
 "source": ["aws.mediaconvert"],
 "detail-type": ["MediaConvert Job State Change"],
 "detail": {
  "status": ["ERROR"]
 }
}

3. EventBridge 가 패턴 캐치하여 mediaconvert_error Lambda 에서 카프카 토픽으로 메세지 전송

@app.on_cw_event({
                  "source": [
                    "aws.mediaconvert"
                  ],
                  "detail-type": [
                    "MediaConvert Job State Change"
                  ],
                  "detail": {
                    "status": [
                      "ERROR"
                    ]
                  }
                })
def lambda_handler(event):
    try:
        # eventbridge에서 컨버팅 실패 이벤트가 캐치되면 업로드 실패 메세지 발행
        kafka.send_media_video_topic(result['job_info'])

        ...(로직)

#5. Step Functions & MediaConvert

AWS Step Functions

stepfunction_trigger 람다에서 step functions 이 실행(execute)되면 아래 그래프에 나와있는 단계대로 각 람다 함수가 실행합니다.

• S3에 트리거되는 람다를 만들고 해당 람다에서 boto3 로 step functions 클라이언트 생성 후 실행

{
  "Comment": "비디오 인코딩 프로세스 output dev",
  "StartAt": "ffmpeg",
  "States": {
    "ffmpeg": {
      "Type": "Task",
      "Resource": "<ffmpeg-lambda-arn>",
      "InputPath": "$",
      "ResultPath": "$",
      "Next": "outputLambda"
    },
    "outputLambda": {
      "Type": "Task",
      "Resource": "<output-lambda-arn>",
      "InputPath": "$",
      "ResultPath": "$",
      "End": true
    }
  }
}

step functions 실행 그래프

AWS MediaConvert

1. 업로드된 동영상의 썸네일 이미지 추출
2. 업로드된 동영상이 60초가 넘을경우 60초로 조정

S3에서 트리거 되어 실행되는 람다 설정을 할때 버킷 속성에서 **S3 이벤트 알림** 을 사용하게 됩니다.

prefix 가 특정 경로를 이미 사용하고 있을경우 동일한 경로를 포함한 하위 경로를 가진 이벤트 알림을 생성 할 수 없습니다.

예를들어, /test 디렉토리에 이벤트 알림이 설정되어있을경우, /test/video 디렉토리의 이벤트 알림은 설정 할 수 없습니다.

#6. Consumer (Application) 설정

producer 역할은 media-lambda (python)가 수행합니다.

Consumer

• 비디오 인코딩 상태 변경(인코딩 시작, 완료, 실패)

Topic Name

${spring.config.activate.on-profile}.common.media-video

• {환경}.{메인도메인}.{서브도메인}

Consumer Group

video-encoding-status-${domain}-events-group

• {컨슈머 그룹이하는 작업에대한 정의}-{도메인}-events-group
• 미디어 업로드는 모든 서비스에서 presignedurl 을 처리하는 람다를 사용해 한곳에서 발행하는 토픽이지만, 컨슈머그룹은 서비스별로 나뉘어 져있기때문에 도메인도 포함합니다.

Task

• 동영상을 업로드하면 람다에(Producer)서 일련의 과정을 거치며 토픽으로 메세지를 전송하고, Consumer가 Consume한 페이로드의 상태값과 메타데이터를 포함한 정보를 DB에 업데이트 합니다.
• 공통으로 처리하는 하나의 람다에서 모든 토픽을 발행하기때문에 이를 구독하는 Consumer 는 담당 서비스에서 업로드된 파일만 받을 수 있게 분기처리 하기위한 필터링이 존재합니다.

# MediaConsumer.class

/** 
  각 Lambda 에서 발행하는 실패, 성공 메세지가 가지고 있는 action 정보
  - CREATED
  - COMPLETED
  - DATABASE_FAILED
  - ENCODING_FAILED
  - CREATE_JOB_FAILED

  통합토픽 어노테이션 (allAction = true) - 모든 action을 허용
**/
@KafkaListenerWithAction(topics = MEDIA_VIDEO_TOPIC, groupId = VIDEO_ENCODING_STATUS_GROUP, allAction = true)
    public void videoEncodingStatusMessage(Topic topic) {
        log.info("비디오 인코딩 상태변경 토픽 수신");

        // 서비스별 분기를 위한 필드가 추가된 객체로 다운캐스팅
        MediaTopic mediaTopic = (MediaTopic) topic;

        // 서비스별 분기 (커뮤니티 serviceId만 통과)
        if (VideoServiceType.isNoneAccessService(mediaTopic.getVideoServiceId())) {
            return;
        }

    ...(로직)
}

----------------------------------------------------------------------------

# TopicDeserializer.class

@Override
public Topic deserialize(String topic, byte[] data) {
    try {
        // MEDIA_VIDEO_TOPIC이 포함되어 있다면, 메세지를 MediaTopic 클래스로 역직렬화합니다
        if (topic.contains(MEDIA_VIDEO_TOPIC)) {
            return objectMapper.readValue(data, MediaTopic.class);
        }
        return objectMapper.readValue(data, Topic.class);

    ...(로직)
}

마무리

이번 포스트를 작성하며 그간의 과정을 간단히 돌아보면, 환경별 손쉬운 배포 방식 적용, 불필요한 함수 및 로직 제거, 그리고 역할 분리에 대한 개선이 미디어 고도화의 주요한 변경 사항이었습니다.

Chalice 배포 방식으로 배포를 간소화하고, 더 이상 사용하지 않는 람다 함수와 하드코딩된 더미 데이터는 모두 제거하여 실제 필요한 부분만 남도록 정리했습니다.

또한, MSA 아키텍처의 핵심 원칙에 맞춰 람다 간 호출을 제거하고, 각 람다를 독립적인 역할을 가진 Producer로 구성하여 Kafka로 메시지를 전송하는 방식으로 전환했습니다. 덕분에 각 서비스가 유연한 구조를 가지며 독립적으로 동작 가능하게 되었습니다.

마지막으로, Step Functions를 통해 람다 간 호출 순서를 제어하여, 안정적인 데이터 흐름을 보장하게 되었습니다 :)

이번 포스트에서의 내용은 구조 변경을 통해 유지보수를 용이하게 만드는 데 중점을 두었습니다.

이 글이 비슷한 문제를 겪고 있는 분에게 조금이라도 도움이 되었으면 좋겠습니다!

긴 글 읽어주셔서 감사합니다 ⭐