1. Home
2. Platform
3. セキュリティ
4. 全般的なセキュリティ

クラスターレジストリ

Kafka クラスター管理者は Confluent Platform cluster registry を使用して、Kafka クラスターをメタデータサービス（MDS）に一元的に登録し、よりユーザーフレンドリーな RBAC ロールバインディングを可能にしたり、監査ログ記録を一元化したりできます。

クラスターレジストリを使用すると、MDS に Kafka クラスターを登録し、選択した名前で識別できます。通常の Kafka クラスター ID は、LRx92c9yQ+ws786HYosuBn などの Kafka GUID（Globally Unique Identifier）であるため、指定したクラスター名の方が使いやすく、覚えやすい可能性があります。Kafka クラスターを Confluent Platform cluster registry に登録した後は、新しいクラスター名を RBAC ロールバインディングで使用できます。たとえば、Connect クラスターのロールバインディングを作成する場合に、（Kafka クラスター ID と Connect クラスター ID の両方ではなく）登録済みのクラスター名を指定するだけで済みます。

注釈

"古い" GUID のクラスター ID も後方互換性のために残されており、引き続き使用できます。ただし、GUID のクラスター ID を使用すると、cluster registry に新しい名前を登録する意味がなくなります。

クラスターの名前を変更しても、そのクラスターを含むロールバインディングに変更はありません。プログラム上は、クラスター名は実際にはエイリアスであるためです。

一元的な監査ログでは、Confluent Metadata API リファレンス によって cluster registry を活用し、Confluent Platform cluster registry に登録されているすべての Kafka クラスターに、監査ログ構成のアップデートをプッシュできます。詳細については、「CLI による監査ログの構成」および「MDS API 監査ログ構成エンドポイント」を参照してください。

cluster registry と MDS

cluster registry は、MDS をホストしている Kafka クラスターで動的なブローカー構成として保管されます。複数のクラスターがあり、そのうちの 1 つが MDS ホストクラスターとして機能している場合は、クラスター間の関係がクラスターの登録方法にどのように影響するかを理解することが重要です。

たとえば、次の 3 つのクラスターがあるとします。

• クラスター A は MDS をホストしています。
• クラスター B および C は、クラスター A によって管理される MDS クラスターです。

クラスター B と C がクラスター A に依存する Connect、ksqlDB、または Schema Registry クラスターである場合は、最初にこれらのクラスターを起動し、クラスター ID を取得して、ロールバインディングを定義する前にクラスター A の cluster registry に登録する必要があります。

登録が必要な新しいクラスターを起動する場合、タスクの順序は次のようになります。

1. Kafka ブローカーを起動し、クラスター ID を取得します。
2. クラスターレジストリの MDS にクラスター ID を登録します。前述のとおり、MDS に依存するクラスターは、MDS ホストクラスターの cluster registry に登録する必要があります。
3. コンポーネントのロールバインディングを作成します （ksqlDB、Connect、Schema Registry など）。
4. クラスターを起動します。

前提条件

• MDS の構成で RBAC を有効にする必要があります。

• ブートストラップクラスター登録の場合は、ユーザーが MDS Kafka クラスターの super.user であるか、MDS Kafka クラスターの SystemAdmin ロールに属している必要があります。

  注釈

  "クラスターのブートストラップ" とは、クラスターを初めて起動するときに、スタートアップに必要な適切なホストや情報を既に含んでいる別のサーバー/クラスターにリンクまたは依存することを意味します。新しいクラスターを起動するたびに固有のデータをすべて入力する必要はありません。これにより、時間とリソースを節約しながら、信頼性の高い方法でクラスターを起動できます。

• 他のすべてのクラスター登録およびアップデートの場合は、Kafka クラスターの SystemAdmin ロールまたは ClusterAdmin ロールのいずれかが必要です。

ブートストラップクラスター登録と他のすべてのクラスター登録の違いは、cluster registry で使用されるゲートキーピングロジックです。新しいクラスターには、SystemAdmin ロールまたは ClusterAdmin ロールが割り当てられていません。このため、新しいクラスターのブートストラップ登録を行う方法には、次の 2 種類があります。

• MDS super.user または SystemAdmin がクラスターを登録する。
• MDS super.user または SystemAdmin が、（クラスター ID を使用して）クラスター上で別のユーザー（user:"alice"）へのロールバインディングを許可する。次に、そのユーザーがクラスターを登録する。

