Parallel Workflows on Kubernetes

Applications are now increasingly distributed, running on multiple machines and accessed by multiple users from all over the world. By bundling the application code, the application runtime, and the libraries, containers, and container orchestrators, we have addressed many of the challenges of building distributed systems. With container runtimes like Docker, we can deploy our applications on different environments. And with container orchestration tools like Kubernetes, we can scale our applications. We frequently have to break our distributed application into a collection of multiple containers running on different machines. This requires us to coordinate execution and facilitate communication among different containers. A scenario where this situation is encountered is when we compose a workflow using multiple containers. In this article, we will learn how to build such workflows using a container workflow engine, Argo, for Kubernetes. We will develop workflows for the following two examples of so-called "embarrassingly parallel" problems for which Kubernetes is the ideal scaling platform.

• N-Queens using genetic algorithms

• Distributed search

N-Queens Using Genetic Algorithms

The N-Queens problem is to place N queens on an NxN chessboard so that no two attack. Since each queen must be on a different row and column, we can assume that queen i is placed in i-th column. All solutions to the N-Queens problem can, therefore, be represented as N-tuples (q1, q2, …, qN) that are permutations of an N-tuple (1, 2, 3, …, N). The position of a number in the tuple represents the queen's column position, while its value represents the queen's row position. The complexity of the search is N*(N-1)* ... *1 = N!. 

Genetic Algorithms

Genetic algorithms can be used to search for a permutation of N non-attacking queens. The pseudo-code shown below is a simple genetic algorithm to search for a solution.

create initial population
evaluate initial population

while not done
    select 3 individuals
    run mutation operator
    evaluate offspring
    if solution found, set done = true
end

The choice of initial population, random here, determines how fast the solution is found. There are different ways to parallelize this algorithm to speed it up. One of the simplest ways is to run this same computation on multiple workers with a different randomly-selected initial population. Once any worker has exited with success, no other worker should still be doing any work or writing any output. They should all exit as soon as one of the workers finds the solution.

Argo Workflow

Let us create a workflow to be run on Kubernetes that runs this genetic algorithm along with other subsequent processing tasks. A Kubernetes Job with a specified number of completions and parallelism is used for parallel execution of pods. We create a Job object with 5 completions and 5 parallelism that will launch 5 pods in parallel to search for a solution. A Redis queue service is used for pubsub. The pod that finishes first will publish a message on a channel on the Redis server. All workers are subscribed to the channel. On receiving "finished" message, all other workers stop the search and exit. 

The Docker image containing the Python implementation of the genetic algorithm as described above can be pulled from the DockerHub. This image requires a Redis service to be running on the Kubernetes cluster. Refer to the Kubernetes documentation on how to create and configure a Redis service on Kubernetes.

> docker pull randhirkumars/n-queens-genetic-redis

The workflow then consists of the following steps:

1. Create a parallelized Kubernetes Job which launches 5 parallel workers. Once any pod has exited with success, no other pod will be doing any work. Return the job name and job uid as output parameters.

2. Using the uid of the job, query any of its associated pods and print the result to the stdout.

3. Delete the job using the job name.

