# Kafka イベントソースマッピングのログ記録
<a name="esm-logging"></a>

Kafka イベントソースマッピングのシステムレベルのログ記録を設定すると、Lambda イベントポーラーが CloudWatch に送信するシステムログを有効化およびフィルタリングできます。

この機能は Kafka イベントソースマッピングでのみ使用可能で、[プロビジョニングモード](https://docs.aws.amazon.com/lambda/latest/dg/kafka-scaling-modes.html#kafka-provisioned-mode)を指定して実行できます。

ログ記録設定を含むイベントソースマッピングの場合、コンソールの **[Lambda]** > **[その他のリソース]** > **[イベントソースマッピング]** ページにある **[モニタリング]** タブで、事前定義済みのログクエリによって返されたシステムログを確認することもできます。

## ログ記録の仕組み
<a name="esm-logging-overview"></a>

イベントソースマッピングでログレベルのログ記録設定を設定すると、Lambda イベントポーラーは対応するログ (イベントソースマッピングのシステムログ) を送信します。

イベントソースマッピングでは、Lambda 関数が使用しているのと同じ[ログ送信先](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-logs.html#configuring-log-destinations)が再利用されます。Lambda 関数の実行ロールに必要なログ記録権限があることを確認します。

イベントソースマッピングには独自のログストリーム機能があり、ログストリーム名は日付とイベントソースマッピングの UUID で構成されます (`2020/01/01/12345678-1234-1234-1234-12345678901`)。

イベントソースマッピングのシステムログについては、次のログレベルから選択できます。


| ログレベル | 使用方法 | 
| --- | --- | 
| DEBUG (詳細度が最も高い) | イベントソース処理の進行状況に関する詳細情報 | 
| 情報 | イベントソースマッピングの通常の動作に関するメッセージ | 
| WARN (詳細度が最も低い) | 予期しない動作につながる可能性のある潜在的な警告およびエラーに関するメッセージ | 

ログレベルを選択すると、Lambda イベントポーラーはそのレベル以下のログを送信します。例えば、イベントソースマッピングのシステムログレベルを INFO に設定すると、イベントポーラーは DEBUG レベルでログ出力を送信しません。

## ログの設定
<a name="esm-logging-configure"></a>

Kafka イベントソースマッピングを作成または更新するときに、ログ記録設定を設定できます。

### ログ記録の設定 (コンソール)
<a name="esm-logging-console"></a>

**ログ記録を設定するには (コンソール）**

1. Lambda コンソールの [[関数]](https://console.aws.amazon.com/lambda/home#/functions) ページを開きます。

1. 関数の名前を選択します。

1. 次のいずれかを行います。
   + 新しい Kafka トリガーを追加するには、**[関数の概要]** で **[トリガーを追加]** を選択します。
   + 既存の Kafka トリガーを変更するには、トリガーを選択し、**[編集]** を選択します。

1. **[イベントポーラーの設定]** の **[プロビジョニングモード]** で、**[設定]** チェックボックスを有効にします。**[ログレベル]** 設定が表示されます。

1.  **[ログレベル]** ドロップダウンリストをクリックし、イベントソースマッピングのレベルを選択します。

1. 下部にある **[追加]** または **[保存]** を選択して、イベントソースマッピングを作成または更新します。

### ログの設定 (AWS CLI)
<a name="esm-logging-cli"></a>

#### ログ記録付きイベントソースマッピングの作成
<a name="esm-logging-cli-create"></a>

次の例では、ログ記録設定を含む 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"}'
```

#### ログ記録設定の更新
<a name="esm-logging-cli-update"></a>

`update-event-source-mapping` コマンドを使用すると、ログ記録設定を追加または変更できます。

```
aws lambda update-event-source-mapping \
  --uuid 12345678-1234-1234-1234-123456789012 \
  --logging-config '{"SystemLogLevel":"WARN"}'
```

## Kafka イベントソースマッピングのシステムログのレコード形式
<a name="esm-logging-record-format"></a>

Lambda イベントポーラーがログを送信する場合、各ログエントリには一般的なイベントソースマッピングメタデータとイベント固有のコンテンツが含まれています。

### WARN ログレコード
<a name="esm-logging-warn-record"></a>

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 ログレコード
<a name="esm-logging-info-record"></a>

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 ログレコード
<a name="esm-logging-debug-record"></a>

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
    }
}
```