AMQ Streams on OpenShift を使い始める

手順

1. OpenShift Web コンソールで Home > Projects ページに移動し、インストール用のプロジェクト (namespace) を作成します。

   この例では、amq-streams-kafka という名前のプロジェクトを使用します。

2. Operators > OperatorHub ページに移動します。

3. スクロール、または Filter by keyword ボックスにキーワードを入力して、Red Hat Integration - AMQ Streams Operator を見つけます。

   Operator は、Streaming & Messaging カテゴリーにあります。

4. Red Hat Integration - AMQ Streams をクリックして、Operator 情報を表示します。
5. Operator に関する情報を確認し、Install をクリックします。

6. 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 コンポーネント専用とすることが推奨されます。

   • Update approval: デフォルトでは、OLM (Operator Lifecycle Manager) によって、AMQ Streams Operator が自動的に最新の AMQ Streams バージョンにアップグレードされます。今後のアップグレードを手動で承認する場合は、Manual を選択します。詳細は、OpenShift ドキュメントの Operators ガイドを参照してください。

7. Install をクリックして、選択した namespace に Operator をインストールします。

   AMQ Streams Operator によって、Cluster Operator、CRD、およびロールベースアクセス制御 (RBAC) リソースは選択された namespace にデプロイされます。

8. Operator を使用する準備ができたら、Operators > Installed Operators に移動して、Operator が選択した namespace にインストールされていることを確認します。

   ステータスは Succeeded と表示されます。

   これで、AMQ Streams Operator を使用して、Kafka クラスターから始めて Kafka コンポーネントをデプロイできます。

手順

1. Web コンソールで Operators > Installed Operators ページに移動し、Red Hat Integration - AMQ Streams をクリックして、Operator の詳細を表示します。

   提供されている API から、Kafka コンポーネントのインスタンスを作成できます。

2. Kafka の下の Create instance をクリックして、Kafka インスタンスを作成します。

   デフォルトでは、3 つの Kafka ブローカーノードと 3 つの ZooKeeper ノードを持つ my-cluster という名の Kafka クラスターを作成します。クラスターはエフェメラルストレージを使用します。

3. Create をクリックして、Kafka のインストールを開始します。

   ステータスが Ready に変わるまで待ちます。

手順

1. Web コンソールで Operators > Installed Operators ページに移動し、Red Hat Integration - AMQ Streams を選択して、Operator の詳細を表示します。
2. Kafka ページを選択して、インストールされている Kafka クラスターを表示します。

3. 設定している Kafka クラスターの名前をクリックして、その詳細を表示します。

   この例では、my-cluster という名前の Kafka クラスターを使用します。

4. Kafka クラスター my-cluster の YAML ページを選択します。

5. ルートリスナー設定を追加して、listener1 という名前の OpenShift ルートを作成します。

   リスナー設定は、route タイプに設定する必要があります。Kafka 設定の listeners の下にリスナー設定を追加します。

   外部ルートリスナーの設定

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     namespace: amq-streams-kafka
   spec:
     kafka:
       # ...
       listeners:
         # ...
         - name: listener1
           port: 9094
           type: route
           tls: true
   # ...

   クライアントはデフォルトのルーターポートであるポート 443 に接続しますが、トラフィックは設定するポート (この例では 9094) にルーティングされます。

6. 更新された設定を保存します。

7. Kafka クラスター my-cluster の Resources ページを選択して、クライアントに必要な接続情報を見つけます。

   Resources ページから、Kafka クラスターに接続するために必要なルートリスナーと公開クラスター証明書の詳細を確認できます。

8. Kafka クラスター用に作成された my-cluster-kafka-listener1-bootstrap ルートの名前をクリックして、ルートの詳細を表示します。

9. ホスト名をメモします。

   ホスト名は、Kafka クラスターに接続するためのブートストラップアドレスとして、Kafka クライアントのポート 443 で指定されます。

   Networking > Routes に移動し、amq-streams-kafka プロジェクトを選択して、namespace に作成されたルートを表示することにより、ブートストラップアドレスを見つけることもできます。

   または、oc ツールを使用してブートストラップの詳細を抽出できます。

   ブートストラップ情報の抽出

   oc get routes my-cluster-kafka-listener1-bootstrap -o=jsonpath='{.status.ingress[0].host}{"\n"}'

10. Resources ページに戻り、my-cluster-cluster-ca-cert の名前をクリックして、Kafka クラスターにアクセスするためのシークレットの詳細を表示します。

    ca.crt 証明書ファイルには、Kafka クラスターの公開証明書が含まれています。

    Kafka ブローカーにアクセスするには証明書が必要です。

