[RHOAIENG-20548] - Env variable removed from ISVC are not removed from predictor pod (#516) (#520)

* [RHOAIENG-20548] - Env variable removed from ISVC are not removed from predictor pod

chore:  Fixes the reconcile loop to effectivelly update the EnvVars and replicas count.
        Also address this issue: https://issues.redhat.com/browse/RHOAIENG-20634



* review comments

---------

Signed-off-by: Spolti <[email protected]>

Author: spolti

pkg/controller/v1beta1/inferenceservice/rawkube_controller_test.go

@@ -74,20 +74,19 @@ var _ = Describe("v1beta1 inference service controller", func() {
 		}
 	)
 
-	Context("When creating inference service with raw kube predictor", func() {
-		configs := map[string]string{
-			"explainers": `{
+	configs := map[string]string{
+		"explainers": `{
 				"alibi": {
 					"image": "kserve/alibi-explainer",
 					"defaultImageVersion": "latest"
 				}
 			}`,
-			"ingress": `{
+		"ingress": `{
 				"ingressGateway": "knative-serving/knative-ingress-gateway",
 				"localGateway": "knative-serving/knative-local-gateway",
 				"localGatewayService": "knative-local-gateway.istio-system.svc.cluster.local"
 			}`,
-			"storageInitializer": `{
+		"storageInitializer": `{
 				"image" : "kserve/storage-initializer:latest",
 				"memoryRequest": "100Mi",
 				"memoryLimit": "1Gi",
@@ -97,7 +96,8 @@ var _ = Describe("v1beta1 inference service controller", func() {
 				"caBundleVolumeMountPath": "/etc/ssl/custom-certs",
 				"enableDirectPvcVolumeMount": false
 			}`,
-		}
+	}
+	Context("When creating inference service with raw kube predictor", func() {
 
 		It("Should have ingress/service/deployment/hpa created", func() {
 			By("By creating a new InferenceService")
@@ -1383,6 +1383,199 @@ var _ = Describe("v1beta1 inference service controller", func() {
 				ShouldNot(Succeed())
 		})
 	})
+
+	Context("When updating ISVC envs", func() {
+		It("Should reconcile the deployment if isvc envs are updated", func() {
+			defaultEnvs := []v1.EnvVar{
+				{
+					Name:  "env1",
+					Value: "val1",
+				},
+				{
+					Name:  "env2",
+					Value: "val2",
+				},
+				{
+					Name:  "env3",
+					Value: "val3",
+				},
+			}
+
+			// Create configmap
+			var configMap = &v1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      constants.InferenceServiceConfigMapName,
+					Namespace: constants.KServeNamespace,
+				},
+				Data: configs,
+			}
+			Expect(k8sClient.Create(context.TODO(), configMap)).NotTo(HaveOccurred())
+			defer k8sClient.Delete(context.TODO(), configMap)
+			// Create ServingRuntime
+			servingRuntime := &v1alpha1.ServingRuntime{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "tf-serving-raw",
+					Namespace: "default",
+				},
+				Spec: v1alpha1.ServingRuntimeSpec{
+					SupportedModelFormats: []v1alpha1.SupportedModelFormat{
+						{
+							Name:       "tensorflow",
+							Version:    proto.String("1"),
+							AutoSelect: proto.Bool(true),
+						},
+					},
+					ServingRuntimePodSpec: v1alpha1.ServingRuntimePodSpec{
+						Containers: []v1.Container{
+							{
+								Name:    "kserve-container",
+								Image:   "tensorflow/serving:1.14.0",
+								Command: []string{"/usr/bin/tensorflow_model_server"},
+								Args: []string{
+									"--port=9000",
+									"--rest_api_port=8080",
+									"--model_base_path=/mnt/models",
+									"--rest_api_timeout_in_ms=60000",
+								},
+								Resources: defaultResource,
+							},
+						},
+					},
+					Disabled: proto.Bool(false),
+				},
+			}
+			k8sClient.Create(context.TODO(), servingRuntime)
+			defer k8sClient.Delete(context.TODO(), servingRuntime)
+			serviceName := "raw-test-env"
+			var expectedRequest = reconcile.Request{NamespacedName: types.NamespacedName{Name: serviceName, Namespace: "default"}}
+			var serviceKey = expectedRequest.NamespacedName
+			// create isvc
+			var storageUri = "s3://test/mnist/export"
+			isvcOriginal := &v1beta1.InferenceService{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      serviceKey.Name,
+					Namespace: serviceKey.Namespace,
+					Annotations: map[string]string{
+						"serving.kserve.io/deploymentMode":              "RawDeployment",
+						"serving.kserve.io/autoscalerClass":             "hpa",
+						"serving.kserve.io/metrics":                     "cpu",
+						"serving.kserve.io/targetUtilizationPercentage": "75",
+					},
+				},
+				Spec: v1beta1.InferenceServiceSpec{
+					Predictor: v1beta1.PredictorSpec{
+						ComponentExtensionSpec: v1beta1.ComponentExtensionSpec{
+							MinReplicas: v1beta1.GetIntReference(1),
+							MaxReplicas: 3,
+						},
+						Tensorflow: &v1beta1.TFServingSpec{
+							PredictorExtensionSpec: v1beta1.PredictorExtensionSpec{
+
+								StorageURI:     &storageUri,
+								RuntimeVersion: proto.String("1.14.0"),
+								Container: v1.Container{
+									Name:      constants.InferenceServiceContainerName,
+									Resources: defaultResource,
+									Env:       defaultEnvs,
+								},
+							},
+						},
+					},
+				},
+			}
+
+			isvcOriginal.DefaultInferenceService(nil, nil, &v1beta1.SecurityConfig{AutoMountServiceAccountToken: false}, nil)
+			Expect(k8sClient.Create(ctx, isvcOriginal)).Should(Succeed())
+
+			inferenceService := &v1beta1.InferenceService{}
+			Eventually(func() bool {
+				err := k8sClient.Get(ctx, serviceKey, inferenceService)
+				if err != nil {
+					return false
+				}
+				return true
+			}, timeout, interval).Should(BeTrue())
+
+			deployed1 := &appsv1.Deployment{}
+			predictorDeploymentKey := types.NamespacedName{Name: constants.PredictorServiceName(serviceKey.Name),
+				Namespace: serviceKey.Namespace}
+			Eventually(func() error {
+				return k8sClient.Get(context.TODO(), predictorDeploymentKey, deployed1)
+			}, timeout, interval).Should(Succeed())
+			Expect(deployed1.Spec.Template.Spec.Containers[0].Env).To(ContainElements(defaultEnvs))
+
+			// Now, update the isvc with new env
+			newEnvs := []v1.EnvVar{
+				{
+					Name:  "newEnv1",
+					Value: "newValue1",
+				},
+				{
+					Name:  "newEnv2",
+					Value: "delete",
+				},
+			}
+
+			// Update the isvc to add new envs
+			fmt.Fprintln(GinkgoWriter, "### Adding new envs")
+			isvcUpdated1 := &v1beta1.InferenceService{}
+			Eventually(func() bool {
+				// get the latest deployed version
+				err := k8sClient.Get(ctx, serviceKey, inferenceService)
+				if err != nil {
+					return false
+				}
+
+				isvcUpdated1 = inferenceService.DeepCopy()
+				isvcUpdated1.Spec.Predictor.Model.Env = append(isvcUpdated1.Spec.Predictor.Model.Env, newEnvs...)
+				err = k8sClient.Update(ctx, isvcUpdated1)
+				if err != nil {
+					return false
+				}
+				return true
+			}, timeout, interval).Should(BeTrue())
+
+			// The deployment should be reconciled
+			deployed2 := &appsv1.Deployment{}
+			appendedEnvs := append(defaultEnvs, newEnvs...)
+			Eventually(func() []v1.EnvVar {
+				_ = k8sClient.Get(context.TODO(), predictorDeploymentKey, deployed2)
+				return deployed2.Spec.Template.Spec.Containers[0].Env
+			}, timeout, interval).Should(ContainElements(appendedEnvs))
+
+			// Now remove the default envs and update the isvc
+			fmt.Fprintln(GinkgoWriter, "### Removing default envs")
+			isvcUpdated2 := &v1beta1.InferenceService{}
+			Eventually(func() bool {
+				// get the latest deployed version
+				err := k8sClient.Get(ctx, serviceKey, isvcUpdated1)
+				if err != nil {
+					return false
+				}
+
+				isvcUpdated2 = isvcUpdated1.DeepCopy()
+				isvcUpdated2.Spec.Predictor.Model.Env = newEnvs
+				// Make sure the default envs were removed before updating the isvc
+				Expect(isvcUpdated2.Spec.Predictor.Model.Env).ToNot(ContainElements(defaultEnvs))
+
+				err = k8sClient.Update(ctx, isvcUpdated2)
+				if err != nil {
+					return false
+				}
+				return true
+			}, timeout, interval).Should(BeTrue())
+
+			deployed3 := &appsv1.Deployment{}
+			Eventually(func() []v1.EnvVar {
+				_ = k8sClient.Get(context.TODO(), predictorDeploymentKey, deployed3)
+				return deployed3.Spec.Template.Spec.Containers[0].Env
+			}, timeout, interval).Should(Not(ContainElements(defaultEnvs)))
+
+			Expect(deployed3.Spec.Template.Spec.Containers[0].Env).ToNot(ContainElement(HaveField("Value", "env_marked_for_deletion")))
+			Expect(deployed3.Spec.Template.Spec.Containers[0].Env).To(ContainElements(newEnvs))
+		})
+	})
+
 	Context("When creating inference service with raw kube predictor and empty ingressClassName", func() {
 		configs := map[string]string{
 			"explainers": `{

pkg/controller/v1beta1/inferenceservice/reconcilers/deployment/deployment_reconciler.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/client-go/kubernetes"
 
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -40,7 +41,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
-	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"knative.dev/pkg/kmp"
 	kclient "sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
@@ -412,13 +412,15 @@ func (r *DeploymentReconciler) checkDeploymentExist(client kclient.Client, deplo
 		log.Error(err, "Failed to perform dry-run update of deployment", "Deployment", deployment.Name)
 		return constants.CheckResultUnknown, nil, err
 	}
+
 	processedExistingDep := v1beta1utils.RemoveCookieSecretArg(*existingDeployment)
 	processedNewDep := v1beta1utils.RemoveCookieSecretArg(*deployment)
-	if diff, err := kmp.SafeDiff(processedNewDep.Spec, processedExistingDep.Spec, ignoreFields); err != nil {
+	if diff, err := kmp.SafeDiff(processedExistingDep.Spec, processedNewDep.Spec, ignoreFields); err != nil {
+		log.Error(err, "Failed to diff deployments", "Deployment", deployment.Name)
 		return constants.CheckResultUnknown, nil, err
-	} else if diff != "" {
+	} else if len(diff) > 0 {
 		log.Info("Deployment Updated", "Diff", diff)
-		return constants.CheckResultUpdate, existingDeployment, nil
+		return constants.CheckResultUpdate, processedNewDep, nil
 	}
 	return constants.CheckResultExisted, existingDeployment, nil
 }
@@ -554,6 +556,7 @@ func addGPUResourceToDeployment(deployment *appsv1.Deployment, targetContainerNa
 func (r *DeploymentReconciler) Reconcile() ([]*appsv1.Deployment, error) {
 	for _, deployment := range r.DeploymentList {
 		// Reconcile Deployment
+		originalDeployment := &appsv1.Deployment{}
 		checkResult, _, err := r.checkDeploymentExist(r.client, deployment)
 		if err != nil {
 			return nil, err
@@ -565,26 +568,58 @@ func (r *DeploymentReconciler) Reconcile() ([]*appsv1.Deployment, error) {
 		case constants.CheckResultCreate:
 			opErr = r.client.Create(context.TODO(), deployment)
 		case constants.CheckResultUpdate:
-			curJson, err := json.Marshal(deployment)
+			// get the current deployment
+			_ = r.client.Get(context.TODO(), types.NamespacedName{Name: deployment.Name, Namespace: deployment.Namespace}, originalDeployment)
+			// we need to remove the Replicas field from the deployment spec
+			originalDeployment.Spec.Replicas = nil
+			curJson, err := json.Marshal(originalDeployment)
 			if err != nil {
 				return nil, err
 			}
 
+			// Check if there are any envs to remove
+			// If there, its value will be set to "delete" so we can update the patchBytes with
+			// "patch": "delete"
+			// The strategic merge patch does not remove items from list just by removing it from the patch,
+			// to delete lists items using strategic merge patch, the $patch delete pattern is used.
+			// Example:
+			// - env:
+			//   - "name": "ENV1",
+			//     "$patch": "delete"
+			for i, deploymentC := range deployment.Spec.Template.Spec.Containers {
+				envs := []corev1.EnvVar{}
+				for _, OriginalC := range originalDeployment.Spec.Template.Spec.Containers {
+					if deploymentC.Name == OriginalC.Name {
+						envsToRemove, envsToKeep := utils.CheckEnvsToRemove(deploymentC.Env, OriginalC.Env)
+						if len(envsToRemove) > 0 {
+							envs = append(envs, envsToKeep...)
+							envs = append(envs, envsToRemove...)
+						} else {
+							envs = deploymentC.Env
+						}
+					}
+				}
+				deployment.Spec.Template.Spec.Containers[i].Env = envs
+			}
+
 			// To avoid the conflict between HPA and Deployment,
 			// we need to remove the Replicas field from the deployment spec
-			modDeployment := deployment.DeepCopy()
-			modDeployment.Spec.Replicas = nil
+			deployment.Spec.Replicas = nil
 
-			modJson, err := json.Marshal(modDeployment)
+			modJson, err := json.Marshal(deployment)
 			if err != nil {
 				return nil, err
 			}
+
 			// Generate the strategic merge patch between the current and modified JSON
 			patchByte, err := strategicpatch.StrategicMergePatch(curJson, modJson, appsv1.Deployment{})
 			if err != nil {
 				return nil, err
 			}
 
+			// override the envs that needs to be removed with  "$patch": "delete"
+			patchByte = []byte(strings.ReplaceAll(string(patchByte), "\"value\":\""+utils.PLACEHOLDER_FOR_DELETION+"\"", "\"$patch\":\"delete\""))
+
 			// Patch the deployment object with the strategic merge patch
 			opErr = r.client.Patch(context.TODO(), deployment, kclient.RawPatch(types.StrategicMergePatchType, patchByte))
 		}

pkg/utils/utils.go

@@ -310,3 +310,29 @@ func IsValidCustomGPUArray(s string) bool {
 
 	return true
 }
+
+const PLACEHOLDER_FOR_DELETION = "env_marked_for_deletion"
+
+// CheckEnvsToRemove checks the current envs against the desired ones and returns the envs that needs to be
+// removed from the target env list and envs that need to be kept.
+// Returns envsToRemove, envsToKeep
+func CheckEnvsToRemove(desired, current []v1.EnvVar) ([]v1.EnvVar, []v1.EnvVar) {
+	var envsToRemove []v1.EnvVar
+	var envsToKeep []v1.EnvVar
+	for _, currentEnv := range current {
+		found := false
+		for _, desiredEnv := range desired {
+			if currentEnv.Name == desiredEnv.Name {
+				envsToKeep = append(envsToKeep, currentEnv)
+				found = true
+				break
+			}
+		}
+		if !found {
+			// replace the value of the found env to a placeholder to mark it for deletion
+			currentEnv.Value = PLACEHOLDER_FOR_DELETION
+			envsToRemove = append(envsToRemove, currentEnv)
+		}
+	}
+	return envsToRemove, envsToKeep
+}