DynamoDB 이벤트 소스에서 폐기된 레코드 보존하기

DynamoDB 이벤트 소스에서 폐기된 레코드 보존하기 (Retain discarded records for a DynamoDB event source in Lambda)

DynamoDB 이벤트 소스 매핑의 오류 처리는 오류가 함수 호출 전에 발생하는지, 함수 호출 중에 발생하는지에 따라 달라져요.

호출 전: Lambda 이벤트 소스 매핑이 스로틀링이나 다른 문제로 함수를 호출할 수 없으면, 레코드가 만료되거나(MaximumRecordAgeInSeconds) 이벤트 소스 매핑에 구성된 최대 보존 기간을 초과할 때까지 재시도해요.
호출 중: 함수가 호출됐지만 오류를 반환하면, 레코드가 만료되거나 최대 보존 기간(MaximumRecordAgeInSeconds)을 초과하거나 구성된 재시도 할당량(MaximumRetryAttempts)에 도달할 때까지 Lambda가 재시도해요. 함수 오류에는 BisectBatchOnFunctionError를 구성할 수도 있는데, 이 옵션은 실패한 배치를 두 개의 더 작은 배치로 나눠 잘못된 레코드를 격리하고 타임아웃을 피해요. 배치 분할은 재시도 할당량을 소모하지 않아요.

오류 처리 조치가 실패하면 Lambda는 레코드를 폐기하고 스트림에서 배치 처리를 계속해요. 기본 설정에서는 잘못된 레코드 하나가 영향받는 샤드(shard)의 처리를 최대 하루 동안 막을 수 있어요. 이를 피하려면 사용 사례에 맞는 합리적인 재시도 횟수와 최대 레코드 보존 기간으로 함수의 이벤트 소스 매핑을 구성하세요.

실패한 호출용 대상(destination) 구성하기

실패한 이벤트 소스 매핑 호출의 레코드를 보존하려면 함수의 이벤트 소스 매핑에 대상을 추가해요. 대상으로 보내는 각 레코드는 실패한 호출에 대한 메타데이터를 포함하는 JSON 문서예요. Amazon S3 대상의 경우 Lambda는 메타데이터와 함께 전체 호출 레코드도 보내요. 대상으로 Amazon SNS 토픽, Amazon SQS 큐, Amazon S3 버킷, 또는 Kafka를 구성할 수 있어요.

Amazon S3 대상에서는 Amazon S3 Event Notifications 기능으로 대상 S3 버킷에 객체가 업로드될 때 알림을 받을 수 있어요. 실패한 배치에 자동 처리를 수행하도록 S3 Event Notifications가 다른 Lambda 함수를 호출하게 구성할 수도 있어요.

실행 역할에는 대상에 대한 권한이 있어야 해요.

SQS 대상: sqs:SendMessage
SNS 대상: sns:Publish
S3 대상: s3:PutObject 및 s3:ListBucket
Kafka 대상: kafka-cluster:WriteData

Kafka 이벤트 소스 매핑의 실패 시 대상(on-failure destination)으로 Kafka 토픽을 구성할 수 있어요. Lambda가 재시도 횟수를 모두 소진한 뒤에도 레코드를 처리할 수 없거나 레코드가 최대 보존 기간을 초과하면, 실패한 레코드를 지정된 Kafka 토픽으로 보내 나중에 처리하게 해요. Kafka 토픽을 실패 시 대상으로 사용하기를 참고하세요.

S3 대상에 대해 자체 KMS 키로 암호화를 활성화했다면, 함수의 실행 역할에도 kms:GenerateDataKey를 호출할 권한이 있어야 해요. KMS 키와 S3 버킷 대상이 Lambda 함수·실행 역할과 다른 계정에 있다면, 실행 역할이 kms:GenerateDataKey를 허용하도록 KMS 키를 구성해요.

콘솔로 실패 시 대상을 구성하려면 다음 단계를 따라요.

Lambda 콘솔의 Functions 페이지를 열어요.
함수를 선택해요.
Function overview 아래에서 Add destination을 선택해요.
Source에서 Event source mapping invocation을 선택해요.
Event source mapping에서 이 함수에 구성된 이벤트 소스를 선택해요.
Condition에서 On failure를 선택해요. 이벤트 소스 매핑 호출에서는 이것이 유일하게 허용되는 조건이에요.
Destination type에서 Lambda가 호출 레코드를 보내는 대상 유형을 선택해요.
Destination에서 리소스를 선택해요.
Save를 선택해요.

AWS Command Line Interface(AWS CLI)로도 실패 시 대상을 구성할 수 있어요. 예를 들어 다음 create-event-source-mapping 명령은 SQS 실패 시 대상과 함께 이벤트 소스 매핑을 MyFunction에 추가해요.

aws lambda create-event-source-mapping \
--function-name "MyFunction" \
--event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525 \
--destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:us-east-1:123456789012:dest-queue"}}'

다음 update-event-source-mapping 명령은 이벤트 소스 매핑을 업데이트해서, 재시도 2회 후 또는 레코드가 1시간보다 오래됐을 때 실패한 호출 레코드를 SNS 대상으로 보내요.

aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--maximum-retry-attempts 2 \
--maximum-record-age-in-seconds 3600 \
--destination-config '{"OnFailure": {"Destination": "arn:aws:sns:us-east-1:123456789012:dest-topic"}}'

업데이트된 설정은 비동기적으로 적용되며, 프로세스가 완료될 때까지 출력에 반영되지 않아요. get-event-source-mapping 명령으로 현재 상태를 확인해요.

