在 Argo workflow 中使用插件减少并行 Pod 数量

在之前写过的使用 Argo workflow 调用公有云客户端软件实现运维过程的文章中,可以看到,使用 Argo workflow 的容器模板,简单的将既有运维能力容器化,就能使用 Argo workflow 对这些能力进行编排了。

不过近期一个测试中,遇到个小麻烦——在一个 With 循环里,我输入了 500 个任务,结果是 6 节点 CCE 集群爆满,流程卡住——集群规模的事情很简单,我直接将 Argo workflow 部署到 CCE Autopilot 集群中,随着流程启动,Auto pilot 集群非常给力,不到一分钟就扩容到了上百节点。然而新的问题出现了,Argo workflow 容器模板使用的镜像托管在 quay.io 上,我被限流了——无法拉取镜像,工作流自然也就无法执行了。

如果说必须要限流的话,Argo workflow 提供了多种机制,在不同粒度上对工作流的并发进行控制:

  1. 在模板中,使用 parallelism 参数,限制流程实例内的并发数。
  2. 在 Workflow Controller 的 Configmap(workflow-controller-configmap)中,使用 parallelism 或者 namespaceParallelism,在集群范围内,限制总体并发的流程数量。
  3. 模板中使用 synchronization,使用同样的共享锁的流程实例将会被有效限流。

不难看出,在有限集群的规模下,通过对并发的控制,以及垃圾回收策略的定义,都能有效的限制集群规模——毕竟上百节点是要花不少银子的。在这种情况下,还有一条路就是,使用执行插件。例如如下工作流:

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: python-example-
spec:
  entrypoint: main
  arguments:
    parameters:
      - name: value
        value: "1"  
  templates:
    - name: main
      steps:
      - - name: evaluate
          template: evaluate
          arguments:
            parameters:
              - name: value
                value: "{{workflow.parameters.value}}"
          withSequence:
            count: "50"         
    - name: evaluate
      inputs:
        parameters:
          - name: value    
      plugin:
        python:
          expression: |
            {"sum": int(parameters["value"]) + 1}

这里使用 plugin.python 的方式引用了一个插件,执行时,循环了 50 次,提交后,我们会发现,这里只执行了一个 Pod:python-example-hlc5t-1340600742-agent,也就是说,这一个 Pod 承载了所有的 50 个任务。如何实现的呢?这里就要看看 Argo workflow 的插件机制了。

Argo workflow 默认是不启用插件的,要启用插件,需要给控制器加入环境变量:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: workflow-controller
spec:
  template:
    spec:
      containers:
        - name: workflow-controller
          env:
            - name: ARGO_EXECUTOR_PLUGINS
              value: "true"

重启后,就可以启用上面工作流引用的插件了,启用插件的方式很有意思,提交一个 Configmap 即可:

# This is an auto-generated file. DO NOT EDIT
apiVersion: v1
data:
  sidecar.container: |
    args:
....
kind: ConfigMap
metadata:
...
    workflows.argoproj.io/version: '>= v3.3'
  creationTimestamp: null
  labels:
    workflows.argoproj.io/configmap-type: ExecutorPlugin
  name: python-executor-plugin

这方式有点奇怪,Configmap 里面包含了一堆 Python 代码。以及似乎是 Sidecar 的容器定义。应用之后,就能够运行上述工作流了。

注意 Configmap 中的注释说明:这是一个自动生成的文件,哪里来的呢?

实际上,Argo workflow 插件是由 argo executor-plugin build 命令构建出来的,一个插件的原始文件主要包含三个部分:

  1. 插件清单(plugin.yaml):这里实际上是对一个容器的定义,其中包含了容器镜像、资源使用等。
  2. 启动文件:一个命名为 server.* 的文本文件,可以是 Shell 或者 Python 脚本,他会在插件启动时被执行。
  3. 插件镜像:上述文本文件可能无法描述一些业务逻辑,因此,可以将二进制文件封装到镜像里,给启动文件调用。

例如前边用到的 Python 插件的 plugin.yaml

kind: ExecutorPlugin
apiVersion: argoproj.io/v1alpha1
metadata:
  name: python
...
    workflows.argoproj.io/version: '>= v3.3'
spec:
  sidecar:
    container:
      command:
        - python
        - -c
      image: python:alpine
      name: python-executor-plugin
...

不难看出,这个定义和上边的 Configmap 是一致的。再看看 server.py

import json
from http.server import BaseHTTPRequestHandler, HTTPServer


class Plugin(BaseHTTPRequestHandler):

    def args(self):
        return json.loads(self.rfile.read(int(self.headers.get('Content-Length'))))

    def reply(self, reply):
        self.send_response(200)
        self.end_headers()
        self.wfile.write(json.dumps(reply).encode("UTF-8"))

    def unsupported(self):
        self.send_response(404)
        self.end_headers()

    def do_POST(self):
        if self.path == '/api/v1/template.execute':
            args = self.args()

            template = args['template']
            plugin = template.get('plugin', {})

            if 'python' in plugin:
                spec = plugin['python']

                # convert parameters into easy to use dict
                # artifacts are not supported
                parameters = {}
                for parameter in template.get('inputs', {}).get('parameters', []):
                    parameters[parameter['name']] = parameter['value']

                try:
                    code = compile(spec['expression'], "<string>", "eval")
...


if __name__ == '__main__':
    httpd = HTTPServer(('', 7984), Plugin)
    httpd.serve_forever()

上边的代码,不难看出,这里只是启动了一个简单的 Python HTTP Server,监听 /api/v1/template.execute 的 Post 请求,并对其进行处理。

上述的 YAML 和启动代码都编写完成之后,就可以使用 argo executor-plugin build 命令来构建 Configmap 了。

当然也可以使用自己定义的基础镜像。

Avatar
崔秀龙

简单,是大师的责任;我们凡夫俗子,能做到清楚就很不容易了。

comments powered by Disqus
上一页