Strimzi を使用して Apache Kafka を GKE にデプロイする

環境を準備する

このチュートリアルでは、Cloud Shell を使用して Google Cloud でホストされるリソースを管理します。Cloud Shell には、このチュートリアルに必要なソフトウェア（kubectl、gcloud CLI、Helm、Terraform など）がプリインストールされています

Cloud Shell を使用して環境を設定するには、次の操作を行います。

1. Google Cloud コンソールで 「Cloud Shell をアクティブにする」をクリックして、Google Cloud コンソールから Cloud Shell セッションを起動します。Google Cloud コンソールの下部ペインでセッションが起動します。

2. 環境変数を設定します。

   export PROJECT_ID=PROJECT_ID
   export KUBERNETES_CLUSTER_PREFIX=kafka
   export REGION=us-central1
   

   PROJECT_ID の置き換え: Google Cloud のプロジェクト ID に置き換えます。

3. GitHub リポジトリのクローンを作成します。

   git clone https://github.com/GoogleCloudPlatform/kubernetes-engine-samples
   

4. 作業ディレクトリを変更します。

   cd kubernetes-engine-samples/streaming/
   

クラスタ インフラストラクチャを作成する

このセクションでは、Terraform スクリプトを実行して、限定公開の高可用性リージョン GKE クラスタを作成します。次の手順では、コントロール プレーンへの公開アクセスを許可します。アクセスを制限するため、限定公開クラスタを作成します。

オペレーターは、Standard または Autopilot クラスタを使用してインストールできます。

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-standard init
terraform -chdir=kafka/terraform/gke-standard apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

...
Apply complete! Resources: 14 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials strimzi-cluster --region us-central1"

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-autopilot init
terraform -chdir=kafka/terraform/gke-autopilot apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

...
Apply complete! Resources: 12 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials strimzi-cluster --region us-central1"

クラスタへの接続

クラスタと通信を行うように kubectl を構成します。

gcloud container clusters get-credentials ${KUBERNETES_CLUSTER_PREFIX}-cluster --region ${REGION}

Strimzi オペレーターをクラスタにデプロイする

このセクションでは、Helm チャートを使用して Strimzi オペレーターをデプロイします。Strimzi をデプロイする方法は他にもいくつかあります。

1. Strimzi Helm チャート リポジトリを追加します。

   helm repo add strimzi https://strimzi.io/charts/
   

2. Strimzi オペレーターと Kafka クラスタの Namespace を追加します。

   kubectl create ns kafka
   

3. Helm を使用して Strimzi クラスタ オペレーターをデプロイします。

   helm install strimzi-operator strimzi/strimzi-kafka-operator -n kafka
   

   Strimzi クラスタ オペレーターと Kafka クラスタを異なる Namespace にデプロイする場合は、Helm コマンドに --set watchNamespaces="{kafka-namespace,kafka-namespace-2,...}" パラメータを追加します。

4. Helm を使用して、Strimzi クラスタ オペレーターが正常にデプロイされたことを確認します。

   helm ls -n kafka
   

   出力は次のようになります。

   NAME            NAMESPACE    REVISION    UPDATED                              STATUS    CHART                        APP VERSION
   strimzi-operator    kafka      1       2023-06-27 11:22:15.850545 +0200 CEST    deployed    strimzi-kafka-operator-0.35.0    0.35.0
   

Kafka をデプロイする

オペレーターがクラスタにデプロイされると、Kafka クラスタ インスタンスをデプロイする準備が整います。

このセクションでは、Kafka を基本構成にデプロイしてから、可用性、セキュリティ、オブザーバビリティの要件を満たすため、さまざまな高度な構成シナリオを試してみます。

基本構成

Kafka インスタンスの基本構成には、次のコンポーネントが含まれています。

