一元的な監査ログを管理するための MDS の構成¶

Confluent Platform 6.0 以降では、一元的な監査ログを使用して監査ログ構成を動的にアップデートできます。Confluent CLI を介して行われた変更は、MDS（Metadata Service）からすべての登録済みクラスターにプッシュされます。これにより、監査ログ構成の一元管理が可能になり、登録されているすべてのクラスターの監査ログイベントが同じ送信先 Kafka クラスターにパブリッシュされます。

MDS は、管理クライアントを使用して送信先 Kafka クラスターに接続し、API リクエストに応じて送信先トピックを検査、作成、および変更します。

一元的な監査ログを使用するには、いずれかの Kafka クラスターで Metadata Service（MDS）が実行されるように構成する必要があります。これにより、組織内の Kafka クラスターのリストを登録し、そのクラスターの監査ログ構成を一元管理するための API エンドポイントを提供できます。この監査ログ構成 API は、登録されているすべてのクラスターに、キャプチャされるイベントの種類と送信先を管理するルールをプッシュします。また、不足している送信先トピックを作成し、送信先トピックの保持時間ポリシーを監査ログ構成ポリシーと同期させます。

構成が変更されるまで、MDS は単独のクラスターであるという想定に基づいて動作します。つまり、独自の内部管理クライアントを使用して送信先トピックが構成されます。ブートストラップサーバーは指定されず、監査ログの送信先トピックは同じクラスター上に生成されます。デフォルトの動作は、必要になる構成が少ないため、開発環境での初期セットアップ用に便利です。

本稼働環境の設定では、すべての Kafka クラスターが監査ログを単一の一元的な送信先クラスターにパブリッシュする必要があります。各マネージド型クラスターの構成ファイルには、送信先クラスターの接続および認証情報を含めます。ただし、送信先トピックの自動作成は無効にし、confluent.security.event.router.config は未指定にしておく必要があります。

以下のセクションでは、一元的な監査ログを管理するために Kafka クラスターと MDS を構成する方法について説明します。

前提条件¶

すべての Kafka クラスターから従来の監査ログ構成を移行して、結合された JSON ポリシーに統合する必要があります。

重要

すべての Kafka クラスターを登録する前に、この前提条件が満たされている必要があります。
すべての Kafka クラスター（MDS クラスターを含む）をクラスターレジストリに登録します。

注釈

MDS クラスターの登録は、デフォルトでは実行されません。MDS クラスターは、他のクラスターよりも前に、クラスターレジストリに対して明示的に登録する必要があります。
すべての登録済みクラスターで、RBAC 用に同じ MDS が使用されるように構成します。
MDS クラスターは、管理クライアントを使用して登録済みクラスター（マネージド型クラスター）と通信します。すべての登録済みクラスターで認証トークンリスナーを公開し（listener.name.external.sasl.enabled.mechanisms=OAUTHBEARER など）、そのリスナーのポートをクラスターレジストリに登録することにより、MDS 管理クライアントからそれらのクラスターに接続できることを確認します。SASL_SSL を使用する場合は、クライアントのトラストストアの証明書による検証が可能な TLS キーのみを使用してください。
監査ログを受信するようにクラスターを構成します。そのクラスターで監査ログライターユーザー（auditlogwriter などの名前を使用）をセットアップし、送信先トピックへの書き込みを許可します。たとえば、プレフィックス confluent-audit-log-events が付いたトピックに対して、DeveloperWrite ロールを付与します。詳細については、「送信先クラスターに接続するための監査ログライターの構成」を参照してください。
すべての Kafka クラスターで、監査ログ構成を管理するユーザーまたはグループに、 AuditAdmin ロールを付与します。
注釈

アクセス許可を付与するために推奨される方法は、監査ログ管理者がいずれかの Confluent CLI confluent audit-log コマンドを実行することです。返されるエラーメッセージに、そのユーザーへの付与が推奨されるロールバインディングのリストが含まれています。
```
confluent login --url "http://mds.example.com:8090" # authenticate as user "alice"
confluent audit-log config describe
Error: 403 Forbidden
User:alice not permitted to DescribeConfigs on one or more clusters. Fix it:
confluent iam rbac role-binding create --role AuditAdmin --principal User:alice --kafka-cluster-id DBS26_qTQ-mT23p5opUK_g
confluent iam rbac role-binding create --role AuditAdmin --principal User:alice --kafka-cluster-id prz9a_-xqqlRgmekDoLw4U
```

