概要
Confluent for Kubernetes (CFK)は、プライベートクラウド環境(今回は Azure Kubernetes Service(AKS))に Confluent をデプロイして管理するためのクラウドネイティブのコントロールプレーンです。宣言型 API で Confluent Platform をカスタマイズ、デプロイ、管理するための標準的で簡素なインターフェイスが備わっています。
CFK を使用して AKS上にデプロイされている Confluent Platform に Topic を作成するためのワークフローの概要は以下のとおりです。
- Kubernetes 環境を準備します(事前準備で完了)
- Confluent for Kubernetes をデプロイします(事前準備で完了)
- Confluent Platform をデプロイします(事前準備で完了)
- Confluent Platform の追加構成をします
- Confluent Platform の追加デプロイ(Topic作成)をします
ローカル環境
- macOS Monterey 12.3.1
- python 3.8.12
- Azure CLI 2.34.1
- helm v3.6.3
- kubectl v1.21.3
事前準備
Confluent Platform の追加構成
Bearer authentication secret の作成(Bearer認証)
Topic作成のための Kafka Rest Class 利用時の認証用 ID/Password の定義ファイルを作成します
username=kafka
password=kafka-secret
Kafka Rest Class(Admin Rest API) の yaml ファイルの作成
apiVersion: platform.confluent.io/v1beta1
kind: KafkaRestClass
metadata:
name: default
namespace: akscfk231
spec:
kafkaRest:
authentication:
type: bearer
bearer:
secretRef: rest-credential
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 を確認します。
後処理
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
Confluent Admin REST API の管理
confluent-kubernetes-examples/security/production-secure-deploy/
Kafka REST API のデプロイのサンプル
Topic作成用のサンプル