11. ca.crt 公開証明書ファイルのローカルコピーを作成します。

    証明書の詳細をコピーするか、OpenShift oc ツールを使用してそれらを抽出できます。

    公開証明書の抽出

    oc extract secret/my-cluster-cluster-ca-cert --keys=ca.crt --to=- > ca.crt

12. keytool を使用して、公開クラスター証明書のローカルトラストストアを作成します。

    ローカルトラストストアの作成

    keytool -keystore client.truststore.jks -alias CARoot -import -file ca.crt

    プロンプトが表示されたら、トラストストアにアクセスするためのパスワードを作成します。

    トラストストアは、Kafka クラスターへのアクセスを認証するために Kafka クライアントで指定されます。

    メッセージの送受信を開始する準備が整いました。

1. oc コマンドラインインターフェイスを使用して、Kafka プロデューサーをデプロイします。

   この例では、Kafka クラスター my-cluster に接続する Kafka プロデューサーをデプロイします。

   my-topic という名前のトピックが作成されます。

   Kafka プロデューサーの OpenShift へのデプロイ

   oc run kafka-producer -ti \
   --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 \
   --rm=true \
   --restart=Never \
   -- bin/kafka-console-producer.sh \
   --bootstrap-server my-cluster-kafka-bootstrap:9092 \
   --topic my-topic

   注記

   接続に失敗した場合は、Kafka クラスターが実行中で、正しいクラスター名が bootstrap-server として指定されていることを確認してください。

2. コマンドプロンプトから、いくつかのメッセージを入力します。
3. OpenShift Web コンソールで Home > Projects ページに移動し、作成した amq-streams-kafka プロジェクトを選択します。
4. Pod のリストから、kafka-producer をクリックして、プロデューサー Pod の詳細を表示します。
5. Logs ページを選択して、入力したメッセージが存在することを確認します。

6. oc コマンドラインインターフェイスを使用して、Kafka コンシューマーをデプロイします。

   Kafka コンシューマーの OpenShift へのデプロイ

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

   コンシューマーは my-topic に生成されたメッセージを消費しました。

7. コマンドプロンプトから、コンシューマーコンソールに着信メッセージが表示されていることを確認します。
8. OpenShift Web コンソールで Home > Projects ページに移動し、作成した amq-streams-kafka プロジェクトを選択します。
9. Pod のリストから、kafka-consumer をクリックして、コンシューマー Pod の詳細を表示します。
10. Logs ページを選択して、消費したメッセージが存在することを確認します。

1. AMQ Streams ソフトウェアダウンロードページ から AMQ Streams <version> バイナリーをダウンロードして展開します。

   amq-streams-<version>-bin.zip ファイルを任意の場所に解凍します。

2. コマンドラインインターフェイスを開き、トピック my-topic と TLS の認証プロパティーを使用して Kafka コンソールプロデューサーを起動します。

   OpenShift ルートを使用して Kafka ブローカーにアクセスする のに必要なプロパティーを追加します。

   • 使用している OpenShift ルートのホスト名およびポート 443 を使用します。

   • パスワードと、ブローカー証明書用に作成したトラストストアへの参照を使用します。

     ローカル Kafka プロデューサーの起動

     kafka-console-producer.sh \
     --bootstrap-server my-cluster-kafka-listener1-bootstrap-amq-streams-kafka.apps.ci-ln-50kcyvt-72292.origin-ci-int-gce.dev.rhcloud.com:443 \
     --producer-property security.protocol=SSL \
     --producer-property ssl.truststore.password=password \
     --producer-property ssl.truststore.location=client.truststore.jks \
     --topic my-topic

3. プロデューサーが実行しているコマンドラインインターフェイスにメッセージを入力します。
4. Enter を押してメッセージを送信します。

5. 新しいコマンドラインインターフェイスタブまたはウィンドウを開き、Kafka コンソールコンシューマーを起動してメッセージを受信します。

   プロデューサーと同じ接続の詳細を使用します。

   ローカル Kafka コンシューマーの起動

   kafka-console-consumer.sh \
   --bootstrap-server my-cluster-kafka-listener1-bootstrap-amq-streams-kafka.apps.ci-ln-50kcyvt-72292.origin-ci-int-gce.dev.rhcloud.com:443 \
   --consumer-property security.protocol=SSL \
   --consumer-property ssl.truststore.password=password \
   --consumer-property ssl.truststore.location=client.truststore.jks \
   --topic my-topic --from-beginning

6. コンシューマーコンソールに受信メッセージが表示されることを確認します。
7. Crtl+C を押して、Kafka コンソールプロデューサーとコンシューマーを終了します。