Confluent Operator でのストレージの構成¶
ストレージのオプション¶
Operator ではデフォルトで、ストレージ管理に Kubernetes が提供する動的ストレージプロビジョニングが使用されています。
静的にプロビジョニングされたストレージボリュームの利用が必要な場合は、ストレージを手動でプロビジョニングして Kubernetes ワーカーノードに接続し、それをプラットフォームに PersistentVolume として公開したうえで、Confluent Operator を使用して Confluent Platform クラスターをデプロイすることで、ブローカーインスタンスによって PersistentVolume がマウントされます。
Confluent Operator に対するストレージ要件の指定方法に応じて、以下のオプションがあります。
- StorageClass を作成し、Confluent Operator で使用されるクラス名を指定する
- デフォルトの Kubernetes StorageClass を使用する
- Confluent Operator Helm チャートにストレージプロビジョナーなどの詳細を指定して StorageClass を作成する
Confluent Operator では、ストレージクラス間での移行はサポートされていません。
定義済み StorageClass の使用¶
Confluent Operator 5.5 以降、作成されるすべての PersistentVolume で特定の StorageClass を使用するよう Operator に指示できます。
Confluent Platform 全体で使用するストレージクラスの指定、または ZooKeeper、Kafka、ksqlDB、Control Center などのコンポーネントごとに異なるストレージクラスの指定ができます。
Kubernetes クラスターで使用する定義済み StorageClass を作成または使用します。
Confluent Operator で使用するストレージクラスで
volumeBindingModeがWaitForFirstConsumerに設定されていることを確認してください。詳細については、Kubernetes ストレージクラスのボリュームバインディングモード の説明を参照してください。既存の StorageClass を使用しないで新たに作成する場合は、Kubernetes クラスターにおいて StorageClass を作成および変更できるアクセス許可が必要です。
構成ファイル（
$VALUES_FILE）で、Confluent Platform のデプロイに使用する StorageClass の名前を指定します。デプロイされた全コンポーネントで使用する StorageClass を指定するには、ストレージクラス名を
global.storageClassNameに指定します。global: storageClassName:
特定のコンポーネントで使用する StorageClass を指定するには、ストレージクラス名を
<component>.storageClassNameに指定します。Kafka の場合の例を以下に示します。kafka: storageClassName:
Kubernetes のデフォルト StorageClass の使用¶
Confluent Operator 5.5 以降、Kubernetes のデフォルトストレージクラスを使用するように Confluent Operator を構成できます。
このオプションを使用するには、volumeBindingMode が WaitForFirstConsumer に設定されていることを確認してください。詳細については、Kubernetes ストレージクラスのボリュームバインディングモード の説明を参照してください。
静的にプロビジョニングされたストレージを使用するための手順は、上記と同じです。PersistentVolume の定義に指定されている storageClassName が Kubernetes クラスターのデフォルト StorageClass と一致していることを確認してください。
Kubernetes のデフォルトストレージクラスを使用するには、構成ファイル（ $VALUES_FILE ）を以下のようにします。
- グローバルレベルの
storageClassName値を指定しません。または、空の文字列（""）を設定します。 - コンポーネントレベルの
storageClassName値を指定しません。または、空の文字列（""）を設定します。 global.provider.storageオブジェクトを指定しません。
関連付けられたボリュームでは、Kubernetes クラスターのデフォルト StorageClass が使用されます。バージョン 1.11 以降の Kubernetes では、デフォルト StorageClass のサポートがデフォルトで有効になっています。
Confluent Operator Helm チャートによって作成された StorageClass の使用¶
Confluent Platform クラスター用のストレージクラスを Confluent Operator Helm チャートで作成するには、構成ファイル（ $VALUES_FILE ）で以下を設定します。
global:
provider:
kubernetes:
deployment:
zones:
- us-central1-a
storageClassName:"" ----- [1]
storage: ----- [2]
provisioner: kubernetes.io/gce-pd
allowVolumeExpansion: ----- [3]
parameters:
type: pd-ssd
[1]
storageClassName:""を設定します。または、グローバルレベルかコンポーネントレベルでstorageClassNameを指定しません。[2] マルチゾーンクラスターを構成している場合は、Confluent Operator によって
global.provider.kubernetes.deployment.zoneに指定されているゾーンごとにストレージクラスが作成されます。構成例については、ストレージクラスのプロビジョナー の説明を参照してください。この例では、GCE 永続ディスクストレージ（
gce-pd）と SSD（pd-ssd）を、GCE PD の説明に従って使用します。[3]
allowVolumeExpansion: falseを設定して、自動ストレージ拡張を無効にします。Confluent Operator では、Confluent Operator Helm チャートによって作成されたストレージクラスについて、ストレージの自動拡張がデフォルトでサポートされます。
Confluent Platform クラスターの作成時には、Confluent Operator によって複数の StorageClass が、global.provider.storage のデータを各 StorageClass の仕様として用いてオンザフライで作成されます。それぞれ以下のパターンに従って命名されます。
{cp-component-helm-chart-name}-standard-ssd-{zone}
こうする場合、静的にプロビジョニングされたストレージを使用するための手順は前述と同じです。
ストレージオプションの優先順位¶
可能なストレージ構成オプションについての優先ルールは以下のとおりです。
storageClassNameがグローバルレベル（globalオブジェクトで）とコンポーネントレベル（コンポーネントオブジェクトで）の両方で指定されている場合は、コンポーネントレベルのstorageClassNameが使用されます。storageClassNameがコンポーネントレベルで指定されており、global.provider.storageオブジェクトが指定されている場合は、コンポーネントレベルのstorageClassNameが使用されます。- グローバルレベルの
storageClassNameとglobal.provider.storageとがどちらも指定されている場合は、Operator からエラーが返されます。
階層型ストレージ¶
Confluent Platform 6.0 以降、Operator では AWS S3 または Google GCS を使用した 階層型ストレージ を有効にできます。
階層型ストレージの有効化¶
階層型ストレージを有効にする場合は、Kafka で以下を構成する必要があります。
- 使用する Blob ストレージの種類。現状では、AWS S3 と GCS がサポートされています。
- 使用するストレージバケットの名前。このバケットはあらかじめ作成しておく必要があります。このバケットがユーザーに代わって Operator によって作成されることはありません。
また、Kafka ブローカーにストレージバケットへの適切なアクセス許可があることを確認する必要があります。以下のいずれかのオプションを使用します。
サービスアカウント を使用する
クラウド IAM へのアクセス許可を、Kafka ブローカーポッドに関連付けられている Kubernetes ServiceAccount にマップできます。
AWS には、AWS IAM へのアクセス許可を EKS の ServiceAccount にネイティブで関連付ける機能が用意されています。
同様に、GCP には、IAM へのアクセス許可を GKE の ServiceAccount にネイティブでマップする機能が用意されています。
適切なバケットのアクセス許可は、Kafka のデプロイ先となる Kubernetes 名前空間のデフォルトの ServiceAccount にマップできます。または、別個の ServiceAccount にマップし、Operator を使用して Kafka ブローカーポッドがその ServiceAccount と関連付けられていることを確認することもできます。このアプローチの主な利点は、Confluent Platform を Operator からデプロイする際に、バケットにアクセスするための機密の認証情報を具体的に管理する必要がないことです。
EKS におけるサービスアカウントの AWS IAM ロールへの関連付けについては、「サービスアカウントの IAM ロール」を参照してください。
GKE におけるサービスアカウントの GCP IAM ロールへの関連付けについては、「Workload Identity の使用」を参照してください。
Operator によって管理されている Confluent Platform コンポーネントと関連付ける Kubernetes サービスアカウントの構成については、「サービスアカウントの用意」を参照してください。
シークレット オブジェクトの使用
AWS または GCP の認証情報をシークレットオブジェクトに指定し、Operator で Kafka をデプロイする際にそのオブジェクトの認証情報を使用するように Kafka を構成できます。
上記の必須設定に加えて、他の階層型ストレージ設定を kafka セクションの configOverrides 設定を使用して構成できます。可能な設定については、「階層型ストレージ」を参照してください。
Kafka クラスターが削除されると、Operator による階層型ストレージバケットのコンテンツに対するガベージコレクションは行われません。設定間隔だけ待機するか、階層型ストレージバケットでオブジェクトを手動で削除します。詳細については、「トピック削除の間隔」を参照してください。
AWS S3 の階層型ストレージの構成¶
AWS S3 で階層型ストレージを有効にして構成するには、構成ファイル（ $VALUES_FILE ）で Kafka に対する以下の構成オーバーライドを設定します。
kafka:
configOverrides:
server:
- confluent.tier.feature=true ----- [1]
- confluent.tier.enable=true ----- [2]
- inter.broker.protocol.version= ----- [3]
- confluent.tier.backend=S3 ----- [4]
- confluent.tier.s3.bucket= ----- [5]
- confluent.tier.s3.region= ----- [6]
- confluent.tier.s3.cred.file.path= ----- [7]
- confluent.tier.topic.delete.check.interval.ms -- [8]
mountedSecrets: ----- [9]
- secretRef:
keyItems:
- key:
path:
[1]
confluent.tier.feature=trueと設定して、階層型ストレージを有効にします。[2]
confluent.tier.enableを、作成されたトピックのデフォルト値に指定します。trueに設定すると、未圧縮のトピックがすべて階層型になります。[3]
inter.broker.protocol.versionを2.4以上に設定します。2.4 が、階層型ストレージをサポートしている Kafka の最小バージョンです。[4]
confluent.tier.backendをS3に設定します。[5]
confluent.tier.s3.bucketを、使用する S3 バケットに設定します。[6]
confluent.tier.s3.regionをリージョンに設定します。[7]（省略可能）階層型ストレージの認証情報の提供に シークレット を使用する場合は、
confluent.tier.s3.cred.file.pathを設定します。サービスアカウントを使用する場合、このプロパティは不要です。シークレットのファイルに追加する必要のある内容については、「階層型ストレージ」を参照してください。
[8]（省略可能）
confluent.tier.topic.delete.check.interval.msに、トピックまたはクラスターが削除されてからログセグメントファイルが削除されるまでの間隔を設定します。デフォルト値は 3 時間です。[9]（省略可能）階層型ストレージの認証情報の提供にシークレットを使用する場合に限って必要です。マウントされたシークレットの構成に使用できるオプションについては、「マウントされたシークレットの提供」を参照してください。
以下に例を示します。
kafka:
configOverrides:
server:
- confluent.tier.feature=true
- confluent.tier.enable=true
- inter.broker.protocol.version=2.6
- confluent.tier.backend=S3
- confluent.tier.s3.bucket=my-bucket
- confluent.tier.s3.region=us-west-2
- confluent.tier.s3.cred.file.path=/mnt/secrets/my-secret-aws/aws/creds
mountedSecrets:
- secretRef: my-secret-aws
keyItems:
- key: credentials
path: aws/creds
GCS の階層型ストレージの構成¶
GCS で階層型ストレージを有効にして構成するには、構成ファイル（ $VALUES_FILE ）で Kafka に対する以下の構成オーバーライドを設定します。
kafka:
configOverrides:
server:
- confluent.tier.feature=true ----- [1]
- confluent.tier.enable=true ----- [2]
- inter.broker.protocol.version= ----- [3]
- confluent.tier.backend=GCS ----- [4]
- confluent.tier.gcs.bucket= ----- [5]
- confluent.tier.gcs.region= ----- [6]
- confluent.tier.gcs.cred.file.path= ----- [7]
- confluent.tier.topic.delete.check.interval.ms -- [8]
mountedSecrets: ----- [9]
- secretRef:
keyItems:
- key:
path:
[1]
confluent.tier.feature=trueと設定して、階層型ストレージを有効にします。[2]
confluent.tier.enableを、作成されたトピックのデフォルト値に指定します。trueに設定すると、未圧縮のトピックがすべて階層型になります。[3]
inter.broker.protocol.versionを2.4以上に設定します。2.4 が、階層型ストレージをサポートしている Kafka の最小バージョンです。[4]
confluent.tier.backendをGCSに設定します。[5]
confluent.tier.gcs.bucketを、使用する GCS バケットに設定します。[6]
confluent.tier.gcs.regionを GCS リージョンに設定します。[7]（省略可能）階層型ストレージ用に シークレット を使用する場合は、
confluent.tier.gcs.cred.file.pathを設定します。サービスアカウントを使用する場合、このプロパティは不要です。シークレットのファイルに追加する必要のある内容については、「階層型ストレージ」を参照してください。
[8]（省略可能）
confluent.tier.topic.delete.check.interval.msに、トピックまたはクラスターが削除されてからログセグメントファイルが削除されるまでの間隔を設定します。デフォルト値は 3 時間です。[9]（省略可能）階層型ストレージの認証情報の提供にシークレットを使用する場合に限って必要です。マウントされたシークレットの構成に使用できるオプションについては、「マウントされたシークレットの提供」を参照してください。
以下に例を示します。
kafka:
configOverrides:
server:
- confluent.tier.feature=true
- confluent.tier.enable=true
- inter.broker.protocol.version=2.6
- confluent.tier.backend=GCS
- confluent.tier.gcs.bucket=my-bucket
- confluent.tier.gcs.region=us-central1
- confluent.tier.gcs.cred.file.path=/mnt/secrets/my-secret-gcs/credentials
mountedSecrets:
- secretRef: my-secret-gcs