Metadata Service (MDS) の構成

Confluent Platform Metadata Service(MDS)では、Confluent Platform インストールに関するさまざまなメタデータが管理されます。具体的に言えば、MDS には次のような役割があります。

  • インストールしたクラスターを追跡するための クラスターレジストリ をホストします。
  • クラスター間認可データ(RBAC一元的な ACL など)の記録システムとして機能し、トークンベースの認証に使用できます。
  • 複数のクラスターにわたる 監査ログ構成 を管理するための便利な方法となります。
  • データの認証に使用できます(クライアント認証はサポートされていないことに注意してください)。

MDS は、他の機能を提供する Kafka クラスター内で内部的にセットアップできます。アクセス許可の管理方法は、データベースにログインするユーザーのアクセス許可情報がそのデータベースに保管される形式と同じです。MDS を使用してユーザーデータを保存することもできます。MDS をホストする Kafka クラスターでは、各 Kafka ブローカーで MDS を構成し、その構成をノード間で同期する必要があります。

セキュリティ情報をクライアントデータから分離できるように、専用の Kafka クラスターに MDS をセットアップして、複数のワーカー用 Kafka クラスターにサービスを提供することもできます。ロールベースアクセス制御(RBAC) の場合、MDS は、単一の一元的な構成コンテキストを提供します。これにより、クラスター用のセットアップ後、各リソースのロールを個別に定義して割り当てるという複雑で時間のかかるタスクから管理者を解放できます。MDS は、ホストの Kafka クラスターおよび複数のセカンダリクラスター(Kafka、Connect、Schema Registry など)にわたって、RBAC、一元的な監査ログ、一元的な ACL、およびクラスターレジストリに関するルールを適用できます。このため、MDS をホストする単一の Kafka クラスターを使用して、複数の Kafka、Connect、Schema Registry および ksqlDB のセカンダリクラスターを管理および保護できます。

MDS は、デフォルトポート 8090 で HTTP コマンドをリッスンします。また、認可データのローカルキャッシュを管理し、このキャッシュは _confluent-metadata-auth という名前の内部 Kafka トピックに永続化されます。

MDS を Kafka ブローカーで実行すると、オプションで LDAP と統合し、権限借用に使用する認証と更新可能なベアラートークンを提供できます。権限借用は Confluent コンポーネントに制限されています。RBAC との LDAP 統合の構成については、「LDAP 認証の構成」を参照してください。

このトピックには、以下の構成タスクが含まれています。

前提条件

  • ご使用の環境に合わせて、セルフマネージド型の Confluent Platform を ダウンロード する必要があります。

  • Active Directory(LDAP サービス)を構成する必要があります。このトピックの構成は、Microsoft Active Directory(AD)に基づいています。これらの構成は、お使いの LDAP サービスに合わせてアップデートする必要があります。

    注釈

    ネスト化された LDAP グループはサポートされていません。

  • MDS を実行するブローカーは、ブローカー間通信用に独立したリスナーを使用して構成する必要があります。このリスナーにアクセスするには、ユーザーは認可用に ZooKeeper ベースの ACL一元的な ACL ではない)で構成されている必要があります。必要に応じてユーザーを super.users として構成することもできますが、ロールベースまたはグループベースのアクセスを使用したリソースへのアクセスに依存することはできません。ブローカーユーザーは、スーパーユーザーとして構成するか、 ACL を使用してアクセスを許可する必要があります。

    ちなみに

    ZooKeeper ベースの ACL から 一元的な ACL に移行した後、ブローカーリスナーまたはブローカー間の通信の問題を回避するために、super.users のリストにブローカーユーザーを含めて、アクセスを確認することもできます。ただし、移行後に通信が機能していることを確認した後、必ず super.users からブローカーユーザーを削除してください。

  • ブローカーは、RBAC 認可のメタデータが初期化されるまで、ブローカー間リスナーポートでリクエストを受け入れます。ただし、使用可能な LDAP メタデータを含め、必要なメタデータが初期化された後は、他のポートのリクエストのみが受け入れられます。ブローカーの初期化は、関連するすべてのメタデータが取得およびキャッシュされるまで完了しません。メタデータトピックのレプリケーション係数が 3(デフォルト)の MDS クラスターで複数のブローカーを起動する場合、ブローカーで初期化を完了するには、3 つ以上のブローカーを同時に起動する必要があります。この初期化にはタイムアウト/再試行の制限があることに注意してください。これは confluent.authorizer.init.timeout.ms で指定できます。詳細については、「Confluent Server Authorizer の構成」を参照してください。

  • MDS を使用して AD/LDAP と統合される REST Proxy サービスでは、認可の決定用のユーザープリンシパルとしてユーザーログイン名が使用されます。また、デフォルトではこのプリンシパルが、SASL/GSSAPI(Kerberos)を使用して認証を行うユーザー用にブローカーによって使用されます。ブローカー構成で principal.builder.class または sasl.kerberos.principal.to.local.rules をオーバーライドして別のプリンシパルが作成される場合は、ブローカーで使用されるユーザープリンシパルが、他の Confluent Platform コンポーネントで使用されるプリンシパルと異なることがあります。この場合は、ブローカーリソース用にカスタマイズされたプリンシパルの ACL とロールバインディングを構成する必要があります。