• Kafka ブローカーの 3 つのレプリカ。クラスタの整合性を確保するため、利用可能なレプリカが少なくとも 2 つ必要です。
• クラスタを形成する ZooKeeper ノードの 3 つのレプリカ。
• 2 つの Kafka リスナー: 1 つは認証を利用しません。もう 1 つは Strimzi によって生成された証明書を使用して TLS 認証を利用します。
• Kafka では Java の MaxHeapSize と MinHeapSize が 4 GB に設定され、ZooKeeper では 2 GB に設定されています。
• 1 CPU リクエストの CPU リソースの割り当ては、Kafka、ZooKeeper ともに 2 CPU に制限されています。メモリ リクエストの割り当ては、Kafka の場合 5 GB で、メインサービスに 4 GB、指標エクスポータに 0.5 GB という制限があります。ZooKeeper の場合は 2.5 GB で、メインサービスに 2 GB、指標エクスポータに 0.5 GB という制限があります。
• 次のリクエストと上限を持つ entity-operator。
  • tlsSidecar: 100 m / 500 m CPU と 128 Mi メモリ。
  • topicOperator: 100 m / 500 m CPU と 512 Mi メモリ。
  • userOperator: 500 m CPU と 2 Gi メモリ。
• premium-rwo storageClass を使用して各 Pod に 100 GB のストレージが割り当てられます。
• 各ワークロードに構成された toleration、nodeAffinities、podAntiAffinities。それぞれのノードプールと異なるゾーンを使用して、ノード間で適切に分散されます。
• 自己署名証明書で保護されているクラスタ内の通信: クラスタとクライアント（mTLS）で個別の認証局（CA）。別の認証局を使用するように構成することもできます。

この構成は、本番環境に対応した Kafka クラスタの作成に必要な最小限の設定を表しています。以降のセクションでは、クラスタ セキュリティ、アクセス制御リスト（ACL）、トピック管理、証明書管理などに対処するためのカスタム構成を示します。

基本的な Kafka クラスタを作成する

1. 基本構成を使用して新しい Kafka クラスタを作成します。

   kubectl apply -n kafka -f kafka-strimzi/manifests/01-basic-cluster/my-cluster.yaml
   

   このコマンドは Strimzi オペレーターの Kafka カスタム リソースを作成します。これにより、CPU とメモリのリクエストと上限、ブロック ストレージ リクエスト、プロビジョニングされた Pod を Kubernetes ノード間で分散するための taint とアフィニティの組み合わせが含まれます。

2. Kubernetes が必要なワークロードを開始するまで数分待ちます。

   kubectl wait kafka/my-cluster --for=condition=Ready --timeout=600s -n kafka
   

3. Kafka ワークロードが作成されたことを確認します。

   kubectl get pod,service,deploy,pdb -l=strimzi.io/cluster=my-cluster -n kafka
   

   出力は次のようになります。

   NAME                                            READY   STATUS  RESTARTS   AGE
   pod/my-cluster-entity-operator-848698874f-j5m7f   3/3   Running   0        44m
   pod/my-cluster-kafka-0                          1/1   Running   0        5m
   pod/my-cluster-kafka-1                          1/1   Running   0        5m
   pod/my-cluster-kafka-2                          1/1   Running   0        5m
   pod/my-cluster-zookeeper-0                      1/1   Running   0        6m
   pod/my-cluster-zookeeper-1                      1/1   Running   0        6m
   pod/my-cluster-zookeeper-2                      1/1   Running   0        6m
   
   NAME                                TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                             AGE
   service/my-cluster-kafka-bootstrap  ClusterIP   10.52.8.80   <none>      9091/TCP,9092/TCP,9093/TCP          5m
   service/my-cluster-kafka-brokers    ClusterIP   None         <none>      9090/TCP,9091/TCP,9092/TCP,9093/TCP   5m
   service/my-cluster-zookeeper-client   ClusterIP   10.52.11.144   <none>      2181/TCP                            6m
   service/my-cluster-zookeeper-nodes  ClusterIP   None         <none>      2181/TCP,2888/TCP,3888/TCP          6m
   
   NAME                                       READY   UP-TO-DATE   AVAILABLE   AGE
   deployment.apps/my-cluster-entity-operator   1/1   1          1         44m
   
   NAME                                            MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
   poddisruptionbudget.policy/my-cluster-kafka     2             N/A             1                   5m
   poddisruptionbudget.policy/my-cluster-zookeeper   2             N/A             1                   6m
   

