progress: basic workflow execution from template.

pkg/types.go

@@ -1,6 +1,7 @@
 package v1
 
 import (
+	"github.com/ghodss/yaml"
 	"strings"
 	"time"
 
@@ -60,6 +61,7 @@ type WorkflowTemplate struct {
 	Version    int32
 	IsLatest   bool `db:"is_latest"`
 	IsArchived bool `db:"is_archived"`
+	LatestArgo *wfv1.WorkflowTemplate
 }
 
 func (wt *WorkflowTemplate) GetManifestBytes() []byte {
@@ -76,6 +78,20 @@ func (wt *WorkflowTemplate) GenerateUID() (string, error) {
 	return wt.UID, nil
 }
 
+func (wt *WorkflowTemplate) GetWorkflowManifestBytes() ([]byte, error) {
+	if wt.LatestArgo == nil {
+		return []byte{}, nil
+	}
+
+	wt.LatestArgo.TypeMeta.Kind = "Workflow"
+	wt.LatestArgo.ObjectMeta = metav1.ObjectMeta{
+		GenerateName: wt.LatestArgo.ObjectMeta.GenerateName,
+		Labels:       wt.LatestArgo.ObjectMeta.Labels,
+	}
+
+	return yaml.Marshal(wt.LatestArgo)
+}
+
 const (
 	WorfklowPending   WorkflowExecutionPhase = "Pending"
 	WorfklowRunning   WorkflowExecutionPhase = "Running"

pkg/workflow_execution.go

@@ -293,7 +293,17 @@ func (c *Client) CreateWorkflowExecution(namespace string, workflow *WorkflowExe
 	(*opts.Labels)[workflowTemplateUIDLabelKey] = workflowTemplate.UID
 	(*opts.Labels)[workflowTemplateVersionLabelKey] = fmt.Sprint(workflowTemplate.Version)
 	//UX will prevent multiple workflows
-	workflows, err := UnmarshalWorkflows([]byte(workflowTemplate.Manifest), true)
+	manifest, err := workflowTemplate.GetWorkflowManifestBytes()
+	if err != nil {
+		log.WithFields(log.Fields{
+			"Namespace":        namespace,
+			"WorkflowTemplate": workflowTemplate,
+			"Error":            err.Error(),
+		}).Error("Error with getting WorkflowManifest from workflow template")
+		return nil, err
+	}
+
+	workflows, err := UnmarshalWorkflows(manifest, true)
 	if err != nil {
 		log.WithFields(log.Fields{
 			"Namespace": namespace,
@@ -317,6 +327,17 @@ func (c *Client) CreateWorkflowExecution(namespace string, workflow *WorkflowExe
 		createdWorkflows = append(createdWorkflows, createdWorkflow)
 	}
 
+	if createdWorkflows == nil {
+		err = errors.New("unable to create workflow")
+		log.WithFields(log.Fields{
+			"Namespace":        namespace,
+			"WorkflowTemplate": workflowTemplate,
+			"Error":            err.Error(),
+		}).Error("Error parsing workflow.")
+
+		return nil, err
+	}
+
 	workflow.Name = createdWorkflows[0].Name
 	workflow.CreatedAt = createdWorkflows[0].CreationTimestamp.UTC()
 	workflow.UID = string(createdWorkflows[0].ObjectMeta.UID)

pkg/workflow_template.go

@@ -183,6 +183,7 @@ func (c *Client) getWorkflowTemplate(namespace, uid string, version int32) (work
 		return nil, err
 	}
 	workflowTemplate.Manifest = string(manifest)
+	workflowTemplate.LatestArgo = argoWft
 
 	return
 }

server/workflow_server.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"github.com/onepanelio/core/pkg/util"
 	"google.golang.org/grpc/codes"
-	"gopkg.in/yaml.v2"
 	"math"
 	"sort"
 	"strings"
@@ -60,22 +59,12 @@ func apiWorkflowExecution(wf *v1.WorkflowExecution) (workflow *api.WorkflowExecu
 }
 
 func apiWorkflowTemplate(wft *v1.WorkflowTemplate) *api.WorkflowTemplate {
-	mapping := make(map[interface{}]interface{})
-	if err := yaml.Unmarshal([]byte(wft.Manifest), mapping); err != nil {
-		return nil
-	}
-
-	finalManifest, err := yaml.Marshal(mapping["spec"])
-	if err != nil {
-		return nil
-	}
-
 	return &api.WorkflowTemplate{
 		Uid:        wft.UID,
 		CreatedAt:  wft.CreatedAt.UTC().Format(time.RFC3339),
 		Name:       wft.Name,
 		Version:    wft.Version,
-		Manifest:   string(finalManifest),
+		Manifest:   wft.Manifest,
 		IsLatest:   wft.IsLatest,
 		IsArchived: wft.IsArchived,
 	}