重要

グループロールバインディングに指定するユーザー ID では大文字と小文字が区別され、AD レコードでの指定と一致する必要があります。スーパーユーザーとしてログインするときは、ログイン ID の大文字と小文字も区別され、ロールバインディング内のユーザー ID の指定と一致する必要があることにも注意してください。

PEM キーペアの作成

トークンサービスで使用する PEM(Privacy-Enhanced Mail)キーペアを作成する必要があります。このキーペアは 次のセクションserver.properties ファイルに追加されます。

  1. PEM キーを作成します。

    注釈

    PEM キーの長さは、使用する暗号化方式(AES-256 または RSA-4096)によって異なります。ビット数も、使用する方式のニーズや要件に基づいて選択する必要があります。

    次の例では、2048 ビットの RSA プライベートキーを作成し、<path-to-tokenKeypair.pem> という名前のフォルダーにキーを保管する方法を示します(実際のセットアップに合わせて <> 内のパスを置き換えてください)。

    mkdir <path-to-tokenKeypair.pem> && openssl genrsa -out <path-to-tokenKeypair.pem> 2048
    
  2. パブリックキーを抽出します。

    openssl rsa -in <path-to-tokenKeypair.pem> -outform PEM -pubout -out <path-to-public.pem>
    

注釈

PEM キーファイルを作成するには、OpenSSL のみを使用します。Keytool は使用しないでください。これは有効なオプションではなく、起動時にエラーが発生します。

MDS とロールバインディングをホストするためのプライマリ Kafka クラスターの構成

ちなみに

Confluent CLI confluent secret コマンドを使用すると、パスワードやその他の構成データを安全に保管できます。詳細については、「シークレット管理」を参照してください。

注釈

クラスター内で使用可能なすべてのブローカーで、ここで説明する MDS 構成手順を実行し、完了する必要があります。ここに示す構成例は、スタンドアロンブローカーを備えたクラスター用です。複数ブローカー(ブローカー間)構成の例については、「Kafka ブローカーでの mTLS 認証と RBAC の構成」を参照してください。

以下のセクションでは、MDS をホストするためにプライマリ Kafka クラスターを構成する方法について説明します。

トークン認証の構成で問題が発生した場合は、「トークンの認証」を参照してください。

重要

MDS 構成の すべて のセクションを完了したら、 Confluent Platform を起動します

Confluent Server Authorizer の構成

オーソライザーは操作を認可するために Apache Kafka® で使用されるサーバープラグインです。具体的には、オーソライザーで、プリンシパルとアクセス対象のリソースを基に、操作を認可するかどうかを制御します。

Confluent Server Authorizer では、LDAP ベースのユーザーおよびグループを対象にした独自のロールベースアクセス制御(RBAC)認可と、ACL の設定がサポートされています。Confluent Server Authorizer では、プラグ可能な認可およびグループプロバイダーもサポートされており、実行時に ACL と RBAC プロバイダーを読み込むことができます。

Confluent Authorizer 用に以下の構成を Kafka プロパティファイル(/etc/kafka/server.properties)に追加します。かっこ(&lt;&gt;)で囲まれた内容は、環境に応じてカスタマイズする必要があります。

1############################# Confluent Server Authorizer Settings  #############################
2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
3
4# Specify a list of Access Rule Providers to retain ACLs that have already been enabled and to enable RBAC
5confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT
6
7# Specify when bootstrapping Confluent Platform and to assign a SystemAdmin
8super.users=<User:admin;User:mds>

重要

MDS 構成の すべて のセクションを完了したら、 Confluent Platform を起動します

以下のセクションでは、Confluent Server Authorizer 設定を指定するために使用される構成プロパティについて説明します。

authorizer.class.name

使用するオーソライザーを定義し、Confluent 独自の機能のエントリポイントとしての役割を担います。この MDS 構成では、Confluent Server Authorizer が有効になります。

confluent.authorizer.access.rule.providers

有効になっているアクセスルールプロバイダーのリスト。Kafka Authorizer はアクセスルールプロバイダーを使用して、アクセスの決定にどのルールを使用する必要があるかを判断します。サポートされているアクセスルールプロバイダーは次のとおりです。

  • CONFLUENT: Kafka に保管されている RBAC ロールバインディングと 一元的な ACL を使用して、アクセスルールオブジェクトのセットを生成します。
  • ZK_ACL(デフォルト) : ZooKeeper に保管されている ACL を使用して、アクセスルールオブジェクトのセットを生成します。

