feat: rebase kubernetes overwrites

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.

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

@@ -80,6 +80,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 {
@@ -108,6 +111,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

@@ -1337,6 +1337,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)
@@ -1499,33 +1562,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 == "" {
@@ -1653,6 +1693,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