feat: kubernetes overwrites for PVC and entrypoint

Overview:
Add the ability for a user to overwrite the entrypoint
and PVCs in the sdk.
DEP-157

Details:
It is the first iteration of the feature.
A more generic approach will be developed.

Author: atchernych

deploy/cloud/operator/api/v1alpha1/dynamocomponentdeployment_types.go

@@ -79,6 +79,9 @@ type DynamoComponentDeploymentSharedSpec struct {
 	LivenessProbe  *corev1.Probe `json:"livenessProbe,omitempty"`
 	ReadinessProbe *corev1.Probe `json:"readinessProbe,omitempty"`
 	Replicas       *int32        `json:"replicas,omitempty"`
+
+	// Kubernetes overwrites for the deployment
+	KubernetesOverwrites *KubernetesOverwrites `json:"kubernetesOverwrites,omitempty"`
 }
 
 type RunMode struct {
@@ -107,6 +110,11 @@ type IngressSpec struct {
 	IngressControllerClassName *string           `json:"ingressControllerClassName,omitempty"`
 }
 
+type KubernetesOverwrites struct {
+	Entrypoint  *string         `json:"entrypoint,omitempty"`
+	PVCSettings map[string]*PVC `json:"pvcSettings,omitempty"`
+}
+
 // DynamoComponentDeploymentStatus defines the observed state of DynamoComponentDeployment
 type DynamoComponentDeploymentStatus struct {
 	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster

deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go

@@ -1338,6 +1338,69 @@ func getDynamoComponentRepositoryNameAndDynamoComponentVersion(dynamoComponent *
 	return
 }
 
+func buildPVCVolumesAndMounts(
+	component *v1alpha1.DynamoComponentDeployment,
+) (volumes []corev1.Volume, mounts []corev1.VolumeMount) {
+	used := map[string]bool{}
+
+	// addPVC adds a volume and corresponding volume mount to the pod spec based on the provided PVC configuration.
+	addPVC := func(volumeName string, pvc *v1alpha1.PVC) {
+		if pvc == nil || pvc.Name == nil {
+			return
+		}
+		claimName := *pvc.Name
+		mountPath := "/mnt/default"
+		if pvc.MountPoint != nil && *pvc.MountPoint != "" {
+			mountPath = *pvc.MountPoint
+		}
+		volume := corev1.Volume{
+			Name: volumeName,
+			VolumeSource: corev1.VolumeSource{
+				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+					ClaimName: claimName,
+				},
+			},
+		}
+		mount := corev1.VolumeMount{
+			Name:      volumeName,
+			MountPath: mountPath,
+		}
+		if used[volumeName] {
+			for i, v := range volumes {
+				if v.Name == volumeName {
+					volumes[i] = volume
+					break
+				}
+			}
+			for i, m := range mounts {
+				if m.Name == volumeName {
+					mounts[i] = mount
+					break
+				}
+			}
+		} else {
+			volumes = append(volumes, volume)
+			mounts = append(mounts, mount)
+			used[volumeName] = true
+		}
+	}
+
+	// Handle default PVC
+	if component.Spec.PVC != nil {
+		volumeName := getPvcName(component, component.Spec.PVC.Name)
+		addPVC(volumeName, component.Spec.PVC)
+	}
+
+	// Handle overwrites
+	if ow := component.Spec.KubernetesOverwrites; ow != nil && ow.PVCSettings != nil {
+		for volumeName, pvc := range ow.PVCSettings {
+			addPVC(volumeName, pvc)
+		}
+	}
+
+	return volumes, mounts
+}
+
 //nolint:gocyclo,nakedret
 func (r *DynamoComponentDeploymentReconciler) generatePodTemplateSpec(ctx context.Context, opt generateResourceOption) (podTemplateSpec *corev1.PodTemplateSpec, err error) {
 	podLabels := r.getKubeLabels(opt.dynamoComponentDeployment, opt.dynamoComponent)
@@ -1500,33 +1563,10 @@ func (r *DynamoComponentDeploymentReconciler) generatePodTemplateSpec(ctx contex
 		sharedMemorySizeLimit.SetMilli(memoryLimit.MilliValue() / 2)
 	}
 
-	volumes = append(volumes, corev1.Volume{
-		Name: KubeValueNameSharedMemory,
-		VolumeSource: corev1.VolumeSource{
-			EmptyDir: &corev1.EmptyDirVolumeSource{
-				Medium:    corev1.StorageMediumMemory,
-				SizeLimit: &sharedMemorySizeLimit,
-			},
-		},
-	})
-	volumeMounts = append(volumeMounts, corev1.VolumeMount{
-		Name:      KubeValueNameSharedMemory,
-		MountPath: "/dev/shm",
-	})
-	if opt.dynamoComponentDeployment.Spec.PVC != nil {
-		volumes = append(volumes, corev1.Volume{
-			Name: getPvcName(opt.dynamoComponentDeployment, opt.dynamoComponentDeployment.Spec.PVC.Name),
-			VolumeSource: corev1.VolumeSource{
-				PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
-					ClaimName: getPvcName(opt.dynamoComponentDeployment, opt.dynamoComponentDeployment.Spec.PVC.Name),
-				},
-			},
-		})
-		volumeMounts = append(volumeMounts, corev1.VolumeMount{
-			Name:      getPvcName(opt.dynamoComponentDeployment, opt.dynamoComponentDeployment.Spec.PVC.Name),
-			MountPath: *opt.dynamoComponentDeployment.Spec.PVC.MountPoint,
-		})
-	}
+	// Handle default PVC settings. Apply overwrites if the name matches, add otherwise.
+	pvcVolumes, pvcMounts := buildPVCVolumesAndMounts(opt.dynamoComponentDeployment)
+	volumes = append(volumes, pvcVolumes...)
+	volumeMounts = append(volumeMounts, pvcMounts...)
 
 	imageName := opt.dynamoComponent.GetImage()
 	if imageName == "" {
@@ -1654,6 +1694,21 @@ func (r *DynamoComponentDeploymentReconciler) generatePodTemplateSpec(ctx contex
 		container.SecurityContext.RunAsUser = &[]int64{0}[0]
 	}
 
+	if opt.dynamoComponentDeployment.Spec.KubernetesOverwrites != nil {
+		overwrites := opt.dynamoComponentDeployment.Spec.KubernetesOverwrites
+
+		// Handle Entrypoint overwrite. Entrypoint needs to be renamed to Command.
+		if overwrites.Entrypoint != nil {
+			parts := strings.Fields(*overwrites.Entrypoint)
+			if len(parts) > 0 {
+				container.Command = []string{parts[0]}
+				if len(parts) > 1 {
+					container.Args = parts[1:]
+				}
+			}
+		}
+	}
+
 	containers = append(containers, container)
 
 	debuggerImage := "python:3.12-slim"

deploy/sdk/src/dynamo/sdk/cli/build.py

@@ -116,6 +116,7 @@ class ServiceInfo(BaseModel):
     module_path: str
     class_name: str
     config: ServiceConfig
+    kubernetes_overwrites: t.Optional[t.Dict[str, t.Any]] = None
 
     @classmethod
     def from_service(cls, service: ServiceInterface[T]) -> ServiceInfo:
@@ -145,12 +146,16 @@ def from_service(cls, service: ServiceInterface[T]) -> ServiceInfo:
             http_exposed=len(api_endpoints) > 0,
             api_endpoints=api_endpoints,
         )
+        kubernetes_overwrites = None
+        if hasattr(service, "kubernetes_overwrites") and service.kubernetes_overwrites:
+            kubernetes_overwrites = service.kubernetes_overwrites.model_dump()
 
         return cls(
             name=name,
             module_path=service.__module__,
             class_name=service_class.__name__,
             config=config,
+            kubernetes_overwrites=kubernetes_overwrites,
         )
 
 
@@ -225,6 +230,13 @@ def to_dict(self) -> t.Dict[str, t.Any]:
                 service_dict["config"]["api_endpoints"] = service["config"][
                     "api_endpoints"
                 ]
+
+            # Add kubernetes overwrites if available
+            if service.get("kubernetes_overwrites"):
+                service_dict["config"]["kubernetes_overwrites"] = service[
+                    "kubernetes_overwrites"
+                ]
+
             services_dict.append(service_dict)
         result["services"] = services_dict
         return result

deploy/sdk/src/dynamo/sdk/lib/service.py

@@ -26,6 +26,7 @@
 from _bentoml_sdk.images import Image
 from _bentoml_sdk.service.config import validate
 from fastapi import FastAPI
+from pydantic import BaseModel
 
 from dynamo.sdk.core.protocol.interface import DynamoTransport, LinkedServices
 from dynamo.sdk.lib.decorators import DynamoEndpoint
@@ -45,6 +46,24 @@ class ComponentType(str, Enum):
     # etc.
 
 
+class PVC(BaseModel):
+    """Class for Kubernetes PVC Overwrites."""
+
+    create: Optional[bool] = None
+    name: Optional[str] = None
+    storage_class: Optional[str] = None
+    size: Optional[str] = None  # Represent `resource.Quantity` as a string, e.g. "10Gi"
+    volume_access_mode: Optional[str] = None  # e.g., "ReadWriteOnce", "ReadOnlyMany"
+    mount_point: Optional[str] = None
+
+
+class KubernetesOverwrites(BaseModel):
+    """Class for Kubernetes Overwrites."""
+
+    entrypoint: Optional[str] = None
+    pvc_settings: Optional[Dict[str, PVC]] = None
+
+
 @dataclass
 class DynamoConfig:
     """Configuration for Dynamo components"""
@@ -76,6 +95,7 @@ def __init__(
         envs: Optional[list[dict[str, Any]]] = None,
         dynamo_config: Optional[DynamoConfig] = None,
         app: Optional[FastAPI] = None,
+        kubernetes_overwrites: Optional[KubernetesOverwrites] = None,
     ):
         service_name = inner.__name__
         service_args = self._get_service_args(service_name)
@@ -140,6 +160,7 @@ def __init__(
             self.config["api_endpoints"] = self._api_endpoints.copy()
 
         self._linked_services: List[DynamoService] = []  # Track linked services
+        self.kubernetes_overwrites = kubernetes_overwrites
 
     def _get_service_args(self, service_name: str) -> Optional[dict]:
         """Get ServiceArgs from environment config if specified"""
@@ -306,6 +327,7 @@ def service(
     envs: Optional[list[dict[str, Any]]] = None,
     dynamo: Optional[Union[Dict[str, Any], DynamoConfig]] = None,
     app: Optional[FastAPI] = None,
+    kubernetes_overwrites: Optional[Union[dict, KubernetesOverwrites]] = None,
     **kwargs: Any,
 ) -> Any:
     """Enhanced service decorator that supports Dynamo configuration
@@ -315,8 +337,11 @@ def service(
             - enabled: bool (default True)
             - name: str (default: class name)
             - namespace: str (default: "default")
+        kubernetes_overwrites: kubernetes overwrites i.e. entrypoint.
         **kwargs: Existing BentoML service configuration
     """
+    if isinstance(kubernetes_overwrites, dict):
+        kubernetes_overwrites = KubernetesOverwrites(**kubernetes_overwrites)
     config = kwargs
 
     # Parse dict into DynamoConfig object
@@ -337,6 +362,7 @@ def decorator(inner: type[T]) -> DynamoService[T]:
             envs=envs or [],
             dynamo_config=dynamo_config,
             app=app,
+            kubernetes_overwrites=kubernetes_overwrites,
         )
 
     return decorator(inner) if inner is not None else decorator