IT 铁人赛 k8s 入门30天 -- day19 k8s Task Coarse Parallel Processing Using a Work Queue

前言

这个章节将要来实作 Coarse Parallel Processing Using a Work Queue 这个练习

布署目标

1 建立一个 Message Queue 的服务： 在这个范例使用的是 RabbitMq 也可以使用其他符合 AMPQ 特性的 QUEUE

2 建立一个 QUEUE 并且填入要发送的讯息进去, 每一个讯息代表必须要执行的 Job

3 透过 k8s 发布 Job 来执行从 QUEUE 接收过来的任务： 这个 Job 会开启多个 Pod, 每个 Pod 会从 QUEUE 拿出一个任务来执行, 不断重复直到 QUEUE 的任务都执行结束

建立一个 Message Queue 的服务

建立 rabbitmq-service.yaml 如下：

apiVersion: v1
kind: Service
metadata:
  labels:
    component: rabbitmq
  name: rabbitmq-service
spec:
  ports:
  - port: 5672
  selector:
    app: taskQueue
    component: rabbitmq

建制一个 Service 名称设定为 rabbitmq-service

设定开启的 Port 为 5672

建制指令如下：

kubectl apply -f rabbitmq-service.yaml

建立 rabbitmq-controller.yaml 如下：

apiVersion: v1
kind: ReplicationController
metadata:
  labels:
    component: rabbitmq
  name: rabbitmq-controller
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: taskQueue
        component: rabbitmq
    spec:
      containers:
      - image: rabbitmq
        name: rabbitmq
        ports:
        - containerPort: 5672
        resources:
          limits:
            cpu: 100m

建立一个 ReplicationController 名称设定为 rabbitmq-controller

设定 image 使用 rabbitmq

设定 containerPort 为 5672

设定 resources limit 为 cpu: 100m

设定 replicas 为 1

建制指令如下：

kubectl apply -f rabbitmq-controller.yaml

建立一个 QUEUE 并且填入要发送的讯息进去

首先需要再同一个 Cluster 内建立一个暂时的 Pod

并且安装 ampq 的操作工具

建立指令如下：

kubectl run -i --tty temp --image ubuntu:18.04

在操作的 terminal 输入以下指令来安装操作工具

apt-get update
apt-get install -y curl ca-certificates amqp-tools python dnsutils

察看 rabbitmq-service 的 IP

nslookup rabbitmq-service

开始建立测试 QUEUE

export BROKER_URL=amqp://guest:guest@rabbitmq-service:5672
## create a queue with name foo
/usr/bin/amqp-declare-queue --url=$BROKER_URL -q foo -d
## publish message to foo
/usr/bin/amqp-publish --url=$BROKER_URL -r foo -p -b Hello
## get the message from foo
/usr/bin/amqp-consume --url=$BROKER_URL -q foo -c 1 cat && echo

建立 QUEUE 并且输入讯息

/usr/bin/amqp-declare-queue --url=$BROKER_URL -q job1  -d
for f in apple banana cherry date fig grape lemon melon
do
  /usr/bin/amqp-publish --url=$BROKER_URL -r job1 -p -b $f
done

透过 k8s 发布 Job 来执行从 QUEUE 接收过来的任务

建立 image

建立 Dockerfile 如下：

# Specify BROKER_URL and QUEUE when running
FROM ubuntu:18.04

RUN apt-get update && \
    apt-get install -y curl ca-certificates amqp-tools python \
       --no-install-recommends \
    && rm -rf /var/lib/apt/lists/*
COPY ./worker.py /worker.py

CMD  /usr/bin/amqp-consume --url=$BROKER_URL -q $QUEUE -c 1 /worker.py

建立 worker.py 如下：

#!/usr/bin/env python

# Just prints standard out and sleeps for 10 seconds.
import sys
import time
print("Processing " + sys.stdin.readlines()[0])
time.sleep(10)

修改 worker.py 为可执行权限

chmod +x worker.py

建立 docker image with name job-wq-1

docker build -t job-wq-1 .

这边笔者使用 docker hub 做为 image registry 所以下面指令如下

docker tag job-wq-1 yuanyu90221/job-wq-1
docker push yuanyu90221/job-wq-1

成功 push 的话 就可以到 dockerhub 找到对应的 job-wq-1

建立 Job

建立 job.yaml 如下

apiVersion: batch/v1
kind: Job
metadata:
  name: job-wq-1
spec:
  completions: 8
  parallelism: 2
  template:
    metadata:
      name: job-wq-1
    spec:
      containers:
      - name: c
        image: docker.io/yuanyu90221/job-wq-1
        env:
        - name: BROKER_URL
          value: amqp://guest:guest@rabbitmq-service:5672
        - name: QUEUE
          value: job1
      restartPolicy: OnFailure

建立一个 Job 名称设定为 job-wq-1

completions 设定为 8 代表要执行 8 个 job 才完成

parallelism 设定为 2 代表一次可以启动 2 个 job

注意的是这边的 image 设定跟刚刚 push 上去的 image registry url 相关

像笔者是使用自己的 dockerbub 帐号, 所以使用 docker.io/yuanyu90221/job-wq-1

查询执行结果

kubectl describe jobs/job-wq-1

检验 Pod 内容

kubectl get pod
kubectl logs job-wq-1-zz729

後话

在此范例中, 我们使用 RabbitMq 来做中间的 Message Brokder

必须要透过 RabbitMq 来传递任务

而 k8s 官方有其他的 Job Pattern 可以设定来取代这个作法, 不需要去设定一个 Message Queue Service

而这个范例用一个 Pod 来跑一个 Task 其实太过消耗资源, 而可以使用另一个范例 这个范例是用一个 Pod 执行多个任务可以降低建立多个 Pod 所花费的时间

注意的是 这边 job 里设定的 completions 设定代表所要完成的项目数量

也就是如果 completions 少於 Queue 内的项目代表会有项目没被执行到

而 completions 多余 Queue 内的项目则即使执行完所有 QUEUE 的项目, Job 仍会继续建立 Pod 来等待QUEUE 新项目的出现