こちらはあくあたん工房GWアドベントカレンダー5月5日の記事です．

Workflow Engine とは

Job1があるサービスからデータを取得する，Job2がJob1で取得したデータをもとに利用可能なかたち例えばある数式に基づいた評価スコアに変換する，Job3がJob2で算出されたスコアでランキングを生成する...といった一連のJobの流れがあるとします．
このとき，当然ながらJob2はJob1が終わった後に実行しないと意味がないですし，同様にJob3はJob2が終わった後に実行しないといけません．

このようなJobの依存関係の解決方法として，例えばKubernetesではCronJobのスケジュールをうまく設定することが考えられます．しかし，この方法では一つ一つのJobの実行時間を把握しておかなければいけません．
またJobの数が数十個単位やそれ以上になったとき，そこに新たなJobを追加しようとするのは，もはや職人芸の域と言えるでしょう．

このような問題を解決してくれるのがWorkflow Engineです．
一連のJobの流れをWorkflowといい，Workflow EngineはこのようなWorkflowの実行を管理したり，それぞれのJobのログをとってくれたり等を担ってくれます（なお，Workflow Engineとして持つべき機能は厳密には定義されていないためどのような機能を提供しているかは各ツールによります）．

Workflow Engineの一覧は，こちらにまとめられているため目を通してみると良いかもしれません．

Argo とは

Argo¹はArgoprojに所属するツールの一つで，KubernetesをプラットフォームとするWorkflow Engineです．
カスタムリソースとしてWorkflowリソースを定義し，マニフェストファイルをapplyすることで各Workflowを実行します．

Argoの生い立ちについては公式ブログにまとめられています．

Argoを触ってみる

環境

minikube v1.0.0
argo v2.2.1

インストール

公式ドキュメントを参考に進めます．

argoコマンド

Argoにおけるworkflowリソースの管理はkubectlをラップしたargoコマンドを使います．

# Mac
$ brew install argoproj/tap/argo

# Linux
$ curl -sSL -o /usr/local/bin/argo https://github.com/argoproj/argo/releases/download/v2.2.1/argo-linux-amd64
$ chmod +x /usr/local/bin/argo

シェルを再読込してargoと入力し，コマンドの説明が表示されればインストール成功です．

CRD/Workflow

CRDとしてworkflowを登録し，必要なコントローラ等をデプロイします．

$ kubectl create ns argo
$ kubectl apply -n argo -f https://raw.githubusercontent.com/argoproj/argo/v2.2.1/manifests/install.yaml

kubectl api-resourcesで確認してみます．workflowリソースの略称はwfのようです．

$ kubectl api-resources | grep argo
workflows       wf       argoproj.io        true         Workflow

実行してみる

公式ドキュメント内のexampleを試してみようと思います．

Steps

"-"の数で実行順序を定義します．
マニフェストファイルを見てみましょう．

steps.yml

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: steps-    # サフィックスとして5文字の英数字が割り当てられるため，判別しやすいようにうしろに"-"をつけている
spec:
  entrypoint: hello-hello-hello    # このentorypointに記述されているtemplateが実行されます

  # spec内では"hello-hello-hello"と"whalesay"の二つのtemplateが定義されています．
  templates:
  - name: hello-hello-hello
    steps:    # Stepsで実行することを明記
    - - name: hello1            # hello1が最初に実行されるstepです．
        template: whalesay      # このstepに到達すると，template.whalesayが実行されます
        arguments:
          parameters:
          - name: message
            value: "hello1"
    - - name: hello2a           # "- -" => 前のstepが実行し終わると実行開始されます．
        template: whalesay
        arguments:
          parameters:
          - name: message
            value: "hello2a"
      - name: hello2b           # "-" => 前のstepと並列して実行されます．
        template: whalesay
        arguments:
          parameters:
          - name: message
            value: "hello2b"

  - name: whalesay
    inputs:
      parameters:
      - name: message
    container:
      image: docker/whalesay
      command: [cowsay]
      args: ["{{inputs.parameters.message}}"]

workflowを定義する部分と，各step内のJobTemplateが同列に記述されているのは面白いですね．
実行する際はargo submitを使います．

$ argo submit steps.yml --watch    # watchオプションをつけると実行状況の監視ができる
Name:                steps-rndzz
Namespace:           argo
ServiceAccount:      default
Status:              Succeeded
Created:             Fri May 03 16:20:37 +0900 (18 seconds ago)
Started:             Fri May 03 16:20:37 +0900 (18 seconds ago)
Finished:            Fri May 03 16:20:55 +0900 (now)
Duration:            18 seconds

