Kubernetes での Job 実行機能を拡張する "Furiko" について

## はじめに

Kubernetes では Job を実行するための機能として CronJob と Job というリソースが用意されていますが、そこまで機能が豊富なわけではないのでユースケースによっては機能が足りない場合が出てきます。

その場合には現状では以下のいずれかを用いるケースが多いです。

- ワークフローツールを使用する  
　　e.g. [Argo Workflows](https://argoproj.github.io/argo-workflows/)、[Apache Airflow](https://airflow.apache.org/)　etc


- カスタムコントローラーを使って機能を充実させる
　　e.g. [Kueue](https://github.com/kubernetes-sigs/kueue)、[kube-batch](https://github.com/kubernetes-sigs/kube-batch)　etc

今回はその中で、カスタムコントローラーを使って機能を充実されるアプローチを選択した　[Furiko](https://furiko.io/) というプロダクトについて紹介してきます。


## Furiko とは

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/309176/b700f122-53b9-b179-85dc-a1ceb545f60c.png)
>The word "furiko" (振り子) means "pendulum" in Japanese.
Logo designed by Duan Weiwei, distributed under CC-BY 4.0.

[Rundeck](https://www.rundeck.com/) を用いて　Job を実行するプラットフォームを運用しており、それを k8s 上に移行を検討したところ、既存の k8s の CronJob/Job の機能では要件を満たせたかったために独自に開発されたそうです。

## Furiko で Job を実行する時の考え方

Furiko　では k8s における `CronJob-> Job -> Pod` の関係と同じように `JobConfig -> Job -> Task` という考え方が存在します。
(Furiko で Job という単語を使用する際は k8s の batch/v1　の Job ではなく Furiko 独自の execution.furiko.io/v1alpha1 の　Job のことを意味します。)

実際の Manifest は以下のようになります。

```yaml
apiVersion: execution.furiko.io/v1alpha1
kind: JobConfig
metadata:
  name: example-jobconfig
spec:
  # Concurrency configuration.
  concurrency:
    policy: Forbid # Allow or Forbid or Enqueue
    maxConcurrency: 1

  # Schedule configuration.
  schedule:
    cron:
      expression: "H H/3 * * *"
      timezone: "Asia/Singapore"
    disabled: false
    constraints:
      notBefore: "2022-01-01T00:00:00-03:00"
      notAfter: "2023-01-01T00:00:00-03:00"

  # Job options.
  option:
    options:
      - type: String
        name: username
        label: Username
        string:
          default: Example User
          trimSpaces: true

  # Template for the Job to be created. For more info, see the Job sample configuration.
  template:
    # Any labels and annotations will be automatically added to downstream Jobs.
    metadata:
      annotations:
        annotations.furiko.io/job-group: "cool-jobs"
    spec:
      # Specifies maximum number of attempts for each task, defaults to 1.
      maxAttempts: 3

      # Optional delay between each task retry.
      retryDelaySeconds: 10

      # Optional duration in seconds for how long each task should be pending for
      # until it gets killed.
      taskPendingTimeoutSeconds: 1800

      # The template for each task to be created by the Job.
      taskTemplate:
        # Specify how to create the task as a Pod. This is just a PodTemplateSpec.
        pod:
          spec:
            containers:
              # Notice how we can use context variables and job options inside
              # the PodSpec freely to be substituted at runtime.
              - name: job-container
                args:
                  - echo
                  - "Hello world, ${option.username}!"
                env:
                  - name: JOBCONFIG_NAME
                    value: "${jobconfig.name}"
                  - name: JOB_NAME
                    value: "${job.name}"
                image: "alpine"
                resources:
                  limits:
                    cpu: 100m
                    memory: 64Mi
```

書き方はとても CronJob に似ており、これが Job の定義になります。
基本的な使い方は　`JobConfig` を作成して以下の k8s にデプロイして使用します。

```
$ kubectl apply -f example-jobconfig.yaml
jobconfig.execution.furiko.io/example-jobconfig created
```

Furiko には独自の CLI が用意されているので、それを用いて `JobConfig` から Job を作成してアドホックに実行することもできます。

```
$ furiko run example-jobconfig
Job default/example-jobconfig-klptm created
```

以下のように独自の CLI を用いて Job の情報を取得することができます。

```yaml
$ furiko get job example-jobconfig-klptm
JOB INFO
Name:                example-jobconfig-klptm
Namespace:           default
Type:                Adhoc
Created:             Wed, 14 Dec 2022 01:15:58 +08:00 (22 seconds ago)
Job Config:          forbid-example
Concurrency Policy:  Forbid

JOB STATUS
Phase:         Succeeded
Started:       Wed, 14 Dec 2022 01:15:58 +08:00 (22 seconds ago)
Tasks:         0 Running / 1 Succeeded
Run Duration:  About a second
Result:        Success

LATEST TASK
Name:      example-jobconfig-klptm-gezdqo-0
Created:   Wed, 14 Dec 2022 01:15:58 +08:00 (22 seconds ago)
State:     Terminated
Started:   Wed, 14 Dec 2022 01:16:00 +08:00 (20 seconds ago)
Finished:  Wed, 14 Dec 2022 01:16:00 +08:00 (20 seconds ago)
```

通常の `kubectl get` や `kubectl describe` を使用しても情報を取得することができます。

```yaml
$ kubectl get furikojob example-jobconfig-q4dw9 -o yaml
apiVersion: execution.furiko.io/v1alpha1
kind: Job
metadata:
  creationTimestamp: "2022-12-22T06:31:11Z"
  finalizers:
  - execution.furiko.io/delete-dependents-finalizer
  generateName: example-jobconfig-
  generation: 1
  labels:
    execution.furiko.io/job-config-uid: 7d95384a-3cd4-49eb-8a47-d8d2fbca027f
  name: example-jobconfig-q4dw9
  namespace: default
  ownerReferences:
  - apiVersion: execution.furiko.io/v1alpha1
    blockOwnerDeletion: true
    controller: true
    kind: JobConfig
    name: example-jobconfig
    uid: 7d95384a-3cd4-49eb-8a47-d8d2fbca027f
  resourceVersion: "155365"
  uid: f68e732f-64b3-40a1-953a-6eced8e6c536
spec:
  startPolicy:
    concurrencyPolicy: Forbid
  substitutions:
    jobconfig.cron_schedule: '*/5 * * * *'
    jobconfig.cron_timezone: ""
    jobconfig.name: example-jobconfig
    jobconfig.namespace: default
    jobconfig.uid: 7d95384a-3cd4-49eb-8a47-d8d2fbca027f
  template:
    maxAttempts: 1
    taskTemplate:
      pod:
        metadata:
          creationTimestamp: null
        spec:
          containers:
          - command:
            - echo
            - Hello World
            image: alpine
            name: container
            resources: {}
          restartPolicy: Never
  ttlSecondsAfterFinished: 3600
  type: Adhoc
status:
  condition:
    finished:
      finishTime: "2022-12-22T06:31:15Z"
      latestCreationTimestamp: "2022-12-22T06:31:11Z"
      latestRunningTimestamp: "2022-12-22T06:31:15Z"
      result: Success
  createdTasks: 1
  phase: Succeeded
  runningTasks: 0
  startTime: "2022-12-22T06:31:11Z"
  tasks:
  - containerStates:
    - containerID: containerd://b644e9726aa7cb553f2f1502a980ad660fc68d5b2f88cacc728a72d70dd5ff34
      exitCode: 0
      reason: Completed
    creationTimestamp: "2022-12-22T06:31:11Z"
    deletedStatus:
      result: Succeeded
      state: Terminated
    finishTimestamp: "2022-12-22T06:31:15Z"
    name: example-jobconfig-q4dw9-gezdqo-0
    nodeName: kind-worker
    parallelIndex:
      indexNumber: 0
    retryIndex: 0
    runningTimestamp: "2022-12-22T06:31:15Z"
    status:
      result: Succeeded
      state: Terminated
```

## Furiko の特徴

k8s の既存の CronJob/Job と比較した際の Furiko の特徴については、[Furiko の公式ドキュメント](https://furiko.io/docs/features)の機能比較表を元に、k8s にはない機能に絞って以降で説明していきます。

公式の比較表は、比較対象となっている　k8s　のバージョンが低いようでしたので、k8s v1.26.0 の機能との差分を元に再整理した表が以下になります。

- Scheduling

Feature | Furiko | batch/v1
-- | -- | --
Cron expressions | ✅ | ✅ (cronjob.spec.schedule により実現)
Cron timezone | ✅ | ✅ (k8s 1.25 より Beta。cronjob.spec.timeZone により実現)
Scheduling constraints | ✅ | ❌
Cron load balancing | ✅ | ❌
Enqueue jobs for later | ✅ | ❌
Forbid concurrent with adhoc execution | ✅ | ❌
Back-scheduling | ✅ | ✅ (cronjob.spec.startingDeadlineSeconds により実現)


- Task Execution

Feature | Furiko | batch/v1
-- | -- | --
Retries using separate Pods | ✅ | ❌
Pending timeouts for dead nodes | ✅ | ❌
Multiple parallel Pods per Job | ✅ | ✅ (k8s 1.22 より Beta。job.spec.completionMode により実現)
Parallel expansion by list and matrix | ✅ | ❌

- General

Feature | Furiko | batch/v1
-- | -- | --
Automatic cleanup with TTL | ✅ | ✅ (k8s 1.21 より Beta。job.spec.ttlSecondsAfterFinished により実現)
Parameterization of job inputs | ✅ | ❌

上記のうち、 Furiko にのみ ✅ がついている機能について紹介していきます。

> :pencil: のマークがついているのもは筆者の個人的な感想になります。

### Scheduling

スケジュールに関する機能になります。

#### Scheduling constraints 

schedule に対して `notBefore` と `notAfter` で CronJob が実行できる範囲を指定できます。

```diff
  schedule:
    cron:
      expression: "H H/3 * * *"
    disabled: false
+    constraints:
+      notBefore: "2022-01-01T00:00:00-03:00"
+      notAfter: "2023-01-01T00:00:00-03:00"
```

:pencil: 該当の Job は何日以降から実行したいケースはあると思うので結構便利な機能だなと思います。


#### Cron load balancing

Furiko では [unix-cron](https://man7.org/linux/man-pages/man5/crontab.5.html) に加えて `H` を使用できます。

```diff
  schedule:
    cron:
+      expression: "H H/3 * * *"
```

`H` を使用すると指定した時間区間の範囲内で一意のランダムな値が設定されるので、多数の CronJob が存在する場合に実行時刻がずらせるというメリットがあります。

`H` については以下の 4 つの使い方がサポートされています。

使用方法 | 説明 
-- | --  
`H` |`H * * * *` : この場合は、1時間毎に `H` 分に1回実行されます。
`H/６` |  `0 H/6 * * *`: AM 0〜5時のいずれかで実行され、以降は6時間毎に実行されます。<br>H=0: 00:00 に実行され、以降は6時間毎に実行<br>H=2: 02:00 に実行され、以降は6時間毎に実行
`H(5-19)` | `H(5-19) * * * *`: 1時間毎に1回実行されます。その際の実行分は5〜19分のいずれかになります。<br>H=5: 00:05 に実行される。<br>H=12: 00:12 に実行される。
`H(5-19)/５` | `H(5-19)/5 * * * *`: 5〜19分の範囲で実行され、以降は5分毎に実行されます。<br>H=0: 00:05, 00:10, 00:15, 01:05, 01:10, 01:15, ... <br>H=2: 00:07, 00:12, 00:17, 01:07, 01:12, 01:17, ..

k8s の CronJob で実行時刻をずらそうとすると以下の2つのいずれかをやる必要があると思います。

- CronJob 毎に実行時刻を変える
- 実行する Job に対して jitter を実装する

CronJob の時刻を変えるのは管理が煩雑で面倒です。jitter を入れる場合は機能追加するのが手間なのに加えて、Pod が割り当てられた後に実行を待つので request に指定された分のリソースが確保された状態で、アプリケーションが起動しない状態となるので、リソースの管理の面で少し勿体無く感じます。

:pencil: 正直、この機能は k8s の方に導入してほしい。実行時間をずらす作業がだるいので私、すぐ使いたいです。

#### Enqueue jobs for later

以下のように JobConfig に対して同時実行に対するポリシーを設定できます。

```diff
spec:
  concurrency:
+    policy: Forbid # Allow or Forbid or Enqueue
+    maxConcurrency: 1
```

設定できるポリシーは以下の3種類です。

Policy | 説明
-- | --
`Allow` | 同時に実行できる Job に制限はなし
`Forbid` | 複数の Job の同時実行を禁止
`Enqueue` | Job をキューに入れて他の Job が完了した後で実行を開始する

k8s にある `Replace` がなくて代わりに `Enqueue` が指定できるのが違いになります。

:pencil: k8s の CronJob で `parallelism=1` を設定した時も似たような挙動にできそうなので、ウリはウリなんだと思いますが、そこまで差別化のポイントでもないのかなと思っています。

#### Forbid concurrent with adhoc execution

Job をアドホックに実行する際に JobConfig で設定した `Forbid` のポリシーが Job でも適用されるのが k8s との差分になります。
 
Furiko では独自の CLI により JobConfig から Job をアドホックに生成することができます。k8s で Cronjob から Job を生成した時にはポリシーが適用されないけど、Furiko だと適用されるというのがウリになります。

- Furiko の CLI で JobConfig から Job を作成する場合

```
$ furiko run <JobConfig名>
```

- k8s で CronJob から Job を作成する場合

```
$ kubectl create job <NAME> --from=cronjob/<name>
```

:pencil: 個人的にそんなにここを推さなくてもいいのでは？と思った機能になります。実際の運用中に Job が失敗した時のリカバリで手動で実行する時に制限がかかると嬉しいかも。開発期間中の手動実行だと同時実行ポリシーはなくても嬉しいケースもあるかなと思いました。


### Task Execution

#### Retries using separate Pods

ノードの設定誤りやホストの問題で Job が実行した際に、同一のノードで再度 Job を実行しても失敗する可能性があるためにリトライ時に別のノードで Job を実行する機能が追加されています。

:pencil: k8s の Job でも `spec.restartPolicy=Never` に設定すれば失敗時に Pod の再生成をしてくれますし。ノードの問題についてはノードのヘルチェックの方で確認して、問題がある場合はノードを `unschedulable` にして Pod のスケジュール対象にならないようにするアプローチの方が良いかなと個人的には思うので、あまり心は惹かれなかったです。

:pencil: リトライに関しては Job の失敗時の遅延時間を指定できる `retryDelaySeconds` の機能が便利だなと思いました。Job の失敗時に一旦、時間を置きたいケースは多いと思うので汎用的に使えるいい機能だなと思いました。

```diff
    spec:
      # Optional delay between each task retry.
+      retryDelaySeconds: 10
```

#### Pending timeouts for dead nodes

これは Pending に対してタイムアウトを設定できる機能になります。

```diff
    spec:
      # Optional duration in seconds for how long each task should be pending for
      # until it gets killed.
+      taskPendingTimeoutSeconds: 1800
```

:pencil: Job が Pending のステータスのままだと、失敗もしないのでアラートで気付けずにそのままになるケースも考えられるので、この機能を使って Job を失敗にできるっていうのは結構便利だし、汎用的に使えるんじゃないかと思いました。こういう機能を見るのは運用していく上で勉強になりますね。

#### Parallel expansion by list and matrix

並列に Job を実行する際に各 Job 毎に個別にパラメータを与えられる機能になります。
文章だけだとわかりにくいので、実際の例を用いて説明します

以下のように `spec.parallelism.withMatrix` に付与したいパラメータの key/value を設定します。
実行する Task の環境変数の方に `${task.index_matrix.goos}` のよう形で設定します。

```diff
apiVersion: execution.furiko.io/v1alpha1
kind: Job
spec:
  parallelism:
+    withMatrix:
+      goos:
+        - linux
+        - darwin
+        - windows
+      goarch:
+        - amd64
+        - arm64
    completionStrategy: AllSuccessful
  template:
    taskTemplate:
      pod:
        spec:
          containers:
            - name: job-container
              image: my-image
              env:
+                - name: GOOS
+                  value: ${task.index_matrix.goos}
+                - name: GOARCH
+                  value: ${task.index_matrix.goarch}
              args:
                - go build ./...
```

この時 `goos` の組み合わせが3パターン、`goarch` の組み合わせが2パターンとなり、組み合わせとして 3 * 2 = 6 パターンがあるので、6個の Task が以下のように並列で実行されます。

タスク | goos | goarch
-- | -- | --
0 | linux | amd64
1 | linux | arm64
2 | darwin | amd64
3 | darwin | arm64
4 | windows | amd64
5 | windows | arm64

:pencil: k8s でも k8s 1.22 から IndexedJob の機能が利用できるようになり、並列実行時の Job に対してユニークな番号を振れるようになったので、それと Configmap を組み合わせたら似たようなことはできるのですが、こういう形で機能として提供されてると便利ですね。自分で実装するとなると少し手間ですし、こちらの方がパラーメータのバリエーションがそこまで多くないなら見やすいかなと思いました。


### Task Execution

#### Parameterization of job inputs

JobConfig に入力パラメータの定義を追加でき、Job の作成時に入力パラメータをチェックして Task で実行できる機能になります。

これも分かりにくいので、実際の例を用いて説明します。

- JobConfig

```diff
apiVersion: execution.furiko.io/v1alpha1
kind: JobConfig
metadata:
  generateName: jobconfig-sample
spec:
  option:
    options:
+      - type: String
+        name: username
        label: Username
        string:
          default: Guest User
          trimSpaces: true
          required: true
```

上記のように String での入力を期待する `username` を定義します。


- PodTemplateSpec(Taskの定義)

```diff
spec:
  containers:
    - name: job-container
      args:
        - echo
+        - "Hello world, ${option.username}!"
```

JobConfig で定義した値を Task で利用する際には `option.<オプション名>` で利用することができます。

入力パラメータについては、JobConfig から Job を作成する際にチェックされます。

- JobConfig から CLI で Job を作成する際の例

```diff
$ furiko run jobconfig-sample
+Please input option values.
+? Username  　　　<-`required: true` に設定したため、入力を求めれらています。

Job default/jobconfig-sample-fds52 created
```

定義できる種類は以下になります。

Name | Input Type | Description
-- | -- | --
String | string | 文字列
Bool | bool | 真偽値
Select | string | リスト形式の文字列
Multi | []string | 複数の文字列のリスト
Date | string | RFC3339 形式


:pencil: Job の定義だけを用意しておいて、実行時にパラメータを指定するみたいなことを k8s してしようと思ったことがなかったので面白いなと思いました。普段 shell とかで作ってるものを Job で置き換えたりするのに使うのかな？ この機能の開発にするに至った背景とかも併せて知りたいですね。

## 所感

こういう機能開発には、その環境での開発ノウハウとかが詰まっているので調べみると勉強になって面白かったです。こういう独自のコントローラーで実装されて、その中で k8s でも汎用的に使える機能が　k8s の方に取り込まれてプロダクトとして成長していくのかなと感じることができました。