fix(backend/sdk): Fix REST API outputs for pipeline versions with invalid platform specs (kubeflow#12183)

* Skip setting spec.platformSpec when platform spec is empty

An empty platform spec causes a bug where the REST API will not return a
pipeline spec because it errantly thinks the empty platform spec is the
pipeline spec.

Signed-off-by: mprahl <[email protected]>

* Fix REST API outputs for pipeline versions with invalid platform specs

When a platform spec did not contain a `kubernetes` platform when
defined as a Kubernetes manifest (e.g. `spec.platformSpec: {}`),
`YamlStringToPipelineSpecStruct` was errantly detecting it as the
pipeline spec when outputting the pipeline version through the REST API.

This caused the REST API to not return a pipeline spec in the case of it
being empty, and caused the UI to load the pipeline and to properly
create a run.

The solution is when translating from the Kubernetes object to a model
representation, the invalid platform specs are not included. This allows
for proper REST API output while also not modifying the user's spec in
the mutating webhook which would cause GitOps tools to always reconcile.

Signed-off-by: mprahl <[email protected]>

---------

Signed-off-by: mprahl <[email protected]>

Author: mprahl

backend/src/crd/kubernetes/v2beta1/pipelineversion_types.go

@@ -179,6 +179,15 @@ func (p *PipelineVersion) ToModel() (*model.PipelineVersion, error) {
 		piplineSpecAndPlatformSpec = append(piplineSpecAndPlatformSpec, platformSpecBytes...)
 	}
 
+	// The pipeline spec in model.PipelineVersion ignores platform specs that don't have a "kubernetes" platform.
+	// This additional parsing filters out platform specs that normally are excluded when the pipeline version is
+	// created through the REST API. This is done rather than modifying the mutating webhook to remove these
+	// platform specs so that GitOps tools don't see a diff from what is in Git and what is on the cluster.
+	v2Spec, err := template.NewV2SpecTemplate(piplineSpecAndPlatformSpec, false, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse the pipeline spec: %w", err)
+	}
+
 	pipelineVersionStatus := model.PipelineVersionCreating
 
 	for _, condition := range p.Status.Conditions {
@@ -214,7 +223,7 @@ func (p *PipelineVersion) ToModel() (*model.PipelineVersion, error) {
 		Status:          pipelineVersionStatus,
 		CodeSourceUrl:   p.Spec.CodeSourceURL,
 		Description:     p.Spec.Description,
-		PipelineSpec:    string(piplineSpecAndPlatformSpec),
+		PipelineSpec:    string(v2Spec.Bytes()),
 		PipelineSpecURI: p.Spec.PipelineSpecURI,
 	}, nil
 }

backend/src/crd/kubernetes/v2beta1/pipelineversion_types_test.go

@@ -404,6 +404,67 @@ func TestToModel_PipelineAndPlatformSpecs(t *testing.T) {
 	assert.Contains(t, result.PipelineSpec, "---")
 }
 
+func TestToModel_PipelineAndPlatformSpecs_WithNonKubernetesPlatform(t *testing.T) {
+	pipelineVersion := &PipelineVersion{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-version",
+			Namespace: "default",
+			UID:       "version-456",
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: GroupVersion.String(),
+					Kind:       "Pipeline",
+					UID:        "pipeline-123",
+					Name:       "test-pipeline",
+				},
+			},
+		},
+		Spec: PipelineVersionSpec{
+			DisplayName:     "Test Version",
+			Description:     "A test version",
+			PipelineName:    "test-pipeline",
+			CodeSourceURL:   "https://github.com/test/pipeline",
+			PipelineSpecURI: "gs://bucket/pipeline.yaml",
+			PipelineSpec: IRSpec{
+				Value: map[string]interface{}{
+					"pipelineInfo": map[string]interface{}{
+						"name":        "test-pipeline",
+						"displayName": "Test Pipeline",
+					},
+					"root": map[string]interface{}{
+						"dag": map[string]interface{}{
+							"tasks": map[string]interface{}{},
+						},
+					},
+					"schemaVersion": "2.1.0",
+					"sdkVersion":    "kfp-2.13.0",
+				},
+			},
+			PlatformSpec: &IRSpec{
+				Value: map[string]interface{}{
+					"platforms": map[string]interface{}{
+						"not-kubernetes": map[string]interface{}{
+							"pipelineConfig": map[string]interface{}{
+								"workspace": map[string]interface{}{
+									"size": "10Gi",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	result, err := pipelineVersion.ToModel()
+	require.NoError(t, err)
+	assert.NotNil(t, result)
+
+	// Check that the platform spec is not included
+	assert.NotContains(t, result.PipelineSpec, "platforms:")
+	assert.NotContains(t, result.PipelineSpec, "---")
+}
+
 func TestToModel_WithStatus(t *testing.T) {
 	pipelineVersion := &PipelineVersion{
 		ObjectMeta: metav1.ObjectMeta{
@@ -426,6 +487,11 @@ func TestToModel_WithStatus(t *testing.T) {
 					"pipelineInfo": map[string]interface{}{
 						"name": "test-pipeline",
 					},
+					"root": map[string]interface{}{
+						"dag": map[string]interface{}{
+							"tasks": map[string]interface{}{},
+						},
+					},
 					"schemaVersion": "2.1.0",
 				},
 			},
@@ -471,6 +537,11 @@ func TestToModel_DisplayNameFallback(t *testing.T) {
 					"pipelineInfo": map[string]interface{}{
 						"name": "test-pipeline",
 					},
+					"root": map[string]interface{}{
+						"dag": map[string]interface{}{
+							"tasks": map[string]interface{}{},
+						},
+					},
 					"schemaVersion": "2.1.0",
 				},
 			},
@@ -498,6 +569,11 @@ func TestToModel_InvalidPipelineSpec(t *testing.T) {
 					"pipelineInfo": map[string]interface{}{
 						"name": "test-pipeline",
 					},
+					"root": map[string]interface{}{
+						"dag": map[string]interface{}{
+							"tasks": map[string]interface{}{},
+						},
+					},
 					"schemaVersion": "2.1.0",
 				},
 			},
@@ -526,6 +602,11 @@ func TestToModel_InvalidPlatformSpec(t *testing.T) {
 					"pipelineInfo": map[string]interface{}{
 						"name": "test-pipeline",
 					},
+					"root": map[string]interface{}{
+						"dag": map[string]interface{}{
+							"tasks": map[string]interface{}{},
+						},
+					},
 					"schemaVersion": "2.1.0",
 				},
 			},

sdk/python/kfp/compiler/compiler_test.py

@@ -1100,8 +1100,18 @@ def my_pipeline(input1: str):
                              pipeline_name)
             self.assertEqual(pipeline_version_manifest['metadata']['namespace'],
                              namespace)
+            self.assertNotIn('platformSpec', pipeline_version_manifest['spec'])
+
+            # Test with include_pipeline_manifest=False and has a platform spec
+            @dsl.pipeline(
+                name='my-pipeline',
+                description='A simple test pipeline with platform spec',
+                pipeline_config=dsl.PipelineConfig(
+                    workspace=dsl.WorkspaceConfig(size='25Gi'),),
+            )
+            def my_pipeline(input1: str):
+                print_op(message=input1)
 
-            # Test with include_pipeline_manifest=False
             package_path2 = os.path.join(tmpdir, 'pipeline2.yaml')
             kubernetes_manifest_options2 = KubernetesManifestOptions(
                 pipeline_name=pipeline_name,
@@ -1136,6 +1146,21 @@ def my_pipeline(input1: str):
                              pipeline_name)
             self.assertEqual(
                 pipeline_version_manifest2['metadata']['namespace'], namespace)
+            self.assertEqual(
+                pipeline_version_manifest2['spec']['platformSpec'], {
+                    'platforms': {
+                        'kubernetes': {
+                            'pipelineConfig': {
+                                'workspace': {
+                                    'kubernetes': {
+                                        'pvcSpecPatch': {}
+                                    },
+                                    'size': '25Gi'
+                                }
+                            }
+                        }
+                    }
+                })
 
 
 class TestCompilePipelineCaching(unittest.TestCase):

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -2121,9 +2121,12 @@ def _write_kubernetes_manifest_to_file(
             'displayName': pipeline_version_display_name,
             'pipelineName': pipeline_name,
             'pipelineSpec': pipeline_spec_dict,
-            'platformSpec': platform_spec_dict,
         },
     }
+
+    if platform_spec_dict:
+        pipeline_version_manifest['spec']['platformSpec'] = platform_spec_dict
+
     if pipeline_description:
         pipeline_version_manifest['spec']['description'] = pipeline_description
     documents.append(pipeline_version_manifest)