7.2. OpenShift 外クライアントのアクセスの設定

1. TLS 暗号化および認証、ならびに Kafka 簡易承認 を有効にして、Kafka クラスターに外部リスナーが設定されます。
2. 簡易承認 用に TLS 認証および アクセス制御リスト (ACL) を定義して、クライアントに KafkaUser が作成されます。

手順

1. external Kafka リスナーと共に Kafka クラスターを設定します。

   • リスナーを通じて Kafka ブローカーにアクセスするのに必要な認証を定義します。

   • Kafka ブローカーで承認を有効にします。

     以下に例を示します。

     apiVersion: kafka.strimzi.io/v1beta2
     kind: Kafka
     metadata:
       name: my-cluster
       namespace: myproject
     spec:
       kafka:
         # ...
         listeners: 1
         - name: external 2
           port: 9094 3
           type: LISTENER-TYPE 4
           tls: true 5
           authentication:
             type: tls 6
           configuration:
             preferredNodePortAddressType: InternalDNS 7
             bootstrap and broker service overrides 8
             #...
         authorization: 9
           type: simple
           superUsers:
             - super-user-name 10
       # ...

     1

     外部リスナーを有効にする設定オプションは、汎用 Kafka リスナースキーマ参照 に記載されています。

     2

     リスナーを識別するための名前。Kafka クラスター内で一意である必要があります。

     3

     Kafka 内でリスナーによって使用されるポート番号。ポート番号は指定の Kafka クラスター内で一意である必要があります。許可されるポート番号は 9092 以上ですが、すでに Prometheus および JMX によって使用されているポート 9404 および 9999 以外になります。リスナーのタイプによっては、ポート番号は Kafka クライアントに接続するポート番号と同じではない場合があります。

     4

     route、loadbalancer、nodeport、または ingress として指定される外部リスナータイプ。内部リスナーは internal として指定されます。

     5

     リスナーで TLS による暗号化を有効にします。デフォルトは false です。route リスナーには TLS 暗号化は必要ありません。

     6

     認証は tls として指定されます。

     7

     (任意設定: nodeport リスナーのみ) ノードアドレスとして AMQ Streams によって使用される最初のアドレスタイプの希望を指定します。

     8

     (任意設定) AMQ Streams はクライアントに公開するアドレスを自動的に決定します。アドレスは OpenShift によって自動的に割り当てられます。AMQ Streams を実行しているインフラストラクチャーが正しい ブートストラップおよびブローカーサービスのアドレス を提供しない場合、そのアドレスを上書きできます。検証はオーバーライドに対しては実行されません。オーバーライド設定はリスナーのタイプによって異なります。たとえば、route の場合はホストを、 loadbalancer の場合は DNS 名または IP アドレスを、また nodeport の場合はノードポートを、それぞれ上書きすることができます。

     9

     simple と指定された承認 (AclAuthorizer Kafka プラグインを使用する)。

     10

     (任意設定) スーパーユーザーは、ACL で定義されたアクセス制限に関係なく、すべてのブローカーにアクセスできます。

     警告

     OpenShift Route アドレスは、Kafka クラスターの名前、リスナーの名前、および作成される namespace の名前で構成されます。たとえば、my-cluster-kafka-listener1-bootstrap-myproject (CLUSTER-NAME-kafka-LISTENER-NAME-bootstrap-NAMESPACE) となります。route リスナータイプを使用している場合、アドレス全体の長さが上限の 63 文字を超えないように注意してください。

2. Kafka リソースを作成または更新します。

   oc apply -f <kafka_configuration_file>

   Kafka クラスターは、TLS 認証を使用する Kafka ブローカーリスナーと共に設定されます。

   Kafka ブローカー Pod ごとにサービスが作成されます。

   サービスが作成され、Kafka クラスターに接続するための ブートストラップアドレス として機能します。

   サービスは、nodeport リスナーを使用した Kafka クラスターへの外部接続用 外部ブートストラップアドレス としても作成されます。

   kafka ブローカーのアイデンティティーを検証するクラスター CA 証明書もシークレット <cluster_name>-cluster-ca-cert に作成されます。

   注記

   外部リスナーの使用時に Kafka クラスターをスケーリングする場合に、すべての Kafka ブローカーのローリングアップデートがトリガーされる可能性があります。これは設定によって異なります。