オペレーターが次のリソースを作成します。

• Kafka と ZooKeeper に 2 つの StrimziPodSets。
• Kafka ブローカー レプリカ用に 3 つの Pod。
• ZooKeeper レプリカ用に 3 つの Pod。
• 2 つの PodDisruptionBudgets。クラスタの整合性のため、少なくとも 2 つのレプリカの可用性を確保します。
• my-cluster-kafka-bootstrap という名前の Service。Kubernetes クラスタ内から接続する Kafka クライアントのブートストラップ サーバーとして機能します。この Service では、すべての内部 Kafka リスナーを使用できます。
• my-cluster-kafka-brokers という名前のヘッドレス Service。Kafka ブローカー Pod IP アドレスの DNS 解決を直接有効にします。この Service はブローカー間の通信に使用されます。
• my-cluster-zookeeper-client という名前の Service。Kafka ブローカーがクライアントとして ZooKeeper ノードに接続できるようにします。
• ZooKeeper Pod IP アドレスの DNS 解決を直接行う my-cluster-zookeeper-nodes という名前のヘッドレス Service。この Service は ZooKeeper レプリカ間の接続に使用されます。
• my-cluster-entity-operator という名前の Deployment。topic-operator と user-operator を含み、カスタム リソースである KafkaTopics と KafkaUsers の簡単を容易にします。

2 つの NetworkPolicies を構成して、任意の Pod と Namespace から Kafka リスナーへの接続を容易にすることもできます。また、これらのポリシーは、ZooKeeper への接続をブローカーに制限し、クラスタ Pod とクラスタ通信専用の内部 Service ポート間の通信を有効にします。

認証とユーザー管理

このセクションでは、Kafka リスナーを保護し、クライアントと認証情報を共有するために認証と認可を有効にする方法について説明します。

Strimzi では、Kubernetes ネイティブの方法でユーザー管理を行います。個別の User Operator とそれに対応し、ユーザー構成が定義されている Kubernetes カスタム リソース（KafkaUser）を使用します。ユーザー構成には認証と認可の設定が含まれており、対応するユーザーを Kafka にプロビジョニングします。

Strimzi では、ユーザー名とパスワードに基づく認証（SCRAM-SHA-512）や TLS など、複数の認証メカニズムをサポートする Kafka リスナーとユーザーを作成できます。また、OAuth 2.0 認証も使用できます。この認証方法は、セキュリティと外部で認証情報を管理するという点で、パスワードや証明書を使用する方法よりも優れていると考えられています。

Kafka クラスタをデプロイする

このセクションでは、ユーザー管理機能を実行する Strimzi オペレーターのデプロイ方法を示します。ここでは、次のものをデプロイします。

• いずれかのリスナーでパスワード ベースの認証（SCRAM-SHA-512）が有効になっている Kafka クラスタ。
• 3 つのレプリカを持つ KafkaTopic。
• トピックに対するユーザーの読み取り権限と書き込み権限を指定する ACL を持つ KafkaUser。

1. パスワード ベースの SCRAM-SHA-512 認証と単純な認証を行うリスナーをポート 9094 で使用するように、Kafka クラスタを構成します。

   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-cluster.yaml
   

2. Topic、User、Kafka クラスタにコマンドを実行するクライアント Pod を作成します。

   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/topic.yaml
   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-user.yaml
   

   ユーザー認証情報を含む Secret my-user は、Volume としてクライアント Pod にマウントされます。

   これらの認証情報により、パスワード ベースの認証（SCRAM-SHA-512）を有効にしたリスナーを使用してトピックにメッセージをパブリッシュする権限がユーザーにあることが確認されます。