STEP                   PODNAME                        DURATION  MESSAGE
 ✔ shanpu-steps-rndzz
 ├---✔ hello1          steps-rndzz-2539174266  7s
 └-·-✔ hello2a         steps-rndzz-1532723075  7s
   └-✔ hello2b         steps-rndzz-1549500694  10s

DAG

DAG²ではdependenciesに依存関係にあるJobを明示的に指定します．
マニフェストファイルを見てみましょう．

dag.yml

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: dag-diamond-
spec:
  entrypoint: diamond
  templates:
  - name: echo
    inputs:
      parameters:
      - name: message
    container:
      image: alpine:3.7
      command: [echo, "{{inputs.parameters.message}}"]

  - name: diamond
    dag:
      tasks:
      - name: A
        template: echo
        arguments:
          parameters: [{name: message, value: A}]
      - name: B
        dependencies: [A]    # Aの実行終了後Bが実行される
        template: echo
        arguments:
          parameters: [{name: message, value: B}]
      - name: C
        dependencies: [A]    # Aの実行終了後Cが実行される
        template: echo
        arguments:
          parameters: [{name: message, value: C}]
      - name: D
        dependencies: [B, C]    # B，Cの実行終了後Dが実行される
        template: echo
        arguments:
          parameters: [{name: message, value: D}]

DAGもSteps同様，argo submitで実行します．

$ argo submit dag.yml --watch
Name:                dag-diamond-5dkj2
Namespace:           argo
ServiceAccount:      default
Status:              Succeeded
Created:             Fri May 03 16:36:28 +0900 (17 seconds ago)
Started:             Fri May 03 16:36:28 +0900 (17 seconds ago)
Finished:            Fri May 03 16:36:45 +0900 (now)
Duration:            17 seconds

STEP                  PODNAME                      DURATION  MESSAGE
 ✔ dag-diamond-5dkj2
 ├-✔ A                dag-diamond-5dkj2-322730072  9s
 ├-✔ B                dag-diamond-5dkj2-373062929  3s
 ├-✔ C                dag-diamond-5dkj2-356285310  3s
 └-✔ D                dag-diamond-5dkj2-406618167  2s

実行したWorkflowの確認

argoコマンドを使うことでsubmitされたworkflowの確認ができます．

argo list:　submitされたworkflowの一覧を表示

$ argo list
NAME                 STATUS      AGE   DURATION
dag-diamond-5dkj2    Succeeded   4m    17s
steps-rndzz          Succeeded   20m   18s

argo get [workflow_name]: 　submitされたworkflowの詳細を表示

$ argo get dag-diamond-5dkj2                                                                                                                                                                                                2019/05/03 16:40
Name:                dag-diamond-5dkj2
Namespace:           argo
ServiceAccount:      default
Status:              Succeeded
Created:             Fri May 03 16:36:28 +0900 (4 minutes ago)
Started:             Fri May 03 16:36:28 +0900 (4 minutes ago)
Finished:            Fri May 03 16:36:45 +0900 (4 minutes ago)
Duration:            17 seconds

STEP                  PODNAME                      DURATION  MESSAGE
 ✔ dag-diamond-5dkj2
 ├-✔ A                dag-diamond-5dkj2-322730072  9s
 ├-✔ B                dag-diamond-5dkj2-373062929  3s
 ├-✔ C                dag-diamond-5dkj2-356285310  3s
 └-✔ D                dag-diamond-5dkj2-406618167  2s

argo logs [job_name]: 　workflow内の任意のJobの実行ログを表示

$ argo logs dag-diamond-5dkj2-322730072
A

なお各コマンドやオプションについては--helpで確認することができます（当たり前だけど忘れがち）．

$ argo --help
argo is the command line interface to Argo

Usage:
  argo [flags]
  argo [command]

Available Commands:
...

UI

Argoではコマンドを用いた方法以外にもUIで各workflowの確認ができます（ただし何故かLogだけ何も表示されない...）．

$ kubectl -n argo port-forward deployment/argo-ui 8001:8001
# コマンド実行後，ブラウザでlocalhost:8001にアクセスする

（diamond...なるほど......）

Workflow Controlerのスコープ範囲

デフォルトではどのnamespaceからsubmitしてもworkflowは実行されます．
しかし，ConfigMapにスコープ範囲を記述することでWorkflow Controllerの関心の範囲を限定することができます（sample）．

例えばnamespace/argoに限定する場合は次のようなConfigMapを適用します．

scope-config.yml

apiVersion: v1
kind: ConfigMap
metadata:
  name: workflow-controller-configmap
  namespace: argo
data:
  config: |
    namespace: argo

$ kubectl apply -f scope-config.yml

