【KEDA入門】イベント駆動でKafka Streamsをオートスケールさせる

# KEDAとは

Kubernetesクラスタ上で動作するOSSのイベント駆動オートスケーラー(**K**ubernetes-based **E**vent **D**riven **A**utoscaler = `KEDA`)のことであり、外部システムにおける特定メトリクスを閾値によって監視し、スケーリング対象アプリケーション(Kubernetes Deploymentなど)のPod数を自動で調整します。

メトリクス監視はKEDA本体によって実施されますが、スケーリング対象アプリケーションのPod数の自動調整は`Horizontal Pod Autoscaler` (以降、HPAと記載) と呼ばれるKubernetesの標準リソースによって実施されます。

また、監視対象の接続先情報やメトリクス閾値、Pod数の調整範囲などの設定項目は、`ScaledObject`と呼ばれるKEDA独自のリソースに定義します。

各リソースとシステム間の関係性は、以下のようにまとめられます。

```mermaid
graph TD;
KEDA([KEDA])
SO[ScaledObject]
HPA[Horizontal Pod Autoscaler]
ES(External System)
AP(Target App)
SO-.define.->KEDA
SO-.define.->HPA
KEDA--monitoring-->ES
KEDA--notify-->HPA
HPA--scaling-->AP
```

# スケーリング対象アプリケーション

