AMQ Streams on OpenShift の使用

手順

1. 手動で更新する Kafka Pod を制御する StatefulSet の名前を見つけます。

   たとえば、Kafka クラスターの名前が my-cluster の場合、対応する StatefulSet の名前は my-cluster-kafka になります。

2. OpenShift で StatefulSet リソースにアノテーションを付けます。たとえば、oc annotate を使用すると以下のようになります。

   oc annotate statefulset cluster-name-kafka strimzi.io/manual-rolling-update=true

3. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。アノテーションが調整プロセスで検出されれば、アノテーションが付いた StatefulSet 内のすべての Pod でローリングアップデートがトリガーされます。すべての Pod のローリングアップデートが完了すると、アノテーションは StatefulSet から削除されます。

手順

1. 手動で更新する ZooKeeper Pod を制御する StatefulSet の名前を見つけます。

   たとえば、Kafka クラスターの名前が my-cluster の場合、ZooKeeper の対応する StatefulSet の名前は my-cluster-zookeeper になります。

2. OpenShift で StatefulSet リソースにアノテーションを付けます。たとえば、oc annotate を使用すると以下のようになります。

   oc annotate statefulset cluster-name-zookeeper strimzi.io/manual-rolling-update=true

3. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。アノテーションが調整プロセスで検出されれば、アノテーションが付いた StatefulSet 内のすべての Pod でローリングアップデートがトリガーされます。すべての Pod のローリングアップデートが完了すると、アノテーションは StatefulSet から削除されます。

手順

1. 削除する Pod の名前を見つけます。

   たとえば、クラスターの名前が cluster-name の場合、Pod の名前は cluster-name-kafka-index になります。index はゼロで始まり、レプリカーの合計数で終わる値です。

2. OpenShift で Pod リソースにアノテーションを付けます。

   oc annotate を使用します。

   oc annotate pod cluster-name-kafka-index strimzi.io/delete-pod-and-pvc=true

3. 基盤となる永続ボリューム要求 (Persistent Volume Claim) でアノテーションが付けられた Pod が削除され、再作成されるときに、次の調整の実行を待ちます。

手順

1. 削除する Pod の名前を見つけます。

   たとえば、クラスターの名前が cluster-name の場合、Pod の名前は cluster-name-zookeeper-index になります。index はゼロで始まり、レプリカーの合計数で終わる値です。

2. OpenShift で Pod リソースにアノテーションを付けます。

   oc annotate を使用します。

   oc annotate pod cluster-name-zookeeper-index strimzi.io/delete-pod-and-pvc=true

3. 基盤となる永続ボリューム要求 (Persistent Volume Claim) でアノテーションが付けられた Pod が削除され、再作成されるときに、次の調整の実行を待ちます。

手順

1. strimzi.io/force-renew アノテーションを、更新対象の CA 証明書が含まれる Secret に適用します。

   表2.1 証明書の更新を強制する Secret のアノテーション。

   証明書	Secret	annotate コマンド
   クラスター CA	KAFKA-CLUSTER-NAME-cluster-ca-cert	oc annotate secret KAFKA-CLUSTER-NAME-cluster-ca-cert strimzi.io/force-renew=true
   クライアント CA	KAFKA-CLUSTER-NAME-clients-ca-cert	oc annotate secret KAFKA-CLUSTER-NAME-clients-ca-cert strimzi.io/force-renew=true

   次回の調整で、アノテーションを付けた Secret の新規 CA 証明書が Cluster Operator によって生成されます。メンテナンス時間枠が設定されている場合、Cluster Operator によって、最初の調整時に次のメンテナンス時間枠内で新規 CA 証明書が生成されます。

   Cluster Operator によって更新されたクラスターおよびクライアント CA 証明書をクライアントアプリケーションがリロードする必要があります。

2. CA 証明書が有効である期間を確認します。

   たとえば、openssl コマンドを使用します。

   oc get secret CA-CERTIFICATE-SECRET -o 'jsonpath={.data.CA-CERTIFICATE}' | base64 -d | openssl x509 -subject -issuer -startdate -enddate -noout

   CA-CERTIFICATE-SECRETはSecretの名前で、クラスタCA証明書の場合はKAFKA-CLUSTER-NAME-cluster-ca-certであり、クライアントCA証明書の場合はKAFKA-CLUSTER-NAME-clients-ca-certとなります。

   CA-CERTIFICATEは、jsonpath={.data.ca\.crt}のように、CA証明書の名前です。

   このコマンドは、CA 証明書の有効期間である notBefore および notAfter の日付を返します。

   たとえば、クラスター CA 証明書の場合は以下のようになります。

   subject=O = io.strimzi, CN = cluster-ca v0
   issuer=O = io.strimzi, CN = cluster-ca v0
   notBefore=Jun 30 09:43:54 2020 GMT
   notAfter=Jun 30 09:43:54 2021 GMT

