argo/client/workflow.go

package client

import (
	"github.com/pkg/errors"
	"net/http"
)

const (
	createWorkflowRelativeUrl          = "/api/v1/workflows/{namespace}"
	deleteWorkflowRelativeUrl          = "/api/v1/workflows/{namespace}/{name}"
	getWorkflowsInNamespaceRelativeUrl = "/api/v1/workflows/{namespace}"
	getWorkflowRelativeUrl             = "/api/v1/workflows/{namespace}/{name}"
	lintWorkflowRelativeUrl            = "/api/v1/workflows/{namespace}/lint"
	submitWorkflowRelativeUrl          = "/api/v1/workflows/{namespace}/submit"
	resubmitWorkflowRelativeUrl        = "/api/v1/workflows/{namespace}/{name}/resubmit"
	retryWorkflowRelativeUrl           = "/api/v1/workflows/{namespace}/{name}/retry"
	stopWorkflowRelativeUrl            = "/api/v1/workflows/{namespace}/{name}/stop"
	terminateWorkflowRelativeUrl       = "/api/v1/workflows/{namespace}/{name}/terminate"
	setWorkflowRelativeUrl             = "/api/v1/workflows/{namespace}/{name}/set"
	suspendWorkflowRelativeUrl         = "/api/v1/workflows/{namespace}/{name}/suspend"
	resumeWorkflowRelativeUrl          = "/api/v1/workflows/{namespace}/{name}/resume"
	getWorkflowEventsStreamRelativeUrl = "/api/v1/workflow-events/{namespace}"
)

type CreateWorkflowParams struct {
	Namespace          string
	WorkflowDefinition map[string]any
}