3. Kafka リソースのステータスからブートストラップアドレスおよびポートを見つけます。

   oc get kafka KAFKA-CLUSTER-NAME -o jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}'

   Kafka クライアントのブートストラップアドレスを使用して、Kafka クラスターに接続します。

4. Kafka クラスターにアクセスする必要があるクライアントに対応するユーザーを作成または変更します。

   • Kafka リスナーと同じ認証タイプを指定します。

   • 簡易承認に承認 ACL を指定します。

     以下に例を示します。

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaUser
     metadata:
       name: my-user
       labels:
         strimzi.io/cluster: my-cluster 1
     spec:
       authentication:
         type: tls 2
       authorization:
         type: simple
         acls: 3
           - resource:
               type: topic
               name: my-topic
               patternType: literal
             operation: Read
           - resource:
               type: topic
               name: my-topic
               patternType: literal
             operation: Describe
           - resource:
               type: group
               name: my-group
               patternType: literal
             operation: Read

     1

     ラベルは、作成するユーザーの Kafka クラスターのラベルと一致する必要があります。

     2

     認証は tls として指定されます。

     3

     簡易承認には、ユーザーに適用する ACL ルールのリストが必要です。ルールは、ユーザー名 (my-user) を基に Kafka リソースで許可される操作を定義します。

5. KafkaUser リソースを作成または変更します。

   oc apply -f USER-CONFIG-FILE

   KafkaUser リソースと同じ名前の Secret と共に、ユーザーが作成されます。Secret には、TLS クライアント認証の秘密鍵と公開鍵が含まれます。

   以下に例を示します。

   apiVersion: v1
   kind: Secret
   metadata:
     name: my-user
     labels:
       strimzi.io/kind: KafkaUser
       strimzi.io/cluster: my-cluster
   type: Opaque
   data:
     ca.crt: PUBLIC-KEY-OF-THE-CLIENT-CA
     user.crt: USER-CERTIFICATE-CONTAINING-PUBLIC-KEY-OF-USER
     user.key: PRIVATE-KEY-OF-USER
     user.p12: P12-ARCHIVE-FILE-STORING-CERTIFICATES-AND-KEYS
     user.password: PASSWORD-PROTECTING-P12-ARCHIVE

6. パブリッククラスター CA 証明書を必要な証明書形式に展開します。

   oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

7. パスワードファイルからパスワードを抽出します。

   oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password

8. パブリッククラスター証明書の認証情報でクライアントを設定します。

   クライアントコードのサンプル

   properties.put("security.protocol","SSL"); 1
   properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,"/path/to/ca.p12"); 2
   properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,CA-PASSWORD); 3
   properties.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG,"PKCS12"); 4

   1

   (TLS クライアント認証ありまたはなしで) TLS による暗号化を有効にします。

   2

   証明書がインポートされたトラストストアの場所を指定します。

   3

   トラストストアにアクセスするためのパスワードを指定します。このプロパティーは、トラストストアで必要なければ省略できます。

   4

   トラストストアのタイプを識別します。

   注記

   セキュリティープロトコルを使用します。 (TLS で SCRAM-SHA 認証を使用する場合)

9. ユーザー Secret から必要な証明書形式にユーザー CA 証明書を抽出します。

   oc get secret USER-NAME -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12

10. パスワードファイルからパスワードを抽出します。

    oc get secret USER-NAME -o jsonpath='{.data.user\.password}' | base64 -d > user.password

11. ユーザー CA 証明書の認証情報でクライアントを設定します。

    クライアントコードのサンプル

    properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,"/path/to/user.p12"); 1
    properties.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,"<user.password>"); 2
    properties.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG,"PKCS12"); 3

    1

    証明書がインポートされたキーストアの場所を指定します。

    2

    キーストアにアクセスするためのパスワードを指定します。このプロパティーは、キーストアで必要なければ省略できます。パブリックユーザー証明書は、作成時にクライアント CA により署名されます。

    3

    キーストアのタイプを識別します。

12. Kafka クラスターに接続するためのブートストラップアドレスおよびポートを追加します。

    bootstrap.servers: BOOTSTRAP-ADDRESS:PORT