super.users

MDS 構成では、super.users 属性を割り当てて、Metadata Service (MDS) クラスター内のすべてのリソースへのフルアクセス権限を持つユーザーを定義する必要があります。super.users の主な用途は、Confluent Platform をブートストラップし、SystemAdmin を割り当てることです。MDS クラスターで、super.user は他のすべてのクラスターのロールバインディングを作成できます。super.user によって付与されるアクセス許可は、super.user 属性が指定されているブローカーにのみ適用され、他のブローカー、クラスター、Confluent Platform コンポーネントには適用されません。super.users として定義されたユーザーには認可が適用されないため、この属性は限られた数のユーザー(ブートストラップを担当する 1 〜 2 人のユーザーなど)のみに指定することを強くお勧めします。

注釈

Confluent Platform のブートストラップでは、クラスターを初めて起動するときに、起動に必要な適切なホスト情報やその他の構成情報が既に含まれている別のサーバー/クラスターにリンクまたは依存します。新しいクラスターを起動するたびに、すべての起動属性を入力する必要がありません。ブートストラップを使用すると、信頼性の高い方法でクラスターを起動すると同時に、時間とリソースを節約できます。

LDAP ID プロバイダーの構成

以下で、MDS の基本的な LDAP 構成について説明します。この構成は、MDS に対して LDAP ユーザーとグループを識別するための LDAP コンテキストを示しています。ネスト化された LDAP グループはサポートされていない点にご注意ください。

LDAP 構成で指定する必要があるため、以下の情報を準備してください。

  • ホスト(LDAP サーバーの URL、LDAPSERVER.EXAMPLE.COM など)、ポート(389 など)、その他のセキュリティメカニズム(TLS/SSL など)
  • LDAP ユーザーの完全な DN(識別名)
  • 複雑な LDAP ディレクトリツリーがある場合は、MDS が LDAP 検索結果を絞り込むことができるように、検索フィルターの提供を検討してください

注釈

LDAP 構成後(ただし MDS を構成する前に)LDAP サーバーに接続して照会し、LDAP 接続情報を確認することをお勧めします。これを行うには、LDAP ツール(JXplorer など)の使用をお勧めします。

注釈

LDAP コールバックハンドラー を追加して Kafka クライアントの LDAP 認証を有効にした場合(この構成には示されていません):

  • LDAP サーバーで単純バインドがサポートされていない場合にのみ ldap.user.password.attribute を指定します。
  • このプロパティ(io.confluent.security.auth.provider.ldap.LdapAuthenticateCallbackHandler)を定義すると、LDAP がユーザー検索を実行してパスワードを Kafka に返し、Kafka がパスワードチェックを実行します。
  • LDAP サーバーからはユーザーのハッシュ化されたパスワードが返されるため、ユーザーのプロパティファイルでもハッシュ化されたパスワードを使用していない限り、Kafka はユーザーを認証できません。

ID プロバイダー(LDAP)用に以下の構成を Kafka プロパティファイル(/etc/kafka/server.properties )に追加します。かっこ(&lt;&gt;)で囲まれた内容は、環境に応じてカスタマイズする必要があります。

 1############################# Identity Provider Settings (LDAP) #############################
 2# Search groups for group-based authorization.
 3ldap.group.name.attribute=<sAMAccountName>
 4ldap.group.object.class=group
 5ldap.group.member.attribute=member
 6ldap.group.member.attribute.pattern=CN=(.*),DC=rbac,DC=confluent,DC=io
 7ldap.group.search.base=CN=Users,DC=rbac,DC=confluent,DC=io
 8#Limit the scope of searches to subtrees off of base
 9ldap.user.search.scope=2
10#Enable filters to limit search to only those groups needed
11ldap.group.search.filter=(|(CN=<specific group>)(CN=<specific group>))
12
13# Kafka authenticates to the directory service with the bind user.
14ldap.java.naming.provider.url=ldap://<hostname>:389
15ldap.java.naming.security.authentication=simple
16ldap.java.naming.security.credentials=<password>
17ldap.java.naming.security.principal=<mds-user-DN>
18
19# Locate users. Make sure that these attributes and object classes match what is in your directory service.
20ldap.user.name.attribute=<sAMAccountName>
21ldap.user.object.class=user
22ldap.user.search.base=<user-search-base-DN>

重要

MDS 構成の すべて のセクションを完了したら、 Confluent Platform を起動します

