argo

package module
v0.0.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 13 Imported by: 0

README

workflow/argo

workflow/argo 是 Argo Workflows 的轻量封装，通过 Argo Server REST API 操作 Kubernetes 中的 Workflow 资源。模块没有引入 Argo 官方 Go SDK 或 client-go，适合只需要提交、查询、控制和读取日志的场景。

模块能力

  • 创建并关闭 Argo REST 客户端。
  • 提交 Workflow，支持命名空间覆盖、服务端 dry-run 和 key=value 参数追加。
  • 查询、列表、删除 Workflow。
  • 挂起、恢复、终止、重试、重新提交和停止 Workflow。
  • 获取整个 Workflow 或指定 Pod 的日志。
  • 提供轻量的 Argo Workflow JSON 结构体，覆盖 container、script、DAG、steps 和参数等常见字段。

安装

go get github.com/liujitcn/kratos-kit/workflow/argo@latest

客户端

client, err := argo.NewClient(argo.ClientOptions{
	ServerURL:         "https://localhost:2746",
	Namespace:         "default",
	Token:             "your-token",
	InsecureSkipVerify: true,
})
if err != nil {
	return err
}
defer func() { _ = client.Close() }()

Token 只填写 token 原文，模块会自动设置 Authorization: Bearer <token>。InsecureSkipVerify 仅建议本地开发或测试环境使用。

Workflow 操作

提交

SubmitWorkflow(ctx, wf, opts) 会把 Workflow 包装为 Argo Server 所需的提交请求。SubmitOptions.Parameters 会追加到 workflow.spec.arguments.parameters，不会修改调用方传入的 Workflow 对象。

wf, err := client.SubmitWorkflow(ctx, &argo.Workflow{
	APIVersion: "argoproj.io/v1alpha1",
	Kind:       "Workflow",
	Metadata: argo.ObjectMeta{
		GenerateName: "hello-",
	},
	Spec: argo.WorkflowSpec{
		Entrypoint: "main",
		Templates: []argo.Template{
			{
				Name: "main",
				Container: &argo.Container{
					Image:   "alpine:3.20",
					Command: []string{"sh", "-c"},
					Args:    []string{"echo hello"},
				},
			},
		},
	},
}, &argo.SubmitOptions{
	Parameters: []string{"message=hello"},
})
if err != nil {
	return err
}
查询与列表
wf, err := client.GetWorkflow(ctx, "hello-abcde", "")
list, err := client.ListWorkflows(ctx, &argo.ListOptions{
	LabelSelector: "workflows.argoproj.io/completed=true",
	Limit:         20,
})

第三个命名空间参数为空时使用 ClientOptions.Namespace。ListOptions.Namespace 可覆盖客户端默认命名空间。

生命周期
方法 说明
SuspendWorkflow(ctx, name, namespace) 挂起运行中的 Workflow
ResumeWorkflow(ctx, name, namespace) 恢复已挂起的 Workflow
TerminateWorkflow(ctx, name, namespace) 终止运行中的 Workflow
RetryWorkflow(ctx, name, namespace) 重试失败的 Workflow，并返回更新后的资源
ResubmitWorkflow(ctx, name, namespace) 基于已有 Workflow 重新提交一次执行
StopWorkflow(ctx, name, namespace, message) 停止 Workflow，可附带停止原因
DeleteWorkflow(ctx, name, namespace) 删除 Workflow
日志
logs, err := client.GetWorkflowLogs(ctx, "hello-abcde", "", "")
podLogs, err := client.GetWorkflowLogs(ctx, "hello-abcde", "", "hello-abcde-main")

podName 为空时读取 Workflow 级日志；非空时读取指定 Pod 日志。

配置

ClientOptions
字段 说明 默认值
ServerURL Argo Server API 地址 https://localhost:2746
Namespace 默认 Kubernetes 命名空间 default
Token Bearer token 原文 空
InsecureSkipVerify 是否跳过 TLS 证书校验 false
SubmitOptions
字段 说明 默认值
Namespace 提交目标命名空间，覆盖客户端默认值 空
ServerDryRun 只在服务端校验，不实际创建 Workflow false
Parameters key=value 格式参数，追加到 Workflow arguments 空
ListOptions
字段 说明
Namespace 查询目标命名空间，覆盖客户端默认值
LabelSelector Kubernetes label selector
FieldSelector Kubernetes field selector
Limit 返回数量上限
Offset 分页偏移

类型与状态

Phase 取值包括 Pending、Running、Succeeded、Failed、Error。可调用 phase.IsTerminal() 判断是否已经到达终态，其中 Succeeded、Failed、Error 为终态。

参考

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Arguments 

type Arguments struct {
	Parameters []Parameter `json:"parameters,omitempty"`
}

Arguments 表示 Workflow 或步骤参数。

