Repository: numaproj/numaflow
Pull request #1379 (2023-11-20 15:05:49)

fix: add pipeline update validation checks
* Should not update the type of an existing vertex.
* Should not change the storage strategy of a reduce vertex.
* Should not change the partition count of a reduce vertex.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/docs/operations/validating-webhook.md b/docs/operations/validating-webhook.md
index 7df88fd4..d1d0199b 100644
--- a/docs/operations/validating-webhook.md
+++ b/docs/operations/validating-webhook.md
@@ -45,4 +45,18 @@ to:
 Resource: "numaflow.numaproj.io/v1alpha1, Resource=pipelines", GroupVersionKind: "numaflow.numaproj.io/v1alpha1, Kind=Pipeline"
 Name: "simple-pipeline", Namespace: "numaflow-system"
 for: "examples/1-simple-pipeline.yaml": error when patching "examples/1-simple-pipeline.yaml": admission webhook "webhook.numaflow.numaproj.io" denied the request: Cannot update pipeline with different interStepBufferServiceName
-```
\ No newline at end of file
+```
+
+Other validations include:
+
+Pipeline:
+
+1. cannot change the type of an existing vertex 
+2. cannot change the partition count of a reduce vertex
+3. cannot change the storage class of a reduce vertex
+4. etc.
+
+InterStepBufferService:
+
+1. cannot change the persistence configuration of an ISB Service
+2. etc.
\ No newline at end of file
diff --git a/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md b/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md
index d9fe734a..9262a45a 100644
--- a/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md
+++ b/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md
@@ -56,7 +56,7 @@ window-based partition into smaller partitions by adding keys along with the
 window will help us horizontally scale the distribution.
 
 Keyed partitions are heavily used to aggregate data and are frequently seen
-throughout the processing pipeline. We could also convert and non-keyed problem
+throughout the processing pipeline. We could also convert a non-keyed problem
 to a set of keyed problems and apply a non-keyed function at the end. This will
 help solve the original problem in a scalable manner without affecting the
 result's completeness and/or accuracy.
@@ -66,7 +66,7 @@ vertex for parallel processing.
 
 ### Usage
 
