Kinesis Data Streams 이벤트 소스에서 버려진 일괄 레코드 보존

Kinesis Data Streams 이벤트 소스에서 버려진 일괄 레코드 보존 (Retain discarded batch records for a Kinesis Data Streams event source in Lambda)

Kinesis 이벤트 소스 매핑의 오류 처리는 오류가 함수 호출 전에 발생하는지 호출 중에 발생하는지에 따라 달라집니다:

호출 전: Lambda 이벤트 소스 매핑이 제한(throttling) 또는 기타 문제로 함수를 호출하지 못하면 레코드가 만료되거나 이벤트 소스 매핑에 구성된 최대 수명(MaximumRecordAgeInSeconds)을 초과할 때까지 재시도합니다.
호출 중: 함수가 호출되었지만 오류를 반환하면 Lambda는 레코드가 만료되거나 최대 수명(MaximumRecordAgeInSeconds)을 초과하거나 구성된 재시도 할당량(MaximumRetryAttempts)에 도달할 때까지 재시도합니다. 함수 오류의 경우 실패한 일괄 처리를 두 개의 더 작은 일괄 처리로 분할해 잘못된 레코드를 격리하고 시간 초과를 피하는 BisectBatchOnFunctionError를 구성할 수도 있습니다. 일괄 처리 분할은 재시도 할당량을 소비하지 않습니다.

오류 처리 조치가 실패하면 Lambda는 레코드를 버리고 스트림의 일괄 처리를 계속 처리합니다. 기본 설정을 사용하면 잘못된 레코드 하나가 영향을 받는 샤드의 처리를 최대 1주일 동안 차단할 수 있습니다. 이를 피하려면 사용 사례에 맞는 합리적인 재시도 횟수와 최대 레코드 수명으로 함수의 이벤트 소스 매핑을 구성하세요.

출처: AWS Lambda 개발자 안내서

본문

실패한 호출에 대한 대상 구성

실패한 이벤트 소스 매핑 호출의 레코드를 보존하려면 함수의 이벤트 소스 매핑에 대상(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 주제를 Kafka 이벤트 소스 매핑의 on-failure 대상으로 구성할 수 있습니다. Lambda가 재시도 시도 후에도 레코드를 처리할 수 없거나 레코드가 최대 수명을 초과하면 Lambda는 실패한 레코드를 지정된 Kafka 주제로 보내 나중에 처리하게 합니다. Kafka 주제를 on-failure 대상으로 사용을 참조하세요.

S3 대상에 대해 자체 KMS 키로 암호화를 활성화한 경우 함수의 실행 역할에도 kms:GenerateDataKey를 호출할 권한이 있어야 합니다. KMS 키와 S3 버킷 대상이 Lambda 함수 및 실행 역할과 다른 계정에 있으면 KMS 키가 실행 역할을 신뢰하도록 구성해 kms:GenerateDataKey를 허용하세요.

콘솔을 사용해 on-failure 대상을 구성하려면 다음 단계를 따르세요:

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)를 사용해 on-failure 대상을 구성할 수도 있습니다. 예를 들어 다음 create-event-source-mapping 명령은 SQS on-failure 대상을 MyFunction에 추가합니다:

aws lambda create-event-source-mapping \
--function-name "MyFunction" \
--event-source-arn arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream \
--destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:us-east-1:123456789012:dest-queue"}}'

다음 update-event-source-mapping 명령은 이벤트 소스 매핑을 업데이트해 두 번의 재시도 후 또는 레코드가 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

예시 Amazon SNS 및 Amazon SQS 호출 레코드

다음 예제는 실패한 Kinesis 이벤트 소스 호출에 대해 Lambda가 SQS 대기열 또는 SNS 주제로 보내는 내용을 보여줍니다. Lambda는 이러한 대상 유형에 대해 메타데이터만 보내므로 streamArn, shardId, startSequenceNumber, endSequenceNumber 필드를 사용해 전체 원본 레코드를 얻습니다. KinesisBatchInfo 속성에 표시된 모든 필드는 항상 존재합니다.

{
    "requestContext": {
        "requestId": "c9b8fa9f-5a7f-xmpl-af9c-0c604cde93a5",
        "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:38:06.021Z",
    "KinesisBatchInfo": {
        "shardId": "shardId-000000000001",
        "startSequenceNumber": "49601189658422359378836298521827638475320189012309704722",
        "endSequenceNumber": "49601189658422359378836298522902373528957594348623495186",
        "approximateArrivalOfFirstRecord": "2019-11-14T00:38:04.835Z",
        "approximateArrivalOfLastRecord": "2019-11-14T00:38:05.580Z",
        "batchSize": 500,
        "streamArn": "arn:aws:kinesis:us-east-2:123456789012:stream/mystream"
    }
}

이 정보를 사용해 문제 해결을 위해 스트림에서 영향을 받은 레코드를 검색할 수 있습니다. 실제 레코드는 포함되지 않으므로 이 레코드를 처리하고 만료되어 손실되기 전에 스트림에서 검색해야 합니다.

예시 Amazon S3 호출 레코드

다음 예제는 실패한 Kinesis 이벤트 소스 호출에 대해 Lambda가 Amazon S3 버킷으로 보내는 내용을 보여줍니다. SQS 및 SNS 대상에 대한 이전 예제의 모든 필드 외에도 payload 필드에 원본 호출 레코드가 이스케이프된 JSON 문자열로 포함됩니다.

{
    "requestContext": {
        "requestId": "c9b8fa9f-5a7f-xmpl-af9c-0c604cde93a5",
        "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:38:06.021Z",
    "KinesisBatchInfo": {
        "shardId": "shardId-000000000001",
        "startSequenceNumber": "49601189658422359378836298521827638475320189012309704722",
        "endSequenceNumber": "49601189658422359378836298522902373528957594348623495186",
        "approximateArrivalOfFirstRecord": "2019-11-14T00:38:04.835Z",
        "approximateArrivalOfLastRecord": "2019-11-14T00:38:05.580Z",
        "batchSize": 500,
        "streamArn": "arn:aws:kinesis:us-east-2:123456789012:stream/mystream"
    },
    "payload": "" // Only available in S3
}

호출 레코드를 포함하는 S3 객체는 다음 명명 규칙을 사용합니다:

aws/lambda/<function-name>/<region>/<year>/<month>/<day>/<YYYY-MM-DDTHH.MM.SS>-<event-id>

더 알아보기 (Learn more)

Lambda 이벤트 소스 매핑 대상
Kinesis 이벤트 소스 오류 처리
Lambda 이벤트 소스 매핑 모니터링