type ClientOptions 

type ClientOptions struct {
	// ServerURL 是 Argo Server API 地址，例如 "https://localhost:2746"。
	ServerURL string

	// Namespace 是执行 Workflow 操作的 Kubernetes 命名空间。
	Namespace string

	// Token 是用于认证的 Bearer token，不需要包含 Bearer 前缀。
	Token string

	// InsecureSkipVerify 控制是否跳过 TLS 证书校验，仅建议开发环境使用。
	InsecureSkipVerify bool
}

ClientOptions 配置 Argo Workflows API 客户端。

type Container 

type Container struct {
	Image   string   `json:"image,omitempty"`
	Command []string `json:"command,omitempty"`
	Args    []string `json:"args,omitempty"`
}

Container 表示容器模板。

type DAGTask 

type DAGTask struct {
	Name         string    `json:"name,omitempty"`
	Template     string    `json:"template,omitempty"`
	Arguments    Arguments `json:"arguments,omitempty"`
	Dependencies []string  `json:"dependencies,omitempty"`
}

DAGTask 表示 DAG 模板中的任务。

type DAGTemplate 

type DAGTemplate struct {
	Tasks []DAGTask `json:"tasks,omitempty"`
}

DAGTemplate 表示 DAG 模板。

type ListOptions 

type ListOptions struct {
	// Namespace 是目标命名空间，会覆盖客户端默认命名空间。
	Namespace string

	// LabelSelector 按 Kubernetes label 过滤 Workflow。
	LabelSelector string

	// FieldSelector 按 Kubernetes field 过滤 Workflow。
	FieldSelector string

	// Limit 限制返回结果数量。
	Limit int64

	// Offset 是分页偏移量。
	Offset int64
}

ListOptions 表示查询 Workflow 列表时的过滤与分页参数。

type NodeStatus 

type NodeStatus struct {
	ID           string     `json:"id,omitempty"`
	Name         string     `json:"name,omitempty"`
	TemplateName string     `json:"templateName,omitempty"`
	Phase        Phase      `json:"phase,omitempty"`
	StartedAt    *time.Time `json:"startedAt,omitempty"`
	FinishedAt   *time.Time `json:"finishedAt,omitempty"`
	Message      string     `json:"message,omitempty"`
}

NodeStatus 表示 Workflow 中节点的状态。

type ObjectMeta 

type ObjectMeta struct {
	Name              string            `json:"name,omitempty"`
	GenerateName      string            `json:"generateName,omitempty"`
	Namespace         string            `json:"namespace,omitempty"`
	UID               string            `json:"uid,omitempty"`
	Labels            map[string]string `json:"labels,omitempty"`
	Annotations       map[string]string `json:"annotations,omitempty"`
	CreationTimestamp *time.Time        `json:"creationTimestamp,omitempty"`
}

ObjectMeta 表示 Kubernetes 对象元数据。

type ParallelSteps 

type ParallelSteps []Step

ParallelSteps 表示同一阶段内可并行执行的一组 steps。 Argo 原生 JSON 结构要求 steps 是二维数组：外层表示顺序阶段，内层表示并行步骤。

type Parameter 

type Parameter struct {
	Name  string `json:"name,omitempty"`
	Value string `json:"value,omitempty"`
}

Parameter 表示 Workflow 参数。

type Phase 

type Phase string

Phase 表示 Workflow 当前阶段。 

const (
	PhasePending   Phase = "Pending"
	PhaseRunning   Phase = "Running"
	PhaseSucceeded Phase = "Succeeded"
	PhaseFailed    Phase = "Failed"
	PhaseError     Phase = "Error"
)

func (Phase) IsTerminal 

func (p Phase) IsTerminal() bool

IsTerminal 判断当前阶段是否为终态。

type Script 

type Script struct {
	Image   string   `json:"image,omitempty"`
	Command []string `json:"command,omitempty"`
	Source  string   `json:"source,omitempty"`
}

Script 表示脚本模板。

type Step 

type Step struct {
	Name      string    `json:"name,omitempty"`
	Template  string    `json:"template,omitempty"`
	Arguments Arguments `json:"arguments,omitempty"`
}

Step 表示 steps 模板中的单个步骤。

type SubmitOptions 

type SubmitOptions struct {
	// Namespace 是目标命名空间，会覆盖客户端默认命名空间。
	Namespace string

	// ServerDryRun 要求服务端只校验请求，不实际创建 Workflow。
	ServerDryRun bool

	// Parameters 是 "key=value" 格式的 Workflow 参数。
	Parameters []string
}

SubmitOptions 表示提交 Workflow 时的可选参数。

type Template 

type Template struct {
	Name      string          `json:"name,omitempty"`
	Container *Container      `json:"container,omitempty"`
	Script    *Script         `json:"script,omitempty"`
	DAG       *DAGTemplate    `json:"dag,omitempty"`
	Steps     []ParallelSteps `json:"steps,omitempty"`
}

