Red Hat OpenShift on IBM CloudにAMQ Streamsを入れてみる

#はじめに
IBM CloudのManaged OpenShift環境で、Kafkaクラスタの検証を行うため、Red Hat AMQ StreamsをOperatorで入れた手順です。

#環境概要
・IBM Cloud Gen2 VPC上のOpenShift v4.5 (Multi-zone cluster)
・AMQ Streams(Kafka) v1.6.2
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/cc898b64-b488-dc16-2d13-1e7f6f2a2a52.png)



#導入手順

## OpenShiftクラスタの準備
事前準備として、IBM Cloud上にOpenShiftクラスタを作成します。
本検証では、OpenShift4.5(Gen2 VPC)のクラスタを1つ作成し、AMQ Streams用にWorker Node(Flavor:bx2.4x16)を3つのAZに合計3台用意しました。
参照：https://cloud.ibm.com/docs/openshift?topic=openshift-clusters


## AMQ Streamsの導入
### 1. Projectの作成
・OpenShiftのWebコンソールにログインし、Home→Projectsから任意の名前で新規プロジェクト`amqstreams`を作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/5dfa1676-7567-73b7-9342-3fc538f39878.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/404f8f5f-4938-fd1b-53ca-5d3ff614cc5c.png)


### 2. AMQ Streams Operatorの導入
まず、OperatorHubからAMQ StreamsのOperatorを導入します。
・コンソール左のOperators→OperatorHubを選択し、All Itemsの検索ボックスからAMQ Streamsを検索します。
provided by Red Hatの`Red Hat Integration - AMQ Streams`を探します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/f9aeec61-9cdd-a122-f132-a28891163cdf.png)

・AMQ Streamsを選択し、Installをクリックします。
*AMQ StreamsのOperatorは、Capability LevelがL4 Deep Insightsとなっており、バックアップ・リストアのFull Lifecycleに加え、ロギング、アラート、メトリクス取得も対応していることがわかります。
参考:https://operatorframework.io/operator-capabilities/
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/529e6ef6-bf15-865e-0e39-8de3dafeb4d9.png)

・Update Channelはsubscriptionのバージョン更新の時に利用されるチャネルですが、ここでは`stable`のままとします。
・Installation Modeは、本検証でのKafka導入のProjectは最初に作成したProjectに限定するため、`A specific namespace on the cluster`を選択の上、Installed Namespaceに作成済みProjectを選びます。
・Approval Strategyは、更新の自動化を指定しますが、ここでは`Automatic`のままとします。
・Installをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/f575b61b-213d-1ed0-2b8f-2f4ff7d298a3.png)

・Installed Operators画面でStatusがSucceededになることを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/7ae41994-a795-10ee-0aae-c376fdc8519c.png)

・選択したProjectでoperatorのPodがRunningになっていることが確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/aa36e6f7-9d58-f3dd-83b8-725ae82eb14a.png)

### 3. Kafkaクラスタの構成
続いて、KafkaクラスタおよびZookeeperクラスタの構成を行います。
・Installed Operators画面から導入済みの`Red Hat Integration - AMQ Streams`を選択します。
・Provided APIsの、Kafkaの`Create Instance`をクリックします。
*AMQ StremasのOperatorには、Kafkaクラスタ以外にもTopic作成、Kafka Connectorなどに加え、ノード追加時などにpartition replicaをリバランスするKafka Rebalanceも含まれています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/2a5ea080-8ac3-c557-9af3-87d97d2d5565.png)

・Kafkaの初期設定はForm Viewより構成します。Kafkaクラスタ名は任意の名前にします。
*もしForm Viewで提供されていない設定を編集したい場合はYAML Viewから修正します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/acd87e8d-e384-a4be-3c69-a6c01c40cd68.png)
・本検証では、IBM CloudのBlock StorageをPersistent VolumeとしてDynamic Provisioningするため、Storageを開き、Kafka storageに`persistent-claim`を選択し、size(推奨100Gi)とclassを指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/6eff138e-c399-4b8e-dfa3-6908f3f5430f.png)

・Kafka Brokersは3ノード構成のため`3`、バージョン`2.6.0`はデフォルトのままとします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/99823523-f7b4-4895-3c1f-fb45d2c062a3.png)



・本検証では、OpenShiftクラスタ内に別用途のworker nodeも構成予定のため、KafkaのPodが稼働するノードをAffinityで指定します。
・Node Affinityを開いて、ここではRequired During Scheduling Ignored DuringのNode Selectorに、今回配置したいノードのラベルを記載します。この環境ではworker-pool-nameというラベルで識別します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/1c605a12-7e65-575c-f8b1-2c2d298ec2e7.png)


・Rackは、zoneや物理ラックなどの障害時のデータロスを避けるため、Kafkaクラスタ内のPartition Replicaを地理的に分散配置する設定です。本検証では3つのAZを使っているため、zoneawareのTopology Keyを追加します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/c227e2da-6315-e6d5-d2ff-8a3e1eb64d41.png)
・その他はデフォルトのままとします。
*クラスタ外部からTLS通信した場合にはTLS Sidecarの設定をする必要があります。

### 4. Zookeeperクラスタの構成
Kafka構成のForm Viewの続きで、Zookeeperクラスタの構成を行います。

・ZookeeperもStorageを開いてPV構成のために`persistent-claim`を選択し、classとsizeを指定します。(最小10Gi)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/9b788a51-a700-260e-3f4f-d3ff0d782139.png)