必要な構成アップデート¶

一元的な監査ログを使用するには、次のことを行う必要があります。

MDS クラスターの構成
各登録済みクラスターの構成
監査ログの送信先クラスターの構成

MDS クラスターの構成¶

MDS クラスターの構成を次のようにアップデートする必要があります。

マネージド型クラスターへの監査ログ構成の伝播
送信先クラスターでの監査ログトピック管理の構成
送信先クラスターに接続するための監査ログライターの構成

マネージド型クラスターへの監査ログ構成の伝播¶

MDS は、監査ログ構成のアップデートをマネージド型クラスターに伝播します。MDS がこれを行うには、すべてのマネージド型クラスターの OAUTHBEARER エンドポイントがクラスターレジストリに登録されている必要があります（「MDS クラスターからのリスナー接続の構成」を参照）。

マネージド型クラスターのリスナーが SASL_SSL 用に構成されている場合は、MDS クラスターの /etc/kafka/server.properties ファイルで TLS クライアント設定を構成する必要があります。以下に例を示します。

# Configuration used when propagating changes to managed clusters with SASL_SSL endpoints
ssl.confluent.metadata.server.truststore.location=<path-to-truststore.jks>
ssl.confluent.metadata.server.truststore.password=<trust-store-password>

すべての Kafka クラスターで、監査ログ構成を管理するユーザーまたはグループに、AuditAdmin ロールを付与する必要があります（まだ付与していない場合）。

# Repeat for each Kafka cluster
confluent iam rbac role-binding create --role AuditAdmin --principal User:<audit-admin-user> --kafka-cluster-id <managed-kafka-cluster-id>
# Alternatively, repeat for each Kafka cluster, but specify the role binding for the group
confluent iam rbac role-binding create --role AuditAdmin --principal Group:<audit-admin-group> --kafka-cluster-id <managed-kafka-cluster-id>

送信先クラスターでの監査ログトピック管理の構成¶

MDS は、送信先クラスター上の監査ログトピックを管理し、不足しているトピックを作成して、それらのトピックの保持時間ポリシーと監査ログ構成ポリシーとの同期を維持します。MDS でこれを行うには、MDS が使用する管理クライアントが送信先クラスターに接続できるように構成する必要があります。

MDS クラスターの server.properties ファイルで管理クライアントを構成するときは、confluent.security.event.logger.destination.admin. プレフィックスを使用します。プレフィックス要件を除くと、この構成は他の管理クライアント構成と同様です。この接続は、このクラスターおよびすべてのマネージド型クラスターのプロデューサー構成と一致している必要があります。ここで指定されているプロパティの詳細については、「AdminClient の構成」および Kafka の「AdminClient」を参照してください。

SASL_SSL の構成

confluent.security.event.logger.destination.admin.bootstrap.servers=<logs1.example.com:9092,logs2.example.com:9092>
confluent.security.event.logger.destination.admin.security.protocol=SASL_SSL
confluent.security.event.logger.destination.admin.sasl.mechanism=PLAIN
confluent.security.event.logger.destination.admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="<audit-log-admin-username>" \
      password="<audit-log-admin-password>";
confluent.security.event.logger.destination.admin.ssl.truststore.location=<path-to-truststore.jks>
confluent.security.event.logger.destination.admin.ssl.truststore.password=<truststore-password>

注釈

管理クライアント構成に推奨されるセキュリティプロトコルは SASL_SSL です。送信先クラスターで別のセキュリティプロトコル（SASL_PLAINTEXT など）が使用されている場合は、それを confluent.security.event.logger.destination.admin.security.protocol の値に反映する必要があります。SASL_PLAINTEXT の場合、前述した TLS/SSL トラストストアのプロパティを指定する必要はありません。

confluent.security.event.logger.destination.admin.bootstrap.servers で指定したサーバーは、MDS に保存されている監査ログ構成 JSON オブジェクトで（この設定をオーバーライドして）送信先 bootstrap_servers が指定されるまで、フォールバックとしてのみ使用されます。

送信先クラスターに接続するための監査ログライターの構成¶

