ArgoWorkflow Parallel Nested Dag 구현하기 ( with Kubernetes )

- 목차

 

함께 보면 좋은 자료.

https://westlife0615.tistory.com/407

KinD 알아보기.

https://westlife0615.tistory.com/516

[Kubernetes] Argo Workflows 시작하기

소개.

Argo Workflow 를 통해서 여러 Job 들을 Scheduling 할 수 있습니다.

저의 경우는 두 종류의 Parameter 들을 조합하여 Nested Loop 을 구현해야 했던 경험이 있습니다.

예를 들어, 사용자 별로 2가지의 Task 를 수행해야하는 경우가 있었죠.

아래와 같은 2가지 배열이 존재할 때, 각각의 사용자에게 say_hello, say_goodbye Task 를 수행하는 경우입니다.

user_list=( "Andy" "Bob" "Chris" "Daniel" )
task_list=( "say_hello" "say_goodbye" )

 

이를 Shell Script 로 구성하면 아래와 같습니다.

user_list=( "Andy" "Bob" "Chris" "Daniel" )
task_list=( "say_hello" "say_goodbye" )

for user in $user_list; do
  for task in $task_list; do
    if [ "$task" = "say_hello" ]; then
    	echo hello $user
    elif [ "$task" = "say_goodbye" ]; then
		echo goodbye $user
    fi
  done;
  echo 
done

hello Andy
goodbye Andy

hello Bob
goodbye Bob

hello Chris
goodbye Chris

hello Daniel
goodbye Daniel

 

이번 글에서는 위와 같은 구조의 Nested Loop 를 Argo Workflow 에서 구현하는 방법에 대해서 알아보려고 합니다.

Argo 설치하기.

저는 Kubernetes 환경에서 Argo Workflow 를 구현하도록 하겠습니다.

그리고 KinD 를 통해서 Local 환경에 Kubernetes 를 세팅하였습니다.

아래 명령어를 실행하여 Argo 관련 Namespace, CRD, Deployment 를 생성하겠습니다.

 

kubectl create namespace argo
kubectl apply -n argo -f https://github.com/argoproj/argo-workflows/releases/download/v3.4.4/install.yaml

cat <<EOF> /tmp/argo-server-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: argo-server
  namespace: argo
spec:
  selector:
    matchLabels:
      app: argo-server
  template:
    metadata:
      labels:
        app: argo-server
    spec:
      containers:
      - args:
        - server
        - --secure=false
        - --auth-mode=server
        env: []
        image: quay.io/argoproj/argocli:v3.4.4
        name: argo-server
        ports:
        - containerPort: 2746
          name: web
        readinessProbe:
          httpGet:
            path: /
            port: 2746
            scheme: HTTP
          initialDelaySeconds: 10
          periodSeconds: 20
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            drop:
            - ALL
          readOnlyRootFilesystem: true
          runAsNonRoot: true
        volumeMounts:
        - mountPath: /tmp
          name: tmp
      nodeSelector:
        kubernetes.io/os: linux
      securityContext:
        runAsNonRoot: true
      serviceAccountName: argo-server
      volumes:
      - emptyDir: {}
        name: tmp
EOF

kubectl -n argo apply -f /tmp/argo-server-deployment.yaml

 

위 명령어를 통해서 Argo 의 Namespace, CRD, Deployment  생성을 마쳤다면,

아래 명령어를 통해서 argo-server 를 port-forward 합니다.

이를 통해서 Argo Workflow 의 Web UI 로 진입할 수 있습니다.

kubectl -n argo port-forward service/argo-server 2746:2746

 

 

Workflow 생성하기.

간단한 예시로 아래 페이지에서 제공되는 기본적인 parallelism-nested Workflow 를 실행해보도록 하겠습니다.

https://github.com/argoproj/argo-workflows/blob/main/examples/parallelism-nested.yaml

cat <<EOF> /tmp/test-workflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: parallelism-nested-
spec:
  arguments:
    parameters:
    - name: seq-list
      value: |
        ["a","b","c","d"]
    - name: parallel-list
      value: |
        [1,2,3,4]

  entrypoint: parallel-worker
  templates:
  - name: parallel-worker
    inputs:
      parameters:
      - name: seq-list
      - name: parallel-list
    steps:
    - - name: parallel-worker
        template: seq-worker
        arguments:
          parameters:
          - name: seq-list
            value: "{{inputs.parameters.seq-list}}"
          - name: parallel-id
            value: "{{item}}"
        withParam: "{{inputs.parameters.parallel-list}}"

  - name: seq-worker
    parallelism: 1
    inputs:
      parameters:
      - name: seq-list
      - name: parallel-id
    steps:
    - - name: seq-step
        template: one-job
        arguments:
          parameters:
          - name: parallel-id
            value: "{{inputs.parameters.parallel-id}}"
          - name: seq-id
            value: "{{item}}"
        withParam: "{{inputs.parameters.seq-list}}"

  - name: one-job
    inputs:
      parameters:
      - name: seq-id
      - name: parallel-id
    container:
      image: alpine
      command: ['/bin/sh', '-c']
      args: ["echo {{inputs.parameters.parallel-id}} {{inputs.parameters.seq-id}}; sleep 10"]
EOF

kubectl create -f /tmp/test-workflow.yaml

 

위와 같이 동시적으로 Job 들이 수행됩니다.

 

 

Workflow 리소스의 속성 알아보기.

spec.arguments[*].parameters[*]

parameters 는 Workflow 에서 사용할 입력 데이터입니다.

각각의 step 에 해당하는 Job 의 Container 내부로 파라미터를 전달할 수 있습니다.

아래와 같이 표현할 수 있고,

["a","b","c","d"], [1,2,3,4] 인 2개의 파라미터 배열을 사용합니다.

spec:
  arguments:
    parameters:
    - name: seq-list
      value: |
        ["a","b","c","d"]
    - name: parallel-list
      value: |
        [1,2,3,4]

 

 

spec.entrypoint

spec.entrypoint 는 처음으로 실행할 Template 을 의미합니다.

Workflow 는 여러개의 Template 를 다루는데요.

각 Template 들은 아래의 개별적인 단위를 생성할 수 있는데요.

(동그라미 + 체크) 표시로 된 단위를 생성합니다.

 

아래 yaml 이 template 를 설정하는 예시입니다.

templates:
  - name: parallel-worker
    inputs:
      parameters:
      - name: seq-list
      - name: parallel-list
    steps:
    - - name: parallel-worker
        template: seq-worker
        arguments:
          parameters:
          - name: seq-list
            value: "{{inputs.parameters.seq-list}}"
          - name: parallel-id
            value: "{{item}}"
        withParam: "{{inputs.parameters.parallel-list}}"

  - name: seq-worker
    parallelism: 1
    inputs:
      parameters:
      - name: seq-list
      - name: parallel-id
    steps:
    - - name: seq-step
        template: one-job
        arguments:
          parameters:
          - name: parallel-id
            value: "{{inputs.parameters.parallel-id}}"
          - name: seq-id
            value: "{{item}}"
        withParam: "{{inputs.parameters.seq-list}}"

  - name: one-job
    inputs:
      parameters:
      - name: seq-id
      - name: parallel-id
    container:
      image: alpine
      command: ['/bin/sh', '-c']
      args: ["echo {{inputs.parameters.parallel-id}} {{inputs.parameters.seq-id}}; sleep 10"]

 

 

spec.templates[*].inputs

template 의 inputs 은 template 의 input 으로 사용할 데이터를 의미합니다.

보통 이전 tempate 의 output 을 input 으로 사용하게 됩니다.