Runs and jobs can be created from pipeline version

backend/api/pipeline.proto

@@ -120,6 +120,13 @@ service PipelineService {
     //   delete: "/apis/v1beta1/pipeline_versions/{version_id}"
     // };
   }
+
+  rpc GetPipelineVersionTemplate(GetPipelineVersionTemplateRequest) returns (GetTemplateResponse) {
+    // TODO(jingzhang36): uncomment when exposing this API method.
+    // option (google.api.http) = {
+    //   get: "/apis/v1beta1/pipeline_versions/{version_id}/templates"
+    // };
+  }
 }
 
 message Url {
@@ -240,8 +247,7 @@ message Pipeline {
   // Output only. The default version of the pipeline. As of now, the latest
   // version is used as default. (In the future, if desired by customers, we
   // can allow them to set default version.)
-  // TODO(jingzhang36): expose this in API pipeline definition with FE changes.
-  // PipelineVersion default_version = 8;
+  PipelineVersion default_version = 8;
 }
 
 message PipelineVersion {
@@ -268,4 +274,4 @@ message PipelineVersion {
   // Input. Required. E.g., specify which pipeline this pipeline version belongs
   // to.
   repeated ResourceReference resource_references = 7;
-}
+}

backend/api/resource_reference.proto

@@ -22,6 +22,7 @@ enum ResourceType {
   EXPERIMENT = 1;
   JOB = 2;
   PIPELINE = 3;
+  PIPELINE_VERSION = 4;
 }
 
 enum Relationship {

backend/src/apiserver/common/const.go

@@ -23,10 +23,11 @@ type ResourceType string
 type Relationship string
 
 const (
-	Experiment ResourceType = "Experiment"
-	Job        ResourceType = "Job"
-	Run        ResourceType = "Run"
-	Pipeline   ResourceType = "pipeline"
+	Experiment      ResourceType = "Experiment"
+	Job             ResourceType = "Job"
+	Run             ResourceType = "Run"
+	Pipeline        ResourceType = "pipeline"
+	PipelineVersion ResourceType = "PipelineVersion"
 )
 
 const (
@@ -40,6 +41,8 @@ func ToModelResourceType(apiType api.ResourceType) (ResourceType, error) {
 		return Experiment, nil
 	case api.ResourceType_JOB:
 		return Job, nil
+	case api.ResourceType_PIPELINE_VERSION:
+		return PipelineVersion, nil
 	default:
 		return "", util.NewInvalidInputError("Unsupported resource type: %s", api.ResourceType_name[int32(apiType)])
 	}

backend/src/apiserver/resource/model_converter.go

@@ -37,17 +37,17 @@ func (r *ResourceManager) ToModelRunMetric(metric *api.RunMetric, runUUID string
 // The input run might not contain workflowSpecManifest, but instead a pipeline ID.
 // The caller would retrieve workflowSpecManifest and pass in.
 func (r *ResourceManager) ToModelRunDetail(run *api.Run, runId string, workflow *util.Workflow, workflowSpecManifest string) (*model.RunDetail, error) {
-	params, err := toModelParameters(run.PipelineSpec.Parameters)
+	params, err := toModelParameters(run.GetPipelineSpec().GetParameters())
 	if err != nil {
 		return nil, util.Wrap(err, "Unable to parse the parameter.")
 	}
-	resourceReferences, err := r.toModelResourceReferences(runId, common.Run, run.ResourceReferences)
+	resourceReferences, err := r.toModelResourceReferences(runId, common.Run, run.GetResourceReferences())
 	if err != nil {
 		return nil, util.Wrap(err, "Unable to convert resource references.")
 	}
 	var pipelineName string
-	if run.PipelineSpec.GetPipelineId() != "" {
-		pipelineName, err = r.getResourceName(common.Pipeline, run.PipelineSpec.GetPipelineId())
+	if run.GetPipelineSpec().GetPipelineId() != "" {
+		pipelineName, err = r.getResourceName(common.Pipeline, run.GetPipelineSpec().GetPipelineId())
 		if err != nil {
 			return nil, util.Wrap(err, "Error getting the pipeline name")
 		}
@@ -63,7 +63,7 @@ func (r *ResourceManager) ToModelRunDetail(run *api.Run, runId string, workflow
 			Description:        run.Description,
 			ResourceReferences: resourceReferences,
 			PipelineSpec: model.PipelineSpec{
-				PipelineId:           run.PipelineSpec.GetPipelineId(),
+				PipelineId:           run.GetPipelineSpec().GetPipelineId(),
 				PipelineName:         pipelineName,
 				WorkflowSpecManifest: workflowSpecManifest,
 				Parameters:           params,
@@ -76,17 +76,17 @@ func (r *ResourceManager) ToModelRunDetail(run *api.Run, runId string, workflow
 }
 
 func (r *ResourceManager) ToModelJob(job *api.Job, swf *util.ScheduledWorkflow, workflowSpecManifest string) (*model.Job, error) {
-	params, err := toModelParameters(job.PipelineSpec.Parameters)
+	params, err := toModelParameters(job.GetPipelineSpec().GetParameters())
 	if err != nil {
 		return nil, util.Wrap(err, "Error parsing the input job.")
 	}
-	resourceReferences, err := r.toModelResourceReferences(string(swf.UID), common.Job, job.ResourceReferences)
+	resourceReferences, err := r.toModelResourceReferences(string(swf.UID), common.Job, job.GetResourceReferences())
 	if err != nil {
 		return nil, util.Wrap(err, "Error to convert resource references.")
 	}
 	var pipelineName string
-	if job.PipelineSpec.GetPipelineId() != "" {
-		pipelineName, err = r.getResourceName(common.Pipeline, job.PipelineSpec.GetPipelineId())
+	if job.GetPipelineSpec().GetPipelineId() != "" {
+		pipelineName, err = r.getResourceName(common.Pipeline, job.GetPipelineSpec().GetPipelineId())
 		if err != nil {
 			return nil, util.Wrap(err, "Error getting the pipeline name")
 		}
@@ -103,7 +103,7 @@ func (r *ResourceManager) ToModelJob(job *api.Job, swf *util.ScheduledWorkflow,
 		MaxConcurrency:     job.MaxConcurrency,
 		ResourceReferences: resourceReferences,
 		PipelineSpec: model.PipelineSpec{
-			PipelineId:           job.PipelineSpec.GetPipelineId(),
+			PipelineId:           job.GetPipelineSpec().GetPipelineId(),
 			PipelineName:         pipelineName,
 			WorkflowSpecManifest: workflowSpecManifest,
 			Parameters:           params,
@@ -238,6 +238,12 @@ func (r *ResourceManager) getResourceName(resourceType common.ResourceType, reso
 			return "", util.Wrap(err, "Referred run not found.")
 		}
 		return run.DisplayName, nil
+	case common.PipelineVersion:
+		version, err := r.GetPipelineVersion(resourceId)
+		if err != nil {
+			return "", util.Wrap(err, "Referred pipeline version not found.")
+		}
+		return version.Name, nil
 	default:
 		return "", util.NewInvalidInputError("Unsupported resource type: %s", string(resourceType))
 	}

backend/src/apiserver/resource/resource_manager.go

@@ -230,10 +230,16 @@ func (r *ResourceManager) GetPipelineTemplate(pipelineId string) ([]byte, error)
 }
 
 func (r *ResourceManager) CreateRun(apiRun *api.Run) (*model.RunDetail, error) {
-	// Get workflow from pipeline spec, which might be pipeline ID or an argo workflow
+	// Get workflow from either of the two places:
+	// (1) raw pipeline manifest in pipeline_spec
+	// (2) pipeline version in resource_references
+	var workflowSpecManifestBytes []byte
 	workflowSpecManifestBytes, err := r.getWorkflowSpecBytes(apiRun.GetPipelineSpec())
 	if err != nil {
-		return nil, util.Wrap(err, "Failed to fetch workflow spec.")
+		workflowSpecManifestBytes, err = r.getWorkflowSpecBytesFromPipelineVersion(apiRun.GetResourceReferences())
+		if err != nil {
+			return nil, util.Wrap(err, "Failed to fetch workflow spec.")
+		}
 	}
 	uuid, err := r.uuid.NewRandom()
 	if err != nil {
@@ -282,7 +288,7 @@ func (r *ResourceManager) CreateRun(apiRun *api.Run) (*model.RunDetail, error) {
 	}
 
 	// Add a reference to the default experiment if run does not already have a containing experiment
-	ref, err := r.getDefaultExperimentIfNoExperiment(apiRun.ResourceReferences)
+	ref, err := r.getDefaultExperimentIfNoExperiment(apiRun.GetResourceReferences())
 	if err != nil {
 		return nil, err
 	}
@@ -441,11 +447,18 @@ func (r *ResourceManager) GetJob(id string) (*model.Job, error) {
 }
 
 func (r *ResourceManager) CreateJob(apiJob *api.Job) (*model.Job, error) {
-	// Get workflow from pipeline spec, which might be pipeline ID or an argo workflow
+	// Get workflow from either of the two places:
+	// (1) raw pipeline manifest in pipeline_spec
+	// (2) pipeline version in resource_references
+	var workflowSpecManifestBytes []byte
 	workflowSpecManifestBytes, err := r.getWorkflowSpecBytes(apiJob.GetPipelineSpec())
 	if err != nil {
-		return nil, util.Wrap(err, "Failed to fetch workflow spec.")
+		workflowSpecManifestBytes, err = r.getWorkflowSpecBytesFromPipelineVersion(apiJob.GetResourceReferences())
+		if err != nil {
+			return nil, util.Wrap(err, "Failed to fetch workflow spec.")
+		}
 	}
+
 	var workflow util.Workflow
 	err = json.Unmarshal(workflowSpecManifestBytes, &workflow)
 	if err != nil {
@@ -454,7 +467,7 @@ func (r *ResourceManager) CreateJob(apiJob *api.Job) (*model.Job, error) {
 	}
 
 	// Verify no additional parameter provided
-	err = workflow.VerifyParameters(toParametersMap(apiJob.PipelineSpec.Parameters))
+	err = workflow.VerifyParameters(toParametersMap(apiJob.GetPipelineSpec().GetParameters()))
 	if err != nil {
 		return nil, util.Wrap(err, "Create job failed")
 	}
@@ -473,7 +486,7 @@ func (r *ResourceManager) CreateJob(apiJob *api.Job) (*model.Job, error) {
 			MaxConcurrency: &apiJob.MaxConcurrency,
 			Trigger:        *toCRDTrigger(apiJob.Trigger),
 			Workflow: &scheduledworkflow.WorkflowResource{
-				Parameters: toCRDParameter(apiJob.PipelineSpec.Parameters),
+				Parameters: toCRDParameter(apiJob.GetPipelineSpec().GetParameters()),
 				Spec:       workflow.Spec,
 			},
 		},
@@ -484,12 +497,12 @@ func (r *ResourceManager) CreateJob(apiJob *api.Job) (*model.Job, error) {
 	}
 
 	// Add a reference to the default experiment if run does not already have a containing experiment
-	ref, err := r.getDefaultExperimentIfNoExperiment(apiJob.ResourceReferences)
+	ref, err := r.getDefaultExperimentIfNoExperiment(apiJob.GetResourceReferences())
 	if err != nil {
 		return nil, err
 	}
 	if ref != nil {
-		apiJob.ResourceReferences = append(apiJob.ResourceReferences, ref)
+		apiJob.ResourceReferences = append(apiJob.GetResourceReferences(), ref)
 	}
 
 	job, err := r.ToModelJob(apiJob, util.NewScheduledWorkflow(newScheduledWorkflow), string(workflowSpecManifestBytes))
@@ -688,6 +701,8 @@ func (r *ResourceManager) checkRunExist(runID string) (*model.RunDetail, error)
 }
 
 func (r *ResourceManager) getWorkflowSpecBytes(spec *api.PipelineSpec) ([]byte, error) {
+	// TODO(jingzhang36): after FE is enabled to use pipeline version to create
+	// run, we'll only check for the raw manifest in pipeline_spec.
 	if spec.GetPipelineId() != "" {
 		var workflow util.Workflow
 		err := r.objectStore.GetFromYamlFile(&workflow, storage.CreatePipelinePath(spec.GetPipelineId()))
@@ -702,6 +717,25 @@ func (r *ResourceManager) getWorkflowSpecBytes(spec *api.PipelineSpec) ([]byte,
 	return nil, util.NewInvalidInputError("Please provide a valid pipeline spec")
 }
 
+func (r *ResourceManager) getWorkflowSpecBytesFromPipelineVersion(references []*api.ResourceReference) ([]byte, error) {
+	var pipelineVersionId = ""
+	for _, reference := range references {
+		if reference.Key.Type == api.ResourceType_PIPELINE_VERSION && reference.Relationship == api.Relationship_CREATOR {
+			pipelineVersionId = reference.Key.Id
+		}
+	}
+	if len(pipelineVersionId) == 0 {
+		return nil, util.NewInvalidInputError("No pipeline version.")
+	}
+	var workflow util.Workflow
+	err := r.objectStore.GetFromYamlFile(&workflow, storage.CreatePipelinePath(pipelineVersionId))
+	if err != nil {
+		return nil, util.Wrap(err, "Get pipeline YAML failed.")
+	}
+
+	return []byte(workflow.ToStringForStore()), nil
+}
+
 // Used to initialize the Experiment database with a default to be used for runs
 func (r *ResourceManager) CreateDefaultExperiment() (string, error) {
 	// First check that we don't already have a default experiment ID in the DB.
@@ -896,3 +930,18 @@ func (r *ResourceManager) DeletePipelineVersion(pipelineVersionId string) error
 
 	return nil
 }
+
+func (r *ResourceManager) GetPipelineVersionTemplate(versionId string) ([]byte, error) {
+	// Verify pipeline version exist
+	_, err := r.pipelineStore.GetPipelineVersion(versionId)
+	if err != nil {
+		return nil, util.Wrap(err, "Get pipeline version template failed")
+	}
+
+	template, err := r.objectStore.GetFile(storage.CreatePipelinePath(fmt.Sprint(versionId)))
+	if err != nil {
+		return nil, util.Wrap(err, "Get pipeline version template failed")
+	}
+
+	return template, nil
+}