// CreateWorkflow 创建工作流
func (c *Client) CreateWorkflow(params CreateWorkflowParams) (string, error) {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
		}).
		SetBody(map[string]any{
			"namespace": params.Namespace,
			"workflow":  params.WorkflowDefinition,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Post(createWorkflowRelativeUrl)
	if err != nil {
		return "", errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		metadata, ok := responseMap["metadata"]
		if !ok {
			return "", errors.New("metadata为空")
		}

		metadataMap, ok := metadata.(map[string]any)
		if !ok {
			return "", errors.New("metadata不是map")
		}

		workflowName, ok := metadataMap["name"]
		if !ok {
			return "", errors.New("metadata中没有工作流名称")
		}

		workflowNameStr, ok := workflowName.(string)
		if !ok {
			return "", errors.New("工作流名称不是字符串")
		}

		return workflowNameStr, nil
	case http.StatusConflict:
		return "", errors.New("工作流已存在")
	default:
		message, ok := responseMap["message"]
		if !ok {
			return "", errors.Errorf("%v", resp.Status())
		}

		return "", errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type DeleteWorkflowParams struct {
	Namespace string
	Name      string
}

// DeleteWorkflow 删除工作流
func (c *Client) DeleteWorkflow(params DeleteWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Delete(deleteWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type GetWorkflowsInNamespaceParams struct {
	Namespace string
}

// GetWorkflowsInNamespace 获取命名空间下的工作流
func (c *Client) GetWorkflowsInNamespace(params GetWorkflowsInNamespaceParams) ([]map[string]any, error) {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Get(getWorkflowsInNamespaceRelativeUrl)
	if err != nil {
		return nil, errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		itemsValue, ok := responseMap["items"]
		if !ok {
			return nil, errors.New("没有获取到items参数")
		}

		items, ok := itemsValue.([]any)
		if !ok {
			return nil, errors.New("items不是slice")
		}

		templateDefinitions := make([]map[string]any, len(items))
		for i, item := range items {
			templateDefinition, ok := item.(map[string]any)
			if !ok {
				return nil, errors.New("item无法转换为map[string]any")
			}

			templateDefinitions[i] = templateDefinition
		}

		return templateDefinitions, nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return nil, errors.Errorf("%v", resp.Status())
		}

		return nil, errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type GetWorkflowParams struct {
	Namespace string
	Name      string
}

// GetWorkflow 获取指定的工作流
func (c *Client) GetWorkflow(params GetWorkflowParams) (map[string]any, error) {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Get(getWorkflowRelativeUrl)
	if err != nil {
		return nil, errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return responseMap, nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return nil, errors.Errorf("%v", resp.Status())
		}

		return nil, errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type LintWorkflowParams struct {
	Namespace          string
	WorkflowDefinition map[string]any
}

// LintWorkflow 检查工作流定义语法
func (c *Client) LintWorkflow(params LintWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
		}).
		SetBody(map[string]any{
			"namespace": params.Namespace,
			"workflow":  params.WorkflowDefinition,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Post(lintWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

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

// SubmitWorkflowFromWorkflowTemplate 基于工作流模板提交工作流
func (c *Client) SubmitWorkflowFromWorkflowTemplate(params SubmitWorkflowFromWorkflowTemplateParams) (string, error) {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
		}).
		SetBody(map[string]any{
			"namespace":    params.Namespace,
			"resourceKind": "WorkflowTemplate",
			"resourceName": params.TemplateName,
			"submitOptions": map[string]any{
				"parameters": params.Parameters,
			},
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Post(submitWorkflowRelativeUrl)
	if err != nil {
		return "", errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		metadata, ok := responseMap["metadata"]
		if !ok {
			return "", errors.New("metadata为空")
		}

		metadataMap, ok := metadata.(map[string]any)
		if !ok {
			return "", errors.New("metadata不是map")
		}

		workflowName, ok := metadataMap["name"]
		if !ok {
			return "", errors.New("metadata中没有工作流名称")
		}

		workflowNameStr, ok := workflowName.(string)
		if !ok {
			return "", errors.New("工作流名称不是字符串")
		}

		return workflowNameStr, nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return "", errors.Errorf("%v", resp.Status())
		}

		return "", errors.Errorf("%v, %v", resp.Status(), message)
	}
}

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

// ResubmitWorkflow 重提交工作流
// 有三种方式可以用来重复提交(可以结合使用)：重新运行，基于缓存的和传递重提交参数
// 基于缓存的必须在Error和Failed的工作流上才可以使用
func (c *Client) ResubmitWorkflow(params ResubmitWorkflowParams) (string, error) {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace":  params.Namespace,
			"name":       params.Name,
			"memoized":   params.Memoized,
			"parameters": params.ResubmitParameters,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(resubmitWorkflowRelativeUrl)
	if err != nil {
		return "", errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		metadata, ok := responseMap["metadata"]
		if !ok {
			return "", errors.New("metadata为空")
		}

		metadataMap, ok := metadata.(map[string]any)
		if !ok {
			return "", errors.New("metadata不是map")
		}

		workflowName, ok := metadataMap["name"]
		if !ok {
			return "", errors.New("metadata中没有工作流名称")
		}

		workflowNameStr, ok := workflowName.(string)
		if !ok {
			return "", errors.New("工作流名称不是字符串")
		}

		return workflowNameStr, nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return "", errors.Errorf("%v", resp.Status())
		}

		return "", errors.Errorf("%v, %v", resp.Status(), message)
	}
}

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

// RetryWorkflow 重新运行工作流(默认只能失败的工作流上重新运行)
func (c *Client) RetryWorkflow(params RetryWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace":         params.Namespace,
			"name":              params.Name,
			"parameters":        params.RetryParameters,
			"restartSuccessful": params.RetryOnSuccessfulWorkflow,
			"nodeFieldSelector": params.RetryWorkflowNodeFieldSelector,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(retryWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

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

// StopWorkflow 终止工作流运行，会调用所有的退出处理器
func (c *Client) StopWorkflow(params StopWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace":         params.Namespace,
			"name":              params.Name,
			"nodeFieldSelector": params.NodeFieldSelector,
			"message":           params.Message,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(stopWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type TerminateWorkflowParams struct {
	Namespace string
	Name      string
}

// TerminateWorkflow 终止工作流运行，不调用所有的退出处理器
func (c *Client) TerminateWorkflow(params TerminateWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(terminateWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

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

// SetWorkflow 设置工作流的参数
func (c *Client) SetWorkflow(params SetWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace":         params.Namespace,
			"name":              params.Name,
			"nodeFieldSelector": params.NodeFieldSelector,
			"outputParameters":  params.OutputParameters,
			"message":           params.Message,
			"phase":             params.Phase,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(setWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type SuspendWorkflowParams struct {
	Namespace string
	Name      string
}

// SuspendWorkflow 挂起工作流
func (c *Client) SuspendWorkflow(params SuspendWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(suspendWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type ResumeWorkflowParams struct {
	Namespace         string
	Name              string
	NodeFieldSelector string
}

// ResumeWorkflow 恢复被挂起的工作流
func (c *Client) ResumeWorkflow(params ResumeWorkflowParams) error {
	responseMap := make(map[string]any)

	resp, err := c.restyClient.R().
		SetHeader("Content-Type", "application/json").
		SetAuthToken(c.token).
		SetPathParams(map[string]string{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetBody(map[string]any{
			"namespace": params.Namespace,
			"name":      params.Name,
		}).
		SetResult(&responseMap).
		SetError(&responseMap).
		Put(resumeWorkflowRelativeUrl)
	if err != nil {
		return errors.New(err.Error())
	}

	switch resp.StatusCode() {
	case http.StatusOK:
		return nil
	default:
		message, ok := responseMap["message"]
		if !ok {
			return errors.Errorf("%v", resp.Status())
		}

		return errors.Errorf("%v, %v", resp.Status(), message)
	}
}

type EventCallback func(event map[string]any, eventErr error) error

type GetWorkflowEventsStreamParams struct {
	Namespace string
	Name      string
}

// GetWorkflowEventsStream 监听工作流事件
func (c *Client) GetWorkflowEventsStream(params GetWorkflowEventsStreamParams, callback EventCallback) error {
	return nil
}