# Kafka 이벤트 소스 매핑 로깅
Lambda 이벤트 폴러가 CloudWatch로 전송하는 시스템 로그를 활성화하고 필터링하도록 Kafka 이벤트 소스 매핑에 대한 시스템 수준 로깅을 구성할 수 있습니다.
이 기능은 [프로비저닝된 모드](https://docs.aws.amazon.com/lambda/latest/dg/kafka-scaling-modes.html#kafka-provisioned-mode)를 사용하는 Kafka 이벤트 소스 매핑에만 사용 가능합니다.
로깅 구성을 사용한 이벤트 소스 매핑의 경우, 지금 콘솔 **Lambda** > **추가 리소스** > **이벤트 소스 매핑** 페이지의 **모니터** 탭에서 사전 구축된 로그 쿼리의 시스템 로그를 확인할 수도 있습니다.
## 로깅 작동 방식
이벤트 소스 매핑에서 로그 수준으로 로깅 구성을 설정하면 Lambda 이벤트 폴러가 해당 로그(이벤트 소스 매핑 시스템 로그)를 전송합니다.
이벤트 소스 매핑은 Lambda 함수와 동일한 [로그 대상](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-logs.html#configuring-log-destinations)을 재사용합니다. Lambda 함수의 실행 역할에 필요한 로깅 권한이 있어야 합니다.
이벤트 소스 매핑에는 `2020/01/01/12345678-1234-1234-1234-12345678901`과 같은 로그 스트림 이름으로 날짜 및 이벤트 소스 매핑 UUID가 포함된 자체 로그 스트림이 있습니다.
이벤트 소스 매핑 시스템 로그의 경우 다음 로그 수준 중에서 선택할 수 있습니다.
| 로그 수준 | 사용법 |
| --- | --- |
| DEBUG(최대 세부 정보) | 이벤트 소스 처리 진행 상황에 대한 세부 정보 |
| INFO | 이벤트 소스 매핑 정상 작동에 대한 메시지 |
| WARN(최소 세부 정보) | 예기치 않은 동작으로 이어질 수 있는 잠재적 오류에 대한 메시지 |
로그 수준을 선택하면 Lambda 이벤트 폴러는 해당 수준 이하의 로그를 전송합니다. 예를 들어, 이벤트 소스 매핑 시스템 로그 수준을 INFO로 설정하면 이벤트 폴러는 DEBUG 수준에서 로그 출력을 전송하지 않습니다.
## 로깅 구성
Kafka 이벤트 소스 매핑을 생성 또는 업데이트할 때 로깅 구성을 설정할 수 있습니다.
### 로깅 구성(콘솔)
**로깅을 구성하려면(콘솔)**
1. Lambda 콘솔의 [함수 페이지](https://console.aws.amazon.com/lambda/home#/functions)를 엽니다.
1. 함수 이름을 선택합니다.
1. 다음 중 하나를 수행하세요.
+ 새 Kafka 트리거를 추가하려면 **함수 개요**에서 **트리거 추가**를 선택합니다.
+ 기존 Kafka 트리거를 수정하려면 트리거를 선택하고 **편집**을 선택합니다.
1. **이벤트 폴러 구성**의 **프로비저닝된 모드**에서 **구성** 확인란을 선택합니다. 그러면 **로그 수준** 설정이 나타납니다.
1. **로그 수준** 드롭다운 목록을 클릭하고 이벤트 소스 매핑의 수준을 선택합니다.
1. 맨 아래에서 **추가** 또는 **저장**을 선택하여 이벤트 소스 매핑을 생성하거나 업데이트합니다.
### 로깅 구성(AWS CLI)
#### 로깅을 사용하여 이벤트 소스 매핑 생성
다음 예제에서는 로깅 구성을 사용하여 Amazon MSK 이벤트 소스 매핑을 생성합니다.
```
aws lambda create-event-source-mapping \
--function-name my-kafka-function \
--topics AWSKafkaTopic \
--event-source-arn arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abc123 \
--starting-position LATEST \
--provisioned-poller-config MinimumPollers=1,MaximumPollers=3 \
--logging-config '{"SystemLogLevel":"DEBUG"}'
```
자체 관리형 Kafka의 경우 동일한 구문을 사용합니다.
```
aws lambda create-event-source-mapping \
--function-name my-kafka-function \
--topics AWSKafkaTopic \
--self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc.xyz.com:9092"]}}' \
--starting-position LATEST \
--provisioned-poller-config MinimumPollers=1,MaximumPollers=3 \
--logging-config '{"SystemLogLevel":"DEBUG"}'
```
#### 로깅 구성 업데이트
`update-event-source-mapping` 명령을 사용하여 로깅 구성을 추가 또는 수정합니다.
```
aws lambda update-event-source-mapping \
--uuid 12345678-1234-1234-1234-123456789012 \
--logging-config '{"SystemLogLevel":"WARN"}'
```
## Kafka 이벤트 소스 매핑 시스템 로그의 레코드 형식
Lambda 이벤트 폴러가 로그를 전송할 때 각 로그 항목에는 일반 이벤트 소스 매핑 메타데이터뿐 아니라 이벤트별 콘텐츠도 포함됩니다.
### WARN 로그 레코드
WARN 레코드에는 이벤트 폴러의 오류 또는 경고가 포함되며, 이 레코드는 이벤트가 발생할 때 출력됩니다. 예제:
```
{
"eventType": "ESM_PROCESSING_EVENT",
"timestamp": 1546347650000,
"resourceArn": "arn:aws:lambda:us-east-1:123456789012:event-source-mapping:12345678-1234-1234-1234-123456789012",
"eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/tests-cluster/87654321-4321-4321-4321-876543221-s1",
"eventProcessorId": "12345678-1234-1234-1234-123456789012/0",
"logLevel": "WARN",
"error": {
"errorMessage": "Timeout expired while fetching topic metadata",
"errorCode": "org.apache.kafka.common.errors.TimeoutException"
}
}
```
### INFO 로그 레코드
INFO 레코드에는 각 이벤트 폴러의 Kafka 소비자 클라이언트 구성이 포함되며, 이 레코드는 소비자가 빌드 또는 변경될 때 출력됩니다. 예제:
```
{
"eventType": "POLLER_STATUS_EVENT",
"timestamp": 1546347660000,
"resourceArn": "arn:aws:lambda:us-east-1:123456789012:event-source-mapping:12345678-1234-1234-1234-123456789012",
"eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/tests-cluster/87654321-4321-4321-4321-876543221-s1",
"eventProcessorId": "12345678-1234-1234-1234-123456789012/0",
"logLevel": "INFO",
"kafkaEventSourceConnection": {
"brokerEndpoints": "boot-abcd1234.c2.kafka-serverless.us-east-1.amazonaws.com:9098",
"consumerId": "12345678-1234-1234-1234-123456789012-0",
"topics": [
"test"
],
"consumerGroupId": "12345678-1234-1234-1234-123456789012",
"securityProtocol": "SASL_SSL",
"saslMechanism": "AWS_MSK_IAM",
"totalPartitionCount": 2,
"assignedPartitionCount": 2,
"partitionsAssignmentGeneration": 5,
"assignedPartitions": [
"test-0",
"test-1"
],
"networkConfig": {
"ipAddresses": [
"10.100.141.1"
],
"subnetCidrBlock": "10.100.128.0/20",
"securityGroups": [
"sg-abcdefabcdefabcdef"
]
}
}
}
```
### DEBUG 로그 레코드
DEBUG 로그에는 이벤트 소스 매핑 처리의 Kafka 오프셋 관련 정보가 포함되며, 오프셋 정보는 1분마다 출력됩니다. 예제:
```
{
"eventType": "KAFKA_STATUS_EVENT",
"timestamp": 1546347670000,
"resourceArn": "arn:aws:lambda:us-east-1:123456789012:event-source-mapping:12345678-1234-1234-1234-123456789012",
"eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/tests-cluster/87654321-4321-4321-4321-876543221-s1",
"eventProcessorId": "12345678-1234-1234-1234-123456789012/0",
"logLevel": "DEBUG",
"kafkaPartitionOffsets": {
"partition": "test-1",
"endOffset": 5004,
"consumedOffset": 5003,
"processedOffset": 5003,
"committedOffset": 5004
}
}
```