3. クライアント Pod を作成します。

   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/kafkacat.yaml
   

4. クライアント Pod が Ready になるまで数分待ってから接続します。

   kubectl wait --for=condition=Ready pod --all -n kafka --timeout=600s
   kubectl exec -it kafkacat -n kafka -- /bin/sh
   

5. my-user 認証情報を使用して新しいメッセージを生成し、使用します。

   echo "Message from my-user" |kcat \
     -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=SCRAM-SHA-512 \
     -X sasl.username=my-user \
     -X sasl.password=$(cat /my-user/password) \
     -t my-topic -P
   kcat -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=SCRAM-SHA-512 \
     -X sasl.username=my-user \
     -X sasl.password=$(cat /my-user/password) \
     -t my-topic -C
   

   出力は次のようになります。

   Message from my-user
   % Reached end of topic my-topic [0] at offset 0
   % Reached end of topic my-topic [2] at offset 1
   % Reached end of topic my-topic [1] at offset 0
   

   「CTRL+C」と入力して、コンシューマー プロセスを停止します。

6. Pod のシェルを終了します。

   exit
   

バックアップと障害復旧

Strimzi オペレーターにはバックアップ機能が組み込まれていませんが、特定のパターンに従うことで効率的なバックアップ戦略を実装できます。

Backup for GKE を使用して、次のものをバックアップできます。

• Kubernetes リソース マニフェスト。
• バックアップを行うクラスタの Kubernetes API サーバーから抽出された Strimzi API カスタム リソースとその定義。
• マニフェスト内の PersistentVolumeClaim リソースに対応する Volume。

Backup for GKE を使用して Kafka クラスタのバックアップと復元を行う方法については、障害復旧の準備を行うをご覧ください。

Strimzi オペレーターを使用して、デプロイされた Kafka クラスタのバックアップを実行することもできます。次のデータをバックアップする必要があります。

• Kafka 構成。KafkaTopics や KafkaUsers など、Strimzi API のすべてのカスタム リソースが含まれます。
• Kafka ブローカーの PersistentVolume に保存されているデータ。

Strimzi の構成を含む Kubernetes リソース マニフェストを Git リポジトリに保存すると、必要に応じてリソースを新しい Kubernetes クラスタに再適用できるため、Kafka 構成のバックアップを別途行う必要がなくなります。

Kafka サーバー インスタンスまたは Kafka がデプロイされた Kubernetes クラスタが失われた場合に Kafka のデータを復旧できるようにするため、reclaimPolicy オプションを Retain に設定して、Kafka ブローカーのボリュームのプロビジョニングに使用される Kubernetes ストレージ クラスを構成することをおすすめします。また、Kafka ブローカー ボリュームのスナップショットを取得することもおすすめします。

次のマニフェストでは、reclaimPolicy オプション Retain を使用する StorageClass を記述しています。

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: premium-rwo-retain
...
reclaimPolicy: Retain
volumeBindingMode: WaitForFirstConsumer

次の例は、Kafka クラスタのカスタム リソースの spec に追加された StorageClass を示しています。

# ...
spec:
  kafka:
    # ...
    storage:
      type: persistent-claim
      size: 100Gi
      class: premium-rwo-retain

この構成では、対応する PersistentVolumeClaim が削除されても、ストレージ クラスを使用してプロビジョニングされた PersistentVolume は削除されません。

既存の構成とブローカー インスタンス データを使用して、新しい Kubernetes クラスタに Kafka インスタンスを復元するには:

1. 既存の Strimzi Kafka カスタム リソース（Kakfa、KafkaTopic、KafkaUser など）を新しい Kubernetes クラスタに適用します。
2. PersistentVolumeClaim の spec.volumeName プロパティを使用して、新しい Kafka ブローカー インスタンスの名前を持つ PersistentVolumeClaim を古い PersistentVolume に戻します。