クラスターレジストリのルールと要件

• クラスター名に含めることのできる文字は、英語のアルファベット、数字、アンダースコア（_）、ダッシュ（-）のみです。
• クラスター名では大文字と小文字が区別されます。
• レジストリ内のクラスターごとに、少なくとも 1 つのホストエントリが（IP/DNS およびポート識別子を使用して）定義されている必要があります。

次のルールのいずれかに違反すると、cluster registry から競合エラーが返されます。

• （クラスター ID で定義されている）2 つの異なる物理クラスターに同じ名前を使用しない。
• 同じ物理クラスターに 2 つの異なる名前を使用しない。

登録済みのクラスターの名前を変更する必要がある場合は、「クラスターレジストリのエントリのアップデート」を参照してください。

Confluent Metadata API リファレンス または CLI を使用して、複数のクラスター登録を 1 つずつスクリプト化する場合、データの損失を防止するには、各呼び出しの応答を確認してから、次の呼び出しを行う必要があります。もう 1 つの方法として、（たとえば自動化されたマルチクラスターデプロイの一部として）API を使用し、複数の cluster registry 登録をスクリプト化することもできます。これにより、単一のトランザクションで一連のクラスターを処理できます。

クラスターの登録

クラスターを登録するには、curl コマンド または Confluent Platform CLI を使用します。

cluster registry に Kafka クラスターを登録するときは、以下の情報を指定する必要があります。

クラスター名

RBAC ロールバインディングおよび一元的な監査ログで使用される Kafka クラスターの新しい名前。

クラスター ID

クラスター ID を特定する必要がある場合は、「クラスター ID の表示」を参照してください。

ホスト名とポート番号

クラスター用に定義されたホストとポートには、RBAC トークン認証 をサポートするポートのみを含める必要があります。たとえば、Kafka クラスターでは、ブローカー間ポート、外部 Kerberos ポート、または mTLS ポートを使用しないでください。このことは、Confluent Metadata API リファレンス を使用する場合に最も重要です。構成のアップデートを既知の Kafka クラスターにプッシュするときにポート情報が活用されるためです。

ホストとポートで使用されるプロトコル

プロトコルは、Kafka クラスターの場合は SASL_SSL（本稼働環境以外の Kafka クラスターの場合は SASL_PLAINTEXT）で、Connect、ksqlDB、または Schema Registry クラスターの場合は HTTP または HTTPS である必要があります。

新しく登録されたクラスターでは、適切な RBAC ロール （ClusterAdmin および SystemAdmin）を必ず付与してください。これにより、他の構成のユーザーによるアクセスおよび使用が可能になります。また、一元的な監査ログ構成を管理するプリンシパルには、AuditAdmin ロールを必ず付与してください。登録済みクラスターでのロールの付与については、「登録済みクラスターのロールバインディングの構成」を参照してください。

重要

一元的な監査ログ に登録済みクラスターを使用する場合は、登録済みの各クラスターを MDS で管理 できるように構成する必要があります。一元的な監査ログ記録用に登録済みクラスターを構成する方法の詳細については、「一元的な監査ログを管理するための MDS の構成」を参照してください。クラスターをクラスターレジストリに登録する場合、登録するエンドポイントでは SASL_SSL または SASL_PLAINTEXT がサポートされ、OAUTHBEARER 認証が使用されている必要があります。

CLI を使用したクラスターの登録

CLI を使用して新しい Kafka クラスターを MDS に登録するには:

1. 実行中の MDS クラスターにログインします。

   # confluent login --url <protocol>://<host>:<port>
   confluent login --url http://localhost:8090
   

2. 新しい Kafka クラスターを登録します。

   # Substitute the value in <> with your information
   confluent cluster register \
     --cluster-name <exampleKafka> \
     --kafka-cluster-id <kafka-ID> \
     --hosts <10.6.6.6:9000,10.3.3.3:9003> \
     --protocol SASL_PLAINTEXT
   

新しい Kafka クラスターに関連付けられた新しい Connect クラスターを登録するには:

# Substitute the value in <> with your information
confluent cluster register \
  --cluster-name <exampleConnect> \
  --kafka-cluster-id <kafka-ID> \
  --connect-cluster-id <connect-ID> \
  --hosts <10.6.6.7:8443> \
  --protocol HTTP

curl を使用したクラスターの登録

curl を使用して新しい Kafka クラスターを MDS に登録するには:

