Argo Workflows ExecutorPlugin 详解从 CRD 模型到 workflow-level 插件实战【免费下载链接】argo-workflowsWorkflow Engine for Kubernetes项目地址: https://gitcode.com/gh_mirrors/ar/argo-workflowsExecutorPlugin 是 Argo Workflows 中用于扩展工作流执行能力的关键 API 类型它允许用户以边车sidecar容器 HTTP RPC的方式为 Workflow 提供自定义模板执行逻辑如对接 Slack、Tekton 等第三方系统。本文以IoArgoprojWorkflowV1alpha1ExecutorPlugin这一 Java SDK 生成的 API 文档为骨架结合仓库中的 CRD 源码、官方插件指南与完整示例系统讲解该 API 的对象结构、字段语义、启用方式、RPC 契约、故障处理与发布流程帮助你掌握如何在 Argo Workflows 中编写、配置和运维属于自己的 Executor Plugin。一、ExecutorPlugin API 模型总览IoArgoprojWorkflowV1alpha1ExecutorPlugin是 Argo Workflows Java SDKswagger-codegen 生成中对应 Kubernetes CRDExecutorPlugin的数据模型文档对它的定位只有一句话ExecutorPlugin describes workflow-level executor pluginExecutorPlugin 描述工作流级别的执行器插件。它属于argoproj.io/v1alpha1API 组其 Go 侧定义位于 pkg/apis/workflow/v1alpha1/workflow_types.go// ExecutorPlugin describes workflow-level executor plugin type ExecutorPlugin struct { metav1.ObjectMeta json:metadata protobuf:bytes,2,opt,namemetadata Spec ExecutorPluginSpec json:spec protobuf:bytes,1,opt,namespec } type ExecutorPluginSpec struct { Sidecar ExecutorPluginSidecar json:sidecar protobuf:bytes,1,opt,namesidecar } type ExecutorPluginSidecar struct { // AutomountServiceAccountToken enables mounting the service account token. // The service account must be named plugin-name-executor-plugin. AutomountServiceAccountToken bool json:automountServiceAccountToken,omitempty protobuf:varint,1,opt,nameautomountServiceAccountToken // Container defines the Kubernetes container specification for the sidecar. Container apiv1.Container json:container protobuf:bytes,2,opt,namecontainer }Java SDK 文档以属性表形式呈现了该模型的完整字段两个字段与源码一一对应NameType说明metadataio.kubernetes.client.openapi.models.V1ObjectMeta标准的 Kubernetes 对象元数据名称、标签、注解等与源码中的metav1.ObjectMeta对应specIoArgoprojWorkflowV1alpha1ExecutorPluginSpec插件规格定义插件边车的运行方式说明io.kubernetes.client.openapi.models.V1ObjectMeta与IoArgoprojWorkflowV1alpha1ExecutorPluginSpec的 Java 文档同样位于sdks/java/client/docs/目录下分别对应 Go 源码中的metav1.ObjectMeta与ExecutorPluginSpec。1. spec.sidecar插件的核心执行载体IoArgoprojWorkflowV1alpha1ExecutorPluginSpec只有一个必填字段sidecar其类型为IoArgoprojWorkflowV1alpha1ExecutorPluginSidecarJava 文档给出的字段如下NameTypeDescriptionNotesautomountServiceAccountTokenBooleanAutomountServiceAccountToken enables mounting the service account token. The service account must be named plugin-name-executor-plugin.[optional]containerio.kubernetes.client.openapi.models.V1Container两个字段的语义对应 workflow_types.goautomountServiceAccountToken可选是否将 ServiceAccount Token 挂载进插件边车。若启用则要求集群中存在名为plugin-name-executor-plugin的 ServiceAccount——也就是说插件名metadata.name直接参与 ServiceAccount 的命名约定这是插件需要访问 Kubernetes API 或第三方系统凭证时的常见做法。container必填标准的 Kubernetes 容器定义apiv1.Container即V1Container用于描述插件 HTTP 服务器所在边车的镜像、命令、端口、资源限制、安全上下文、环境变量等全部细节。2. 两种使用形态全局 ConfigMap 与 workflow 级 spec从源码结构看ExecutorPlugin存在两个使用入口全局插件以ExecutorPluginCRD 形式定义最终以 ConfigMap 形式存放在 Controller 的配置中标签workflows.argoproj.io/configmap-typeExecutorPlugin。工作流级插件直接写在 Workflow 的spec.executorPlugins字段中类型为[]ExecutorPlugin定义见 workflow_types.go// Specifies executor plugins at the workflow level. // // This field is effective only when the ARGO_WORKFLOW_LEVEL_EXECUTOR_PLUGINS // feature gate is enabled. // // If this field contains one or more executor plugins, executor plugin // settings from the controller ConfigMap are ignored. // // If this field is empty or not set, the controller falls back to the // ConfigMap configuration. ExecutorPlugins []ExecutorPlugin json:executorPlugins,omitempty protobuf:bytes,44,rep,nameexecutorPlugins源码注释明确了三件事该字段仅在ARGO_WORKFLOW_LEVEL_EXECUTOR_PLUGINS特性开关开启时生效一旦字段非空Controller ConfigMap 中的全局插件配置对本工作流将被忽略字段为空时则回退到全局配置。这正是工作流级插件优先的覆盖规则。二、启用 ExecutorPlugin 功能开关Executor Plugins 默认是禁用的需要显式开启两个开关相互独立见 docs/executor_plugins.mdARGO_EXECUTOR_PLUGINStrue启用定义在 Controller ConfigMap 中的全局 Executor Plugins。ARGO_WORKFLOW_LEVEL_EXECUTOR_PLUGINStrue允许使用直接写在 Workflow spec 中的插件设置且这些设置优先于全局配置。两个开关可以只开其一也可以同时开启。在 Workflow Controller 的 Deployment 中启用两者的配置如下apiVersion: apps/v1 kind: Deployment metadata: name: workflow-controller spec: template: spec: containers: - name: workflow-controller env: - name: ARGO_EXECUTOR_PLUGINS value: true - name: ARGO_WORKFLOW_LEVEL_EXECUTOR_PLUGINS value: true如果使用 Helm Chart 安装则在values.yaml中增加controller: extraEnv: - name: ARGO_EXECUTOR_PLUGINS value: true - name: ARGO_WORKFLOW_LEVEL_EXECUTOR_PLUGINS value: true三、实战一个工作流级 ExecutorPlugin 的完整示例仓库在 examples/workflow-level-executor-plugin.yaml 中提供了一个可直接运行的 workflow 级插件示例它同时展示了IoArgoprojWorkflowV1alpha1ExecutorPlugin的metadata与spec两个字段如何落地。核心结构如下apiVersion: argoproj.io/v1alpha1 kind: Workflow metadata: generateName: workflow-level-executor-plugin- labels: workflows.argoproj.io/no-test: environment spec: entrypoint: main executorPlugins: - metadata: name: workflow-level-hello-executor-plugin spec: sidecar: container: name: workflow-level-hello-executor-plugin image: python:alpine3.23 command: - python - -c args: - | 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() if workflowLevelHello in args[template].get(plugin, {}): self.reply({ node: { phase: Succeeded, message: Hello from a workflow-level executor plugin! } }) else: self.reply({}) else: self.unsupported() if __name__ __main__: httpd HTTPServer((, 4356), Plugin) httpd.serve_forever() ports: - containerPort: 4356 resources: limits: cpu: 200m memory: 64Mi requests: cpu: 100m memory: 32Mi securityContext: allowPrivilegeEscalation: false capabilities: drop: - ALL readOnlyRootFilesystem: true runAsNonRoot: true templates: - name: main plugin: workflowLevelHello: {}示例要点插件边车是一个标准的container定义运行一个内嵌在args中的 Python HTTP 服务器监听 4356 端口——与官方指南建议一致自用插件建议使用大于 10000 的端口避免与常用端口冲突。插件的名字是workflow-level-hello-executor-plugin而模板通过plugin.workflowLevelHello引用它插件通过检查请求体中template.plugin是否包含该 key 来决定是否认领这个模板。边车遵循了资源请求/限制与安全上下文的强制要求runAsNonRoot: true、drop ALL capabilities、只读根文件系统。运行该示例的前提是 Controller 以ARGO_WORKFLOW_LEVEL_EXECUTOR_PLUGINStrue启动。四、ExecutorPlugin 的 RPC 契约/api/v1/template.execute插件边车本质上是运行在 Agent Pod 内的 HTTP 服务通过 RPC 风格接口响应执行器的调用。完整的 API 契约见 docs/executor_swagger.md其中与本主题直接相关的端点与数据模型如下端点POST /api/v1/template.execute请求体ExecuteTemplateArgs两个必填字段NameType说明templateTemplate待执行的模板workflowWorkflow当前 Workflow 的完整对象响应体ExecuteTemplateReplyNameType说明nodeNodeResult节点执行结果requeueDuration重新入队时间用于异步长任务场景NodeResult字段NameType说明messagestring节点消息outputsOutputs节点输出参数/工件phaseNodePhase节点阶段如Succeeded、Failed、Running、PendingprogressProgress进度从实现侧看仓库中的 workflow/executor/plugins/rpc/plugin.go 正是该契约的 Go 客户端实现type plugin struct{ rpc.Client } func New(address, token string) executorplugins.TemplateExecutor { return plugin{Client: rpc.New(address, token, 30*time.Second, wait.Backoff{ Duration: time.Second, Jitter: 0.2, Factor: 2, Steps: 5, })} } func (p *plugin) ExecuteTemplate(ctx context.Context, args executorplugins.ExecuteTemplateArgs, reply *executorplugins.ExecuteTemplateReply) error { return p.Call(ctx, template.execute, args, reply) }可以看到执行器以 30 秒为超时、指数退避初始 1 秒、抖动 0.2、倍增因子 2、最多 5 步调用template.execute方法方法名即 URL 路径/api/v1/template.execute。在编写插件时需要注意只需要实现你需要的方法返回 404 表示该插件不支持此 RPC 方法之后不会被再次调用。路径就是 RPC 方法名self.path /api/v1/template.execute即对应ExecuteTemplate。必须校验鉴权请求头Authorization应包含与/var/run/argo/token相同的值不匹配时返回 403。响应体可携带节点结果包括phase如Succeeded/Failed、message以及outputs可返回参数如示例中的{name: foo, value: bar}。返回{}表示不认领例如插件是 Slack 插件但当前模板是 Tekton 任务则返回空对象表示无法执行该 Plugin 模板。五、插件发现、密钥注入与安全约束发现机制工作流运行时插件从两个位置加载Workflow 所在的命名空间Argo 安装命名空间通常是argo。如果两个命名空间中存在同名插件则只加载 Workflow 所在命名空间中的那个工作流命名空间优先。Secrets 注入插件要对接第三方系统时不要把密钥写死在plugin.yaml的spec中应通过 Kubernetes Secret 注入环境变量spec: sidecar: container: env: - name: URL valueFrom: secretKeyRef: name: slack-executor-plugin key: URL这样既避免了密钥泄露到清单文件也便于轮换与审计。automountServiceAccountToken字段则用于按需把plugin-name-executor-pluginServiceAccount 的 Token 挂载进边车供插件访问 Kubernetes API 或云厂商凭证使用。资源与安全上下文强制项为了让任何人都无法创建内存无上限或以 root 运行的恶意插件资源请求/限制与安全上下文是强制校验的官方指南明确为 mandatoryspec: sidecar: container: resources: requests: cpu: 100m memory: 32Mi limits: cpu: 200m memory: 64Mi securityContext: runAsNonRoot: true runAsUser: 1000六、故障语义与重试行为官方文档 docs/executor_plugins.md 对插件故障做了明确分级故障类型语义连接/套接字错误视为**瞬时transient**错误会重试超时Timeout视为瞬时错误会重试404方法不被插件支持同一次工作流内不再调用该方法503视为瞬时错误会重试其他 4xx/5xx视为**致命fatal**错误瞬时错误会按指数退避策略重试对应plugin.go中的wait.Backoff配置致命错误则会导致步骤失败。重新入队Re-Queue异步长任务模式当插件无法立即完成工作例如启动了一个长时间运行的外部任务时可以在响应中返回Running/Pending阶段并附带requeue时间让执行器稍后再次调用{ node: { phase: Running, message: Long-running task started }, requeue: 2m }该示例中任务会在 2 分钟后被重新入队届时template.execute会被再次调用——这正是ExecuteTemplateReply.requeue字段的用途。七、调试、列出与发布插件查看插件日志插件运行在 Agent Pod 的边车中直接查看对应容器日志即可kubectl -n argo logs ${agentPodName} -c hello-executor-plugin列出已安装的插件全局插件本质上是 ConfigMap可用标签过滤kubectl get cm -l workflows.argoproj.io/configmap-typeExecutorPlugin构建与安装以官方 Python 示例hello插件为例使用 CLI 将插件构建为 ConfigMap 清单再kubectl apply安装argo executor-plugin build . kubectl -n argo apply -f hello-executor-plugin-configmap.yaml安装成功后 Controller 日志会出现levelinfo msgExecutor plugin added namehello-controller-plugin随后即可运行引用该插件的 Workflow。发布共享如果你希望将插件分享给社区可参考 docs/plugin-directory.md 中的插件目录并提交 Pull Request 将你的插件加入其中。八、小结IoArgoprojWorkflowV1alpha1ExecutorPlugin虽然在 Java SDK 文档中只是一个两字段的属性表但它在 Argo Workflows 中承担着插件即边车、边车即 RPC 服务的完整扩展机制metadata提供插件身份与命名还参与 ServiceAccount 的命名约定spec.sidecar.container承载插件实际运行逻辑二者共同支撑起工作流级spec.executorPlugins与全局ConfigMap两种插件形态。结合 workflow_types.go 的类型定义、executor_swagger.md 的 RPC 契约、workflow/executor/plugins/rpc/plugin.go 的客户端实现以及 examples/workflow-level-executor-plugin.yaml 的完整示例你可以据此快速上手开启特性开关 → 编写 HTTP 边车 → 配置插件清单 → 在工作流模板中以plugin字段引用从而把任意外部系统无缝接入 Argo Workflows 的执行链路。【免费下载链接】argo-workflowsWorkflow Engine for Kubernetes项目地址: https://gitcode.com/gh_mirrors/ar/argo-workflows创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