3. Secret から古い証明書を削除します。

   コンポーネントで新しい証明書が使用される場合でも、古い証明書がアクティブであることがあります。古い証明書を削除して、潜在的なセキュリティーリスクを取り除きます。

手順

1. KafkaConnect または KafkaConnectS2I リソースの spec プロパティーを編集します。

   設定可能なプロパティーは以下の例のとおりです。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect 1
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 2
   spec:
     replicas: 3 3
     authentication: 4
       type: tls
       certificateAndKey:
         certificate: source.crt
         key: source.key
         secretName: my-user-source
     bootstrapServers: my-cluster-kafka-bootstrap:9092 5
     tls: 6
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     config: 7
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     externalConfiguration: 8
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey
     resources: 9
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 10
       type: inline
       loggers:
         log4j.rootLogger: "INFO"
     readinessProbe: 11
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metrics: 12
       lowercaseOutputName: true
       lowercaseOutputLabelNames: true
       rules:
         - pattern: kafka.connect<type=connect-worker-metrics><>([a-z-]+)
           name: kafka_connect_worker_$1
           help: "Kafka Connect JMX metric worker"
           type: GAUGE
         - pattern: kafka.connect<type=connect-worker-rebalance-metrics><>([a-z-]+)
           name: kafka_connect_worker_rebalance_$1
           help: "Kafka Connect JMX metric rebalance information"
           type: GAUGE
     jvmOptions: 13
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 14
     template: 15
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 16
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   必要に応じて KafkaConnect または KafkaConnectS2I を使用します。

   2

   Kafka Connect クラスターの KafkaConnectors を有効にします。

   3

   レプリカノードの数。

   4

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、Kafka Connect クラスターの認証。デフォルトでは、Kafka Connect はプレーンテキスト接続を使用して Kafka ブローカーに接続します。

   5

   Kafka Connect クラスターに接続するためのブートストラップサーバー。

   6

   クラスターの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   7

   ワーカー の Kafka Connect 設定 (コネクターではない)。標準の Apache Kafka 設定が提供されることがありますが、AMQ Streams によって直接管理されないプロパティーに制限されます。

   8

   ここで示す環境変数や、ボリュームを使用した Kafka コネクターの外部設定

   9

   サポートされているリソース（現在は cpu と memory）の予約と、消費可能な最大リソースを指定するための制限を要求します。

   10

   指定された Kafka loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。Kafka Connect log4j.rootLogger ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   11

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するためのヘルスチェック。

   12

   Prometheus メトリクス。この例では、Prometheus JMX エクスポーターの設定で有効になっています。metrics: {} を使用すると追加設定なしでメトリクスを有効にすることができます。

   13

   Kafka Connect を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   14

   高度なオプション：特別な場合のみ 推奨される コンテナーイメージの設定。

   15

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   16

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

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

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

3. Kafka Connect の承認が有効である場合、Kafka Connect ユーザーを設定し、Kafka Connect のコンシューマーグループおよびトピックへのアクセスを有効にします。

手順

1. KafkaUser リソースの authorization プロパティーを編集し、アクセス権限をユーザーに付与します。

   以下の例では、literal の名前の値を使用して Kafka Connect トピックおよびコンシューマーグループにアクセス権限が設定されます。

   プロパティー	名前
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Read
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Read
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Read
           host: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operation: Read
           host: "*"

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

   oc apply -f KAFKA-USER-CONFIG-FILE

手順

