Repository: numaproj/numaflow
Pull request #1358 (2023-11-14 21:54:06)

fix: add more checks to isbsvc validation
1. Use IsDNS1053Label to validate the ISBSVC name.
2. Replace gin ShouldBindJson with the json decoder to decode the request body to the spec object. json decoder supports disallowing unknown fields. This catches typos such as `replica`.
3. Disallow updating ISBSVC name and namespace from UI.

<!--

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/pkg/reconciler/isbsvc/controller.go b/pkg/reconciler/isbsvc/controller.go
index 87b2c754..89e83250 100644
--- a/pkg/reconciler/isbsvc/controller.go
+++ b/pkg/reconciler/isbsvc/controller.go
@@ -19,9 +19,6 @@ package isbsvc
 import (
 	"context"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/reconciler"
-	"github.com/numaproj/numaflow/pkg/reconciler/isbsvc/installer"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -31,6 +28,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
+	"github.com/numaproj/numaflow/pkg/reconciler/isbsvc/installer"
 )
 
 const (
diff --git a/pkg/reconciler/isbsvc/validate.go b/pkg/reconciler/isbsvc/validate.go
index 31e65b83..b9d4fa58 100644
--- a/pkg/reconciler/isbsvc/validate.go
+++ b/pkg/reconciler/isbsvc/validate.go
@@ -19,31 +19,39 @@ package isbsvc
 import (
 	"fmt"
 
+	k8svalidation "k8s.io/apimachinery/pkg/util/validation"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
-// ValidateInterStepBufferService accepts an isbs and performs validation against it
-func ValidateInterStepBufferService(isbs *dfv1.InterStepBufferService) error {
-	if isbs.Spec.Redis != nil && isbs.Spec.JetStream != nil {
+// ValidateInterStepBufferService accepts an ISB Service and performs validation against it
+func ValidateInterStepBufferService(isbsvc *dfv1.InterStepBufferService) error {
+	if isbsvc == nil {
+		return fmt.Errorf("nil ISB Service")
+	}
+	if errs := k8svalidation.IsDNS1035Label(isbsvc.Name); len(errs) > 0 {
+		return fmt.Errorf("invalid ISB Service name %q, %v", isbsvc.Name, errs)
+	}
+	if isbsvc.Spec.Redis != nil && isbsvc.Spec.JetStream != nil {
 		return fmt.Errorf(`invalid spec: "spec.redis" and "spec.jetstream" can not be defined together`)
 	}
-	if isbs.Spec.Redis == nil && isbs.Spec.JetStream == nil {
+	if isbsvc.Spec.Redis == nil && isbsvc.Spec.JetStream == nil {
 		return fmt.Errorf(`invalid spec: either "spec.redis" or "spec.jetstream" needs to be specified`)
 	}
-	if isbs.Spec.Redis != nil {
-		if isbs.Spec.Redis.Native != nil && isbs.Spec.Redis.External != nil {
+	if isbsvc.Spec.Redis != nil {
+		if isbsvc.Spec.Redis.Native != nil && isbsvc.Spec.Redis.External != nil {
 			return fmt.Errorf(`"native" and "external" can not be defined together`)
 		}
-		if isbs.Spec.Redis.Native == nil && isbs.Spec.Redis.External == nil {
+		if isbsvc.Spec.Redis.Native == nil && isbsvc.Spec.Redis.External == nil {
 			return fmt.Errorf(`either "native" or "external" must be defined`)
 		}
-		if native := isbs.Spec.Redis.Native; native != nil {
+		if native := isbsvc.Spec.Redis.Native; native != nil {
 			if native.Version == "" {
 				return fmt.Errorf(`invalid spec: "spec.redis.native.version" is not defined`)
 			}
 		}
 	}
-	if x := isbs.Spec.JetStream; x != nil {
+	if x := isbsvc.Spec.JetStream; x != nil {
 		if x.Version == "" {
 			return fmt.Errorf(`invalid spec: "spec.jetstream.version" is not defined`)
 		}
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index 7f53b340..c41dbf53 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -236,38 +236,34 @@ func (h *handler) GetClusterSummary(c *gin.Context) {
 // CreatePipeline is used to create a given pipeline
 func (h *handler) CreatePipeline(c *gin.Context) {
 	ns := c.Param("namespace")
-	// dryRun is used to check if the operation is just a validation or an actual create
+	// dryRun is used to check if the operation is just a validation or an actual creation
 	dryRun := strings.EqualFold("true", c.DefaultQuery("dry-run", "false"))
 
 	var pipelineSpec dfv1.Pipeline
-	err := c.ShouldBindJSON(&pipelineSpec)
-	if err != nil {
-		h.respondWithError(c, fmt.Sprintf("Failed to get JSON request body, %s", err.Error()))
+	if err := bindJson(c, &pipelineSpec); err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to decode JSON request body to pipeline spec, %s", err.Error()))
 		return
 	}
 
-	err = validateNamespace(h, &pipelineSpec, ns)
-	if err != nil {
-		h.respondWithError(c, err.Error())
+	if requestedNs := pipelineSpec.Namespace; !isValidNamespaceSpec(requestedNs, ns) {
+		h.respondWithError(c, fmt.Sprintf("namespace mismatch, expected %s, got %s", ns, requestedNs))
 		return
 	}
 	pipelineSpec.Namespace = ns
-	err = validatePipelineSpec(h, nil, &pipelineSpec, ValidTypeCreate)
-	if err != nil {
+
+	if err := validatePipelineSpec(h, nil, &pipelineSpec, ValidTypeCreate); err != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to validate pipeline spec, %s", err.Error()))
 		return
 	}
-	// if Validation flag "dryRun" is set to true, return without creating the pipeline
+	// if the validation flag "dryRun" is set to true, return without creating the pipeline
 	if dryRun {
 		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 		return
 	}
-
 	if _, err := h.numaflowClient.Pipelines(ns).Create(context.Background(), &pipelineSpec, metav1.CreateOptions{}); err != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to create pipeline %q, %s", pipelineSpec.Name, err.Error()))
 		return
 	}
-
 	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 }
 
@@ -378,32 +374,29 @@ func (h *handler) UpdatePipeline(c *gin.Context) {
 	}
 
 	var updatedSpec dfv1.Pipeline
-	err = c.ShouldBindJSON(&updatedSpec)
-	if err != nil {
-		h.respondWithError(c, fmt.Sprintf("Failed to get JSON request body, %s", err.Error()))
+	if err = bindJson(c, &updatedSpec); err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to decode JSON request body to pipeline spec, %s", err.Error()))
 		return
 	}
 
-	// Validate the namespace of the request
-	err = validateNamespace(h, &updatedSpec, ns)
-	if err != nil {
-		h.respondWithError(c, err.Error())
+	if requestedNs := updatedSpec.Namespace; !isValidNamespaceSpec(requestedNs, ns) {
+		h.respondWithError(c, fmt.Sprintf("namespace mismatch, expected %s, got %s", ns, requestedNs))
 		return
 	}
+	updatedSpec.Namespace = ns
 
-	// pipeline name in the URL should be same as spec name
+	// pipeline name in the URL should be the same as spec name
 	if pipeline != updatedSpec.Name {
 		h.respondWithError(c, fmt.Sprintf("pipeline name %q is immutable", pipeline))
 		return
 	}
 
-	updatedSpec.Namespace = ns
 	isValid := validatePipelineSpec(h, oldSpec, &updatedSpec, ValidTypeUpdate)
 	if isValid != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to update pipeline %q, %s", pipeline, isValid.Error()))
 		return
 	}
-	// If Validation flag is set to true, return without updating the pipeline
+	// If the validation flag is set to true, return without updating the pipeline
 	if dryRun {
 		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 		return
@@ -457,25 +450,29 @@ func (h *handler) PatchPipeline(c *gin.Context) {
 	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 }
 
-// CreateInterStepBufferService is used to create a given interstep buffer service
 func (h *handler) CreateInterStepBufferService(c *gin.Context) {
 	ns := c.Param("namespace")
 	// dryRun is used to check if the operation is just a validation or an actual update
 	dryRun := strings.EqualFold("true", c.DefaultQuery("dry-run", "false"))
 
 	var isbsvcSpec dfv1.InterStepBufferService
-	err := c.ShouldBindJSON(&isbsvcSpec)
-	if err != nil {
-		h.respondWithError(c, fmt.Sprintf("Failed to get JSON request body, %s", err.Error()))
+	if err := bindJson(c, &isbsvcSpec); err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to decode JSON request body to interstepbuffer service spec, %s", err.Error()))
+		return
+	}
+
+	if isbsvcNs := isbsvcSpec.Namespace; !isValidNamespaceSpec(isbsvcNs, ns) {
+		h.respondWithError(c, fmt.Sprintf("namespace mismatch, expected %s, got %s", ns, isbsvcNs))
 		return
 	}
+	isbsvcSpec.Namespace = ns
 
-	isValid := validateISBSVCSpec(h, nil, &isbsvcSpec, ValidTypeCreate)
+	isValid := validateISBSVCSpec(nil, &isbsvcSpec, ValidTypeCreate)
 	if isValid != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to create interstepbuffer service spec, %s", isValid.Error()))
 		return
 	}
-	// If Validation flag is set to true, return without creating the ISB
+	// If the validation flag is set to true, return without creating the ISB service
 	if dryRun {
 		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 		return
@@ -522,7 +519,6 @@ func (h *handler) GetInterStepBufferService(c *gin.Context) {
 	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, resp))
 }
 
-// UpdateInterStepBufferService is used to delete the inter-step buffer service
 func (h *handler) UpdateInterStepBufferService(c *gin.Context) {
 	ns, isbsvcName := c.Param("namespace"), c.Param("isb-service")
 	// dryRun is used to check if the operation is just a validation or an actual update
@@ -535,19 +531,17 @@ func (h *handler) UpdateInterStepBufferService(c *gin.Context) {
 	}
 
 	var updatedSpec dfv1.InterStepBufferService
-	err = c.ShouldBindJSON(&updatedSpec)
-	if err != nil {
-		h.respondWithError(c, fmt.Sprintf("Failed to get JSON request body, %s", err.Error()))
+	if err = bindJson(c, &updatedSpec); err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to decode JSON request body to interstepbuffer service spec, %s", err.Error()))
 		return
 	}
 
-	err = validateISBSVCSpec(h, isbSVC, &updatedSpec, ValidTypeUpdate)
-	if err != nil {
+	if err = validateISBSVCSpec(isbSVC, &updatedSpec, ValidTypeUpdate); err != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to validate interstepbuffer service spec, %s", err.Error()))
 		return
 	}
 
-	// If Validation flag is set to true, return without updating the ISB service
+	// If the validation flag is set to true, return without updating the ISB service
 	if dryRun {
 		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 		return
@@ -558,16 +552,13 @@ func (h *handler) UpdateInterStepBufferService(c *gin.Context) {
 		h.respondWithError(c, fmt.Sprintf("Failed to update the interstep buffer service: namespace %q isb-services %q: %s", ns, isbsvcName, err.Error()))
 		return
 	}
-
 	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, updatedISBSvc))
 }
 
-// DeleteInterStepBufferService is used to update the spec of the inter step buffer service
 func (h *handler) DeleteInterStepBufferService(c *gin.Context) {
 	ns, isbsvcName := c.Param("namespace"), c.Param("isb-service")
 
 	pipelines, err := h.numaflowClient.Pipelines(ns).List(context.Background(), metav1.ListOptions{})
-	// Get(context.Background(), pipeline, metav1.GetOptions{})
 	if err != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to get pipelines in namespace %q, %s", ns, err.Error()))
 		return
@@ -639,7 +630,7 @@ func (h *handler) UpdateVertex(c *gin.Context) {
 		inputVertexName = c.Param("vertex")
 		pipeline        = c.Param("pipeline")
 		ns              = c.Param("namespace")
-		// dryRun is used to check if the operation is just a validation or an actual create
+		// dryRun is used to check if the operation is just a validation or an actual creation
 		dryRun = strings.EqualFold("true", c.DefaultQuery("dry-run", "false"))
 	)
 
@@ -650,9 +641,8 @@ func (h *handler) UpdateVertex(c *gin.Context) {
 		return
 	}
 
-	if err = c.ShouldBindJSON(&requestBody); err != nil {
-		h.respondWithError(c, fmt.Sprintf("Failed to update the vertex: namespace %q pipeline %q vertex %q: %s", ns,
-			pipeline, inputVertexName, err.Error()))
+	if err = bindJson(c, &requestBody); err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to decode JSON request body to vertex spec, %s", err.Error()))
 		return
 	}
 
@@ -673,18 +663,13 @@ func (h *handler) UpdateVertex(c *gin.Context) {
 			break
 		}
 	}
-	err = validateNamespace(h, pl, ns)
-	if err != nil {
-		h.respondWithError(c, err.Error())
-		return
-	}
-	pl.Namespace = ns
+
 	err = validatePipelineSpec(h, nil, pl, ValidTypeCreate)
 	if err != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to validate pipeline spec, %s", err.Error()))
 		return
 	}
-	// if Validation flag "dryRun" is set to true, return without creating the pipeline
+	// if the validation flag "dryRun" is set to true, return without creating the pipeline
 	if dryRun {
 		c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, nil))
 		return
@@ -969,11 +954,20 @@ func validatePipelineSpec(h *handler, oldPipeline *dfv1.Pipeline, newPipeline *d
 }
 
 // validateISBSVCSpec is used to validate the ISB service spec
-func validateISBSVCSpec(h *handler, prevSpec *dfv1.InterStepBufferService,
+func validateISBSVCSpec(prevSpec *dfv1.InterStepBufferService,
 	newSpec *dfv1.InterStepBufferService, validType string) error {
-	ns := newSpec.Namespace
-	isbClient := h.numaflowClient.InterStepBufferServices(ns)
-	valid := validator.NewISBServiceValidator(h.kubeClient, isbClient, prevSpec, newSpec)
+	// UI-specific validations: updating namespace and name from UI is not allowed
+	if validType == ValidTypeUpdate && prevSpec != nil {
+		if !isValidNamespaceSpec(newSpec.Namespace, prevSpec.Namespace) {
+			return fmt.Errorf("updating an inter-step buffer service's namespace is not allowed, expected %s, got %s", prevSpec.Namespace, newSpec.Namespace)
+		}
+		if prevSpec.Name != newSpec.Name {
+			return fmt.Errorf("updating an inter-step buffer service's name is not allowed, expected %s, got %s", prevSpec.Name, newSpec.Name)
+		}
+	}
+
+	// the rest of the code leverages the webhook validator to validate the ISB service spec.
+	valid := validator.NewISBServiceValidator(prevSpec, newSpec)
 	var resp *admissionv1.AdmissionResponse
 	switch validType {
 	case ValidTypeCreate:
@@ -988,12 +982,19 @@ func validateISBSVCSpec(h *handler, prevSpec *dfv1.InterStepBufferService,
 	return nil
 }
 
-// validateNamespace is used to validate the namespace for a pipeline spec
-// For a request, the namespace provided as parameter should be same as the namespace in the pipeline spec
-func validateNamespace(h *handler, pipeline *dfv1.Pipeline, ns string) error {
-	if pipeline.Namespace != "" && pipeline.Namespace != ns {
-		errMsg := fmt.Errorf("namespace mismatch, expected %s", ns)
-		return errMsg
+// isValidNamespaceSpec validates
+// that the requested namespace should be either empty or the same as the current namespace
+func isValidNamespaceSpec(requested, current string) bool {
+	return requested == "" || requested == current
+}
+
+// bindJson is used to bind the request body to a given object
+// It also validates the request body to ensure that it does not contain any unknown fields
+func bindJson(c *gin.Context, obj interface{}) error {
+	decoder := json.NewDecoder(c.Request.Body)
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(obj); err != nil {
+		return err
 	}
 	return nil
 }
diff --git a/webhook/validator/isbsvc.go b/webhook/validator/isbsvc.go
index 93974ddf..0264f9af 100644
--- a/webhook/validator/isbsvc.go
+++ b/webhook/validator/isbsvc.go
@@ -20,39 +20,31 @@ import (
 	"context"
 
 	admissionv1 "k8s.io/api/admission/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"
 	isbsvccontroller "github.com/numaproj/numaflow/pkg/reconciler/isbsvc"
 )
 
 type isbsvcValidator struct {
-	client kubernetes.Interface
-	isbscv v1alpha1.InterStepBufferServiceInterface
-
 	oldISBService *dfv1.InterStepBufferService
 	newISBService *dfv1.InterStepBufferService
 }
 
-// NewISBServiceValidator returns ISBService validator
-func NewISBServiceValidator(client kubernetes.Interface, isbsvc v1alpha1.InterStepBufferServiceInterface, old, new *dfv1.InterStepBufferService) Validator {
-	return &isbsvcValidator{client: client, isbscv: isbsvc, oldISBService: old, newISBService: new}
+func NewISBServiceValidator(old, new *dfv1.InterStepBufferService) Validator {
+	return &isbsvcValidator{oldISBService: old, newISBService: new}
 }
 
-func (v *isbsvcValidator) ValidateCreate(ctx context.Context) *admissionv1.AdmissionResponse {
+func (v *isbsvcValidator) ValidateCreate(_ context.Context) *admissionv1.AdmissionResponse {
 	if err := isbsvccontroller.ValidateInterStepBufferService(v.newISBService); err != nil {
 		return DeniedResponse(err.Error())
 	}
 	return AllowedResponse()
 }
 
-func (v *isbsvcValidator) ValidateUpdate(ctx context.Context) *admissionv1.AdmissionResponse {
-
+func (v *isbsvcValidator) ValidateUpdate(_ context.Context) *admissionv1.AdmissionResponse {
 	if err := isbsvccontroller.ValidateInterStepBufferService(v.newISBService); err != nil {
 		return DeniedResponse(err.Error())
 	}
-
 	switch {
 	case v.oldISBService.Spec.JetStream != nil:
 		if v.newISBService.Spec.Redis != nil {
@@ -63,6 +55,5 @@ func (v *isbsvcValidator) ValidateUpdate(ctx context.Context) *admissionv1.Admis
 			return DeniedResponse("Can not change ISB Service type from Redis to Jetstream")
 		}
 	}
-
 	return AllowedResponse()
 }
diff --git a/webhook/validator/mock_isb_client.go b/webhook/validator/mock_isb_client.go
index ed79e4d8..2c85a362 100644
--- a/webhook/validator/mock_isb_client.go
+++ b/webhook/validator/mock_isb_client.go
@@ -32,7 +32,7 @@ type MockInterStepBufferServices struct {
 
 // Get takes name of the interStepBufferService, and returns the corresponding interStepBufferService object, and an error if there is any.
 func (c *MockInterStepBufferServices) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1alpha1.InterStepBufferService, err error) {
-	isbsvc := fakeISBSvc()
+	isbsvc := fakeRedisISBSvc()
 	isbsvc.Status.MarkDeployed()
 	return isbsvc, err
 }
diff --git a/webhook/validator/pipeline.go b/webhook/validator/pipeline.go
index 7aca54ae..2e1b0209 100644
--- a/webhook/validator/pipeline.go
+++ b/webhook/validator/pipeline.go
@@ -54,32 +54,28 @@ func (v *pipelineValidator) ValidateCreate(ctx context.Context) *admissionv1.Adm
 	if err := pipelinecontroller.ValidatePipeline(v.newPipeline); err != nil {
 		return DeniedResponse(err.Error())
 	}
-	// Check that the ISB service exists
+	// check that the ISB service exists
 	var isbName string
 	if v.newPipeline.Spec.InterStepBufferServiceName != "" {
 		isbName = v.newPipeline.Spec.InterStepBufferServiceName
 	} else {
 		isbName = dfv1.DefaultISBSvcName
 	}
-
 	if err := v.checkISBSVCExists(ctx, isbName); err != nil {
 		return DeniedResponse(err.Error())
 	}
 	return AllowedResponse()
 }
 
-func (v *pipelineValidator) ValidateUpdate(ctx context.Context) *admissionv1.AdmissionResponse {
-
-	// check that update is valid pipeline
+func (v *pipelineValidator) ValidateUpdate(_ context.Context) *admissionv1.AdmissionResponse {
+	// check that the new pipeline spec is valid
 	if err := pipelinecontroller.ValidatePipeline(v.newPipeline); err != nil {
 		return DeniedResponse(err.Error())
 	}
-
-	// can't change pipeline's isbsvc name
+	// 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")
 	}
-
 	return AllowedResponse()
 }
 
diff --git a/webhook/validator/validator.go b/webhook/validator/validator.go
index 52ff41f2..bbbf1a4d 100644
--- a/webhook/validator/validator.go
+++ b/webhook/validator/validator.go
@@ -57,8 +57,7 @@ func GetValidator(ctx context.Context, client kubernetes.Interface, NumaClient v
 				return nil, err
 			}
 		}
-		isbSvcClient := NumaClient.InterStepBufferServices(newSpec.Namespace)
-		return NewISBServiceValidator(client, isbSvcClient, oldSpec, newSpec), nil
+		return NewISBServiceValidator(oldSpec, newSpec), nil
 	case dfv1.PipelineGroupVersionKind.Kind:
 		var newSpec *dfv1.Pipeline
 		if len(newBytes) > 0 {
@@ -84,7 +83,7 @@ func GetValidator(ctx context.Context, client kubernetes.Interface, NumaClient v
 	}
 }
 
-// DeniedResponse constructs a denied AdmissionResonse
+// DeniedResponse constructs a denied AdmissionResponse
 func DeniedResponse(reason string, args ...interface{}) *admissionv1.AdmissionResponse {
 	result := apierrors.NewBadRequest(fmt.Sprintf(reason, args...)).Status()
 	return &admissionv1.AdmissionResponse{
@@ -93,7 +92,7 @@ func DeniedResponse(reason string, args ...interface{}) *admissionv1.AdmissionRe
 	}
 }
 
-// AllowedResponse constructs an allowed AdmissionResonse
+// AllowedResponse constructs an allowed AdmissionResponse
 func AllowedResponse() *admissionv1.AdmissionResponse {
 	return &admissionv1.AdmissionResponse{
 		Allowed: true,
diff --git a/webhook/webhook.go b/webhook/webhook.go
index 3ce926e7..274a14df 100644
--- a/webhook/webhook.go
+++ b/webhook/webhook.go
@@ -101,7 +101,7 @@ func (ac *AdmissionController) Run(ctx context.Context) error {
 	serverStartErrCh := make(chan struct{})
 	go func() {
 		if err := server.ListenAndServeTLS("", ""); err != nil {
-			logger.Errorw("ListenAndServeTLS for admission webhook erorred out", zap.Error(err))
+			logger.Errorw("ListenAndServeTLS for admission webhook errored out", zap.Error(err))
 			close(serverStartErrCh)
 		}
 	}()
@@ -111,7 +111,6 @@ func (ac *AdmissionController) Run(ctx context.Context) error {
 	case <-serverStartErrCh:
 		return fmt.Errorf("webhook server failed to start")
 	}
-
 }
 
 // Register registers the validating admission webhook


Test patch:
diff --git a/pkg/reconciler/isbsvc/controller_test.go b/pkg/reconciler/isbsvc/controller_test.go
index 9230cd6b..1192db8a 100644
--- a/pkg/reconciler/isbsvc/controller_test.go
+++ b/pkg/reconciler/isbsvc/controller_test.go
@@ -20,8 +20,6 @@ import (
 	"context"
 	"testing"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/reconciler"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 	appv1 "k8s.io/api/apps/v1"
@@ -33,6 +31,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 )
 
 const (
diff --git a/pkg/reconciler/isbsvc/validate_test.go b/pkg/reconciler/isbsvc/validate_test.go
index 15787b1f..b3a8269c 100644
--- a/pkg/reconciler/isbsvc/validate_test.go
+++ b/pkg/reconciler/isbsvc/validate_test.go
@@ -19,9 +19,10 @@ package isbsvc
 import (
 	"testing"
 
-	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
 var (
@@ -105,4 +106,18 @@ func TestValidateInterStepBuffer(t *testing.T) {
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "is not defined")
 	})
+
+	t.Run("test nil ISB Service", func(t *testing.T) {
+		err := ValidateInterStepBufferService(nil)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "nil ISB Service")
+	})
+
+	t.Run("test invalid name", func(t *testing.T) {
+		isbs := testJetStreamIsbs.DeepCopy()
+		isbs.Name = "UpperCaseName"
+		err := ValidateInterStepBufferService(isbs)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid ISB Service name")
+	})
 }
diff --git a/test/api-e2e/api_test.go b/test/api-e2e/api_test.go
index f945d0c9..a654068e 100644
--- a/test/api-e2e/api_test.go
+++ b/test/api-e2e/api_test.go
@@ -178,7 +178,7 @@ func (s *APISuite) TestPipeline0() {
 		Status(200).Body().Raw()
 	assert.Contains(s.T(), pausePipeline1, patchPipelineSuccessExpect)
 
-	resumePipeline1 := HTTPExpect(s.T(), "https://localhost:8443").PATCH(fmt.Sprintf("/api/v1/namespaces/%s/pipelines/%s", Namespace, testPipeline1Name)).WithBytes([]byte(testPipeline1Resume)).
+	resumePipeline1 := HTTPExpect(s.T(), "https://localhost:8443").PATCH(fmt.Sprintf("/api/v1/namespaces/%s/pipelines/%s", Namespace, testPipeline1Name)).WithBytes(testPipeline1Resume).
 		Expect().
 		Status(200).Body().Raw()
 	assert.Contains(s.T(), resumePipeline1, patchPipelineSuccessExpect)
diff --git a/webhook/validator/isbsvc_test.go b/webhook/validator/isbsvc_test.go
index a2ca05c4..a8fd630b 100644
--- a/webhook/validator/isbsvc_test.go
+++ b/webhook/validator/isbsvc_test.go
@@ -20,22 +20,34 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
 func TestValidateISBServiceCreate(t *testing.T) {
-	isbsvc := fakeISBSvc()
-	v := NewISBServiceValidator(fakeK8sClient, &fakeISBSvcClient, nil, isbsvc)
+	isbsvc := fakeRedisISBSvc()
+	v := NewISBServiceValidator(nil, isbsvc)
 	r := v.ValidateCreate(contextWithLogger(t))
 	assert.True(t, r.Allowed)
 }
 
 func TestValidateISBServiceUpdate(t *testing.T) {
-	isbsvc := fakeISBSvc()
-	t.Run("test ISBSvc spec change", func(t *testing.T) {
-		JetStreamISBSvc := fakeJetStreamISBSvc()
-		newISBSvc := JetStreamISBSvc.DeepCopy()
-		v := NewISBServiceValidator(fakeK8sClient, &fakeISBSvcClient, isbsvc, newISBSvc)
-		r := v.ValidateUpdate(contextWithLogger(t))
-		assert.False(t, r.Allowed)
-	})
+	testCases := []struct {
+		name string
+		old  *dfv1.InterStepBufferService
+		new  *dfv1.InterStepBufferService
+		want bool
+	}{
+		{name: "invalid new ISBSvc spec", old: fakeRedisISBSvc(), new: nil, want: false},
+		{name: "changing ISB Service type is not allowed - redis to jetstream", old: fakeRedisISBSvc(), new: fakeJetStreamISBSvc(), want: false},
+		{name: "changing ISB Service type is not allowed - jetstream to redis", old: fakeJetStreamISBSvc(), new: fakeRedisISBSvc(), want: false},
+		{name: "valid new ISBSvc spec", old: fakeRedisISBSvc(), new: fakeRedisISBSvc(), want: true},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			v := NewISBServiceValidator(tc.old, tc.new)
+			r := v.ValidateUpdate(contextWithLogger(t))
+			assert.Equal(t, tc.want, r.Allowed)
+		})
+	}
 }
diff --git a/webhook/validator/test_utils.go b/webhook/validator/test_utils.go
index f011f302..a300057c 100644
--- a/webhook/validator/test_utils.go
+++ b/webhook/validator/test_utils.go
@@ -14,12 +14,11 @@ const (
 
 var (
 	fakeK8sClient      = fakeClient.NewSimpleClientset()
-	fakeISBSvcClient   = fake.FakeInterStepBufferServices{}
 	fakePipelineClient = fake.FakePipelines{}
 	fakeNumaClient     = fake.FakeNumaflowV1alpha1{}
 )
 
-func fakeISBSvc() *dfv1.InterStepBufferService {
+func fakeRedisISBSvc() *dfv1.InterStepBufferService {
 	return &dfv1.InterStepBufferService{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: testNamespace,
@@ -38,7 +37,7 @@ func fakeISBSvc() *dfv1.InterStepBufferService {
 func fakeJetStreamISBSvc() *dfv1.InterStepBufferService {
 	return &dfv1.InterStepBufferService{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "test-ns",
+			Namespace: testNamespace,
 			Name:      dfv1.DefaultISBSvcName,
 		},
 		Spec: dfv1.InterStepBufferServiceSpec{
@@ -53,7 +52,7 @@ func fakePipeline() *dfv1.Pipeline {
 	return &dfv1.Pipeline{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-pl",
-			Namespace: "test-ns",
+			Namespace: testNamespace,
 		},
 		Spec: dfv1.PipelineSpec{
 			Vertices: []dfv1.AbstractVertex{
diff --git a/webhook/validator/validator_test.go b/webhook/validator/validator_test.go
index 62137129..3b729f91 100644
--- a/webhook/validator/validator_test.go
+++ b/webhook/validator/validator_test.go
@@ -34,7 +34,7 @@ func contextWithLogger(t *testing.T) context.Context {
 
 func TestGetValidator(t *testing.T) {
 	t.Run("test get InterStepBufferService validator", func(t *testing.T) {
-		bytes, err := json.Marshal(fakeISBSvc())
+		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)