대상을 제거하려면 destination-config 파라미터의 인자로 빈 문자열을 제공해요.

aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--destination-config '{"OnFailure": {"Destination": ""}}'

Amazon S3 대상의 보안 모범 사례

대상으로 구성된 S3 버킷을 함수의 구성에서 대상 설정을 제거하지 않고 삭제하면 보안 위험이 생길 수 있어요. 다른 사용자가 대상 버킷의 이름을 알면 자기 AWS 계정에서 그 버킷을 다시 만들 수 있죠. 그러면 실패한 호출 레코드가 그 사용자의 버킷으로 보내져 함수의 데이터가 노출될 수 있어요.

경고 함수의 호출 레코드가 다른 AWS 계정의 S3 버킷으로 보내질 수 없게 하려면, 함수의 실행 역할에 s3:PutObject 권한을 계정의 버킷으로 제한하는 조건을 추가하세요.

다음 예시는 함수의 s3:PutObject 권한을 계정의 버킷으로 제한하는 IAM 정책이에요. 이 정책은 또한 Lambda가 S3 버킷을 대상으로 사용하는 데 필요한 s3:ListBucket 권한을 부여해요.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3BucketResourceAccountWrite",
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::*/*",
        "arn:aws:s3:::*"
      ],
      "Condition": {
        "StringEquals": {
          "s3:ResourceAccount": "111122223333"
        }
      }
    }
  ]
}

AWS Management Console 또는 AWS CLI로 함수의 실행 역할에 권한 정책을 추가하려면 다음 절차의 지침을 참고하세요.

콘솔 — 함수 실행 역할에 권한 정책 추가하기

Lambda 콘솔의 Functions 페이지를 열어요.
실행 역할을 수정하려는 Lambda 함수를 선택해요.
Configuration 탭에서 Permissions를 선택해요.
Execution role 탭에서 함수의 Role name을 선택해 역할의 IAM 콘솔 페이지를 열어요.
다음으로 역할에 권한 정책을 추가해요. Permissions policies 창에서 Add permissions를 선택하고 Create inline policy를 선택해요. Policy editor에서 JSON을 선택해요. 추가할 정책을 편집기에 붙여넣고(기존 JSON을 대체) Next를 선택해요. Policy details 아래에서 Policy name을 입력해요. Create policy를 선택해요.

CLI — 함수 실행 역할에 권한 정책 추가하기

필요한 권한이 있는 JSON 정책 문서를 만들어 로컬 디렉터리에 저장해요.
IAM put-role-policy CLI 명령으로 함수의 실행 역할에 권한을 추가해요. JSON 정책 문서를 저장한 디렉터리에서 다음 명령을 실행하고 역할 이름, 정책 이름, 정책 문서를 여러분 값으로 바꿔요.

aws iam put-role-policy \
--role-name my_lambda_role \
--policy-name LambdaS3DestinationPolicy \
--policy-document file://my_policy.json

다음 예시는 Lambda가 DynamoDB 스트림에 대해 SQS 또는 SNS 대상으로 보내는 호출 레코드를 보여줘요.

{
  "requestContext": {
    "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
    "functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction",
    "condition": "RetryAttemptsExhausted",
    "approximateInvokeCount": 1
  },
  "responseContext": {
    "statusCode": 200,
    "executedVersion": "$LATEST",
    "functionError": "Unhandled"
  },
  "version": "1.0",
  "timestamp": "2019-11-14T00:13:49.717Z",
  "DDBStreamBatchInfo": {
    "shardId": "shardId-00000001573689847184-864758bb",
    "startSequenceNumber": "800000000003126276362",
    "endSequenceNumber": "800000000003126276362",
    "approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z",
    "approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z",
    "batchSize": 1,
    "streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388"
  }
}

이 정보로 스트림에서 영향받은 레코드를 검색해 문제를 해결할 수 있어요. 실제 레코드는 포함되지 않으므로, 이 레코드를 처리해서 만료되기 전에 스트림에서 검색해야 해요.

예시 Amazon S3 호출 레코드

다음 예시는 Lambda가 DynamoDB 스트림에 대해 S3 버킷으로 보내는 호출 레코드를 보여줘요. SQS와 SNS 대상에 대한 이전 예시의 모든 필드 외에도, payload 필드에 원래 호출 레코드가 이스케이프된 JSON 문자열로 담겨 있어요.

{
  "requestContext": {
    "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
    "functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction",
    "condition": "RetryAttemptsExhausted",
    "approximateInvokeCount": 1
  },
  "responseContext": {
    "statusCode": 200,
    "executedVersion": "$LATEST",
    "functionError": "Unhandled"
  },
  "version": "1.0",
  "timestamp": "2019-11-14T00:13:49.717Z",
  "DDBStreamBatchInfo": {
    "shardId": "shardId-00000001573689847184-864758bb",
    "startSequenceNumber": "800000000003126276362",
    "endSequenceNumber": "800000000003126276362",
    "approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z",
    "approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z",
    "batchSize": 1,
    "streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388"
  },
  "payload": "<Whole Event>" // Only available in S3
}

호출 레코드를 담은 S3 객체는 다음 명명 규칙을 사용해요.

aws/lambda/<ESM-UUID>/<shardID>/YYYY/MM/DD/YYYY-MM-DDTHH.MM.SS-<Random UUID>

출처: AWS Lambda 개발자 안내서