以下のセクションでは、ユーザーおよびグループベースの認可に使用されるベースライン LDAP 構成オプションについて詳しく説明します。LDAP 構成の詳細については、「MDS の LDAP グループベース認可の構成」および「LDAP 認証の構成」を参照してください。

ldap.group.name.attribute

LDAP 検索を使用して取得したグループエントリ内のグループ名が含まれます。ldap.group.name.attribute.pattern を構成することにより、ACL で使用されるグループ名をこの属性から抽出するための正規表現パターンを指定できます。<sAMAccountName> は、さまざまなバージョンの Windows OS を実行しているクライアントとサーバーをサポートするために使用される、Microsoft Active Directory に固有のログイン名です。LDAP 構成が異なる場合には、使用する値を変更します。この構成オプションのデフォルトは cn (common name: 一般名)です。

ldap.group.object.class

ディレクトリサービスでユーザーを定義する LDAP オブジェクトクラス値を指定します。グループベースの認可を行うためにグループを検索するには、group を指定します。group には多数の用途がありますが、本質的にはゼロ個以上のデジタル ID のリストです。

ldap.group.member.attribute

LDAP 検索を使用して取得したグループエントリ内でグループのメンバーが格納されている属性の名前です。デフォルト値は member です。ldap.group.member.attribute.pattern を構成することにより、ユーザープリンシパルをこの属性から抽出するための正規表現(regex)パターンを指定できます。

ldap.group.member.attribute.pattern

ldap.group.member.attribute を使用して指定された LDAP 属性からグループメンバーエントリを取得し、そのエントリからグループメンバーのユーザープリンシパルを抽出する Java 正規表現パターン。デフォルトでは、属性の完全な値が使用されます。

ldap.group.search.base

この属性は、指定された値を使用して検索ベースをグループベースの検索に制限するように LDAP に指定します。デフォルトは ou=groups です。

ldap.group.search.scope

グループベース検索用の LDAP 検索スコープ。値 2 を指定すると、指定されたベースからすべてのサブツリーが含まれるように検索が開始されます。この指定を行うと、多くの場合は検索対象範囲が広すぎて、タイムアウトが発生することがあります。このオプションを指定した場合は、ldap.group.search.filter も指定する必要があります。デフォルト値は 1 です。

ldap.group.search.filter

グループベース検索用の LDAP 検索フィルターです。フィルターを有効にして、検索対象を必要なグループのみに限定します。検索に使用するすべてのグループをリストすることをお勧めします。大規模な組織の LDAP ツリーはきわめて大きくなる傾向にあります。すべてを検索しようとするとタイムアウトになるため、通常はこの処理が必要です。たとえば、ldap.group.search.scope に、すべてのサブツリーを検索するためのスコープ 2 を追加した後、検索に含めるグループを絞り込む必要があります。この検索フィルターには、任意の数のグループを含めることができます。

以下のセクションでは、Kafka でバインドユーザーによるディレクトリサービスへの認証に使用されるベースライン LDAP 構成オプションについて、詳細に説明します。

ldap.java.naming.provider.url

このオプションは、LDAP サーバーへの接続に使用する URL を定義します。デフォルトのホスト名は localhost、デフォルトのポートは 389 です。このオプションは MDS 構成に指定する必要があります。

ldap.java.naming.security.authentication

LDAP サーバーでパスワード認証が有効になっている場合は、ブローカーが単純認証を使用して LDAP サーバーで認証できるようにユーザープリンシパルとパスワードを構成できます。LDAP サーバーでの認証を行わない場合は、none を指定します。MDS 構成を稼働させるための推奨値は、simple (セキュリティを提供しない PLAINTEXT 認証プロトコル)です。本稼働環境のインスタンスの場合は、LDAP サーバーでサポートされている、より安全な SASL メソッド( SASL_GSSAPI など)を指定する必要があります。

ldap.java.naming.security.credentials

LDAP 検索を実行するプリンシパルのセキュリティ認証情報(パスワード)を指定します。

ldap.java.naming.security.principal

LDAP 検索を実行する LDAP ユーザーの識別名であるセキュリティプリンシパルを指定します。この構成では、DN を使用して MDS ユーザーを指定します。DN とは LDAP 識別名であり、コンマで接続された相対識別名(RDN: Relative Distinguished Name)のシーケンスです。

以下のセクションでは、ユーザーの場所の特定に使用するオプションについて詳しく説明します。これらの属性とオブジェクトクラスが、お使いのディレクトリサービスに存在するものと一致することを確認してください。

ldap.user.name.attribute