・Zookeeper Nodesは最小の`3`とします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/f1fef878-52af-ce14-694c-f5b82a0ff5f0.png)

・Kafkaクラスタと同様のノードに配置するため、上記と同じNode Affinityを設定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/a3e1b07a-6737-b238-fa04-6e2b699dc729.png)

・その他はデフォルトのままとし、最後の`Create`をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/71b5c45a-94d4-d937-9302-0c0acf31720c.png)

・・・
・余所見していた十数分後にはもうKafkaクラスタができていました。
All Instancesから作成したKafkaクラスタがConditions:Readyになっていることを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/d9e96125-761c-5ec1-fba5-a0b6c409bb9d.png)
*`Warning`が出てますが。。affinityのpathがdeprecatedみたいです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/79993413-21e4-4123-40a8-0f3c637f737c.png)
YAML viewから修正してみます。zookeeperも同様に。
_`before`_
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/666dd63c-c1e8-c785-f7c8-98ddd2a622a1.png)
_`after`_
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/61cf64ce-391c-64f6-7abe-375684dfdce6.png)
修正してSaveしたらWarningは消えたみたいです。良かった。。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/d6f26973-7631-a345-47f4-c629d7107c7a.png)

・Resourcesを見ると、たくさんリソースができています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/9c574d31-f596-4dfd-2add-50fab26924b8.png)
・KafkaとZookeeperのStatefulSetのPodがそれぞれ3つできていることを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/588120ae-016b-5e7c-686c-7f3a0b098ee9.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/463a8dbd-4e7f-3239-c5e4-db946ab93785.png)

・PV,PVCや接続用のbootstrap Serviceもできていることを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/cd4b22b3-714b-4491-8c25-b8b3c88b78bf.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/47d689f5-e23a-ea41-398d-2f7a6790269e.png)

### 5. Topicの作成
次に、AMQ StreamsのOperatorのProvided APIからKafka Topicを作成します。Kafka Topicの`Create Instance`をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/4e48d39a-60fa-3f24-4f3b-cb8430374e3f.png)

・Form Viewより、トピック名、Partition数、Replication factor数`3`を指定し、`Create`をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/e2778666-8c1a-a3fa-b68c-ae2226426598.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/3927a17f-eb6b-f8c9-29d0-a271b1c09307.png)

・Condition:ReadyでTopicが作成されたことを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/176470/fcaf8b6f-c563-d0e8-d51d-eb8b4673bdf4.png)

### 6. メッセージのPublish/Subscribeのテスト
Kafka producerとKafka comsumerのコンテナをメッセージの出し入れを確認します。
・Kafkaクラスタと同一Project(namespace)にて実行します。
・Kafkaクラスタのbootstrapサービス名を確認します。

```shell
$ oc project
Using project "amqstreams" on server "https://xxxxx.containers.cloud.ibm.com:xxxxx".
$ oc get svc
NAME                          TYPE        CLUSTER-IP       EXTERNAL-IP   PORT(S)                      AGE
my-cluster-kafka-bootstrap    ClusterIP   172.21.163.220   <none>        9091/TCP,9092/TCP,9093/TCP   12h
my-cluster-kafka-brokers      ClusterIP   None             <none>        9091/TCP,9092/TCP,9093/TCP   12h
my-cluster-zookeeper-client   ClusterIP   172.21.230.221   <none>        2181/TCP                     12h
my-cluster-zookeeper-nodes    ClusterIP   None             <none>        2181/TCP,2888/TCP,3888/TCP   12h
```

・Kafka producerをデプロイし、kafka-console-producer.shを実行し、プロンプトに任意のメッセージを入れます。

```shell
$ oc run kafka-producer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9092 --topic my-topic
If you don't see a command prompt, try pressing enter.
>hello001
>hello002
>hello003
>hello004
>^Cpod "kafka-producer" deleted
pod amqstreams/kafka-producer terminated (Error)
```
・メッセージを確認するため、次にKafka consumerをデプロイし、kafka-console-consumer.shを実行します。

```shell
$ oc run kafka-consumer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic my-topic --from-beginning
If you don't see a command prompt, try pressing enter.
hello001
hello002
hello004
hello003
```

・メッセージが確認できました！

導入確認はここまでにして、次回はKafkaのメトリクス取得を行いたいと思います。



参考文献：
1.[Using AMQ Streams on OpenShift](https://access.redhat.com/documentation/en-us/red_hat_amq/2020.q4/html-single/using_amq_streams_on_openshift/index)
2.[Chapter 4. Deploying AMQ Streams](https://access.redhat.com/documentation/en-us/red_hat_amq/2020.q4/html/deploying_and_upgrading_amq_streams_on_openshift/deploy-tasks_str)
3.[Chapter 5. Setting up client access to the Kafka cluster](https://access.redhat.com/documentation/en-us/red_hat_amq/2020.q4/html/deploying_and_upgrading_amq_streams_on_openshift/deploy-verify_str)
4.[OpenShift + AMQ Streams(Apache Kafka) + QUARKUS で実現するアプリケーション](https://kenta-kosugi.medium.com/openshift-amq-streams-apache-kafka-quarkus-%E3%81%A7%E5%AE%9F%E7%8F%BE%E3%81%99%E3%82%8B%E3%82%A2%E3%83%97%E3%83%AA%E3%82%B1%E3%83%BC%E3%82%B7%E3%83%A7%E3%83%B3-17481f0dd63d)