fix: pdb validity check & deletion (#599)

# Description
## Background
When creating Pod Disruption Budget, Kubernetes gives two ways for user
to specify the allowed disruption amount, use `minAvailable` or
`maxUnavailable` . The example of behavior is as below:
```
current replica = 2

Use MinAvailable:
min available = 80%
allowed disruption = current replica - min available replica = 2 - ceil(80% * 2) = 2 - 2 = 0 

Use MaxUnavailable:
max unavailable = 20%
allowed disruption = ceil(20% * 2) = ceil(0.4) = 1
```

## Issue 

### 1. Miscalculation on PDB validity check

Merlin do a validity check on the PDB configuration, if the
configuration doesn’t allow for any disruption, then Merlin will not
create the PDB, this to avoid Kubernetes can’t remove any replica
because all replica must be up. When the configuration in Merlin use the
`maxUnavailability`, the calculation will be
([ref1](https://github.com/caraml-dev/merlin/blob/8c4930dc2c3f3edf3f1a862a99dd3cb5730c50cb/api/cluster/pdb.go#L88)
&
[ref2](https://github.com/caraml-dev/merlin/blob/8c4930dc2c3f3edf3f1a862a99dd3cb5730c50cb/api/cluster/pdb.go#L94-L96)):
```
minPercentage = (100 - maxUnavailability)%

if not (minPercentage * minReplica < minReplica)
	don't create PDB 
```
Which doesn’t reflect the real calculation on Kubernetes. Ex: 
```
replica: 5
maxUnavailability: 10%

Kubernetes calculation:
allowed disruption = ceil(10% * 5) = ceil(0.5) = 1

Merlin calculation:
minPercentage = (100 - 10)% = 90%
minAvailableReplica = ceil(90% * 5) = ceil(4.5) = 5 
allowed disruption = 5 - 5 = 0 
```
And therefore in the example above, Merlin will not create any PDB
configuration because it thinks that if the PDB is created, it might not
allowed any pod to be disrupted. Merlin avoid this because in the event
of node scale down, the pods can’t be removed and causing deadlock. But
actually, the Kubernetes calculation does allow for 1 pod to be
disrupted.

### 2. Unused PDB not deleted
When a model is redeployed, it will create a new revision ID, and
there's a chance that unused PDB for the previous revision is not
deleted. Scenario:
1. Current state of model: has predictor & transformer PDB 
2. User redeploy a new model version and this version doesn't have PDB
3. Newer model version is deployed, old unused PDB is not deleted
(because Merlin doesn't check or delete if previously PDB exist)

**Notes:** the old PDB will not affect the new model deployment, because
in the `selector.matchLabels` there's a label of
`inferenceservice:{modelName}-{versionID}-{revisionID}`, and the
revisionID will be incremented on every new deployment. So it is safe to
not delete PDB, but it will going to confuse user.


# Modifications
Changes: 

- Fix the validity checking of the PDB. The `minAvailable` will use the
current behavior, but if Merlin config uses `maxUnavailable` it will
create PDB as long as the number is bigger than 0.
- Merlin will create PDB with a fix name of
`{name}-{versionID}-{componentType}-pdb`, where `componentType` is
either `predictor` or `transformer`. The changes in the code will
compare those two fix names with the new PDBs name, then for name that
doesn't exist in new PDBs name, delete it.

# Tests
<!-- Besides the existing / updated automated tests, what specific
scenarios should be tested? Consider the backward compatibility of the
changes, whether corner cases are covered, etc. Please describe the
tests and check the ones that have been completed. Eg:
- [x] Deploying new and existing standard models
- [ ] Deploying PyFunc models
-->

# Checklist
- [x] Added PR label
- [x] Added unit test, integration, and/or e2e tests
- [x] Tested locally
- [ ] Updated documentation
- [ ] Update Swagger spec if the PR introduce API changes
- [ ] Regenerated Golang and Python client if the PR introduces API
changes

# Release Notes
<!--
Does this PR introduce a user-facing change?
If no, just write "NONE" in the release-note block below.
If yes, a release note is required. Enter your extended release note in
the block below.
If the PR requires additional action from users switching to the new
release, include the string "action required".

For more information about release notes, see kubernetes' guide here:
http://git.k8s.io/community/contributors/guide/release-notes.md
-->

```release-note
fix wrong validation check of PDB and delete unused PDB from previous version model
```

api/cluster/controller.go

@@ -288,12 +288,21 @@ func (c *controller) Deploy(ctx context.Context, modelService *models.Service) (
 		inferenceURL = vsCfg.getInferenceURL(vs)
 	}
 
-	// Delete previous inference service
+	// Delete previous inference service and pdb
 	if modelService.CurrentIsvcName != "" {
 		if err := c.deleteInferenceService(ctx, modelService.CurrentIsvcName, modelService.Namespace); err != nil {
 			log.Errorf("unable to delete prevision revision %s with error %v", modelService.CurrentIsvcName, err)
 			return nil, errors.Wrapf(err, fmt.Sprintf("%v (%s)", ErrUnableToDeletePreviousInferenceService, modelService.CurrentIsvcName))
 		}
+
+		unusedPdbs, err := c.getUnusedPodDisruptionBudgets(ctx, modelService)
+		if err != nil {
+			log.Warnf("unable to get model name %s, version %s unused pdb: %v", modelService.ModelName, modelService.ModelVersion, err)
+		} else {
+			if err := c.deletePodDisruptionBudgets(ctx, unusedPdbs); err != nil {
+				log.Warnf("unable to delete model name %s, version %s unused pdb: %v", modelService.ModelName, modelService.ModelVersion, err)
+			}
+		}
 	}
 
 	return &models.Service{

api/cluster/controller_test.go

@@ -353,7 +353,7 @@ func TestController_DeployInferenceService_NamespaceCreation(t *testing.T) {
 }
 
 func TestController_DeployInferenceService(t *testing.T) {
-	defaultMaxUnavailablePDB := 20
+	minAvailablePercentage := 80
 	deployTimeout := 2 * tickDurationSecond * time.Second
 
 	model := &models.Model{
@@ -729,8 +729,8 @@ func TestController_DeployInferenceService(t *testing.T) {
 				DefaultModelResourceRequests:       &config.ResourceRequests{},
 				DefaultTransformerResourceRequests: &config.ResourceRequests{},
 				PodDisruptionBudget: config.PodDisruptionBudgetConfig{
-					Enabled:                  true,
-					MaxUnavailablePercentage: &defaultMaxUnavailablePDB,
+					Enabled:                true,
+					MinAvailablePercentage: &minAvailablePercentage,
 				},
 				StandardTransformer: config.StandardTransformerConfig{
 					ImageName:             "ghcr.io/caraml-dev/merlin-transformer-test",
@@ -758,6 +758,223 @@ func TestController_DeployInferenceService(t *testing.T) {
 	}
 }
 
+func TestController_DeployInferenceService_PodDisruptionBudgetsRemoval(t *testing.T) {
+	err := models.InitKubernetesLabeller("gojek.com/", "dev")
+	assert.Nil(t, err)
+
+	minAvailablePercentage := 80
+
+	model := &models.Model{
+		Name: "my-model",
+	}
+	project := mlp.Project{
+		Name: "my-project",
+	}
+	version := &models.Version{
+		ID: 1,
+	}
+	revisionID := models.ID(5)
+
+	isvcName := models.CreateInferenceServiceName(model.Name, version.ID.String(), revisionID.String())
+	statusReady := createServiceReadyStatus(isvcName, project.Name, baseUrl)
+	statusReady.Components = make(map[kservev1beta1.ComponentType]kservev1beta1.ComponentStatusSpec)
+	pdb := &policyv1.PodDisruptionBudget{}
+	vs := fakeVirtualService(model.Name, version.ID.String())
+
+	metadata := models.Metadata{
+		App:       "mymodel",
+		Component: models.ComponentModelVersion,
+		Stream:    "mystream",
+		Team:      "myteam",
+	}
+	modelSvc := &models.Service{
+		Name:            isvcName,
+		ModelName:       model.Name,
+		ModelVersion:    version.ID.String(),
+		RevisionID:      revisionID,
+		Namespace:       project.Name,
+		Metadata:        metadata,
+		CurrentIsvcName: isvcName,
+	}
+
+	defaultMatchLabel := map[string]string{
+		"gojek.com/app":                      "mymodel",
+		"gojek.com/component":                "model-version",
+		"gojek.com/environment":              "dev",
+		"gojek.com/orchestrator":             "merlin",
+		"gojek.com/stream":                   "mystream",
+		"gojek.com/team":                     "myteam",
+		"serving.kserve.io/inferenceservice": "mymodel-1-r4",
+		"model-version-id":                   "1",
+	}
+
+	tests := []struct {
+		name             string
+		modelService     *models.Service
+		existingPdbs     *policyv1.PodDisruptionBudgetList
+		createPdbResult  *pdbReactor
+		deletedPdbResult *pdbReactor
+		wantPdbs         []string
+	}{
+		{
+			name:         "success: remove pdbs",
+			modelService: modelSvc,
+			existingPdbs: &policyv1.PodDisruptionBudgetList{
+				Items: []policyv1.PodDisruptionBudget{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "my-model-1-r4-predictor-pdb",
+							Labels:    defaultMatchLabel,
+							Namespace: modelSvc.Namespace,
+						},
+					},
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "my-model-1-r4-transformer-pdb",
+							Labels:    defaultMatchLabel,
+							Namespace: modelSvc.Namespace,
+						},
+					},
+				},
+			},
+			createPdbResult:  &pdbReactor{pdb, nil},
+			deletedPdbResult: &pdbReactor{err: nil},
+			wantPdbs: []string{
+				"my-model-1-r4-predictor-pdb",
+				"my-model-1-r4-transformer-pdb",
+			},
+		},
+		{
+			name:         "success: only remove pdb with same labels",
+			modelService: modelSvc,
+			existingPdbs: &policyv1.PodDisruptionBudgetList{
+				Items: []policyv1.PodDisruptionBudget{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "my-model-1-r4-predictor-pdb",
+							Labels:    defaultMatchLabel,
+							Namespace: modelSvc.Namespace,
+						},
+					},
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "my-model-2-r4-predictor-pdb",
+							Labels: map[string]string{
+								"gojek.com/app":                      "mymodel",
+								"gojek.com/component":                "model-version",
+								"gojek.com/environment":              "dev",
+								"gojek.com/orchestrator":             "merlin",
+								"gojek.com/stream":                   "mystream",
+								"gojek.com/team":                     "myteam",
+								"component":                          "predictor",
+								"serving.kserve.io/inferenceservice": "mymodel-2-r4",
+								"model-version-id":                   "2",
+							},
+							Namespace: modelSvc.Namespace,
+						},
+					},
+				},
+			},
+			createPdbResult:  &pdbReactor{pdb, nil},
+			deletedPdbResult: &pdbReactor{err: nil},
+			wantPdbs: []string{
+				"my-model-1-r4-predictor-pdb",
+			},
+		},
+		{
+			name:         "success: no pdb found or to delete",
+			modelService: modelSvc,
+			existingPdbs: &policyv1.PodDisruptionBudgetList{
+				Items: []policyv1.PodDisruptionBudget{},
+			},
+			createPdbResult:  &pdbReactor{pdb, nil},
+			deletedPdbResult: &pdbReactor{err: nil},
+			wantPdbs:         []string{},
+		},
+		{
+			name:         "fail deleting pdb should not return error",
+			modelService: modelSvc,
+			existingPdbs: &policyv1.PodDisruptionBudgetList{
+				Items: []policyv1.PodDisruptionBudget{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "my-model-1-r4-predictor-pdb",
+							Labels:    defaultMatchLabel,
+							Namespace: modelSvc.Namespace,
+						},
+					},
+				},
+			},
+			createPdbResult:  &pdbReactor{pdb, nil},
+			deletedPdbResult: &pdbReactor{err: ErrUnableToDeletePDB},
+			wantPdbs:         []string{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			knClient := knservingfake.NewSimpleClientset()
+
+			kfClient := fakekserve.NewSimpleClientset().ServingV1beta1().(*fakekservev1beta1.FakeServingV1beta1)
+			isvcWatchReactor := newIsvcWatchReactor(&kservev1beta1.InferenceService{
+				ObjectMeta: metav1.ObjectMeta{Name: isvcName, Namespace: project.Name},
+				Status:     statusReady,
+			})
+			kfClient.WatchReactionChain = []ktesting.WatchReactor{isvcWatchReactor}
+
+			kfClient.PrependReactor(getMethod, inferenceServiceResource, func(action ktesting.Action) (handled bool, ret runtime.Object, err error) {
+				return true, &kservev1beta1.InferenceService{Status: statusReady}, nil
+			})
+
+			v1Client := fake.NewSimpleClientset().CoreV1()
+
+			policyV1Client := fake.NewSimpleClientset(tt.existingPdbs).PolicyV1().(*fakepolicyv1.FakePolicyV1)
+			policyV1Client.Fake.PrependReactor(patchMethod, pdbResource, func(action ktesting.Action) (handled bool, ret runtime.Object, err error) {
+				return true, tt.createPdbResult.pdb, tt.createPdbResult.err
+			})
+
+			deletedPdbNames := []string{}
+			policyV1Client.Fake.PrependReactor(deleteMethod, pdbResource, func(action ktesting.Action) (handled bool, ret runtime.Object, err error) {
+				deletedPdbNames = append(deletedPdbNames, action.(ktesting.DeleteAction).GetName())
+				return true, nil, tt.deletedPdbResult.err
+			})
+
+			istioClient := fakeistio.NewSimpleClientset().NetworkingV1beta1().(*fakeistionetworking.FakeNetworkingV1beta1)
+			istioClient.PrependReactor(patchMethod, virtualServiceResource, func(action ktesting.Action) (handled bool, ret runtime.Object, err error) {
+				return true, vs, nil
+			})
+
+			deployConfig := config.DeploymentConfig{
+				DeploymentTimeout:                  2 * tickDurationSecond * time.Second,
+				NamespaceTimeout:                   2 * tickDurationSecond * time.Second,
+				DefaultModelResourceRequests:       &config.ResourceRequests{},
+				DefaultTransformerResourceRequests: &config.ResourceRequests{},
+				MaxAllowedReplica:                  4,
+				PodDisruptionBudget: config.PodDisruptionBudgetConfig{
+					Enabled:                true,
+					MinAvailablePercentage: &minAvailablePercentage,
+				},
+				StandardTransformer:                   config.StandardTransformerConfig{},
+				UserContainerCPUDefaultLimit:          userContainerCPUDefaultLimit,
+				UserContainerCPULimitRequestFactor:    userContainerCPULimitRequestFactor,
+				UserContainerMemoryLimitRequestFactor: userContainerMemoryLimitRequestFactor,
+			}
+
+			containerFetcher := NewContainerFetcher(v1Client, clusterMetadata)
+			templater := clusterresource.NewInferenceServiceTemplater(deployConfig)
+
+			ctl, _ := newController(knClient.ServingV1(), kfClient, v1Client, nil, policyV1Client, istioClient, deployConfig, containerFetcher, templater)
+			iSvc, err := ctl.Deploy(context.Background(), tt.modelService)
+
+			if tt.deletedPdbResult.err == nil {
+				assert.ElementsMatch(t, deletedPdbNames, tt.wantPdbs)
+			}
+			assert.NoError(t, err)
+			assert.NotNil(t, iSvc)
+		})
+	}
+}
+
 func TestGetCurrentDeploymentScale(t *testing.T) {
 	testNamespace := "test-namespace"
 	var testDesiredReplicas int32 = 5