第6章 Kafka クラスターへのクライアントアクセスの設定

手順

1. Kafka プロデューサーをデプロイします。

   oc run kafka-producer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-27-rhel7:1.7.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list cluster-name-kafka-bootstrap:9092 --topic my-topic

   Copy to Clipboard Toggle word wrap
2. プロデューサーが稼働しているコンソールにメッセージを入力します。
3. Enter を押してメッセージを送信します。

4. Kafka コンシューマーをデプロイします。

   oc run kafka-consumer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-27-rhel7:1.7.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server cluster-name-kafka-bootstrap:9092 --topic my-topic --from-beginning

   Copy to Clipboard Toggle word wrap
5. コンシューマーコンソールに受信メッセージが表示されることを確認します。

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

     
       # ...

     Copy to Clipboard Toggle word wrap

     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-CONFIG-FILE

   Copy to Clipboard Toggle word wrap

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

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

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

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

   kafka ブローカーの ID を検証するためのクラスター CA 証明書も、Kafka リソースと同じ名前で作成されます。

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

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

   Copy to Clipboard Toggle word wrap

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

4. 生成された KAFKA-CLUSTER-NAME-cluster-ca-cert Secret から、公開クラスター CA 証明書およびパスワードを抽出します。

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   Kafka クライアントの証明書およびパスワードを使用して、TLS 暗号化により Kafka クラスターに接続します。

   注記

   デフォルトでは、クラスター CA 証明書は自動的に更新されます。専用の Kafka リスナー証明書を使用している場合は、証明書を手動で更新する 必要があります。

5. 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

     Copy to Clipboard Toggle word wrap

     1

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

     2

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

     3

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

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

   oc apply -f USER-CONFIG-FILE

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

7. Kafka クラスターへのセキュアな接続を確立するのに必要なプロパティーを使用して Kafka クラスターに接続するように、クライアントを設定します。

   1. パブリッククラスター証明書の認証詳細を追加します。

      security.protocol: SSL 

      1

      
      ssl.truststore.location: PATH-TO/ssl/keys/truststore 

      2

      
      ssl.truststore.password: CLUSTER-CA-CERT-PASSWORD 

      3

      
      ssl.truststore.type=PKCS12 

      4

      Copy to Clipboard Toggle word wrap

      1

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

      2

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

      3

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

      4

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

      注記

      TLS 経由で SCRAM-SHA 認証を使用する場合は、security.protocol: SASL_SSL を使用します。

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

      bootstrap.servers: BOOTSTRAP-ADDRESS:PORT

      Copy to Clipboard Toggle word wrap

   3. パブリックユーザー証明書の認証情報を追加します。

      ssl.keystore.location: PATH-TO/ssl/keys/user1.keystore 

      1

      
      ssl.keystore.password: USER-CERT-PASSWORD 

      2

      Copy to Clipboard Toggle word wrap

      1

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

      2

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

      パブリックユーザー証明書は、作成時にクライアント CA により署名されます。