今回は、[事前に作成したKafkaStreamsアプリケーション](https://github.com/ogi-iii/kafka-streams-pipeline)を対象にKEDAでオートスケールさせます。

アプリケーションの処理内容は、以下のようなETL処理となります。

1. Apache Kafkaから処理対象のデータを取得する。(Extract: 抽出)
2. データから必要な情報のみを選び、DBに挿入可能な形式に加工する。(Transform: 変換)
3. Mongo DBに対して加工したデータを挿入する。(Load: 格納)

```mermaid
graph LR;
    AK([Apache Kafka])
    subgraph Transform
    PA(Streams Pipeline App)
    end
    MD[(Mongo DB)]
    AK--Extract-->PA
    PA--Load-->MD
```

※アプリケーションの詳細については、[ソースコード](https://github.com/ogi-iii/kafka-streams-pipeline)を参照ください。

# KEDAによるオートスケーリングの実装

それでは、実際にKEDAを動かしてアプリケーションがオートスケーリングされる様子を観察してみます。

※実際に動作可能なソースコード一式は、[こちらのリポジトリ](https://github.com/ogi-iii/keda-auto-scalable-pipeline)に格納しています。

## 動作環境と前提条件

シェルスクリプトを利用するため、MacまたはLinux(及びWSL)での動作を前提としています。

また、以下のツールが必要となります。

- Git
- Docker
- Kubernetes (クラスタ1つとkubectlコマンド)
- Helm (バージョン3)

## 1. 関連システムの事前セットアップ

Kafka Streamsアプリケーションで利用するApache Kafka及びMongo DBを予めセットアップしておきます。

```bash
if [ ! -d kafka-streams-pipeline/ ]; then
    git clone https://github.com/ogi-iii/kafka-streams-pipeline.git
fi
cd kafka-streams-pipeline/
git pull

docker compose -f ./demo/docker-compose.yml up -d
./demo/scripts/create-topic.sh

cd ..
```

## 2. スケーリング対象アプリケーションの事前デプロイ

KEDAのスケーリング対象となるKafka Streamsアプリケーションを予めデプロイしておきます。

```bash
if [ ! -d keda-auto-scalable-pipeline/ ]; then
    git clone https://github.com/ogi-iii/keda-auto-scalable-pipeline.git
fi
cd keda-auto-scalable-pipeline/
git pull

docker image build -t streams-pipeline-app .
kubectl create namespace pipeline
kubectl apply -f kubernetes/pipeline-deployment.yaml

cd ..
```

## 3. KEDAのセットアップ

Helmを利用し、KEDAをインストールします。

```bash
helm repo add kedacore https://kedacore.github.io/charts
helm repo update
```

利用するkubernetesクラスタにKEDA専用のnamespaceを作成し、KEDAをkubernetesクラスタへインストールします。

```bash
kubectl create namespace keda
helm install keda kedacore/keda -n keda
```

## 4. ScaledObjectのデプロイ

KEDA独自のScaledObjectリソースをyamlファイルとして以下のように定義します。

```yaml:scaled-object.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: streams-pipeline-app-scaler
  namespace: pipeline
spec:
  scaleTargetRef:
    apiVersion: apps/v1                # Optional. Default: apps/v1
    kind:       Deployment             # Optional. Default: Deployment
    name:       streams-pipeline-app   # Mandatory. Must be in the same namespace as the ScaledObject
  pollingInterval:  5                  # Optional. Default: 30 seconds
  cooldownPeriod:   10                 # Optional. Default: 300 seconds
  idleReplicaCount: 0                  # Optional. Must be less than minReplicaCount
  minReplicaCount:  1                  # Optional. Default: 1
  maxReplicaCount:  5                  # Optional. Default: 100 (up to the number of partitions in Kafka Brokers)
  triggers:
  - type: kafka
    metadata:
      bootstrapServers: host.docker.internal:29092
      consumerGroup: streams-pipeline-app
      topic: pizzaOrders
      lagThreshold: '10'
      offsetResetPolicy: latest
      allowIdleConsumers: "false"
      version: 1.0.0
```

作成したyamlファイルは以下のようにデプロイさせます。

```bash
kubectl apply -f scaled-object.yaml
```

※メトリクス監視対象のApache Kafkaとうまく接続できなかった場合は、一度ScaledObjectリソースを削除し、再デプロイしてみてください。

```bash
# リソースの削除
kubectl delete -f ./kubernetes/scaled-object.yaml

# リソースの再デプロイ
kubectl apply -f scaled-object.yaml
```

## オートスケーリングの動作確認

Kafka StreamsアプリケーションのPod数を事前に確認します。

```bash
kubectl get pods -n pipeline
# 【出力結果の例】
# NAME                                    READY   STATUS    RESTARTS   AGE
# streams-pipeline-app-aaaaaaa-aaaaaaaa   1/1     Running   0          115s
```

Apache Kafkaに検証用のソースデータを流し込みます。

```bash
if [ ! -d kafka-streams-pipeline/ ]; then
    git clone https://github.com/ogi-iii/kafka-streams-pipeline.git
fi
cd kafka-streams-pipeline/
git pull

./demo/scripts/create-datagen.sh

cd ..
```

ソースデータの流入後、しばらくしてからKafka StreamsアプリケーションのPod数を再度表示すると、Pod数が増加しスケールアウトされていることが確認できます。

```bash
kubectl get pods -n pipeline
# 【出力結果の例】
# NAME                                    READY   STATUS    RESTARTS   AGE
# streams-pipeline-app-ddddddd-dddddddd   1/1     Running   0          40s
# streams-pipeline-app-ccccccc-cccccccc   1/1     Running   0          40s
# streams-pipeline-app-bbbbbbb-bbbbbbbb   1/1     Running   0          40s
# streams-pipeline-app-aaaaaaa-aaaaaaaa   1/1     Running   0          6m23s
```

# 参考リンク
- [Prometheusメトリクスに基づいて Kubernetes Pod をオートスケーリングさせてみた - QG Tech Blog](https://tech.quickguard.jp/posts/pod-autoscaling-using-prometheus/)
- [ContainerSolutions/keda-kafka](https://github.com/ContainerSolutions/keda-kafka)
- [ogi-iii/keda-auto-scalable-pipeline](https://github.com/ogi-iii/keda-auto-scalable-pipeline)
- [ogi-iii/kafka-streams-pipeline](https://github.com/ogi-iii/kafka-streams-pipeline)