OpenShift での AMQ Streams のデプロイおよびアップグレード
OpenShift Container Platform での AMQ Streams 2.1 のデプロイ
概要
多様性を受け入れるオープンソースの強化
Red Hat では、コード、ドキュメント、Web プロパティーにおける配慮に欠ける用語の置き換えに取り組んでいます。まずは、マスター (master)、スレーブ (slave)、ブラックリスト (blacklist)、ホワイトリスト (whitelist) の 4 つの用語の置き換えから始めます。これは大規模な取り組みであるため、これらの変更は今後の複数のリリースで段階的に実施されます。詳細は、Red Hat CTO である Chris Wright のメッセージをご覧ください。
第1章 デプロイメントの概要
AMQ Streams は、OpenShift クラスターで Apache Kafka を実行するプロセスを簡素化します。
本ガイドでは、AMQ Streams をデプロイおよびアップグレードするすべての方法の手順を取り上げ、デプロイメントの対象や、OpenShift クラスターで Apache Kafka を実行するために必要なデプロイメントの順序について説明します。
デプロイメントの手順を説明する他に、デプロイメントを準備および検証するためのデプロイメントの前および後の手順についても説明します。本ガイドでは、メトリクスを導入するための追加のデプロイメントオプションについても説明しています。
アップグレードの手順は、「AMQ Streams および Kafka のアップグレード」を参照してください。
AMQ Streams は、パブリックおよびプライベートクラウドからデプロイメントを目的とするローカルデプロイメントまで、ディストリビューションに関係なくすべてのタイプの OpenShift クラスターで動作するよう設計されています。
1.1. デプロイメントの設定
本ガイドのデプロイメント手順は、デプロイメントの初期構造の設定に役立つように設計されています。構造の設定後、カスタムリソースを使用して、正確なニーズに合わせてデプロイメントを設定できます。デプロイメント手順では、AMQ Streams に同梱されているインストールファイルのサンプルを使用します。手順では、検討する必要がある重要な設定事項について説明しますが、使用できる設定オプションをすべて取り上げるわけではありません。
AMQ Streams をデプロイする前に、Kafka コンポーネントに使用できる設定オプションを確認できます。設定オプションの詳細は、Configuring AMQ Streams on OpenShiftを参照してください。
1.1.1. Kafka のセキュリティー
デプロイメントでは、Cluster Operator はクラスター内でのデータの暗号化および認証に対して自動で TLS 証明書を設定します。
AMQ Streams では、暗号化、認証、および 承認 の追加の設定オプションが提供されます。
- Kafka へのセキュアなアクセスを管理して、Kafka クラスターとクライアント間のデータ交換をセキュアにします。
- 承認サーバーが OAuth 2.0 認証 および OAuth 2.0 承認 を使用するように、デプロイメントを設定します。
- 独自の証明書を使用して Kafka をセキュアにします。
1.1.2. デプロイメントの監視
AMQ Streams は、デプロイメントを監視する追加のデプロイメントオプションをサポートします。
- Prometheus および Grafana を Kafka クラスターでデプロイし、メトリクスを抽出して、Kafka コンポーネントを監視します。
- Kafka Exporter を Kafka クラスターでデプロイし、特にコンシューマーラグの監視に関する追加のメトリクスを抽出します。
- 分散トレーシングを設定し、エンドツーエンドのメッセージ追跡を行います。
1.1.3. CPU およびメモリーのリソース制限および要求
デフォルトでは、AMQ Streams Cluster Operator はデプロイするオペランドの CPU およびメモリーリソースの要求および制限を指定しません。
Kafka などのアプリケーションでリソースの安定性を確保してパフォーマンスを向上させるには、十分なリソースを確保する必要があります。
使用する適切なリソース量は、特定の要件やユースケースによって異なります。
CPU およびメモリーリソースの設定を検討してください。AMQ Streams カスタムリソース の各コンテナーのリソース要求および制限を設定できます。
1.2. AMQ Streams のカスタムリソース
AMQ Streams を使用した Kafka コンポーネントの OpenShift クラスターへのデプロイメントは、カスタムリソースの適用により高度な設定が可能です。カスタムリソースは、OpenShift リソースを拡張するために CRD (カスタムリソース定義、Custom Resource Definition) によって追加される API のインスタンスとして作成されます。
CRD は、OpenShift クラスターでカスタムリソースを記述するための設定手順として機能し、デプロイメントで使用する Kafka コンポーネントごとに AMQ Streams で提供されます。CRD およびカスタムリソースは YAML ファイルとして定義されます。YAML ファイルのサンプルは AMQ Streams ディストリビューションに同梱されています。
また、CRD を使用すると、CLI へのアクセスや設定検証などのネイティブ OpenShift 機能を AMQ Streams リソースで活用することもできます。
1.2.1. AMQ Streams カスタムリソースの例
AMQ Streams 固有リソースのインスタンス化および管理に使用されるスキーマを定義するため、CRD をクラスターに 1 度インストールする必要があります。
CRD をインストールして新規カスタムリソースタイプをクラスターに追加した後に、その仕様に基づいてリソースのインスタンスを作成できます。
クラスターの設定によりますが、インストールには通常、クラスター管理者権限が必要です。
カスタムリソースの管理は、AMQ Streams 管理者のみが行えます。詳細は、AMQ Streams 管理者の指定 を参照してください。
kind:Kafka
などの新しい kind
リソースは、OpenShift クラスター内で CRD によって定義されます。
Kubernetes API サーバーを使用すると、kind
を基にしたカスタムリソースの作成が可能になり、カスタムリソースが OpenShift クラスターに追加されたときにカスタムリソースの検証および格納方法を CRD から判断します。
CRD が削除されると、そのタイプのカスタムタイプも削除されます。さらに、Pod や Statefulset などのカスタムリソースによって作成されたリソースも削除されます。
AMQ Streams 固有の各カスタムリソースは、リソースの kind
の CRD によって定義されるスキーマに準拠します。AMQ Streams コンポーネントのカスタムリソースには、spec
で定義される共通の設定プロパティーがあります。
CRD とカスタムリソースの関係を理解するため、Kafka トピックの CRD の例を見てみましょう。
Kafka トピックの CRD
apiVersion: kafka.strimzi.io/v1beta2 kind: CustomResourceDefinition metadata: 1 name: kafkatopics.kafka.strimzi.io labels: app: strimzi spec: 2 group: kafka.strimzi.io versions: v1beta2 scope: Namespaced names: # ... singular: kafkatopic plural: kafkatopics shortNames: - kt 3 additionalPrinterColumns: 4 # ... subresources: status: {} 5 validation: 6 openAPIV3Schema: properties: spec: type: object properties: partitions: type: integer minimum: 1 replicas: type: integer minimum: 1 maximum: 32767 # ...
- 1
- CRD を識別するためのトピック CRD、その名前および名前のメタデータ。
- 2
- この CRD に指定された項目には、トピックの API にアクセスするため URL に使用されるグルShortNameープ (ドメイン) 名、複数名、およびサポートされるスキーマバージョンが含まれます。他の名前は、CLI のインスタンスリソースを識別するために使用されます。たとえば、
oc get kafkaShortNametopic my-topic
やoc get kafkatopics
などです。 - 3
- ShortName は CLI コマンドで使用できます。たとえば、
oc get kafkatopic
の代わりにoc get kt
を略名として使用できます。 - 4
- カスタムリソースで
get
コマンドを使用する場合に示される情報。 - 5
- リソースの スキーマ参照 に記載されている CRD の現在の状態。
- 6
- openAPIV3Schema 検証によって、トピックカスタムリソースの作成が検証されます。たとえば、トピックには 1 つ以上のパーティションと 1 つのレプリカが必要です。
ファイル名に、インデックス番号とそれに続く「Crd」が含まれるため、AMQ Streams インストールファイルと提供される CRD YAML ファイルを識別できます。
KafkaTopic
カスタムリソースに該当する例は次のとおりです。
Kafka トピックカスタムリソース
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic 1 metadata: name: my-topic labels: strimzi.io/cluster: my-cluster 2 spec: 3 partitions: 1 replicas: 1 config: retention.ms: 7200000 segment.bytes: 1073741824 status: conditions: 4 lastTransitionTime: "2019-08-20T11:37:00.706Z" status: "True" type: Ready observedGeneration: 1 / ...
- 1
kind
およびapiVersion
によって、インスタンスであるカスタムリソースの CRD が特定されます。- 2
- トピックまたはユーザーが属する Kafka クラスターの名前 (
Kafka
リソースの名前と同じ) を定義する、KafkaTopic
およびKafkaUser
リソースのみに適用可能なラベル。 - 3
- 指定内容には、トピックのパーティション数およびレプリカ数や、トピック自体の設定パラメーターが示されています。この例では、メッセージがトピックに保持される期間や、ログのセグメントファイルサイズが指定されています。
- 4
KafkaTopic
リソースのステータス条件。lastTransitionTime
でtype
条件がReady
に変更されています。
プラットフォーム CLI からカスタムリソースをクラスターに適用できます。カスタムリソースが作成されると、Kubernetes API の組み込みリソースと同じ検証が使用されます。
KafkaTopic
の作成後、Topic Operator は通知を受け取り、該当する Kafka トピックが AMQ Streams で作成されます。
1.3. Kafka Bridge を使用した Kafka クラスターへの接続
AMQ Streams Kafka Bridge API を使用して、コンシューマーを作成および管理し、ネイティブ Kafka プロトコルではなく HTTP を介してレコードを送受信できます。
Kafka Bridge を設定する場合、Kafka クラスターへの HTTP アクセスを設定します。その後、Kafka Bridge を使用して、クラスターからのメッセージを生成および消費したり、REST インターフェースを介して他の操作を実行することができます。
関連情報
- Kafka Bridge のインストールおよび使用に関する詳細は、AMQ Streams Kafka Bridge の使用を参照してください。
1.4. 本書の表記慣例
ユーザーが置き換えた値
ユーザーが置き換えた値は、置換可能とも呼ばれ、山かっこ(<>)を付けて斜体で表示されます。アンダースコア(_)は、複数単語の値に使用されます。値がコードまたはコマンドを参照すると、monospace
も使用されます。
たとえば、以下のコードでは <my_namespace>
を namespace の名前に置き換えます。
sed -i 's/namespace: .*/namespace: <my_namespace>/' install/cluster-operator/*RoleBinding*.yaml
1.5. 関連情報
第2章 AMQ Streams のインストール方法
AMQ Streams on OpenShift 4.6 から 4.10 にインストールする方法は 2 つあります。
インストール方法 | 説明 |
---|---|
AMQ Streams ソフトウェアダウンロードページ から、Red Hat AMQ Streams 2.1 OpenShift インストールおよびサンプルファイル をダウンロードします。
| |
OperatorHub で Red Hat Integration - AMQ Streams を使用し、AMQ Streams を単一の namespace またはすべての namespace にデプロイします。 |
柔軟性が重要な場合は、インストールアーティファクトによる方法を選択します。OperatorHub メソッドは標準的な設定を提供し、、自動更新を利用できるようにします。
第3章 AMQ Streams でデプロイされるもの
Apache Kafka コンポーネントは、AMQ Streams ディストリビューションを使用して OpenShift にデプロイするために提供されます。Kafka コンポーネントは通常、クラスターとして実行され、可用性を確保します。
Kafka コンポーネントが組み込まれた通常のデプロイメントには以下が含まれます。
- ブローカーノードの Kafka クラスター
- レプリケートされた ZooKeeper インスタンスの zookeeper クラスター
- 外部データ接続用の Kafka Connect クラスター
- セカンダリークラスターで Kafka クラスターをミラーリングする Kafka MirrorMaker クラスター
- 監視用に追加のKafka メトリクスデータを抽出する Kafka Exporter
- Kafka クラスターに対して HTTP ベースの要求を行う Kafka Bridge
少なくとも Kafka および ZooKeeper は必要ですが、上記のコンポーネントがすべて必須なわけではありません。MirrorMaker や Kafka Connect など、一部のコンポーネントでは Kafka なしでデプロイできます。
3.1. デプロイメントの順序
OpenShift クラスターへのデプロイメントで必要な順序は次のとおりです。
- Cluster Operator をデプロイし、Kafka クラスターを管理します。
- ZooKeeper クラスターとともに Kafka クラスターをデプロイし、Topic Operator および User Operator がデプロイメントに含まれるようにします。
任意で以下をデプロイします。
- Topic Operator および User Operator (Kafka クラスターとともにデプロイしなかった場合)
- Kafka Connect
- Kafka MirrorMaker
- Kafka Bridge
- メトリクスを監視するためのコンポーネント
Cluster Operator は、 Deployment
、Service
、および Pod
リソースなどのコンポーネントの OpenShift リソースを作成します。OpenShift リソース名には、デプロイ時にコンポーネントに指定された名前が追加されます。たとえば、my-kafka-cluster
という名前の Kafka クラスターには、 my-kafka-cluster-kafka
という名前のサービスがあります。
第4章 AMQ Streams デプロイメントの準備
ここでは、AMQ Streams デプロイメントを準備する方法を説明します。
本ガイドのコマンドを実行するには、クラスターユーザーに RBAC (ロールベースアクセス制御) および CRD を管理する権限を付与する必要があります。
4.1. デプロイメントの前提条件
AMQ Streams をデプロイするには、以下が必要です。
OpenShift 4.6 から 4.10 クラスター。
AMQ Streams は AMQ Streams Strimzi 0.28.x をベースとしている。
-
oc
コマンドラインツールがインストールされ、稼働中のクラスターに接続するように設定されていること。
4.2. AMQ Streams リリースアーティファクトのダウンロード
デプロイメントファイルを使用して AMQ Streams をインストールするには、AMQ Streams ソフトウェアダウンロードページ から AMQ Streams の <version> インストールとサンプルファイル アーカイブをダウンロードして展開します。
AMQ Streams のリリースアーティファクトには、YAML ファイルが含まれています。これらのファイルは、AMQ Streams コンポーネントの OpenShift へのデプロイ、共通の操作の実行、および Kafka クラスターの設定に便利です。
oc
を使用して、ダウンロードした ZIP ファイルの install/cluster-operator
フォルダーから Cluster Operator をデプロイします。Cluster Operator のデプロイメントおよび設定に関する詳細は、「Cluster Operator のデプロイ」 を参照してください。
また、AMQ Streams Cluster Operator によって管理されない Kafka クラスターをトピックおよび User Operator のスタンドアロンインストールと共に使用する場合は、install/topic-operator
および install/user-operator
フォルダーからデプロイできます。
AMQ Streams コンテナーイメージは、Red Hat Ecosystem Catalog から使用することもできます。しかし、提供される YAML ファイルを使用して AMQ Streams をデプロイすることが推奨されます。
4.3. 設定ファイルとデプロイメントファイルの例
AMQ Streams で提供される設定およびデプロイメントファイルの例を使用して、異なる設定で Kafka コンポーネントをデプロイし、デプロイメントを監視します。カスタムリソースの設定ファイルの例には、重要なプロパティーおよび値が含まれています。これは、独自のデプロイメントでサポートされる追加の設定プロパティーで拡張できます。
4.3.1. ファイルの場所の例
サンプルファイルは、AMQ Streams ソフトウェアダウンロードページからダウンロード可能なリリースアーティファクトとともに提供されます。
oc
コマンドラインツールを使用してサンプルをダウンロードおよび適用できます。これらの例は、デプロイメントに独自の Kafka コンポーネント設定を構築する際の開始点として使用できます。
Operator を使用して AMQ Streams をインストールした場合、サンプルファイルをダウンロードし、それらを使用して設定をアップロードできます。
4.3.2. AMQ Streams で提供されるサンプルファイル
リリースアーティファクトには、設定例が含まれる examples
ディレクトリーが含まれます。
Examples ディレクトリー
examples ├── user 1 ├── topic 2 ├── security 3 │ ├── tls-auth │ ├── scram-sha-512-auth │ └── keycloak-authorization ├── mirror-maker 4 ├── metrics 5 ├── kafka 6 ├── cruise-control 7 ├── connect 8 └── bridge 9
- 1
- User Operator によって管理される
KafkaUser
カスタムリソース設定。 - 2
- Topic Operator によって管理される
KafkaTopic
カスタムリソースの設定。 - 3
- Kafka コンポーネントの認証および承認設定。TLS および SCRAM-SHA-512 認証の設定例が含まれています。Red Hat Single Sign-On の例には、
Kafka
カスタムリソース設定および Red Hat Single Sign-On レルム仕様が含まれています。この例を使用して、Red Hat Single Sign-On 承認サービスを試すことができます。 - 4
- Mirror Maker のデプロイメント用の
Kafka
カスタムリソース設定。レプリケーションポリシーおよび同期頻度の設定例が含まれます。 - 5
- Prometheus インストールおよび Grafana ダッシュボードファイルが含まれるメトリクス設定。
- 6
- Kafka のデプロイメント用の
Kafka
カスタムリソース設定。一時または永続で、単一または複数ノードのデプロイメントの設定例が含まれます。 - 7
- Cruise Control のデプロイメント設定のある
Kafka
カスタムリソース。デフォルトまたはユーザー最適化ゴールを使用する設定の例とともに、Cruise Control から最適化プロポーザルを生成するためのKafkaRebalance
カスタムリソースが含まれます。 - 8
- Kafka Connect のデプロイメントの
KafkaConnect
およびKafkaConnector
カスタムリソース設定。単一または複数ノードのデプロイメントの設定例が含まれています。 - 9
- KafkaBridge のデプロイメント用の
KafkaBridge
カスタムリソース設定。
4.4. コンテナーイメージを独自のレジストリーにプッシュ
AMQ Streams のコンテナーイメージは Red Hat Ecosystem Catalog にあります。AMQ Streams によって提供されるインストール YAML ファイルは、直接 Red Hat Ecosystem Catalog からイメージをプルします。
Red Hat Ecosystem Catalog にアクセスできない場合や独自のコンテナーリポジトリーを使用する場合は以下を行います。
- リストにある すべての コンテナーイメージをプルします。
- 独自のレジストリーにプッシュします。
- インストール YAML ファイルのイメージ名を更新します。
リリースに対してサポートされる各 Kafka バージョンには別のイメージがあります。
コンテナーイメージ | namespace/リポジトリー | 説明 |
---|---|---|
Kafka |
| 次を含む、Kafka を実行するための AMQ Streams イメージ。
|
Operator |
| Operator を実行するための AMQ Streams イメージ。
|
Kafka Bridge |
| AMQ Streams Kafka Bridge を稼働するための AMQ Streams イメージ |
AMQ Streams Drain Cleaner |
| AMQ Streams Drain Cleaner を実行するための AMQ Streams イメージ |
4.5. AMQ Streams の管理者の指名
AMQ Streams では、デプロイメントの設定にカスタムリソースが提供されます。デフォルトでは、これらのリソースを表示、作成、編集、および削除する権限は OpenShift クラスター管理者に制限されます。AMQ Streams には、このような権限を他のユーザーに割り当てするために使用する 2 つのクラスターロールがあります。
-
strimzi-view
ロールを指定すると、ユーザーは AMQ Streams リソースを表示できます。 -
strimzi-admin
ロールを指定すると、ユーザーは AMQ Streams リソースを作成、編集、または削除することもできます。
これらのロールをインストールすると、これらの権限は自動でデフォルトの OpenShift クラスターロールに集約 (追加) されます。strimzi-view
は view
ロールに集約され、strimzi-admin
は edit
および admin
ロールに集約されます。集約により、これらのロールを同様の権限を持つユーザーに割り当てする必要がない可能性があります。
以下の手順では、クラスター管理者でないユーザーが AMQ Streams リソースを管理できるようにする strimzi-admin
ロールの割り当て方法を説明します。
システム管理者は、Cluster Operator のデプロイ後に AMQ Streams の管理者を指名できます。
前提条件
- Cluster Operator でデプロイとデプロイされた CRD (カスタムリソース定義) を管理する AMQ Streams の CRD リソースおよび RBAC (ロールベースアクセス制御) リソースが必要です。
手順
OpenShift で
strimzi-view
およびstrimzi-admin
クラスターロールを作成します。oc create -f install/strimzi-admin
必要な場合は、ユーザーに必要なアクセス権限を付与するロールを割り当てます。
oc create clusterrolebinding strimzi-admin --clusterrole=strimzi-admin --user=user1 --user=user2
第5章 Web コンソールを使用した OperatorHub からの AMQ Streams のインストール
OpenShift Container Platform Web コンソールの OperatorHub から Red Hat Integration - AMQ Streams Operator をインストールします。
本セクションの手順では以下の方法を説明します。
5.1. Red Hat Integration Operator を使用した AMQ Streams Operator のインストール
Red Hat Integration Operator (非推奨) を使用すると、Red Hat Integration コンポーネントを管理する Operator を選択およびインストールできます。複数の Red Hat Integration サブスクリプションがある場合、Red Hat Integration Operator を使用して、AMQ Streams Operator およびサブスクライブしている Red Hat Integration コンポーネントのすべての Operator をインストールおよび更新できます。
AMQ Streams Operator の場合は、Operator Lifecycle Manager (OLM) を使用して、OCP コンソールの OperatorHub から OpenShift Container Platform (OCP) クラスターに Red Hat Integration Operator をインストールできます。
Red Hat Integration Operator は非推奨となり、今後削除予定です。OpenShift4.6 から 4.10 の OperatorHub で利用できるようになります。
関連情報
Red Hat Integration Operator のインストールおよび使用に関する詳細は、『Red Hat Integration Operator のインストール』を参照してください。
5.2. OperatorHub からの AMQ Streams Operator のインストール
OpenShift Container Platform Web コンソールの OperatorHub を使用して、AMQ Streams オペレーターをインストールしてサブスクライブできます。
この手順では、プロジェクトを作成し、そのプロジェクトに AMQ Streams オペレーターをインストールする方法について説明します。プロジェクトは名前空間の表現です。管理しやすくするために、namespace を使用して関数を分離することをお勧めします。
適切な更新チャネルを使用するようにしてください。サポートされるバージョンの OpenShift を使用している場合、デフォルトの stable チャネルから安全に AMQ Streams をインストールできます。ただし、stable チャネルで自動更新を有効にすることは推奨されません。自動アップグレードでは、アップグレード前に必要な手順がスキップされます。バージョン固有のチャネルでのみ自動アップグレードを使用します。
前提条件
-
cluster-admin
またはstrimzi-admin
パーミッションを持つアカウントを使用して OpenShift Container Platform Web コンソールにアクセスできること。
手順
OpenShift Web コンソールで Home > Projects ページに移動し、インストール用のプロジェクト (namespace) を作成します。
この例では、
amq-streams-kafka
という名前のプロジェクトを使用します。- Operators > OperatorHub ページに移動します。
スクロール、または Filter by keyword ボックスにキーワードを入力して、Red Hat Integration - AMQ Streams Operator を見つけます。
Operator は、Streaming & Messaging カテゴリーにあります。
- Red Hat Integration - AMQ Streams をクリックして、Operator 情報を表示します。
- Operator に関する情報を確認し、Install をクリックします。
Install Operator ページで、次のインストールおよび更新オプションから選択します。
Update Channel (更新チャネル):Operator の更新チャネルを選択します。
- stable チャネル (デフォルト) には最新の更新とリリースがすべて含まれます。これには、十分なテストを行った上、安定していることが想定される、メジャー、マイナー、およびマイクロリリースが含まれます。
- amq-streams-X.x チャネルには、メジャーリリースのマイナーリリースの更新およびマイクロリリースの更新が含まれます。X は、メジャーリリースのバージョン番号に置き換えられます。
- amq-streams-X.Y.x チャネルには、マイナーリリースのマイクロリリースの更新が含まれます。X はメジャーリリースのバージョン番号、Y はマイナーリリースのバージョン番号に置き換えられます。
Installation Mode (インストールモード):作成したプロジェクトを選択して、特定の namespace に Operator をインストールします。
AMQ Streams Operator をクラスターのすべての namespace (デフォルトのオプション) にインストールするか、特定の namespace にインストールするかを選択できます。特定の namespace を Kafka クラスターおよびその他の AMQ Streams コンポーネントの専用とすることが推奨されます。
- 更新の承認:デフォルトでは、OLM (Operator Lifecycle Manager) によって、AMQ Streams Operator は自動的に最新の AMQ Streams バージョンにアップグレードされます。今後のアップグレードを手動で承認する場合は、Manual を選択します。詳細は、OpenShift ドキュメントの Operators ガイドを参照してください。
Install をクリックして、選択した namespace に Operator をインストールします。
AMQ Streams Operator によって、Cluster Operator、CRD、およびロールベースアクセス制御 (RBAC) リソースは選択された namespace にデプロイされます。
Operator を使用する準備ができたら、Operators > Installed Operators に移動して、Operator が選択した namespace にインストールされていることを確認します。
ステータスは Succeeded と表示されます。
これで、AMQ Streams Operator を使用して、Kafka クラスターから始めて Kafka コンポーネントをデプロイできます。
5.3. AMQ Streams Operator を使用した Kafka コンポーネントのデプロイ
Openshift にインストールすると、AMQ Streams Operator は、ユーザーインターフェイスから Kafka コンポーネントをインストールできるようにします。
次の Kafka コンポーネントをインストールできます。
- Kafka
- Kafka Connect
- Kafka MirrorMaker
- Kafka MirrorMaker 2
- Kafka Topic
- Kafka User
- Kafka Bridge
- Kafka Connector
- Kafka Rebalance
コンポーネントを選択して、インスタンスを作成します。少なくとも、Kafka インスタンスを作成します。この手順では、デフォルト設定を使用して Kafka インスタンスを作成する方法を説明します。インストールを実行する前に、デフォルトのインストール仕様を設定できます。
プロセスは、他の Kafka コンポーネントのインスタンスを作成する場合と同じです。
前提条件
- AMQ Streams オペレーターは、OpenShift クラスターにインストールされます。
手順
Web コンソールで Operators > Installed Operators ページに移動し、Red Hat Integration - AMQ Streams をクリックして、Operator の詳細を表示します。
提供されている API から、Kafka コンポーネントのインスタンスを作成できます。
Kafka の下の Create instance をクリックして、Kafka インスタンスを作成します。
デフォルトでは、3 つの Kafka ブローカーノードと 3 つの ZooKeeper ノードを持つ
my-cluster
という名の Kafka クラスターを作成します。クラスターはエフェメラルストレージを使用します。Create をクリックして、Kafka のインストールを開始します。
ステータスが Ready に変わるまで待ちます。
第6章 インストールアーティファクトを使用した AMQ Streams のデプロイ
AMQ Streams のデプロイメント環境を準備 したら、AMQ Streams を OpenShift クラスターにデプロイできます。リリースアーティファクトで提供されるデプロイメントファイルを使用できます。
デプロイメントファイルを使用して Kafka クラスターを作成します。
任意で、要件に応じて以下の Kafka コンポーネントをデプロイできます。
AMQ Streams は Strimzi 0.28.x をベースとしています。AMQ Streams 2.1 は、OpenShift 4.6 から 4.10 にデプロイできます。
本ガイドのコマンドを実行するには、クラスターユーザーに RBAC (ロールベースアクセス制御) および CRD を管理する権限を付与する必要があります。
6.1. Kafka クラスターの作成
Cluster Operator で Kafka クラスターを管理できるようにするには、これを Kafka
リソースとしてデプロイする必要があります。AMQ Streams では、これ用のデプロイメントファイルのサンプルが同梱されています。これらのファイルを使用して、Topic Operator および User Operator を同時にデプロイできます。
Kafka クラスターを Kafka
リソースとしてデプロイしていない場合は、Cluster Operator を使用してこれを管理できません。これは、たとえば OpenShift 外で実行されている Kafka クラスターに適用されます。ただし、Topic Operator および User Operator は、スタンドアロンコンポーネントとしてデプロイし、使用することができます。
Cluster Operator は、OpenShift クラスターの 1 つ、複数、またはすべての namespace を監視できます。Topic Operator および User Operator は、Kafka クラスターデプロイメントの単一の namespace で KafkaTopics
および KafkaUsers
を監視します。
Kafka クラスターを Topic Operator および User Operator とデプロイ
AMQ Streams によって管理される Kafka クラスターを Topic Operator および User Operator と使用する場合は、このデプロイメント手順を実行します。
- Cluster Operator をデプロイします。
Cluster Operator を使用して以下をデプロイします。
スタンドアロン Topic Operator および User Operator のデプロイ
AMQ Streams によって管理されない Kafka クラスターを Topic Operator および User Operator と使用する場合は、このデプロイメント手順を実行します。
6.1.1. Cluster Operator のデプロイ
Cluster Operator は、OpenShift クラスター内で Apache Kafka クラスターのデプロイおよび管理を行います。
本セクションの手順では、以下のいずれかを監視するように Cluster Operator をデプロイする方法を説明します。
6.1.1.1. Cluster Operator デプロイメントの監視オプション
Cluster Operator の稼働中に、Kafka リソースの更新に対する監視が開始されます。
Cluster Operator をデプロイして、以下からの Kafka リソースの監視を選択できます。
- 単一の namespace (Cluster Operator が含まれる同じ namespace)
- 複数の namespace
- すべての namespace
AMQ Streams では、デプロイメントの処理を簡単にするため、YAML ファイルのサンプルが提供されます。
Cluster Operator では、以下のリソースの変更が監視されます。
-
Kafka クラスターの
Kafka
。 -
Kafka Connect クラスターの
KafkaConnect
。 -
Kafka Connect クラスターでコネクターを作成および管理するための
KafkaConnector
。 -
Kafka MirrorMaker インスタンスの
KafkaMirrorMaker
。 -
Kafka MirrorMaker 2.0 インスタンスの
KafkaMirrorMaker2
。 -
Kafka Bridge インスタンスの
KafkaBridge
。 -
Cruise Control の最適化リクエストの
KafkaRebalance
。
OpenShift クラスターでこれらのリソースの 1 つが作成されると、Operator によってクラスターの詳細がリソースより取得されます。さらに、StatefulSet、Service、および ConfigMap などの必要な OpenShift リソースが作成され、リソースの新しいクラスターの作成が開始されます。
Kafka リソースが更新されるたびに、リソースのクラスターを構成する OpenShift リソースで該当する更新が Operator によって実行されます。
クラスターの望ましい状態がリソースのクラスターに反映されるようにするため、リソースへのパッチ適用後またはリソースの削除後にリソースが再作成されます。この操作は、サービスの中断を引き起こすローリングアップデートの原因となる可能性があります。
リソースが削除されると、Operator によってクラスターがアンデプロイされ、関連する OpenShift リソースがすべて削除されます。
6.1.1.2. 単一の namespace を監視対象とする Cluster Operator のデプロイメント
この手順では、OpenShift クラスターの単一の namespace で AMQ Streams リソースを監視するように Cluster Operator をデプロイする方法を説明します。
前提条件
-
この手順では、
CustomResourceDefinitions
、ClusterRoles
、およびClusterRoleBindings
を作成できる OpenShift ユーザーアカウントを使用する必要があります。通常、OpenShift クラスターでロールベースアクセス制御 (RBAC) を使用する場合、これらのリソースを作成、編集、および削除する権限を持つユーザーはsystem:admin
などの OpenShift クラスター管理者に限定されます。
手順
Cluster Operator がインストールされる namespace を使用するように、AMQ Streams のインストールファイルを編集します。
たとえば、この手順では Cluster Operator は
<my_cluster_operator_namespace>
という namespace にインストールされます。Linux の場合は、以下を使用します。
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
MacOS の場合は、以下を使用します。
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
Cluster Operator をデプロイします。
oc create -f install/cluster-operator -n <my_cluster_operator_namespace>
デプロイメントのステータスを確認します。
oc get deployments -n <my_cluster_operator_namespace>
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1
READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.1.1.3. 複数の namespace を監視対象とする Cluster Operator のデプロイメント
この手順では、OpenShift クラスターの複数の namespace 全体で AMQ Streams リソースを監視するように Cluster Operator をデプロイする方法を説明します。
前提条件
-
この手順では、
CustomResourceDefinitions
、ClusterRoles
、およびClusterRoleBindings
を作成できる OpenShift ユーザーアカウントを使用する必要があります。通常、OpenShift クラスターでロールベースアクセス制御 (RBAC) を使用する場合、これらのリソースを作成、編集、および削除する権限を持つユーザーはsystem:admin
などの OpenShift クラスター管理者に限定されます。
手順
Cluster Operator がインストールされる namespace を使用するように、AMQ Streams のインストールファイルを編集します。
たとえば、この手順では Cluster Operator は
<my_cluster_operator_namespace>
という namespace にインストールされます。Linux の場合は、以下を使用します。
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
MacOS の場合は、以下を使用します。
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml
ファイルを編集し、Cluster Operator によって監視されるすべての namespace のリストをSTRIMZI_NAMESPACE
環境変数に追加します。たとえば、この手順では Cluster Operator は
watched-namespace-1
、watched-namespace-2
、およびwatched-namespace-3
という namespace を監視します。apiVersion: apps/v1 kind: Deployment spec: # ... template: spec: serviceAccountName: strimzi-cluster-operator containers: - name: strimzi-cluster-operator image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.1.0 imagePullPolicy: IfNotPresent env: - name: STRIMZI_NAMESPACE value: watched-namespace-1,watched-namespace-2,watched-namespace-3
リストした各 namespace に
RoleBindings
をインストールします。この例では、コマンドの
watched-namespace
を前述のステップでリストした namespace に置き換えます。watched-namespace-1
、watched-namespace-2
、およびwatched-namespace-3
に対してこれを行います。oc create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace> oc create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>
Cluster Operator をデプロイします。
oc create -f install/cluster-operator -n <my_cluster_operator_namespace>
デプロイメントのステータスを確認します。
oc get deployments -n <my_cluster_operator_namespace>
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1
READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.1.1.4. すべての namespace を対象とする Cluster Operator のデプロイメント
この手順では、OpenShift クラスターのすべての namespace 全体で AMQ Streams リソースを監視するように Cluster Operator をデプロイする方法を説明します。
このモードで実行している場合、Cluster Operator によって、新規作成された namespace でクラスターが自動的に管理されます。
前提条件
-
この手順では、
CustomResourceDefinitions
、ClusterRoles
、およびClusterRoleBindings
を作成できる OpenShift ユーザーアカウントを使用する必要があります。通常、OpenShift クラスターでロールベースアクセス制御 (RBAC) を使用する場合、これらのリソースを作成、編集、および削除する権限を持つユーザーはsystem:admin
などの OpenShift クラスター管理者に限定されます。
手順
Cluster Operator がインストールされる namespace を使用するように、AMQ Streams のインストールファイルを編集します。
たとえば、この手順では Cluster Operator は
<my_cluster_operator_namespace>
という namespace にインストールされます。Linux の場合は、以下を使用します。
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
MacOS の場合は、以下を使用します。
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml
ファイルを編集し、STRIMZI_NAMESPACE
環境変数の値を*
に設定します。apiVersion: apps/v1 kind: Deployment spec: # ... template: spec: # ... serviceAccountName: strimzi-cluster-operator containers: - name: strimzi-cluster-operator image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.1.0 imagePullPolicy: IfNotPresent env: - name: STRIMZI_NAMESPACE value: "*" # ...
クラスター全体ですべての namespace にアクセスできる権限を Cluster Operator に付与する
ClusterRoleBindings
を作成します。oc create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount <my_cluster_operator_namespace>:strimzi-cluster-operator oc create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount <my_cluster_operator_namespace>:strimzi-cluster-operator
<my_cluster_operator_namespace>
は、Cluster Operator をインストールする namespace に置き換えます。Cluster Operator を OpenShift クラスターにデプロイします。
oc create -f install/cluster-operator -n <my_cluster_operator_namespace>
デプロイメントのステータスを確認します。
oc get deployments -n <my_cluster_operator_namespace>
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1
READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.1.2. Kafka のデプロイ
Apache Kafka は、耐障害性のリアルタイムデータフィードを実現する、オープンソースの分散型 publish/subscribe メッセージングシステムです。
本セクションの手順では、以下を説明します。
Cluster Operator を使用して以下をデプロイする方法
- 一時 または 永続 Kafka クラスター
Topic Operator および User Operator (
Kafka
カスタムリソースを設定してデプロイする)
Topic Operator および User Operator の代替のスタンドアロンデプロイメント手順
Kafka をインストールする場合、AMQ Streams によって ZooKeeper クラスターもインストールされ、Kafka と ZooKeeper との接続に必要な設定が追加されます。
6.1.2.1. Kafka クラスターのデプロイメント
この手順では、Cluster Operator を使用して Kafka クラスターを OpenShift クラスターにデプロイする方法を説明します。
デプロイメントでは、YAML ファイルの仕様を使って Kafka
リソースが作成されます。
AMQ Streams には、設定ファイルのサンプルが含まれています。Kafka デプロイメントでは、以下の例が提供されます。
kafka-persistent.yaml
- 3 つの Zookeeper ノードと 3 つの Kafka ノードを使用して永続クラスターをデプロイします。
kafka-jbod.yaml
- それぞれが複数の永続ボリューを使用する、3 つの ZooKeeper ノードと 3 つの Kafka ノードを使用して、永続クラスターをデプロイします。
kafka-persistent-single.yaml
- 1 つの ZooKeeper ノードと 1 つの Kafka ノードを使用して、永続クラスターをデプロイします。
kafka-ephemeral.yaml
- 3 つの ZooKeeper ノードと 3 つの Kafka ノードを使用して、一時クラスターをデプロイします。
kafka-ephemeral-single.yaml
- 3 つの ZooKeeper ノードと 1 つの Kafka ノードを使用して、一時クラスターをデプロイします。
この手順では、一時 および 永続 Kafka クラスターデプロイメントの例を使用します。
- 一時クラスター
-
通常、Kafka の一時クラスターは開発およびテスト環境での使用に適していますが、本番環境での使用には適していません。このデプロイメントでは、ブローカー情報 (ZooKeeper) と、トピックまたはパーティション (Kafka) を格納するための
emptyDir
ボリュームが使用されます。emptyDir
ボリュームを使用すると、その内容は厳密に Pod のライフサイクルと関連し、Pod がダウンすると削除されます。 - 永続クラスター
-
Kafka の永続クラスターでは、
PersistentVolumes
を使用して ZooKeeper および Kafka データを格納します。PersistentVolumeClaim
を使用してPersistentVolume
が取得され、PersistentVolume
の実際のタイプには依存しません。たとえば、YAML ファイルを変更しなくても Amazon AWS デプロイメントで Amazon EBS ボリュームを使用できます。PersistentVolumeClaim
でStorageClass
を使用し、自動ボリュームプロビジョニングをトリガーすることができます。
サンプル YAML ファイルは、サポートされる最新の Kafka バージョンを指定し、サポートされるログメッセージ形式バージョンの設定とブローカー間のプロトコルバージョンの設定を指定します。Kafka config
の inter.broker.protocol.version
プロパティーは、指定された Kafka バージョン (spec.kafka.version
) によってサポートされるバージョンである必要があります。このプロパティーは、Kafka クラスターで使用される Kafka プロトコルのバージョンを表します。
Kafka 3.0.0 以降、inter.broker.protocol.version
が 3.0
以上に設定されていると、log.message.format.version
オプションは無視されるため、設定する必要はありません。
Kafka のアップグレード 時に、inter.broker.protocol.version
への更新が必要です。
サンプルクラスターの名前はデフォルトで my-cluster
になります。クラスター名はリソースの名前によって定義され、クラスターがデプロイされた後に変更できません。クラスターをデプロイする前にクラスター名を変更するには、関連する YAML ファイルにある Kafka
リソースの Kafka.metadata.name
プロパティーを編集します。
デフォルトのクラスター名および指定された Kafka バージョン
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: kafka: version: 3.1.0 #... config: #... log.message.format.version: "3.1" inter.broker.protocol.version: "3.1" # ...
手順
一時 または 永続 クラスターを作成およびデプロイします。
開発またはテストでは、一時クラスターの使用が適している可能性があります。永続クラスターはどのような状況でも使用することができます。
一時 クラスターを作成およびデプロイするには、以下を実行します。
oc apply -f examples/kafka/kafka-ephemeral.yaml
永続 クラスターを作成およびデプロイするには、以下を実行します。
oc apply -f examples/kafka/kafka-persistent.yaml
デプロイメントのステータスを確認します。
oc get pods -n <my_cluster_operator_namespace>
出力には、Pod 名および readiness が表示されます。
NAME READY STATUS RESTARTS my-cluster-entity-operator 3/3 Running 0 my-cluster-kafka-0 1/1 Running 0 my-cluster-kafka-1 1/1 Running 0 my-cluster-kafka-2 1/1 Running 0 my-cluster-zookeeper-0 1/1 Running 0 my-cluster-zookeeper-1 1/1 Running 0 my-cluster-zookeeper-2 1/1 Running 0
my-cluster
は Kafka クラスターの名前です。デフォルトのデプロイメントでは、Entity Operator クラスター、3 つの Kafka Pod、および 3 つの ZooKeeper Pod をインストールします。
READY
は、Ready/expected 状態のレプリカ数を表示します。STATUS
がRunning
と表示されると、デプロイメントは成功しています。
関連情報
6.1.2.2. Cluster Operator を使用した Topic Operator のデプロイ
この手順では、Cluster Operator を使用して Topic Operator をデプロイする方法を説明します。
Kafka
リソースの entityOperator
プロパティーを設定し、topicOperator
が含まれるようにします。デフォルトでは、Topic Operator は Kafka クラスターデプロイメントの namespace で KafkaTopics
を監視します。
AMQ Streams によって管理されない Kafka クラスターを Topic Operator と使用する場合は、Topic Operator をスタンドアロンコンポーネントとしてデプロイする必要があります。
entityOperator
および topicOperator
プロパティーの設定に関する詳細は、Configuring the Entity Operator を参照してください。
手順
Kafka
リソースのentityOperator
プロパティーを編集し、topicOperator
が含まれるようにします。apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: #... entityOperator: topicOperator: {} userOperator: {}
「
EntityTopicOperatorSpec
schema reference」に記載されているプロパティーを使用して、Topic Operator のspec
を設定します。すべてのプロパティーにデフォルト値を使用する場合は、空のオブジェクト (
{}
) を使用します。リソースを作成または更新します。
次のように
oc apply
を使用します。oc apply -f <your-file>
6.1.2.3. Cluster Operator を使用した User Operator のデプロイ
この手順では、Cluster Operator を使用して User Operator をデプロイする方法を説明します。
Kafka
リソースの entityOperator
プロパティーを設定し、userOperator
が含まれるようにします。デフォルトでは、User Operator は Kafka クラスターデプロイメントの namespace で KafkaUsers
を監視します。
AMQ Streams によって管理されない Kafka クラスターを User Operator と使用する場合は、User Operator をスタンドアロンコンポーネントとしてデプロイする必要があります。
entityOperator
および userOperator
プロパティーの設定に関する詳細は、Configuring the Entity Operator を参照してください。
手順
Kafka
リソースのentityOperator
プロパティーを編集し、userOperator
が含まれるようにします。apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: #... entityOperator: topicOperator: {} userOperator: {}
「
EntityUserOperatorSpec
スキーマ参照」に記載されているプロパティーを使用して、User Operator のspec
を設定します。すべてのプロパティーにデフォルト値を使用する場合は、空のオブジェクト (
{}
) を使用します。リソースを作成または更新します。
oc apply -f <your-file>
6.1.3. AMQ Streams Operator の代替のスタンドアロンデプロイメントオプション
Topic Operator および User Operator のスタンドアロンデプロイメントを実行できます。Cluster Operator によって管理されない Kafka クラスターを使用している場合は、これらの Operator のスタンドアロンデプロイメントを検討してください。
Operator を OpenShift にデプロイします。Kafka は OpenShift 外で実行できます。たとえば、Kafka を管理対象サービスとして使用する場合があります。スタンドアロン Operator のデプロイメント設定を調整し、Kafka クラスターのアドレスと一致するようにします。
6.1.3.1. スタンドアロン Topic Operator のデプロイ
この手順では、Topic Operator をトピック管理のスタンドアロンコンポーネントとしてデプロイする方法を説明します。スタンドアロン Topic Operator を Cluster Operator によって管理されない Kafka クラスターと使用できます。
スタンドアロンデプロイメントは、任意の Kafka クラスターと操作できます。
スタンドアロンデプロイメントファイルは AMQ Streams で提供されます。05-Deployment-strimzi-topic-operator.yaml
デプロイメントファイルを使用して、Topic Operator をデプロイします。Kafka クラスターへの接続に必要な環境変数を追加または設定します。
前提条件
Topic Operator が接続する Kafka クラスターを実行しています。
スタンドアロンの Topic Operator が接続用に正しく設定されている限り、Kafka クラスターはベアメタル環境、仮想マシン、または管理対象クラウドアプリケーションサービスで実行できます。
手順
install/topic-operator/05-Deployment-strimzi-topic-operator.yaml
スタンドアロンデプロイメントファイルのenv
プロパティーを編集します。スタンドアロンの Topic Operator デプロイメント設定の例
apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-topic-operator labels: app: strimzi spec: # ... template: # ... spec: # ... containers: - name: strimzi-topic-operator # ... env: - name: STRIMZI_NAMESPACE 1 valueFrom: fieldRef: fieldPath: metadata.namespace - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2 value: my-kafka-bootstrap-address:9092 - name: STRIMZI_RESOURCE_LABELS 3 value: "strimzi.io/cluster=my-cluster" - name: STRIMZI_ZOOKEEPER_CONNECT 4 value: my-cluster-zookeeper-client:2181 - name: STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 5 value: "18000" - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6 value: "120000" - name: STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS 7 value: "6" - name: STRIMZI_LOG_LEVEL 8 value: INFO - name: STRIMZI_TLS_ENABLED 9 value: "false" - name: STRIMZI_JAVA_OPTS 10 value: "-Xmx=512M -Xms=256M" - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 11 value: "-Djavax.net.debug=verbose -DpropertyName=value" - name: STRIMZI_PUBLIC_CA 12 value: "false" - name: STRIMZI_TLS_AUTH_ENABLED 13 value: "false" - name: STRIMZI_SASL_ENABLED 14 value: "false" - name: STRIMZI_SASL_USERNAME 15 value: "admin" - name: STRIMZI_SASL_PASSWORD 16 value: "password" - name: STRIMZI_SASL_MECHANISM 17 value: "scram-sha-512" - name: STRIMZI_SECURITY_PROTOCOL 18 value: "SSL"
- 1
KafkaTopic
リソースを監視する Topic Operator の OpenShift namespace。Kafka クラスターの namespace を指定します。- 2
- Kafka クラスターのすべてのブローカーを検出し、接続するブートストラップブローカーアドレスのホストとポートのペア。サーバーがダウンした場合に備えて、コンマ区切りリストを使用して 2 つまたは 3 つのブローカーアドレスを指定します。
- 3
- Topic Operator によって管理される
KafkaTopic
リソースを識別するためのラベルセレクター。 - 4
- ZooKeeper クラスターに接続するためのアドレスのホストおよびポートのペア。これは、Kafka クラスターが使用する ZooKeeper クラスターと同じである必要があります。
- 5
- ZooKeeper セッションのタイムアウト (秒単位)。デフォルトは
18000
(18 秒) です。 - 6
- 定期的な調整の間隔 (秒単位)。デフォルトは
120000
(2 分) です。 - 7
- Kafka からトピックメタデータの取得を試行する回数。各試行の間隔は、指数バックオフとして定義されます。パーティションまたはレプリカの数が原因で、トピックの作成に時間がかかる場合は、この値を大きくすることを検討してください。デフォルトの試行回数は
6
回です。 - 8
- ロギングメッセージの出力レベル。レベルを、
ERROR
、WARNING
、INFO
、DEBUG
、またはTRACE
に設定できます。 - 9
- Kafka ブローカーとの暗号化された通信の TLS サポートを有効にします。
- 10
- (任意) Topic Operator を実行する JVM に使用される Java オプション。
- 11
- (任意) Topic Operator に設定されたデバッグ (
-D
) オプション。 - 12
- (オプション)TLS が
STRIMZI_TLS_ENABLED
によって有効になっている場合、トラストストア証明書の生成を省略します。この環境変数が有効になっている場合、ブローカーは TLS 証明書に公的に信頼できる認証局を使用する必要があります。デフォルトはfalse
です。 - 13
- (オプション)相互 TLS 認証用のキーストア証明書を生成します。これを
false
に設定すると、TLS を使用した Kafka ブローカーへのクライアント認証が無効になります。デフォルトはtrue
です。 - 14
- (オプション)Kafka ブローカーに接続するときにクライアント認証の SASL サポートを有効にします。デフォルトは
false
です。 - 15
- (任意)クライアント認証用の SASL ユーザー名。SASL が
STRIMZI_SASL_ENABLED
によって有効化された場合のみ必須です。 - 16
- (任意)クライアント認証用の SASL パスワード。SASL が
STRIMZI_SASL_ENABLED
によって有効化された場合のみ必須です。 - 17
- (任意)クライアント認証用の SASL メカニズム。SASL が
STRIMZI_SASL_ENABLED
によって有効化された場合のみ必須です。この値はplain
、scram-sha-256
、またはscram-sha-512
に設定できます。 - 18
- (任意)Kafka ブローカーとの通信に使用されるセキュリティープロトコル。デフォルト値は「PLAINTEXT」です。値は
PLAINTEXT
、SSL
、SASL_PLAINTEXT
、またはSASL_SSL
に設定できます。
-
公開認証局から証明書を使用している Kafkaブ ローカーに接続する場合は、
STRIMZI_PUBLIC_CA
をtrue
に設定します。たとえば、Amazon AWS MSK サービスを使用している場合は、このプロパティーをtrue
に設定します。 STRIMZI_TLS_ENABLED
環境変数で TLS を有効にした場合、Kafka クラスターへの接続を認証するために使用されるキーストアおよびトラストストアを指定します。TLS 設定の例
# .... env: - name: STRIMZI_TRUSTSTORE_LOCATION 1 value: "/path/to/truststore.p12" - name: STRIMZI_TRUSTSTORE_PASSWORD 2 value: "TRUSTSTORE-PASSWORD" - name: STRIMZI_KEYSTORE_LOCATION 3 value: "/path/to/keystore.p12" - name: STRIMZI_KEYSTORE_PASSWORD 4 value: "KEYSTORE-PASSWORD" # ...
Topic Operator をデプロイします。
oc create -f install/topic-operator
デプロイメントのステータスを確認します。
oc get deployments
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE strimzi-topic-operator 1/1 1 1
READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.1.3.2. スタンドアロン User Operator のデプロイ
この手順では、ユーザー管理のスタンドアロンコンポーネントとして User Operator をデプロイする方法を説明します。スタンドアロン User Operator を Cluster Operator によって管理されない Kafka クラスターと使用できます。
スタンドアロンデプロイメントは、任意の Kafka クラスターと操作できます。
スタンドアロンデプロイメントファイルは AMQ Streams で提供されます。05-Deployment-strimzi-user-operator.yaml
デプロイメントファイルを使用して、User Operator をデプロイします。Kafka クラスターへの接続に必要な環境変数を追加または設定します。
前提条件
User Operator が接続する Kafka クラスターを実行しています。
スタンドアロンの User Operator が接続用に正しく設定されている限り、Kafka クラスターはベアメタル環境、仮想マシン、または管理対象クラウドアプリケーションサービスで実行できます。
手順
以下の
env
プロパティーをinstall/user-operator/05-Deployment-strimzi-user-operator.yaml
スタンドアロンデプロイメントファイルで編集します。スタンドアロン User Operator デプロイメント設定の例
apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-user-operator labels: app: strimzi spec: # ... template: # ... spec: # ... containers: - name: strimzi-user-operator # ... env: - name: STRIMZI_NAMESPACE 1 valueFrom: fieldRef: fieldPath: metadata.namespace - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2 value: my-kafka-bootstrap-address:9092 - name: STRIMZI_CA_CERT_NAME 3 value: my-cluster-clients-ca-cert - name: STRIMZI_CA_KEY_NAME 4 value: my-cluster-clients-ca - name: STRIMZI_LABELS 5 value: "strimzi.io/cluster=my-cluster" - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6 value: "120000" - name: STRIMZI_LOG_LEVEL 7 value: INFO - name: STRIMZI_GC_LOG_ENABLED 8 value: "true" - name: STRIMZI_CA_VALIDITY 9 value: "365" - name: STRIMZI_CA_RENEWAL 10 value: "30" - name: STRIMZI_JAVA_OPTS 11 value: "-Xmx=512M -Xms=256M" - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 12 value: "-Djavax.net.debug=verbose -DpropertyName=value" - name: STRIMZI_SECRET_PREFIX 13 value: "kafka-" - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 14 value: "true"
- 1
KafkaUser
リソースを監視する User Operator の OpenShift namespace。指定できる namespace は 1 つだけです。- 2
- Kafka クラスターのすべてのブローカーを検出し、接続するブートストラップブローカーアドレスのホストとポートのペア。サーバーがダウンした場合に備えて、コンマ区切りリストを使用して 2 つまたは 3 つのブローカーアドレスを指定します。
- 3
- TLS クライアント認証に対して新しいユーザー証明書に署名する認証局の公開鍵 (
ca.crt
) の値が含まれる OpenShiftSecret
。 - 4
- TLS クライアント認証に対して新しいユーザー証明書に署名する認証局の秘密鍵 (
ca.key
) の値が含まれる OpenShiftSecret
。 - 5
- User Operator によって管理される
KafkaUser
リソースを識別するために使用されるラベルセレクター。 - 6
- 定期的な調整の間隔 (秒単位)。デフォルトは
120000
(2 分) です。 - 7
- ロギングメッセージの出力レベル。レベルを、
ERROR
、WARNING
、INFO
、DEBUG
、またはTRACE
に設定できます。 - 8
- ガベッジコレクション (GC) ロギングを有効にします。デフォルトは
true
です。 - 9
- 認証局の有効期間。デフォルトは
365
日です。 - 10
- 認証局の更新期間。更新期間は、現在の証明書の有効期日から逆算されます。デフォルトでは、古い証明書が期限切れになる前の証明書の更新期間は
30
日です。 - 11
- (任意) User Operator を実行する JVM に使用される Java オプション。
- 12
- (任意) User Operator に設定されたデバッグ (
-D
) オプション。 - 13
- (オプション)User Operator によって作成される OpenShift シークレットの名前のプレフィックス。
- 14
- (任意)Kafka クラスターが Kafka Admin API を使用した承認 ACL ルールの管理をサポートするかどうかを示します。
false
に設定すると、User Operator はsimple
承認 ACL ルールを持つすべてのリソースを拒否します。これは、Kafka クラスターログで不要な例外を回避するのに役立ちます。デフォルトはtrue
です。
TLS を使用して Kafka クラスターに接続する場合は、接続の認証に使用されるシークレットを指定します。それ以外の場合は、次のステップに進みます。
TLS 設定の例
# .... env: - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 1 value: my-cluster-cluster-ca-cert - name: STRIMZI_EO_KEY_SECRET_NAME 2 value: my-cluster-entity-operator-certs # ..."
User Operator をデプロイします。
oc create -f install/user-operator
デプロイメントのステータスを確認します。
oc get deployments
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE strimzi-user-operator 1/1 1 1
READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.2. Kafka Connect のデプロイ
Kafka Connect は、Apache Kafka と外部システムとの間でデータをストリーミングするためのツールです。
AMQ Streams では、Kafka Connect は分散 (distributed) モードでデプロイされます。Kafka Connect はスタンドアロンモードでも動作しますが、AMQ Streams ではサポートされません。
Kafka Connect では、コネクター の概念を使用し、スケーラビリティーと信頼性を維持しながら Kafka クラスターで大量のデータを出し入れするためのフレームワークが提供されます。
Kafka Connect は通常、Kafka を外部データベース、ストレージシステム、およびメッセージングシステムと統合するために使用されます。
本セクションの手順では以下の方法を説明します。
コネクター という用語は、Kafka Connect クラスター内で実行されているコネクターインスタンスや、コネクタークラスと同じ意味で使用されます。本ガイドでは、本文の内容で意味が明確である場合に コネクター という用語を使用します。
6.2.1. Kafka Connect の OpenShift クラスターへのデプロイ
この手順では、Cluster Operator を使用して Kafka Connect クラスターを OpenShift クラスターにデプロイする方法を説明します。
Kafka Connect クラスターは Deployment
として実装されます。その Deployment には、コネクターのワークロードを タスク として分布するノード (ワーカー とも呼ばれる) の設定可能な数が含まれるため、メッセージフローのスケーラビリティーや信頼性が高くなります。
デプロイメントでは、YAML ファイルの仕様を使って KafkaConnect
リソースが作成されます。
AMQ Streams には、設定ファイルのサンプルが含まれています。この手順では、以下のサンプルファイルを使用します。
-
examples/connect/kafka-connect.yaml
手順
Kafka Connect を OpenShift クラスターにデプロイします。
examples/connect/kafka-connect.yaml
ファイルを使用して Kafka Connect をデプロイします。oc apply -f examples/connect/kafka-connect.yaml
デプロイメントのステータスを確認します。
oc get deployments -n <my_cluster_operator_namespace>
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE my-connect-cluster-connect 1/1 1 1
my-connect-cluster
は、Kafka Connect クラスターの名前です。READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.2.2. 複数インスタンスの Kafka Connect 設定
Kafka Connect のインスタンスを複数実行している場合は、以下の config
プロパティーのデフォルト設定を変更する必要があります。
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect spec: # ... config: group.id: connect-cluster 1 offset.storage.topic: connect-cluster-offsets 2 config.storage.topic: connect-cluster-configs 3 status.storage.topic: connect-cluster-status 4 # ... # ...
これら 3 つのトピックの値は、同じ group.id
を持つすべての Kafka Connect インスタンスで同じする必要があります。
デフォルト設定を変更しないと、同じ Kafka クラスターに接続する各 Kafka Connect インスタンスは同じ値でデプロイされます。その結果、事実上はすべてのインスタンスが結合されてクラスターで実行され、同じトピックが使用されます。
複数の Kafka Connect クラスターが同じトピックの使用を試みると、Kafka Connect は想定どおりに動作せず、エラーが生成されます。
複数の Kafka Connect インスタンスを実行する場合は、インスタンスごとにこれらのプロパティーの値を変更してください。
6.2.3. コネクタープラグインでの Kafka Connect の拡張
Kafka Connect の AMQ Streams コンテナーイメージには、ファイルベースのデータを Kafka クラスターで出し入れするために 2 つの組み込みコネクターが含まれています。
ファイルコネクター | 説明 |
---|---|
| ファイル (ソース) から Kafka クラスターにデータを転送します。 |
| Kafka クラスターからファイル (シンク) にデータを転送します。 |
ここの手順では、以下を行って、独自のコネクタークラスをコネクターイメージに追加する方法を説明します。
Kafka Connect REST API または KafkaConnector
カスタムリソースを使用 して、直接コネクターの設定を作成します。
6.2.3.1. AMQ Streams を使用した新しいコンテナーイメージの自動作成
この手順では、AMQ Streams が追加のコネクターで新しいコンテナーイメージを自動的にビルドするように Kafka Connect を設定する方法を説明します。コネクタープラグインは、KafkaConnect
カスタムリソースの .spec.build.plugins
プロパティーを使用して定義します。AMQ Streams はコネクタープラグインを自動的にダウンロードし、新しいコンテナーイメージに追加します。コンテナーは、.spec.build.output
に指定されたコンテナーリポジトリーにプッシュされ、Kafka Connect デプロイメントで自動的に使用されます。
前提条件
- Cluster Operator がデプロイされている。
- コンテナーレジストリー。
イメージをプッシュ、保存、およびプルできる独自のコンテナーレジストリーを提供する必要があります。AMQ Streams は、プライベートコンテナーレジストリーだけでなく、Quay や Docker Hub などのパブリックレジストリーもサポートします。
手順
.spec.build.output
でコンテナーレジストリーを指定し、.spec.build.plugins
で追加のコネクターを指定して、KafkaConnect
カスタムリソースを設定します。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec: 1 #... build: output: 2 type: docker image: my-registry.io/my-org/my-connect-cluster:latest pushSecret: my-registry-credentials plugins: 3 - name: debezium-postgres-connector artifacts: - type: tgz url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03 - name: camel-telegram artifacts: - type: tgz url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479 #...
リソースを作成または更新します。
$ oc apply -f KAFKA-CONNECT-CONFIG-FILE
- 新しいコンテナーイメージがビルドされ、Kafka Connect クラスターがデプロイされるまで待ちます。
- Kafka Connect REST API または KafkaConnector カスタムリソースを使用して、追加したコネクタープラグインを使用します。
関連情報
詳細は、Using Strimzi ガイドを参照してください。
6.2.3.2. Kafka Connect ベースイメージからの Docker イメージの作成
この手順では、カスタムイメージを作成し、/opt/kafka/plugins
ディレクトリーに追加する方法を説明します。
Red Hat Ecosystem Catalog の Kafka コンテナーイメージを、追加のコネクタープラグインで独自のカスタムイメージを作成するためのベースイメージとして使用できます。
AMQ Stream バージョンの Kafka Connect は起動時に、/opt/kafka/plugins
ディレクトリーに含まれるサードパーティーのコネクタープラグインをロードします。
手順
registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0
をベースイメージとして使用して、新規のDockerfile
を作成します。FROM registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 USER root:root COPY ./my-plugins/ /opt/kafka/plugins/ USER 1001
プラグインファイルの例
$ tree ./my-plugins/ ./my-plugins/ ├── debezium-connector-mongodb │ ├── bson-3.4.2.jar │ ├── CHANGELOG.md │ ├── CONTRIBUTE.md │ ├── COPYRIGHT.txt │ ├── debezium-connector-mongodb-0.7.1.jar │ ├── debezium-core-0.7.1.jar │ ├── LICENSE.txt │ ├── mongodb-driver-3.4.2.jar │ ├── mongodb-driver-core-3.4.2.jar │ └── README.md ├── debezium-connector-mysql │ ├── CHANGELOG.md │ ├── CONTRIBUTE.md │ ├── COPYRIGHT.txt │ ├── debezium-connector-mysql-0.7.1.jar │ ├── debezium-core-0.7.1.jar │ ├── LICENSE.txt │ ├── mysql-binlog-connector-java-0.13.0.jar │ ├── mysql-connector-java-5.1.40.jar │ ├── README.md │ └── wkb-1.0.2.jar └── debezium-connector-postgres ├── CHANGELOG.md ├── CONTRIBUTE.md ├── COPYRIGHT.txt ├── debezium-connector-postgres-0.7.1.jar ├── debezium-core-0.7.1.jar ├── LICENSE.txt ├── postgresql-42.0.0.jar ├── protobuf-java-2.6.1.jar └── README.md
注記この例では、MongoDB、MySQL、および PostgreSQL 用の Debezium コネクターを使用します。Kafka Connect で実行されている Debezium は、他の Kafka Connect タスクと同じように表示されます。
- コンテナーイメージをビルドします。
- カスタムイメージをコンテナーレジストリーにプッシュします。
新しいコンテナーイメージを示します。
以下のいずれかを行います。
KafkaConnect
カスタムリソースのKafkaConnect.spec.image
プロパティーを編集します。設定された場合、このプロパティーによって Cluster Operator の
STRIMZI_KAFKA_CONNECT_IMAGES
変数がオーバーライドされます。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec: 1 #... image: my-new-container-image 2 config: 3 #...
または
-
install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml
ファイルのSTRIMZI_KAFKA_CONNECT_IMAGES
変数を編集して新しいコンテナーイメージを示すようにした後、Cluster Operator を再インストールします。
6.2.4. コネクターの作成および管理
コネクタープラグインのコンテナーイメージを作成したら、Kafka Connect クラスターにコネクターインスタンスを作成する必要があります。その後、稼働中のコネクターインスタンスを設定、監視、および管理できます。
コネクターは特定の コネクタークラス のインスタンスで、メッセージに関して関連する外部システムとの通信方法を認識しています。コネクターは多くの外部システムで使用でき、独自のコネクターを作成することもできます。
ソース および シンク タイプのコネクターを作成できます。
- ソースコネクター
- ソースコネクターは、外部システムからデータを取得し、それをメッセージとして Kafka に提供するランタイムエンティティーです。
- シンクコネクター
- シンクコネクターは、Kafka トピックからメッセージを取得し、外部システムに提供するランタイムエンティティーです。
6.2.4.1. コネクターの作成および管理用の API
AMQ Streams では、コネクターの作成および管理に 2 つの API が提供されます。
-
KafkaConnector
カスタムリソース (KafkaConnectors と呼ばれます) - Kafka Connect REST API
API を使用すると、以下を行うことができます。
- コネクターインスタンスのステータスの確認。
- 稼働中のコネクターの再設定。
- コネクターインスタンスのコネクタータスク数の増減。
- コネクターの再起動。
- 失敗したタスクを含むコネクタータスクの再起動。
- コネクターインスタンスの一時停止。
- 一時停止したコネクターインスタンスの再開。
- コネクターインスタンスの削除。
KafkaConnector カスタムリソース
KafkaConnectors を使用すると、Kafka Connect のコネクターインスタンスを OpenShift ネイティブに作成および管理できるため、cURL などの HTTP クライアントが必要ありません。その他の Kafka リソースと同様に、コネクターの望ましい状態を OpenShift クラスターにデプロイされた KafkaConnector
YAML ファイルに宣言し、コネクターインスタンスを作成します。KafkaConnector
リソースは、リンク先の Kafka Connect クラスターと同じ namespace にデプロイする必要があります。
該当する KafkaConnector
リソースを更新して稼働中のコネクターインスタンスを管理した後、更新を適用します。該当する KafkaConnector
を削除して、コネクターを削除します。
下位バージョンの AMQ Streams との互換性を維持するため、KafkaConnectors はデフォルトで無効になっています。Kafka Connect クラスターの KafkaConnectors を有効にするには、KafkaConnect
リソースで strimzi.io/use-connector-resources
アノテーションを true
に設定します。手順は、Configuring Kafka Connect を参照してください。
KafkaConnectors が有効になると、Cluster Operator によって監視が開始されます。KafkaConnectors に定義された設定と一致するよう、稼働中のコネクターインスタンスの設定を更新します。
AMQ Streams には KafkaConnector
設定ファイルの例が含まれます。これをもとに、 を使用して FileStreamSourceConnector
と FileStreamSinkConnector
を作成、管理できます。
KafkaConnector
リソースにアノテーションを付けて、コネクターの再起動 または コネクタータスクの再起動 ができます。
Kafka Connect API
Kafka Connect REST API でサポートされる操作は、Apache Kafka Connect API のドキュメント で説明されています。
Kafka Connect API の使用から KafkaConnectors の使用への切り替え
Kafka Connect API の使用から KafkaConnectors の使用に切り替えると、コネクターを管理できます。スイッチの作成は、次の作業を以下の順序で行います。
-
設定で
KafkaConnector
リソースをデプロイし、コネクターインスタンスを作成します。 -
strimzi.io/use-connector-resources
アノテーションをtrue
に設定して、Kafka Connect 設定で KafkaConnectors を有効にします。
リソースを作成する前に KafkaConnectors を有効にすると、すべてのコネクターが削除されます。
KafkaConnectors の使用から Kafka Connect API の使用に切り替えるには、まず Kafka Connect 設定から KafkaConnectors を有効にするアノテーションを削除します。それ以外の場合、Kafka Connect REST API を使用して直接行われた手動による変更は、 Cluster Operator によって元に戻されます。
6.2.4.2. サンプル KafkaConnector リソースのデプロイ
Kafka Connect で KafkaConnectors を使用し、Kafka クラスターとの間でデータをストリーミングします。
AMQ Streams には、設定ファイルのサンプルが含まれています。この手順では、以下のサンプルファイルを使用します。
-
examples/connect/source-connector.yaml
このファイルは、以下のコネクターインスタンスの作成に使用されます。
-
Kafka ライセンスファイル(ソース)から各行を読み取り、データをメッセージとして単一の Kafka トピックに書き込む
FileStreamSourceConnector
インスタンス。 -
Kafka トピックからメッセージを読み取り、メッセージを一時ファイル (シンク) に書き込む
FileStreamSinkConnector
インスタンス。
実稼働環境で、「コネクタープラグインでの Kafka Connect の拡張」 の説明どおりに、必要な Kafka Connect コネクターが含まれるコンテナーイメージを準備します。
FileStreamSourceConnector
および FileStreamSinkConnector
が例として提供されています。ここで説明するように、コンテナーでこれらのコネクターを実行することは、実稼働のユースケースには適していません。
前提条件
- Kafka Connect デプロイメント。
- Kafka Connect デプロイメントで KafkaConnectors が有効になっている。
- Cluster Operator が稼働している必要があります。
手順
examples/connect/source-connector.yaml
ファイルを編集します。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-source-connector 1 labels: strimzi.io/cluster: my-connect-cluster 2 spec: class: org.apache.kafka.connect.file.FileStreamSourceConnector 3 tasksMax: 2 4 config: 5 file: "/opt/kafka/LICENSE" 6 topic: my-topic 7 # ...
- 1
- コネクターの名前として使用される
KafkaConnector
リソースの名前。OpenShift リソースで有効な名前を使用します。 - 2
- コネクターインスタンスを作成する Kafka Connect クラスターの名前。コネクターは、リンク先の Kafka Connect クラスターと同じ namespace にデプロイする必要があります。
- 3
- コネクタークラスのフルネームまたはエイリアス。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。
- 4
- コネクターが作成できる Kafka Connect
Tasks
の最大数。 - 5
- キーと値のペアとしてのコネクター設定。
- 6
- このサンプルソースコネクター設定では、
/opt/kafka/LICENSE
ファイルからデータが読み取られます。 - 7
- ソースデータのパブリッシュ先となる Kafka トピック。
OpenShift クラスターでソース
KafkaConnector
を作成します。oc apply -f examples/connect/source-connector.yaml
examples/connect/sink-connector.yaml
ファイルを作成します。touch examples/connect/sink-connector.yaml
以下の YAML を
sink-connector.yaml
ファイルに貼り付けます。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-sink-connector labels: strimzi.io/cluster: my-connect spec: class: org.apache.kafka.connect.file.FileStreamSinkConnector 1 tasksMax: 2 config: 2 file: "/tmp/my-file" 3 topics: my-topic 4
OpenShift クラスターにシンク
KafkaConnector
を作成します。oc apply -f examples/connect/sink-connector.yaml
コネクターリソースが作成されたことを確認します。
oc get kctr --selector strimzi.io/cluster=MY-CONNECT-CLUSTER -o name my-source-connector my-sink-connector
MY-CONNECT-CLUSTER を Kafka Connect クラスターに置き換えます。
コンテナーで、
kafka -console-consumer.sh
を実行して、ソースコネクターによってトピックに書き込まれたメッセージを読み取ります。oc exec MY-CLUSTER-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server MY-CLUSTER-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning
ソースおよびシンクコネクターの設定オプション
コネクター設定は、KafkaConnector
リソースの spec.config
プロパティーで定義されます。
FileStreamSourceConnector
クラスおよび FileStreamSinkConnector
クラスは、Kafka Connect REST API と同じ設定オプションをサポートします。他のコネクターは異なる設定オプションをサポートします。
名前 | タイプ | デフォルト値 | 説明 |
---|---|---|---|
| 文字列 | Null | メッセージを書き込むソースファイル。指定のない場合は、標準入力が使用されます。 |
| リスト | Null | データのパブリッシュ先となる Kafka トピック。 |
名前 | タイプ | デフォルト値 | 説明 |
---|---|---|---|
| 文字列 | Null | メッセージを書き込む宛先ファイル。指定のない場合は標準出力が使用されます。 |
| リスト | Null | データの読み取り元となる 1 つ以上の Kafka トピック。 |
| 文字列 | Null | データの読み取り元となる 1 つ以上の Kafka トピックと一致する正規表現。 |
6.2.4.3. Kafka コネクターの再起動の実行
この手順では、OpenShift アノテーションを使用して Kafka コネクターの再起動を手動でトリガーする方法を説明します。
前提条件
- Cluster Operator が稼働している必要があります。
手順
再起動する Kafka コネクターを制御する
KafkaConnector
カスタムリソースの名前を見つけます。oc get KafkaConnector
コネクターを再起動するには、OpenShift で
KafkaConnector
リソースにアノテーションを付けます。たとえば、oc annotate
を使用すると以下のようになります。oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart=true
次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。
アノテーションが調整プロセスで検出されれば、Kafka コネクターは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは
KafkaConnector
カスタムリソースから削除されます。
6.2.4.4. Kafka コネクタータスクの再起動の実行
この手順では、OpenShift アノテーションを使用して Kafka コネクタータスクの再起動を手動でトリガーする方法を説明します。
前提条件
- Cluster Operator が稼働している必要があります。
手順
再起動する Kafka コネクタータスクを制御する
KafkaConnector
カスタムリソースの名前を見つけます。oc get KafkaConnector
KafkaConnector
カスタムリソースから再起動するタスクの ID を検索します。タスク ID は 0 から始まる負の値ではない整数です。oc describe KafkaConnector KAFKACONNECTOR-NAME
コネクタータスクを再起動するには、OpenShift で
KafkaConnector
リソースにアノテーションを付けます。たとえば、oc annotate
を使用してタスク 0 を再起動します。oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart-task=0
次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。
アノテーションが調整プロセスで検出されれば、Kafka コネクタータスクは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは
KafkaConnector
カスタムリソースから削除されます。
6.2.4.5. Kafka Connect API の公開
KafkaConnector
リソースを使用してコネクターを管理する代わりに、Kafka Connect REST API を使用します。Kafka Connect REST API は、<connect_cluster_name>-connect-api:8083
で実行しているサービスとして利用できます。ここで、<connect_cluster_name> は、お使いの Kafka Connect クラスターの名前になります。サービスは、Kafka Connect インスタンスの作成時に作成されます。
strimzi.io/use-connector-resources
アノテーションは KafkaConnectors を有効にします。アノテーションを KafkaConnect
リソース設定に適用した場合、そのアノテーションを削除して Kafka Connect API を使用する必要があります。それ以外の場合、Kafka Connect REST API を使用して直接行われた手動による変更は、 Cluster Operator によって元に戻されます。
コネクター設定を JSON オブジェクトとして追加できます。
コネクター設定を追加するための curl 要求の例
curl -X POST \ http://my-connect-cluster-connect-api:8083/connectors \ -H 'Content-Type: application/json' \ -d '{ "name": "my-source-connector", "config": { "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector", "file": "/opt/kafka/LICENSE", "topic":"my-topic", "tasksMax": "4", "type": "source" } }'
API には OpenShift クラスター内でのみアクセスできます。OpenShift クラスター外部で実行しているアプリケーションに Kafka Connect API がアクセスできるようにする場合は、以下の機能のいずれかを使用して Kafka Connect API を手動で公開できます。
-
LoadBalancer
またはNodePort
タイプのサービス -
Ingress
リソース - OpenShift ルート
接続は安全ではないため、外部からのアクセスはよく考えてから許可してください。
サービスを作成する場合には、<connect_cluster_name>-connect-api
サービスの selector
からラベルを使用して、サービスがトラフィックをルーティングする Pod を設定します。
サービスのセレクター設定
# ... selector: strimzi.io/cluster: my-connect-cluster 1 strimzi.io/kind: KafkaConnect strimzi.io/name: my-connect-cluster-connect 2 #...
また、外部クライアントからの HTTP 要求を許可する NetworkPolicy
を作成する必要もあります。
Kafka Connect API への要求を許可する NetworkPolicy の例
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: my-custom-connect-network-policy
spec:
ingress:
- from:
- podSelector: 1
matchLabels:
app: my-connector-manager
ports:
- port: 8083
protocol: TCP
podSelector:
matchLabels:
strimzi.io/cluster: my-connect-cluster
strimzi.io/kind: KafkaConnect
strimzi.io/name: my-connect-cluster-connect
policyTypes:
- Ingress
- 1
- API への接続が許可される Pod のラベル。
クラスター外でコネクター設定を追加するには、curl コマンドで API を公開するリソースの URL を使用します。
6.3. Kafka MirrorMaker のデプロイ
Cluster Operator によって、1 つ以上の Kafka MirrorMaker のレプリカがデプロイされ、Kafka クラスターの間でデータが複製されます。このプロセスはミラーリングと言われ、Kafka パーティションのレプリケーションの概念と混同しないようにします。MirrorMaker は、ソースクラスターからメッセージを消費し、これらのメッセージをターゲットクラスターにパブリッシュします。
6.3.1. Kafka MirrorMaker の OpenShift クラスターへのデプロイ
この手順では、Cluster Operator を使用して Kafka MirrorMaker クラスターを OpenShift クラスターにデプロイする方法を説明します。
デプロイメントでは、YAML ファイルの仕様を使って、デプロイされた MirrorMaker のバージョンに応じて KafkaMirrorMaker
または KafkaMirrorMaker2
リソースが作成されます。
Kafka MirrorMaker 1 (ドキュメントでは単に MirrorMaker と呼ばれる) は Apache Kafka 3.0.0 で非推奨となり、Apache Kafka 4.0.0 で削除されます。そのため、Kafka MirrorMaker 1 のデプロイに使用される KafkaMirrorMaker
カスタムリソースも、AMQ Streams で非推奨となりました。Apache Kafka 4.0.0 を導入すると、KafkaMirrorMaker
リソースは AMQ Streams から削除されます。代わりに、IdentityReplicationPolicy
で KafkaMirrorMaker2
カスタムリソースを使用します。
AMQ Streamsは設定ファイルの例を提供しています。この手順では、以下のサンプルファイルを使用します。
-
examples/mirror-maker/kafka-mirror-maker.yaml
-
examples/mirror-maker/kafka-mirror-maker-2.yaml
手順
Kafka MirrorMaker を OpenShift クラスターにデプロイします。
MirrorMaker の場合
oc apply -f examples/mirror-maker/kafka-mirror-maker.yaml
MirrorMaker 2.0 の場合
oc apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml
デプロイメントのステータスを確認します。
oc get deployments -n <my_cluster_operator_namespace>
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE my-mirror-maker-mirror-maker 1/1 1 1 my-mm2-cluster-mirrormaker2 1/1 1 1
my-mirror-maker
は Kafka MirrorMaker クラスターの名前で、my-mm2-cluster
は、Kafka MirrorMaker 2.0 クラスターの名前です。READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.4. Kafka Bridge のデプロイ
Cluster Operator によって、1 つ以上の Kafka Bridge のレプリカがデプロイされ、HTTP API 経由で Kafka クラスターとクライアントの間でデータが送信されます。
6.4.1. Kafka Bridge を OpenShift クラスターへデプロイ
この手順では、Cluster Operator を使用して Kafka Bridge クラスターを OpenShift クラスターにデプロイする方法を説明します。
デプロイメントでは、YAML ファイルの仕様を使って KafkaBridge
リソースが作成されます。
AMQ Streams には、設定ファイルのサンプルが含まれています。この手順では、以下のサンプルファイルを使用します。
-
examples/bridge/kafka-bridge.yaml
手順
Kafka Bridge を OpenShift クラスターにデプロイします。
oc apply -f examples/bridge/kafka-bridge.yaml
デプロイメントのステータスを確認します。
oc get deployments -n <my_cluster_operator_namespace>
出力には、デプロイメント名と準備状態が表示されます。
NAME READY UP-TO-DATE AVAILABLE my-bridge-bridge 1/1 1 1
my-bridge
は、Kafka Bridge クラスターの名前です。READY
は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE
出力に1
が表示されれば、デプロイメントは成功しています。
6.4.2. Kafka Bridge サービスのローカルマシンへの公開
ポート転送を使用して AMQ Streams の Kafka Bridge サービスを http://localhost:8080 上でローカルマシンに公開します。
ポート転送は、開発およびテストの目的でのみ適切です。
手順
OpenShift クラスターの Pod の名前をリストします。
oc get pods -o name pod/kafka-consumer # ... pod/quickstart-bridge-589d78784d-9jcnr pod/strimzi-cluster-operator-76bcf9bc76-8dnfm
ポート
8080
で Kafka Bridge Pod に接続します。oc port-forward pod/quickstart-bridge-589d78784d-9jcnr 8080:8080 &
注記ローカルマシンのポート 8080 がすでに使用中の場合は、代わりの HTTP ポート (
8008
など) を使用します。
これで、API リクエストがローカルマシンのポート 8080 から Kafka Bridge Pod のポート 8080 に転送されるようになります。
6.4.3. OpenShift 外部の Kafka Bridge へのアクセス
デプロイメント後、AMQ Streams Kafka Bridge には同じ OpenShift クラスターで実行しているアプリケーションのみがアクセスできます。これらのアプリケーションは、<kafka_bridge_name>-bridge-service
サービスを使用して API にアクセスします。
OpenShift クラスター外部で実行しているアプリケーションに Kafka Bridge がアクセスできるようにする場合は、以下の機能のいずれかを作成して Kafka Bridge を手動で公開できます。
-
LoadBalancer
またはNodePort
タイプのサービス -
Ingress
リソース - OpenShift ルート
サービスを作成する場合には、<kafka_bridge_name>-bridge-service
サービスの selector
からラベルを使用して、サービスがトラフィックをルーティングする Pod を設定します。
# ...
selector:
strimzi.io/cluster: kafka-bridge-name 1
strimzi.io/kind: KafkaBridge
#...
- 1
- OpenShift クラスターでの Kafka Bridge カスタムリソースの名前。
第7章 Kafka クラスターへのクライアントアクセスの設定
AMQ Streams のデプロイ 後、本章では以下の操作を行う方法について説明します。
- サンプルプロデューサーおよびコンシューマークライアントをデプロイし、これを使用してデプロイメントを検証する
Kafka クラスターへの外部クライアントアクセスを設定する
OpenShift 外部のクライアントに Kafka クラスターへのアクセスを設定する手順はより複雑で、Kafka コンポーネントの設定手順 に精通している必要があります。
7.1. サンプルクライアントのデプロイ
この手順では、ユーザーが作成した Kafka クラスターを使用してメッセージを送受信するプロデューサーおよびコンシューマークライアントの例をデプロイする方法を説明します。
前提条件
- クライアントが Kafka クラスターを使用できる必要があります。
手順
Kafka プロデューサーをデプロイします。
oc run kafka-producer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list cluster-name-kafka-bootstrap:9092 --topic my-topic
- プロデューサーが稼働しているコンソールにメッセージを入力します。
- Enter を押してメッセージを送信します。
Kafka コンシューマーをデプロイします。
oc run kafka-consumer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server cluster-name-kafka-bootstrap:9092 --topic my-topic --from-beginning
- コンシューマーコンソールに受信メッセージが表示されることを確認します。
7.2. OpenShift 外クライアントのアクセスの設定
以下の手順では、OpenShift 外部からの Kafka クラスターへのクライアントアクセスを設定する方法を説明します。
Kafka クラスターのアドレスを使用して、異なる OpenShift namespace または完全に OpenShift 外のクライアントに外部アクセスを提供できます。
アクセスを提供するために、外部 Kafka リスナーを設定します。
以下のタイプの外部リスナーがサポートされます。
-
OpenShift
Route
およびデフォルトの HAProxy ルーターを使用するroute
-
ロードバランサーサービスを使用する
loadbalancer
-
OpenShift ノードのポートを使用する
nodeport
-
OpenShift Ingress と NGINX Ingress Controller for Kubernetes を使用する
ingress
要件ならびにお使いの環境およびインフラストラクチャーに応じて、選択するタイプは異なります。たとえば、ロードバランサーは、ベアメタル等の特定のインフラストラクチャーには適さない場合があります。ベアメタルでは、ノードポートがより適したオプションを提供します。
以下の手順では、
- TLS 暗号化および認証、ならびに Kafka 簡易承認 を有効にして、Kafka クラスターに外部リスナーが設定されます。
-
簡易承認 用に TLS 認証および アクセス制御リスト (ACL) を定義して、クライアントに
KafkaUser
が作成されます。
TLS、SCRAM-SHA-512、または OAuth 2.0 認証を使用するようにリスナーを設定できます。TLS は常に暗号化を使用しますが、SCRAM-SHA-512 および OAuth 2.0 認証でも暗号化を使用することが推奨されます。
Kafka ブローカーにシンプルな、OAuth 2.0、OPA、またはカスタム承認を設定できます。承認を有効にすると、承認は有効なすべてのリスナーに適用されます。
KafkaUser
認証および承認メカニズムを設定する場合、必ず同等の Kafka 設定と一致するようにしてください。
-
KafkaUser.spec.authentication
はKafka.spec.kafka.listeners[*].authentication
と一致します。 -
KafkaUser.spec.authorization
はKafka.spec.kafka.authorization
と一致します。
KafkaUser
に使用する認証をサポートするリスナーが少なくとも 1 つ必要です。
Kafka ユーザーと Kafka ブローカー間の認証は、それぞれの認証設定によって異なります。たとえば、TLS が Kafka 設定で有効になっていない場合は、TLS でユーザーを認証できません。
AMQ Streams Operator により設定プロセスが自動されます。
- Cluster Operator はリスナーを作成し、クラスターおよびクライアント認証局 (CA) 証明書を設定して Kafka クラスター内で認証を有効にします。
- User Operator はクライアントに対応するユーザーを作成すると共に、選択した認証タイプに基づいて、クライアント認証に使用されるセキュリティークレデンシャルを作成します。
この手順では、Cluster Operator によって生成された証明書が使用されますが、独自の証明書をインストール してそれらを置き換えることができます。外部認証局によって管理される Kafka リスナー証明書を使用するようにリスナーを設定することもできます。
PKCS #12 (.p12) 形式および PEM (.crt) 形式の証明書を利用できます。この手順では、PKCS #12 証明書を説明します。
前提条件
- クライアントが Kafka クラスターを使用できる必要があります。
- Cluster Operator および User Operator がクラスターで実行されている必要があります。
- OpenShift クラスター外のクライアントが Kafka クラスターに接続できる必要があります。
手順
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 文字を超えないように注意してください。
Kafka
リソースを作成または更新します。oc apply -f <kafka_configuration_file>
Kafka クラスターは、TLS 認証を使用する Kafka ブローカーリスナーと共に設定されます。
Kafka ブローカー Pod ごとにサービスが作成されます。
サービスが作成され、Kafka クラスターに接続するための ブートストラップアドレス として機能します。
サービスは、
nodeport
リスナーを使用した Kafka クラスターへの外部接続用 外部ブートストラップアドレス としても作成されます。kafka ブローカーのアイデンティティーを検証するクラスター CA 証明書もシークレット
<cluster_name>-cluster-ca-cert
に作成されます。注記外部リスナーの使用時に Kafka クラスターをスケーリングする場合に、すべての Kafka ブローカーのローリングアップデートがトリガーされる可能性があります。これは設定によって異なります。
Kafka
リソースのステータスからブートストラップアドレスおよびポートを見つけます。oc get kafka KAFKA-CLUSTER-NAME -o jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}'
Kafka クライアントのブートストラップアドレスを使用して、Kafka クラスターに接続します。
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
-
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
パブリッククラスター CA 証明書を必要な証明書形式に展開します。
oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12
パスワードファイルからパスワードを抽出します。
oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password
パブリッククラスター証明書の認証情報でクライアントを設定します。
クライアントコードのサンプル
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
注記セキュリティープロトコルを使用します。
(TLS で SCRAM-SHA 認証を使用する場合)ユーザー Secret から必要な証明書形式にユーザー CA 証明書を抽出します。
oc get secret USER-NAME -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12
パスワードファイルからパスワードを抽出します。
oc get secret USER-NAME -o jsonpath='{.data.user\.password}' | base64 -d > user.password
ユーザー 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
Kafka クラスターに接続するためのブートストラップアドレスおよびポートを追加します。
bootstrap.servers: BOOTSTRAP-ADDRESS:PORT
関連情報
- Listener authentication options
- Kafka authorization options
- 承認サーバーを使用している場合は、トークンベースの OAuth 2.0 認証 および OAuth 2.0 承認 を使用できます。
第8章 AMQ Streams のメトリクスおよびダッシュボードの設定
Prometheus および Grafana を使用して、AMQ Streams デプロイメントを監視できます。
ダッシュボードでキーメトリクスを表示し、特定の条件下でトリガーされるアラートを設定すると、AMQ Streams デプロイメントを監視できます。メトリクスは、AMQ Streams の各コンポーネントで利用できます。
AMQ Streams は、メトリクス情報を提供するために、Prometheus ルールと Grafana ダッシュボードを使用します。
Prometheus に AMQ Streams の各コンポーネントのルールセットが設定されている場合、Prometheus はクラスターで稼働している Pod からキーメトリクスを使用します。次に、Grafana はこれらのメトリクスをダッシュボードで可視化します。AMQ Streams には、デプロイメントに合わせてカスタマイズできる Grafana ダッシュボードのサンプルが含まれています。
AMQ Streams は、ユーザー定義プロジェクトのモニタリング (OpenShift 機能) を使用して、Prometheus の設定プロセスを単純化します。
要件に応じて以下を行うことができます。
Prometheus および Grafana が設定されると、監視に AMQ Streams が提供する Grafana ダッシュボードのサンプルを使用できます。
さらに、分散トレーシングを設定して、エンドツーエンドのメッセージ追跡を行うようにデプロイメントを設定することもできます。
AMQ Streams は、Prometheus と Grafana のインストールファイルの例を提供します。AMQ Streams の監視を試みる際に、このファイルを開始点として使用できます。さらにサポートするには、Prometheus および Grafana 開発者コミュニティーに参加してみてください。
メトリクスおよびモニタリングツールのサポートドキュメント
メトリクスおよびモニタリングツールの詳細は、サポートドキュメントを参照してください。
- Prometheus
- Prometheus の設定
- Kafka Exporter
- Grafana Labs
- Apache Kafka Monitoring では、Apache Kafka により公開される JMX メトリクスについて解説しています。
- ZooKeeper JMX では、Apache Zookeeper により公開される JMX メトリックについて解説しています。
8.1. Kafka Exporter でのコンシューマーラグの監視
Kafka Exporter は、Apache Kafka ブローカーおよびクライアントの監視を強化するオープンソースプロジェクトです。Kafka クラスターで Kafka Exporter をデプロイ するように、Kafka
リソースを設定できます。Kafka Exporter は、オフセット、コンシューマーグループ、コンシューマーラグ、およびトピックに関連する Kafka ブローカーから追加のメトリクスデータを抽出します。一例として、メトリクスデータを使用すると、低速なコンシューマーの識別に役立ちます。ラグデータは Prometheus メトリクスとして公開され、解析のために Grafana で使用できます。
Kafka Exporter は、コンシューマーラグおよびコンシューマーオフセットに関連する追加のメトリクスのみを提供します。通常の Kafka メトリクスでは、Kafka ブローカー で、Prometheus メトリクスを設定する必要があります。
コンシューマーラグは、メッセージの生成と消費の差を示しています。具体的には、指定のコンシューマーグループのコンシューマーラグは、パーティションの最後のメッセージと、そのコンシューマーが現在ピックアップしているメッセージとの時間差を示しています。
ラグには、パーティションログの最後を基準とする、コンシューマーオフセットの相対的な位置が反映されます。
プロデューサーおよびコンシューマーオフセット間のコンシューマーラグ
この差は、Kafka ブローカートピックパーティションの読み取りと書き込みの場所である、プロデューサーオフセットとコンシューマーオフセットの間の デルタ とも呼ばれます。
あるトピックで毎秒 100 個のメッセージがストリーミングされる場合を考えてみましょう。プロデューサーオフセット (トピックパーティションの先頭) と、コンシューマーが読み取った最後のオフセットとの間のラグが 1000 個のメッセージであれば、10 秒の遅延があることを意味します。
コンシューマーラグ監視の重要性
可能な限りリアルタイムのデータの処理に依存するアプリケーションでは、コンシューマーラグを監視して、ラグが過度に大きくならないようにチェックする必要があります。ラグが大きくなるほど、リアルタイム処理の達成から遠ざかります。
たとえば、パージされていない古いデータの大量消費や、予定外のシャットダウンが、コンシューマーラグの原因となることがあります。
コンシューマーラグの削減
Grafana のチャートを使用して、ラグを分析し、ラグ削減の方法が対象のコンシューマーグループに影響しているかどうかを確認します。たとえば、ラグを減らすように Kafka ブローカーを調整すると、ダッシュボードには コンシューマーグループごとのラグ のチャートが下降し 毎分のメッセージ消費 のチャートが上昇する状況が示されます。
通常、ラグを削減するには以下を行います。
- 新規コンシューマーを追加してコンシューマーグループをスケールアップします。
- メッセージがトピックに留まる保持時間を延長します。
- ディスク容量を追加してメッセージバッファーを増強します。
コンシューマーラグを減らす方法は、基礎となるインフラストラクチャーや、AMQ Streams によりサポートされるユースケースによって異なります。たとえば、ラグが生じているコンシューマーの場合、ディスクキャッシュからフェッチリクエストに対応できるブローカーを活用できる可能性は低いでしょう。場合によっては、コンシューマーの状態が改善されるまで、自動的にメッセージをドロップすることが許容されることがあります。
8.2. Cruise Control 操作の監視
Cruise Control は、ブローカー、トピック、およびパーティションの使用状況を追跡するために Kafka ブローカーを監視します。Cruise Control は、独自のパフォーマンスを監視するためのメトリクスのセットも提供します。
Cruise Control メトリクスレポーターは、Kafka ブローカーから未加工のメトリクスデータを収集します。データは、Cruise Control によって自動的に作成されるトピックに生成されます。メトリクスは、Kafka クラスターの最適化提案の生成 に使用されます。
Cruise Control メトリクスは、Cruise Control 操作のリアルタイム監視で利用できます。たとえば、Cruise Control メトリクスを使用して、実行中のリバランス操作のステータスを監視したり、操作のパフォーマンスで検出された異常についてアラートを提供したりできます。
Cruise Control 設定で Prometheus JMX Exporter を有効にして Cruise Control メトリクスを公開します。
センサー として知られる利用可能な Cruise Control メトリクスの完全なリストは、Cruise Control のドキュメントを 参照してください。
8.2.1. Cruise Control メトリクスの公開
Cruise Control 操作でメトリクスを公開する場合は、Kafka
リソースを設定して、Cruise Control をデプロイし、デプロイメントで Prometheus メトリクスを有効にします。独自の設定を使用するか、AMQ Streams によって提供される kafka-cruise-control-metrics.yaml
ファイルのサンプルを使用できます。
設定を Kafka
リソースの CruiseControl
プロパティーの metricsConfig
に追加します。この設定により、Prometheus JMX Exporter が有効化され、HTTP エンドポイント経由で Cruise Control メトリクスが公開されます。HTTP エンドポイントは Prometheus サーバーによってスクレープされます。
Cruise Control のメトリクス設定例
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
name: my-cluster
Spec:
# ...
cruiseControl:
# ...
metricsConfig:
type: jmxPrometheusExporter
valueFrom:
configMapKeyRef:
name: cruise-control-metrics
key: metrics-config.yml
---
kind: ConfigMap
apiVersion: v1
metadata:
name: cruise-control-metrics
labels:
app: strimzi
data:
metrics-config.yml: |
# metrics configuration...
8.2.2. Cruise Control メトリクスの表示
Cruise Control メトリクスを公開したら、Prometheus または別の適切なモニタリングシステムを使用して、メトリクスデータの情報を表示することができます。Streams for Apache Kafka は、Cruise Control メトリクスの視覚化を表示する Grafana ダッシュボードのサンプル を提供します。ダッシュボードは strimzi-cruise-control.json
という名前の JSON ファイルです。公開されるメトリクスは、Grafana ダッシュボードを有効にする 際に監視データを提供します。
8.2.2.1. 分散スコアの監視
Cruise Control メトリクスには、分散スコアが含まれます。分散度は、Kafka クラスター内でワークロードがどの程度均等に分散されているかを示す尺度です。
分散度スコア (balancedness-score
) の Cruise Control メトリクスは、KafkaRebalance
リソースの分散度スコアとは異なる可能性があります。Cruise Control は anomaly.detection.goals
を使用して各スコアを計算します。これは、KafkaRebalance
リソースで使用される default.goals
と同じでない可能性があります。anomaly.detection.goals
は、Kafka
カスタムリソースの spec.cruiseControl.config
に指定されます。
KafkaRebalance
リソースを更新すると、最適化プロポーザルをフェッチします。以下の条件のいずれかが適用されると、キャッシュされた最新の最適化プロポーザルがフェッチされます。
-
KafkaRebalance
goals
は、Kafka
リソースのdefault.goals
セクションに設定されたゴールと一致します。 -
KafkaRebalance
goals
は指定されていません。
これ以外の場合は、Cruise Control は KafkaRebalance goals
に基づいて、新しい最適化プロポーザルを生成します。更新ごとに新しいプロポーザルが生成されると、パフォーマンスの監視に影響を及ぼす可能性があります。
8.2.2.2. 異常検出へのアラート
Cruise Control の 異常検出 は、ブローカーの障害などの最適化ゴールの生成をブロックする条件のメトリクスデータを提供します。可視性を高めたい場合は、異常検出器が提供するメトリックを使用して、アラートを設定し、通知を送信できます。Cruise Control の 異常通知機能 を設定して、指定された通知チャネルを介してこれらのメトリクスに基づいてアラートをルーティングできます。または、Prometheus を設定して、異常検出器によって提供されるメトリクスデータをスクレープし、アラートを生成することもできます。その後、Prometheus Alertmanager は Prometheus で生成されるアラートをルーティングできます。
Cruise Control ドキュメント には、AnomalyDetector
メトリクスおよび異常通知機能に関する情報が記載されています。
8.3. メトリクスファイルの例
Grafana ダッシュボードおよびその他のメトリクス設定ファイルの例は、AMQ Streams によって提供される 設定ファイルの例 を参照してください。
AMQ Streams で提供されるサンプルメトリクスファイル
metrics ├── grafana-dashboards 1 │ ├── strimzi-cruise-control.json │ ├── strimzi-kafka-bridge.json │ ├── strimzi-kafka-connect.json │ ├── strimzi-kafka-exporter.json │ ├── strimzi-kafka-mirror-maker-2.json │ ├── strimzi-kafka.json │ ├── strimzi-operators.json │ └── strimzi-zookeeper.json ├── grafana-install │ └── grafana.yaml 2 ├── prometheus-additional-properties │ └── prometheus-additional.yaml 3 ├── prometheus-alertmanager-config │ └── alert-manager-config.yaml 4 ├── prometheus-install │ ├── alert-manager.yaml 5 │ ├── prometheus-rules.yaml 6 │ ├── prometheus.yaml 7 │ ├── strimzi-pod-monitor.yaml 8 ├── kafka-bridge-metrics.yaml 9 ├── kafka-connect-metrics.yaml 10 ├── kafka-cruise-control-metrics.yaml 11 ├── kafka-metrics.yaml 12 └── kafka-mirror-maker-2-metrics.yaml 13
- 1
- 異なる AMQ Streams コンポーネントの Grafana ダッシュボードの例。
- 2
- Grafana イメージのインストールファイル。
- 3
- CPU、メモリー、およびディスクボリュームの使用状況についてのメトリクスをスクレープする追加の設定。これらのメトリクスは、ノード上の OpenShift cAdvisor エージェントおよび kubelet から直接提供されます。
- 4
- Alertmanager による通知送信のためのフック定義。
- 5
- Alertmanager をデプロイおよび設定するためのリソース。
- 6
- Prometheus Alertmanager と使用するアラートルールの例 (Prometheus とデプロイ)。
- 7
- Prometheus イメージのインストールリソースファイル。
- 8
- Prometheus Operator によって Prometheus サーバーのジョブに変換される PodMonitor の定義。これにより、Pod から直接メトリクスデータをスクレープできます。
- 9
- メトリクスが有効になっている Kafka Bridge リソース。
- 10
- Kafka Connect に対する Prometheus JMX Exporter の再ラベル付けルールを定義するメトリクス設定。
- 11
- Cruise Control に対する Prometheus JMX Exporter の再ラベル付けルールを定義するメトリクス設定。
- 12
- Kafka および ZooKeeper に対する Prometheus JMX Exporter の再ラベル付けルールを定義するメトリクス設定。
- 13
- Kafka Mirror Maker 2.0 に対する Prometheus JMX Exporter の再ラベル付けルールを定義するメトリクス設定。
8.3.1. Prometheus メトリクス設定の例
AMQ Streams は、Prometheus JMX Exporter を使用して、Prometheus サーバーによってスクレープできる HTTP エンドポイント経由でメトリクスを公開します。
Grafana ダッシュボードが依存する Prometheus JMX Exporter の再ラベル付けルールは、カスタムリソース設定として AMQ Streams コンポーネントに対して定義されます。
ラベルは名前と値のペアです。再ラベル付けは、ラベルを動的に書き込むプロセスです。たとえば、ラベルの値は Kafka サーバーおよびクライアント ID の名前から派生されます。
AMQ Streams では、再ラベル付けルールが含まれるカスタムリソース設定用の YAML ファイルのサンプルが提供されます。Prometheus メトリクス設定をデプロイする場合、カスタムリソースのサンプルをデプロイすることや、メトリクス設定を独自のカスタムリソース定義にコピーすることができます。
コンポーネント | カスタムリソース | サンプル YAML ファイル |
---|---|---|
Kafka および ZooKeeper |
|
|
Kafka Connect |
|
|
Kafka MirrorMaker 2.0 |
|
|
Kafka Bridge |
|
|
Cruise Control |
|
|
8.3.2. アラート通知の Prometheus ルールの例
アラート通知の Prometheus ルールの例は、AMQ Streams によって提供される メトリクス設定ファイルの例 と共に提供されます。ルールは、Prometheus デプロイメント で使用するための prometheus-rules.yaml
ファイルのサンプルに指定されています。
アラートルールによって、メトリクスで監視される特定条件についての通知が提供されます。ルールは Prometheus サーバーで宣言されますが、アラート通知は Prometheus Alertmanager で対応します。
Prometheus アラートルールでは、継続的に評価される PromQL 表現を使用して条件が記述されます。
アラート表現が true になると、条件が満たされ、Prometheus サーバーからアラートデータが Alertmanager に送信されます。次に Alertmanager は、そのデプロイメントに設定された通信方法を使用して通知を送信します。
アラートルールの定義に関する一般的な留意点:
-
for
プロパティーはルールと併用され、アラートがトリガーされる前に条件が維持されなければならない期間を決定します。 -
ティック (tick) は ZooKeeper の基本的な時間単位です。ミリ秒単位で測定され、
Kafka.spec.zookeeper.config
のtickTime
パラメーターを使用して設定されます。たとえば、ZooKeeper でtickTime=3000
の場合、3 ティック (3 x 3000) は 9000 ミリ秒と等しくなります。 -
ZookeeperRunningOutOfSpace
メトリクスおよびアラートを利用できるかどうかは、使用される OpenShift 設定およびストレージ実装によります。特定のプラットフォームのストレージ実装では、メトリクスによるアラートの提供に必要な利用可能な領域について情報が提供されない場合があります。
Alertmanager は、電子メール、チャットメッセージなどの通知方法を使用するように設定できます。ルールの例のデフォルト設定は、特定のニーズに合わせて調整してください。
8.3.2.1. ルールの変更例
prometheus-rules.yaml
ファイルには、以下のコンポーネントのルールの例が含まれます。
- Kafka
- ZooKeeper
- Entitiy Operator
- Kafka Connect
- Kafka Bridge
- MirrorMaker
- Kafka Exporter
各ルールの例の説明は、ファイルに記載されています。
8.3.3. Grafana ダッシュボードのサンプル
Prometheus をデプロイしてメトリクスを提供する場合は、AMQ Streams で提供される Grafana ダッシュボードのサンプルを使用して、AMQ Streams コンポーネントを監視できます。
ダッシュボードのサンプルは、examples/metrics/grafana-dashboards
ディレクトリーに JSON ファイルで提供されます。
すべてのダッシュボードは、JVM メトリクスに加えてコンポーネントに固有のメトリクスを提供します。たとえば、AMQ Streams Operator の Grafana ダッシュボードは、調整の数または処理中のカスタムリソースに関する情報を提供します。
ダッシュボードのサンプルには、Kafka でサポートされるすべてのメトリクスは表示されません。ダッシュボードには、監視用の代表的なメトリクスのセットが表示されます。
Component | JSON ファイルの例: |
---|---|
AMQ Streams の Operator |
|
Kafka |
|
ZooKeeper |
|
Kafka Connect |
|
Kafka MirrorMaker 2.0 |
|
Kafka Bridge |
|
Cruise Control |
|
Kafka Exporter |
|
8.4. Prometheus メトリクス設定のデプロイ
Prometheus メトリクス設定をデプロイし、AMQ Streams で Prometheus を使用します。metricsConfig
プロパティーを使用して、Prometheus メトリクスを有効化および設定します。
独自の設定、または AMQ Streams で提供されるサンプルのカスタムリソース設定ファイル を使用できます。
-
kafka-metrics.yaml
-
kafka-connect-metrics.yaml
-
kafka-mirror-maker-2-metrics.yaml
-
kafka-bridge-metrics.yaml
-
kafka-cruise-control-metrics.yaml
設定ファイルのサンプルには、再ラベル付けルールと Prometheus メトリクスの有効化に必要な設定があります。Prometheus は、ターゲット HTTP エンドポイントからメトリクスを収集します。サンプルファイルは、AMQ Streams で Prometheus を試すのに適した方法です。
再ラベル付けルールおよびメトリクス設定を適用するには、以下のいずれかを行います。
- 独自のカスタムリソースに設定例をコピーする。
- メトリクス設定でカスタムリソースをデプロイする。
Kafka Exporter メトリクスを含める場合は、kafkaExporter
設定を Kafka
リソースに追加します。
Kafka Exporter は、コンシューマーラグおよびコンシューマーオフセットに関連する追加のメトリクスのみを提供します。通常の Kafka メトリクスでは、Kafka ブローカー で、Prometheus メトリクスを設定する必要があります。
この手順では、Kafka
リソースに Prometheus メトリクス設定をデプロイする方法を説明します。このプロセスは、他のリソースのサンプルファイルを使用する場合と同じです。
手順
Prometheus 設定でカスタムリソースのサンプルをデプロイします。
たとえば、
Kafka
リソースごとにkafka-metrics.yaml
ファイルを適用します。サンプル設定のデプロイ
oc apply -f kafka-metrics.yaml
または、
kafka-metrics.yaml
の設定例を独自のKafka
リソースにコピーすることもできます。サンプル設定のコピー
oc edit kafka <kafka-configuration-file>
metricsConfig
プロパティーと、Kafka
リソースを参照するConfigMap
をコピーします。Kafka のメトリクス設定例
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: kafka: # ... metricsConfig: 1 type: jmxPrometheusExporter valueFrom: configMapKeyRef: name: my-config-map key: my-key --- kind: ConfigMap 2 apiVersion: v1 metadata: name: kafka-metrics labels: app: strimzi data: kafka-metrics-config.yml: | # metrics configuration...
注記Kafka Bridge の場合、
enableMetrics
プロパティーを指定し、これをtrue
に設定します。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaBridge metadata: name: my-bridge spec: # ... bootstrapServers: my-cluster-kafka:9092 http: # ... enableMetrics: true # ...
Kafka Exporter をデプロイするには、
kafkaExporter
設定を追加します。KafkaExporter
設定は、Kafka
リソースでのみ指定されます。Kafka Exporter のデプロイの設定例
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: # ... kafkaExporter: image: my-registry.io/my-org/my-exporter-cluster:latest 1 groupRegex: ".*" 2 topicRegex: ".*" 3 resources: 4 requests: cpu: 200m memory: 64Mi limits: cpu: 500m memory: 128Mi logging: debug 5 enableSaramaLogging: true 6 template: 7 pod: metadata: labels: label1: value1 imagePullSecrets: - name: my-docker-credentials securityContext: runAsUser: 1000001 fsGroup: 0 terminationGracePeriodSeconds: 120 readinessProbe: 8 initialDelaySeconds: 15 timeoutSeconds: 5 livenessProbe: 9 initialDelaySeconds: 15 timeoutSeconds: 5 # ...
8.5. OpenShift での Kafka メトリクスおよびダッシュボードの表示
AMQ Streams が OpenShift Container Platform にデプロイされると、ユーザー定義プロジェクトのモニタリング によりメトリクスが提供されます。この OpenShift 機能により、開発者は独自のプロジェクト (例: Kafka
プロジェクト) を監視するために別の Prometheus インスタンスにアクセスできます。
ユーザー定義プロジェクトのモニタリングが有効である場合、openshift-user-workload-monitoring
プロジェクトには以下のコンポーネントが含まれます。
- Prometheus Operator
- Prometheus インスタンス (Prometheus Operator によって自動的にデプロイされます)
- Thanos Ruler インスタンス
AMQ Streams は、これらのコンポーネントを使用してメトリクスを消費します。
クラスター管理者は、ユーザー定義プロジェクトのモニタリングを有効にし、開発者およびその他のユーザーに独自のプロジェクト内のアプリケーションを監視するパーミッションを付与する必要があります。
Grafana のデプロイメント
Grafana インスタンスを、Kafka クラスターが含まれるプロジェクトにデプロイできます。その後、Grafana ダッシュボードのサンプルを使用して、AMQ Streams の Prometheus メトリクスを Grafana ユーザーインターフェースで可視化できます。
openshift-monitoring
プロジェクトはコアプラットフォームコンポーネントのモニタリングを提供します。このプロジェクトの Prometheus および Grafana コンポーネントを使用して、OpenShift Container Platform 4.x 上の AMQ Streams の監視を設定しないでください。
手順の概要
OpenShift Container Platform で AMQ Streams のモニタリングを設定するには、以下の手順を順番に行います。
8.5.1. 前提条件
- YAML ファイルのサンプルを使用して、Prometheus メトリクス設定がデプロイされている 必要があります。
-
ユーザー定義プロジェクトの監視が有効になっている必要があります。クラスター管理者は、OpenShift クラスターに
cluster-monitoring-config
設定マップを作成しています。 -
クラスター管理者は、
monitoring-rules-edit
またはmonitoring-edit
ロールを割り当てている。
cluster-monitoring-config
設定マップの作成およびユーザー定義プロジェクトをモニターするためのパーミッションをユーザーに付与する方法の詳細については、OpenShift Container Platform の Monitoring を参照してください。
8.5.2. 関連情報
- OpenShift Container Platform Monitoring
8.5.3. Prometheus リソースのデプロイ
Prometheus を使用して、Kafka クラスターのモニタリングデータを取得します。
独自の Prometheus デプロイメントを使用するか、AMQ Streams によって提供される メトリクス設定ファイルのサンプル を使用して Prometheus をデプロイできます。サンプルファイルを使用するには、PodMonitor
リソースを設定し、デプロイします。PodMonitors
は、Apache Kafka、ZooKeeper、Operator、Kafka Bridge、および Cruise Control から直接データをスクレープします。
次に、Alertmanager のアラートルールのサンプルをデプロイします。
前提条件
- 稼働中の Kafka クラスターが必要です。
- AMQ Streams で 提供されるアラートルールのサンプル を確認します。
手順
ユーザー定義プロジェクトのモニタリングが有効であることを確認します。
oc get pods -n openshift-user-workload-monitoring
有効であると、モニタリングコンポーネントの Pod が返されます。以下に例を示します。
NAME READY STATUS RESTARTS AGE prometheus-operator-5cc59f9bc6-kgcq8 1/1 Running 0 25s prometheus-user-workload-0 5/5 Running 1 14s prometheus-user-workload-1 5/5 Running 1 14s thanos-ruler-user-workload-0 3/3 Running 0 14s thanos-ruler-user-workload-1 3/3 Running 0 14s
Pod が返されなければ、ユーザー定義プロジェクトのモニタリングは無効になっています。「OpenShift での Kafka メトリクスおよびダッシュボードの表示」 の前提条件を参照してください。
複数の
PodMonitor
リソースは、examples/metrics/prometheus-install/strimzi-pod-monitor.yaml
で定義されます。PodMonitor
リソースごとにspec.namespaceSelector.matchNames
プロパティーを編集します。apiVersion: monitoring.coreos.com/v1 kind: PodMonitor metadata: name: cluster-operator-metrics labels: app: strimzi spec: selector: matchLabels: strimzi.io/kind: cluster-operator namespaceSelector: matchNames: - <project-name> 1 podMetricsEndpoints: - path: /metrics port: http # ...
- 1
- メトリクスをスクレープする Pod が実行されているプロジェクト (例:
Kafka
)。
strimzi-pod-monitor.yaml
ファイルを、Kafka クラスターが稼働しているプロジェクトにデプロイします。oc apply -f strimzi-pod-monitor.yaml -n MY-PROJECT
Prometheus ルールのサンプルを同じプロジェクトにデプロイします。
oc apply -f prometheus-rules.yaml -n MY-PROJECT
8.5.4. Grafana のサービスアカウントの作成
AMQ Streams の Grafana インスタンスは、cluster-monitoring-view
ロールが割り当てられたサービスアカウントで実行する必要があります。
Grafana を使用してモニタリングのメトリクスを表示する場合は、サービスアカウントを作成します。
前提条件
手順
Grafana の
ServiceAccount
を作成します。ここでは、リソースの名前はgrafana-serviceaccount
です。apiVersion: v1 kind: ServiceAccount metadata: name: grafana-serviceaccount labels: app: strimzi
ServiceAccount
を、Kafka クラスターが含まれるプロジェクトにデプロイします。oc apply -f GRAFANA-SERVICEACCOUNT -n MY-PROJECT
cluster-monitoring-view
ロールを GrafanaServiceAccount
に割り当てるClusterRoleBinding
リソースを作成します。ここでは、リソースの名前はgrafana-cluster-monitoring-binding
です。apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: grafana-cluster-monitoring-binding labels: app: strimzi subjects: - kind: ServiceAccount name: grafana-serviceaccount namespace: <my-project> 1 roleRef: kind: ClusterRole name: cluster-monitoring-view apiGroup: rbac.authorization.k8s.io
- 1
- プロジェクトの名前。
ClusterRoleBinding
を、Kafka クラスターが含まれるプロジェクトにデプロイします。oc apply -f <grafana-cluster-monitoring-binding> -n <my-project>
8.5.5. Prometheus データソースを使用した Grafana のデプロイ
Grafana をデプロイし、Prometheus メトリクスを表示します。Grafana アプリケーションには、OpenShift Container Platform モニタリングスタックの設定が必要です。
OpenShift Container Platform では、openshift-monitoring
プロジェクトに Thanos Querier インスタンスが含まれています。Thanos Querier は、プラットフォームメトリクスを集約するために使用されます。
必要なプラットフォームメトリクスを使用するには、Grafana インスタンスには Thanos Querier に接続できる Prometheus データソースが必要です。この接続を設定するには、トークンを使用し、Thanos Querier と並行して実行される oauth-proxy
サイドカーに対して認証を行うconfig map を作成します。datasource.yaml
ファイルは config map のソースとして使用されます。
最後に、Kafka クラスターが含まれるプロジェクトにボリュームとしてマウントされた config map で Grafana アプリケーションをデプロイします。
手順
Grafana
ServiceAccount
のアクセストークンを取得します。oc serviceaccounts get-token grafana-serviceaccount -n MY-PROJECT
次のステップで使用するアクセストークンをコピーします。
Grafana の Thanos Querier 設定が含まれる
datasource.yaml
ファイルを作成します。以下に示すように、アクセストークンを
httpHeaderValue1
プロパティーに貼り付けます。apiVersion: 1 datasources: - name: Prometheus type: prometheus url: https://thanos-querier.openshift-monitoring.svc.cluster.local:9091 access: proxy basicAuth: false withCredentials: false isDefault: true jsonData: timeInterval: 5s tlsSkipVerify: true httpHeaderName1: "Authorization" secureJsonData: httpHeaderValue1: "Bearer ${GRAFANA-ACCESS-TOKEN}" 1 editable: true
- 1
GRAFANA-ACCESS-TOKEN
:GrafanaServiceAccount
のアクセストークンの値。
datasource.yaml
ファイルからgrafana-config
という名前の config map を作成します。oc create configmap grafana-config --from-file=datasource.yaml -n MY-PROJECT
Deployment
およびService
で構成される Grafana アプリケーションを作成します。grafana-config
config map はデータソース設定のボリュームとしてマウントされます。apiVersion: apps/v1 kind: Deployment metadata: name: grafana labels: app: strimzi spec: replicas: 1 selector: matchLabels: name: grafana template: metadata: labels: name: grafana spec: serviceAccountName: grafana-serviceaccount containers: - name: grafana image: grafana/grafana:7.5.15 ports: - name: grafana containerPort: 3000 protocol: TCP volumeMounts: - name: grafana-data mountPath: /var/lib/grafana - name: grafana-logs mountPath: /var/log/grafana - name: grafana-config mountPath: /etc/grafana/provisioning/datasources/datasource.yaml readOnly: true subPath: datasource.yaml readinessProbe: httpGet: path: /api/health port: 3000 initialDelaySeconds: 5 periodSeconds: 10 livenessProbe: httpGet: path: /api/health port: 3000 initialDelaySeconds: 15 periodSeconds: 20 volumes: - name: grafana-data emptyDir: {} - name: grafana-logs emptyDir: {} - name: grafana-config configMap: name: grafana-config --- apiVersion: v1 kind: Service metadata: name: grafana labels: app: strimzi spec: ports: - name: grafana port: 3000 targetPort: 3000 protocol: TCP selector: name: grafana type: ClusterIP
Grafana アプリケーションを、Kafka クラスターが含まれるプロジェクトにデプロイします。
oc apply -f <grafana-application> -n <my-project>
8.5.6. Grafana サービスへのルートの作成
Grafana サービスを公開するルートを介して、Grafana ユーザーインターフェースにアクセスできます。
手順
grafana
サービスへのルートの作成:oc create route edge <my-grafana-route> --service=grafana --namespace=KAFKA-NAMESPACE
8.5.7. Grafana ダッシュボードサンプルのインポート
Grafana を使用して、カスタマイズ可能なダッシュボードで Prometheus メトリクスを視覚化します。
AMQ Streams は、Grafana のダッシュボード設定ファイルのサンプル を JSON 形式で提供します。
-
examples/metrics/grafana-dashboards
この手順では、Grafana ダッシュボードのサンプルを使用します。
ダッシュボードのサンプルは、キーメトリクスを監視するを開始点として適していますが、Kafka でサポートされるすべてのメトリクスは表示されません。使用するインフラストラクチャーに応じて、ダッシュボードのサンプルの編集や、他のメトリクスの追加を行うことができます。
前提条件
手順
Grafana サービスへのルートの詳細を取得します。以下に例を示します。
oc get routes NAME HOST/PORT PATH SERVICES MY-GRAFANA-ROUTE MY-GRAFANA-ROUTE-amq-streams.net grafana
- Web ブラウザーで、Route ホストおよびポートの URL を使用して Grafana ログイン画面にアクセスします。
ユーザー名とパスワードを入力し、続いて Log In をクリックします。
デフォルトの Grafana ユーザー名およびパスワードは、どちらも
admin
です。初回ログイン後に、パスワードを変更できます。- Configuration > Data Sources で、Prometheus データソースが作成済みであることを確認します。データソースは 「Prometheus データソースを使用した Grafana のデプロイ」 に作成されています。
- + アイコンをクリックしてから、Import をクリックします。
-
examples/metrics/grafana-dashboards
で、インポートするダッシュボードの JSON をコピーします。 - JSON をテキストボックスに貼り付け、Load をクリックします。
- 他の Grafana ダッシュボードのサンプル用に、ステップ 5-7 を繰り返します。
インポートされた Grafana ダッシュボードは、Dashboards ホームページから表示できます。
第9章 AMQ Streams のアップグレード
AMQ Streams をバージョン 2.1 にアップグレードすると、新機能および改良された機能、パフォーマンスの向上、およびセキュリティーオプションを利用できます。
このアップグレード中に、Kafka をサポートされる最新バージョンにアップグレードします。各 Kafka リリースによって、AMQ Streams デプロイメントに新機能、改善点、およびバグ修正が導入されます。
新しいバージョンで問題が発生した場合は、AMQ Streams を以前のバージョンに ダウングレード できます。
リリースされた AMQ Streams バージョンは、AMQ Streams ソフトウェアダウンロードページ から入手できます。
アップグレードパス
2 つのアップグレードパスが可能です。
- インクリメント
- AMQ Streams を以前のマイナーバージョンからバージョン 2.1 にアップグレードします。
- マルチバージョン
AMQ Streams を 1 回で古いバージョンからバージョン 2.1 にアップグレードします (1 つ以上の中間バージョンを飛ばします)。
たとえば、AMQ Streams 1.8 から直接 AMQ Streams 2.1 にアップグレードします。
Kafka バージョンのサポート
- Kafka 3.1.0 は実稼働環境での使用でサポートされます。
- Kafka 3.0.0 は、AMQ Streams 2.1 にアップグレードする目的でのみサポートされます。
ご使用のバージョンの AMQ Streams によってサポートされれば、上位バージョンの Kafka にアップグレードできます。サポートされる下位バージョンの Kafka にダウングレードできる場合もあります。
ダウンタイムと可用性
高可用性に対してトピックが設定されている場合、AMQ Streams をアップグレードしても、これらのトピックからデータをパブリッシュおよび読み取るコンシューマーとプロデューサーのダウンタイムは発生しません。高可用性トピックのレプリケーション係数は 3 以上であり、パーティションはブローカー間で均等に分散されます。
AMQ Streams をアップグレードするとローリングアップデートがトリガーされ、プロセスのさまざまな段階ですべてのブローカーが順に再起動されます。ローリングアップデート中に、すべてのブローカーがオンライン状態ではないため、クラスター全体の可用性 は一時的に低下します。クラスターの可用性が低下すると、ブローカーの障害によってメッセージが失われる可能性が高くなります。
9.1. 必要なアップグレードシーケンス
ダウンタイムなしでブローカーとクライアントをアップグレードするには、以下の順序でアップグレード手順を 必ず 完了してください。
Kubernetes クラスターのバージョンがサポートされていることを確認してください。
AMQ Streams 2.1 は、OpenShift 4.6 から 4.10 でサポートされます。
-
AMQ Streams を 1.7 以前からアップグレードする場合は、既存のカスタムリソースを更新して、
v1beta2
API バージョンをサポートします。 - 新しい AMQ Streams バージョンに Cluster Operator を更新 します。
- サポートされる最新の Kafka バージョンに、すべての Kafka ブローカーとクライアントアプリケーションをアップグレードします。
- オプション:パーティションの再分散に Incremental Cooperative Rebalance プロトコルを使用するために、コンシューマーと Kafka Streams アプリケーションをアップグレードします。
9.1.1. Cluster Operator のアップグレードオプション
Cluster Operator のアップグレード方法は、デプロイ方法によって異なります。
- インストールファイルの使用
- インストール用の YAML ファイルを使用して Cluster Operator をデプロイした場合は、「Cluster Operator のアップグレード」の説明に従って、Operator のインストールファイルを変更してアップグレードを実行します。
- OperatorHub の使用
OperatorHub から AMQ Streams をデプロイした場合は、Operator Lifecycle Manager (OLM) を使用して AMQ Streams Operator の更新チャネルを新しい AMQ Streams バージョンに変更します。
選択したアップグレードストラテジーに応じて、チャネルの更新後に以下のいずれかを実行します。
- 自動アップグレードが開始されます。
手動アップグレードでは、インストールを開始する前に承認が必要です。
OperatorHub を使用した Operator のアップグレードについての詳細は、OpenShift ドキュメントの「Upgrading installed Operators」を参照してください。
9.1.2. OperatorHub を使用した AMQ Streams 1.7 以前からのアップグレード
OperatorHub を使用して AMQ Streams 1.7 以前からアップグレードする場合に必要となるアクション
Red Hat Integration - AMQ Streams Operator は、v1beta2
カスタムリソースのみをサポートします。OperatorHub で AMQ Streams Operator をバージョン 2.1 に アップグレードする前 に、カスタムリソースを v1beta2
にアップグレードする必要があります。
すべてのカスタムリソースの v1beta2
API バージョンが AMQ Streams 1.7 で導入されました。AMQ Streams 1.8 では、KafkaTopic
および KafkaUser
を除くすべての AMQ Streams カスタムリソースから v1alpha1
および v1beta1
API バージョンが削除されました。
バージョン 1.7 より前の AMQ Streams バージョンからアップグレードする場合は、以下を行います。
- AMQ Streams 1.7 へのアップグレード
- AMQ Streams ソフトウェアのダウンロードページ から、AMQ Streams 1.8 で提供される Red Hat AMQ Streams API Conversion Tool をダウンロードします。
カスタムリソースおよび CRD を
v1beta2
に変換します。詳細は、AMQ Streams 1.7 upgrade documentation を参照してください。
- OperatorHub で、Red Hat Integration - AMQ Streams Operator のバージョン 1.7.0 を削除します。
存在する場合は、Red Hat Integration - AMQ Streams Operator のバージョン 2.1.0 を削除します。
存在しない場合は、次のステップに進みます。
AMQ Streams Operator の Approval Strategy が Automatic に設定されている場合、Operator のバージョン 2.1.0 がすでにクラスターに存在する可能性があります。リリース前にカスタムリソースおよび CRD を
v1beta2
API バージョンに 変換しなかった 場合、Operator が管理するカスタムリソースおよび CRD は古い API バージョンを使用します。その結果、2.1.0 Operator は Pending ステータスで停止します。このような場合、Red Hat Integration - AMQ Streams Operator のバージョン 2.1.0 およびバージョン 1.7.0 を削除する必要があります。両方の Operator を削除すると、新しい Operator バージョンがインストールされるまで、調整は一時停止されます。カスタムリソースへの変更が遅延しないように、次の手順を直ちに実行します。
OperatorHub で、Red Hat Integration - AMQ Streams Operator のバージョン 2.1.0 を即時にインストールします。
インストールされた 2.1.0 Operator はクラスターの監視を開始し、ローリングアップデートを実行します。このプロセス中に、クラスターのパフォーマンスが一時的に低下する場合があります。
別の方法として、バージョン 1.7 からカスタムリソースをインストールし、リソースを変換してから 1.8 以降にアップグレードすることもできます。
9.2. 最小限のダウンタイムでの OpenShift のアップグレード
OpenShift をアップグレードする場合は、OpenShift アップグレードのドキュメントを参照して、アップグレードパスとノードを正しくアップグレードする手順を確認してください。OpenShift をアップグレードする前に、お使いの AMQ Streams バージョンでサポートされるバージョン を確認してください。
アップグレードを実行する際に、Kafka クラスターを利用できるようにしておくことを推奨します。
以下のストラテジーのいずれかを使用できます。
- Pod の Disruption Budget (停止状態の予算) を設定します。
以下の方法の 1 つで Pod をローリングします。
- AMQ Streams Drain Cleaner の使用
- アノテーションを Pod に適用して手動で
Pod の Disruption Budget(停止状態の予算)を設定して Pod をロールオーバーする方法のいずれかを使用する必要があります。
Kafka を稼働し続けるには、高可用性のためにトピックも複製する必要があります。これには、少なくとも 3 つのレプリケーション係数と、レプリケーション係数よりも 1 つ少ない In-Sync レプリカの最小数を指定するトピック設定が必要です。
高可用性のためにレプリケートされた Kafka トピック
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic labels: strimzi.io/cluster: my-cluster spec: partitions: 1 replicas: 3 config: # ... min.insync.replicas: 2 # ...
高可用性環境では、Cluster Operator はアップグレードプロセス時にトピックの In-Sync レプリカの最小数を維持し、ダウンタイムが発生しないようにします。
9.2.1. AMQ Streams Drain Cleaner を使用した Pod のローリング
AMQ Streams Drain Cleaner ツールを使用して、アップグレード時にノードをエビクトできます。AMQ Streams Drain Cleaner は、ローリングアップデート Pod アノテーションを Pod に付けます。これにより、Cluster Operator に、エビクトされた Pod のローリングアップデートを実行するように指示します。
Pod の Disruption Budget(停止状態の予算)により、指定された数の Pod のみが、特定の時点で利用できなくなります。Kafka ブローカー Pod の計画的なメンテナンス中に、Pod の Disruption Budget (停止状態の予算) により、Kafka が高可用性環境で引き続き実行されます。
Kafka コンポーネントの template
のカスタマイズを使用して、Pod の Disruption Budget (停止状態の予算) を指定します。デフォルトでは、Pod の Disruption Budget (停止状態の予算) は、単一の Pod のみを指定時に利用できないようにします。
これを実行するには、maxUnavailable
を 0
(ゼロ) に設定します。Pod の Disruption Budget (停止状態の予算) の最大値をゼロに減らすと、自発的な中断が回避されるため、Pod を手動でエビクトする必要があります。
Pod の Disruption Budget (停止状態の予算) の指定
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster namespace: myproject spec: kafka: # ... template: podDisruptionBudget: maxUnavailable: 0 # ...
9.2.2. トピックを利用可能な状態に維持しながらの手動での Pod のローリング
アップグレード時に、Cluster Operator 経由で Pod の手動ローリングアップデートをトリガーできます。Pod
リソースを使用して、ローリングアップデートは新規 Pod でリソースの Pod を再起動します。AMQ Streams Drain Cleaner を使用する場合と同様に、Pod の Disruption Budget の maxUnavailable
の値をゼロに設定する必要があります。
ドレイン(解放)する必要のある Pod を監視する必要があります。次に Pod アノテーションを追加して更新を行います。
ここで、アノテーションは Kafka ブローカーを更新します。
Kafka ブローカー Pod での手動ローリングアップデートの実行
oc annotate pod <cluster_name>-kafka-<index> strimzi.io/manual-rolling-update=true
<cluster_name> は、クラスターの名前に置き換えます。Kafka ブローカー Pod の名前は <cluster-name>-kafka-<index> です。ここで、<index> はゼロで始まり、レプリカの合計数から 1 を引いた数で終了します。例: my-cluster-kafka-0
9.3. AMQ Streams カスタムリソースのアップグレード
AMQ Streams を 1.7 以前から 2.1 にアップグレードする場合、カスタムリソースが API バージョン v1beta2
を使用していることを確認する必要があります。AMQ Streams 1.8 以降にアップグレードする前に、カスタムリソース定義およびカスタムリソースをアップグレードする必要があります。アップグレードを実行するには、AMQ Streams 1.7 で提供される API 変換ツール を使用できます。詳細は、AMQ Streams 1.7 upgrade documentation を参照してください。
9.4. Cluster Operator のアップグレード
この手順では、AMQ Streams 2.1 を使用するように Cluster Operator デプロイメントをアップグレードする方法を説明します。
インストール YAML ファイルを使用して Cluster Operator をデプロイした場合は、以下の手順に従います。
Cluster Operator によって管理される Kafka クラスターの可用性は、アップグレード操作による影響を受けません。
特定バージョンの AMQ Streams へのアップグレード方法については、そのバージョンをサポートするドキュメントを参照してください。
前提条件
- 既存の Cluster Operator デプロイメントを利用できる必要があります。
- AMQ Streams 2.1 のリリースアーティファクトがダウンロード済みである必要があります。
手順
-
既存の Cluster Operator リソース (
/install/cluster-operator
ディレクトリー内) に追加した設定変更を覚えておきます。すべての変更は、新しいバージョンの Cluster Operator によって上書きされます。 - カスタムリソースを更新して、AMQ Streams バージョン 2.1 で使用できるサポート対象の設定オプションを反映します。
Cluster Operator を更新します。
Cluster Operator を実行している namespace に従い、新しい Cluster Operator バージョンのインストールファイルを編集します。
Linux の場合は、以下を使用します。
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
MacOS の場合は、以下を使用します。
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
-
既存の Cluster Operator
Deployment
で 1 つ以上の環境変数を編集した場合、install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml
ファイルを編集し、これらの環境変数を使用します。
設定を更新したら、残りのインストールリソースとともにデプロイします。
oc replace -f install/cluster-operator
ローリングアップデートが完了するのを待ちます。
新しい Operator バージョンがアップグレード元の Kafka バージョンをサポートしなくなった場合、Cluster Operator はバージョンがサポートされていないことを示すエラーメッセージを返します。そうでない場合は、エラーメッセージは返されません。
エラーメッセージが返される場合は、新しい Cluster Operator バージョンでサポートされる Kafka バージョンにアップグレードします。
-
Kafka
カスタムリソースを編集します。 -
spec kafka.version
プロパティーをサポートされる Kafka バージョンに変更します。
-
- エラーメッセージが返されない場合は、次のステップに進みます。Kafka のバージョンを後でアップグレードします。
Kafka Pod のイメージを取得して、アップグレードが正常に完了したことを確認します。
oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'
イメージタグには、新しい Operator のバージョンが表示されます。以下に例を示します。
registry.redhat.io/amq7/amq-streams-kafka-30-rhel8:2.1.0
Cluster Operator はバージョン 2.1 にアップグレードされましたが、管理するクラスターで稼働している Kafka のバージョンは変更されていません。
Cluster Operator のアップグレードの次に、Kafka のアップグレードを実行する必要があります。
9.5. Kafka のアップグレード。
Cluster Operator を 2.1 にアップグレードした後、次にすべての Kafka ブローカーをサポートされる最新バージョンの Kafka にアップグレードします。
Kafka のアップグレードは、Kafka ブローカーのローリングアップデートによって Cluster Operator によって実行されます。
Cluster Operator は、Kafka クラスターの設定に基づいてローリングアップデートを開始します。
Kafka.spec.kafka.config に以下が含まれている場合 | Cluster Operator によって開始されるもの |
---|---|
|
単一のローリングアップデート更新後、 |
| 2 つのローリングアップデート |
| 2 つのローリングアップデート |
Kafka 3.0.0 以降、inter.broker.protocol.version
が 3.0
以上に設定されていると、log.message.format.version
オプションは無視されるため、設定する必要はありません。ブローカーの log.message.format.version
プロパティーおよびトピックの message.format.version
プロパティーは、非推奨となり、Kafka の今後のリリースで削除されます。
Cluster Operator は、Kafka のアップグレードの一環として、ZooKeeper のローリングアップデートを開始します。
- ZooKeeper バージョンが変更されなくても、単一のローリングアップデートが発生します。
- 新しいバージョンの Kafka に新しいバージョンの ZooKeeper が必要な場合、追加のローリングアップデートが発生します。
9.5.1. Kafka バージョン
Kafka のログメッセージ形式バージョンとブローカー間のプロトコルバージョンは、それぞれメッセージに追加されるログ形式バージョンとクラスターで使用される Kafka プロトコルのバージョンを指定します。正しいバージョンが使用されるようにするため、アップグレードプロセスでは、既存の Kafka ブローカーの設定変更と、クライアントアプリケーション (コンシューマーおよびプロデューサー) のコード変更が行われます。
以下の表は、Kafka バージョンの違いを示しています。
Kafka のバージョン | Interbroker プロトコルのバージョン | ログメッセージ形式のバージョン | ZooKeeper のバージョン |
---|---|---|---|
3.1.0 | 3.1 | 3.1 | 3.6.3 |
3.0.0 | 3.0 | 3.0 | 3.6.3 |
ブローカー間のプロトコルバージョン
Kafka では、ブローカー間の通信に使用されるネットワークプロトコルはブローカー間プロトコル (Inter-broker protocol) と呼ばれます。Kafka の各バージョンには、互換性のあるバージョンのブローカー間プロトコルがあります。上記の表が示すように、プロトコルのマイナーバージョンは、通常 Kafka のマイナーバージョンと一致するように番号が増加されます。
ブローカー間プロトコルのバージョンは、Kafka
リソースでクラスター全体に設定されます。これを変更するには、Kafka.spec.kafka.config
の inter.broker.protocol.version
プロパティーを編集します。
ログメッセージ形式のバージョン
プロデューサーが Kafka ブローカーにメッセージを送信すると、特定の形式を使用してメッセージがエンコードされます。この形式は Kafka のリリース間で変更される可能性があるため、メッセージにはエンコードに使用されたメッセージ形式のバージョンが指定されます。
特定のメッセージ形式のバージョンを設定するために使用されるプロパティーは以下のとおりです。
-
トピック用の
message.format.version
プロパティー -
Kafka ブローカーの
log.message.format.version
プロパティー
Kafka 3.0.0 以降、メッセージ形式のバージョンの値は inter.broker.protocol.version
と一致すると見なされ、設定する必要はありません。値は、使用される Kafka バージョンを反映します。
Kafka 3.0.0 以降にアップグレードする場合、inter.broker.protocol.version
を更新する際にこれらの設定を削除できます。それ以外の場合は、アップグレード先の Kafka バージョンに基づいてメッセージ形式のバージョンを設定します。
トピックの message.format.version
のデフォルト値は、Kafka ブローカーに設定される log.message.format.version
によって定義されます。トピックの message.format.version
は、トピック設定を編集すると手動で設定できます。
9.5.2. クライアントをアップグレードするストラテジー
クライアントアプリケーション (Kafka Connect コネクターを含む) のアップグレードに適切な方法は、特定の状況によって異なります。
消費するアプリケーションは、そのアプリケーションが理解するメッセージ形式のメッセージを受信する必要があります。その状態であることを、以下のいずれかの方法で確認できます。
- プロデューサーをアップグレードする 前に、トピックのすべてのコンシューマーをアップグレードする。
- ブローカーでメッセージをダウンコンバートする。
ブローカーのダウンコンバートを使用すると、ブローカーに余分な負荷が加わるので、すべてのトピックで長期にわたりダウンコンバートに頼るのは最適な方法ではありません。ブローカーの実行を最適化するには、ブローカーがメッセージを一切ダウンコンバートしないようにしてください。
ブローカーのダウンコンバートは 2 通りの方法で設定できます。
-
トピックレベルの
message.format.version
では単一のとピックが設定されます。 -
ブローカーレベルの
log.message.format.version
は、トピックレベルのmessage.format.version
が設定されてないトピックのデフォルトです。
新バージョンの形式でトピックにパブリッシュされるメッセージは、コンシューマーによって認識されます。これは、メッセージがコンシューマーに送信されるときでなく、ブローカーがプロデューサーからメッセージを受信するときに、ブローカーがダウンコンバートを実行するからです。
クライアントのアップグレードに使用できる一般的なストラテジーを以下に示します。クライアントアプリケーションをアップグレードするストラテジーは他にもあります。
Kafka 3.0.0 以降にアップグレードすると、各ストラテジーで概説されている手順がわずかに変わります。Kafka 3.0.0 以降、メッセージ形式のバージョンの値は inter.broker.protocol.version
と一致すると見なされ、設定する必要はありません。
ブローカーレベルのコンシューマーの最初のストラテジー
- コンシューマーとして機能するアプリケーションをすべてアップグレードします。
-
ブローカーレベル
log.message.format.version
を新バージョンに変更します。 - プロデューサーとして機能するアプリケーションをアップグレードします。
このストラテジーは分かりやすく、ブローカーのダウンコンバートの発生をすべて防ぎます。ただし、所属組織内のすべてのコンシューマーを整然とアップグレードできることが前提になります。また、コンシューマーとプロデューサーの両方に該当するアプリケーションには通用しません。さらにリスクとして、アップグレード済みのクライアントに問題がある場合は、新しい形式のメッセージがメッセージログに追加され、以前のコンシューマーバージョンに戻せなくなる場合があります。
トピックレベルのコンシューマーの最初のストラテジー
トピックごとに以下を実行します。
- コンシューマーとして機能するアプリケーションをすべてアップグレードします。
-
トピックレベルの
message.format.version
を新バージョンに変更します。 - プロデューサーとして機能するアプリケーションをアップグレードします。
このストラテジーではブローカーのダウンコンバートがすべて回避され、トピックごとにアップグレードできます。この方法は、同じトピックのコンシューマーとプロデューサーの両方に該当するアプリケーションには通用しません。ここでもリスクとして、アップグレード済みのクライアントに問題がある場合は、新しい形式のメッセージがメッセージログに追加される可能性があります。
ダウンコンバージョンを使用したトピックレベルのコンシューマーの最初のストラテジー
トピックごとに以下を実行します。
-
トピックレベルの
message.format.version
を、旧バージョンに変更します (または、デフォルトがブローカーレベルのlog.message.format.version
のトピックを利用します)。 - コンシューマーおよびプロデューサーとして機能するアプリケーションをすべてアップグレードします。
- アップグレードしたアプリケーションが正しく機能することを確認します。
-
トピックレベルの
message.format.version
を新バージョンに変更します。
このストラテジーにはブローカーのダウンコンバートが必要ですが、ダウンコンバートは一度に 1 つのトピック (またはトピックの小さなグループ) のみに必要になるので、ブローカーへの負荷は最小限に抑えられます。この方法は、同じトピックのコンシューマーとプロデューサーの両方に該当するアプリケーションにも通用します。この方法により、新しいメッセージ形式バージョンを使用する前に、アップグレードされたプロデューサーとコンシューマーが正しく機能することが保証されます。
この方法の主な欠点は、多くのトピックやアプリケーションが含まれるクラスターでの管理が複雑になる場合があることです。
複数のストラテジーを適用することもできます。たとえば、最初のいくつかのアプリケーションとトピックに、「トピック単位でコンシューマーを最初にアップグレード、ダウンコンバートあり」のストラテジーを適用します。これが問題なく適用されたら、より効率的な別のストラテジーの使用を検討できます。
9.5.3. Kafka バージョンおよびイメージマッピング
Kafka のアップグレード時に、STRIMZI_KAFKA_IMAGES
環境変数と Kafka.spec.kafka.version
プロパティーの設定について考慮してください。
-
それぞれの
Kafka
リソースはKafka.spec.kafka.version
で設定できます。 Cluster Operator の
STRIMZI_KAFKA_IMAGES
環境変数により、Kafka のバージョンと、指定のKafka
リソースでそのバージョンが要求されるときに使用されるイメージをマッピングできます。-
Kafka.spec.kafka.image
を設定しないと、そのバージョンのデフォルトのイメージが使用されます。 -
Kafka.spec.kafka.image
を設定すると、デフォルトのイメージがオーバーライドされます。
-
Cluster Operator は、Kafka ブローカーの想定されるバージョンが実際にイメージに含まれているかどうかを検証できません。所定のイメージが所定の Kafka バージョンに対応することを必ず確認してください。
9.5.4. Kafka ブローカーおよびクライアントアプリケーションのアップグレード
この手順では、AMQ Streams Kafka クラスターを最新のサポートされる Kafka バージョンにアップグレードする方法を説明します。
新しい Kafka バージョンを現在のバージョンと比較すると、新しいバージョンは ログメッセージ形式の上位バージョン、ブローカー間プロトコルの上位バージョン、またはその両方をサポートする可能性があります。必要に応じて、これらのバージョンをアップグレードする手順を実行します。詳細は、「Kafka バージョン」 を参照してください。
クライアントをアップグレードするストラテジーを選択する必要もあります。Kafka クライアントは、この手順の 6 でアップグレードされます。
前提条件
Kafka
リソースをアップグレードするには、以下を確認します。
- 両バージョンの Kafka をサポートする Cluster Operator が稼働している。
-
Kafka.spec.kafka.config
には、新しい Kafka バージョンでサポートされないオプションが含まれていない。
手順
Kafka クラスター設定を更新します。
oc edit kafka my-cluster
設定されている場合、
Kafka.spec.kafka.config
にlog.message.format.version
があり、inter.broker.protocol.version
が現在の Kafka バージョンのデフォルトに設定されていることを確認します。たとえば、Kafka のバージョン 3.0.0 から 3.1.0 へのアップグレードは以下のようになります。
kind: Kafka spec: # ... kafka: version: 3.0.0 config: log.message.format.version: "3.0" inter.broker.protocol.version: "3.0" # ...
log.message.format.version
およびinter.broker.protocol.version
が設定されていない場合、AMQ Streams では、次のステップの Kafka バージョンの更新後、これらのバージョンを現在のデフォルトに自動的に更新します。注記log.message.format.version
およびinter.broker.protocol.version
の値は、浮動小数点数として解釈されないように文字列である必要があります。Kafka.spec.kafka.version
を変更して、新しい Kafka バージョンを指定します。現在の Kafka バージョンのデフォルトでlog.message.format.version
およびinter.broker.protocol.version
のままにします。注記kafka.version
を変更すると、クラスターのすべてのブローカーがアップグレードされ、新しいブローカーバイナリの使用が開始されます。このプロセスでは、一部のブローカーは古いバイナリーを使用し、他のブローカーはすでに新しいバイナリーにアップグレードされています。Inter.broker.protocol.version
を変更しないと、ブローカーはアップグレード中も相互に通信を継続できます。たとえば、Kafka 3.0.0 から 3.1.0 へのアップグレードは以下のようになります。
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka spec: # ... kafka: version: 3.1.0 1 config: log.message.format.version: "3.0" 2 inter.broker.protocol.version: "3.0" 3 # ...
警告新しい Kafka バージョンの
inter.broker.protocol.version
が変更された場合は、Kafka をダウングレードできません。ブローカー間プロトコルのバージョンは、__consumer_offsets
に書き込まれたメッセージなど、ブローカーによって保存される永続メタデータに使用されるスキーマを判断します。ダウングレードされたクラスターはメッセージを理解しません。Kafka クラスターのイメージが
Kafka.spec.kafka.image
の Kafka カスタムリソースで定義されている場合、image
を更新して、新しい Kafka バージョンでコンテナーイメージを示すようにします。「Kafka バージョンおよびイメージマッピング」を参照してください。
エディターを保存して終了し、ローリングアップデートの完了を待ちます。
Pod の状態の遷移を監視して、ローリングアップデートの進捗を確認します。
oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'
ローリングアップデートにより、各 Pod が新バージョンの Kafka のブローカーバイナリーを使用するようになります。
クライアントのアップグレードに選択したストラテジーに応じて、新バージョンのクライアントバイナリーを使用するようにすべてのクライアントアプリケーションをアップグレードします。
必要に応じて、Kafka Connect および MirrorMaker の
version
プロパティーを新バージョンの Kafka として設定します。-
Kafka Connect では、
KafkaConnect.spec.version
を更新します。 -
MirrorMaker では、
KafkaMirrorMaker.spec.version
を更新します。 -
MirrorMaker 2.0 の場合は、
KafkaMirrorMaker2.spec.version
を更新します。
-
Kafka Connect では、
設定されている場合、新しい
inter.broker.protocol.version
バージョンを使用するように Kafka リソースを更新します。それ以外の場合は、ステップ 9 に進みます。たとえば、Kafka 3.1.0 へのアップグレードでは以下のようになります。
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka spec: # ... kafka: version: 3.1.0 config: log.message.format.version: "3.0" inter.broker.protocol.version: "3.1" # ...
- Cluster Operator によってクラスターが更新されるまで待ちます。
設定されている場合、新しい
log.message.format.version
バージョンを使用するように Kafka リソースを更新します。それ以外の場合は、ステップ 10 に進みます。たとえば、Kafka 3.1.0 へのアップグレードでは以下のようになります。
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka spec: # ... kafka: version: 3.1.0 config: log.message.format.version: "3.1" inter.broker.protocol.version: "3.1" # ...
重要Kafka 3.0.0 以降、
inter.broker.protocol.version
が3.0
以上に設定されていると、log.message.format.version
オプションは無視されるため、設定する必要はありません。Cluster Operator によってクラスターが更新されるまで待ちます。
- これで、Kafka クラスターおよびクライアントが新バージョンの Kafka を使用するようになります。
- ブローカーは、ブローカー間プロトコルバージョンと新バージョンの Kafka のメッセージ形式バージョンを使用して、メッセージを送信するように設定されます。
Kafka のアップグレードに従い、必要な場合は以下を行うことができます。
9.6. コンシューマーの Cooperative Rebalancing へのアップグレード
Kafka コンシューマーおよび Kafka Streams アプリケーションをアップグレードすることで、パーティションの再分散にデフォルトの Eager Rebalance プロトコルではなく Incremental Cooperative Rebalance プロトコルを使用できます。この新しいプロトコルが Kafka 2.4.0 に追加されました。
コンシューマーは、パーティションの割り当てを Cooperative Rebalance で保持し、クラスターの分散が必要な場合にプロセスの最後でのみ割り当てを取り消します。これにより、コンシューマーグループまたは Kafka Streams アプリケーションが使用不可能になる状態が削減されます。
Incremental Cooperative Rebalance プロトコルへのアップグレードは任意です。Eager Rebalance プロトコルは引き続きサポートされます。
前提条件
- Kafka 3.1.0 に Kafka ブローカーおよびクライアントアプリケーションをアップグレード済みであることが必要です。
手順
Incremental Cooperative Rebalance プロトコルを使用するように Kafka コンシューマーをアップグレードするには以下を行います。
-
Kafka クライアント
.jar
ファイルを新バージョンに置き換えます。 -
コンシューマー設定で、
partition.assignment.strategy
にcooperative-sticky
を追加します。たとえば、range
ストラテジーが設定されている場合は、設定をrange, cooperative-sticky
に変更します。 - グループ内の各コンシューマーを順次再起動し、再起動後に各コンシューマーがグループに再度参加するまで待ちます。
-
コンシューマー設定から前述の
partition.assignment.strategy
を削除して、グループの各コンシューマーを再設定し、cooperative-sticky
ストラテジーのみを残します。 - グループ内の各コンシューマーを順次再起動し、再起動後に各コンシューマーがグループに再度参加するまで待ちます。
Incremental Cooperative Rebalance プロトコルを使用するように Kafka Streams アプリケーションをアップグレードするには以下を行います。
-
Kafka Streams の
.jar
ファイルを新バージョンに置き換えます。 -
Kafka Streams の設定で、
upgrade.from
設定パラメーターをアップグレード前の Kafka バージョンに設定します (例: 2.3)。 - 各ストリームプロセッサー (ノード) を順次再起動します。
-
upgrade.from
設定パラメーターを Kafka Streams 設定から削除します。 - グループ内の各コンシューマーを順次再起動します。
第10章 AMQ Streams のダウングレード
アップグレードしたバージョンの AMQ Streams で問題が発生した場合は、インストールを直前のバージョンに戻すことができます。
以下のダウングレードを実行できます。
Cluster Operator を以前の AMQ Streams バージョンに戻します。
すべての Kafka ブローカーとクライアントアプリケーションを、以前の Kafka バージョンにダウングレードします。
以前のバージョンの AMQ Streams では使用している Kafka バージョンがサポートされない場合、メッセージに追加されるログメッセージ形式のバージョンが一致すれば Kafka をダウングレードすることができます。
10.1. Cluster Operator の以前のバージョンへのダウングレード
AMQ Streams で問題が発生した場合は、インストールを元に戻すことができます。
この手順では、Cluster Operator デプロイメントを以前のバージョンにダウングレードする方法を説明します。
前提条件
- 既存の Cluster Operator デプロイメントを利用できる必要があります。
- 以前のバージョンのインストールファイルがダウンロード済みである必要があります。
手順
-
既存の Cluster Operator リソース (
/install/cluster-operator
ディレクトリー内) に追加した設定変更を覚えておきます。すべての変更は、以前のバージョンの Cluster Operator によって上書きされます。 - カスタムリソースを元に戻して、ダウングレードする AMQ Streams バージョンで利用可能なサポート対象の設定オプションを反映します。
Cluster Operator を更新します。
Cluster Operator を実行している namespace に従い、以前のバージョンのインストールファイルを編集します。
Linux の場合は、以下を使用します。
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
MacOS の場合は、以下を使用します。
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml
-
既存の Cluster Operator
Deployment
で 1 つ以上の環境変数を編集した場合、install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml
ファイルを編集し、これらの環境変数を使用します。
設定を更新したら、残りのインストールリソースとともにデプロイします。
oc replace -f install/cluster-operator
ローリングアップデートが完了するのを待ちます。
Kafka Pod のイメージを取得して、アップグレードが正常に完了したことを確認します。
oc get pod my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'
イメージタグには、新しい AMQ Streams バージョンと Kafka バージョンが順に示されます。例:
NEW-STRIMZI-VERSION-kafka-CURRENT-KAFKA-VERSION
Cluster Operator は以前のバージョンにダウングレードされました。
10.2. Kafka のダウングレード
Kafka バージョンのダウングレードは、Cluster Operator によって実行されます。
10.2.1. ダウングレードでの Kafka バージョンの互換性
Kafka のダウングレードは、互換性のある現在およびターゲットの Kafka バージョン と、メッセージがログに記録された状態に依存します。
そのバージョンが、クラスターでこれまで使用された inter.broker.protocol.version
設定をサポートしない場合、または新しい log.message.format.version
を使用するメッセージメッセージにメッセージが追加された場合は、下位バージョンの Kafka に戻すことはできません。
Inter.broker.protocol.version
は、__consumer_offsets
に書き込まれたメッセージのスキーマなど、ブローカーによって保存される永続メタデータに使用されるスキーマを判断します。クラスターで以前使用された inter.broker.protocol.version
が認識されない Kafka バージョンにダウングレードすると、ブローカーが認識できないデータが発生します。
ダウングレードする Kafka のバージョンの関係は次のとおりです。
-
ダウングレードする Kafka バージョンの
log.message.format.version
が現行バージョンと 同じ である場合、Cluster Operator は、ブローカーのローリング再起動を 1 回実行してダウングレードを行います。 別の
log.message.format.version
の場合、ダウングレード後の Kafka バージョンが使用するバージョンに設定されたlog.message.format.version
が常に 実行中のクラスターに存在する場合に限り、ダウングレードが可能です。通常は、アップグレードの手順がlog.message.format.version
の変更前に中止された場合にのみ該当します。その場合、ダウングレードには以下が必要です。- 2 つのバージョンで Interbroker プロトコルが異なる場合、ブローカーのローリング再起動が 2 回必要です。
- 両バージョンで同じ場合は、ローリング再起動が 1 回必要です。
以前のバージョンでサポートされない log.message.format.version
が新バージョンで使われていた場合 (log.message.format.version
のデフォルト値が使われていた場合など)、ダウングレードは実行 できません。たとえば以下のリソースの場合、log.message.format.version
が変更されていないので、Kafka バージョン 3.0.0 にダウングレードできます。
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka spec: # ... kafka: version: 3.1.0 config: log.message.format.version: "3.0" # ...
log.message.format.version
が "3.1"
に設定されているか、値がない(このためパラメーターに 3.1.0 ブローカーのデフォルト値 3.1 が採用される)場合は、ダウングレードを実施できません。
Kafka 3.0.0 以降、inter.broker.protocol.version
が 3.0
以上に設定されていると、log.message.format.version
オプションは無視されるため、設定する必要はありません。
10.2.2. Kafka ブローカーおよびクライアントアプリケーションのダウングレード
この手順では、AMQ Streams Kafka クラスターを Kafka の下位 (以前の) バージョンにダウングレードする方法 (3.1.0 から 3.0.0 へのダウングレードなど) を説明します。
前提条件
AMQ Streams Kafka クラスターをダウングレードする前に、Kafka
リソースについて以下を確認してください。
- 重要:Kafka バージョンの互換性。
- 両バージョンの Kafka をサポートする Cluster Operator が稼働している。
-
Kafka.spec.kafka.config
に、ダウングレードする Kafka バージョンでサポートされていないオプションが含まれていない。 Kafka.spec.kafka.config
に、ダウングレード先の Kafka バージョンでサポートされるlog.message.format.version
とinter.broker.protocol.version
がある。Kafka 3.0.0 以降、
inter.broker.protocol.version
が3.0
以上に設定されていると、log.message.format.version
オプションは無視されるため、設定する必要はありません。
手順
Kafka クラスター設定を更新します。
oc edit kafka KAFKA-CONFIGURATION-FILE
Kafka.spec.kafka.version
を変更して、以前のバージョンを指定します。たとえば、Kafka 3.1.0 から 3.0.0 へのダウングレードは以下のようになります。
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka spec: # ... kafka: version: 3.0.0 1 config: log.message.format.version: "3.0" 2 inter.broker.protocol.version: "3.0" 3 # ...
注記log.message.format.version
およびinter.broker.protocol.version
の値は、浮動小数点数として解釈されないように文字列である必要があります。Kafka バージョンのイメージが Cluster Operator の
STRIMZI_KAFKA_IMAGES
に定義されているイメージとは異なる場合は、Kafka.spec.kafka.image
を更新します。「Kafka バージョンおよびイメージマッピング」 を参照してください。
エディターを保存して終了し、ローリングアップデートの完了を待ちます。
更新をログで確認するか、または Pod 状態の遷移を監視して確認します。
oc logs -f CLUSTER-OPERATOR-POD-NAME | grep -E "Kafka version downgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"
oc get pod -w
Cluster Operator ログで
INFO
レベルのメッセージを確認します。Reconciliation #NUM(watch) Kafka(NAMESPACE/NAME): Kafka version downgrade from FROM-VERSION to TO-VERSION, phase 1 of 1 completed
すべてのクライアントアプリケーション (コンシューマー) をダウングレードして、以前のバージョンのクライアントバイナリーを使用します。
これで、Kafka クラスターおよびクライアントは以前の Kafka バージョンを使用するようになります。
トピックメタデータの保存に ZooKeeper を使用する 1.7 よりも前のバージョンの AMQ Streams に戻す場合は、Kafka クラスターから内部トピックストアのトピックを削除します。
oc run kafka-admin -ti --image=registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 --rm=true --restart=Never -- ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi-topic-operator-kstreams-topic-store-changelog --delete && ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi_store_topic --delete
第11章 AMQ Streams のアンインストール
OpenShift Container Platform Web コンソールまたは CLI を使用して OperatorHub から AMQ Streams on OpenShift 4.6 - 4.10 をアンインストールできます。
AMQ Streams のインストールに使用したのと同じアプローチを使用します。
AMQ Streams をアンインストールする場合は、デプロイメント専用に作成され、AMQ Streams リソースから参照されるリソースを特定する必要があります。
このようなリソースには以下があります。
- シークレット (カスタム CA および証明書、Kafka Connect Secrets、その他の Kafka シークレット)
-
ロギング
ConfigMap
(external
タイプ)
Kafka
、KafkaConnect
、KafkaMirrorMaker
、KafkaBridge
のいずれかの設定で参照されるリソースです。
CustomResourceDefinitions
を削除すると、対応するカスタムリソース(Kafka
、KafkaConnect
、KafkaMirrorMaker
、または KafkaBridge
)、およびそれらに依存するリソース(Deployments、StatefulSets、およびその他の依存リソース)のガベージコレクションが実行されます。
11.1. Web コンソールを使用した OperatorHub からの AMQ Streams のアンインストール
この手順では、OperatorHub から AMQ Streams をアンインストールし、デプロイメントに関連するリソースを削除する方法を説明します。
コンソールから手順を実行したり、別の CLI コマンドを使用したりできます。
前提条件
-
cluster-admin
またはstrimzi-admin
パーミッションを持つアカウントを使用して OpenShift Container Platform Web コンソールにアクセスできること。 削除するリソースを特定している。
AMQ Streams をアンインストールしたら、以下の
oc
CLI コマンドを使用してリソースを検索して、削除されていることを確認できます。AMQ Streams デプロイメントに関連するリソースを検索するコマンド
oc get <resource_type> --all-namespaces | grep <kafka_cluster_name>
<resource_type> は、
secret
またはconfigmap
などのチェックするリソースのタイプに置き換えます。
手順
- OpenShift Web コンソールで、Operators > Installed Operators に移動します。
インストールされた Red Hat Integration - AMQ Streams Operator については、オプションアイコン(3 つの縦の点)を選択し、Uninstall Operator をクリックします。
Operator が Installed Operators から削除されます。
- Home > Projects に移動し、AMQ Streams と Kafka コンポーネントがインストールされているプロジェクトを選択します。
Inventory のオプションをクリックして関連リソースを削除します。
リソースには以下が含まれます。
- デプロイメント
- StatefulSets
- Pod
- サービス
- ConfigMap
- シークレット
ヒント検索を使用して、Kafka クラスターの名前で始まる関連リソースを検索します。また、Workloads でもリソースを検索できます。
代わりの CLI コマンド
CLI コマンドを使用して、OperatorHub から AMQ Streams をアンインストールできます。
AMQ Streams サブスクリプションを削除します。
oc delete subscription amq-streams -n openshift-operators
クラスターサービスバージョン (CSV) を削除します。
oc delete csv amqstreams.<version> -n openshift-operators
関連する CRD を削除します。
oc get crd -l app=strimzi -o name | xargs oc delete
11.2. CLI を使用した AMQ Streams のアンインストール
この手順では、oc
コマンドラインツールを使用して AMQ Streams をアンインストールし、デプロイメントに関連するリソースを削除する方法を説明します。
前提条件
削除するリソースを特定している。
AMQ Streams をアンインストールしたら、以下の
oc
CLI コマンドを使用してリソースを検索して、削除されていることを確認できます。AMQ Streams デプロイメントに関連するリソースを検索するコマンド
oc get <resource_type> --all-namespaces | grep <kafka_cluster_name>
<resource_type> は、
secret
またはconfigmap
などのチェックするリソースのタイプに置き換えます。
手順
Cluster Operator
Deployment
、関連するCustomResourceDefinitions
、およびRBAC
リソースを削除します。Cluster Operator のデプロに使用されるインストールファイルを指定します。
oc delete -f amq-streams-<version>/install/cluster-operator
前提条件で特定したリソースを削除します。
oc delete <resource_type> <resource_name> -n <namespace>
<resource_type> は削除するリソースのタイプに、 <resource_name> はリソースの名前に置き換えます。
シークレットの削除例
oc delete secret my-cluster-clients-ca -n my-project
第12章 AMQ Streams でのメータリングの使用
OCP 4 で利用可能なメータリングツールを使用して、異なるデータソースからメータリングレポートを生成できます。クラスター管理者として、メータリングを使用してクラスターの内容を分析できます。独自のクエリーを作成するか、または事前定義 SQL クエリーを使用して、利用可能な異なるデータソースからデータを処理する方法を定義できます。Prometheus をデフォルトのデータソースとして使用すると、Pod、namespace、およびその他ほとんどの Kubernetes リソースのレポートを生成できます。
OpenShift の Metering Operator を使用すると、インストールされた AMQ Streams コンポーネントを分析し、Red Hat サブスクリプションに準拠しているかどうかを判断することができます。
AMQ Streams でメータリングを使用するには、まず OpenShift Container Platform に Metering operator をインストールし、設定する必要があります。
12.1. メータリングリソース
メータリングには、メータリングのデプロイメントやインストール、およびメータリングが提供するレポート機能を管理するために使用できるリソースが多数含まれています。メータリングは以下の CRD を使用して管理されます。
名前 | 説明 |
---|---|
| デプロイメントのメータリングスタックを設定します。メータリング スタックを構成する各コンポーネントを制御するカスタマイズおよび設定オプションが含まれます。 |
| 使用するクエリー、クエリーを実行するタイミングおよび頻度、および結果を保存する場所を制御します。 |
|
|
| ReportQuery および Report で利用可能なデータを制御します。メータリング内で使用できるように複数の異なるデータベースへのアクセスの設定を可能にします。 |
12.2. AMQ Streams のメータリングラベル
以下の表では、AMQ Streams インフラストラクチャーコンポーネントおよびインテグレーションのメータリングラベルが一覧表示されています。
ラベル | 可能な値 |
---|---|
|
|
|
|
|
|
|
|
|
|
| インフラストラクチャー
|
アプリケーション
| |
|
|
例
インフラストラクチャーの例 (インフラストラクチャーコンポーネントが entity-operator の場合)
com.company=Red_Hat rht.prod_name=Red_Hat_Integration rht.prod_ver=2022.Q2 rht.comp=AMQ_Streams rht.comp_ver=2.1 rht.subcomp=entity-operator rht.comp_t=infrastructure
アプリケーションの例 (インテグレーションのデプロイメント名が kafka-bridge の場合)
com.company=Red_Hat rht.prod_name=Red_Hat_Integration rht.prod_ver=2022.Q2 rht.comp=AMQ_Streams rht.comp_ver=2.1 rht.subcomp=kafka-bridge rht.comp_t=application
付録A サブスクリプションの使用
AMQ Streams は、ソフトウェアサブスクリプションから提供されます。サブスクリプションを管理するには、Red Hat カスタマーポータルでアカウントにアクセスします。
アカウントへのアクセス
- access.redhat.com に移動します。
- アカウントがない場合は、作成します。
- アカウントにログインします。
サブスクリプションのアクティベート
- access.redhat.com に移動します。
- サブスクリプション に移動します。
- Activate a subscription に移動し、16 桁のアクティベーション番号を入力します。
Zip および Tar ファイルのダウンロード
zip または tar ファイルにアクセスするには、カスタマーポータルを使用して、ダウンロードする関連ファイルを検索します。RPM パッケージを使用している場合は、この手順は必要ありません。
- ブラウザーを開き、access.redhat.com/downloads で Red Hat カスタマーポータルの Product Downloads ページにログインします。
- INTEGRATION AND AUTOMATION カテゴリーで、AMQ Streams for Apache Kafka エントリーを見つけます。
- 必要な AMQ Streams 製品を選択します。Software Downloads ページが開きます。
- コンポーネントの Download リンクをクリックします。
改訂日時: 2022-07-12 13:27:23 +1000