この属性は、LDAP 検索を使用して取得したユーザーエントリ内のユーザープリンシパルを識別します。ldap.user.name.attribute.pattern を構成することにより、ユーザープリンシパルをこの属性から抽出するための正規表現パターンを指定できます。 <sAMAccountName> は、さまざまなバージョンの Windows OS を実行しているクライアントとサーバーをサポートするために使用される、Active Directory に固有のログイン名です。LDAP 構成が異なる場合には、この構成を変更します。このオプションのデフォルトは cn (common name: 一般名)です。

ldap.user.object.class

ディレクトリサービスでユーザーを定義する LDAP オブジェクトクラス値を指定します。ユーザーベースの認可のための検索を行うには、user を指定します。

ldap.user.search.base

ユーザーベース検索用の LDAP 検索ベースを指定するために使用します。デフォルト値は ou=users です。

MDS サーバーの構成

このセクションでは、MDS 用のキーペアを作成し、ブローカーで MDS を構成するためのオプションについて説明します。キーファイルへのパスは、セットアップに応じてアップデートする必要があります。

注釈

Confluent Server は、サーブレットアプリケーションの REST Proxy と MDS をサポートします。

  • MDS が構成されている場合、MDS のリスナー構成は REST Proxy のリスナー構成より優先されます。
  • MDS が無効になっている場合は、REST Proxy リスナーの構成が優先されます。

MDS サーバーの設定用に以下の構成を Kafka プロパティファイル(/etc/kafka/server.properties )に追加します。かっこ(&lt;&gt;)で囲まれた内容は、環境に応じてカスタマイズする必要があります。

1############################# MDS Server Settings #############################
2# Bind Metadata Service HTTP service to port 8090.
3confluent.metadata.server.listeners=http://0.0.0.0:8090
4# The key to encrypt the token (when you issue you a token)
5confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
6# Supported authentication methods
7confluent.metadata.server.authentication.method=BEARER

重要

MDS 構成のすべてのセクションを完了したら、 Confluent Platform を起動 します。

以下のセクションでは、MDS サーバーの設定を指定するために使用される構成オプションについて説明します。

confluent.metadata.server.listeners

HTTP(または HTTPS)サービスをポートにバインドするために使用します。ここでは、デフォルト値(8090)が指定されています。

confluent.metadata.server.token.key.path

トークンの署名と検証に使用される、 PEM エンコードされたパブリックキー/プライベートキーのペア の場所です。トークンサービスでサポートされるのは RS256 署名のみであるため、キーペアは RSA アルゴリズムを使用して生成する必要があります。

confluent.metadata.server.authentication.method

MDS 側でのトークン認証のサポートを指定するために使用します。構成でベアラートークン認証が有効になっていることを示すには、BEARER を使用します。

トークンリスナーの構成

MDS は、ユーザー名およびパスワードと引き換えにトークンを提供できます。また、MDS は、ユーザーまたはクライアントからトークンを受け取り、そのトークンによってプリンシパルを認証することもできます(初回以降はこれを使用して認証を継続できます)。

トークンは、Kafka で構成された OAUTHBEARER リスナーに対する認証に使用されます。このセクションでは、MDS トークンサービスの有効化および構成を行う方法を示します。

  • コンポーネントとブローカーの間では、必ず同じパブリックキーを使用してください。
  • OAUTH に接続する MDS サービスまたはコンポーネントごとに一意のキーを使用しないでください。
  • ブローカー間で同じプライベートキーを使用してください。そうすることで、MDS が権限借用を行う際にトークンを適切に復号化できます。
  • MDS をセットアップした後、OAUTH リスナーでは、MDS 復号化キー(tokenKeypair.pempublic.pem )が異なっても、特にそれらに関するエラーが表示されない点に注意してください。

トークンリスナーの設定用に以下の構成を Kafka プロパティファイル(/etc/kafka/server.properties )に追加します。かっこ(&lt;&gt;)で囲まれた内容は、環境に応じてカスタマイズする必要があります。

 1############################# Token Listener Settings #############################
 2# Add named listener TOKEN to existing listeners and advertised.listeners
 3advertised.listeners=<advertised.listeners>,TOKEN://localhost:9092
 4listeners=<listeners>,TOKEN://:9092
 5# Set SASL callback handler for handling tokens on login. This is essentially a noop if not used for inter-broker communication.
 6listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
 7# Set SASL callback handler for verifying authentication token signatures
 8listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
 9# Configure the public key used to verify RBAC Metadata Service signatures
10# Note: username, password and metadataServerUrls must be set if used for inter-broker communication
11listener.name.token.oauthbearer.sasl.jaas.config= \
12   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
13   publicKeyPath="<path-to-public.pem>";
14# Add protocol mapping for newly-added named listener TOKEN
15listener.security.protocol.map=<listener.map>,TOKEN:SASL_PLAINTEXT
16listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER

重要

MDS 構成のすべてのセクションを完了したら、 Confluent Platform を起動 します。

