AWS MediaConvert SDK 사용 방법 (3) - SNS 연결하기

AWS MediaConvert 주제의 마지막 포스팅 입니다. 지난 포스팅에서는 AWS MediaConvert 를 사용하기 위한 role 생성과 MediaConvert SDK 를 이용하여 Job 을 생성하는 방법을 알아 보았습니다.

2023.08.17 - [AWS] - AWS MediaConvert SDK 사용 방법 (1) - role 생성하기

2023.08.18 - [AWS] - AWS MediaConvert SDK 사용 방법 (2) - Job 생성 코드

이번 포스팅에서는 AWS MediaConvert 와 SNS를 연동하여 MediaConvert Job 수행 완료 이벤트를 받는 방법에 대해서 설명합니다.

AWS MediaConvert 모니터링에 대한 문서는 아래 링크에서 확인할 수 있습니다.

https://docs.aws.amazon.com/mediaconvert/latest/ug/setting-up-cloudwatch-event-rules.html

 

AWS SNS 로 부터 완료 메시지를 수신하기 위한 REST API 를 생성

REST API 를 통해서 MediaConvert Job 완료 메시지를 수신할 것이기 때문에 REST API URL 이 먼저 있어야 합니다.

package com.example.mediaconvert.controller;

import com.example.mediaconvert.dto.MediaConvertSnsMessageDto;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.awspring.cloud.sns.annotation.endpoint.NotificationMessageMapping;
import io.awspring.cloud.sns.annotation.endpoint.NotificationSubscriptionMapping;
import io.awspring.cloud.sns.annotation.endpoint.NotificationUnsubscribeConfirmationMapping;
import io.awspring.cloud.sns.annotation.handlers.NotificationMessage;
import io.awspring.cloud.sns.annotation.handlers.NotificationSubject;
import io.awspring.cloud.sns.handlers.NotificationStatus;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(path = "/media-convert/event")
@Slf4j
@RequiredArgsConstructor
public class MediaConvertEventController {
	//AWS SNS 구독 생성시에 확인 처리용 API 입니다.
	//AWS SNS 구독 요청시 확인을 위해서 전송되는 subscribe url 을 호출합니다.
	@NotificationSubscriptionMapping
	public void handleSubscriptionMessage(NotificationStatus status) {
		log.info("sns subscribe confirmation: {}", status);
		status.confirmSubscription();
	}

	//AWS SNS 메시지를 수신하기 위한 API 입니다.
	@NotificationMessageMapping
	public void handleNotificationMessage(@NotificationSubject String subject,
                                          @NotificationMessage String message) 
                                          throws JsonProcessingException {
		log.info("received sns message, subject={}, message={}", subject, message);
		ObjectMapper objectMapper = new ObjectMapper();
		//String JSON 메시지를 DTO 로 변환
		MediaConvertSnsMessageDto mediaConvertSnsMessageDto = 
        	objectMapper.readValue(message, MediaConvertSnsMessageDto.class);
		log.info("json object: {}", mediaConvertSnsMessageDto);
        
		//DTO 를 기반으로 수신된 AWS SNS 메시지를 이용하여 후처리 작업을 진행합니다.
	}
}

AWS SNS 메시지를 수신하기 위한 Controller 입니다.

SNS 메시지를 수신하기 위한 endpoint 는 /media-convert/event 입니다.

package global.voda.operator.dto;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;

import java.util.List;
import java.util.Map;

@JsonIgnoreProperties(ignoreUnknown = true)
public record MediaConvertSnsMessageDto(@JsonProperty("version") String version,
                                        @JsonProperty("id") String id,
                                        @JsonProperty("detail-type") String detailType,
                                        @JsonProperty("source") String source,
                                        @JsonProperty("account") String account,
                                        @JsonProperty("time") String time,
                                        @JsonProperty("region") String region,
                                        @JsonProperty("resources") List<String> resources,
                                        @JsonProperty("detail") Detail detail,
                                        @JsonProperty("SigningCertURL") String signingCertURL,
                                        @JsonProperty("UnsubscribeURL") String unsubscribeURL) {
    @JsonIgnoreProperties(ignoreUnknown = true)
    public record Detail(@JsonProperty("timestamp") Long timestamp,
                         @JsonProperty("accountId") String accountId,
                         @JsonProperty("queue") String queue,
                         @JsonProperty("jobId") String jobId,
                         @JsonProperty("status") String status,
                         @JsonProperty("errorCode") Integer errorCode,
                         @JsonProperty("errorMessage") String errorMessage,
                         @JsonProperty("userMetadata") Map<String, String> userMetadata,
                         @JsonProperty("outputGroupDetails") List<OutputGroupDetail> outputGroupDetails,
                         @JsonProperty("paddingInserted") Integer paddingInserted,
                         @JsonProperty("blackVideoDetected") Integer blackVideoDetected) {

        @JsonIgnoreProperties(ignoreUnknown = true)
        public record OutputGroupDetail(@JsonProperty("outputDetails") List<OutputDetail> outputDetails,
                                        @JsonProperty("type") String type) {

            @JsonIgnoreProperties(ignoreUnknown = true)
            public record OutputDetail(@JsonProperty("outputFilePaths") List<String> outputFilePaths,
                                       @JsonProperty("durationInMs") Integer duration,
                                       @JsonProperty("videoDetails") VideoDetail videoDetail) {
                @JsonIgnoreProperties(ignoreUnknown = true)
                public record VideoDetail(@JsonProperty("widthInPx") Integer width,
                                          @JsonProperty("heightInPx") Integer height,
                                          @JsonProperty("averageBitrate") Integer averageBitrate) {
                }
            }
        }
    }
}

