AWS

구독
다운로드 551
업데이트 2026. 10. 27.

4. AWS Aurora 로그 수집 환경 구성

SQS 대기열 생성

유형, 이름, 액세스 정책은 다음에 열거된 값으로 사용하세요. 나머지 옵션은 기본값을 사용하거나, 운영 환경에 맞게 선택하세요.

  • 유형: FIFO 선택

  • 이름: logpresso-aws-aurora-das.fifo 입력

  • 액세스 정책: 고급 선택 후 다음 액세스 정책을 참고하여 설정(REGION, ACCOUNT_ID를 환경에 맞게 수정)

    {
  "Version": "2012-10-17",
  "Id": "__default_policy_ID",
  "Statement": [
    {
      "Sid": "__owner_statement",
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::ACCOUNT_ID:root"
      },
      "Action": "SQS:*",
      "Resource": "arn:aws:sqs:REGION:ACCOUNT_ID:logpresso-aws-aurora-das.fifo"
    },
    {
      "Sid": "__receiver_statement",
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::ACCOUNT_ID:user/logpresso"
      },
      "Action": [
        "SQS:ChangeMessageVisibility",
        "SQS:DeleteMessage",
        "SQS:ReceiveMessage"
      ],
      "Resource": "arn:aws:sqs:REGION:ACCOUNT_ID:logpresso-aws-aurora-das.fifo"
    }
  ]
}
Caution
SQS 대기열 생성 후 Lambda 트리거를 설정하지 마세요. Lambda는 다음 단계에서 별도로 설정합니다.

SQS 대기열 생성 직후 화면은 아래와 같습니다.

SQS 대기열 생성

Lambda 함수 생성

Lambda 서비스 화면에서 함수를 생성하세요. 이 함수는 S3에 로그 파일이 저장될 때마다 호출되어 SQS 대기열에 로그를 적재합니다.

  1. 함수 생성 방법은 새로 작성을 선택하고 기본 정보 옵션을 다음과 같이 입력하고 함수 생성 버튼을 클릭하세요. 여기에 명시되지 않은 다른 옵션들은 기본값을 사용하거나, 운영 환경에 맞게 선택하세요.

    • 함수 이름: logpresso-aws-aurora-das-log-producer 입력

    • 런타임: Python 3.12

  2. 함수 생성 직후 화면은 아래와 같습니다.

    Lambda 함수 생성

    코드 탭 화면에서 lambda_function.py의 내용을 다음 코드로 대체해야 합니다. 이때 변수 sqs_queue_url에 할당된 문자열을 SQS 대기열 logpresso-aws-aurora-das.fifo의 URL로 수정하세요.

    import json
import boto3
import gzip

s3_client = boto3.client('s3')
sqs_client = boto3.client('sqs')
sqs_queue_url = 'https://sqs.REGION.amazonaws.com/ACCOUNT_ID/QUEUE'

def lambda_handler(event, context):
    # process_s3_object('BUCKET_NAME', 'FILE_NAME')
    for record in event['Records']:
         bucket_name = record['s3']['bucket']['name']
         object_key = record['s3']['object']['key']
         process_s3_object(bucket_name, object_key)

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

def process_s3_object(bucket_name, object_key):
    response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
    data = gzip.decompress(response['Body'].read()).decode('utf-8')

    lines = data.splitlines()
    filtered = []
    i = 0
    for line in lines:
        try:
            i = i + 1
            log_entry = json.loads(line)

            # NOTE: Add your own filtering condition below
            if 'type' in log_entry:
                filtered.append(json.dumps(log_entry))
        except: 
            print('error object key ' +  object_key + ' row #' + str(i))
            continue

    dup_id = 0
    if filtered:
        max_message_size = 128 * 1024  # 128KB
        current_message = []
        current_size = 0

        for log in filtered:
            log_size = len(log.encode('utf-8')) + 1 
            if current_size + log_size > max_message_size:
                send_message_to_sqs(sqs_queue_url, "\n".join(current_message), object_key, dup_id)
                dup_id = dup_id + 1
                current_message = [log]
                current_size = log_size
            else:
                current_message.append(log)
                current_size += log_size

        if current_message:
            send_message_to_sqs(sqs_queue_url, "\n".join(current_message), object_key, dup_id)
            dup_id = dup_id + 1



def send_message_to_sqs(queue_url, message_body, object_key, dup_id):
    message_group_id = 'logpresso'
    message_deduplication_id = object_key.split('/')[-1]

    response = sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody=message_body,
        MessageGroupId=message_group_id,
        MessageDeduplicationId=object_key + str(dup_id)
    )

    print(response)
    return response

가상 이벤트 테스트

Lambda 함수가 정상적으로 동작하는지 시험합니다. Test 버튼을 클릭하고 아래와 같이 가상 이벤트를 편집하세요.

Lambda 테스트 이벤트 구성

간접 호출 버튼을 클릭해서 테스트가 잘 실행되는지 확인하세요.

Lambda 인라인 정책

테스트가 잘 실행되었다면 Lambda 함수가 S3 서비스에 접근할 수 있도록 권한을 부여하세요.

  1. 구성 탭 화면에서 역할 이름을 클릭하세요. 화면이 IAM 서비스로 이동됩니다.

    역할 이름을 클릭해 IAM 서비스로 이동

  2. IAM 역할 화면에서 권한 추가 > 인라인 정책 생성 버튼을 클릭하세요.

    인라인 정책 생성 실행

  3. 정책 에디터 화면에 아래 JSON 파일을 복사하여 IAM 정책을 생성하세요(REGION, ACCOUNT_ID, BUCKET_NAME, QUEUE를 환경에 맞게 수정).

    {
	"Version": "2012-10-17",
	"Statement": [
		{
			"Sid": "Statement1",
			"Effect": "Allow",
			"Action": [
				"s3:GetObject",
				"sqs:SendMessage"
			],
			"Resource": [
				"arn:aws:s3:::BUCKET_NAME",
				"arn:aws:s3:::BUCKET_NAME/*",
				"arn:aws:sqs:REGION:ACCOUNT_ID:QUEUE.fifo"
			]
		}
	]
}

Lambda 함수 트리거 추가

이제 함수 트리거를 추가합니다. S3에 로그 파일이 저장될 때마다 lambda 함수가 호출되도록 트리거 조건을 추가하세요.

  1. 함수 개요 화면에서 트리거 추가 버튼을 클릭하세요.

  2. 트리거 추가 화면에서 S3 버킷을 선택하고 이벤트 유형으로 모든 객체 생성 이벤트를 선택하세요. 경고 문구가 나타나면 확인을 체크하고 추가 버튼을 클릭하세요.

    Lambda 함수 트리거 추가

동작 확인

이제 모니터링 탭 화면에서 S3에 로그가 쌓일 때마다 Lambda 함수가 호출되는지 확인하세요.

Lambda 함수 모니터링