MDS では監査ログイベントも生成されるため、監査ログ送信先クラスターに接続できるようにプロデューサーを構成する必要があります。

# Configure the producer from the MDS cluster to export its own audit log events
confluent.security.event.logger.enable=true
confluent.security.event.logger.exporter.kafka.topic.create=false
confluent.security.event.logger.exporter.kafka.bootstrap.servers=<logs1.example.com:9092,logs2.example.com:9092>
confluent.security.event.logger.exporter.kafka.security.protocol=SASL_SSL
confluent.security.event.logger.exporter.kafka.sasl.mechanism=PLAIN
confluent.security.event.logger.exporter.kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="<audit-log-writer-username>" \
      password="<audit-log-writer-password>";
confluent.security.event.logger.exporter.kafka.ssl.truststore.location=<path-to-truststore.jks>
confluent.security.event.logger.exporter.kafka.ssl.truststore.password=<truststore-password>

注釈

confluent.security.event.logger.exporter.kafka.sasl.jaas.config で指定されたユーザーは、送信先クラスターでプレフィックス confluent-audit-log-events が付いたすべてのトピックに対する DeveloperWrite 書き込みロールを持つプリンシパルである必要があります。ただし、すべての登録済みクラスターに、ここで指定されているものと同じプリンシパルを使用する必要はありません。必要であれば、すべてに異なるプリンシパルを使用することもできますが、これらすべてのプリンシパルに適切なアクセス許可を付与する必要があります。

送信先トピックの作成とアップデートは、一元化された構成機能によって行われるため、すべての Kafka クラスター（MDS クラスターを含む）で confluent.security.event.logger.exporter.kafka.topic.create=false を設定して、自動作成メカニズムを無効にすることをお勧めします。

confluent.security.event.logger.exporter.kafka.bootstrap.servers で指定したサーバーは、MDS に保存されている監査ログ構成 JSON オブジェクトで（この設定をオーバーライドして）送信先 bootstrap_servers が指定されるまで、フォールバックとしてのみ使用されます。

注釈

Confluent Platform 5.4 または 5.5 の既存の監査ログ構成を Confluent Platform 6.0 に移行する場合、一元的な監査ログを使用する一方で、古い監査ログにも引き続きアクセスする必要があれば、その期間中は元のクラスターと送信先クラスターの両方から消費するか、クラスター内の元のトピックから送信先クラスタートピックに古いログをコピーする必要があります。

各登録済みクラスターの構成¶

各登録済みクラスターの構成を次のようにアップデートする必要があります。

クラスターレジストリで各マネージド型クラスターがセットアップされていることを確認します。
MDS クラスターからのリスナー接続の構成
監査ログの送信先クラスターにパブリッシュするためのプロデューサーの構成

MDS クラスターからのリスナー接続の構成¶

sasl.enabled.mechanisms で指定された OAUTHBEARER とともに SASL_SSL（またはSASL_PLAINTEXT）認証プロトコルが使用されるように、登録済みの各マネージド型クラスターのリスナーを構成する必要があります。この構成の詳細については、「ブローカーの構成」を参照してください。

監査ログの送信先クラスターにパブリッシュするためのプロデューサーの構成¶

一元的な監査ログ用に監査ログの送信先クラスターに接続してパブリッシュできるように、各登録済みクラスターのプロデューサーを構成する必要があります。以下の SASL_SSL 構成を server.properties ファイルに指定します。

# Publish to the audit log destination cluster
confluent.security.event.logger.enable=true
confluent.security.event.logger.exporter.kafka.topic.create=false
confluent.security.event.logger.exporter.kafka.bootstrap.servers=<logs1.example.com:9092,logs2.example.com:9092>
confluent.security.event.logger.exporter.kafka.security.protocol=SASL_SSL
confluent.security.event.logger.exporter.kafka.sasl.mechanism=PLAIN
confluent.security.event.logger.exporter.kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="<audit-log-writer-username>" \
    password="<audit-log-writer-password>";
confluent.security.event.logger.exporter.kafka.ssl.truststore.location=<path-to-truststore.jks>
confluent.security.event.logger.exporter.kafka.ssl.truststore.password=<truststore-password>

監査ログの送信先クラスターの構成¶

監査ログの送信先クラスターの構成を次のようにアップデートする必要があります。