1. 登録に使用する MDS 認証トークンを取得します。

   # Example
   TOKEN=$(curl -X GET "http://localhost:8090/security/1.0/authenticate" \
   -H "accept: application/json" \
   -u mds:mds | jq -r .auth_token)
   
   # Template
   TOKEN=$(curl -X GET "http://HOST:8090/security/1.0/authenticate" \
   -H "accept: application/json" \
   -u <ADMINUSER>:<PASSWORD> | jq -r .auth_token)
   

2. exampleConnect という名前のクラスターを登録します。

   curl -X POST "http://localhost:8090/security/1.0/registry/clusters" \
   -H "Content-Type: application/json" -H "accept: application/json" \
   -H "Authorization: Bearer $TOKEN" \
   -d '[{"clusterName":"exampleConnect","scope":{"clusters":{"kafka-cluster":"kafka-Id","connect-cluster":"connect-Id"}},"hosts":[{"host":"10.6.6.7","port":8443},{"host":"10.3.3.3","port":9003}]}]'
   

クラスターレジストリのエントリのアップデート

cluster registry の既存のクラスターをアップデートするには、「クラスターの登録」に指定されているものと同じ手順を使用します。既存のエントリは、指定したクラスター名に上書きされます。

クラスターレジストリのエントリの表示

クラスターで SystemAdmin、ClusterAdmin、または Operator のいずれかのロールが付与されているユーザーは、完全なクラスター情報を表示できます。

confluent cluster list
>       Name      |      Type       |  Kafka ID | Component ID |     Hosts                    |    Protocol
>+----------------+-----------------+-----------+--------------+------------------------------+----------------+
>  exampleConnect | connect-cluster | kafka-Id  | connect-Id   | 10.6.6.7:8443                | HTTP
>  exampleKafka   | kafka-cluster   | kafka-Id  |              | 10.6.6.6:9000,10.3.3.3:9003  | SASL_PLAINTEXT

クラスター上で他のロール（DeveloperRead など）が付与されているユーザーは、クラスター名とクラスター ID のみを表示できます。

confluent cluster list
>       Name      |      Type       |  Kafka ID | Component ID |     Hosts     |    Protocol
>+----------------+-----------------+-----------+--------------+---------------+----------------+
>  exampleConnect | connect-cluster | kafka-Id  | connect-Id   |               |
>  exampleKafka   | kafka-cluster   | kafka-Id  |              |               |

クラスターの登録解除

クラスターの登録と登録解除には、同じロールが必要です。詳細については、「前提条件」を参照してください。

cluster registry からクラスターを登録解除するには:

# Substitute the value in <> with your information
confluent cluster unregister --cluster-name <exampleConnect>
> Successfully unregistered the cluster exampleConnect from the Cluster Registry.

登録済みクラスターのロールバインディングの構成

クラスターを登録した後は、新しいクラスター名を RBAC ロールバインディングで使用できます。未登録のクラスターのロールバインディングの場合とは異なり、"古い" GUID のクラスター ID は含めないでください。

登録済みクラスターのロールバインディングの構成（CLI）

次の CLI 例は、登録済みクラスター exampleConnect 用に ClusterAdmin ロールをプリンシパル alice に付与する RBAC ロールバインディングの作成方法を示しています。

# Substitute the value in <> with your information
confluent iam rolebinding create --role ClusterAdmin \
  --principal User:<alice> \
  --cluster-name <exampleConnect>

次の例は、プリンシパル audit-admin に AuditAdmin ロールを付与する RBAC ロールバインディングの作成方法を示しています。このプリンシパルは、登録済みのマネージド型 Kafka クラスターの 一元的な監査ログ構成 を管理します。

# Substitute the value in <> with your information
confluent iam rolebinding create --role AuditAdmin \
   --principal User:<audit-admin> \
   --cluster-name <managed-kafka-cluster-id>

登録済みクラスターのロールバインディングの構成（curl）

次の例は、curl を使用して同じ RBAC ロールバインディングを作成する方法を示しています。

curl -X POST "http://localhost:8090/security/1.0/principals/User:alice/roles/ClusterAdmin" \
-H "accept: application/json" -H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-d '{"clusterName":"exampleConnect"}'

登録済みクラスターの保護

登録済クラスターでクライアントからの安全な接続を受け入れることができるようにセキュリティを有効にするには、server.properties ファイルで構成を指定します。

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

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>