1. KafkaMirrorMaker リソースの spec プロパティーを編集します。

   設定可能なプロパティーは以下の例のとおりです。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     replicas: 3 1
     consumer:
       bootstrapServers: my-source-cluster-kafka-bootstrap:9092 2
       groupId: "my-group" 3
       numStreams: 2 4
       offsetCommitInterval: 120000 5
       tls: 6
         trustedCertificates:
         - secretName: my-source-cluster-ca-cert
           certificate: ca.crt
       authentication: 7
         type: tls
         certificateAndKey:
           secretName: my-source-secret
           certificate: public.crt
           key: private.key
       config: 8
         max.poll.records: 100
         receive.buffer.bytes: 32768
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 9
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 10
     producer:
       bootstrapServers: my-target-cluster-kafka-bootstrap:9092
       abortOnSendFailure: false 11
       tls:
         trustedCertificates:
         - secretName: my-target-cluster-ca-cert
           certificate: ca.crt
       authentication:
         type: tls
         certificateAndKey:
           secretName: my-target-secret
           certificate: public.crt
           key: private.key
       config:
         compression.type: gzip
         batch.size: 8192
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 12
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 13
     whitelist: "my-topic|other-topic" 14
     resources: 15
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 16
       type: inline
       loggers:
         mirrormaker.root.logger: "INFO"
     readinessProbe: 17
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metrics: 18
       lowercaseOutputName: true
       rules:
         - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*><>Count"
           name: "kafka_server_$1_$2_total"
         - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*,
           topic=(.+)><>Count"
           name: "kafka_server_$1_$2_total"
           labels:
             topic: "$3"
     jvmOptions: 19
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 20
     template: 21
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 22
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing: 23
       type: jaeger

   1

   レプリカノードの数。

   2

   コンシューマーおよびプロデューサーのブートストラップサーバー。

   3

   コンシューマーのグループ ID。

   4

   コンシューマーストリームの数。

   5

   オフセットの自動コミット間隔 (ミリ秒単位)。

   6

   コンシューマーまたはプロデューサーの TLS 証明書が X.509 形式で保存される、キー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   7

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、コンシューマーおよびプロデューサーの認証。

   8

   コンシューマー および プロデューサー の Kafka 設定オプション。

   9

   TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。

   10

   HTTPS に設定することで、ホスト名の検証が有効になります。空の文字列を指定すると検証が無効になります。

   11

   abortOnSendFailure プロパティが true に設定されている場合、メッセージの送信に失敗した後、Kafka MirrorMaker は終了し、コンテナは再起動します。

   12

   TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。

   13

   HTTPS に設定することで、ホスト名の検証が有効になります。空の文字列を指定すると検証が無効になります。

   14

   ソースからターゲット Kafka クラスターにミラーリングされたトピックの 許可リスト。

   15

   サポートされているリソース（現在は cpu と memory）の予約と、消費可能な最大リソースを指定するための制限を要求します。

   16

   指定された loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。MirrorMaker には mirrormaker.root.logger と呼ばれる単一のロガーがあります。ログレベルは INFO、ERROR、WARN、TRACE、DEBUG、FATAL、または OFF に設定できます。

   17

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するためのヘルスチェック。

   18

   Prometheus メトリクス。この例では、Prometheus JMX エクスポーターの設定で有効になっています。metrics: {} を使用すると追加設定なしでメトリクスを有効にすることができます。

   19

   Kafka MirrorMaker を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   20

   高度なオプション：特別な場合のみ 推奨される コンテナーイメージの設定。

   21

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   22

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

   23

   Jaeger の分散トレーシングは有効になっています。

   警告

   abortOnSendFailure プロパティーが false に設定されると、プロデューサーはトピックの次のメッセージを送信しようとします。失敗したメッセージは再送されないため、元のメッセージが失われる可能性があります。

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

   oc apply -f <your-file>

手順