以下のセクションでは、トークンリスナーの設定を指定するために使用される構成オプションについて説明します。

advertised.listeners

他のブローカーやクライアントで使用する URI とリスナー名のコンマ区切りのリストです。IaaS 環境では、場合によりブローカーがバインドするインターフェイスと異なるものにする必要があります。これが設定されていない場合、listeners の値が使用されます。

listeners

HTTP または HTTPS で API リクエストをリッスンするリスナーのコンマ区切りリスト。各リスナーには、ホスト名およびポートが含まれている必要があります。

listener.name.rbac.oauthbearer.sasl.login.callback.handler.class

MDS JSON ウェブトークンおよび渡されたユーザー名/パスワードを理解および認証できるように、Kafka API 呼び出し(生成や消費など)を構成および有効化するために使用します。詳細については、「トークン認証の構成」を参照してください。

listener.name.rbac.oauthbearer.sasl.server.callback.handler.class

MDS JSON ウェブトークンおよび渡されたユーザー名/パスワードを理解および認証できるように、Kafka API 呼び出し(生成や消費など)を構成および有効化するために使用します。詳細については、「トークン認証の構成」を参照してください。

listener.name.rbac.oauthbearer.sasl.jaas.config

着信した JSON Web Token(JWT)を検証するために、listener.name.rbac.oauthbearer.sasl.login.callback.handler.class によって使用されます。

listener.security.protocol.map

使用するセキュリティープロトコルのキー/値のペアをリスナー名ごとに定義する Kafka ブローカー構成オプションです。リスナー名とセキュリティプロトコルの間でのマッピングを行います。同じセキュリティプロトコルを複数のポートまたは IP で使用するには、これを定義する必要があります。たとえば、内部トラフィックと外部トラフィックの両方に TLS/SSL が必要であっても、これらのトラフィックを分離できます。具体的には、ユーザーは INTERNAL および EXTERNAL という名前のリスナーを定義し、プロパティを INTERNAL:SSL,EXTERNAL:SSL のように定義できます。このように、キーと値はコロンで区切り、マップエントリはコンマで区切ります。各リスナー名は、マップに 1 回だけ指定できます。リスナーごとに異なるセキュリティ(TLS/SSL および SASL)設定を構成できます。その場合は、正規化されたプレフィックス(リスナー名は小文字)を構成名に追加します。

listener.name.rbac.sasl.enabled.mechanisms

RBAC リスナーで有効になっている SASL メカニズムのコンマ区切りリストです。

重要

外部クライアント通信には、トークンサービスまたは OAUTHBEARER の SASL メカニズム(listener.name.rbac.sasl.enabled.mechanisms=OAUTHBEARER)を使用しないでください。RBAC が有効になっている場合、トークンサービスの対象は Confluent Platform コンポーネント間の内部通信のみとなり(たとえば、Schema Registry ライセンスクライアントには有効)、長時間実行されるサービスプリンシパルやクライアント認証は対象になりません。OAUTHBEARER 設定は内部用であり、変更される可能性があります。また、フル機能の OAuth プロトコルは実装されていません。このため、長期またはクライアントのユースケースには、SASL や mTLS(相互 TLS)など、サポートされているいずれかの認証方法を使用してください。詳細については、「認証方法の概要」を参照してください。

重要

MDS 構成のすべてのセクションを完了したら、 Confluent Platform を起動 します。

プライマリ Kafka クラスターの完全な MDS 構成

この例では、MDS とロールバインディングをホストしているプライマリ Kafka クラスターの完全な構成を示しています。

 1############################# Confluent Server Authorizer Settings  #############################
 2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
 3
 4# Specify a list of Access Rule Providers to retain ACLs that have already been enabled and to enable RBAC
 5confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT
 6
 7# Specify when bootstrapping Confluent Platform and to assign a SystemAdmin
 8super.users=<User:admin;User:mds>
 9