수신된 SNS 메시지를 Deserialization 하기 위한 DTO 클래스 입니다.

 

 

AWS SNS topic / subscription 생성

첫번째로 MediaConvert 의 Job 완료 메시지를 수신하기 위해서는 우선 AWS SNS 토픽을 생성해야 합니다. 이때 생성하는 토픽 타입은 Standard 로 설정해야 합니다.

SNS topic 생성

생성된 MediaConvertJobCompleteErrorAlert 토픽 상세 페이지 하단에 [Create subscription] 을 통해서 subscription 을 생성합니다. 생성 전에 SNS 구독 생성 이벤트를 수신할 API 서버가 실행되고 있어야 합니다. 

Create subscription > Details

Enable raw message delivery 체크는 해제된 상태로 설정합니다.

구독 생성시에 controller 의 handleSubscriptionMessage 가 호출되며 아래와 같은 로그가 생성되는 것을 확인할 수 있습니다.

 

 

AWS EventBridge Rule 생성

MediaConvert 의 Job 상태에 따라서 SNS 메시지를 전달 시키기 위해서 AWS EventBridge Rules 에 상태 정의를 해야합니다. 

AWS EventBridge > Rules > Create rule 을 선택합니다. 

Define rule detail

Build event pattern > Event pattern

Event pattern 섹션에서는 Specific state(s) 항목에 COMPLETE, ERROR 를 선택하여 MediaConvert Job 이 정상적으로 완료된 경우와 에러가 발생한 경우에 대해서만 SNS 이벤트를 발생시키도록 설정하였습니다.

Select target(s)

현재 생성하고 있는 EventBridge rule 을 SNS topic 에 연결합니다.

Configure tags > Next 다음  Review and create 에서 설정 정보 확인 후 Create rule 을 선택하여 rule 을 생성합니다. 

 

AWS SNS topic 과 EventBridge rules 연결 확인

EventBridge rule 을 생성 후에 SNS topic 과 연결이 되었는지는 관리콘솔에서 AWS SNS > Topics > Access policy 탭에서 확인할 수 있습니다.

{
  "Version": "2008-10-17",
  "Id": "__default_policy_ID",
  "Statement": [
    {
      "Sid": "__default_statement_ID",
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": [
        "SNS:GetTopicAttributes",
        "SNS:SetTopicAttributes",
        "SNS:AddPermission",
        "SNS:RemovePermission",
        "SNS:DeleteTopic",
        "SNS:Subscribe",
        "SNS:ListSubscriptionsByTopic",
        "SNS:Publish"
      ],
      "Resource": "<AWS SNS topic arn>",
      "Condition": {
        "StringEquals": {
          "AWS:SourceOwner": "<account ID>"
        }
      }
    },
    {
      "Sid": "<AWSEvents_MediaConvertJobStateCompleteError_Sid>",
      "Effect": "Allow",
      "Principal": {
        "Service": "events.amazonaws.com"
      },
      "Action": "sns:Publish",
      "Resource": "<AWS SNS topic arn>"
    }
  ]
}

 

 

MediaConvert Job Complete Event 수신 확인

MediaConvert Job 을 생성 후 해당 Job 이 완료된 경우 SNS 이벤트를 수신하는 컨트롤러에는 아래와 같은 로그가 생성됨을 확인 할 수 있습니다.

received sns message, subject=, message={
    "version": "0",
    "id": "<id>",
    "detail-type": "MediaConvert Job State Change",
    "source": "aws.mediaconvert",
    "account": "<account Id>",
    "time": "2023-08-18T10:13:51Z",
    "region": "<region>",
    "resources": [
        "<MediaConvert job arn>"
    ],
    "detail": {
        "timestamp": 1692353631573,
        "accountId": "<account Id>",
        "queue": "<queue Default>",
        "jobId": "<Job Id>",
        "status": "COMPLETE",
        "userMetadata": {},
        "outputGroupDetails": [
            {
                "outputDetails": [
                    {
                        "outputFilePaths": [
                            "<S3 output object path>"
                        ],
                        "durationInMs": 15000,
                        "videoDetails": {
                            "widthInPx": 600,
                            "heightInPx": 120,
                            "averageBitrate": 36599,
                            "qvbrAvgQuality": 10,
                            "qvbrMinQuality": 10,
                            "qvbrMaxQuality": 10,
                            "qvbrMinQualityLocation": 0,
                            "qvbrMaxQualityLocation": 0
                        }
                    }
                ],
                "type": "FILE_GROUP"
            }
        ],
        "paddingInserted": 0,
        "blackVideoDetected": 0,
        "warnings": [
            {
                "code": 230001,
                "count": 1
            },
            {
                "code": 230005,
                "count": 1
            }
        ]
    }
}

 

3개의 포스팅에 걸쳐서 AWS MediaConvert SDK 를 사용하는 방법을 알아 보았습니다. 다뤄야 할 내용이 더 많지만 이 글을 읽으시는 분에게 도움이 되었기를 바랍니다.