feat(backend): Selectively block v1 recurring runs (#13302)

Signed-off-by: Monica Zhang <zhangmonica1@gmail.com>

Author: MonicaZhang1

backend/src/crd/controller/scheduledworkflow/controller.go

@@ -17,6 +17,8 @@ package main
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"strings"
 	"time"
 
 	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
@@ -33,6 +35,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	log "github.com/sirupsen/logrus"
+	"github.com/spf13/viper"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/structpb"
 	corev1 "k8s.io/api/core/v1"
@@ -577,6 +580,13 @@ func (c *Controller) submitNewWorkflowIfNotAlreadySubmitted(
 
 	// If the workflow is not found, we need to create it.
 	if swf.Spec.Workflow != nil && swf.Spec.Workflow.Spec != nil {
+		// V1 recurring runs bypass the API server by embedding the workflow spec directly in the ScheduledWorkflow CRD,
+		// so the V1 pipeline block needs to be enforced at the controller level as well.
+
+		if shouldEnforceV1Block(swf) {
+			return false, "", fmt.Errorf(
+				"namespace %s is not allowed to run v1 pipelines; please migrate to KFP v2 pipelines", swf.Namespace)
+		}
 		newWorkflow, err := swf.NewWorkflow(nextScheduledEpoch, nowEpoch)
 		if err != nil {
 			return false, "", err
@@ -689,3 +699,36 @@ func (c *Controller) updateStatus(
 	}
 	return nil
 }
+
+// isV1PipelineBlocked checks if the given namespace is blocked from running V1 pipelines
+// based on the BLOCK_V1_PIPELINES and V1_ALLOWED_NAMESPACES environment variables.
+func isV1PipelineBlocked(namespace string) bool {
+	blockV1Value := viper.GetString("BLOCK_V1_PIPELINES")
+	blockV1, err := strconv.ParseBool(blockV1Value)
+	if err != nil {
+		log.WithError(err).Warnf("Invalid BLOCK_V1_PIPELINES value %q; V1 pipelines are not blocked", blockV1Value)
+		blockV1 = false
+	}
+	if !blockV1 {
+		return false
+	}
+
+	allowedNamespaces := viper.GetString("V1_ALLOWED_NAMESPACES")
+	if allowedNamespaces == "" {
+		return true
+	}
+
+	targetNamespace := strings.ToLower(strings.TrimSpace(namespace))
+	for _, n := range strings.Split(allowedNamespaces, ",") {
+		if strings.ToLower(strings.TrimSpace(n)) == targetNamespace {
+			return false
+		}
+	}
+	return true
+}
+
+// shouldEnforceV1Block checks if the V1 pipeline block should be enforced for the given ScheduledWorkflow.
+// v2 ScheduledWorkflows can also have embedded workflow spec, but should not be blocked with this feature flag.
+func shouldEnforceV1Block(swf *util.ScheduledWorkflow) bool {
+	return strings.HasPrefix(swf.APIVersion, commonutil.ApiVersionV1) && isV1PipelineBlocked(swf.Namespace)
+}

backend/src/crd/controller/scheduledworkflow/controller_test.go

@@ -0,0 +1,150 @@
+// Copyright 2018 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 main
+
+import (
+	"testing"
+
+	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
+	commonutil "github.com/kubeflow/pipelines/backend/src/common/util"
+	util "github.com/kubeflow/pipelines/backend/src/crd/controller/scheduledworkflow/util"
+	swfapi "github.com/kubeflow/pipelines/backend/src/crd/pkg/apis/scheduledworkflow/v1beta1"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestIsV1PipelineBlocked(t *testing.T) {
+	tests := []struct {
+		name              string
+		blockV1           string
+		allowedNamespaces string
+		namespace         string
+		expected          bool
+	}{
+		{
+			name:      "Blocking disabled - not blocked",
+			blockV1:   "false",
+			namespace: "ns1",
+			expected:  false,
+		},
+		{
+			name:      "Blocking not set - not blocked",
+			blockV1:   "",
+			namespace: "ns1",
+			expected:  false,
+		},
+		{
+			name:      "Blocking enabled, no allowed namespaces - blocked",
+			blockV1:   "true",
+			namespace: "ns1",
+			expected:  true,
+		},
+		{
+			name:              "Blocking enabled, namespace allowed - not blocked",
+			blockV1:           "true",
+			allowedNamespaces: "ns1",
+			namespace:         "ns1",
+			expected:          false,
+		},
+		{
+			name:              "Blocking enabled, namespace not in allowed list - blocked",
+			blockV1:           "true",
+			allowedNamespaces: "ns2,ns3",
+			namespace:         "ns1",
+			expected:          true,
+		},
+		{
+			name:              "Blocking enabled, namespace in allowed list - not blocked",
+			blockV1:           "true",
+			allowedNamespaces: "ns1,ns2,ns3",
+			namespace:         "ns2",
+			expected:          false,
+		},
+		{
+			name:              "Blocking enabled, case insensitive namespace match - not blocked",
+			blockV1:           "true",
+			allowedNamespaces: "NS1",
+			namespace:         "ns1",
+			expected:          false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			viper.Set(common.BlockV1Pipelines, tt.blockV1)
+			viper.Set(common.V1NamespaceWhitelist, tt.allowedNamespaces)
+			defer func() {
+				viper.Set(common.BlockV1Pipelines, "")
+				viper.Set(common.V1NamespaceWhitelist, "")
+			}()
+
+			result := isV1PipelineBlocked(tt.namespace)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestShouldEnforceV1Block(t *testing.T) {
+	tests := []struct {
+		name       string
+		apiVersion string
+		blockV1    string
+		namespace  string
+		expected   bool
+	}{
+		{
+			name:       "V1 SWF, blocking enabled - blocked",
+			apiVersion: commonutil.ApiVersionV1,
+			blockV1:    "true",
+			namespace:  "ns1",
+			expected:   true,
+		},
+		{
+			name:       "V1 SWF, blocking disabled - not blocked",
+			apiVersion: commonutil.ApiVersionV1,
+			blockV1:    "false",
+			namespace:  "ns1",
+			expected:   false,
+		},
+		{
+			name:       "V2 SWF, blocking enabled - not blocked",
+			apiVersion: commonutil.ApiVersionV2,
+			blockV1:    "true",
+			namespace:  "ns1",
+			expected:   false,
+		},
+		{
+			name:       "V2 SWF, blocking disabled - not blocked",
+			apiVersion: commonutil.ApiVersionV2,
+			blockV1:    "false",
+			namespace:  "ns1",
+			expected:   false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			viper.Set(common.BlockV1Pipelines, tt.blockV1)
+			defer viper.Set(common.BlockV1Pipelines, "")
+
+			swf := util.NewScheduledWorkflow(&swfapi.ScheduledWorkflow{
+				TypeMeta:   metav1.TypeMeta{APIVersion: tt.apiVersion},
+				ObjectMeta: metav1.ObjectMeta{Namespace: tt.namespace},
+			})
+			assert.Equal(t, tt.expected, shouldEnforceV1Block(swf))
+		})
+	}
+}