1. KafkaMirrorMaker2 リソースの spec プロパティーを編集します。

   設定可能なプロパティーは以下の例のとおりです。

   apiVersion: kafka.strimzi.io/v1alpha1
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 2.6.0 1
     replicas: 3 2
     connectCluster: "my-cluster-target" 3
     clusters: 4
     - alias: "my-cluster-source" 5
       authentication: 6
         certificateAndKey:
           certificate: source.crt
           key: source.key
           secretName: my-user-source
         type: tls
       bootstrapServers: my-cluster-source-kafka-bootstrap:9092 7
       tls: 8
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-source-cluster-ca-cert
     - alias: "my-cluster-target" 9
       authentication: 10
         certificateAndKey:
           certificate: target.crt
           key: target.key
           secretName: my-user-target
         type: tls
       bootstrapServers: my-cluster-target-kafka-bootstrap:9092 11
       config: 12
         config.storage.replication.factor: 1
         offset.storage.replication.factor: 1
         status.storage.replication.factor: 1
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 13
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 14
       tls: 15
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-target-cluster-ca-cert
     mirrors: 16
     - sourceCluster: "my-cluster-source" 17
       targetCluster: "my-cluster-target" 18
       sourceConnector: 19
         config:
           replication.factor: 1 20
           offset-syncs.topic.replication.factor: 1 21
           sync.topic.acls.enabled: "false" 22
           replication.policy.separator: "" 23
           replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy" 24
       heartbeatConnector: 25
         config:
           heartbeats.topic.replication.factor: 1 26
       checkpointConnector: 27
         config:
           checkpoints.topic.replication.factor: 1 28
       topicsPattern: ".*" 29
       groupsPattern: "group1|group2|group3" 30
     resources: 31
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 32
       type: inline
       loggers:
         connect.root.logger.level: "INFO"
     readinessProbe: 33
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     jvmOptions: 34
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 35
     template: 36
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 37
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger 38
     externalConfiguration: 39
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey

   1

   Kafka Connect の バージョン。

   2

   レプリカノードの数。

   3

   Kafka Connect の クラスターエイリアス。

   4

   同期される Kafka クラスターの 指定。

   5

   ソースの Kafka クラスターの クラスターエイリアス。

   6

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、ソースクラスターの認証。

   7

   ソース Kafka クラスターに接続するための ブートストラップサーバー。

   8

   ソース Kafka クラスターの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   9

   ターゲット Kafka クラスターの クラスターエイリアス。

   10

   ターゲット Kafka クラスターの認証は、ソース Kafka クラスターと同様に設定されます。

   11

   ターゲット Kafka クラスターに接続するための ブートストラップサーバー。

   12

   Kafka Connect の設定。標準の Apache Kafka 設定が提供されることがありますが、AMQ Streams によって直接管理されないプロパティーに限定されます。

   13

   TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。

   14

   HTTPS に設定することで、ホスト名の検証が有効になります。空の文字列を指定すると検証が無効になります。

   15

   ターゲット Kafka クラスターの TLS による暗号化は、ソース Kafka クラスターと同様に設定されます。

   16

   MirrorMaker 2.0 コネクター。

   17

   MirrorMaker 2.0 コネクターによって使用されるソースクラスターの クラスターエイリアス。

   18

   MirrorMaker 2.0 コネクターによって使用されるターゲットクラスターの クラスターエイリアス。

   19

   リモートトピックを作成する MirrorSourceConnectorの設定。デフォルトの設定オプションは config によって上書きされます。

   20

   ターゲットクラスターで作成されるミラーリングされたトピックのレプリケーション係数。

   21

   ソースおよびターゲットクラスターのオフセットをマップする MirrorSourceConnector offset-syncs 内部トピックのレプリケーション係数。

   22

   ACL ルールの同期 が有効になっていると、同期されたトピックに ACL が適用されます。デフォルトは true です。

   23

   リモートトピック名の変更に使用する区切り文字を定義します。

   24

   リモートトピック名の自動変更をオーバーライドするポリシーを追加します。その名前の前にソースクラスターの名前を追加する代わりに、トピックが元の名前を保持します。このオプションの設定は、active/passive バックアップおよびデータ移行に役立ちます。

   25

   接続チェックを実行する MirrorHeartbeatConnector の設定。デフォルトの設定オプションは config によって上書きされます。

   26

   ターゲットクラスターで作成されたハートビートトピックのレプリケーション係数。

   27

   オフセットを追跡する MirrorCheckpointConnectorの設定。デフォルトの設定オプションは config によって上書きされます。

   28

   ターゲットクラスターで作成されたチェックポイントトピックのレプリケーション係数。

   29

   正規表現パターンとして定義されたソースクラスターからのトピックレプリケーション。ここで、すべてのトピックを要求します。

   30

   正規表現パターンとして定義されたソースクラスターからのコンシューマーグループレプリケーション。ここで、3 つのコンシューマーグループを名前で要求します。コンマ区切りリストを使用できます。

   31

   サポートされているリソース（現在は cpu と memory）の予約と、消費可能な最大リソースを指定するための制限を要求します。

   32

   指定された Kafka loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。Kafka Connect log4j.rootLogger ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   33

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するためのヘルスチェック。

   34

   Kafka MirrorMaker を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   35

   高度なオプション：特別な場合のみ 推奨される コンテナーイメージの設定。

   36

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   37

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

   38

   Jaeger の分散トレーシングは有効になっています。

   39

   環境変数として Kafka MirrorMaker にマウントされた OpenShift Secret の外部設定。

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

   oc apply -f <your-file>

手順

