こんにちは、アトラエでインフラエンジニアをやっているくーまです。

今回は掲題の通り、Argo Eventsでの並列実行数の制御の方法について書いていこうと思います。

前提としては↓のようなものがあります。

ArgoEventsを使用している
ArgoWorkflowで処理を実行している

ArgoEventsを導入している場合、何かしら非同期に処理を行いたいといった要件が多いと思われます。

そういった要件の中には、処理に時間がかかったり、そこそこのコンピューティングリソースを必要とするものもあるでしょう。

そうなると必要とされるリソース量に応じてスケーリングしていきたいとなるわけですが、特にDBに負荷がかかる場合などはスケーリングの速度が追いつかなかったり、非同期処理のためだけにスケーリングに最適化されたDBを採用できなかったりという理由で、非同期処理の並列実行数を制御してしまうことが一番現実的、ということもあり得るのではないかと思います。

ということで今回は、ArgoEventsで並列実行数をどう制御するのかについて書いていきます。

同時実行数の制御はArgoEvents側ではなくArgoWorkflow側で行う

いきなりタイトルを破壊している感じの見出しになってしまいましたが、実際こうなりました。

色々と調べてみたところ、ArgoEvents側には同時実行数の制御を行う仕組みは備わっておらず、ArgoWorkflow等の実行側で制限をかけることを想定しているようです。

そのため、以下はArgoWorkflowでの同時実行数の制御の方法と、その挙動になります。

※当然ですが、ArgoWorkflowである必要はありません

同時実行数を制御するYAMLの記述

ArgoWorkflowにはsynchronizationという設定を入れることができます。

これは、該当するWorkflowのみに同時実行の制限を入れられる、というものです。

ワークフローAは同時実行に制限なし、ワークフローBは最大3つまで…みたいな感じにできる、ということです。

今回は実際の挙動まで書いていくため、サンプルのSensorのYAMLを載せます。

なお、今回は引数で指定した秒数sleepするコンテナを動かす想定で書いています。

また、EventSource等は良い感じに作成済みという前提で、Sensorのみ記載します。

アーキテクチャは

AWS SQS → EventSource → EventBus → Sensor → Workflow → Pod

といった感じです。

SQSにJSONを詰め、そのメッセージ内容を引数としてsleepをかける感じです。

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: test-sleep
spec:
  template:
    serviceAccountName: test-sleep-sa
  eventBusName: test-sleep
  dependencies:
  - name: test-sleep
    eventSourceName: test-sleep
    eventName: test-sleep

  triggers:
  - template:
      name: test-sleep-wf
      k8s:
        group: argoproj.io
        version: v1alpha1
        resource: workflows
        operation: create
        source:
          resource:
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: test-sleep-
            spec:
              entrypoint: test-sleep
              synchronization:
                semaphore:
                  configMapKeyRef:
                    name: test-sleep-conf
                    key: PARALLELISM

              arguments:
                parameters:
                - name: time
                  value: value
              templates:
              - name: test-sleep
                serviceAccountName: test-sleep-sa
                inputs:
                  parameters:
                  - name: time
                container:
                  image: alpine:latest
                  args: [sleep, '{{inputs.parameters.time}}']
                  resources:
                    requests:
                      memory: 100Mi
                      cpu: 100m
                    limits:
                      memory: 100Mi
                      cpu: 100m
        parameters:
        - src:
            dependencyName: test-sleep
            dataKey: body.time
          dest: spec.arguments.parameters.0.value

ここで大事なのは真ん中あたりにある

              synchronization:
                semaphore:
                  configMapKeyRef:
                    name: test-sleep-conf
                    key: PARALLELISM

ここの部分です。

ConfigMapで定義したPARALLELISMという設定値を読み込み、その数値の数までしか同時実行されないようになる、という記述です。

今回は3と指定してみます。