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

はじめに

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

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

• ワークフローツールを使用する
  　　e.g. Argo Workflows、Apache Airflow　etc

• カスタムコントローラーを使って機能を充実させる
  　　e.g. Kueue、kube-batch　etc

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

Furiko とは

The word "furiko" (振り子) means "pendulum" in Japanese.
Logo designed by Duan Weiwei, distributed under CC-BY 4.0.

Rundeck を用いて　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 は以下のようになります。

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 の情報を取得することができます。

$ 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 を使用しても情報を取得することができます。

$ 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 の公式ドキュメントの機能比較表を元に、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 にのみ ✅ がついている機能について紹介していきます。

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

Scheduling

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

Scheduling constraints

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

  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"

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

Cron load balancing

Furiko では unix-cron に加えて H を使用できます。

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

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

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

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

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

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

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

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

Enqueue jobs for later

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

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

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

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

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

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>

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

Task Execution

Retries using separate Pods

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

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

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

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

Pending timeouts for dead nodes

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

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

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

Parallel expansion by list and matrix

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

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

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

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

Task Execution

Parameterization of job inputs

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

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

• JobConfig

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の定義)

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

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

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

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

$ 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 形式

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

所感

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