監査ログトピックに対するユーザーアクセス許可の付与
MDS および登録済みマネージド型クラスターからのリスナー接続の構成
監査ログの送信先クラスターでのトピック用のデフォルトプロパティの構成

監査ログトピックに対するユーザーアクセス許可の付与¶

一元的な監査ログ用に監査ログの送信先クラスターを構成するには、以下のユーザーアクセス許可を付与する必要があります。

送信先クラスターにログを書き込むことができる 1 つまたは複数のプリンシパルを指定します。プリンシパルは、各 Kafka クラスターの server.properties ファイルの confluent.security.event.logger.exporter.kafka.sasl.jaas.config プロパティに含まれているものと同じである必要があります。
```
confluent iam rbac role-binding create --principal User:<audit-log-writer> --role DeveloperWrite --resource Topic:confluent-audit-log-events --prefix --kafka-cluster-id <kafka-destination-cluster-id>
```
送信先クラスターで監査ログトピックを管理できるプリンシパルを指定します。プリンシパルは、MDS クラスターの server.properties ファイルの confluent.security.event.logger.destination.admin.sasl.jaas.config プロパティに含まれているものと同じである必要があります。
```
confluent iam rbac role-binding create --principal User:<audit-log-admin> --role ResourceOwner --resource Topic:confluent-audit-log-events --prefix --kafka-cluster-id <kafka-destination-cluster-id>
```

MDS および登録済みマネージド型クラスターからのリスナー接続の構成¶

送信先サーバーには、すべてのクライアントからの接続を許可するリスナーが必要です。

MDS 管理クライアント
MDS 監査ログライターとマネージド型クラスターの監査ログライター

listeners=INTERNAL://:9092,BROKER://:9091,EXTERNAL://:9093
listener.security.protocol.map=INTERNAL:SASL_SSL,BROKER:SASL_SSL,EXTERNAL:SASL_SSL
...
listener.name.external.sasl.enabled.mechanisms=PLAIN
listener.name.external.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" password="admin-secret" \
  user_audit-log-admin-username="<audit-log-admin-password>" \
  user_audit-log-writer-username="<audit-log-writer-password>"
listener.name.external.ssl.key.password=<key-password>
listener.name.external.ssl.keystore.location=<path-to-keystore.jks>
listener.name.external.ssl.keystore.password=<key-store-password>
listener.name.external.ssl.truststore.location=<path-to-truststore.jks>
listener.name.external.ssl.truststore.password=<trust-store-password>

この構成の詳細については、「ブローカーの構成」を参照してください。

監査ログの送信先クラスターでのトピック用のデフォルトプロパティの構成¶

送信先クラスターで作成されたトピックには、送信先クラスターの server.properties ファイルで指定されているデフォルトのプロパティが使用されます（num.partitions、min.insync.replicas、default.replication.factor など）。

次のステップ¶

これで必要な構成の更新は完了です。次は、Confluent Platform CLI confluent audit-log config コマンドまたは Confluent MDS API を使用して、登録されているすべてのクラスターに対する監査ログを構成できます。トラブルシューティングに関する他のコマンドの例とガイダンスについては、「一元的な監査ログの MDS 構成に関するトラブルシューティング」を参照してください。

一元的な監査ログの MDS 構成に関するトラブルシューティング¶

ここでは、一元的な監査ログの管理に使用される MDS クラスター構成のトラブルシューティングに関する推奨事項をご紹介します。

監査ログの保持時間の設定を確認する¶

以下の手順は、保持ポリシーのみに影響します。若干の変更にとどめることをお勧めします。

Confluent CLI を使用して監査ログ構成を変更し、1 つ以上の送信先トピックの retention_ms をアップデートします。

# Capture the current configuration from MDS
confluent audit-log config describe > /tmp/audit-log-config.json

# View what was captured
cat /tmp/audit-log-config.json
{
  "destinations": {
    "bootstrap_servers": [
      "logs1.example.com:9092",
      "logs2.example.com:9092"
    ],
    "topics": {
      "confluent-audit-log-events": {
        "retention_ms": 7776000000
      }
    }
  },
  "default_topics": {
    "allowed": "confluent-audit-log-events",
    "denied": "confluent-audit-log-events"
  }
}

# Make a small change
vim /tmp/audit-log-config.json # e.g. - change 7776000000 to 7776000001