-Numaflow support both Keyed and Non-Keyed windows. We set `keyed` to either
+Numaflow supports both Keyed and Non-Keyed windows. We set `keyed` to either
 `true` (keyed) or `false` (non-keyed). Please note that the non-keyed windows
 are not horizontally scalable as mentioned above.
 
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index 95fdca78..a5c0146c 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -655,11 +655,6 @@ func (h *handler) UpdateVertex(c *gin.Context) {
 	newPipelineSpec := oldPipelineSpec.DeepCopy()
 	for index, vertex := range newPipelineSpec.Spec.Vertices {
 		if vertex.Name == inputVertexName {
-			if vertex.GetVertexType() != requestBody.GetVertexType() {
-				h.respondWithError(c, fmt.Sprintf("Failed to update the vertex: namespace %q pipeline %q vertex %q: vertex type is immutable",
-					ns, pipeline, inputVertexName))
-				return
-			}
 			newPipelineSpec.Spec.Vertices[index] = requestBody
 			break
 		}
@@ -938,9 +933,8 @@ func getIsbServiceStatus(isbsvc *dfv1.InterStepBufferService) (string, error) {
 // validatePipelineSpec is used to validate the pipeline spec during create and update
 func validatePipelineSpec(h *handler, oldPipeline *dfv1.Pipeline, newPipeline *dfv1.Pipeline, validType string) error {
 	ns := newPipeline.Namespace
-	pipeClient := h.numaflowClient.Pipelines(ns)
 	isbClient := h.numaflowClient.InterStepBufferServices(ns)
-	valid := validator.NewPipelineValidator(h.kubeClient, pipeClient, isbClient, oldPipeline, newPipeline)
+	valid := validator.NewPipelineValidator(isbClient, oldPipeline, newPipeline)
 	var resp *admissionv1.AdmissionResponse
 	switch validType {
 	case ValidTypeCreate:
diff --git a/webhook/validator/pipeline.go b/webhook/validator/pipeline.go
index 2e1b0209..bd7758d0 100644
--- a/webhook/validator/pipeline.go
+++ b/webhook/validator/pipeline.go
@@ -21,8 +21,8 @@ import (
 	"fmt"
 
 	admissionv1 "k8s.io/api/admission/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/client/clientset/versioned/typed/numaflow/v1alpha1"
@@ -30,20 +30,14 @@ import (
 )
 
 type pipelineValidator struct {
-	client    kubernetes.Interface
-	pipeline  v1alpha1.PipelineInterface
-	isbClient v1alpha1.InterStepBufferServiceInterface
-
+	isbClient   v1alpha1.InterStepBufferServiceInterface
 	oldPipeline *dfv1.Pipeline
 	newPipeline *dfv1.Pipeline
 }
 
 // NewPipelineValidator returns a new PipelineValidator
-func NewPipelineValidator(client kubernetes.Interface, pipeline v1alpha1.PipelineInterface,
-	isbClient v1alpha1.InterStepBufferServiceInterface, old, new *dfv1.Pipeline) Validator {
+func NewPipelineValidator(isbClient v1alpha1.InterStepBufferServiceInterface, old, new *dfv1.Pipeline) Validator {
 	return &pipelineValidator{
-		client:      client,
-		pipeline:    pipeline,
 		isbClient:   isbClient,
 		oldPipeline: old,
 		newPipeline: new,
@@ -68,13 +62,16 @@ func (v *pipelineValidator) ValidateCreate(ctx context.Context) *admissionv1.Adm
 }
 
 func (v *pipelineValidator) ValidateUpdate(_ context.Context) *admissionv1.AdmissionResponse {
+	if v.oldPipeline == nil {
+		return DeniedResponse("old pipeline spec is nil")
+	}
 	// check that the new pipeline spec is valid
 	if err := pipelinecontroller.ValidatePipeline(v.newPipeline); err != nil {
-		return DeniedResponse(err.Error())
+		return DeniedResponse(fmt.Sprintf("new pipeline spec is invalid: %s", err.Error()))
 	}
-	// check that the ISB service name is NOT changed
-	if v.newPipeline.Spec.InterStepBufferServiceName != v.oldPipeline.Spec.InterStepBufferServiceName {
-		return DeniedResponse("Cannot update pipeline with different interStepBufferServiceName")
+	// check that the update is valid
+	if err := validatePipelineUpdate(v.oldPipeline, v.newPipeline); err != nil {
+		return DeniedResponse(err.Error())
 	}
 	return AllowedResponse()
 }
@@ -90,3 +87,45 @@ func (v *pipelineValidator) checkISBSVCExists(ctx context.Context, isbSvcName st
 	}
 	return nil
 }
+
+// validatePipelineUpdate validates the update of a pipeline
+func validatePipelineUpdate(old, new *dfv1.Pipeline) error {
+	// rule 1: the ISB service name shall not change
+	if new.Spec.InterStepBufferServiceName != old.Spec.InterStepBufferServiceName {
+		return fmt.Errorf("cannot update pipeline with different ISB service name")
+	}
+	// rule 2: if a vertex is updated, the update must be valid
+	// we consider that a vertex is updated if its name is the same but its spec is different
+	nameMap := make(map[string]dfv1.AbstractVertex)
+	for _, v := range old.Spec.Vertices {
+		nameMap[v.Name] = v
+	}
+	for _, v := range new.Spec.Vertices {
+		if oldV, ok := nameMap[v.Name]; ok {
+			if err := validateVertexUpdate(oldV, v); err != nil {
+				return err
+			}
+		}
+	}
+	// TODO - rule 3: if the structure of the pipeline is updated, the update must be valid
+	// example of structure change can be a vertex is added or removed, an edge is added or removed or updated, etc.
+	// for now, we consider structure change as valid
+	return nil
+}
+
+// validateVertexUpdate validates the update of a vertex
+func validateVertexUpdate(old, new dfv1.AbstractVertex) error {
+	if o, n := old.GetVertexType(), new.GetVertexType(); o != n {
+		return fmt.Errorf("vertex type is immutable, vertex name: %s, expected type %s, got type %s", old.Name, o, n)
+	}
+	if old.GetVertexType() == dfv1.VertexTypeReduceUDF {
+		if o, n := old.GetPartitionCount(), new.GetPartitionCount(); o != n {
+			return fmt.Errorf("partition count is immutable for a reduce vertex, vertex name: %s, expected partition count %d, got partition count %d", old.Name, o, n)
+		}
+		// with both old and new vertex specs being reducer, we can safely assume that the GroupBy field is not nil
+		if o, n := old.UDF.GroupBy.Storage, new.UDF.GroupBy.Storage; !equality.Semantic.DeepEqual(o, n) {
+			return fmt.Errorf("storage is immutable for a reduce vertex, vertex name: %s, expected storage %v, got storage %v", old.Name, o, n)
+		}
+	}
+	return nil
+}
diff --git a/webhook/validator/validator.go b/webhook/validator/validator.go
index bbbf1a4d..d5f2e866 100644
--- a/webhook/validator/validator.go
+++ b/webhook/validator/validator.go
@@ -24,7 +24,6 @@ import (
 	admissionv1 "k8s.io/api/admission/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/client/clientset/versioned/typed/numaflow/v1alpha1"
@@ -37,7 +36,7 @@ type Validator interface {
 }
 
 // GetValidator returns a Validator instance
-func GetValidator(ctx context.Context, client kubernetes.Interface, NumaClient v1alpha1.NumaflowV1alpha1Interface, kind metav1.GroupVersionKind, oldBytes []byte, newBytes []byte) (Validator, error) {
+func GetValidator(ctx context.Context, NumaClient v1alpha1.NumaflowV1alpha1Interface, kind metav1.GroupVersionKind, oldBytes []byte, newBytes []byte) (Validator, error) {
 	log := logging.FromContext(ctx)
 	switch kind.Kind {
 	case dfv1.ISBGroupVersionKind.Kind:
@@ -76,8 +75,7 @@ func GetValidator(ctx context.Context, client kubernetes.Interface, NumaClient v
 			}
 		}
 		isbSvcClient := NumaClient.InterStepBufferServices(newSpec.Namespace)
-		pipelineClient := NumaClient.Pipelines(newSpec.Namespace)
-		return NewPipelineValidator(client, pipelineClient, isbSvcClient, oldSpec, newSpec), nil
+		return NewPipelineValidator(isbSvcClient, oldSpec, newSpec), nil
 	default:
 		return nil, fmt.Errorf("unrecognized kind: %v", kind)
 	}
diff --git a/webhook/webhook.go b/webhook/webhook.go
index a905e25b..df86874a 100644
--- a/webhook/webhook.go
+++ b/webhook/webhook.go
@@ -56,7 +56,7 @@ const (
 	certOrg = "io.numaproj"
 )
 
-// Webhook configuration
+// Options is the webhook configuration
 type Options struct {
 	WebhookName     string
 	ServiceName     string
@@ -68,7 +68,7 @@ type Options struct {
 	ClientAuth      tls.ClientAuthType
 }
 
-// Controller for validation webhook
+// AdmissionController is the validating admission webhook controller
 type AdmissionController struct {
 	Client     kubernetes.Interface
 	NumaClient v1alpha1.NumaflowV1alpha1Interface
@@ -256,7 +256,7 @@ func (ac *AdmissionController) admit(ctx context.Context, request *admissionv1.A
 		log.Infof("Operation not interested: %v %v", request.Kind, request.Operation)
 		return &admissionv1.AdmissionResponse{Allowed: true}
 	}
-	v, err := validator.GetValidator(ctx, ac.Client, ac.NumaClient, request.Kind, request.OldObject.Raw, request.Object.Raw)
+	v, err := validator.GetValidator(ctx, ac.NumaClient, request.Kind, request.OldObject.Raw, request.Object.Raw)
 	if err != nil {
 		return validator.DeniedResponse("failed to get a validator: %v", err)
 	}


Test patch:
diff --git a/pkg/reconciler/pipeline/validate_test.go b/pkg/reconciler/pipeline/validate_test.go
index 38f376a6..64d170b2 100644
--- a/pkg/reconciler/pipeline/validate_test.go
+++ b/pkg/reconciler/pipeline/validate_test.go
@@ -87,7 +87,7 @@ var (
 							Window: dfv1.Window{
 								Fixed: &dfv1.FixedWindow{
 									Length: &metav1.Duration{
-										Duration: time.Duration(60 * time.Second),
+										Duration: 60 * time.Second,
 									},
 								},
 							},
@@ -112,7 +112,7 @@ var (
 							Window: dfv1.Window{
 								Fixed: &dfv1.FixedWindow{
 									Length: &metav1.Duration{
-										Duration: time.Duration(60 * time.Second),
+										Duration: 60 * time.Second,
 									},
 								},
 							},
diff --git a/webhook/validator/pipeline_test.go b/webhook/validator/pipeline_test.go
index e485a67f..80fbc9f9 100644
--- a/webhook/validator/pipeline_test.go
+++ b/webhook/validator/pipeline_test.go
@@ -20,12 +20,14 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
 func TestValidatePipelineCreate(t *testing.T) {
 	pipeline := fakePipeline()
 	fk := MockInterStepBufferServices{}
-	v := NewPipelineValidator(fakeK8sClient, &fakePipelineClient, &fk, nil, pipeline)
+	v := NewPipelineValidator(&fk, nil, pipeline)
 	r := v.ValidateCreate(contextWithLogger(t))
 	assert.True(t, r.Allowed)
 }
@@ -33,11 +35,57 @@ func TestValidatePipelineCreate(t *testing.T) {
 func TestValidatePipelineUpdate(t *testing.T) {
 	pipeline := fakePipeline()
 	fk := MockInterStepBufferServices{}
-	t.Run("test Pipeline interStepBufferServiceName change", func(t *testing.T) {
+	t.Run("test old pipeline spec is nil", func(t *testing.T) {
+		v := NewPipelineValidator(&fk, nil, pipeline)
+		r := v.ValidateUpdate(contextWithLogger(t))
+		assert.False(t, r.Allowed)
+		assert.Contains(t, r.Result.Message, "old pipeline spec is nil")
+	})
+	t.Run("test invalid new pipeline spec", func(t *testing.T) {
+		v := NewPipelineValidator(&fk, pipeline, nil)
+		r := v.ValidateUpdate(contextWithLogger(t))
+		assert.False(t, r.Allowed)
+		assert.Contains(t, r.Result.Message, "new pipeline spec is invalid")
+	})
+	t.Run("test pipeline interStepBufferServiceName change", func(t *testing.T) {
 		newPipeline := pipeline.DeepCopy()
 		newPipeline.Spec.InterStepBufferServiceName = "change-name"
-		v := NewPipelineValidator(fakeK8sClient, &fakePipelineClient, &fk, pipeline, newPipeline)
+		v := NewPipelineValidator(&fk, pipeline, newPipeline)
+		r := v.ValidateUpdate(contextWithLogger(t))
+		assert.False(t, r.Allowed)
+		assert.Contains(t, r.Result.Message, "different ISB service name")
+	})
+	t.Run("test should not change the type of a vertex", func(t *testing.T) {
+		newPipeline := pipeline.DeepCopy()
+		// in our test fake pipeline, the 3nd vertex is a reduce vertex
+		// change it to a map vertex, this ensures that the new pipeline is still valid but the update is not allowed
+		newPipeline.Spec.Vertices[2].UDF = &dfv1.UDF{
+			Builtin: &dfv1.Function{Name: "cat"},
+		}
+		v := NewPipelineValidator(&fk, pipeline, newPipeline)
+		r := v.ValidateUpdate(contextWithLogger(t))
+		assert.False(t, r.Allowed)
+		assert.Contains(t, r.Result.Message, "vertex type is immutable")
+	})
+	t.Run("test should not change the partition count of a reduce vertex", func(t *testing.T) {
+		var oldPartitionCount, newPartitionCount int32 = 2, 3
+		newPipeline := pipeline.DeepCopy()
+		// in our test fake pipeline, the 3rd vertex is a reduce vertex
+		pipeline.Spec.Vertices[2].Partitions = &oldPartitionCount
+		newPipeline.Spec.Vertices[2].Partitions = &newPartitionCount
+		v := NewPipelineValidator(&fk, pipeline, newPipeline)
+		r := v.ValidateUpdate(contextWithLogger(t))
+		assert.False(t, r.Allowed)
+		assert.Contains(t, r.Result.Message, "partition count is immutable for a reduce vertex")
+	})
+	t.Run("test should not change the persistent storage of a reduce vertex", func(t *testing.T) {
+		newPipeline := pipeline.DeepCopy()
+		newPipeline.Spec.Vertices[2].UDF.GroupBy.Storage = &dfv1.PBQStorage{
+			PersistentVolumeClaim: &dfv1.PersistenceStrategy{},
+		}
+		v := NewPipelineValidator(&fk, pipeline, newPipeline)
 		r := v.ValidateUpdate(contextWithLogger(t))
 		assert.False(t, r.Allowed)
+		assert.Contains(t, r.Result.Message, "storage is immutable for a reduce vertex")
 	})
 }
diff --git a/webhook/validator/test_utils.go b/webhook/validator/test_utils.go
index 7e5f08ae..4e6a74b4 100644
--- a/webhook/validator/test_utils.go
+++ b/webhook/validator/test_utils.go
@@ -1,20 +1,18 @@
 package validator
 
 import (
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	fakeClient "k8s.io/client-go/kubernetes/fake"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/client/clientset/versioned/typed/numaflow/v1alpha1/fake"
 )
 
-const (
-	testNamespace = "test-ns"
-)
+const testNamespace = "test-ns"
 
 var (
-	fakeK8sClient        = fakeClient.NewSimpleClientset()
-	fakePipelineClient   = fake.FakePipelines{}
 	fakeNumaClient       = fake.FakeNumaflowV1alpha1{}
 	testStorageClassName = "test-sc"
 )
@@ -71,19 +69,39 @@ func fakePipeline() *dfv1.Pipeline {
 						}},
 				},
 				{
-					Name: "p1",
+					Name: "map",
 					UDF: &dfv1.UDF{
 						Builtin: &dfv1.Function{Name: "cat"},
 					},
 				},
+				{
+					Name: "reduce",
+					UDF: &dfv1.UDF{
+						Container: &dfv1.Container{
+							Image: "test-image",
+						},
+						GroupBy: &dfv1.GroupBy{
+							Window: dfv1.Window{
+								Fixed: &dfv1.FixedWindow{Length: &metav1.Duration{
+									Duration: 60 * time.Second,
+								}},
+							},
+							Keyed: true,
+							Storage: &dfv1.PBQStorage{
+								EmptyDir: &corev1.EmptyDirVolumeSource{},
+							},
+						},
+					},
+				},
 				{
 					Name: "output",
 					Sink: &dfv1.Sink{},
 				},
 			},
 			Edges: []dfv1.Edge{
-				{From: "input", To: "p1"},
-				{From: "p1", To: "output"},
+				{From: "input", To: "map"},
+				{From: "map", To: "reduce"},
+				{From: "reduce", To: "output"},
 			},
 		},
 	}
diff --git a/webhook/validator/validator_test.go b/webhook/validator/validator_test.go
index 3b729f91..6d6fda3c 100644
--- a/webhook/validator/validator_test.go
+++ b/webhook/validator/validator_test.go
@@ -37,7 +37,7 @@ func TestGetValidator(t *testing.T) {
 		bytes, err := json.Marshal(fakeRedisISBSvc())
 		assert.NoError(t, err)
 		assert.NotNil(t, bytes)
-		v, err := GetValidator(contextWithLogger(t), fakeK8sClient, &fakeNumaClient, metav1.GroupVersionKind{Group: "numaflow.numaproj.io", Version: "v1alpha1", Kind: "InterStepBufferService"}, nil, bytes)
+		v, err := GetValidator(contextWithLogger(t), &fakeNumaClient, metav1.GroupVersionKind{Group: "numaflow.numaproj.io", Version: "v1alpha1", Kind: "InterStepBufferService"}, nil, bytes)
 		assert.NoError(t, err)
 		assert.NotNil(t, v)
 	})
@@ -46,7 +46,7 @@ func TestGetValidator(t *testing.T) {
 		bytes, err := json.Marshal(fakePipeline())
 		assert.NoError(t, err)
 		assert.NotNil(t, bytes)
-		v, err := GetValidator(contextWithLogger(t), fakeK8sClient, &fakeNumaClient, metav1.GroupVersionKind{Group: "numaflow.numaproj.io", Version: "v1alpha1", Kind: "Pipeline"}, nil, bytes)
+		v, err := GetValidator(contextWithLogger(t), &fakeNumaClient, metav1.GroupVersionKind{Group: "numaflow.numaproj.io", Version: "v1alpha1", Kind: "Pipeline"}, nil, bytes)
 		assert.NoError(t, err)
 		assert.NotNil(t, v)
 	})