fixing pipeline_upload_client_kubernetes

Signed-off-by: Nelesh Singla <117123879+nsingla@users.noreply.github.com>

Author: nsingla

backend/src/apiserver/common/tag_validation.go

@@ -0,0 +1,59 @@
+// Copyright 2026 The Kubeflow Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	"strings"
+	"unicode/utf8"
+
+	"github.com/kubeflow/pipelines/backend/src/common/util"
+)
+
+// MaxTagKeyLength is the maximum allowed length (in characters) for a tag key.
+// Consistent with Kubernetes label value length limit (63 characters).
+const MaxTagKeyLength = 63
+
+// MaxTagValueLength is the maximum allowed length (in characters) for a tag value.
+// Consistent with Kubernetes label value length limit (63 characters).
+const MaxTagValueLength = 63
+
+// MaxTagsPerEntity is the maximum number of tags allowed on a single entity.
+const MaxTagsPerEntity = 20
+
+// ValidateTags checks that tags conform to all constraints:
+// - maximum number of tags per entity
+// - no empty keys
+// - keys must not contain "." (conflicts with tag filter prefix "tags.")
+// - key and value character lengths within limits
+func ValidateTags(tags map[string]string) error {
+	if len(tags) > MaxTagsPerEntity {
+		return util.NewInvalidInputError("number of tags (%d) exceeds maximum of %d per entity", len(tags), MaxTagsPerEntity)
+	}
+	for key, value := range tags {
+		if key == "" {
+			return util.NewInvalidInputError("tag key cannot be empty")
+		}
+		if strings.Contains(key, ".") {
+			return util.NewInvalidInputError("tag key %q must not contain '.' character", key)
+		}
+		if utf8.RuneCountInString(key) > MaxTagKeyLength {
+			return util.NewInvalidInputError("tag key %q exceeds maximum length of %d characters", key, MaxTagKeyLength)
+		}
+		if utf8.RuneCountInString(value) > MaxTagValueLength {
+			return util.NewInvalidInputError("tag value %q for key %q exceeds maximum length of %d characters", value, key, MaxTagValueLength)
+		}
+	}
+	return nil
+}

backend/src/apiserver/resource/resource_manager.go

