Files
onepanel/manager/workflow_manager.go
2019-12-15 13:47:25 -08:00

113 lines
3.5 KiB
Go

package manager
import (
"encoding/json"
"fmt"
"strconv"
"github.com/onepanelio/core/argo"
"github.com/onepanelio/core/model"
"github.com/onepanelio/core/util"
"github.com/spf13/viper"
"google.golang.org/grpc/codes"
)
func (r *ResourceManager) CreateWorkflow(namespace string, workflow *model.Workflow) (createdWorkflow *model.Workflow, err error) {
workflowTemplate, err := r.GetWorkflowTemplate(namespace, workflow.WorkflowTemplate.UID, workflow.WorkflowTemplate.Version)
if err != nil {
return nil, err
}
// TODO: Need to pull system parameters from k8s config/secret here, example: HOST
opts := &argo.Options{
Namespace: namespace,
}
for _, param := range workflow.Parameters {
opts.Parameters = append(opts.Parameters, argo.Parameter{
Name: param.Name,
Value: param.Value,
})
}
if opts.Labels == nil {
opts.Labels = &map[string]string{}
}
(*opts.Labels)[viper.GetString("k8s.labelKeyPrefix")+"workflow-template-uid"] = workflowTemplate.UID
(*opts.Labels)[viper.GetString("k8s.labelKeyPrefix")+"workflow-template-version"] = fmt.Sprint(workflowTemplate.Version)
createdWorkflows, err := r.argClient.Create(workflowTemplate.GetManifestBytes(), opts)
if err != nil {
return
}
createdWorkflow = workflow
createdWorkflow.Name = createdWorkflows[0].Name
createdWorkflow.UID = string(createdWorkflows[0].ObjectMeta.UID)
return
}
func (r *ResourceManager) GetWorkflow(namespace, name string) (workflow *model.Workflow, err error) {
wf, err := r.argClient.Get(name, &argo.Options{Namespace: namespace})
if err != nil {
return nil, util.NewUserError(codes.NotFound, "Workflow not found.")
}
uid := wf.ObjectMeta.Labels[viper.GetString("k8s.labelKeyPrefix")+"workflow-template-uid"]
version, err := strconv.ParseInt(
wf.ObjectMeta.Labels[viper.GetString("k8s.labelKeyPrefix")+"workflow-template-version"],
10,
32,
)
if err != nil {
return nil, util.NewUserError(codes.InvalidArgument, "Invalid version number.")
}
workflowTemplate, err := r.GetWorkflowTemplate(namespace, uid, int32(version))
if err != nil {
return nil, err
}
status, err := json.Marshal(wf.Status)
if err != nil {
return nil, util.NewUserError(codes.InvalidArgument, "Invalid status.")
}
workflow = &model.Workflow{
UID: string(wf.UID),
Name: wf.Name,
Status: string(status),
WorkflowTemplate: workflowTemplate,
}
return
}
func (r *ResourceManager) CreateWorkflowTemplate(namespace string, workflowTemplate *model.WorkflowTemplate) (createdWorkflowTemplate *model.WorkflowTemplate, err error) {
createdWorkflowTemplate, err = r.workflowRepository.CreateWorkflowTemplate(namespace, workflowTemplate)
if err != nil {
return nil, util.NewUserErrorWrap(err, "Workflow template")
}
return
}
func (r *ResourceManager) GetWorkflowTemplate(namespace, uid string, version int32) (workflowTemplate *model.WorkflowTemplate, err error) {
workflowTemplate, err = r.workflowRepository.GetWorkflowTemplate(namespace, uid, version)
if err != nil {
return nil, util.NewUserError(codes.Unknown, "Unknown error.")
}
if err == nil && workflowTemplate == nil {
return nil, util.NewUserError(codes.NotFound, "Workflow template not found.")
}
return
}
func (r *ResourceManager) ListWorkflowTemplateVersions(namespace, uid string) (workflowTemplateVersions []*model.WorkflowTemplate, err error) {
workflowTemplateVersions, err = r.workflowRepository.ListWorkflowTemplateVersions(namespace, uid)
if err != nil {
return nil, util.NewUserError(codes.NotFound, "Workflow template versions not found.")
}
return
}