# Post the change back to MDS
confluent audit-log config update < /tmp/audit-log-config.json

トピックの retention.ms 設定が送信先クラスターの新しい値を反映していることを確認します。

cat /tmp/destination-cluster-admin-client.properties

bootstrap.servers=<logs1.example.com:9092>
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="<destination-cluster-admin-client-username>" \
  password="<destination-cluster-admin-client-password>";
ssl.endpoint.identification.algorithm=https
sasl.mechanism=PLAIN
truststore.location=<path-to-truststore.jks>
truststore.password=<trust-store-password>

kafka-topics --bootstrap-server <logs1.example.com:9092> \
--command-config /tmp/destination-cluster-admin-client.properties \
--describe --topic <confluent-audit-log-events>

Topic: confluent-audit-log-events  PartitionCount: 6 ReplicationFactor: 3
   Configs: min.insync.replicas=2,cleanup.policy=delete,retention.ms=7776000001
Topic: confluent-audit-log-events Partition: 0  Leader: 2 Replicas: 2,1,0 Isr: 2,0,1
Topic: confluent-audit-log-events Partition: 1  Leader: 1 Replicas: 1,0,2 Isr: 2,0,1
Topic: confluent-audit-log-events Partition: 2  Leader: 0 Replicas: 0,2,1 Isr: 2,0,1
Topic: confluent-audit-log-events Partition: 3  Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: confluent-audit-log-events Partition: 4  Leader: 1 Replicas: 1,2,0 Isr: 2,0,1
Topic: confluent-audit-log-events Partition: 5  Leader: 0 Replicas: 0,1,2 Isr: 2,0,1

送信先クラスターで直接、いずれかの送信先トピックの retention.ms 値を変更します。

kafka-topics --bootstrap-server <destination-cluster>:9092 \
--command-config /tmp/destination-cluster-admin-client.properties \
--alter --topic confluent-audit-log-events \
--config retention.ms=7776000002

監査ログ構成に新しい retention_ms 設定が示されていることを確認します。

confluent audit-log config describe

{
  "destinations": {
    "bootstrap_servers": [
      "logs1.example.com:9092",
      "logs2.example.com:9092"
    ],
    "topics": {
      "confluent-audit-log-events": {
        "retention_ms": 7776000002
      }
    }
  },
  "default_topics": {
    "allowed": "confluent-audit-log-events",
    "denied": "confluent-audit-log-events"
  }
}

このトラブルシューティング手順が適切に機能しない場合は（監査ログが適切に構成されていない場合に describe コマンドを実行しようとするとエラーが発生するなど）、MDS ブローカープロパティ（confluent.security.event.logger.destination.admin. というプレフィックスが付いたプロパティ）で接続と認証情報に誤りがないことを確認します。また、送信先クラスターで管理クライアントプリンシパルに十分なアクセス許可が付与されていることを確認します。最小のロールバインディングでは、送信先クラスターでプレフィックス confluent-audit-log-events が付いたトピックに対して ResourceOwner ロールを付与する必要があることに注意してください。最後に、送信先クラスターが到達可能であり、MDS クラスターのネットワークアドレスからの接続がリッスンされていることを確認します。

監査ログ構成が登録済みクラスターに同期されていることを確認する¶

次のコマンドを使用して、監査ログ構成が登録済みクラスターに同期されていることを確認します。

kafka-configs --bootstrap-server <mds-managed-cluster>:9092 \
--command-config /tmp/managed-cluster-admin-client.properties \
--entity-type brokers \
--entity-default \
--describe \
| grep confluent.security.event.router.config

confluent audit-log config describe を実行した場合と同じ JSON 監査ログ構成が表示されます。監査トピックが送信先クラスターで直接変更されている場合は、retention_ms の値が異なる可能性があります。その場合は、JSON のメタデータも異なる可能性があります。それ以外はすべて同じになるはずです。この検証が失敗した場合は、MDS クラスターレジストリについて以下を確認します。

クラスターで認証トークンリスナーが公開されていること（listener.name.<example>.sasl.enabled.mechanism=OAUTHBEARER）
クラスターの TLS キーが、MDS サーバーのトラストストアの証明書で検証できるものであること。

また、MDS に対して監査ログ API のアップデートリクエストを行ったときに、エラーステータスメッセージがないかどうかを調べてください。