#4 Confluent for Kubernetes を使用して AKS 上に Confluent Platform を構成してみました - Topic作成編

## 概要

Confluent for Kubernetes （CFK）は、プライベートクラウド環境（今回は Azure Kubernetes Service（AKS））に Confluent をデプロイして管理するためのクラウドネイティブのコントロールプレーンです。宣言型 API で Confluent Platform をカスタマイズ、デプロイ、管理するための標準的で簡素なインターフェイスが備わっています。

CFK を使用して AKS上にデプロイされている Confluent Platform に Topic を作成するためのワークフローの概要は以下のとおりです。

1. Kubernetes 環境を準備します（事前準備で完了）
2. Confluent for Kubernetes をデプロイします（事前準備で完了）
3. Confluent Platform をデプロイします（事前準備で完了）
4. Confluent Platform の追加構成をします
5. Confluent Platform の追加デプロイ（Topic作成）をします

デプロイ後のイメージは以下となります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/e1cfdff8-3171-d81c-0c9f-13f7bcba359b.png)

---

### ローカル環境

- macOS Monterey 12.3.1
- python 3.8.12
- Azure CLI 2.34.1
- helm v3.6.3
- kubectl v1.21.3

### 事前準備

1. [この記事](https://qiita.com/turupon/items/b2f679ec6f9a69e30995) を実行して、AKSクラスタ環境が構築されていること
2. [この記事](https://qiita.com/turupon/items/d73771e949dd3e644930) を実行して、AKS 上に Confluent Platform をデプロイできていること

---

## Confluent Platform の追加構成

### Bearer authentication secret の作成（Bearer認証）

Topic作成のための Kafka Rest Class 利用時の認証用 ID/Password の定義ファイルを作成します

```restapibasic.txt
username=kafka
password=kafka-secret
```

### Kafka Rest Class（Admin Rest API） の yaml ファイルの作成

```kafka_rest_class.yaml
apiVersion: platform.confluent.io/v1beta1
kind: KafkaRestClass
metadata:
  name: default
  namespace: akscfk231
spec:
  kafkaRest:
    authentication:
      type: bearer
      bearer:
        secretRef: rest-credential
```

### Topic作成用 の yaml ファイルの定義

```kafka_topic.yaml
---
apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name: topic001
  namespace: akscfk231
spec:
  replicas: 3
  partitionCount: 1
  kafkaRest:
    authentication:
      type: bearer
      bearer:
        secretRef: rest-credential
  configs:
    cleanup.policy: "delete"
---
apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name: topic002
  namespace: akscfk231
spec:
  replicas: 3
  partitionCount: 1
  kafkaRest:
    authentication:
      type: bearer
      bearer:
        secretRef: rest-credential
  configs:
    cleanup.policy: "delete"
```

---

## Confluent Platform への追加デプロイ

### AKS環境に認証情報を作成します

```
$ kubectl create secret generic rest-credential --from-file=bearer.txt=./restapibearer.txt
```

### Admin REST API のデプロイ

Admin REST APIをAKSクラスターの中に作成

```
$ kubectl apply -f kafka_rest_class.yaml
```

### Confluent Platform への Topic の作成

Custom Resource から Admin REST API を通じて Topic を作成

```
$ kubectl apply -f kafka_topic.yaml
```

作成した Topic の確認

```
$ kubectl get topic 
NAME       REPLICAS   PARTITION   STATUS    CLUSTERID                AGE
topic001   3          1           CREATED   b7gKmMeBQK2NEKC139fsog   42s
topic002   3          1           CREATED   b7gKmMeBQK2NEKC139fsog   42s
```

### デプロイされた Confluent Platform リソースの確認

```
$ kubectl get confluent
NAME                                        REPLICAS   PARTITION   STATUS    CLUSTERID                AGE
kafkatopic.platform.confluent.io/topic001   3          1           CREATED   b7gKmMeBQK2NEKC139fsog   2m35s
kafkatopic.platform.confluent.io/topic002   3          1           CREATED   b7gKmMeBQK2NEKC139fsog   2m35s

NAME                                                  REPLICAS   READY   STATUS    AGE
schemaregistry.platform.confluent.io/schemaregistry   1          1       RUNNING   63m

NAME                                        REPLICAS   READY   STATUS    AGE
zookeeper.platform.confluent.io/zookeeper   3          3       RUNNING   63m

NAME                                                  REPLICAS   READY   STATUS    AGE
kafkarestproxy.platform.confluent.io/kafkarestproxy   1          1       RUNNING   63m

NAME                                REPLICAS   READY   STATUS    AGE
kafka.platform.confluent.io/kafka   3          3       RUNNING   63m

NAME                                           AGE
kafkarestclass.platform.confluent.io/default   4m55s

NAME                                  REPLICAS   READY   STATUS    AGE
ksqldb.platform.confluent.io/ksqldb   1          1       RUNNING   63m

NAME                                                REPLICAS   READY   STATUS    AGE
controlcenter.platform.confluent.io/controlcenter   1          1       RUNNING   63m

NAME                                    REPLICAS   READY   STATUS    AGE
connect.platform.confluent.io/connect   1          1       RUNNING   63m


## kafka rest class の詳細確認
$ kubectl describe kafkarestclass          
Name:         default
Namespace:    akscfk231
Labels:       <none>
Annotations:  <none>
API Version:  platform.confluent.io/v1beta1
Kind:         KafkaRestClass
Metadata:
  Creation Timestamp:  2022-06-29T06:28:19Z
  Finalizers:
    kafkarestclass.finalizers.platform.confluent.io
  Generation:  1
  Managed Fields:
    API Version:  platform.confluent.io/v1beta1
    Fields Type:  FieldsV1
    fieldsV1:
      f:metadata:
        f:annotations:
          .:
          f:kubectl.kubernetes.io/last-applied-configuration:
      f:spec:
        .:
        f:kafkaRest:
          .:
          f:authentication:
            .:
            f:bearer:
              .:
              f:secretRef:
            f:type:
    Manager:      kubectl-client-side-apply
    Operation:    Update
    Time:         2022-06-29T06:28:19Z
    API Version:  platform.confluent.io/v1beta1
    Fields Type:  FieldsV1
    fieldsV1:
      f:metadata:
        f:finalizers:
          .:
          v:"kafkarestclass.finalizers.platform.confluent.io":
    Manager:      manager
    Operation:    Update
    Time:         2022-06-29T06:28:19Z
    API Version:  platform.confluent.io/v1beta1
    Fields Type:  FieldsV1
    fieldsV1:
      f:status:
        .:
        f:endpoint:
        f:kafkaClusterID:
    Manager:         manager
    Operation:       Update
    Subresource:     status
    Time:            2022-06-29T06:28:20Z
  Resource Version:  27349
  UID:               15aaae84-8330-4459-89b2-a3f19d2a1dd5
Spec:
  Kafka Rest:
    Authentication:
      Bearer:
        Secret Ref:  rest-credential
      Type:          bearer
Status:
  Endpoint:          http://kafka.akscfk231.svc.cluster.local:8090
  Kafka Cluster ID:  b7gKmMeBQK2NEKC139fsog
Events:              <none>


## kafkatopic の詳細確認
$ kubectl describe kafkatopic       
Name:         topic001
     :
    省略
     :


Name:         topic002
Namespace:    akscfk231
Labels:       <none>
Annotations:  platform.confluent.io/last-spec-hash: WzQ9Vg/QcNB+03C54Sk+/m2scCu9E6oQpByPckcUWJ0=
API Version:  platform.confluent.io/v1beta1
Kind:         KafkaTopic
Metadata:
  Creation Timestamp:  2022-06-29T06:30:39Z
  Finalizers:
    kafkatopic.finalizers.platform.confluent.io
  Generation:  1
  Managed Fields:
    API Version:  platform.confluent.io/v1beta1
    Fields Type:  FieldsV1
    fieldsV1:
      f:metadata:
        f:annotations:
          .:
          f:kubectl.kubernetes.io/last-applied-configuration:
      f:spec:
        .:
        f:configs:
          .:
          f:cleanup.policy:
        f:kafkaRest:
          .:
          f:authentication:
            .:
            f:bearer:
              .:
              f:secretRef:
            f:type:
        f:partitionCount:
        f:replicas:
    Manager:      kubectl-client-side-apply
    Operation:    Update
    Time:         2022-06-29T06:30:39Z
    API Version:  platform.confluent.io/v1beta1
    Fields Type:  FieldsV1
    fieldsV1:
      f:metadata:
        f:annotations:
          f:platform.confluent.io/last-spec-hash:
        f:finalizers:
          .:
          v:"kafkatopic.finalizers.platform.confluent.io":
        f:ownerReferences:
          .:
          k:{"uid":"86f23b17-6529-4e6f-974d-ea6af118c9a0"}:
    Manager:    manager
    Operation:  Update
    Time:       2022-06-29T06:30:40Z
  Owner References:
    API Version:           platform.confluent.io/v1beta1
    Block Owner Deletion:  true
    Controller:            true
    Kind:                  Kafka
    Name:                  kafka
    UID:                   86f23b17-6529-4e6f-974d-ea6af118c9a0
  Resource Version:        28028
  UID:                     df9489f5-b097-4607-9c0a-c50e97355dcf
Spec:
  Configs:
    cleanup.policy:  delete
  Kafka Rest:
    Authentication:
      Bearer:
        Secret Ref:  rest-credential
      Type:          bearer
  Partition Count:   1
  Replicas:          3
Status:
  Kafka Cluster:        akscfk231/kafka
  Kafka Cluster ID:     b7gKmMeBQK2NEKC139fsog
  Kafka Rest Endpoint:  http://kafka.akscfk231.svc.cluster.local:8090
  Partition Count:      1
  Replicas:             3
  State:                CREATED
Events:                 <none>
```

### Confluent プラグインCLIツールを利用しての確認

```
## Confluent コンポーネントのバージョン確認
$ kubectl confluent version
COMPONENT       NAME            VERSION  OPERATOR-VERSION
Zookeeper       zookeeper       7.1.0    v0.435.23
Kafka           kafka           7.1.0    v0.435.23
Connect         connect         1.0      v0.435.23
SchemaRegistry  schemaregistry  7.1.0    v0.435.23
KsqlDB          ksqldb          7.1.0    v0.435.23
ControlCenter   controlcenter   7.1.0    v0.435.23


## Confluent コンポーネントへのアクセスに使用できるエンドポイントの確認
$ kubectl confluent http-endpoints
COMPONENT       NAME            ACCESS    ADDRESS                                                 AUTH   AUTHORIZATION
Kafka           kafka-rest      INTERNAL  http://kafka.akscfk231.svc.cluster.local:8090                  
Connect         connect         INTERNAL  http://connect.akscfk231.svc.cluster.local:8083                
SchemaRegistry  schemaregistry  INTERNAL  http://schemaregistry.akscfk231.svc.cluster.local:8081         
KsqlDB          ksqldb          INTERNAL  http://ksqldb.akscfk231.svc.cluster.local:8088                 
ControlCenter   controlcenter   INTERNAL  http://controlcenter.akscfk231.svc.cluster.local:9021   basic  
```

---

### Confluent Control Center への接続確認

```
$ kubectl confluent dashboard controlcenter
http://localhost:9021
```

ブラウザから Confluent Contorol Center に接続され、ログイン認証後に作成した Topic「topi002」のステータス = Available を確認します。
![スクリーンショット 2022-06-27 22.07.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/97e847dd-e505-4886-1613-8a5d2778575d.png)


---

## 後処理

### Pod / secret / namespace のアンインストール方法

```
## Pod : confluent-operator
$ helm delete confluent-operator             

## Pod : confluent-platform
$ kubectl delete -f confluent_platform_ccc.yaml
$ kubectl delete -f kafka_rest_class.yaml
$ kubectl delete -f kafka_topic.yaml

## secret情報
$ kubectl delete secret rest-credential

## namespace の削除方法（namespace配下のPodは全て削除される）
$ kubectl delete namespace akscfk231
```

### AKSクラスターの停止・起動

```
$ az aks stop -g rg_ituru_aks01 -n aks_ituru_cp01
$ az aks start -g rg_ituru_aks01 -n aks_ituru_cp01
```

### Check for any error messages in events

```
kubectl get events
```

---

## まとめ

Admin REST API を AKSクラスターの中に作って、Custom Resource からそのAPIを通じて Topicを作成できることを確認できました。

### 参考情報

以下の情報を参考にさせていただきました

[Confluent for Kubernetes](https://docs.confluent.io/ja-jp/operator/2.0.1/overview.html)
[Confluent Admin REST API の管理](https://docs.confluent.io/ja-jp/operator/2.0.1/co-manage-rest-api.html)
[confluent-kubernetes-examples/security/production-secure-deploy/](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/production-secure-deploy)
[Kafka REST API のデプロイのサンプル](https://github.com/confluentinc/confluent-kubernetes-examples/blob/master/security/production-secure-deploy/kafkaRestClass.yaml)
[Topic作成用のサンプル](https://github.com/confluentinc/confluent-kubernetes-examples/blob/master/security/production-secure-deploy/topic.yaml)