10############################# Identity Provider Settings (LDAP) #############################
11# Search groups for group-based authorization.
12ldap.group.name.attribute=<sAMAccountName>
13ldap.group.object.class=group
14ldap.group.member.attribute=member
15ldap.group.member.attribute.pattern=CN=(.*),DC=rbac,DC=confluent,DC=io
16ldap.group.search.base=CN=Users,DC=rbac,DC=confluent,DC=io
17#Limit the scope of searches to subtrees off of base
18ldap.user.search.scope=2
19#Enable filters to limit search to only those groups needed
20ldap.group.search.filter=(|(CN=<specific group>)(CN=<specific group>))
21
22# Kafka authenticates to the directory service with the bind user.
23ldap.java.naming.provider.url=ldap://<hostname>:389
24ldap.java.naming.security.authentication=simple
25ldap.java.naming.security.credentials=<password>
26ldap.java.naming.security.principal=<mds-user-DN>
27
28# Locate users. Make sure that these attributes and object classes match what is in your directory service.
29ldap.user.name.attribute=<sAMAccountName>
30ldap.user.object.class=user
31ldap.user.search.base=<user-search-base-DN
32
33############################# MDS Server Settings #############################
34# Bind Metadata Service HTTP service to port 8090.
35confluent.metadata.server.listeners=http://0.0.0.0:8090
36# The key to encrypt the token (when you issue you a token)
37confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
38# Supported authentication methods
39confluent.metadata.server.authentication.method=BEARER
40
41############################# Token Listener Settings #############################
42# Add named listener TOKEN to existing listeners and advertised.listeners
43advertised.listeners=<advertised.listeners>,TOKEN://localhost:9092
44listeners=<listeners>,TOKEN://:9092
45# Set SASL callback handler for handling tokens on login. This is essentially a noop if not used for inter-broker communication.
46listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
47# Set SASL callback handler for verifying authentication token signatures
48listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
49# Configure the public key used to verify RBAC Metadata Service signatures
50# Note: username, password and metadataServerUrls must be set if used for inter-broker communication
51listener.name.token.oauthbearer.sasl.jaas.config= \
52   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
53   publicKeyPath="<path-to-public.pem>";
54# Add protocol mapping for newly-added named listener TOKEN
55listener.security.protocol.map=<listener.map>,TOKEN:SASL_PLAINTEXT
56listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER

MDS の構成の問題のトラブルシューティング

以下のセクションでは、MDS の構成で発生する可能性のある問題のトラブルシューティングに役立つガイドを提供します。

トークンの認証

OAUTHBEARER を使用した MDS へのトークンログインおよび認証は、明白な既知の例外またはエラーが server.log または metadata-service.log のいずれかに示されることなく失敗することがあります。このような失敗の原因は、パブリックキーファイルまたはプライベートキーファイルが破損しているか、異なるパブリックキーファイルとプライベートキーファイルを使用してトークンが生成されたことであると考えられます。

トークン認証のトラブルシューティングを行う場合

  1. server.properties で構成されているフォルダーから tokenKeypair.pempublic.pem の両方を削除して、これらをもう一度生成します。新しいキーファイルは以下の場所に生成されます。

    confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
    
  2. tokenKeypair.pempublic.pem を再生成した後、MDS が実行されているブローカーサーバーを再起動します。

  3. クライアントマシンで、スーパーユーザーがログインした後に、スーパーユーザーのトークンがキャッシュされているローカル CLI キャッシュ(~/.confluent/config.json)を削除します。

  4. CLI を使用してもう一度ログインします。

MDS REST クライアントの構成

MDS と通信するように構成されているコンポーネント(Schema Registry、ksqlDB、Confluent Control Center、Connect など)のクライアントに誤ったユーザー名やパスワードが含まれている場合、認証の試行が無限にループする可能性があります。それにより、REST クライアントの例外ログで例外の継続的なループが生じ、パフォーマンスに影響することがあります。その例を次に示します。

[2021-01-25 05:11:58,330] ERROR [pool-17-thread-1] Error while refreshing active metadata server urls, retrying (io.confluent.security.auth.client.rest.RestClient)
io.confluent.security.auth.client.rest.exceptions.RestClientException: Unauthorized; error code: 401
   at io.confluent.security.auth.client.rest.RestClient$HTTPRequestSender.lambda$submit$0(RestClient.java:353)
   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
   at java.lang.Thread.run(Thread.java:748)

認証の再試行の回数を制御するには、REST クライアントの MDS の構成に次のオプションを含めます。

  • confluent.metadata.server.urls.max.retries
  • confluent.metadata.server.urls.fail.on.401

これらの構成オプションの詳細については、「REST クライアントの構成」を参照してください。

プライマリ Kafka クラスターの MDS によって管理されるセカンダリ Kafka クラスターの構成

以下のセクションでは、プライマリ Kafka クラスターの MDS によって管理されるセカンダリ Kafka クラスターの構成方法について説明します。

重要

MDS 構成のすべてのセクションを完了したら、 Confluent Platform を起動 します。

Confluent Server Authorizer の構成

Confluent Server Authorizer の概要と、ここで推奨されている設定の詳細については、「Confluent Server Authorizer の構成」を参照してください。

以下の MDS 構成を Kafka プロパティファイル(/etc/kafka/server.properties)に追加します。かっこ(&lt;&gt;)で囲まれた内容は、環境に応じてカスタマイズする必要があります。

1############################# Confluent Server Authorizer Settings  #############################
2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
3confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT

重要

MDS 構成の すべて のセクションを完了したら、 Confluent Platform を起動します

MDS の構成