# configmapの内容を反映させるため一度controllerを削除する
# pods/workflow-controllerはdeploymentリソース配下のpodなので，しばらくすると自動的に復活する
$ kubectl delete -n pod workflow-controller-xxxxx

Workflowの定期実行

ネイティブでは時間による定期実行をサポートしていないようです．
考えられる方法としては

argo-eventsのCalendarゲートウェイを使う
CronJobリソースを使う

が挙げられます．
今回は今現在argo-eventsがpre-release状態であること，単に時間による定期実行をしたいだけであることから，後者のCronJobを用いた方法を試してみたいと思います．

具体的な方法としてはCronJobでworkflowを起動するJobを定義＆適切な権限を持ったサービスアカウントを割り当てることで実現させます．
WorkflowのマニフェストはConfigMapに記述し，それをマウントする方法をとりました．
サービスアカウントについてはargo submitしたときにデフォルトでserviceaccounts/defaultが使われるので，これにclusterrole/adminをbindさせます（この方法は公式ドキュメントの3.Configure the service account to run workflowsに記述されています）．

マニフェストファイルは以下のようになります．

cron-wf.yml

apiVersion: v1
kind: ConfigMap
metadata:
  name: manifest-cm
  namespace: argo
data:
  wf-manifest: |
    apiVersion: argoproj.io/v1alpha1
    kind: Workflow
    metadata:
      generateName: dag-diamond-
      namespace: argo
    spec:
      entrypoint: diamond
      templates:
      - name: echo
        inputs:
          parameters:
          - name: message
        container:
          image: alpine:3.7
          command: [echo, "{{inputs.parameters.message}}"]
      - name: diamond
        dag:
          tasks:
          - name: A
            template: echo
            arguments:
              parameters: [{name: message, value: A}]
          - name: B
            dependencies: [A]
            template: echo
            arguments:
              parameters: [{name: message, value: B}]
          - name: C
            dependencies: [A]
            template: echo
            arguments:
              parameters: [{name: message, value: C}]
          - name: D
            dependencies: [B, C]
            template: echo
            arguments:
              parameters: [{name: message, value: D}]

---
apiVersion: batch/v1beta1
kind: CronJob
metadata:
  name: cron-wf
spec:
  schedule: "*/1 * * * *"
  jobTemplate:
    spec:
      backoffLimit: 0
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: exec-workflow
              image: argoproj/argocli:latest
              command: ["argo","submit"]
              args: ["/argo-wf/wf-manifest"]
              volumeMounts:
                - name: wf-manifest
                  mountPath: /argo-wf/
          volumes:
            - name: wf-manifest
              configMap:
                name: manifest-cm

失敗したときの通知

ネイティブでの通知機能はありませんが，workflowの終了ステータスを取得できるのでこれを用いることで失敗または成功の通知が可能です．
exampleコードは公式リポジトリにあります（こちら）．

onExitにメインで実行したいtemplateの後に実行したいtemplate名を記述します．
onExitに記述したtemplate内ではwhenを用いてworkflow.statusに応じて呼び出すJobを変更します．
このとき呼び出されるJobのimageに"Slackへの通知"や"メール通知"を行うものを用いることで通知機能が実現できます．

なお，Workflow内で使える内部変数の一覧はこちらにまとめられています．

困ったときは...

ドキュメントはこちら
exampleコードはこちら
issueはこちら

困ったときはこれらを覗いてみるとヒントがあるかもしれません．

まとめ

ArgoはWorkflowの定義にKubernetes同様のマニフェストファイルを適用するため，普段からKubernetesを使っている人にとって導入はスムーズに行えるのではないかと思います．
特に既存のJobTemplateをほぼそのまま使えるのはいいですね．

今回はver2.2.1を使いましたが，記事執筆時点でver2.3.0がpre-releaseとなっていました．
バグの修正だけでなく，様々な機能が追加されており，導入を検討している方は見てみると良いかもしれません．
またver2.4-Milestoneでは，"argo-eventsと統合した新しいAPIサーバ" との記述があり，argo-eventsとどのように発展していくのか注目したいところです．

apiバージョンはまだalphaと成長途中のArgoですが，完成度は高く，個人的に今のところKubernetes上で動かすWorkflow EngineとしてはArgo一択のような気がしています．
この記事を読んでArgoに興味が湧いた方は，一度GWの空いた時間に触ってみてはいかがでしょうか（といってもGWは今日を含めてあと2日...）．

厳密には"Argo Workflows"という名前ですが，簡単のため本記事では"Argo"と略記します． ↩
DAGは日本語で有向非巡回グラフを意味します． ↩

KubernetesプラットフォームのWorkflow Engine Argoを触ってみた