fix(backend): Fix the pipeline samples (kubeflow#11967)

After making name and display_name separate fields in the API, the
pipeline sample upload code needed to accomodate that, otherwise it
would try to upload the pipeline without a display name, which is a
required field. The backwards compatibility was handled at the API
server level but this code bypassed the API server endpoints and used
the resource manager directly.

Signed-off-by: mprahl <[email protected]>
(cherry picked from commit b477269)

Author: mprahl

backend/src/apiserver/config/config.go

@@ -38,11 +38,15 @@ type deprecatedConfig struct {
 }
 
 type configPipelines struct {
-	Name        string
+	Name string
+	// optional, Name is used for Pipeline if not provided
+	DisplayName string
 	Description string
 	File        string
 	// optional, Name is used for PipelineVersion if not provided
 	VersionName string
+	// optional, VersionName, DisplayName, or Name is used for PipelineVersion if not provided
+	VersionDisplayName string
 	// optional, Description is used for PipelineVersion if not provided
 	VersionDescription string
 }
@@ -93,6 +97,7 @@ func LoadSamples(resourceManager *resource.ResourceManager, sampleConfigPath str
 		for _, cfg := range deprecatedCfg {
 			pipelineConfig.Pipelines = append(pipelineConfig.Pipelines, configPipelines{
 				Name:        cfg.Name,
+				DisplayName: cfg.Name,
 				File:        cfg.File,
 				Description: cfg.Description,
 			})
@@ -124,12 +129,18 @@ func LoadSamples(resourceManager *resource.ResourceManager, sampleConfigPath str
 			return fmt.Errorf("failed to load sample %s. Error: %v", cfg.Name, configErr)
 		}
 
+		pipelineDisplayName := cfg.DisplayName
+		if pipelineDisplayName == "" {
+			pipelineDisplayName = cfg.Name
+		}
+
 		// Create pipeline if it does not already exist
 		p, fetchErr := resourceManager.GetPipelineByNameAndNamespace(cfg.Name, common.GetPodNamespace())
 		if fetchErr != nil {
 			if util.IsUserErrorCodeMatch(fetchErr, codes.NotFound) {
 				p, configErr = resourceManager.CreatePipeline(&model.Pipeline{
 					Name:        cfg.Name,
+					DisplayName: pipelineDisplayName,
 					Description: cfg.Description,
 				})
 				if configErr != nil {
@@ -159,6 +170,17 @@ func LoadSamples(resourceManager *resource.ResourceManager, sampleConfigPath str
 			pvName = cfg.VersionName
 		}
 
+		// Use Pipeline Version Display Name if provided, if not use Pipeline Version Name, and if not provided use
+		// Pipeline Display Name (which defaults to Pipeline Name).
+		var pvDisplayName string
+		if cfg.VersionDisplayName != "" {
+			pvDisplayName = cfg.VersionDisplayName
+		} else if cfg.VersionName != "" {
+			pvDisplayName = cfg.VersionName
+		} else {
+			pvDisplayName = p.DisplayName
+		}
+
 		// If the Pipeline Version exists, do nothing
 		// Otherwise upload new Pipeline Version for
 		// this pipeline.
@@ -168,6 +190,7 @@ func LoadSamples(resourceManager *resource.ResourceManager, sampleConfigPath str
 				_, configErr = resourceManager.CreatePipelineVersion(
 					&model.PipelineVersion{
 						Name:         pvName,
+						DisplayName:  pvDisplayName,
 						Description:  pvDescription,
 						PipelineId:   p.UUID,
 						PipelineSpec: string(pipelineFile),

backend/src/apiserver/config/config_test.go

@@ -102,6 +102,41 @@ func TestLoadSamples(t *testing.T) {
 				VersionName:        "Pipeline 2 - Ver 1",
 				VersionDescription: "Pipeline 2 - Ver 1 Description",
 			},
+			// Test case: Pipeline with display name but no version display name
+			{
+				Name:               "Pipeline 3",
+				DisplayName:        "Display Pipeline 3",
+				Description:        "test description",
+				File:               "testdata/sample_pipeline.yaml",
+				VersionName:        "Pipeline 3 - Ver 1",
+				VersionDescription: "Pipeline 3 - Ver 1 Description",
+			},
+			// Test case: Pipeline with version display name but no pipeline display name
+			{
+				Name:               "Pipeline 4",
+				Description:        "test description",
+				File:               "testdata/sample_pipeline.yaml",
+				VersionName:        "Pipeline 4 - Ver 1",
+				VersionDisplayName: "Display Pipeline 4 - Version 1",
+				VersionDescription: "Pipeline 4 - Ver 1 Description",
+			},
+			// Test case: Pipeline with both display names
+			{
+				Name:               "Pipeline 5",
+				DisplayName:        "Display Pipeline 5",
+				Description:        "test description",
+				File:               "testdata/sample_pipeline.yaml",
+				VersionName:        "Pipeline 5 - Ver 1",
+				VersionDisplayName: "Display Pipeline 5 - Version 1",
+				VersionDescription: "Pipeline 5 - Ver 1 Description",
+			},
+			// Test case: Pipeline with only pipeline name, no version name or display names
+			{
+				Name:               "Pipeline 6",
+				Description:        "test description",
+				File:               "testdata/sample_pipeline.yaml",
+				VersionDescription: "Pipeline 6 - Ver 1 Description",
+			},
 		},
 	}
 
@@ -113,9 +148,51 @@ func TestLoadSamples(t *testing.T) {
 	var pipeline1 *model.Pipeline
 	pipeline1, err = rm.GetPipelineByNameAndNamespace(pc.Pipelines[0].Name, "")
 	require.NoError(t, err)
+	version1, err := rm.GetPipelineVersionByName(pc.Pipelines[0].VersionName)
+	require.NoError(t, err)
+	// Verify that the pipeline display name is set to the pipeline name if not provided
+	assert.Equal(t, pipeline1.DisplayName, pc.Pipelines[0].Name)
+	assert.Equal(t, version1.DisplayName, pc.Pipelines[0].VersionName)
 	var pipeline2 *model.Pipeline
 	pipeline2, err = rm.GetPipelineByNameAndNamespace(pc.Pipelines[1].Name, "")
 	require.NoError(t, err)
+	version2, err := rm.GetPipelineVersionByName(pc.Pipelines[1].VersionName)
+	// Verify that the pipeline display name is set to the pipeline name if not provided
+	assert.Equal(t, pipeline2.DisplayName, pc.Pipelines[1].Name)
+	assert.Equal(t, version2.DisplayName, pc.Pipelines[1].VersionName)
+	require.NoError(t, err)
+
+	// Test display name for Pipeline 3 (pipeline display name only)
+	pipeline3, err := rm.GetPipelineByNameAndNamespace(pc.Pipelines[2].Name, "")
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[2].DisplayName, pipeline3.DisplayName)
+	version3, err := rm.GetPipelineVersionByName(pc.Pipelines[2].VersionName)
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[2].VersionName, version3.DisplayName) // Should use version name when no version display name is provided
+
+	// Test display name for Pipeline 4 (version display name only)
+	pipeline4, err := rm.GetPipelineByNameAndNamespace(pc.Pipelines[3].Name, "")
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[3].Name, pipeline4.DisplayName) // Should use pipeline name
+	version4, err := rm.GetPipelineVersionByName(pc.Pipelines[3].VersionName)
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[3].VersionDisplayName, version4.DisplayName)
+
+	// Test display name for Pipeline 5 (both display names)
+	pipeline5, err := rm.GetPipelineByNameAndNamespace(pc.Pipelines[4].Name, "")
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[4].DisplayName, pipeline5.DisplayName)
+	version5, err := rm.GetPipelineVersionByName(pc.Pipelines[4].VersionName)
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[4].VersionDisplayName, version5.DisplayName)
+
+	// Test display name for Pipeline 6 (only pipeline name)
+	pipeline6, err := rm.GetPipelineByNameAndNamespace(pc.Pipelines[5].Name, "")
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[5].Name, pipeline6.DisplayName)       // Should use pipeline name
+	version6, err := rm.GetPipelineVersionByName(pc.Pipelines[5].Name) // Version name should default to pipeline name
+	require.NoError(t, err)
+	assert.Equal(t, pc.Pipelines[5].Name, version6.DisplayName) // Version display name should default to pipeline name
 
 	_, err = rm.GetPipelineVersionByName(pc.Pipelines[0].VersionName)
 	require.NoError(t, err)
@@ -145,10 +222,12 @@ func TestLoadSamples(t *testing.T) {
 	_, err = rm.GetPipelineVersionByName(pc.Pipelines[1].VersionName)
 	require.NoError(t, err)
 	_, totalSize, _, err = rm.ListPipelineVersions(pipeline2.UUID, opts)
+	require.NoError(t, err)
 	require.Equal(t, totalSize, 2)
 
 	// Confirm previous pipeline version count has not been affected
 	_, totalSize, _, err = rm.ListPipelineVersions(pipeline1.UUID, opts)
+	require.NoError(t, err)
 	require.Equal(t, totalSize, 2)
 
 	// When LoadSamplesOnRestart is false, changes to config should

backend/src/apiserver/config/sample_config.json

@@ -2,12 +2,14 @@
   "loadSamplesOnRestart": true,
   "pipelines": [
     {
-      "name": "[Tutorial] Data passing in python components",
+      "name": "tutorial-data-passing-in-python-components",
+      "displayName": "[Tutorial] Data passing in python components",
       "description": "[source code](https://github.com/kubeflow/pipelines/tree/master/samples/tutorials/Data%20passing%20in%20python%20components) Shows how to pass data between python components.",
       "file": "/samples/tutorials/Data passing in python components/Data passing in python components - Files.py.yaml"
     },
     {
-      "name": "[Tutorial] DSL - Control structures",
+      "name": "tutorial-dsl-control-structures",
+      "displayName": "[Tutorial] DSL - Control structures",
       "description": "[source code](https://github.com/kubeflow/pipelines/tree/master/samples/tutorials/DSL%20-%20Control%20structures) Shows how to use conditional execution and exit handlers. This pipeline will randomly fail to demonstrate that the exit handler gets executed even in case of failure.",
       "file": "/samples/tutorials/DSL - Control structures/DSL - Control structures.py.yaml"
     }

backend/src/apiserver/resource/resource_manager.go

@@ -350,6 +350,11 @@ func (r *ResourceManager) CreatePipeline(p *model.Pipeline) (*model.Pipeline, er
 	if p.Name == "" {
 		return nil, util.NewInvalidInputError("pipeline's name cannot be empty")
 	}
+
+	if p.DisplayName == "" {
+		p.DisplayName = p.Name
+	}
+
 	// Create a record in KFP DB (only pipelines table)
 	newPipeline, err := r.pipelineStore.CreatePipeline(p)
 	if err != nil {
@@ -1697,6 +1702,11 @@ func (r *ResourceManager) CreatePipelineVersion(pv *model.PipelineVersion) (*mod
 		}
 		pv.Name = pipelineSpecName
 	}
+
+	if pv.DisplayName == "" {
+		pv.DisplayName = pv.Name
+	}
+
 	// Parse parameters
 	paramsJSON, err := tmpl.ParametersJSON()
 	if err != nil {