1. KafkaBridge リソースの spec プロパティーを編集します。

   設定可能なプロパティーは以下の例のとおりです。

   apiVersion: kafka.strimzi.io/v1alpha1
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     replicas: 3 1
     bootstrapServers: my-cluster-kafka-bootstrap:9092 2
     tls: 3
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     authentication: 4
       type: tls
       certificateAndKey:
         secretName: my-secret
         certificate: public.crt
         key: private.key
     http: 5
       port: 8080
       cors: 6
         allowedOrigins: "https://strimzi.io"
         allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
     consumer: 7
       config:
         auto.offset.reset: earliest
     producer: 8
       config:
         delivery.timeout.ms: 300000
     resources: 9
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 10
       type: inline
       loggers:
         logger.bridge.level: "INFO"
         # enabling DEBUG just for send operation
         logger.send.name: "http.openapi.operation.send"
         logger.send.level: "DEBUG"
     jvmOptions: 11
       "-Xmx": "1g"
       "-Xms": "1g"
     readinessProbe: 12
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     image: my-org/my-image:latest 13
     template: 14
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       bridgeContainer: 15
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   レプリカノードの数。

   2

   ターゲット Kafka クラスターに接続するための ブートストラップサーバー。

   3

   ソース Kafka クラスターの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   4

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、Kafka Bridge クラスターの認証。デフォルトでは、Kafka Bridge は認証なしで Kafka ブローカーに接続します。

   5

   Kafka ブローカーへの HTTP アクセス。

   6

   選択されたリソースおよびアクセスメソッドを指定する CORS アクセス。リクエストの追加の HTTP ヘッダーには Kafka クラスターへのアクセスが許可されるオリジンが記述されています。

   7

   コンシューマー設定 オプション。

   8

   プロデューサー設定 オプション。

   9

   サポートされているリソース（現在は cpu と memory）の予約と、消費可能な最大リソースを指定するための制限を要求します。

   10

   指定された Kafka Bridge loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。Kafka Bridge ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   11

   Kafka Bridge を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   12

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するためのヘルスチェック。

   13

   高度なオプション：特別な場合のみ 推奨される コンテナーイメージの設定。

   14

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   15

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

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

   oc apply -f KAFKA-BRIDGE-CONFIG-FILE

手順

1. ConfigMap を作成します。

   ConfigMap を YAML ファイルとして作成するか、コマンドラインで oc を使用してプロパティーファイルから Config Map を作成します。

   Kafka のルートロガー定義が含まれる ConfigMap の例:

   kind: ConfigMap
   apiVersion: kafka.strimzi.io/v1beta1
   metadata:
     name: logging-configmap
   data:
     log4j.properties:
       kafka.root.logger.level="INFO"

   プロパティーファイルを使用してコマンドラインから作成します。

   oc create configmap logging-configmap --from-file=log4j.properties

   プロパティーファイルではロギング設定が定義されます。

   # Define the logger
   kafka.root.logger.level="INFO"
   # ...

2. logging.name を ConfigMap の名前に設定し、リソースの spec の external ロギングを定義します。

   spec:
     # ...
     logging:
       type: external
       name: logging-configmap

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

   oc apply -f kafka.yaml

1. 相互 TLS 認証 (TLS による暗号化が有効なリスナーのみ)
2. SCRAM-SHA-512 認証
3. OAuth 2.0 のトークンベースの認証

手順

1. Kafka リソースを設定します。

   1. 承認には authorization プロパティーを設定します。

   2. listeners プロパティーを設定し、認証でリスナーを作成します。

      以下は例になります。

      apiVersion: kafka.strimzi.io/v1beta1
      kind: Kafka
      spec:
        kafka:
          # ...
          authorization: 1
            type: simple
            superUsers: 2
              - CN=client_1
              - user_2
              - CN=client_3
          listeners:
            - name: tls
              port: 9093
              type: internal
              tls: true
              authentication:
                type: tls 3
          # ...
        zookeeper:
          # ...

      1

      Authorizationは、AclAuthorizer Kafka プラグインを使用して、Kafka ブローカーでのsimpleな承認を可能にします。

      2

      Kafka へのアクセスを制限されないユーザープリンシパルのリスト。CN は、TLS による認証が使用される場合のクライアント証明書の共通名です。

      3

      リスナーの認証メカニズムは各リスナーに対して設定でき、相互 TLS、SCRAM-SHA-512、またはトークンベース OAuth 2.0 として指定 できます。

      外部リスナーを設定している場合、設定は選択した接続のメカニズムによって異なります。

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

   oc apply -f KAFKA-CONFIG-FILE

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

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

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

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

手順

