Add PVC support

Add PVC support according to the design and discussions
captured in the issue, kserve#230

Signed-off-by: Chin Huang <[email protected]>

controllers/modelmesh/const.go

@@ -29,6 +29,7 @@ const (
 	SocketVolume    = "domain-socket"
 
 	ConfigStorageMount = "storage-config"
+	PVCMount           = "pvc-mount"
 
 	//The name of the puller container
 	PullerContainerName = "puller"
@@ -45,12 +46,18 @@ const (
 	//The env variable puller uses to configure the config dir (secrets)
 	PullerEnvStorageConfigDir = "STORAGE_CONFIG_DIR"
 
+	//The env variable puller uses to configure the pvc dir (secrets)
+	PullerEnvPVCDir = "PVC_DIR"
+
 	//The puller default port number
 	PullerPortNumber = 8086
 
 	//The puller model mount path
 	PullerModelPath = "/models"
 
+	//The puller model PVC path
+	PullerPVCPath = "/pvc_mounts"
+
 	//The puller model config path
 	PullerConfigPath = "/storage-config"
 

controllers/modelmesh/modelmesh.go

@@ -53,6 +53,7 @@ type Deployment struct {
 	RESTProxyImage     string
 	RESTProxyResources *corev1.ResourceRequirements
 	RESTProxyPort      uint16
+	PVCs               []string
 	// internal fields used when templating
 	ModelMeshLimitCPU          string
 	ModelMeshRequestsCPU       string

controllers/modelmesh/puller.go

@@ -67,6 +67,9 @@ func addPullerSidecar(rts *kserveapi.ServingRuntimeSpec, deployment *appsv1.Depl
 			}, {
 				Name:  PullerEnvStorageConfigDir,
 				Value: PullerConfigPath,
+			}, {
+				Name:  PullerEnvPVCDir,
+				Value: PullerPVCPath,
 			},
 		},
 		Image:   pullerImage,
@@ -89,6 +92,11 @@ func addPullerSidecar(rts *kserveapi.ServingRuntimeSpec, deployment *appsv1.Depl
 				MountPath: PullerConfigPath,
 				ReadOnly:  true,
 			},
+			{
+				Name:      PVCMount,
+				MountPath: PullerPVCPath,
+				ReadOnly:  true,
+			},
 		},
 	}
 

controllers/modelmesh/runtime.go

@@ -28,6 +28,7 @@ import (
 
 const (
 	ModelsDir     string  = "/models"
+	PVCRootDir    string  = "/pvc_mounts"
 	ModelDirScale float64 = 1.5
 )
 
@@ -81,6 +82,21 @@ func (m *Deployment) addVolumesToDeployment(deployment *appsv1.Deployment) error
 		volumes = append(volumes, storageVolume)
 	}
 
+	// need to add pvc volumes
+	for _, pvcName := range m.PVCs {
+		pvcVolume := corev1.Volume{
+			Name: pvcName,
+			VolumeSource: corev1.VolumeSource{
+				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+					ClaimName: pvcName,
+					ReadOnly:  true,
+				},
+			},
+		}
+
+		volumes = append(volumes, pvcVolume)
+	}
+
 	deployment.Spec.Template.Spec.Volumes = volumes
 
 	return nil
@@ -124,6 +140,16 @@ func (m *Deployment) addRuntimeToDeployment(deployment *appsv1.Deployment) error
 		},
 	}
 