このセクションでは、MDS クラスターと通信してロールバインディングを "消費" できるように Kafka を構成します。以下の構成例では SASL_PLAINTEXT/PLAIN が使用されていますが、MDS を実行している Kafka ブローカーに必要なセキュリティメカニズムを使用してください。また、他のブローカーによって公開されているポートを指定する必要があります。詳細については、「Metadata Service の構成オプション」を参照してください。

1############################# MDS Settings #############################
2confluent.metadata.bootstrap.servers=<kafka-with-mds-host-1>:<port>,<kafka-with-mds-host-2>:<port>,...
3confluent.metadata.security.protocol=SASL_PLAINTEXT
4confluent.metadata.sasl.mechanism=PLAIN
5confluent.metadata.sasl.jaas.config= \
6   org.apache.kafka.common.security.plain.PlainLoginModule required \
7   username="<broker-username>" \
8   password="<broker-password>";

重要

MDS 構成の すべて のセクションを完了したら、 Confluent Platform を起動します

以下のセクションでは、MDS の設定を指定するために使用される構成オプションについて説明します。

confluent.metadata.bootstrap.servers

MDS をホストするクラスターの Kafka ホストとポート。ブローカーが MDS に接続する方法を決定するために使用されます。このオプションはマネージド型ブローカーでは必須であり、デフォルトでブローカー間の値に設定される MDS ブローカーではオプションです。たとえば ec2-22-222-22-222.compute-1.amazonaws.com:9092 のような値になります。

confluent.metadata.security.protocol

外部 MDS に接続するときに使用されるセキュリティオプションを定義し、マネージド型ブローカーに RBAC データを消費する機能を提供します。

confluent.metadata.sasl.mechanism

外部 MDS に接続するときに使用されるセキュリティオプションを定義し、マネージド型ブローカーに RBAC データを消費する機能を提供します。

confluent.metadata.sasl.jaas.config

マネージド型クラスターがロールバインディングデータに接続して消費するための JAAS 構成を定義します。これにより、マネージド型クラスターが Kafka API による自身への直接呼び出しで RBAC をローカル適用できるようになります。

トークンリスナーの構成

MDS 構成のこのセクションでは、権限借用に使用される OAUTHBEARER SASL メカニズムによってリスナーを有効にします。詳細については、「Kafka ブローカーの推奨構成」を参照してください。MDS 構成でのトークンリスナー設定の詳細については、「トークンリスナーの構成」を参照してください。

ここで使用されているトークンリスナーの設定と、前述の「トークンリスナーの構成」で説明されている設定との違いについては、以下で詳しく示します。

1############################# Token-based Listener Settings #############################
2listeners=<listeners>,TOKEN://:9092
3advertised.listeners=<advertised.listeners>,TOKEN://<hostname>:9092
4listener.security.protocol.map=<advertised.listeners>,TOKEN:SASL_PLAINTEXT
5listener.name.token.oauthbearer.sasl.jaas.config= \
6   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
7   publicKeyPath="<path-to-public.pem>";
8listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
9listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler

listener.name.token.oauthbearer.sasl.jaas.config

定義済みの LISTENER_NAMEoauthbearer.sasl.jaas.config です。ここで、リスナーの名前は TOKEN です。つまり、listener.name.token.oauthbearer.sasl.jaas.configlistener.name.<LISTENER_NAME>.oauthbearer.sasl.jaas.config のインスタンスです。

重要

MDS 構成の すべて のセクションを完了したら、 Confluent Platform を起動します

セカンダリ Kafka クラスターの完全な MDS 構成

この例では、プライマリ Kafka クラスターの MDS によって管理されるセカンダリ Kafka クラスターの完全な構成を示しています。

 1############################# Confluent Server Authorizer Settings  #############################
 2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
 3confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT
 4
 5############################# MDS Settings #############################
 6confluent.metadata.bootstrap.servers=<kafka-with-mds-host-1>:<port>,<kafka-with-mds-host-2>:<port>,...
 7confluent.metadata.security.protocol=SASL_PLAINTEXT
 8confluent.metadata.sasl.mechanism=PLAIN
 9confluent.metadata.sasl.jaas.config= \
10   org.apache.kafka.common.security.plain.PlainLoginModule required \
11   username="<broker-username>" \
12   password="<broker-password>";
13
14############################# Token-based Listener Settings #############################
15listeners=<listeners>,TOKEN://:9092
16advertised.listeners=<advertised.listeners>,TOKEN://<hostname>:9092
17listener.security.protocol.map=<advertised.listeners>,TOKEN:SASL_PLAINTEXT
18listener.name.token.oauthbearer.sasl.jaas.config= \
19   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
20   publicKeyPath="<path-to-public.pem>";
21listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
22listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler