Argo WorkflowsとRabbitMQを利用して、Kubernetesでバッチ処理基盤を構築（その1）

# 本記事で行うこと

[前回の記事](https://qiita.com/kanedaq/items/e230ab6b930559535924)を発展させて、Kubernetesでバッチ処理基盤を構築してみます。
その第1弾として、以下の構成図を考えました。
緑の四角はオープンソース、紫の四角は本記事で実装するアプリです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/8dc6afdf-134b-c351-860d-93fa3955f384.png)

## 緑の四角（オープンソース）について

- [CronWorkflow](https://argoproj.github.io/argo-workflows/cron-workflows/)
[Argo Workflows](https://argoproj.github.io/argo-workflows/)が提供するKubernetesリソースの一つで、スケジュール実行を行うワークフローエンジンです。

- [argo-server](https://argoproj.github.io/argo-workflows/argo-server/)（Webアプリ）
Argo Workflowsが提供する、ワークフローを管理するためのUIです。

- [RabbitMQ](https://www.rabbitmq.com/)
メッセージキューイングシステムです。
本記事では、バッチジョブがメッセージをパブリッシュし、それをフロントエンドアプリケーションが受け取るように作ります。
将来は、バッチ処理のTODOをキューに積んで、バッチ処理を駆動する等に利用したいです。

- [RabbitMQ Management UI](https://www.rabbitmq.com/management.html)（Webアプリ）
RabbitMQが提供する、キューイングシステム管理のためのUIです。

## 紫の四角（本記事で実装）について

- batch job
バッチジョブ本体です。
本記事ではRabbitMQにメッセージをパブリッシュするだけの処理を行います。
Pythonで実装しましたが（send.py）、Dockerコンテナで動かすので言語は何でも良いです。

- Frontend Application
本記事では、RabbitMQからメッセージを受け取ってコンソールに表示するだけのPythonアプリ（receive.py）を実装します。
将来はWebフロントエンドアプリにしたいです。

- Admin Server（Web APIサーバー）
処理内容は[前回の記事](https://qiita.com/kanedaq/items/e230ab6b930559535924)と同じで、CronWorkflowのscheduleを確認したり変更したりするAPIを提供します。
Go言語で[client-go](https://github.com/kubernetes/client-go)を使用して実装します。
将来は、Workflowを手動実行するAPIを追加する等、機能を充実させたいです。
また、APIだけでなく、運用担当者向けに管理画面も作りたいです。

## Webアプリの外部公開方法について

本記事では、上記の構成図のポート番号でWebアプリを外部公開します。

- 本番ではhttpsにすべきですが、本記事では全てhttpで統一しました。
- KubernetesでWebアプリを公開する時、[Ingress](https://kubernetes.io/ja/docs/concepts/services-networking/ingress/)リソースを使うことが多いと思います。
しかしIngressの標準機能では、80/443以外のポートでは公開できないそうです。
ただし[このページ](https://saturncloud.io/blog/can-i-set-custom-ports-for-a-kubernetes-ingress-to-listen-on-besides-80-443/)によれば、テクニックを使えば80/443以外のポートで公開できるそうですが、Ingressよりも[Gateway API](https://gateway-api.sigs.k8s.io/)が主流になる可能性を考慮し、本記事では手間を省いて[NodePortタイプのService](https://kubernetes.io/ja/docs/concepts/services-networking/service/#nodeport)を使うに留めます。

# 主な参考サイト（感謝します）

- [rabbitmq-on-kubernetes](https://github.com/latonaio/rabbitmq-on-kubernetes)
- [RabbitMQのチュートリアルを雑に日本語に訳してみた -Hello World-](https://qiita.com/gracefulm/items/be246f5737b6091ca63d)
- [KubernetesのService Accountについて調べてみた](https://qiita.com/knqyf263/items/ecc799650fe247dce9c5)
- [kubernetes – ServiceAccount](https://hyoublog.com/2022/05/02/kubernetes-serviceaccount/)

# 環境

Kubernetesは[kind（Kubernetes in Docker）](https://kind.sigs.k8s.io/)を使用しました。

```
kind version
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/e4b1c654-54de-f5a0-2fa6-efd3ffac4081.png)

# 事前準備

## Argo CLIのインストール

[このページ](https://github.com/argoproj/argo-workflows/releases/)の手順でインストールしました。
本記事では手順は省略します。

# 作業手順

## Kubernetesクラスタ作成

[このページ](https://kind.sigs.k8s.io/docs/user/quick-start/)を参考に、kind用にマニフェストファイルを用意します。
hostPortに、構成図に記載した公開用portを設定します。
containerPortには、後にNodePortタイプのServiceを作るときに、マニフェストで設定する番号を入れます。

```batch-cluster.yaml
kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
nodes:
- role: control-plane
  extraPortMappings:
  - containerPort: 31323
    hostPort: 1323
    listenAddress: "0.0.0.0"
  - containerPort: 32746
    hostPort: 2746
    listenAddress: "0.0.0.0"
  - containerPort: 32094
    hostPort: 5672
    listenAddress: "0.0.0.0"
  - containerPort: 32095
    hostPort: 15672
    listenAddress: "0.0.0.0"
```

kindコマンドでKubernetesクラスタを作成します。
クラスタ名をbatch-clusterにしましたが、何でも良いです。

```shell
kind create cluster --name batch-cluster --config=batch-cluster.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/88c4cc81-35d0-865b-93f6-15d32a10aa35.png)

Kubernetesクラスタを作成できました。

## namespace作成

RabbitMQ用、Argo Workflows用、バッチ処理用のnamespaceをそれぞれ作成します。
バッチ処理用namespace名をfooにしましたが、何でも良いです。

```shell
kubectl create namespace rabbitmq
kubectl create namespace argo
kubectl create namespace foo
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/e09748f6-22a6-dc38-986d-a6f9d2956643.png)

namespaceを作成できました。

## RabbitMQ（その1）：導入～管理UI起動まで

### Deploymentを起動

[このリポジトリ](https://github.com/latonaio/rabbitmq-on-kubernetes)のマニフェストを、そのまま使わせていただきました。感謝します。

```shell
kubectl -n rabbitmq apply -f https://raw.githubusercontent.com/latonaio/rabbitmq-on-kubernetes/main/deployment.yml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/cb077352-da13-2d8f-8e90-cbddc6c5f7d2.png)

起動確認

```shell
kubectl -n rabbitmq get deployment
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/1e3997bc-0963-ab3f-03f1-6da2735e4a19.png)

### RabbitMQ Management UI（port 15672）を起動

外部からブラウザで起動します。
your_hostには、Kubernetesを動かしているホストを指定します。

```
http://your_host:15672
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/67666896-d59a-167f-a5ae-854525c2b8b9.png)

Username、Pasword共に「guest」でログインできます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/44726d9f-73bb-9af2-aaf0-49ccc76a675f.png)

RabbitMQの導入とUI起動が成功しました。

## batch job（その1）：docker buildまで

CronWorkflowの中で実行する各ジョブを開発します。
CronWorkflowは複数の種類のジョブを実行できますが、本記事では一種類のみ作りました。

### ファイル一覧とソースコード

```
src directory
  ├── send.py
  ├── requirements.txt
  └── Dockerfile
```

#### send.py

[pika](https://github.com/pika/pika)ライブラリを使用して、RabbitMQにメッセージをパブリッシュします。
3つのプログラム引数を渡す必要があります。
- host：RabbitMQのホスト名
- queue：キュー名
- message：メッセージ

```send.py
import json
import sys
import traceback
import pika


def main():
    host = sys.argv[1]
    queue = sys.argv[2]
    message = sys.argv[3]

    param = pika.ConnectionParameters(host=host)
    connection = pika.BlockingConnection(param)
    channel = connection.channel()
    channel.queue_declare(queue=queue)

    channel.basic_publish(exchange="", routing_key=queue, body=json.dumps({"message": message}))
    connection.close()


if __name__ == "__main__":
    try:
        main()
    except Exception:
        traceback.print_exc()
```

#### requirements.txt

```requirements.txt
pika
```

#### Dockerfile

```Dockerfile
FROM python:3-slim-bullseye

RUN set -e; \
    apt-get update -y && apt-get install -y \
    lsb-release; \
    apt-get clean

RUN pip install --upgrade pip
RUN pip install --upgrade setuptools
ENV PYTHONUNBUFFERED 1

ENV APP_HOME /app
WORKDIR $APP_HOME
COPY ./send.py ./
COPY ./requirements.txt ./

RUN pip install -r requirements.txt

ENV PORT 5672
ENTRYPOINT ["python", "send.py"]
CMD ["dummy"]
```

### Dockerビルド

```shell
docker build ./ -t batch-job
```

### Dockerイメージをkindに登録

```shell
kind load docker-image batch-job --name=batch-cluster
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/04ac36c5-a962-6ee0-1d95-089c72f70529.png)

batch jobの開発＆docker buildまでできました。

## RabbitMQ（その2）＆ batch job（その2）：RabbitMQとbatch jobを同時に動作確認

### パブリッシュ側（send.py）の動作確認

kubectl debugコマンドを使って動作確認を行います。
[記事執筆時点のマニフェスト](https://raw.githubusercontent.com/latonaio/rabbitmq-on-kubernetes/main/deployment.yml)を使った場合は、以下のコマンドでデバッグ作業に入れました。
マニフェストが更新されたら、コマンドを調整する必要があるかもしれません。

```shell
kubectl -n rabbitmq debug \
    $(kubectl -n rabbitmq get --no-headers=true pods -l run=rabbitmq -o custom-columns=:metadata.name) \
    --image=batch-job \
    --image-pull-policy=Never \
    -it \
    -- sh
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/4754d62d-be19-5736-a2f9-6ac57252f659.png)

デバッグに入ってのシェルで、send.pyを実行します。

```shell
python send.py rabbitmq.rabbitmq foo-job test-message
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/0fbf35ce-4e9d-1221-dedd-677ea14b7b3f.png)

RabbitMQのUIを見て、キューにメッセージが溜まれば、ひとまずOKです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/db7cc40f-8835-67e4-2848-988665697651.png)

### サブスクライブ側（receive.py）の開発と動作確認

外部のマシンで以下のコードを用意します。
[pika](https://github.com/pika/pika)ライブラリを使用して、RabbitMQからメッセージを受信します。
2つのプログラム引数を渡す必要があります。
- host：RabbitMQのホスト名
- queue：キュー名

```receive.py
import sys
import traceback
import pika


def main():
    host = sys.argv[1]
    queue = sys.argv[2]

    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)

    param = pika.ConnectionParameters(host=host)
    connection = pika.BlockingConnection(param)
    channel = connection.channel()
    channel.queue_declare(queue=queue)

    channel.basic_consume(queue=queue, on_message_callback=callback, auto_ack=True)
    channel.start_consuming()


if __name__ == "__main__":
    try:
        main()
    except Exception:
        traceback.print_exc()
```

pip等でpikaライブラリをインストール後、以下を実行します。
your_hostには、Kubernetesを動かしているホストを指定します。

```shell
python receive.py your_host foo-job
```

RabbitMQのキューに溜まっていたテストメッセージが受信されればOKです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/7b97af30-7c7b-18af-85ba-d9cc09400123.png)

管理UIでも、キューのメッセージが0件になったことを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/3ae93424-ab03-4564-dc8b-b7958e26c204.png)

RabbitMQとbatch jobの動作確認ができました。
receive.pyを終了させ、Kerbernetes側のホストでOSのシェルに戻っておきます。

## Argo Workflowsのインストールとargo-serverのUI起動

### インストール

[このページ](https://argoproj.github.io/argo-workflows/quick-start/)の手順でインストールします。
本記事執筆時点で、Argo Workflowsの最新バージョンはv3.4.11でした。

```shell
kubectl -n argo apply -f https://github.com/argoproj/argo-workflows/releases/download/v3.4.11/install.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/c19c0c03-bca8-734d-b257-a31e64139d2a.png)

### argo-serverの設定変更

本記事ではhttpsでなくhttpに統一しますので、kubectl patchコマンドを使って、argo-serverの設定をそのように変更します。

```shell
kubectl patch deployment \
  argo-server \
  --namespace argo \
  --type='json' \
  -p='[
          {
              "op": "replace",
              "path": "/spec/template/spec/containers/0/readinessProbe/httpGet/scheme",
              "value": "HTTP"
          },
          {
              "op": "replace",
              "path": "/spec/template/spec/containers/0/args",
              "value": [
                  "server",
                  "--auth-mode=server",
                  "--secure=false"
              ]
          }
      ]'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/f3aac6ea-e7ee-0bd0-4bf8-a73cdf370795.png)

変更内容確認

```
kubectl -n argo get deployment argo-server -o yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/752d34a3-7123-6374-90e5-621ddf94e3f2.png)

### argo-server用のServiceを起動

以下のマニフェストファイルを用意します。
記事冒頭で書きましたように、NodePortタイプのServiceで外部公開します。

```argo-server-nodeport.yaml
apiVersion: v1
kind: Service
metadata:
  name: argo-server-nodeport
  namespace: argo
spec:
  type: NodePort
  ports:
  - name: web
    port: 2746
    targetPort: 2746
    nodePort: 32746
  selector:
    app: argo-server
```

Serviceを起動します。

```shell
kubectl apply -f argo-server-nodeport.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/b73d8eda-c662-78d2-5f53-b989c226f715.png)

起動確認

```shell
kubectl -n argo get svc
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/1cee9644-9b90-2af0-b6bf-6f638b8fc053.png)

### argo-server（port 2746）を起動

外部からブラウザで起動します。
your_hostには、Kubernetesを動かしているホストを指定します。

```
http://your_host:2746
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/101d0d48-143c-8cc0-c503-2acc4d3b96d2.png)

argo-serverのUIを起動できました。

## Admin Server（Go言語）

※ [この記事](https://qiita.com/kanedaq/items/e230ab6b930559535924)と内容がダブります。

[client-go](https://github.com/kubernetes/client-go)を使用して、CronJobやCronWorkflowのscheduleを変更するAPIを開発します。
[こちらのソースコード](https://github.com/kubernetes/client-go/tree/master/examples/in-cluster-client-configuration)を参考にしました。

### ファイル一覧とソースコード

```
src directory
  ├── admin_server.go
  ├── go.mod
  └── Dockerfile
```

#### admin_server.go

Webフレームワークは[Echo](https://echo.labstack.com/)を使用しました。
起動時にプログラム引数として、namespace名を1つ渡す必要があります。
APIは以下の4つです。

- GET /&lt;&lt;namespace&gt;&gt;/cronjobs
namespaceに属するCronJobの一覧を取得します。

- PUT /&lt;&lt;namespace&gt;&gt;/cronjobs
CronJobのscheduleをupdateします。

- GET /&lt;&lt;namespace&gt;/cronworkflows
namespaceに属するCronWorkflowの一覧を取得します。

- PUT /&lt;&lt;namespace&gt;&gt;/cronworkflows
CronWorkflowのscheduleをupdateします。

※ 本記事ではCronJobは使用しません。

```admin_server.go
package main

import (
	"context"
	argoclientset "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned"
	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
	"net/http"
	"os"
	"path/filepath"
)

type httpError struct {
	Error string `json:"error"`
}

var (
	namespace     string
	clientset     *kubernetes.Clientset
	argoClientset *argoclientset.Clientset
)

func getCronjobs(c echo.Context) error {
	// レスポンス初期化
	type cronjob struct {
		Name     string `json:"name"`
		Schedule string `json:"schedule"`
	}
	type responseType struct {
		Data []cronjob `json:"data"`
	}
	var response responseType
	response.Data = []cronjob{}

	// CronJobリスト取得
	cjobs, err := clientset.BatchV1().CronJobs(namespace).List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		return c.JSON(http.StatusInternalServerError, httpError{Error: err.Error()})
	}

	// 各CronJobをレスポンスに追加
	for _, cjob := range cjobs.Items {
		cj := cronjob{
			Name:     cjob.Name,
			Schedule: cjob.Spec.Schedule,
		}
		response.Data = append(response.Data, cj)
	}

	return c.JSON(http.StatusOK, response)
}

func updateCronjob(c echo.Context) error {
	// リクエストボディ＆レスポンスの型
	type cronjob struct {
		Name     string `json:"name"`
		Schedule string `json:"schedule"`
	}

	// リクエストボディを構造体にバインド
	reqb := cronjob{}
	if err := c.Bind(&reqb); err != nil {
		return c.JSON(http.StatusBadRequest, httpError{Error: err.Error()})
	}

	cjClient := clientset.BatchV1().CronJobs(namespace)

	// CronJob取得
	cj, err := cjClient.Get(context.TODO(), reqb.Name, metav1.GetOptions{})
	if err != nil {
		return c.JSON(http.StatusBadRequest, httpError{Error: err.Error()})
	}

	// スケジュール変更
	cj.Spec.Schedule = reqb.Schedule

	// 更新適用
	resultCj, err := cjClient.Update(context.TODO(), cj, metav1.UpdateOptions{})
	if err != nil {
		return c.JSON(http.StatusInternalServerError, httpError{Error: err.Error()})
	}

	// 更新された内容をレスポンスにする
	response := cronjob{
		Name:     resultCj.Name,
		Schedule: resultCj.Spec.Schedule,
	}
	return c.JSON(http.StatusOK, response)
}

func getCronworkflows(c echo.Context) error {
	// レスポンス初期化
	type cronworkflow struct {
		Name     string `json:"name"`
		Schedule string `json:"schedule"`
	}
	type responseType struct {
		Data []cronworkflow `json:"data"`
	}
	var response responseType
	response.Data = []cronworkflow{}

	// CronWorkflowリスト取得
	cworkflows, err := argoClientset.ArgoprojV1alpha1().CronWorkflows(namespace).List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		return c.JSON(http.StatusInternalServerError, httpError{Error: err.Error()})
	}

	// 各CronWorkflowをレスポンスに追加
	for _, cworkflow := range cworkflows.Items {
		cw := cronworkflow{
			Name:     cworkflow.Name,
			Schedule: cworkflow.Spec.Schedule,
		}
		response.Data = append(response.Data, cw)
	}

	return c.JSON(http.StatusOK, response)
}

func updateCronworkflow(c echo.Context) error {
	// リクエストボディ＆レスポンスの型
	type cronworkflow struct {
		Name     string `json:"name"`
		Schedule string `json:"schedule"`
	}

	// リクエストボディを構造体にバインド
	reqb := cronworkflow{}
	if err := c.Bind(&reqb); err != nil {
		return c.JSON(http.StatusBadRequest, httpError{Error: err.Error()})
	}

	cwClient := argoClientset.ArgoprojV1alpha1().CronWorkflows(namespace)

	// CronWorkflow取得
	cw, err := cwClient.Get(context.TODO(), reqb.Name, metav1.GetOptions{})
	if err != nil {
		return c.JSON(http.StatusBadRequest, httpError{Error: err.Error()})
	}

	// スケジュール変更
	cw.Spec.Schedule = reqb.Schedule

	// 更新適用
	resultCw, err := cwClient.Update(context.TODO(), cw, metav1.UpdateOptions{})
	if err != nil {
		return c.JSON(http.StatusInternalServerError, httpError{Error: err.Error()})
	}

	// 更新された内容をレスポンスにする
	response := cronworkflow{
		Name:     resultCw.Name,
		Schedule: resultCw.Spec.Schedule,
	}
	return c.JSON(http.StatusOK, response)
}

// 開発用。本番コードでは不要
func outClusterConfig() (*rest.Config, error) {
	// Create client
	kubeconfig, ok := os.LookupEnv("KUBECONFIG")
	if !ok {
		kubeconfig = filepath.Join(homedir.HomeDir(), ".kube", "config")
	}
	return clientcmd.BuildConfigFromFlags("", kubeconfig)
}

func main() {
	// Program argsでKubernetesのnamespaceを指定
	if len(os.Args) != 2 {
		panic("assert len(os.Args) == 2")
	}
	namespace = os.Args[1]

	// creates the in-cluster config
	config, err := rest.InClusterConfig()
	if err != nil {
		config, err = outClusterConfig() // 開発用。本番コードでは消す
		if err != nil {
			panic(err.Error())
		}
	}

	// creates the clientset and argoClientset
	clientset, err = kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}
	argoClientset, err = argoclientset.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}

	// EchoでAPIサーバー
	e := echo.New()
	e.Use(middleware.Logger())
	e.Use(middleware.Recover())

	// Routing
	api := e.Group("/" + namespace)
	api.GET("/cronjobs", getCronjobs)
	api.PUT("/cronjobs", updateCronjob)
	api.GET("/cronworkflows", getCronworkflows)
	api.PUT("/cronworkflows", updateCronworkflow)

	e.Logger.Fatal(e.Start(":1323"))
}
```

#### go.mod

```go.mod
module admin_server

go 1.20

require (
	github.com/argoproj/argo-workflows/v3 v3.4.11
	github.com/labstack/echo/v4 v4.11.1
	k8s.io/apimachinery v0.28.2
	k8s.io/client-go v0.28.2
)

require (
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/emicklei/go-restful/v3 v3.10.0 // indirect
	github.com/go-logr/logr v1.2.4 // indirect
	github.com/go-openapi/jsonpointer v0.19.6 // indirect
	github.com/go-openapi/jsonreference v0.20.2 // indirect
	github.com/go-openapi/swag v0.22.3 // indirect
	github.com/gogo/protobuf v1.3.2 // indirect
	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
	github.com/golang/protobuf v1.5.3 // indirect
	github.com/google/gnostic-models v0.6.8 // indirect
	github.com/google/go-cmp v0.5.9 // indirect
	github.com/google/gofuzz v1.2.0 // indirect
	github.com/google/uuid v1.3.0 // indirect
	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
	github.com/imdario/mergo v0.3.13 // indirect
	github.com/josharian/intern v1.0.0 // indirect
	github.com/json-iterator/go v1.1.12 // indirect
	github.com/labstack/gommon v0.4.0 // indirect
	github.com/mailru/easyjson v0.7.7 // indirect
	github.com/mattn/go-colorable v0.1.13 // indirect
	github.com/mattn/go-isatty v0.0.19 // indirect
	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
	github.com/modern-go/reflect2 v1.0.2 // indirect
	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
	github.com/sirupsen/logrus v1.9.3 // indirect
	github.com/spf13/pflag v1.0.5 // indirect
	github.com/valyala/bytebufferpool v1.0.0 // indirect
	github.com/valyala/fasttemplate v1.2.2 // indirect
	golang.org/x/crypto v0.12.0 // indirect
	golang.org/x/net v0.14.0 // indirect
	golang.org/x/oauth2 v0.11.0 // indirect
	golang.org/x/sys v0.11.0 // indirect
	golang.org/x/term v0.11.0 // indirect
	golang.org/x/text v0.12.0 // indirect
	golang.org/x/time v0.3.0 // indirect
	google.golang.org/appengine v1.6.7 // indirect
	google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130 // indirect
	google.golang.org/genproto/googleapis/api v0.0.0-20230706204954-ccb25ca9f130 // indirect
	google.golang.org/genproto/googleapis/rpc v0.0.0-20230720185612-659f7aaaa771 // indirect
	google.golang.org/grpc v1.56.2 // indirect
	google.golang.org/protobuf v1.31.0 // indirect
	gopkg.in/inf.v0 v0.9.1 // indirect
	gopkg.in/yaml.v2 v2.4.0 // indirect
	gopkg.in/yaml.v3 v3.0.1 // indirect
	k8s.io/api v0.28.2 // indirect
	k8s.io/klog/v2 v2.100.1 // indirect
	k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9 // indirect
	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 // indirect
	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
	sigs.k8s.io/yaml v1.3.0 // indirect
)
```

#### Dockerfile

```Dockerfile
FROM golang:bullseye as builder
WORKDIR /app
COPY ./admin_server.go ./
COPY ./go.mod ./
RUN go mod tidy
RUN go build -o admin_server admin_server.go

FROM debian:bullseye
WORKDIR /app
COPY --from=builder /app/admin_server /app/
ENTRYPOINT ["/app/admin_server"]
```

### Dockerビルド

```shell
docker build ./ -t admin-server
```

### Dockerイメージをkindに登録

```shell
kind load docker-image admin-server --name=batch-cluster
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/21cd9c2c-66ca-92dd-66e1-6f977a682c13.png)

### Admin Server用サービスアカウントの作成

Admin ServerをPodで動かすために必要な、サービスアカウントを作成します。
以下のマニフェストファイルを用意します。

```admin-sa.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: admin-sa
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: admin-role
rules:
  - apiGroups:
      - batch
    verbs:
      - list
      - get
      - update
    resources:
      - cronjobs
  - apiGroups:
      - argoproj.io
    verbs:
      - list
      - get
      - update
    resources:
      - cronworkflows
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: admin-role-binding
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: admin-role
subjects:
  - kind: ServiceAccount
    name: admin-sa
```

サービスアカウントを作成します。

```shell
kubectl -n foo apply -f admin-sa.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/569900ac-1044-371e-222f-8b552ae7f5a1.png)

### Admin Server用Serviceの作成

以下のマニフェストファイルを用意します。
- 先ほど作成したサービスアカウント「admin-sa」を指定します。
- Admin Server起動時のプログラム引数として、namespace名（本記事では"foo"）を渡すように設定します。
- 記事冒頭で書きましたように、NodePortタイプのServiceで外部公開します。

```foo-admin-nodeport.yaml
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: foo-admin-deployment
  namespace: foo
  labels:
    app: foo-admin-server
spec:
  replicas: 1
  selector:
    matchLabels:
      app: foo-admin-server
  template:
    metadata:
      labels:
        app: foo-admin-server
    spec:
      serviceAccountName: admin-sa
      containers:
      - name: foo-admin-container
        image: admin-server:latest
        imagePullPolicy: Never
        args: ["foo"]
---
apiVersion: v1
kind: Service
metadata:
  name: foo-admin-nodeport
  namespace: foo
  labels:
    app: foo-admin-server
spec:
  type: NodePort
  ports:
  - port: 1323
    targetPort: 1323
    nodePort: 31323
  selector:
    app: foo-admin-server
```

Serviceを起動します。

```shell
kubectl apply -f foo-admin-nodeport.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/67e3dda7-c6ef-bba3-59e3-66348cd76186.png)

起動確認

```shell
kubectl -n foo get pod
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/4aebd504-a50d-1489-3078-33b944baf8d9.png)

### Serviceの動作確認

ここで一旦、Admin ServerのAPIを動作確認しておきます。

#### 動作確認用CronWorkflowを起動

公式サイトのexamplesのマニフェストファイルをそのまま使用しました。

```shell
kubectl -n foo apply -f https://raw.githubusercontent.com/argoproj/argo-workflows/master/examples/cron-workflow.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/e9101071-95f7-b721-a6fd-17cd7f97da7d.png)

起動確認

```shell
argo -n foo cron list
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/4a365d99-dcef-988e-fd42-7a7a23f41417.png)

CronWorkflow名がhello-worldであることを覚えておきます。

#### APIを呼んでみる

外から2つのAPIを呼んでみます。
your_hostには、Kubernetesを動かしているホストを指定します。

```shell
curl http://your_host:1323/foo/cronworkflows
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/5cda2aeb-323b-598c-01f9-aeeab2a20338.png)

```shell
curl --location --request PUT 'http://your_host:1323/foo/cronworkflows' --header 'Content-Type: application/json' --data '{"name": "hello-world", "schedule": "1 0 * * *"}'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/3e4043df-725f-6b49-8a0a-859f5a81d5fa.png)

APIが正常に動くことが確認できました。

#### 動作確認用CronWorkflowを削除

kubectl applyコマンドで作ったCronWorkflowを、kubectl deleteコマンドで削除します。

```shell
kubectl -n foo delete -f https://raw.githubusercontent.com/argoproj/argo-workflows/master/examples/cron-workflow.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/33818c9a-c32c-798b-c537-5ab2b6dc2243.png)

別の方法として、argo-serverのUIから削除することもできます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/4df461dd-a7d1-4708-c393-574c4d2f2373.png)

Admin Serverの一旦の動作確認まで完了しました。

## CronWorkflow

CronWorkflowを起動する前に、あらかじめreceive.pyを外部で起動しておきます。
your_hostには、Kubernetesを動かしているホストを指定します。

```shell
python receive.py your_host foo-job
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/840b2569-6142-74ae-b3c3-4a46a8733bfb.png)

次に、1分毎にスケジュール実行を行うCronWorkflowを作成します。
以下のマニフェストファイルを用意します。
既に構築したbatch-jobを2回、直列に実行します。

```foo-cronworkflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: foo-cronworkflow
  namespace: foo
spec:
  schedule: "* * * * *"
  timezone: "Asia/Tokyo"
  concurrencyPolicy: "Forbid"
  startingDeadlineSeconds: 0
  workflowSpec:
    entrypoint: foo-entrypoint
    onExit: finally-container
    templates:
    - name: foo-entrypoint
      dag:
        tasks:
        - name: job1
          template: job1-container
        - name: job2
          depends: job1
          template: job2-container
    - name: job1-container
      container:
        image: batch-job
        imagePullPolicy: Never
        command: ["python", "send.py"]
        args: ["rabbitmq.rabbitmq", "foo-job", "job1"]
    - name: job2-container
      container:
        image: batch-job
        imagePullPolicy: Never
        command: ["python", "send.py"]
        args: ["rabbitmq.rabbitmq", "foo-job", "job2"]
    - name: finally-container
      container:
        image: batch-job
        imagePullPolicy: Never
        command: ["sh", "-c"]
        args:
        - |
          echo finally
```

CronWorkflowを起動します。

```shell
kubectl apply -f foo-cronworkflow.yaml
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/2c82b723-93bb-6434-c074-53f4d2ea43a6.png)

起動確認

```shell
argo -n foo cron list
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/7ba27ffa-8eb3-4d02-d3b6-a16829750b34.png)

事前に起動したreceive.py側を見ると、1分毎にワークフローからのメッセージが受信され続けました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/99a745dc-2caa-a353-212e-ba4974c92a11.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/9ae5d217-944b-5fb1-f161-a7b6352c5ebe.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/a48a4c84-9b3f-df06-437b-e98d6dbf18cd.png)

argo-serverのUIで、ワークフローをリアルタイムに確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/eb2f8bfb-b88e-f96f-35e7-caf5f9036a70.png)

CronWorkflowを動作確認できました。

## Admin Serverで、起動中のCronWorkflowのscheduleを操作する

外部からAdmin ServerのAPIを呼んで、CronWorkflowのscheduleをリアルタイムに変更してみます。
your_hostには、Kubernetesを動かしているホストを指定します。


まずGET APIで現在のスケジュールを確認すると、1分毎の実行になっています。

```shell
curl http://your_host:1323/foo/cronworkflows
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/a699faaf-9951-771e-b789-75b9927311c7.png)

次にPUT APIでスケジュールを毎日0時0分に変更します。

```shell
curl --location --request PUT 'http://your_host:1323/foo/cronworkflows' --header 'Content-Type: application/json' --data '{"name": "foo-cronworkflow", "schedule": "0 0 * * *"}'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/e2ab9a25-bbb4-8580-0602-a578bcffb65f.png)

receive.py側を見ると、メッセージの受信が止まっていました。
0時0分までこのまま動かないはずです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/f95bb33e-0d73-85f3-9d1a-cc6d241bdcb4.png)

Admin ServerのAPIで、CronWorkflowのscheduleを変更できました。

## 後始末

Kubernetesクラスタを削除するコマンドです。

```shell
kind delete clusters batch-cluster
````