argo/client/workflow.go

package client

import (
	"github.com/argoproj/argo-workflows/v3/pkg/apiclient/workflow"
	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	"github.com/pkg/errors"
)

type CreateWorkflowParams struct {
	Namespace          string
	WorkflowDefinition string
}

// CreateWorkflow 创建工作流
func (c *Client) CreateWorkflow(params CreateWorkflowParams) (string, error) {
	wf := new(v1alpha1.Workflow)
	err := unmarshalYamlToJsonStruct([]byte(params.WorkflowDefinition), wf)
	if err != nil {
		return "", err
	}

	response, err := c.workflowService.CreateWorkflow(c.ctx, &workflow.WorkflowCreateRequest{
		Namespace: params.Namespace,
		Workflow:  wf,
	})
	if err != nil {
		return "", errors.New(err.Error())
	}

	return response.Name, nil
}

type DeleteWorkflowParams struct {
	Namespace string
	Name      string
}

// DeleteWorkflow 删除工作流
func (c *Client) DeleteWorkflow(params DeleteWorkflowParams) error {
	_, err := c.workflowService.DeleteWorkflow(c.ctx, &workflow.WorkflowDeleteRequest{
		Namespace: params.Namespace,
		Name:      params.Name,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type GetWorkflowsInNamespaceParams struct {
	Namespace string
}

// GetWorkflowsInNamespace 获取命名空间下的工作流
func (c *Client) GetWorkflowsInNamespace(params GetWorkflowsInNamespaceParams) (v1alpha1.Workflows, error) {
	response, err := c.workflowService.ListWorkflows(c.ctx, &workflow.WorkflowListRequest{
		Namespace: params.Namespace,
	})
	if err != nil {
		return nil, errors.New(err.Error())
	}

	return response.Items, nil
}

type GetWorkflowParams struct {
	Namespace string
	Name      string
}

// GetWorkflow 获取指定的工作流
func (c *Client) GetWorkflow(params GetWorkflowParams) (*v1alpha1.Workflow, error) {
	response, err := c.workflowService.GetWorkflow(c.ctx, &workflow.WorkflowGetRequest{
		Namespace: params.Namespace,
		Name:      params.Name,
	})
	if err != nil {
		return nil, errors.New(err.Error())
	}

	return response, nil
}

type LintWorkflowParams struct {
	Namespace          string
	WorkflowDefinition string
}

// LintWorkflow 检查工作流定义语法
func (c *Client) LintWorkflow(params LintWorkflowParams) error {
	wf := new(v1alpha1.Workflow)
	err := unmarshalYamlToJsonStruct([]byte(params.WorkflowDefinition), wf)
	if err != nil {
		return err
	}

	_, err = c.workflowService.LintWorkflow(c.ctx, &workflow.WorkflowLintRequest{
		Namespace: params.Namespace,
		Workflow:  wf,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type SubmitWorkflowFromWorkflowTemplateParams struct {
	Namespace    string
	TemplateName string
	Parameters   []string
}

// SubmitWorkflowFromWorkflowTemplate 基于工作流模板提交工作流
func (c *Client) SubmitWorkflowFromWorkflowTemplate(params SubmitWorkflowFromWorkflowTemplateParams) (string, error) {
	response, err := c.workflowService.SubmitWorkflow(c.ctx, &workflow.WorkflowSubmitRequest{
		Namespace:    params.Namespace,
		ResourceKind: "WorkflowTemplate",
		ResourceName: params.TemplateName,
		SubmitOptions: &v1alpha1.SubmitOpts{
			Parameters: params.Parameters,
		},
	})
	if err != nil {
		return "", errors.New(err.Error())
	}

	return response.Name, nil
}

type ResubmitWorkflowParams struct {
	Namespace          string
	Name               string
	Memoized           bool
	ResubmitParameters []string
}

// ResubmitWorkflow 重提交工作流
// 有三种方式可以用来重复提交(可以结合使用)：重新运行，基于缓存的和传递重提交参数
// 基于缓存的必须在Error和Failed的工作流上才可以使用
func (c *Client) ResubmitWorkflow(params ResubmitWorkflowParams) (string, error) {
	response, err := c.workflowService.ResubmitWorkflow(c.ctx, &workflow.WorkflowResubmitRequest{
		Namespace:  params.Namespace,
		Name:       params.Name,
		Memoized:   params.Memoized,
		Parameters: params.ResubmitParameters,
	})
	if err != nil {
		return "", errors.New(err.Error())
	}

	return response.Name, nil
}

type RetryWorkflowParams struct {
	Namespace                      string
	Name                           string
	RetryParameters                []string
	RetrySuccessfulWorkflow        bool
	RetryWorkflowNodeFieldSelector string
}

// RetryWorkflow 重新运行工作流(默认只能失败的工作流上重新运行)
func (c *Client) RetryWorkflow(params RetryWorkflowParams) error {
	_, err := c.workflowService.RetryWorkflow(c.ctx, &workflow.WorkflowRetryRequest{
		Namespace:         params.Namespace,
		Name:              params.Name,
		RestartSuccessful: params.RetrySuccessfulWorkflow,
		NodeFieldSelector: params.RetryWorkflowNodeFieldSelector,
		Parameters:        params.RetryParameters,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type StopWorkflowParams struct {
	Namespace         string
	Name              string
	NodeFieldSelector string
	Message           string
}

// StopWorkflow 终止工作流运行，会调用所有的退出处理器
func (c *Client) StopWorkflow(params StopWorkflowParams) error {
	_, err := c.workflowService.StopWorkflow(c.ctx, &workflow.WorkflowStopRequest{
		Namespace:         params.Namespace,
		Name:              params.Name,
		NodeFieldSelector: params.NodeFieldSelector,
		Message:           params.Message,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type TerminateWorkflowParams struct {
	Namespace string
	Name      string
}

// TerminateWorkflow 终止工作流运行，不调用所有的退出处理器
func (c *Client) TerminateWorkflow(params TerminateWorkflowParams) error {
	_, err := c.workflowService.TerminateWorkflow(c.ctx, &workflow.WorkflowTerminateRequest{
		Namespace: params.Namespace,
		Name:      params.Name,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type SetWorkflowParams struct {
	Namespace         string
	Name              string
	NodeFieldSelector string
	OutputParameters  string
	Message           string
	Phase             string
}

// SetWorkflow 设置工作流的参数
func (c *Client) SetWorkflow(params SetWorkflowParams) error {
	_, err := c.workflowService.SetWorkflow(c.ctx, &workflow.WorkflowSetRequest{
		Namespace:         params.Namespace,
		Name:              params.Name,
		NodeFieldSelector: params.NodeFieldSelector,
		Message:           params.Message,
		Phase:             params.Phase,
		OutputParameters:  params.OutputParameters,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type SuspendWorkflowParams struct {
	Namespace string
	Name      string
}

// SuspendWorkflow 挂起工作流
func (c *Client) SuspendWorkflow(params SuspendWorkflowParams) error {
	_, err := c.workflowService.SuspendWorkflow(c.ctx, &workflow.WorkflowSuspendRequest{
		Namespace: params.Namespace,
		Name:      params.Name,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}

type ResumeWorkflowParams struct {
	Namespace         string
	Name              string
	NodeFieldSelector string
}

// ResumeWorkflow 恢复被挂起的工作流
func (c *Client) ResumeWorkflow(params ResumeWorkflowParams) error {
	_, err := c.workflowService.ResumeWorkflow(c.ctx, &workflow.WorkflowResumeRequest{
		Namespace:         params.Namespace,
		Name:              params.Name,
		NodeFieldSelector: params.NodeFieldSelector,
	})
	if err != nil {
		return errors.New(err.Error())
	}

	return nil
}