@@ -23,9 +23,7 @@ import (
 	"net/url"
 	"reflect"
 	"strconv"
-	"strings"
 	"time"
-	"unicode/utf8"
 
 	apiv2beta1 "github.com/kubeflow/pipelines/backend/api/v2beta1/go_client"
 	scheduledworkflow "github.com/kubeflow/pipelines/backend/src/crd/pkg/apis/scheduledworkflow/v1beta1"
@@ -394,48 +392,22 @@ func (r *ResourceManager) UpdatePipelineDefaultVersion(pipelineId string, versio
 
 // MaxTagKeyLength is the maximum allowed length (in characters) for a tag key.
 // Consistent with Kubernetes label value length limit (63 characters).
-const MaxTagKeyLength = 63
+const MaxTagKeyLength = common.MaxTagKeyLength
 
 // MaxTagValueLength is the maximum allowed length (in characters) for a tag value.
 // Consistent with Kubernetes label value length limit (63 characters).
-const MaxTagValueLength = 63
+const MaxTagValueLength = common.MaxTagValueLength
 
 // MaxTagsPerEntity is the maximum number of tags allowed on a single pipeline or pipeline version.
-const MaxTagsPerEntity = 20
-
-// validateTags checks that tags conform to all constraints:
-// - maximum number of tags per entity
-// - no empty keys
-// - keys must not contain "." (conflicts with tag filter prefix "tags.")
-// - key and value character lengths within limits
-func validateTags(tags map[string]string) error {
-	if len(tags) > MaxTagsPerEntity {
-		return util.NewInvalidInputError("number of tags (%d) exceeds maximum of %d per entity", len(tags), MaxTagsPerEntity)
-	}
-	for key, value := range tags {
-		if key == "" {
-			return util.NewInvalidInputError("tag key cannot be empty")
-		}
-		if strings.Contains(key, ".") {
-			return util.NewInvalidInputError("tag key %q must not contain '.' character", key)
-		}
-		if utf8.RuneCountInString(key) > MaxTagKeyLength {
-			return util.NewInvalidInputError("tag key %q exceeds maximum length of %d characters", key, MaxTagKeyLength)
-		}
-		if utf8.RuneCountInString(value) > MaxTagValueLength {
-			return util.NewInvalidInputError("tag value %q for key %q exceeds maximum length of %d characters", value, key, MaxTagValueLength)
-		}
-	}
-	return nil
-}
+const MaxTagsPerEntity = common.MaxTagsPerEntity
 
 // UpdatePipeline updates mutable fields of a pipeline (display_name, tags).
 // Both fields are updated in a single transaction via UpdatePipelineFields.
 func (r *ResourceManager) UpdatePipeline(pipelineID string, displayName string, tags map[string]string) (*model.Pipeline, error) {
 	if pipelineID == "" {
 		return nil, util.NewInvalidInputError("pipeline id cannot be empty when updating pipeline")
 	}
-	if err := validateTags(tags); err != nil {
+	if err := common.ValidateTags(tags); err != nil {
 		return nil, err
 	}
 	// Update fields and tags in a single transaction to prevent deadlocks.
@@ -452,7 +424,7 @@ func (r *ResourceManager) UpdatePipelineVersion(pipelineVersionID string, displa
 	if pipelineVersionID == "" {
 		return nil, util.NewInvalidInputError("pipeline version id cannot be empty when updating pipeline version")
 	}
-	if err := validateTags(tags); err != nil {
+	if err := common.ValidateTags(tags); err != nil {
 		return nil, err
 	}
 	// Update fields and tags in a single transaction to prevent deadlocks.
@@ -474,7 +446,7 @@ func (r *ResourceManager) CreatePipeline(p *model.Pipeline) (*model.Pipeline, er
 		p.DisplayName = p.Name
 	}
 
-	if err := validateTags(p.Tags); err != nil {
+	if err := common.ValidateTags(p.Tags); err != nil {
 		return nil, err
 	}
 
@@ -498,10 +470,10 @@ func (r *ResourceManager) CreatePipeline(p *model.Pipeline) (*model.Pipeline, er
 // Creates a pipeline and a pipeline version.
 // This is used when two resources need to be created in a single DB transaction.
 func (r *ResourceManager) CreatePipelineAndPipelineVersion(p *model.Pipeline, pv *model.PipelineVersion) (*model.Pipeline, *model.PipelineVersion, error) {
-	if err := validateTags(p.Tags); err != nil {
+	if err := common.ValidateTags(p.Tags); err != nil {
 		return nil, nil, err
 	}
-	if err := validateTags(pv.Tags); err != nil {
+	if err := common.ValidateTags(pv.Tags); err != nil {
 		return nil, nil, err
 	}
 
@@ -1909,7 +1881,7 @@ func (r *ResourceManager) CreatePipelineVersion(pv *model.PipelineVersion) (*mod
 	pv.Status = model.PipelineVersionCreating
 	pv.PipelineSpec = model.LargeText(string(tmpl.Bytes()))
 
-	if err := validateTags(pv.Tags); err != nil {
+	if err := common.ValidateTags(pv.Tags); err != nil {
 		return nil, err
 	}
 

backend/src/apiserver/resource/resource_manager_test.go

@@ -18,8 +18,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net/url"
 	"io"
+	"net/url"
 	"strings"
 	"testing"
 	"time"
@@ -4620,7 +4620,7 @@ func TestValidateTags(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := validateTags(tt.tags)
+			err := common.ValidateTags(tt.tags)
 			if tt.wantErr {
 				assert.NotNil(t, err)
 				assert.Contains(t, err.Error(), tt.errMsg)

backend/src/common/client/api_server/v2/pipeline_upload_client_kubernetes.go

@@ -156,6 +156,11 @@ func (c *PipelineUploadClientKubernetes) Upload(parameters *params.UploadPipelin
 			fmt.Sprintf("Failed to parse pipeline tags. Params: '%v'", parameters),
 			"Failed to upload pipeline")
 	}
+	if err := common.ValidateTags(tags); err != nil {
+		return nil, util.NewUserError(err,
+			fmt.Sprintf("Failed to validate pipeline tags. Params: '%v'", parameters),
+			"Failed to upload pipeline")
+	}
 
 	pipelineModel := apimodel.Pipeline{
 		Name:        name,
@@ -274,6 +279,11 @@ func (c *PipelineUploadClientKubernetes) UploadPipelineVersion(filePath string,
 			fmt.Sprintf("Failed to parse pipeline version tags. Params: '%v'", parameters),
 			"Failed to upload pipeline version")
 	}
+	if err := common.ValidateTags(tags); err != nil {
+		return nil, util.NewUserError(err,
+			fmt.Sprintf("Failed to validate pipeline version tags. Params: '%v'", parameters),
+			"Failed to upload pipeline version")
+	}
 
 	modelPipelineVersion := apimodel.PipelineVersion{
 		Name:         name,

backend/src/common/client/api_server/v2/pipeline_upload_client_kubernetes_test.go

@@ -144,6 +144,45 @@ func TestUploadPipelineVersionKubernetesClient_InvalidTagsReturnsError(t *testin
 	}
 }
 
+func TestUploadPipelineVersionKubernetesClient_InvalidTagValueLengthReturnsError(t *testing.T) {
+	pipelineID := "pipeline-id"
+	namespace := "kubeflow"
+	pipeline := &k8sapi.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-pipeline",
+			Namespace: namespace,
+			UID:       types.UID(pipelineID),
+		},
+	}
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(pipeline).
+		Build()
+	client := &PipelineUploadClientKubernetes{ctrlClient: fakeClient, namespace: namespace}
+
+	specPath := writePipelineSpecFile(t)
+	rawTags := `{"team":"this-value-is-way-too-long-for-tags-and-exceeds-the-sixty-three-char-limit"}`
+	versionName := "tagged-version"
+	params := params.NewUploadPipelineVersionParams()
+	params.Pipelineid = &pipelineID
+	params.SetName(&versionName)
+	params.SetTags(&rawTags)
+
+	_, err := client.UploadPipelineVersion(specPath, params)
+	if err == nil {
+		t.Fatalf("expected UploadPipelineVersion() to fail for tag value longer than 63 characters")
+	}
+
+	var versions k8sapi.PipelineVersionList
+	if err := fakeClient.List(context.Background(), &versions, &ctrlclient.ListOptions{Namespace: namespace}); err != nil {
+		t.Fatalf("failed to list pipeline versions: %v", err)
+	}
+	if len(versions.Items) != 0 {
+		t.Fatalf("expected no pipeline versions on invalid tag value length, got %d", len(versions.Items))
+	}
+}
+
 func writePipelineSpecFile(t *testing.T) string {
 	t.Helper()
 

backend/test/v2/api/pipeline_upload_api_test.go

@@ -112,7 +112,8 @@ var _ = Describe("Verify Pipeline Upload with Tags >", Label(constants.POSITIVE,
 			tagsJSON, err := testutil.TagsMapToJSONStringPtr(emptyTags)
 			Expect(err).NotTo(HaveOccurred())
 			testContext.Pipeline.UploadParams.Tags = tagsJSON
-			uploadPipelineAndVerify(pipelineSpecFilePath, &testContext.Pipeline.PipelineGeneratedName, nil)
+			createdPipeline := uploadPipelineAndVerify(pipelineSpecFilePath, &testContext.Pipeline.PipelineGeneratedName, nil)
+			Expect(createdPipeline.Tags).To(BeEmpty(), "Pipeline uploaded with empty tags should have empty tags")
 		})
 
 		It(fmt.Sprintf("Upload %s pipeline without tags and verify no tags are returned", helloWorldPipelineFileName), func() {
@@ -306,7 +307,11 @@ var _ = Describe("Verify Pipeline Upload Version with Tags Failure >", Label("Ne
 			Expect(err).NotTo(HaveOccurred())
 			parameters.Tags = tagsJSON
 
-			uploadPipelineVersionAndVerifyFailure(pipelineSpecFilePath, parameters, "Failed to upload pipeline version")
+			expectedErrMsg := "Failed to upload pipeline version"
+			if *config.UploadPipelinesWithKubernetes {
+				expectedErrMsg = "Failed to validate pipeline version tags"
+			}
+			uploadPipelineVersionAndVerifyFailure(pipelineSpecFilePath, parameters, expectedErrMsg)
 		})
 
 		It("Upload a pipeline version with tag value exceeding 63 characters", func() {
@@ -323,7 +328,11 @@ var _ = Describe("Verify Pipeline Upload Version with Tags Failure >", Label("Ne
 			Expect(err).NotTo(HaveOccurred())
 			parameters.Tags = tagsJSON
 
-			uploadPipelineVersionAndVerifyFailure(pipelineSpecFilePath, parameters, "Failed to upload pipeline version")
+			expectedErrMsg := "Failed to upload pipeline version"
+			if *config.UploadPipelinesWithKubernetes {
+				expectedErrMsg = "Failed to validate pipeline version tags"
+			}
+			uploadPipelineVersionAndVerifyFailure(pipelineSpecFilePath, parameters, expectedErrMsg)
 		})
 
 		It("Upload a pipeline version with invalid tags JSON format", func() {
@@ -337,7 +346,11 @@ var _ = Describe("Verify Pipeline Upload Version with Tags Failure >", Label("Ne
 			invalidJSON := "not-valid-json"
 			parameters.SetTags(&invalidJSON)
 
-			uploadPipelineVersionAndVerifyFailure(pipelineSpecFilePath, parameters, "Failed to upload pipeline version")
+			expectedErrMsg := "Failed to upload pipeline version"
+			if *config.UploadPipelinesWithKubernetes {
+				expectedErrMsg = "Failed to parse pipeline version tags"
+			}
+			uploadPipelineVersionAndVerifyFailure(pipelineSpecFilePath, parameters, expectedErrMsg)
 		})
 	})
 })
@@ -465,6 +478,13 @@ func uploadPipelineAndVerify(pipelineFilePath string, pipelineName *string, pipe
 	// Normalize CreatedAt to UTC to avoid timezone location mismatch in deep comparison
 	createdPipelineFromDB.CreatedAt = strfmt.DateTime(time.Time(createdPipelineFromDB.CreatedAt).UTC())
 	createdPipeline.CreatedAt = strfmt.DateTime(time.Time(createdPipeline.CreatedAt).UTC())
+	// Normalize empty tag maps to nil for stable comparisons across storage backends.
+	if len(createdPipelineFromDB.Tags) == 0 {
+		createdPipelineFromDB.Tags = nil
+	}
+	if len(createdPipeline.Tags) == 0 {
+		createdPipeline.Tags = nil
+	}
 	Expect(createdPipelineFromDB).To(Equal(*createdPipeline))
 	matcher.MatchPipelines(&createdPipelineFromDB, testContext.Pipeline.ExpectedPipeline)