The Argo workflow YAML with the above steps is shown below:

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: nqueens-
spec:
  entrypoint: nqueens-job
  templates:
  - name: nqueens-job
    steps:
    - - name: nqueens-genetic-job
        template: nqueens-genetic-job
    - - name: print-solution
        template: print-solution
        arguments:
          parameters:
          - name: job-uid
            value: '{{steps.nqueens-genetic-job.outputs.parameters.job-uid}}'
    - - name: delete-job
        template: delete-job
        arguments:
          parameters:
          - name: job-name
            value: '{{steps.nqueens-genetic-job.outputs.parameters.job-name}}'
  - name: nqueens-genetic-job
    resource:
      action: create
      successCondition: status.succeeded > 1
      failureCondition: status.failed > 0
      manifest: |
        apiVersion: batch/v1
        kind: Job
        metadata:
          namespace: default
          name: nqueens
          labels:
            job: n-queens
        spec:
          parallelism: 5
          completions: 5
          template: 
            metadata:
              labels:
                job: n-queens
            spec:
              containers:
              - name: worker
                image: randhirkumars/n-queens-genetic-redis
              restartPolicy: Never
    outputs:
      parameters:
      - name: job-name
        valueFrom:
          jsonPath: '{.metadata.name}'
      - name: job-uid
        valueFrom:
          jsonPath: '{.metadata.uid}'

  - name: print-solution
    inputs:
      parameters:
      - name: job-uid
    container:
      image: argoproj/argoexec:latest
      command: [sh, -c]
      args: ["
        for i in `kubectl get pods -l controller-uid={{inputs.parameters.job-uid}} -o name`; do
          kubectl logs $i;
        done
      "]

  - name: delete-job
    inputs:
      parameters:
      - name: job-name
    resource:
      action: delete
      manifest: |
        apiVersion: batch/v1
        kind: Job
        metadata:
          name: {{inputs.parameters.job-name}}

Notice that each step in the workflow is a Docker container. The first step uses a Docker image that we have built — randhirkumars/n-queens-genetic-redis — and the second step uses a pre-built image from DockerHub —  argoproj/argoexec:latest.  Submit the workflow using the Argo command line.

> argo submit nqueens.yaml

This launches the workflow and this can be visualized as a graph in Argo UI.

Once the workflow has run to completion, the solution is printed out to the console of the pod associated with the Job. A solution for a 14x14 chessboard is shown below.

Distributed Search

In the N-Queens problem, we learned how to create a workflow that launched parallel jobs on Kubernetes. The parallel jobs used an external Redis queue service for coordination. In this section, we look at another pattern for parallel workflow — distributed document search using the scatter/gather pattern. Here, the task is to search different words across a large database of documents for all documents that contain those words. To parallelize the task, we will scatter the different term requests across nodes in the cluster. All the nodes in the cluster have access to a shared volume hosting the documents to be searched. Then we gather all the responses from worker nodes into a single response. This task can be implemented in a workflow with the following steps.

1. When a request comes into search, parse the request and split the search string into words. 

2. Loop through the words and farm out a leaf pod to search for each word. 

3. Each of the pods returns a list of documents that match one of the words. Collate the search results and print the list of documents.

Argo Workflow

The Argo workflow YAML with the above steps is shown below. Here, we store the documents to be searched and the results of the search on persistent volumes mounted on the pods. 

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: dgrep-
spec:
  entrypoint: dgrep

  volumeClaimTemplates:
  - metadata:
      name: workdir
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 1Gi

  arguments:
    parameters:
    - name: searchstring
      value: hello world

  templates:
  - name: dgrep
    steps:
    - - name: data
        template: data
    - - name: generate
        template: generate
    - - name: search
        template: search
        arguments:
          parameters:
          - name: words
            value: "{{item}}"
        withParam: "{{steps.generate.outputs.result}}"
    - - name: collate
        template: collate

  - name: data
    container:
      image: alpine
      command: [sh, -c]
      args: ["touch /mnt/data/file1; touch /mnt/data/file2; echo -n hello > /mnt/data/file1; echo -n world > /mnt/data/file2"]
      volumeMounts:
      - name: workdir
        mountPath: /mnt/data

  - name: generate
    script:
      image: python:alpine3.6
      command: [python]
      source: |
        import json
        import sys
        json.dump([w for w in "{{workflow.parameters.searchstring}}".split()], sys.stdout)

  - name: search
    inputs:
      parameters:
      - name: words
    container:
      image: alpine
      command: [sh, -c]
      args: ["grep -rl {{inputs.parameters.words}} /mnt/data | awk -F/ '{ print $NF }' >> /mnt/data/output"]
      volumeMounts:
      - name: workdir
        mountPath: /mnt/data

  - name: collate
    container:
      image: alpine
      command: [sh, -c]
      args: ["cat /mnt/data/output | sort | uniq"]
      volumeMounts:
      - name: workdir
        mountPath: /mnt/data

Here we are using the vanilla Docker image alpine for all the steps. The workflow, when submitted using the Argo command line, can be visualized on the Argo UI. The search string consists of two words, "hello" and "world." In the response, two pods are created in parallel, each searching for a word. The individual search results from the pods are collated in the final step to print the list of documents.

Conclusion

We have illustrated how to create parallel workflows on Kubernetes with the help of two examples. We used Argo to create workflows that can be specified as a directed acyclic graph (DAG). Argo allows us to define a container-native workflow on Kubernetes where each step in the workflow is a Docker container. Kubernetes is ideal for running parallel workflows and Argo reduces the complexity of designing such workflows.