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"
	resumeWorkflowRelativeUrl          = "/api/v1/workflows/{namespace}/{name}/resume"
	retryWorkflowRelativeUrl           = "/api/v1/workflows/{namespace}/{name}/retry"
	setWorkflowRelativeUrl             = "/api/v1/workflows/{namespace}/{name}/set"
	stopWorkflowRelativeUrl            = "/api/v1/workflows/{namespace}/{name}/stop"
	suspendWorkflowRelativeUrl         = "/api/v1/workflows/{namespace}/{name}/suspend"
	terminateWorkflowRelativeUrl       = "/api/v1/workflows/{namespace}/{name}/terminate"
	getEventsStreamRelativeUrl         = "/api/v1/stream/events/{namespace}"
	getWorkflowEventsStreamRelativeUrl = "/api/v1/workflow-events/{namespace}"
)

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

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
}

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
}

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
}

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
}

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
	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 SubmitWorkflowFromWorkflowTemplateParams struct {
	Namespace    string
	TemplateName string
	Parameters   []string
}

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
	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 ResubmitWorkflowParams struct {
	Namespace          string
	Name               string
	ResubmitParameters []string
}

func (c *Client) ResubmitWorkflow(params ResubmitWorkflowParams) error {
	return nil
}

type ResumeWorkflowParams struct {
	Namespace string
	Name      string
}

func (c *Client) ResumeWorkflow(params ResumeWorkflowParams) error {
	return nil
}

type RetryWorkflowParams struct {
	Namespace                               string
	Name                                    string
	RetryParameters                         []string
	RetryOnSuccessWorkflowNodeFieldSelector string
}

func (c *Client) RetryWorkflow(params RetryWorkflowParams) error {
	return nil
}

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

func (c *Client) SetWorkflow(params SetWorkflowParams) error {
	return nil
}

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

func (c *Client) StopWorkflow(params StopWorkflowParams) error {
	return nil
}

type SuspendWorkflowParams struct {
	Namespace string
	Name      string
}

func (c *Client) SuspendWorkflow(params SuspendWorkflowParams) error {
	return nil
}

type TerminalWorkflowParams struct {
	Namespace string
	Name      string
}

func (c *Client) TerminalWorkflow(params TerminalWorkflowParams) error {
	return nil
}

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

type GetWorkflowEventsStreamParams struct {
	Namespace string
	Name      string
}

func (c *Client) GetWorkflowEventsStream(params GetWorkflowEventsStreamParams, callback EventCallback) error {
	return nil
}