diff --git a/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java b/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java
index 60966798cbb..ce905b293ff 100644
--- a/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java
+++ b/api/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java
@@ -172,4 +172,5 @@ public interface KubernetesCluster extends ControlledEntity, com.cloud.utils.fsm
     Long getEtcdNodeCount();
     Long getCniConfigId();
     String getCniConfigDetails();
+    boolean isCsiEnabled();
 }
diff --git a/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java b/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java
index 307104af92c..cab545f9228 100644
--- a/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java
+++ b/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java
@@ -135,6 +135,7 @@ public class ApiConstants {
     public static final String CNI_CONFIG_ID = "cniconfigurationid";
     public static final String CNI_CONFIG_DETAILS = "cniconfigdetails";
     public static final String CNI_CONFIG_NAME = "cniconfigname";
+    public static final String CSI_ENABLED = "csienabled";
     public static final String COMPONENT = "component";
     public static final String CPU = "CPU";
     public static final String CPU_CORE_PER_SOCKET = "cpucorepersocket";
@@ -215,6 +216,7 @@ public class ApiConstants {
     public static final String DURATION = "duration";
     public static final String ELIGIBLE = "eligible";
     public static final String EMAIL = "email";
+    public static final String ENABLE_CSI = "enablecsi";
     public static final String END_ASN = "endasn";
     public static final String END_DATE = "enddate";
     public static final String END_IP = "endip";
diff --git a/engine/schema/src/main/resources/META-INF/db/schema-42100to42200.sql b/engine/schema/src/main/resources/META-INF/db/schema-42100to42200.sql
index a14a9fef8ad..5f780994a0b 100644
--- a/engine/schema/src/main/resources/META-INF/db/schema-42100to42200.sql
+++ b/engine/schema/src/main/resources/META-INF/db/schema-42100to42200.sql
@@ -48,6 +48,9 @@ CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.backup_repository', 'cross_zone_inst
 UPDATE `cloud`.`storage_pool_details` SET display = 0 WHERE name LIKE '%password%';
 UPDATE `cloud`.`storage_pool_details` SET display = 0 WHERE name LIKE '%token%';
 
+-- Add csi_enabled column to kubernetes_cluster table to indicate if the cluster is using csi or not
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.kubernetes_cluster', 'csi_enabled', 'TINYINT(1) unsigned NOT NULL DEFAULT 0 COMMENT "true if kubernetes cluster is using csi, false otherwise" ');
+
 -- VMware to KVM migration improvements
 CREATE TABLE IF NOT EXISTS `cloud`.`import_vm_task`(
     `id` bigint unsigned NOT NULL auto_increment COMMENT 'id',
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
index a12648c8e84..e814e35c2c3 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
@@ -96,7 +96,16 @@ import org.apache.cloudstack.api.command.user.network.CreateNetworkACLCmd;
 import org.apache.cloudstack.api.command.user.network.DeleteNetworkACLCmd;
 import org.apache.cloudstack.api.command.user.network.ListNetworkACLsCmd;
 import org.apache.cloudstack.api.command.user.network.ListNetworksCmd;
+import org.apache.cloudstack.api.command.user.snapshot.CreateSnapshotCmd;
+import org.apache.cloudstack.api.command.user.snapshot.DeleteSnapshotCmd;
+import org.apache.cloudstack.api.command.user.snapshot.ListSnapshotsCmd;
 import org.apache.cloudstack.api.command.user.vm.ListVMsCmd;
+import org.apache.cloudstack.api.command.user.volume.AttachVolumeCmd;
+import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
+import org.apache.cloudstack.api.command.user.volume.DeleteVolumeCmd;
+import org.apache.cloudstack.api.command.user.volume.DetachVolumeCmd;
+import org.apache.cloudstack.api.command.user.volume.ListVolumesCmd;
+import org.apache.cloudstack.api.command.user.volume.ResizeVolumeCmd;
 import org.apache.cloudstack.api.response.KubernetesClusterConfigResponse;
 import org.apache.cloudstack.api.response.KubernetesClusterResponse;
 import org.apache.cloudstack.api.response.KubernetesUserVmResponse;
@@ -252,7 +261,16 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     private static final List<Class<?>> PROJECT_KUBERNETES_ACCOUNT_ROLE_ALLOWED_APIS = Arrays.asList(
             QueryAsyncJobResultCmd.class,
             ListVMsCmd.class,
+            ListVolumesCmd.class,
+            CreateVolumeCmd.class,
+            DeleteVolumeCmd.class,
+            AttachVolumeCmd.class,
+            DetachVolumeCmd.class,
+            ResizeVolumeCmd.class,
             ListNetworksCmd.class,
+            CreateSnapshotCmd.class,
+            ListSnapshotsCmd.class,
+            DeleteSnapshotCmd.class,
             ListPublicIpAddressesCmd.class,
             AssociateIPAddrCmd.class,
             DisassociateIPAddrCmd.class,
@@ -880,6 +898,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         response.setMinSize(kubernetesCluster.getMinSize());
         response.setMaxSize(kubernetesCluster.getMaxSize());
         response.setClusterType(kubernetesCluster.getClusterType());
+        response.setCsiEnabled(kubernetesCluster.isCsiEnabled());
         response.setCreated(kubernetesCluster.getCreated());
 
         return response;
@@ -1605,6 +1624,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
                 if (zone.isSecurityGroupEnabled()) {
                     newCluster.setSecurityGroupId(finalSecurityGroup.getId());
                 }
+                newCluster.setCsiEnabled(cmd.getEnableCsi());
                 kubernetesClusterDao.persist(newCluster);
                 addKubernetesClusterDetails(newCluster, defaultNetwork, cmd);
                 return newCluster;
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVO.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVO.java
index 79fc15f6898..7dfd0043e32 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVO.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVO.java
@@ -145,6 +145,9 @@ public class KubernetesClusterVO implements KubernetesCluster {
     @Column(name = "cni_config_details", updatable = true, length = 4096)
     private String cniConfigDetails;
 
+    @Column(name = "csi_enabled")
+    private boolean csiEnabled;
+
     @Override
     public long getId() {
         return id;
@@ -389,6 +392,14 @@ public class KubernetesClusterVO implements KubernetesCluster {
         this.clusterType = clusterType;
     }
 
+    public boolean isCsiEnabled() {
+        return csiEnabled;
+    }
+
+    public void setCsiEnabled(boolean csiEnabled) {
+        this.csiEnabled = csiEnabled;
+    }
+
     public KubernetesClusterVO() {
         this.uuid = UUID.randomUUID().toString();
     }
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java
index 9be5f4c0776..baf717612f8 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java
@@ -232,13 +232,17 @@ public class KubernetesClusterActionWorker {
 
     protected final String deploySecretsScriptFilename = "deploy-cloudstack-secret";
     protected final String deployProviderScriptFilename = "deploy-provider";
+    protected final String deployCsiDriverScriptFilename = "deploy-csi-driver";
+    protected final String deletePvScriptFilename = "delete-pv-reclaimpolicy-delete";
     protected final String autoscaleScriptFilename = "autoscale-kube-cluster";
     protected final String validateNodeScript = "validate-cks-node";
     protected final String removeNodeFromClusterScript = "remove-node-from-cluster";
     protected final String scriptPath = "/opt/bin/";
     protected File deploySecretsScriptFile;
     protected File deployProviderScriptFile;
+    protected File deployCsiDriverScriptFile;
     protected File autoscaleScriptFile;
+    protected File deletePvScriptFile;
     protected KubernetesClusterManagerImpl manager;
     protected String[] keys;
 
@@ -715,12 +719,16 @@ public class KubernetesClusterActionWorker {
     protected void retrieveScriptFiles() {
         deploySecretsScriptFile = retrieveScriptFile(deploySecretsScriptFilename);
         deployProviderScriptFile = retrieveScriptFile(deployProviderScriptFilename);
+        deployCsiDriverScriptFile = retrieveScriptFile(deployCsiDriverScriptFilename);
+        deletePvScriptFile = retrieveScriptFile(deletePvScriptFilename);
         autoscaleScriptFile = retrieveScriptFile(autoscaleScriptFilename);
     }
 
     protected void copyScripts(String nodeAddress, final int sshPort) {
         copyScriptFile(nodeAddress, sshPort, deploySecretsScriptFile, deploySecretsScriptFilename);
         copyScriptFile(nodeAddress, sshPort, deployProviderScriptFile, deployProviderScriptFilename);
+        copyScriptFile(nodeAddress, sshPort, deployCsiDriverScriptFile, deployCsiDriverScriptFilename);
+        copyScriptFile(nodeAddress, sshPort, deletePvScriptFile, deletePvScriptFilename);
         copyScriptFile(nodeAddress, sshPort, autoscaleScriptFile, autoscaleScriptFilename);
     }
 
@@ -821,6 +829,43 @@ public class KubernetesClusterActionWorker {
         }
     }
 
+    protected boolean deployCsiDriver() {
+        File pkFile = getManagementServerSshPublicKeyFile();
+        Pair<String, Integer> publicIpSshPort = getKubernetesClusterServerIpSshPort(null);
+        publicIpAddress = publicIpSshPort.first();
+        sshPort = publicIpSshPort.second();
+
+        try {
+            String command = String.format("sudo %s/%s", scriptPath, deployCsiDriverScriptFilename);
+            Pair<Boolean, String> result = SshHelper.sshExecute(publicIpAddress, sshPort, getControlNodeLoginUser(),
+                    pkFile, null, command, 10000, 10000, 60000);
+
+            // Maybe the file isn't present. Try and copy it
+            if (!result.first()) {
+                logMessage(Level.INFO, "CSI files missing. Adding them now", null);
+                retrieveScriptFiles();
+                copyScripts(publicIpAddress, sshPort);
+
+                if (!createCloudStackSecret(keys)) {
+                    logTransitStateAndThrow(Level.ERROR, String.format("Failed to setup keys for Kubernetes cluster %s",
+                            kubernetesCluster.getName()), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed);
+                }
+
+                // If at first you don't succeed ...
+                result = SshHelper.sshExecute(publicIpAddress, sshPort, getControlNodeLoginUser(),
+                        pkFile, null, command, 10000, 10000, 60000);
+                if (!result.first()) {
+                    throw new CloudRuntimeException(result.second());
+                }
+            }
+            return true;
+        } catch (Exception e) {
+            String msg = String.format("Failed to deploy kubernetes provider: %s : %s", kubernetesCluster.getName(), e.getMessage());
+            logAndThrow(Level.ERROR, msg);
+            return false;
+        }
+    }
+
     public void setKeys(String[] keys) {
         this.keys = keys;
     }
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java
index cf77a83420a..62bd8b4576a 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterDestroyWorker.java
@@ -106,6 +106,9 @@ public class KubernetesClusterDestroyWorker extends KubernetesClusterResourceMod
                         ApiCommandResourceType.VirtualMachine);
                 vmContext.setEventResourceId(vmID);
                 try {
+                    if (clusterVM.isControlNode() && kubernetesCluster.isCsiEnabled()) {
+                        deletePVsWithReclaimPolicyDelete();
+                    }
                     UserVm vm = userVmService.destroyVm(vmID, true);
                     if (!userVmManager.expunge(userVM)) {
                         logger.warn("Unable to expunge VM {}, destroying Kubernetes cluster will probably fail", vm);
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java
index d392612547b..1f976bf3085 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java
@@ -944,4 +944,47 @@ public class KubernetesClusterResourceModifierActionWorker extends KubernetesClu
     protected List<DedicatedResourceVO> listDedicatedHostsInDomain(Long domainId) {
         return dedicatedResourceDao.listByDomainId(domainId);
     }
+
+    public boolean deletePVsWithReclaimPolicyDelete() {
+        File pkFile = getManagementServerSshPublicKeyFile();
+        Pair<String, Integer> publicIpSshPort = getKubernetesClusterServerIpSshPort(null);
+        publicIpAddress = publicIpSshPort.first();
+        sshPort = publicIpSshPort.second();
+        try {
+            String command = String.format("sudo %s/%s", scriptPath, deletePvScriptFilename);
+            logMessage(Level.INFO, "Starting PV deletion script for cluster: " + kubernetesCluster.getName(), null);
+            Pair<Boolean, String> result = SshHelper.sshExecute(publicIpAddress, sshPort, getControlNodeLoginUser(),
+                    pkFile, null, command, 10000, 10000, 600000); // 10 minute timeout
+            if (Boolean.FALSE.equals(result.first())) {
+                logMessage(Level.INFO, "PV delete script missing. Adding it now", null);
+                retrieveScriptFiles();
+                if (deletePvScriptFile != null) {
+                    copyScriptFile(publicIpAddress, sshPort, deletePvScriptFile, deletePvScriptFilename);
+                    logMessage(Level.INFO, "Executing PV deletion script (this may take several minutes)...", null);
+                    result = SshHelper.sshExecute(publicIpAddress, sshPort, getControlNodeLoginUser(),
+                            pkFile, null, command, 10000, 10000, 600000); // 10 minute timeout
+                    if (Boolean.FALSE.equals(result.first())) {
+                        logMessage(Level.ERROR, "PV deletion script failed: " + result.second(), null);
+                        throw new CloudRuntimeException(result.second());
+                    }
+                    logMessage(Level.INFO, "PV deletion script completed successfully", null);
+                } else {
+                    logMessage(Level.WARN, "PV delete script file not found in resources, skipping PV deletion", null);
+                    return false;
+                }
+            } else {
+                logMessage(Level.INFO, "PV deletion script completed successfully", null);
+            }
+
+            if (result.second() != null && !result.second().trim().isEmpty()) {
+                logMessage(Level.INFO, "PV deletion script output: " + result.second(), null);
+            }
+
+            return true;
+        } catch (Exception e) {
+            String msg = String.format("Failed to delete PVs with reclaimPolicy=Delete: %s : %s", kubernetesCluster.getName(), e.getMessage());
+            logMessage(Level.WARN, msg, e);
+            return false;
+        }
+    }
 }
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java
index 8cd539b78e4..62cdd6c5a4e 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java
@@ -143,7 +143,7 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
 
     private Pair<String, String> getKubernetesControlNodeConfig(final String controlNodeIp, final String serverIp,
                                                                 final List<Network.IpAddresses> etcdIps, final String hostName, final boolean haSupported,
-                                                                final boolean ejectIso, final boolean externalCni) throws IOException {
+                                                                final boolean ejectIso, final boolean externalCni, final boolean setupCsi) throws IOException {
         String k8sControlNodeConfig = readK8sConfigFile("/conf/k8s-control-node.yml");
         final String apiServerCert = "{{ k8s_control_node.apiserver.crt }}";
         final String apiServerKey = "{{ k8s_control_node.apiserver.key }}";
@@ -161,6 +161,7 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
         final String certSans = "{{ k8s_control.server_ips }}";
         final String k8sCertificate = "{{ k8s_control.certificate_key }}";
         final String externalCniPlugin = "{{ k8s.external.cni.plugin }}";
+        final String setupCsiDriver = "{{ k8s.setup.csi.driver }}";
 
         final List<String> addresses = new ArrayList<>();
         addresses.add(controlNodeIp);
@@ -212,6 +213,7 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
         k8sControlNodeConfig = k8sControlNodeConfig.replace(certSans, String.format("- %s", serverIp));
         k8sControlNodeConfig = k8sControlNodeConfig.replace(k8sCertificate, KubernetesClusterUtil.generateClusterHACertificateKey(kubernetesCluster));
         k8sControlNodeConfig = k8sControlNodeConfig.replace(externalCniPlugin, String.valueOf(externalCni));
+        k8sControlNodeConfig = k8sControlNodeConfig.replace(setupCsiDriver, String.valueOf(setupCsi));
 
         k8sControlNodeConfig = updateKubeConfigWithRegistryDetails(k8sControlNodeConfig);
 
@@ -246,7 +248,7 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
         Long userDataId = kubernetesCluster.getCniConfigId();
         Pair<String, String> k8sControlNodeConfigAndControlIp = new Pair<>(null, null);
         try {
-            k8sControlNodeConfigAndControlIp = getKubernetesControlNodeConfig(controlNodeIp, serverIp, etcdIps, hostName, haSupported, Hypervisor.HypervisorType.VMware.equals(clusterTemplate.getHypervisorType()), Objects.nonNull(userDataId));
+            k8sControlNodeConfigAndControlIp = getKubernetesControlNodeConfig(controlNodeIp, serverIp, etcdIps, hostName, haSupported, Hypervisor.HypervisorType.VMware.equals(clusterTemplate.getHypervisorType()), Objects.nonNull(userDataId), kubernetesCluster.isCsiEnabled());
         } catch (IOException e) {
             logAndThrow(Level.ERROR, "Failed to read Kubernetes control node configuration file", e);
         }
@@ -858,6 +860,9 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
         }
         taintControlNodes();
         deployProvider();
+        if (kubernetesCluster.isCsiEnabled()) {
+            deployCsiDriver();
+        }
         updateLoginUserDetails(clusterVMs.stream().map(InternalIdentity::getId).collect(Collectors.toList()));
         stateTransitTo(kubernetesCluster.getId(), KubernetesCluster.Event.OperationSucceeded);
         return true;
diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/CreateKubernetesClusterCmd.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/CreateKubernetesClusterCmd.java
index 4e92f9546f8..ad4f61f3e9b 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/CreateKubernetesClusterCmd.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/CreateKubernetesClusterCmd.java
@@ -207,6 +207,9 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
             since = "4.21.0")
     private Map cniConfigDetails;
 
+    @Parameter(name = ApiConstants.ENABLE_CSI, type = CommandType.BOOLEAN, description = "if true, setups up CloudStack CSI driver", since = "4.22.0")
+    private Boolean enableCsi;
+
     @Parameter(name=ApiConstants.AS_NUMBER, type=CommandType.LONG, description="the AS Number of the network")
     private Long asNumber;
 
@@ -371,6 +374,10 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
         return cniConfigId;
     }
 
+    public boolean getEnableCsi() {
+        return Objects.nonNull(enableCsi) ? enableCsi : Boolean.FALSE;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesClusterResponse.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesClusterResponse.java
index b38fdcbc49e..81d068df9b6 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesClusterResponse.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesClusterResponse.java
@@ -204,6 +204,10 @@ public class KubernetesClusterResponse extends BaseResponseWithAnnotations imple
     @Param(description = "Maximum size of the cluster")
     private Long maxSize;
 
+    @SerializedName(ApiConstants.CSI_ENABLED)
+    @Param(description = "Indicates if the CloudStack CSI driver has been setup in the cluster")
+    private Boolean isCsiEnabled;
+
     @SerializedName(ApiConstants.CLUSTER_TYPE)
     @Param(description = "the type of the cluster")
     private KubernetesCluster.ClusterType clusterType;
@@ -515,4 +519,8 @@ public class KubernetesClusterResponse extends BaseResponseWithAnnotations imple
     public void setCniConfigName(String cniConfigName) {
         this.cniConfigName = cniConfigName;
     }
+
+    public void setCsiEnabled(Boolean csiEnabled) {
+        isCsiEnabled = csiEnabled;
+    }
 }
diff --git a/plugins/integrations/kubernetes-service/src/main/resources/conf/k8s-control-node.yml b/plugins/integrations/kubernetes-service/src/main/resources/conf/k8s-control-node.yml
index dc066e10d06..70291dd1c35 100644
--- a/plugins/integrations/kubernetes-service/src/main/resources/conf/k8s-control-node.yml
+++ b/plugins/integrations/kubernetes-service/src/main/resources/conf/k8s-control-node.yml
@@ -179,6 +179,11 @@ write_files:
           mkdir -p /opt/provider
           cp "${BINARIES_DIR}/provider.yaml" /opt/provider/provider.yaml
         fi
+        if [ -e "${BINARIES_DIR}/snapshot-crds.yaml" ]; then
+          mkdir -p /opt/csi
+          cp "${BINARIES_DIR}/snapshot-crds.yaml" /opt/csi/snapshot-crds.yaml
+          cp "${BINARIES_DIR}/manifest.yaml" /opt/csi/manifest.yaml
+        fi
 
         PAUSE_IMAGE=`ctr -n k8s.io images ls -q | grep "pause" | sort | tail -n 1`
         echo $PAUSE_IMAGE
diff --git a/plugins/integrations/kubernetes-service/src/main/resources/script/delete-pv-reclaimpolicy-delete b/plugins/integrations/kubernetes-service/src/main/resources/script/delete-pv-reclaimpolicy-delete
new file mode 100755
index 00000000000..aac0cbf07e4
--- /dev/null
+++ b/plugins/integrations/kubernetes-service/src/main/resources/script/delete-pv-reclaimpolicy-delete
@@ -0,0 +1,156 @@
+#!/bin/bash -e
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+set -e
+
+timestamp() {
+    date '+%Y-%m-%d %H:%M:%S'
+}
+
+echo "$(timestamp) - Starting PV deletion script with reclaimPolicy=Delete"
+
+delete_workloads_using_pvc() {
+    local namespace=$1
+    local pvc_name=$2
+
+    echo "$(timestamp) - Finding workloads using PVC $pvc_name in namespace $namespace..."
+
+    local deleted_count=0
+
+    # Find & delete any deployment using the PVC
+    /opt/bin/kubectl get deployments -n "$namespace" -o json 2>/dev/null | grep -l "$pvc_name" | \
+    while IFS= read -r deployment; do
+        if [ -n "$deployment" ]; then
+            deployment_name=$(echo "$deployment" | cut -d'/' -f2)
+            echo "$(timestamp) -   Deleting Deployment: $deployment_name"
+            /opt/bin/kubectl delete deployment "$deployment_name" -n "$namespace" --ignore-not-found=true
+            deleted_count=$((deleted_count + 1))
+        fi
+    done
+
+    # Find and delete any StatefulSet using the PVC
+    /opt/bin/kubectl get statefulsets -n "$namespace" -o json 2>/dev/null | grep -l "$pvc_name" | \
+    while IFS= read -r sts; do
+        if [ -n "$sts" ]; then
+            sts_name=$(echo "$sts" | cut -d'/' -f2)
+            echo "$(timestamp) -   Deleting StatefulSet: $sts_name"
+            /opt/bin/kubectl delete statefulset "$sts_name" -n "$namespace" --ignore-not-found=true
+            deleted_count=$((deleted_count + 1))
+        fi
+    done
+
+    # Check standalone ReplicaSets (not owned by Deployments)
+    /opt/bin/kubectl get replicasets -n "$namespace" --no-headers -o custom-columns=NAME:.metadata.name | \
+    while read rs_name; do
+        if [ -n "$rs_name" ]; then
+            rs_volumes=$(/opt/bin/kubectl get replicaset "$rs_name" -n "$namespace" -o jsonpath='{.spec.template.spec.volumes[*].persistentVolumeClaim.claimName}' 2>/dev/null || echo "")
+            if echo "$rs_volumes" | grep -q "$pvc_name"; then
+                owner_kind=$(/opt/bin/kubectl get replicaset "$rs_name" -n "$namespace" -o jsonpath='{.metadata.ownerReferences[0].kind}' 2>/dev/null || echo "")
+                if [ "$owner_kind" != "Deployment" ]; then
+                    echo "$(timestamp) -   Deleting standalone ReplicaSet: $rs_name"
+                    /opt/bin/kubectl delete replicaset "$rs_name" -n "$namespace" --ignore-not-found=true
+                    deleted_count=$((deleted_count + 1))
+                fi
+            fi
+        fi
+    done
+
+    # Find and delete any DaemonSet using the PVC
+    /opt/bin/kubectl get daemonsets -n "$namespace" -o json 2>/dev/null | grep -l "$pvc_name" | \
+    while IFS= read -r ds; do
+        if [ -n "$ds" ]; then
+            ds_name=$(echo "$ds" | cut -d'/' -f2)
+            echo "$(timestamp) -   Deleting DaemonSet: $ds_name"
+            /opt/bin/kubectl delete daemonset "$ds_name" -n "$namespace" --ignore-not-found=true
+            deleted_count=$((deleted_count + 1))
+        fi
+    done
+
+    # Find and delete any Job using the PVC
+    /opt/bin/kubectl get jobs -n "$namespace" -o json 2>/dev/null | grep -l "$pvc_name" | \
+    while IFS= read -r job; do
+        if [ -n "$job" ]; then
+            job_name=$(echo "$job" | cut -d'/' -f2)
+            echo "$(timestamp) -   Deleting Job: $job_name"
+            /opt/bin/kubectl delete job "$job_name" -n "$namespace" --ignore-not-found=true
+            deleted_count=$((deleted_count + 1))
+        fi
+    done
+
+    # Find and delete any CronJobs using the PVC
+    /opt/bin/kubectl get cronjobs -n "$namespace" -o json 2>/dev/null | grep -l "$pvc_name" | \
+    while IFS= read -r cronjob; do
+        if [ -n "$cronjob" ]; then
+            cronjob_name=$(echo "$cronjob" | cut -d'/' -f2)
+            echo "$(timestamp) -   Deleting CronJob: $cronjob_name"
+            /opt/bin/kubectl delete cronjob "$cronjob_name" -n "$namespace" --ignore-not-found=true
+            deleted_count=$((deleted_count + 1))
+        fi
+    done
+
+    # Find and delete any standalone Pods using the PVC
+    /opt/bin/kubectl get pods -n "$namespace" --no-headers -o custom-columns=NAME:.metadata.name | \
+    while read pod_name; do
+        if [ -n "$pod_name" ]; then
+            pod_volumes=$(/opt/bin/kubectl get pod "$pod_name" -n "$namespace" -o jsonpath='{.spec.volumes[*].persistentVolumeClaim.claimName}' 2>/dev/null || echo "")
+            if echo "$pod_volumes" | grep -q "$pvc_name"; then
+                owner_kind=$(/opt/bin/kubectl get pod "$pod_name" -n "$namespace" -o jsonpath='{.metadata.ownerReferences[0].kind}' 2>/dev/null || echo "")
+                if [ -z "$owner_kind" ]; then
+                    echo "$(timestamp) -   Deleting standalone Pod: $pod_name"
+                    /opt/bin/kubectl delete pod "$pod_name" -n "$namespace" --ignore-not-found=true
+                    deleted_count=$((deleted_count + 1))
+                fi
+            fi
+        fi
+    done
+
+    if [ $deleted_count -eq 0 ]; then
+        echo "$(timestamp) -   No workloads found using PVC $pvc_name"
+    else
+        echo "$(timestamp) -   Deleted $deleted_count workload(s) using PVC $pvc_name"
+    fi
+
+    echo "$(timestamp) -   Waiting for pods to terminate..."
+    sleep 5
+}
+
+total_pvcs=0
+processed_pvcs=0
+
+echo "$(timestamp) - Scanning for PVCs with associated PVs having reclaimPolicy=Delete..."
+
+while read namespace pvc_name pv_name; do
+    if [ -n "$pv_name" ] && [ "$pv_name" != "<none>" ]; then
+        total_pvcs=$((total_pvcs + 1))
+        reclaim_policy=$(/opt/bin/kubectl get pv "$pv_name" --no-headers -o custom-columns=RECLAIM:.spec.persistentVolumeReclaimPolicy 2>/dev/null || echo "")
+        if [ "$reclaim_policy" = "Delete" ]; then
+            processed_pvcs=$((processed_pvcs + 1))
+            echo "$(timestamp) - Processing PVC $pvc_name in namespace $namespace (PV: $pv_name has reclaimPolicy=Delete)"
+
+            delete_workloads_using_pvc "$namespace" "$pvc_name"
+            echo "$(timestamp) - Deleting PVC $pvc_name in namespace $namespace"
+            /opt/bin/kubectl delete pvc "$pvc_name" -n "$namespace" --ignore-not-found=true
+
+            echo "$(timestamp) - Completed processing PVC $pvc_name"
+            echo "---"
+        fi
+    fi
+done < <(/opt/bin/kubectl get pvc --all-namespaces --no-headers -o custom-columns=NAMESPACE:.metadata.namespace,NAME:.metadata.name,VOLUME:.spec.volumeName)
+
+echo "$(timestamp) - Script completed successfully!"
+echo "$(timestamp) - Summary: Processed $processed_pvcs PVC(s) out of $total_pvcs total PVC(s) found"
diff --git a/plugins/integrations/kubernetes-service/src/main/resources/script/deploy-csi-driver b/plugins/integrations/kubernetes-service/src/main/resources/script/deploy-csi-driver
new file mode 100644
index 00000000000..69684fa32bd
--- /dev/null
+++ b/plugins/integrations/kubernetes-service/src/main/resources/script/deploy-csi-driver
@@ -0,0 +1,46 @@
+#!/bin/bash -e
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+(/opt/bin/kubectl get pods -A | grep cloudstack-csi-controller) && exit 0
+
+if [ -e /opt/csi/snapshot-crds.yaml ]; then
+    /opt/bin/kubectl apply -f /opt/csi/snapshot-crds.yaml
+    sleep 5
+    /opt/bin/kubectl apply -f /opt/csi/manifest.yaml
+    exit 0
+else
+    TARGET_DIR="/opt/csi"
+    mkdir -p "$TARGET_DIR"
+    CSI_URLS=(
+      "https://github.com/cloudstack/cloudstack-csi-driver/releases/download/v3.0.0/snapshot-crds.yaml"
+      "https://github.com/cloudstack/cloudstack-csi-driver/releases/download/v3.0.0/manifest.yaml"
+    )
+    for url in "${CSI_URLS[@]}"; do
+        filename=$(basename "$url")
+
+        curl -sSL ${url} -o ${TARGET_DIR}/${filename}
+        if [ $? -ne 0 ]; then
+          echo "Unable to connect to the internet to download the relevant files to install and setup CloudStack CSI driver"
+          exit 1
+        else
+          /opt/bin/kubectl apply -f /opt/csi/snapshot-crds.yaml
+          /opt/bin/kubectl apply -f /opt/csi/manifest.yaml
+          exit 0
+        fi
+    done
+fi
diff --git a/scripts/util/create-kubernetes-binaries-iso.sh b/scripts/util/create-kubernetes-binaries-iso.sh
index 58aec922b7f..ebaf072771c 100755
--- a/scripts/util/create-kubernetes-binaries-iso.sh
+++ b/scripts/util/create-kubernetes-binaries-iso.sh
@@ -112,6 +112,11 @@ echo "Downloading kubernetes cluster provider ${PROVIDER_URL}"
 provider_conf_file="${working_dir}/provider.yaml"
 curl -sSL ${PROVIDER_URL} -o ${provider_conf_file}
 
+csi_conf_file="${working_dir}/manifest.yaml"
+echo "Including CloudStack CSI Driver manifest"
+wget https://github.com/cloudstack/cloudstack-csi-driver/releases/download/v3.0.0/snapshot-crds.yaml -O ${working_dir}/snapshot-crds.yaml
+wget https://github.com/cloudstack/cloudstack-csi-driver/releases/download/v3.0.0/manifest.yaml -O ${csi_conf_file}
+
 echo "Fetching k8s docker images..."
 ctr -v
 if [ $? -ne 0 ]; then
@@ -143,6 +148,10 @@ output=`printf "%s\n" ${output} ${autoscaler_image}`
 provider_image=`grep "image:" ${provider_conf_file} | cut -d ':' -f2- | tr -d ' '`
 output=`printf "%s\n" ${output} ${provider_image}`
 
+# Extract images from manifest.yaml and add to output
+csi_images=`grep "image:" "${csi_conf_file}" | cut -d ':' -f2- | tr -d ' ' | tr -d "'"`
+output=`printf "%s\n%s" "${output}" "${csi_images}"`
+
 while read -r line; do
     echo "Downloading image $line ---"
     if [[ $line == kubernetesui* ]] || [[ $line == apache* ]] || [[ $line == weaveworks* ]]; then
diff --git a/ui/public/locales/en.json b/ui/public/locales/en.json
index 0b6a4e3ceb9..3a2af9e6e15 100644
--- a/ui/public/locales/en.json
+++ b/ui/public/locales/en.json
@@ -697,6 +697,7 @@
 "label.cron": "Cron expression",
 "label.cron.mode": "Cron mode",
 "label.crosszones": "Cross Zones",
+"label.csienabled": "CSI Enabled",
 "label.currency": "Currency",
 "label.current": "Current",
 "label.current.storage": "Current storage",
@@ -973,6 +974,7 @@
 "label.elastic": "Elastic",
 "label.email": "Email",
 "label.enable.autoscale.vmgroup": "Enable AutoScaling Group",
+"label.enable.csi": "Enable CloudStack CSI Driver",
 "label.enable.custom.action": "Enable Custom Action",
 "label.enable.extension": "Enable Extension",
 "label.enable.host": "Enable Host",
diff --git a/ui/src/config/section/compute.js b/ui/src/config/section/compute.js
index 7688566df42..63d0e365db9 100644
--- a/ui/src/config/section/compute.js
+++ b/ui/src/config/section/compute.js
@@ -573,7 +573,7 @@ export default {
         const filters = ['cloud.managed', 'external.managed']
         return filters
       },
-      details: ['name', 'description', 'zonename', 'kubernetesversionname', 'autoscalingenabled', 'minsize', 'maxsize', 'size', 'controlnodes', 'etcdnodes', 'cpunumber', 'memory', 'keypair', 'cniconfigname', 'associatednetworkname', 'account', 'domain', 'zonename', 'clustertype', 'created'],
+      details: ['name', 'description', 'zonename', 'kubernetesversionname', 'autoscalingenabled', 'csienabled', 'minsize', 'maxsize', 'size', 'controlnodes', 'etcdnodes', 'cpunumber', 'memory', 'keypair', 'cniconfigname', 'associatednetworkname', 'account', 'domain', 'zonename', 'clustertype', 'created'],
       tabs: [
         {
           name: 'k8s',
diff --git a/ui/src/views/compute/CreateKubernetesCluster.vue b/ui/src/views/compute/CreateKubernetesCluster.vue
index f945da73213..68c7a35e3e5 100644
--- a/ui/src/views/compute/CreateKubernetesCluster.vue
+++ b/ui/src/views/compute/CreateKubernetesCluster.vue
@@ -207,6 +207,12 @@
           </template>
           <a-switch v-model:checked="form.advancedmode" />
         </a-form-item>
+        <a-form-item v-if="form.advancedmode" name="enablecsi" ref="enablecsi" :label="$t('label.enable.csi')">
+            <template #label>
+              <tooltip-label :title="$t('label.enable.csi')" :tooltip="apiParams.enablecsi.description"/>
+            </template>
+            <a-switch v-model:checked="form.enablecsi" />
+          </a-form-item>
         <a-form-item v-if="form.advancedmode" name="controlofferingid" ref="controlofferingid">
           <template #label>
             <tooltip-label :title="$t('label.cks.cluster.control.nodes.offeringid')" :tooltip="$t('label.cks.cluster.control.nodes.offeringid')"/>
@@ -901,6 +907,10 @@ export default {
           params.cniconfigurationid = values.cniconfigurationid
         }
 
+        if (values.enablecsi) {
+          params.enablecsi = values.enablecsi
+        }
+
         var idx = 0
         if (this.cniConfigValues) {
           for (const [key, value] of Object.entries(this.cniConfigValues)) {