1. KafkaUser リソースを設定します。

   以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     authentication: 1
       type: tls
     authorization:
       type: simple 2
       acls:
         - 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

   相互 tls または scram-sha-512 として定義されたユーザー認証メカニズム。

   2

   ACL ルールのリストが必要な簡易承認。

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

   oc apply -f USER-CONFIG-FILE

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

手順

1. Kafka リソースを開きます。

2. networkPolicyPeers プロパティーで、Kafka クラスターへのアクセスが許可されるアプリケーション Pod または namespace を定義します。

   以下は、ラベル app が kafka-client に設定されているアプリケーションからの接続のみを許可するよう tls リスナーを設定する例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
           networkPolicyPeers:
             - podSelector:
                 matchLabels:
                   app: kafka-client
       # ...
     zookeeper:
       # ...

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

   次のように oc apply を使用します。

   oc apply -f your-file

1. 承認サーバーをデプロイし、AMQ Streams と統合するためにそのデプロイメントを設定します。
2. OAuth 2.0 を使用するよう設定された Kafka ブローカーリスナーで Kafka クラスターをデプロイまたは更新します。
3. OAuth 2.0 を使用するように Java ベースの Kafka クライアントを更新します。
4. OAuth 2.0 を使用するように Kafka コンポーネントクライアントを更新します。

手順

1. Red Hat Single Sign-On の Admin Console にアクセスするか、Red Hat Single Sign-On の Admin CLI を使用して、OAuth 2.0 認証の設定時に作成した Kafka ブローカークライアントの Authorization Services を有効にします。
2. 承認サービスを使用して、クライアントのリソース、承認スコープ、ポリシー、およびパーミッションを定義します。
3. ロールとグループをユーザーとクライアントに割り当てて、パーミッションをユーザーとクライアントにバインドします。

4. エディターで Kafka リソースの Kafka ブローカー設定 (Kafka.spec.kafka) を更新して、Kafka ブローカーで Red Hat Single Sign-On による承認が使用されるように設定します。

   oc edit kafka my-cluster

5. Kafka ブローカーの kafka 設定を指定して、keycloak による承認を使用し、承認サーバーと Red Hat Single Sign-On の Authorization Services にアクセスできるようにします。

   以下は例になります。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka
     # ...
     authorization:
       type: keycloak 1
       tokenEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token> 2
       clientId: kafka 3
       delegateToKafkaAcls: false 4
       disableTlsHostnameVerification: false 5
       superUsers: 6
         - CN=fred
         - sam
         - CN=edward
       tlsTrustedCertificates: 7
       - secretName: oauth-server-cert
         certificate: ca.crt
       grantsRefreshPeriodSeconds: 60 8
       grantsRefreshPoolSize: 5 9
     #...

   1

   タイプ keycloak によって Red Hat Single Sign-On の承認が有効になります。

   2

   Red Hat Single Sign-On トークンエンドポイントの URI。本番環境では常に HTTP を使用してください。

   3

   承認サービスが有効になっている Red Hat Single Sign-On の OAuth 2.0 クライアント定義のクライアント ID。通常、kafka が ID として使用されます。

   4

   (オプション) Red Hat Single Sign-On Authorization Services ポリシーでアクセスが拒否された場合、Kafka AclAuthorizer に権限を委譲します。デフォルトは false です。

   5

   (任意設定): TLS ホスト名の検証を無効にします。デフォルトは false です。

   6

   (任意設定): 指定の スーパーユーザー。

   7

   (任意設定): 承認サーバーへの TLS 接続用の信用できる証明書。

   8

   (任意設定): 連続する付与 (Grants) 更新実行の間隔。これは、アクティブなセッションが Red Hat Single Sign-On でユーザーのパーミッション変更を検出する最大時間です。デフォルト値は 60 です。

   9

   (任意設定): アクティブなセッションの付与 (Grants) の更新 (並行して) に使用するスレッドの数。デフォルト値は 5 です。

6. エディターを保存して終了し、ローリングアップデートの完了を待ちます。

7. 更新をログで確認するか、または Pod 状態の遷移を監視して確認します。

   oc logs -f ${POD_NAME} -c kafka
   oc get po -w

   ローリングアップデートによって、ブローカーが OAuth 2.0 承認を使用するように設定されます。

8. クライアントまたは特定のロールを持つユーザーとして Kafka ブローカーにアクセスして、設定したパーミッションを検証し、必要なアクセス権限があり、付与されるべきでないアクセス権限がないことを確認します。