+	for _, pvcName := range m.PVCs {
+		volumeMounts = append([]corev1.VolumeMount{
+			{
+				Name:      pvcName,
+				MountPath: PVCRootDir + "/" + pvcName,
+				ReadOnly:  true,
+			},
+		}, volumeMounts...)
+	}
+
 	// Now add the containers specified in serving runtime spec
 	for i := range rts.Containers {
 		// by modifying in-place we rely on the fact that the cacheing

controllers/servingruntime_controller.go

@@ -62,6 +62,10 @@ import (
 	"github.com/kserve/modelmesh-serving/controllers/modelmesh"
 )
 
+const (
+	StoragePVCType = "pvc"
+)
+
 // ServingRuntimeReconciler reconciles a ServingRuntime object
 type ServingRuntimeReconciler struct {
 	client.Client
@@ -75,6 +79,8 @@ type ServingRuntimeReconciler struct {
 	ClusterScope bool
 	// whether the controller is enabled to read and watch ClusterServingRuntimes
 	EnableCSRWatch bool
+	// whether the controller is enabled to read and watch secrets
+	EnableSecretWatch bool
 	// store some information about current runtimes for making scaling decisions
 	runtimeInfoMap      map[types.NamespacedName]*runtimeInfo
 	runtimeInfoMapMutex sync.Mutex
@@ -224,6 +230,25 @@ func (r *ServingRuntimeReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		return ctrl.Result{}, fmt.Errorf("Invalid runtime Spec: %w", err)
 	}
 
+	s := &corev1.Secret{}
+	if err = r.Client.Get(ctx, types.NamespacedName{
+		Name:      modelmesh.StorageSecretName,
+		Namespace: req.Namespace,
+	}, s); err != nil {
+		return RequeueResult, fmt.Errorf("Could not get the storage secret: %w", err)
+	}
+
+	var pvcs []string
+	var storageConfig map[string]string
+	for _, storageData := range s.Data {
+		if err = json.Unmarshal(storageData, &storageConfig); err != nil {
+			return ctrl.Result{}, fmt.Errorf("Could not parse storage configuration json: %w", err)
+		}
+		if storageConfig["type"] == StoragePVCType {
+			pvcs = append(pvcs, storageConfig["name"])
+		}
+	}
+
 	// construct the deployment
 	mmDeployment := modelmesh.Deployment{
 		ServiceName:                cfg.InferenceServiceName,
@@ -249,6 +274,7 @@ func (r *ServingRuntimeReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		PullerResources:            cfg.StorageHelperResources.ToKubernetesType(),
 		Port:                       cfg.InferenceServicePort,
 		GrpcMaxMessageSize:         cfg.GrpcMaxMessageSizeBytes,
+		PVCs:                       pvcs,
 		// Replicas is set below
 		TLSSecretName:       cfg.TLS.SecretName,
 		TLSClientAuth:       cfg.TLS.ClientAuth,
@@ -491,6 +517,13 @@ func (r *ServingRuntimeReconciler) SetupWithManager(mgr ctrl.Manager,
 			}))
 	}
 
+	if r.EnableSecretWatch {
+		builder = builder.Watches(&source.Kind{Type: &corev1.Secret{}},
+			handler.EnqueueRequestsFromMapFunc(func(o client.Object) []reconcile.Request {
+				return r.secretRequests(o.(*corev1.Secret), o.GetNamespace())
+			}))
+	}
+
 	if sourcePluginEvents != nil {
 		builder.Watches(&source.Channel{Source: sourcePluginEvents},
 			handler.EnqueueRequestsFromMapFunc(func(o client.Object) []reconcile.Request {
@@ -624,3 +657,16 @@ func (r *ServingRuntimeReconciler) clusterServingRuntimeRequests(csr *kserveapi.
 
 	return requests
 }
+
+func (r *ServingRuntimeReconciler) secretRequests(secret *corev1.Secret, namespace string) []reconcile.Request {
+	// check whether namespace is modelmesh enabled
+	mme, err := modelMeshEnabled2(context.TODO(), namespace, r.ControllerNamespace, r.Client, r.ClusterScope)
+	if err != nil || !mme {
+		return []reconcile.Request{}
+	}
+	if secret.Name == modelmesh.StorageSecretName {
+		return r.requestsForRuntimes(namespace, nil)
+	}
+
+	return []reconcile.Request{}
+}

main.go

@@ -78,6 +78,7 @@ const (
 	LeaderForLifeLockName             = "modelmesh-controller-leader-for-life-lock"
 	EnableInferenceServiceEnvVar      = "ENABLE_ISVC_WATCH"
 	EnableClusterServingRuntimeEnvVar = "ENABLE_CSR_WATCH"
+	EnableSecretEnvVar                = "ENABLE_SECRET_WATCH"
 	NamespaceScopeEnvVar              = "NAMESPACE_SCOPE"
 	TrueString                        = "true"
 )
@@ -357,6 +358,26 @@ func main() {
 	}
 	enableCSRWatch := checkCSRVar(EnableClusterServingRuntimeEnvVar, "ClusterServingRuntime", &v1alpha1.ClusterServingRuntime{})
 
+	checkSecretVar := func(envVar string, resourceName string, resourceObject client.Object) bool {
+
+		envVarVal, _ := os.LookupEnv(envVar)
+		if envVarVal != "false" {
+			err = cl.Get(context.Background(), client.ObjectKey{Name: "foo", Namespace: ControllerNamespace}, resourceObject)
+			if err == nil || errors.IsNotFound(err) {
+				setupLog.Info(fmt.Sprintf("Reconciliation of %s is enabled", resourceName))
+				return true
+			} else if envVarVal == TrueString {
+				// If env var is explicitly true, require that specified CRD is present
+				setupLog.Error(err, fmt.Sprintf("Unable to access %s resource", resourceName))
+				os.Exit(1)
+			} else {
+				setupLog.Error(err, fmt.Sprintf("%s CRD not accessible, will not reconcile", resourceName))
+			}
+		}
+		return false
+	}
+	enableSecretWatch := checkSecretVar(EnableSecretEnvVar, "Secret", &corev1.Secret{})
+
 	var predictorControllerEvents, runtimeControllerEvents chan event.GenericEvent
 	if len(sources) != 0 {
 		predictorControllerEvents = make(chan event.GenericEvent, 256)
@@ -420,6 +441,7 @@ func main() {
 		ControllerName:      controllerDeploymentName,
 		ClusterScope:        clusterScopeMode,
 		EnableCSRWatch:      enableCSRWatch,
+		EnableSecretWatch:   enableSecretWatch,
 		RegistryMap:         registryMap,
 	}).SetupWithManager(mgr, enableIsvcWatch, runtimeControllerEvents); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ServingRuntime")