Template 表示 Workflow 模板。

type Workflow 

type Workflow struct {
	APIVersion string          `json:"apiVersion,omitempty"`
	Kind       string          `json:"kind,omitempty"`
	Metadata   ObjectMeta      `json:"metadata,omitempty"`
	Spec       WorkflowSpec    `json:"spec,omitempty"`
	Status     *WorkflowStatus `json:"status,omitempty"`
}

Workflow 表示 Argo Workflow 资源。

type WorkflowClient 

type WorkflowClient struct {
	// contains filtered or unexported fields
}

WorkflowClient 通过 REST API 提供 Argo Workflows 高层操作能力。

func NewClient 

func NewClient(opts ClientOptions) (*WorkflowClient, error)

NewClient 创建并初始化 Argo Workflows 客户端。

func (*WorkflowClient) Close 

func (wc *WorkflowClient) Close() error

Close 关闭空闲连接并标记客户端停止运行。

func (*WorkflowClient) DeleteWorkflow 

func (wc *WorkflowClient) DeleteWorkflow(ctx context.Context, name string, optsNamespace string) error

DeleteWorkflow 删除指定 Workflow。

func (*WorkflowClient) GetWorkflow 

func (wc *WorkflowClient) GetWorkflow(ctx context.Context, name string, optsNamespace string) (*Workflow, error)

GetWorkflow 根据名称获取 Workflow。

func (*WorkflowClient) GetWorkflowLogs 

func (wc *WorkflowClient) GetWorkflowLogs(ctx context.Context, name string, optsNamespace string, podName string) (string, error)

GetWorkflowLogs 获取工作流或指定 Pod 的日志。

func (*WorkflowClient) ListWorkflows 

func (wc *WorkflowClient) ListWorkflows(ctx context.Context, opts *ListOptions) (*WorkflowList, error)

ListWorkflows 列出指定命名空间内的 Workflow。

func (*WorkflowClient) ResubmitWorkflow 

func (wc *WorkflowClient) ResubmitWorkflow(ctx context.Context, name string, optsNamespace string) (*Workflow, error)

ResubmitWorkflow 基于已有 Workflow 重新提交一次执行。

func (*WorkflowClient) ResumeWorkflow 

func (wc *WorkflowClient) ResumeWorkflow(ctx context.Context, name string, optsNamespace string) error

ResumeWorkflow 恢复已挂起的 Workflow。

func (*WorkflowClient) RetryWorkflow 

func (wc *WorkflowClient) RetryWorkflow(ctx context.Context, name string, optsNamespace string) (*Workflow, error)

RetryWorkflow 重试失败的 Workflow。

func (*WorkflowClient) StopWorkflow 

func (wc *WorkflowClient) StopWorkflow(ctx context.Context, name string, optsNamespace string, message string) error

StopWorkflow 停止 Workflow，并可附带停止原因。

func (*WorkflowClient) SubmitWorkflow 

func (wc *WorkflowClient) SubmitWorkflow(ctx context.Context, wf *Workflow, opts *SubmitOptions) (*Workflow, error)

SubmitWorkflow 提交新的 Argo Workflow。 opts.Parameters 会追加到 workflow.spec.arguments.parameters，不会修改调用方传入的 Workflow。

func (*WorkflowClient) SuspendWorkflow 

func (wc *WorkflowClient) SuspendWorkflow(ctx context.Context, name string, optsNamespace string) error

SuspendWorkflow 挂起正在运行的 Workflow。

func (*WorkflowClient) TerminateWorkflow 

func (wc *WorkflowClient) TerminateWorkflow(ctx context.Context, name string, optsNamespace string) error

TerminateWorkflow 终止正在运行的 Workflow。

type WorkflowList 

type WorkflowList struct {
	Items []Workflow `json:"items,omitempty"`
}

WorkflowList 表示 Workflow 列表。

type WorkflowSpec 

type WorkflowSpec struct {
	Entrypoint         string     `json:"entrypoint,omitempty"`
	Templates          []Template `json:"templates,omitempty"`
	Arguments          Arguments  `json:"arguments,omitempty"`
	ServiceAccountName string     `json:"serviceAccountName,omitempty"`
}

WorkflowSpec 表示 Workflow 规格。

type WorkflowStatus 

type WorkflowStatus struct {
	Phase      Phase                 `json:"phase,omitempty"`
	StartedAt  *time.Time            `json:"startedAt,omitempty"`
	FinishedAt *time.Time            `json:"finishedAt,omitempty"`
	Message    string                `json:"message,omitempty"`
	Nodes      map[string]NodeStatus `json:"nodes,omitempty"`
}

WorkflowStatus 表示 Workflow 状态。

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.