Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #1711 (2024-05-16 22:13:21)

Add data caching feature with Local SSDs for PD
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
> /kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
/kind feature
> /kind flake

**What this PR does / why we need it**:
Adds feature to leverage Local SSDs as cache for persistent disks to reduce latency. It uses LVM as caching framework.
**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note
Added feature to leverage Local SSDs as cache for persistent disks.
```



Patch:
diff --git a/Dockerfile b/Dockerfile
index a380c6ce..53c94c15 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -25,7 +25,7 @@ RUN GOARCH=$(echo $TARGETPLATFORM | cut -f2 -d '/') GCE_PD_CSI_STAGING_VERSION=$
 FROM gke.gcr.io/debian-base:bullseye-v1.4.3-gke.5 as debian
 # Install necessary dependencies
 # google_nvme_id script depends on the following packages: nvme-cli, xxd, bash
-RUN clean-install util-linux e2fsprogs mount ca-certificates udev xfsprogs nvme-cli xxd bash
+RUN clean-install util-linux e2fsprogs mount ca-certificates udev xfsprogs nvme-cli xxd bash kmod lvm2 mdadm
 
 # Since we're leveraging apt to pull in dependencies, we use `gcr.io/distroless/base` because it includes glibc.
 FROM gcr.io/distroless/base-debian11 as distroless-base
@@ -54,6 +54,35 @@ COPY --from=debian /sbin/e2fsck /sbin/e2fsck
 COPY --from=debian /sbin/fsck /sbin/fsck
 COPY --from=debian /sbin/fsck* /sbin/
 COPY --from=debian /sbin/fsck.xfs /sbin/fsck.xfs
+# Add dependencies for LVM
+COPY --from=debian /etc/lvm /etc/lvm
+COPY --from=debian /etc/lvm* /etc/
+COPY --from=debian /lib/systemd/system/blk-availability.service /lib/systemd/system/blk-availability.service
+COPY --from=debian /lib/systemd/system/lvm2-lvmpolld.service /lib/systemd/system/lvm2-lvmpolld.service
+COPY --from=debian /lib/systemd/system/lvm2-lvmpolld.socket /lib/systemd/system/lvm2-lvmpolld.socket
+COPY --from=debian /lib/systemd/system/lvm2-monitor.service /lib/systemd/system/lvm2-monitor.service
+COPY --from=debian /lib/udev/rules.d/56-lvm.rules /lib/udev/rules.d/56-lvm.rules
+COPY --from=debian /sbin/fsadm /sbin/fsadm
+COPY --from=debian /sbin/lvm /sbin/lvm
+COPY --from=debian /sbin/lvmdump /sbin/lvmdump
+COPY --from=debian /sbin/lvmpolld /sbin/lvmpolld
+COPY --from=debian /usr/lib/tmpfiles.d /usr/lib/tmpfiles.d
+COPY --from=debian /usr/lib/tmpfiles.d/lvm2.conf /usr/lib/tmpfiles.d/lvm2.conf
+COPY --from=debian /sbin/lv* /sbin/
+COPY --from=debian /sbin/pv* /sbin/
+COPY --from=debian /sbin/vg* /sbin/
+COPY --from=debian /sbin/modprobe /sbin/modprobe
+COPY --from=debian /lib/udev /lib/udev
+COPY --from=debian /lib/udev/rules.d /lib/udev/rules.d
+COPY --from=debian /lib/udev/rules.d/55-dm.rules /lib/udev/rules.d/55-dm.rules
+COPY --from=debian /lib/udev/rules.d/60-persistent-storage-dm.rules /lib/udev/rules.d/60-persistent-storage-dm.rules
+COPY --from=debian /lib/udev/rules.d/95-dm-notify.rules /lib/udev/rules.d/95-dm-notify.rules
+COPY --from=debian /sbin/blkdeactivate /sbin/blkdeactivate
+COPY --from=debian /sbin/dmsetup /sbin/dmsetup
+COPY --from=debian /sbin/dmstats /sbin/dmstats
+COPY --from=debian /bin/ls /bin/ls
+# End of dependencies for LVM
+COPY --from=debian /sbin/mdadm /sbin/mdadm
 COPY --from=debian /sbin/mke2fs /sbin/mke2fs
 COPY --from=debian /sbin/mkfs* /sbin/
 COPY --from=debian /sbin/resize2fs /sbin/resize2fs
@@ -76,12 +105,18 @@ COPY --from=debian /lib/${LIB_DIR_PREFIX}-linux-gnu/libpcre.so.3 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libselinux.so.1 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libtinfo.so.6 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libe2p.so.2 \
+                   # The following does not exist in either lib or usr/lib
+                   # /lib/${LIB_DIR_PREFIX}-linux-gnu/libcap.so.2 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libcom_err.so.2 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libdevmapper.so.1.02.1 \
+                   /lib/${LIB_DIR_PREFIX}-linux-gnu/libm.so.6 \
+                   /lib/${LIB_DIR_PREFIX}-linux-gnu/libc.so.6 \
+                   /lib/${LIB_DIR_PREFIX}-linux-gnu/libdevmapper-event.so.1.02.1 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libext2fs.so.2 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libgcc_s.so.1 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/liblzma.so.5 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libreadline.so.8 \
+                   /lib/${LIB_DIR_PREFIX}-linux-gnu/libgpg-error.so.0 \
                    /lib/${LIB_DIR_PREFIX}-linux-gnu/libz.so.1 /lib/${LIB_DIR_PREFIX}-linux-gnu/
 
 COPY --from=debian /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libblkid.so.1 \
@@ -90,6 +125,11 @@ COPY --from=debian /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libblkid.so.1 \
                    /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libmount.so.1 \         
                    /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libudev.so.1 \
                    /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libuuid.so.1 \
+                   /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libzstd.so.1 \
+                   /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libaio.so.1 \
+                   /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libgcrypt.so.20 \
+                   /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libsystemd.so.0 \
+                   /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/liblz4.so.1 \
                    /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libacl.so.1 \
                    /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libattr.so.1 \
                    /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libedit.so.2 \
@@ -104,6 +144,11 @@ COPY --from=debian /usr/lib/${LIB_DIR_PREFIX}-linux-gnu/libblkid.so.1 \
 # Copy NVME support required script and rules into distroless base.
 COPY deploy/kubernetes/udev/google_nvme_id /lib/udev_containerized/google_nvme_id
 
+SHELL ["/bin/bash", "-c"]
+RUN /bin/sed -i -e "s/.*allow_mixed_block_sizes = 0.*/	allow_mixed_block_sizes = 1/" /etc/lvm/lvm.conf
+RUN /bin/sed -i -e "s/.*udev_sync = 1.*/	udev_sync = 0/" /etc/lvm/lvm.conf
+RUN /bin/sed -i -e "s/.*udev_rules = 1.*/	udev_rules = 0/" /etc/lvm/lvm.conf
+
 # Build stage used for validation of the output-image
 # See validate-container-linux-* targets in Makefile
 FROM output-image as validation-image
diff --git a/cmd/gce-pd-csi-driver/main.go b/cmd/gce-pd-csi-driver/main.go
index 83666475..3ee944fe 100644
--- a/cmd/gce-pd-csi-driver/main.go
+++ b/cmd/gce-pd-csi-driver/main.go
@@ -71,6 +71,7 @@ var (
 	formatAndMountTimeout       = flag.Duration("format-and-mount-timeout", 1*time.Minute, "The maximum duration of a format and mount operation before another such operation will be started. Used only if --serialize-format-and-mount")
 	fallbackRequisiteZonesFlag  = flag.String("fallback-requisite-zones", "", "Comma separated list of requisite zones that will be used if there are not sufficient zones present in requisite topologies when provisioning a disk")
 	enableStoragePoolsFlag      = flag.Bool("enable-storage-pools", false, "If set to true, the CSI Driver will allow volumes to be provisioned in Storage Pools")
+	enableDataCacheFlag         = flag.Bool("enable-data-cache", false, "If set to true, the CSI Driver will allow volumes to be provisioned with data cache configuration")
 
 	multiZoneVolumeHandleDiskTypesFlag = flag.String("multi-zone-volume-handle-disk-types", "", "Comma separated list of allowed disk types that can use the multi-zone volumeHandle. Used only if --multi-zone-volume-handle-enable")
 	multiZoneVolumeHandleEnableFlag    = flag.Bool("multi-zone-volume-handle-enable", false, "If set to true, the multi-zone volumeHandle feature will be enabled")
@@ -208,7 +209,7 @@ func handle() {
 		}
 		initialBackoffDuration := time.Duration(*errorBackoffInitialDurationMs) * time.Millisecond
 		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Millisecond
-		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, multiZoneVolumeHandleConfig, listVolumesConfig)
+		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig)
 	} else if *cloudConfigFilePath != "" {
 		klog.Warningf("controller service is disabled but cloud config given - it has no effect")
 	}
@@ -226,12 +227,19 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
-		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter)
+		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, *enableDataCacheFlag)
 		if *maxConcurrentFormatAndMount > 0 {
 			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
 		}
 	}
 
+	if *enableDataCacheFlag {
+		klog.V(2).Info("Raiding local ssds to setup data cache")
+		err := driver.RaidLocalSsds()
+		if err != nil {
+			klog.Fatalf("Failed to Raid local SSDs, unable to setup data caching, got error %v", err)
+		}
+	}
 	err = gceDriver.SetupGCEDriver(driverName, version, extraVolumeLabels, extraTags, identityServer, controllerServer, nodeServer)
 	if err != nil {
 		klog.Fatalf("Failed to initialize GCE CSI Driver: %v", err.Error())
diff --git a/deploy/kubernetes/base/node_linux/node.yaml b/deploy/kubernetes/base/node_linux/node.yaml
index ebf7779c..cc4650af 100644
--- a/deploy/kubernetes/base/node_linux/node.yaml
+++ b/deploy/kubernetes/base/node_linux/node.yaml
@@ -66,6 +66,8 @@ spec:
               mountPath: /run/udev
             - name: sys
               mountPath: /sys
+            - name: lib-modules
+              mountPath: /lib/modules
       volumes:
         - name: registration-dir
           hostPath:
@@ -101,6 +103,10 @@ spec:
           hostPath:
             path: /sys
             type: Directory
+        - name: lib-modules
+          hostPath:
+            path: /lib/modules
+            type: Directory
       # https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/
       # See "special case". This will tolerate everything. Node component should
       # be scheduled on all nodes.
diff --git a/pkg/common/constants.go b/pkg/common/constants.go
index 9851b11d..3c0e013e 100644
--- a/pkg/common/constants.go
+++ b/pkg/common/constants.go
@@ -32,4 +32,14 @@ const (
 
 	// Label that is set on a disk when it is used by a 'multi-zone' VolumeHandle
 	MultiZoneLabel = "goog-gke-multi-zone"
+
+	// Data cache mode
+	DataCacheModeWriteBack    = "writeback"
+	DataCacheModeWriteThrough = "writethrough"
+
+	ContextDataCacheSize = "data-cache-size"
+	ContextDataCacheMode = "data-cache-mode"
+
+	// Keys in the publish context
+	ContexLocalSsdCacheSize = "local-ssd-cache-size"
 )
diff --git a/pkg/common/parameters.go b/pkg/common/parameters.go
index 3b07f077..f6976f8b 100644
--- a/pkg/common/parameters.go
+++ b/pkg/common/parameters.go
@@ -18,7 +18,10 @@ package common
 
 import (
 	"fmt"
+	"strconv"
 	"strings"
+
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -32,7 +35,11 @@ const (
 	ParameterAvailabilityClass                = "availability-class"
 	ParameterKeyEnableConfidentialCompute     = "enable-confidential-storage"
 	ParameterKeyStoragePools                  = "storage-pools"
-	ParameterKeyResourceTags                  = "resource-tags"
+
+	// Parameters for Data Cache
+	ParameterKeyDataCacheSize = "data-cache-size"
+	ParameterKeyDataCacheMode = "data-cache-mode"
+	ParameterKeyResourceTags  = "resource-tags"
 
 	// Parameters for VolumeSnapshotClass
 	ParameterKeyStorageLocations = "storage-locations"
@@ -68,6 +75,15 @@ const (
 	tagKeyCreatedForSnapshotContentName = "kubernetes.io/created-for/volumesnapshotcontent/name"
 )
 
+type DataCacheParameters struct {
+	// Values: {string} in int64 form
+	// Default: ""
+	DataCacheSize string
+	// Values: writethrough, writeback
+	// Default: writethrough
+	DataCacheMode string
+}
+
 // DiskParameters contains normalized and defaulted disk parameters
 type DiskParameters struct {
 	// Values: pd-standard, pd-balanced, pd-ssd, or any other PD disk type. Not validated.
@@ -125,7 +141,7 @@ type StoragePool struct {
 // put them into a well defined struct making sure to default unspecified fields.
 // extraVolumeLabels are added as labels; if there are also labels specified in
 // parameters, any matching extraVolumeLabels will be overridden.
-func ExtractAndDefaultParameters(parameters map[string]string, driverName string, extraVolumeLabels map[string]string, enableStoragePools bool, extraTags map[string]string) (DiskParameters, error) {
+func ExtractAndDefaultParameters(parameters map[string]string, driverName string, extraVolumeLabels map[string]string, enableStoragePools bool, enableDataCache bool, extraTags map[string]string) (DiskParameters, DataCacheParameters, error) {
 	p := DiskParameters{
 		DiskType:             "pd-standard",           // Default
 		ReplicationType:      replicationTypeNone,     // Default
@@ -135,6 +151,12 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 		ResourceTags:         make(map[string]string), // Default
 	}
 
+	// Set data cache mode default
+	d := DataCacheParameters{}
+	if enableDataCache && parameters[ParameterKeyDataCacheSize] != "" {
+		d.DataCacheMode = DataCacheModeWriteThrough
+	}
+
 	for k, v := range extraVolumeLabels {
 		p.Labels[k] = v
 	}
@@ -169,7 +191,7 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 		case ParameterKeyLabels:
 			paramLabels, err := ConvertLabelsStringToMap(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid labels parameter: %w", err)
+				return p, d, fmt.Errorf("parameters contain invalid labels parameter: %w", err)
 			}
 			// Override any existing labels with those from this parameter.
 			for labelKey, labelValue := range paramLabels {
@@ -178,19 +200,19 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 		case ParameterKeyProvisionedIOPSOnCreate:
 			paramProvisionedIOPSOnCreate, err := ConvertStringToInt64(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid provisionedIOPSOnCreate parameter: %w", err)
+				return p, d, fmt.Errorf("parameters contain invalid provisionedIOPSOnCreate parameter: %w", err)
 			}
 			p.ProvisionedIOPSOnCreate = paramProvisionedIOPSOnCreate
 		case ParameterKeyProvisionedThroughputOnCreate:
 			paramProvisionedThroughputOnCreate, err := ConvertMiStringToInt64(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid provisionedThroughputOnCreate parameter: %w", err)
+				return p, d, fmt.Errorf("parameters contain invalid provisionedThroughputOnCreate parameter: %w", err)
 			}
 			p.ProvisionedThroughputOnCreate = paramProvisionedThroughputOnCreate
 		case ParameterAvailabilityClass:
 			paramAvailabilityClass, err := ConvertStringToAvailabilityClass(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid availability class parameter: %w", err)
+				return p, d, fmt.Errorf("parameters contain invalid availability class parameter: %w", err)
 			}
 			if paramAvailabilityClass == ParameterRegionalHardFailoverClass {
 				p.ForceAttach = true
@@ -198,38 +220,59 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 		case ParameterKeyEnableConfidentialCompute:
 			paramEnableConfidentialCompute, err := ConvertStringToBool(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid value for enable-confidential-storage parameter: %w", err)
+				return p, d, fmt.Errorf("parameters contain invalid value for enable-confidential-storage parameter: %w", err)
 			}
 
 			if paramEnableConfidentialCompute {
 				// DiskEncryptionKmsKey is needed to enable confidentialStorage
 				if val, ok := parameters[ParameterKeyDiskEncryptionKmsKey]; !ok || !isValidDiskEncryptionKmsKey(val) {
-					return p, fmt.Errorf("Valid %v is required to enable ConfidentialStorage", ParameterKeyDiskEncryptionKmsKey)
+					return p, d, fmt.Errorf("Valid %v is required to enable ConfidentialStorage", ParameterKeyDiskEncryptionKmsKey)
 				}
 			}
 
 			p.EnableConfidentialCompute = paramEnableConfidentialCompute
 		case ParameterKeyStoragePools:
 			if !enableStoragePools {
-				return p, fmt.Errorf("parameters contains invalid option %q", ParameterKeyStoragePools)
+				return p, d, fmt.Errorf("parameters contains invalid option %q", ParameterKeyStoragePools)
 			}
 			storagePools, err := ParseStoragePools(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid value for %s parameter: %w", ParameterKeyStoragePools, err)
+				return p, d, fmt.Errorf("parameters contain invalid value for %s parameter: %w", ParameterKeyStoragePools, err)
 			}
 			p.StoragePools = storagePools
+		case ParameterKeyDataCacheSize:
+			if !enableDataCache {
+				return p, d, fmt.Errorf("data caching enabled: %v; parameters contains invalid option %q", enableDataCache, ParameterKeyDataCacheSize)
+			}
+			// TODO: need to parse or validate the string
+
+			paramDataCacheSize, err := ConvertGiStringToInt64(v)
+			if err != nil {
+				return p, d, fmt.Errorf("parameters contain invalid dataCacheSize parameter: %w", err)
+			}
+			d.DataCacheSize = strconv.FormatInt(paramDataCacheSize, 10)
+			klog.V(2).Infof("====== Data cache size is %v ======", v)
+		case ParameterKeyDataCacheMode:
+			if !enableDataCache {
+				return p, d, fmt.Errorf("data caching enabled %v; parameters contains invalid option %q", enableDataCache, ParameterKeyDataCacheSize)
+			}
+			if err := ValidateDataCacheMode(v); err != nil {
+				return p, d, fmt.Errorf("parameters contains invalid option: %w", err)
+			}
+			d.DataCacheMode = v
+			klog.V(2).Infof("====== Data cache mode is %v ======", v)
 		case ParameterKeyResourceTags:
 			if err := extractResourceTagsParameter(v, p.ResourceTags); err != nil {
-				return p, err
+				return p, d, err
 			}
 		default:
-			return p, fmt.Errorf("parameters contains invalid option %q", k)
+			return p, d, fmt.Errorf("parameters contains invalid option %q", k)
 		}
 	}
 	if len(p.Tags) > 0 {
 		p.Tags[tagKeyCreatedBy] = driverName
 	}
-	return p, nil
+	return p, d, nil
 }
 
 func ExtractAndDefaultSnapshotParameters(parameters map[string]string, driverName string, extraTags map[string]string) (SnapshotParameters, error) {
diff --git a/pkg/common/runcmd.go b/pkg/common/runcmd.go
new file mode 100644
index 00000000..71240d2a
--- /dev/null
+++ b/pkg/common/runcmd.go
@@ -0,0 +1,71 @@
+package common
+
+import (
+	"fmt"
+	"os/exec"
+	"strings"
+
+	"k8s.io/klog/v2"
+)
+
+const (
+	// Error thrown by exec cmd.Run() when process spawned by cmd.Start() completes before cmd.Wait() is called (see - k/k issue #103753)
+	errNoChildProcesses = "wait: no child processes"
+)
+
+// RunCommand wraps a k8s exec to deal with the no child process error. Same as exec.CombinedOutput.
+// On error, the output is included so callers don't need to echo it again.
+
+func RunCommand(pipeCmd string, pipeCmdArg string, cmd1 string, execCmdArgs ...string) ([]byte, error) {
+	execCmd1 := exec.Command(cmd1, execCmdArgs...)
+
+	if pipeCmd != "" {
+		output, err := execPipeCommand(pipeCmd, pipeCmdArg, execCmd1)
+		if err != nil {
+			return nil, fmt.Errorf("%s %s failed here: %w; output: %s", pipeCmd, pipeCmdArg, err, string(output))
+		}
+		return output, nil
+	}
+	output, err := execCmd1.CombinedOutput()
+	if err != nil {
+		err = checkError(err, *execCmd1)
+		return nil, fmt.Errorf("%s %s failed here 2: %w; output: %s", cmd1, strings.Join(execCmdArgs, " "), err, string(output))
+	}
+
+	return output, nil
+}
+
+func checkError(err error, execCmd exec.Cmd) error {
+	if err.Error() == errNoChildProcesses {
+		if execCmd.ProcessState.Success() {
+			// If the process succeeded, this can be ignored, see k/k issue #103753
+			return nil
+		}
+		// Get actual error
+		klog.Infof("Errored here")
+		err = &exec.ExitError{ProcessState: execCmd.ProcessState}
+	}
+	return err
+}
+func execPipeCommand(pipeCmd string, pipeCmdArg string, execCmd1 *exec.Cmd) ([]byte, error) {
+
+	execPipeCmd := exec.Command(pipeCmd, pipeCmdArg)
+	stdoutPipe, err := execCmd1.StdoutPipe()
+	if err != nil {
+		klog.Errorf("failed command %v: got error:%v", execCmd1, err)
+	}
+	err = execCmd1.Start()
+	if err != nil {
+		klog.Infof("errored running command %v; error %v; ", execCmd1, err)
+	}
+	defer stdoutPipe.Close()
+
+	execPipeCmd.Stdin = stdoutPipe
+	output, err := execPipeCmd.CombinedOutput()
+	if err != nil {
+		err = checkError(err, *execPipeCmd)
+		return nil, fmt.Errorf("%s failed: %w; output: %s", pipeCmd, err, string(output))
+	}
+
+	return output, nil
+}
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index 705d55fd..a4695a1d 100644
--- a/pkg/common/utils.go
+++ b/pkg/common/utils.go
@@ -92,6 +92,8 @@ var (
 		http.StatusNotFound:        codes.NotFound,
 	}
 
+	validDataCacheMode = []string{DataCacheModeWriteBack, DataCacheModeWriteThrough}
+
 	// Regular expressions for validating parent_id, key and value of a resource tag.
 	regexParent = regexp.MustCompile(`(^[1-9][0-9]{0,31}$)|(^[a-z][a-z0-9-]{4,28}[a-z0-9]$)`)
 	regexKey    = regexp.MustCompile(`^[a-zA-Z0-9]([0-9A-Za-z_.-]{0,61}[a-zA-Z0-9])?$`)
@@ -381,6 +383,15 @@ func ConvertMiStringToInt64(str string) (int64, error) {
 	return volumehelpers.RoundUpToMiB(quantity)
 }
 
+// ConvertGiStringToInt64 converts a GiB string to int64
+func ConvertGiStringToInt64(str string) (int64, error) {
+	quantity, err := resource.ParseQuantity(str)
+	if err != nil {
+		return -1, err
+	}
+	return volumehelpers.RoundUpToGiB(quantity)
+}
+
 // ConvertStringToBool converts a string to a boolean.
 func ConvertStringToBool(str string) (bool, error) {
 	switch strings.ToLower(str) {
@@ -578,6 +589,22 @@ func VolumeIdAsMultiZone(volumeId string) (string, error) {
 	return strings.Join(splitId, "/"), nil
 }
 
+func StringInSlice(s string, list []string) bool {
+	for _, v := range list {
+		if v == s {
+			return true
+		}
+	}
+	return false
+}
+
+func ValidateDataCacheMode(s string) error {
+	if StringInSlice(s, validDataCacheMode) {
+		return nil
+	}
+	return fmt.Errorf("invalid data-cache-mode %s. Only \"writeback\" and \"writethrough\" is a valid input", s)
+}
+
 // NewLimiter returns a token bucket based request rate limiter after initializing
 // the passed values for limit, burst (or token bucket) size. If opted for emptyBucket
 // all initial tokens are reserved for the first burst.
diff --git a/pkg/gce-pd-csi-driver/cache.go b/pkg/gce-pd-csi-driver/cache.go
new file mode 100644
index 00000000..5d77690e
--- /dev/null
+++ b/pkg/gce-pd-csi-driver/cache.go
@@ -0,0 +1,380 @@
+package gceGCEDriver
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	csi "github.com/container-storage-interface/spec/lib/go/csi"
+
+	"k8s.io/klog/v2"
+
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+)
+
+const (
+	cacheSuffix        = "csi-fast"
+	mainLvSuffix       = "csi-main"
+	raidedLocalSsdName = "csi-driver-data-cache"
+	raidMode           = "0"
+	raidedLssdPrefix   = "/dev/md/"
+)
+
+var raidedLocalSsdPath = raidedLssdPrefix + raidedLocalSsdName
+
+func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId string) (string, error) {
+	volumeId := req.GetVolumeId()
+	volumeGroupName := getVolumeGroupName(nodeId)
+	mainDevicePath := "/dev/" + volumeGroupName + "/" + getLvName(mainLvSuffix, volumeId)
+	mainLvName := getLvName(mainLvSuffix, volumeId)
+	klog.V(2).Infof("============================== Start LVM PoC NodeStageVolume Steps ==============================")
+	klog.V(2).Infof("============================== volumeGroupName is %v ==============================", volumeGroupName)
+
+	info, err := common.RunCommand("grep", raidedLocalSsdName, "ls", raidedLssdPrefix)
+	if err != nil {
+		klog.Errorf("================== failed while listing raided devices, err: %v, output:%v ===============", err, info)
+	}
+	infoString := strings.TrimSpace(string(info))
+	klog.V(2).Infof("=================== Got Raided LSSD name %v ===================", infoString)
+	raidedLocalSsdPath = raidedLssdPrefix + infoString
+
+	klog.V(2).Infof("============================== vgscan before vgcreate ==============================")
+	args := []string{}
+	info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgscan", args...)
+	if err != nil {
+		klog.Errorf("vgscan error %v: %s", err, info)
+	}
+	klog.V(2).Infof("============================== vgscan info contains volumeGroupName or not %v ==============================", strings.Contains(string(info), volumeGroupName))
+	// Check if the required volume group already exists
+	if strings.Contains(string(info), volumeGroupName) {
+		klog.V(2).Infof("============================== VG exists, now check if PD is part of VG ==============================")
+
+		// Clean up Volume Group before adding the PD
+		reduceVolumeGroup(volumeGroupName, true)
+	} else {
+		err := createVg(volumeGroupName, devicePath, raidedLocalSsdPath)
+		if err != nil {
+			return mainDevicePath, err
+		}
+	}
+
+	// Check if the Physical Volume(PV) is part of some other volume group
+	args = []string{
+		"--select",
+		"pv_name=" + devicePath,
+		"-o",
+		"vg_name",
+	}
+	info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "pvs", args...)
+	if err != nil {
+		klog.Errorf("errored while checking physical volume details %v: %s", err, info)
+		// On error info contains the error message which we cannot use for further steps
+		info = nil
+	}
+
+	infoString = strings.TrimSpace(strings.ReplaceAll(string(info), "\n", " "))
+	infoString = strings.ReplaceAll(infoString, ".", "")
+	infoString = strings.ReplaceAll(infoString, "\"", "")
+	infoSlice := strings.Split(strings.TrimSpace(infoString), " ")
+	vgNameForPv := strings.TrimSpace(infoSlice[(len(infoSlice) - 1)])
+	klog.V(2).Infof("============================== Physical volume is part of Volume group: %v ==============================", vgNameForPv)
+	if vgNameForPv == volumeGroupName {
+		klog.V(2).Infof("============================== Physical Volume(PV) already exists in the Volume Group ==============================")
+	} else if vgNameForPv != "VG" && vgNameForPv != "" {
+
+		klog.V(2).Infof("============================== Deactivate VG %s ==============================", vgNameForPv)
+		info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgchange", []string{"-an", vgNameForPv}...)
+		if err != nil {
+			klog.Errorf("Errored while deactivating VG %v: err: %v: %s", vgNameForPv, err, info)
+		}
+
+		reduceVolumeGroup(vgNameForPv, false)
+		_, isCached := isCachingSetup(mainLvName)
+		// We will continue to uncache even if it errors to check caching as it is not a terminal issue.
+
+		if isCached {
+			klog.Infof("============================== Uncaching the LV %v==============================", mainLvName)
+			// Uncache LV
+			args = []string{
+				"--uncache",
+				vgNameForPv + "/" + mainLvName,
+				"--force",
+				"-y", // force remove cache without flushing data
+			}
+			info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvconvert", args...)
+			if err != nil {
+				klog.Errorf("errored while uncaching main LV. %v: %s", err, info)
+			}
+
+			reduceVolumeGroup(vgNameForPv, false)
+		}
+		klog.V(2).Infof("============================== Merge VG %v to Node VG %v ==============================", vgNameForPv, volumeGroupName)
+		info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgmerge", []string{volumeGroupName, vgNameForPv}...)
+		if err != nil {
+			klog.Errorf("Errored while merging Volume group %s into %s %v: %s", vgNameForPv, volumeGroupName, err, info)
+		}
+
+	} else {
+		klog.V(2).Infof("============================== Extend Node VG %v for PV %v ==============================", volumeGroupName, devicePath)
+		info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgextend", []string{volumeGroupName, devicePath}...)
+		if err != nil {
+			klog.Errorf("Errored while extending VGs %v: %s", err, info)
+		}
+	}
+
+	// Create LV if not already created
+	args = []string{
+		"--select",
+		"vg_name=" + volumeGroupName,
+		"-o",
+		"lv_name",
+	}
+	lvList, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvs", args...)
+	if err != nil {
+		return mainDevicePath, fmt.Errorf("lv list error %w: %s", err, info)
+	}
+	klog.Infof("==============================Got LVs %s on Volume group %s ==============================", string(lvList), volumeGroupName)
+	if !strings.Contains(string(lvList), mainLvName) {
+		// lvcreate -n main -l 100%PVS cachegroup /dev/sdb
+		klog.V(2).Infof("============================== lvcreate main cache layer ==============================")
+		args = []string{
+			"--yes",
+			"-n",
+			mainLvName,
+			"-l",
+			"100%PVS",
+			volumeGroupName,
+			devicePath,
+		}
+		info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvcreate", args...)
+		if err != nil {
+			return mainDevicePath, fmt.Errorf("lvcreate error %w: %s", err, info)
+		}
+
+	}
+	err, isCached := isCachingSetup(mainLvName)
+	if err != nil {
+		klog.Errorf("faild to check if caching ius setup for LV. Continuing to setup caching.")
+	}
+	cacheLvName := getLvName(cacheSuffix, volumeId)
+	if isCached {
+		// Validate that cache is setup for required size
+		klog.V(2).Infof("==============================Assuming valid data cache size and mode, resizing is not supported==============================")
+	} else {
+		fastCacheSize := req.GetPublishContext()[common.ContexLocalSsdCacheSize]
+		chunkSize := "960" // Cannot use default chunk size(64KiB) as it errors on maxChunksAllowed. Unit - KiB
+		klog.V(2).Infof("============================== fastCacheSize is %v ==============================", fastCacheSize)
+		klog.V(2).Infof("============================== lvcreate fast cache layer again with the VolumeGroup %v==============================", volumeGroupName)
+		args = []string{
+			"--yes",
+			"-n",
+			cacheLvName,
+			"-L",
+			fastCacheSize,
+			volumeGroupName,
+			raidedLocalSsdPath,
+		}
+		info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvcreate", args...)
+		if err != nil {
+			klog.V(2).Infof("============================== lvcreate error %v: %s ==============================", err, info)
+			return mainDevicePath, fmt.Errorf("lvcreate error %w: %s", err, info)
+		}
+
+		// Once caching is setup, link the PD to cache
+		klog.V(2).Infof("============================== lvconvert fast and main to cache ==============================")
+		args = []string{
+			"--type",
+			"cache",
+			"--cachevol",
+			cacheLvName,
+			"--zero",
+			"y",
+			"--cachemode",
+			req.GetPublishContext()[common.ContextDataCacheMode],
+			volumeGroupName + "/" + mainLvName,
+			"--chunksize",
+			string(chunkSize),
+			"--force",
+			"-y",
+		}
+		info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvconvert", args...)
+		if err != nil {
+			klog.V(2).Infof("============================== lvconvert error %v: %s ==============================", err, info)
+			return mainDevicePath, fmt.Errorf("lvconvert error %w: %s", err, info)
+		}
+	}
+
+	// activate all the LVs in the Volume group
+	klog.V(2).Infof("============================== Activate Volume group %s ==============================", volumeGroupName)
+	info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgchange", []string{"-ay", volumeGroupName}...)
+	if err != nil {
+		klog.Errorf("Failed to activate VG %v %v:%s", volumeGroupName, err, info)
+	}
+
+	return mainDevicePath, nil
+}
+
+func cleanupCache(volumeId string, nodeId string) error {
+
+	volumeGroupName := getVolumeGroupName(nodeId)
+	mainLvName := getLvName(mainLvSuffix, volumeId)
+	klog.V(2).Infof("============================== Deactivating volume %s/%s ==============================", volumeGroupName, mainLvName)
+	args := []string{
+		"-an",
+		"/dev/" + volumeGroupName + "/" + mainLvName,
+	}
+	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvchange", args...)
+	if err != nil {
+		klog.Errorf("Errored while deactivating the disk  %v: %s", err, info)
+	}
+	args = []string{
+		"--uncache",
+		volumeGroupName + "/" + mainLvName,
+	}
+	info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "lvconvert", args...)
+	if err != nil {
+		return fmt.Errorf("errored while uncaching the disk %w: %s", err, info)
+	}
+	return nil
+}
+
+func getVolumeGroupName(nodePath string) string {
+	nodeSlice := strings.Split(nodePath, "/")
+	nodeId := nodeSlice[len(nodeSlice)-1]
+	return fmt.Sprintf("csi-vg-%s", nodeId)
+}
+
+func getLvName(suffix string, volumeId string) string {
+	pvcNameStringSlice := strings.Split(volumeId, "/")
+	pvcName := pvcNameStringSlice[len(pvcNameStringSlice)-1]
+	return fmt.Sprintf("%s-%s", suffix, pvcName)
+}
+
+func createVg(volumeGroupName string, devicePath string, raidedLocalSsds string) error {
+	klog.V(2).Infof("============================== vgcreate ==============================")
+	args := []string{
+		"--zero",
+		"y",
+		volumeGroupName,
+		raidedLocalSsds,
+	}
+	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgcreate", args...)
+	if err != nil {
+		klog.Errorf("vgcreate error %v: %s", err, info)
+		return fmt.Errorf("vgcreate error %w: %s", err, info)
+	}
+
+	klog.V(2).Infof("============================== vgscan after vgcreate ==============================")
+	args = []string{}
+	info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgscan", args...)
+	if err != nil {
+		klog.Errorf("vgscan error %v: %s", err, info)
+	} else {
+		klog.V(2).Infof("============================== vgscan info %s  ==============================", info)
+	}
+	return nil
+}
+
+func reduceVolumeGroup(volumeGroupName string, force bool) {
+	klog.V(2).Infof("============================== Cleanup VG %s ==============================", volumeGroupName)
+	args := []string{
+		"--removemissing",
+		volumeGroupName,
+	}
+	if force {
+		args = append(args, "--force")
+	}
+	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgreduce", args...)
+	if err != nil {
+		klog.Errorf("Errored while cleaning up volume group %v: %s", err, info)
+	}
+}
+
+func RaidLocalSsds() error {
+	isAlreadyRaided, err := isRaided()
+	if err != nil {
+		klog.V(2).Infof("============================== Errored while scanning for available LocalSSDs err:%v; continuing Raiding ==============================", err)
+	} else if isAlreadyRaided {
+		klog.V(2).Infof("============================== Local SSDs are already RAIDed ==============================")
+		return nil
+	}
+	info, err := common.RunCommand("grep" /* pipedCmd */, "DevicePath" /* pipeCmdArg */, "nvme", []string{"list", "-o", "json"}...)
+	if err != nil {
+		return fmt.Errorf("errored while scanning available NVME disks info: %v; err:%v", info, err)
+	}
+	infoString := strings.ReplaceAll(string(info), "\"", "")
+	infoString = strings.TrimSpace(strings.ReplaceAll(infoString, ",", " "))
+	infoSlice := strings.Split(infoString, "\n")
+	klog.V(2).Infof("============================== NVME list %v ==============================", infoSlice)
+	diskList := []string{}
+	for _, diskInfo := range infoSlice {
+		diskName := strings.TrimSpace(diskInfo)
+		diskName = strings.TrimSpace(strings.Split(diskName, ":")[1])
+		diskList = append(diskList, diskName)
+	}
+	nvmeDiskCount := len(diskList)
+	nvmeDiskList := strings.Join(diskList, " ")
+	if nvmeDiskCount == 0 {
+		klog.Infof("No NVME disks found for RAIDing")
+		return nil
+	}
+	klog.V(2).Infof("============================== nvmeDiskCount %v; nvmeDiskList: %v; diskList %v ==============================", nvmeDiskCount, nvmeDiskList, diskList)
+	args := []string{
+		"--create",
+		raidedLssdPrefix + raidedLocalSsdName,
+		"-l" + raidMode,
+		// Force RAIDing as sometime it might fail for caution if there is just 1 LSSD present as 1 LSSD need not be RAIDed
+		"--force",
+		"-n",
+		strconv.Itoa(nvmeDiskCount),
+	}
+	args = append(args, diskList...)
+	info, err = common.RunCommand("" /* pipedCmd */, "" /* pipeCmdArg */, "mdadm", args...)
+	if err != nil {
+		return fmt.Errorf("errored while RAIDing LSSDs info: %v; err:%v", info, err)
+	}
+	// Validate if Raided successfully
+	isAlreadyRaided, err = isRaided()
+	if err != nil {
+		klog.V(2).Infof("============================== Errored while scanning for available raided LocalSSDs err:%v ==============================", err)
+	}
+	if !isAlreadyRaided {
+		return fmt.Errorf("failed raiding, raided device not found on scanning")
+	}
+	return nil
+}
+
+func isRaided() (bool, error) {
+	args := []string{
+		"--detail",
+		"--scan",
+	}
+	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipeCmdArg */, "mdadm", args...)
+	if err != nil {
+		return false, fmt.Errorf("errored while scanning for raided LSSD %v: %s", err, info)
+	}
+	klog.V(2).Infof("============================== Got LSSDs %v ==============================", string(info))
+	if info != nil && strings.Contains(string(info), raidedLocalSsdName) {
+		return true, nil
+	}
+	return false, nil
+}
+
+func isCachingSetup(mainLvName string) (error, bool) {
+	// Verify caching is setup for PD
+	klog.V(2).Infof("============================== Verifying if caching is setup for %v ==============================", mainLvName)
+	args := []string{
+		"--select",
+		"lv_name=" + mainLvName,
+		"-o",
+		"pool_lv",
+	}
+	poolName, err := common.RunCommand("" /* pipedCmd */, "" /* pipeCmdArg */, "lvs", args...)
+	if err != nil {
+		return fmt.Errorf("lvs error %w", err), false
+	}
+	if strings.Contains(string(poolName), "csi-fast") {
+		return nil, true
+	}
+	return nil, false
+}
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 1bcfdf25..31cc8eb1 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -43,6 +43,11 @@ import (
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
 )
 
+var (
+	// Keys in the volume context.
+	contextForceAttach = "force-attach"
+)
+
 type GCEControllerServer struct {
 	Driver        *GCEDriver
 	CloudProvider gce.GCECompute
@@ -105,6 +110,9 @@ type GCEControllerServer struct {
 	// If set to true, the CSI Driver will allow volumes to be provisioned in Storage Pools.
 	enableStoragePools bool
 
+	// If set to true, the CSI Driver will allow volumes to be provisioned with data cache configuration
+	enableDataCache bool
+
 	multiZoneVolumeHandleConfig MultiZoneVolumeHandleConfig
 
 	listVolumesConfig ListVolumesConfig
@@ -187,9 +195,6 @@ const (
 	// See https://github.com/grpc/grpc/blob/master/include/grpc/impl/codegen/grpc_types.h#L503)
 	maxListVolumesResponseEntries = 500
 
-	// Keys in the volume context.
-	contextForceAttach = "force-attach"
-
 	resourceApiScheme  = "https"
 	resourceApiService = "compute"
 	resourceProject    = "projects"
@@ -309,7 +314,8 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 	// Apply Parameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
-	params, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels, gceCS.enableStoragePools, gceCS.Driver.extraTags)
+	params, dataCacheParams, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels, gceCS.enableStoragePools, gceCS.enableDataCache, gceCS.Driver.extraTags)
+	klog.V(2).Infof("====== dataCacheParams are %v ======", dataCacheParams)
 	diskTypeForMetric = params.DiskType
 	enableConfidentialCompute = strconv.FormatBool(params.EnableConfidentialCompute)
 	hasStoragePools := len(params.StoragePools) > 0
@@ -407,7 +413,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 		// If there is no validation error, immediately return success
 		klog.V(4).Infof("CreateVolume succeeded for disk %v, it already exists and was compatible", volKey)
-		return generateCreateVolumeResponse(existingDisk, zones, params)
+		return generateCreateVolumeResponse(existingDisk, zones, params, dataCacheParams, gceCS.enableDataCache)
 	}
 
 	snapshotID := ""
@@ -521,7 +527,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	}
 
 	klog.V(4).Infof("CreateVolume succeeded for disk %v", volKey)
-	return generateCreateVolumeResponse(disk, zones, params)
+	return generateCreateVolumeResponse(disk, zones, params, dataCacheParams, gceCS.enableDataCache)
 
 }
 
@@ -678,6 +684,16 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 		PublishContext: nil,
 	}
 
+	klog.V(2).Infof("====== ControllerPublishVolume VolumeContext is %v ======", req.GetVolumeContext())
+	// Set data cache publish context
+	if gceCS.enableDataCache && req.GetVolumeContext() != nil {
+		if req.GetVolumeContext()[common.ContextDataCacheSize] != "" {
+			pubVolResp.PublishContext = map[string]string{}
+			pubVolResp.PublishContext[common.ContexLocalSsdCacheSize] = req.GetVolumeContext()[common.ContextDataCacheSize]
+			pubVolResp.PublishContext[common.ContextDataCacheMode] = req.GetVolumeContext()[common.ContextDataCacheMode]
+		}
+	}
+
 	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
 		return nil, status.Errorf(codes.NotFound, "could not split nodeID: %v", err.Error()), nil
@@ -770,6 +786,7 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 	}
 
 	klog.V(4).Infof("ControllerPublishVolume succeeded for disk %v to instance %v", volKey, nodeID)
+	klog.V(2).Infof("====== ControllerPublishVolumeResponse is %v ======", pubVolResp)
 	return pubVolResp, nil, disk
 }
 
@@ -937,7 +954,8 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	}
 
 	// Validate the disk parameters match the disk we GET
-	params, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels, gceCS.enableStoragePools, gceCS.Driver.extraTags)
+	params, dataCacheParams, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels, gceCS.enableStoragePools, gceCS.enableDataCache, gceCS.Driver.extraTags)
+	klog.V(2).Infof("====== dataCacheParams are %v ======", dataCacheParams)
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err.Error())
 	}
@@ -1953,7 +1971,7 @@ func extractVolumeContext(context map[string]string) (*PDCSIContext, error) {
 	return info, nil
 }
 
-func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params common.DiskParameters) (*csi.CreateVolumeResponse, error) {
+func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params common.DiskParameters, dataCacheParams common.DataCacheParameters, enableDataCache bool) (*csi.CreateVolumeResponse, error) {
 	volumeId, err := getResourceId(disk.GetSelfLink())
 	if err != nil {
 		return nil, fmt.Errorf("cannot get volume id from %s: %w", disk.GetSelfLink(), err)
@@ -1974,6 +1992,14 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params co
 			AccessibleTopology: tops,
 		},
 	}
+	// Set data cache volume context
+	if enableDataCache && dataCacheParams != (common.DataCacheParameters{}) {
+		if createResp.Volume.VolumeContext == nil {
+			createResp.Volume.VolumeContext = map[string]string{}
+		}
+		createResp.Volume.VolumeContext[common.ContextDataCacheMode] = dataCacheParams.DataCacheMode
+		createResp.Volume.VolumeContext[common.ContextDataCacheSize] = dataCacheParams.DataCacheSize
+	}
 	snapshotID := disk.GetSnapshotId()
 	imageID := disk.GetImageId()
 	diskID := disk.GetSourceDiskId()
@@ -2008,6 +2034,7 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params co
 		}
 		createResp.Volume.ContentSource = contentSource
 	}
+	klog.V(2).Infof("====== CreateVolumeResponse is %v ======", createResp)
 	return createResp, nil
 }
 
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver.go b/pkg/gce-pd-csi-driver/gce-pd-driver.go
index ba1a0c0b..8f062fa6 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver.go
@@ -143,7 +143,7 @@ func NewIdentityServer(gceDriver *GCEDriver) *GCEIdentityServer {
 	}
 }
 
-func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter) *GCENodeServer {
+func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter, enableDataCache bool) *GCENodeServer {
 	return &GCENodeServer{
 		Driver:          gceDriver,
 		Mounter:         mounter,
@@ -151,10 +151,11 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 		MetadataService: meta,
 		volumeLocks:     common.NewVolumeLocks(),
 		VolumeStatter:   statter,
+		EnableDataCache: enableDataCache,
 	}
 }
 
-func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, errorBackoffInitialDuration, errorBackoffMaxDuration time.Duration, fallbackRequisiteZones []string, enableStoragePools bool, multiZoneVolumeHandleConfig MultiZoneVolumeHandleConfig, listVolumesConfig ListVolumesConfig) *GCEControllerServer {
+func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, errorBackoffInitialDuration, errorBackoffMaxDuration time.Duration, fallbackRequisiteZones []string, enableStoragePools bool, enableDataCache bool, multiZoneVolumeHandleConfig MultiZoneVolumeHandleConfig, listVolumesConfig ListVolumesConfig) *GCEControllerServer {
 	return &GCEControllerServer{
 		Driver:                      gceDriver,
 		CloudProvider:               cloudProvider,
@@ -163,6 +164,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, err
 		errorBackoff:                newCsiErrorBackoff(errorBackoffInitialDuration, errorBackoffMaxDuration),
 		fallbackRequisiteZones:      fallbackRequisiteZones,
 		enableStoragePools:          enableStoragePools,
+		enableDataCache:             enableDataCache,
 		multiZoneVolumeHandleConfig: multiZoneVolumeHandleConfig,
 		listVolumesConfig:           listVolumesConfig,
 	}
diff --git a/pkg/gce-pd-csi-driver/node.go b/pkg/gce-pd-csi-driver/node.go
index 9921a587..6e007d35 100644
--- a/pkg/gce-pd-csi-driver/node.go
+++ b/pkg/gce-pd-csi-driver/node.go
@@ -46,6 +46,7 @@ type GCENodeServer struct {
 	DeviceUtils     deviceutils.DeviceUtils
 	VolumeStatter   mountmanager.Statter
 	MetadataService metadataservice.MetadataService
+	EnableDataCache bool
 
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
@@ -261,7 +262,7 @@ func (ns *GCENodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeU
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
 	}
 	defer ns.volumeLocks.Release(volumeID)
-
+	klog.V(2).Infof("==============Target path %s=========", targetPath)
 	if err := cleanupPublishPath(targetPath, ns.Mounter); err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Unmount failed: %v\nUnmounting arguments: %s\n", err.Error(), targetPath))
 	}
@@ -274,6 +275,7 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 	volumeID := req.GetVolumeId()
 	stagingTargetPath := req.GetStagingTargetPath()
 	volumeCapability := req.GetVolumeCapability()
+	nodeId := ns.MetadataService.GetName()
 	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Volume ID must be provided")
 	}
@@ -283,7 +285,9 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 	if volumeCapability == nil {
 		return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Volume Capability must be provided")
 	}
-
+	if len(nodeId) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Node ID must be provided")
+	}
 	if acquired := ns.volumeLocks.TryAcquire(volumeID); !acquired {
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
 	}
@@ -307,16 +311,28 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		partition = part
 	}
 	devicePath, err := getDevicePath(ns, volumeID, partition)
-
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Error when getting device path: %v", err.Error()))
 	}
 
-	klog.V(4).Infof("Successfully found attached GCE PD %q at device path %s.", volumeKey.Name, devicePath)
+	klog.Infof("Successfully found attached GCE PD %q at device path %s.", volumeKey.Name, devicePath)
+
+	// LVM PoC Steps
+	klog.V(2).Infof("====== NodeStageVolume PublishContext is %v ======", req.GetPublishContext())
+	if ns.EnableDataCache && req.GetPublishContext()[common.ContexLocalSsdCacheSize] != "" {
+		devFsPath, err := filepath.EvalSymlinks(devicePath)
+		if err != nil {
+			klog.Errorf("filepath.EvalSymlinks(%q) failed when trying to create volume group: %v", devicePath, err)
+		}
+		devicePath, err = setupCaching(devFsPath, req, nodeId)
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Error setting up cache: %v", err.Error()))
+		}
+	}
 
 	// Part 2: Check if mount already exists at stagingTargetPath
 	if ns.isVolumePathMounted(stagingTargetPath) {
-		klog.V(4).Infof("NodeStageVolume succeeded on volume %v to %s, mount already exists.", volumeID, stagingTargetPath)
+		klog.V(2).Infof("NodeStageVolume succeeded on volume %v to %s, mount already exists.", volumeID, stagingTargetPath)
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
 
@@ -459,7 +475,17 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 			return nil, status.Errorf(codes.Internal, "NodeUnstageVolume for volume %s failed: %v", volumeID, err)
 		}
 	}
-
+	// The NodeUnstageVolume does not have any volume or publish context, we need to get the info from LVM locally
+	// Check if cache group cache-{volumeID} exist in LVM
+	if ns.EnableDataCache {
+		// LVM PoC Steps
+		klog.V(2).Infof("====== Start LVM PoC NodeUnstageVolume Steps ======")
+		nodeId := ns.MetadataService.GetName()
+		err := cleanupCache(volumeID, nodeId)
+		if err != nil {
+			klog.Errorf("Failed to cleanup cache: %v", err)
+		}
+	}
 	klog.V(4).Infof("NodeUnstageVolume succeeded on %v from %s", volumeID, stagingTargetPath)
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }
diff --git a/vendor/k8s.io/mount-utils/mount_linux.go b/vendor/k8s.io/mount-utils/mount_linux.go
index 07ce76de..88c14067 100644
--- a/vendor/k8s.io/mount-utils/mount_linux.go
+++ b/vendor/k8s.io/mount-utils/mount_linux.go
@@ -598,6 +598,7 @@ func (mounter *SafeFormatAndMount) formatAndMountSensitive(source string, target
 	}
 
 	// Mount the disk
+	// Source should be /dev/{CACHE_GROUP}/main for the LVM cache
 	klog.V(4).Infof("Attempting to mount disk %s in %s format at %s", source, fstype, target)
 	if err := mounter.MountSensitive(source, target, fstype, options, sensitiveOptions); err != nil {
 		return NewMountError(mountErrorValue, err.Error())


Test patch:
diff --git a/pkg/common/parameters_test.go b/pkg/common/parameters_test.go
index b43aeaed..96e7235c 100644
--- a/pkg/common/parameters_test.go
+++ b/pkg/common/parameters_test.go
@@ -25,13 +25,15 @@ import (
 
 func TestExtractAndDefaultParameters(t *testing.T) {
 	tests := []struct {
-		name               string
-		parameters         map[string]string
-		labels             map[string]string
-		enableStoragePools bool
-		extraTags          map[string]string
-		expectParams       DiskParameters
-		expectErr          bool
+		name                  string
+		parameters            map[string]string
+		labels                map[string]string
+		enableStoragePools    bool
+		enableDataCache       bool
+		extraTags             map[string]string
+		expectParams          DiskParameters
+		expectDataCacheParams DataCacheParameters
+		expectErr             bool
 	}{
 		{
 			name:       "defaults",
@@ -350,11 +352,60 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 			labels:             map[string]string{},
 			expectErr:          true,
 		},
+		{
+			name:            "data cache parameters - set default cache mode",
+			enableDataCache: true,
+			parameters:      map[string]string{ParameterKeyType: "pd-balanced", ParameterKeyReplicationType: "none", ParameterKeyDiskEncryptionKmsKey: "foo/key", ParameterKeyLabels: "key1=value1,key2=value2", ParameterKeyDataCacheSize: "1234Gi"},
+			labels:          map[string]string{},
+			expectParams: DiskParameters{
+				DiskType:             "pd-balanced",
+				ReplicationType:      "none",
+				DiskEncryptionKMSKey: "foo/key",
+				Tags:                 map[string]string{},
+				Labels: map[string]string{
+					"key1": "value1",
+					"key2": "value2",
+				},
+				ResourceTags: map[string]string{},
+			},
+			expectDataCacheParams: DataCacheParameters{
+				DataCacheMode: DataCacheModeWriteThrough,
+				DataCacheSize: "1234",
+			},
+		},
+		{
+			name:            "data cache parameters",
+			enableDataCache: true,
+			parameters:      map[string]string{ParameterKeyType: "pd-balanced", ParameterKeyReplicationType: "none", ParameterKeyDiskEncryptionKmsKey: "foo/key", ParameterKeyLabels: "key1=value1,key2=value2", ParameterKeyDataCacheSize: "1234Gi", ParameterKeyDataCacheMode: DataCacheModeWriteBack},
+			labels:          map[string]string{},
+			expectParams: DiskParameters{
+				DiskType:             "pd-balanced",
+				ReplicationType:      "none",
+				DiskEncryptionKMSKey: "foo/key",
+				Tags:                 map[string]string{},
+				Labels: map[string]string{
+					"key1": "value1",
+					"key2": "value2",
+				},
+				ResourceTags: map[string]string{},
+			},
+			expectDataCacheParams: DataCacheParameters{
+				DataCacheMode: DataCacheModeWriteBack,
+				DataCacheSize: "1234",
+			},
+		},
+		{
+			name:            "data cache parameters - enableDataCache is false",
+			enableDataCache: false,
+			parameters:      map[string]string{ParameterKeyType: "pd-balanced", ParameterKeyReplicationType: "none", ParameterKeyDiskEncryptionKmsKey: "foo/key", ParameterKeyLabels: "key1=value1,key2=value2", ParameterKeyDataCacheSize: "1234Gi", ParameterKeyDataCacheMode: DataCacheModeWriteBack},
+			labels:          map[string]string{},
+			expectErr:       true,
+		},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			p, err := ExtractAndDefaultParameters(tc.parameters, "testDriver", tc.labels, tc.enableStoragePools, tc.extraTags)
+			p, d, err := ExtractAndDefaultParameters(tc.parameters, "testDriver", tc.labels, tc.enableStoragePools, tc.enableDataCache, tc.extraTags)
 			if gotErr := err != nil; gotErr != tc.expectErr {
 				t.Fatalf("ExtractAndDefaultParameters(%+v) = %v; expectedErr: %v", tc.parameters, err, tc.expectErr)
 			}
@@ -365,6 +416,10 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 			if diff := cmp.Diff(p, tc.expectParams); diff != "" {
 				t.Errorf("ExtractAndDefaultParameters(%+v): -want, +got \n%s", tc.parameters, diff)
 			}
+
+			if diff := cmp.Diff(d, tc.expectDataCacheParams); diff != "" {
+				t.Errorf("ExtractAndDefaultParameters(%+v): -want, +got \n%s", tc.parameters, diff)
+			}
 		})
 	}
 }
diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go
index 2b610e8e..7d611a91 100644
--- a/pkg/common/utils_test.go
+++ b/pkg/common/utils_test.go
@@ -1042,6 +1042,108 @@ func TestConvertMiStringToInt64(t *testing.T) {
 	}
 }
 
+func TestConvertGiStringToInt64(t *testing.T) {
+	tests := []struct {
+		desc        string
+		inputStr    string
+		expInt64    int64
+		expectError bool
+	}{
+		{
+			desc:        "valid number string",
+			inputStr:    "10000",
+			expInt64:    1,
+			expectError: false,
+		},
+		{
+			desc:        "round Ki to GiB",
+			inputStr:    "1000000Ki",
+			expInt64:    1,
+			expectError: false,
+		},
+		{
+			desc:        "round k to GiB",
+			inputStr:    "1000000k",
+			expInt64:    1,
+			expectError: false,
+		},
+		{
+			desc:        "round Mi to GiB",
+			inputStr:    "1000Mi",
+			expInt64:    1,
+			expectError: false,
+		},
+		{
+			desc:        "round M to GiB",
+			inputStr:    "1000M",
+			expInt64:    1,
+			expectError: false,
+		},
+		{
+			desc:        "round G to GiB",
+			inputStr:    "1000G",
+			expInt64:    932,
+			expectError: false,
+		},
+		{
+			desc:        "round Gi to GiB - most common case",
+			inputStr:    "1234Gi",
+			expInt64:    1234,
+			expectError: false,
+		},
+		{
+			desc:        "round decimal to GiB",
+			inputStr:    "1.2Gi",
+			expInt64:    2,
+			expectError: false,
+		},
+		{
+			desc:        "round big value to GiB",
+			inputStr:    "8191Pi",
+			expInt64:    8588886016,
+			expectError: false,
+		},
+		{
+			desc:        "invalid empty string",
+			inputStr:    "",
+			expInt64:    0,
+			expectError: true,
+		},
+		{
+			desc:        "invalid KiB string",
+			inputStr:    "10KiB",
+			expInt64:    10000,
+			expectError: true,
+		},
+		{
+			desc:        "invalid GB string",
+			inputStr:    "10GB",
+			expInt64:    0,
+			expectError: true,
+		},
+		{
+			desc:        "invalid string",
+			inputStr:    "ew%65",
+			expInt64:    0,
+			expectError: true,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.desc, func(t *testing.T) {
+			actualInt64, err := ConvertGiStringToInt64(tc.inputStr)
+			if err != nil && !tc.expectError {
+				t.Errorf("Got error %v converting string to int64 %s; expect no error", err, tc.inputStr)
+			}
+			if err == nil && tc.expectError {
+				t.Errorf("Got no error converting string to int64 %s; expect an error", tc.inputStr)
+			}
+			if err == nil && actualInt64 != tc.expInt64 {
+				t.Errorf("Got %d for converting string to int64; expect %d", actualInt64, tc.expInt64)
+			}
+		})
+	}
+}
+
 func TestConvertStringToBool(t *testing.T) {
 	tests := []struct {
 		desc        string
@@ -1595,3 +1697,67 @@ func TestUnorderedSlicesEqual(t *testing.T) {
 		})
 	}
 }
+
+func TestStringInSlice(t *testing.T) {
+	testCases := []struct {
+		name            string
+		inputStr        string
+		inputSlice      []string
+		expectedInSlice bool
+	}{
+		{
+			name:            "string is in the slice",
+			inputStr:        "in slice",
+			inputSlice:      []string{"in slice", "other string"},
+			expectedInSlice: true,
+		},
+		{
+			name:       "string is NOT in the slice",
+			inputStr:   "not in slice",
+			inputSlice: []string{"other string"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		actualResult := StringInSlice(tc.inputStr, tc.inputSlice)
+		if actualResult != tc.expectedInSlice {
+			t.Errorf("Expect value is %v but got  %v. inputStr is %s, inputSlice is %v", tc.expectedInSlice, actualResult, tc.inputStr, tc.inputSlice)
+		}
+	}
+}
+
+func TestValidateDataCacheMode(t *testing.T) {
+	testCases := []struct {
+		name        string
+		inputStr    string
+		expectError bool
+	}{
+		{
+			name:     "valid input - writethrough",
+			inputStr: "writethrough",
+		},
+		{
+			name:     "valid input - writeback",
+			inputStr: "writeback",
+		},
+		{
+			name:        "invalid input",
+			inputStr:    "write-back not valid",
+			expectError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		err := ValidateDataCacheMode(tc.inputStr)
+		if err != nil && !tc.expectError {
+			t.Errorf("Got error %v  validate data cache mode %s; expect no error", err, tc.inputStr)
+		}
+
+		if err == nil && tc.expectError {
+			t.Errorf("Got no error validate data cache mode %s; expect an error", tc.inputStr)
+		}
+	}
+
+}
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
index 15306c14..73be32e2 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
@@ -48,10 +48,11 @@ func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute)
 	errorBackoffMaxDuration := 5 * time.Minute
 	fallbackRequisiteZones := []string{}
 	enableStoragePools := false
+	enableDataCache := false
 	multiZoneVolumeHandleConfig := MultiZoneVolumeHandleConfig{}
 	listVolumesConfig := ListVolumesConfig{}
 
-	controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig)
+	controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, enableDataCache, multiZoneVolumeHandleConfig, listVolumesConfig)
 	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, nil, controllerServer, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
diff --git a/pkg/gce-pd-csi-driver/node_test.go b/pkg/gce-pd-csi-driver/node_test.go
index 340412b6..66c96b8a 100644
--- a/pkg/gce-pd-csi-driver/node_test.go
+++ b/pkg/gce-pd-csi-driver/node_test.go
@@ -51,7 +51,8 @@ func getTestGCEDriverWithCustomMounter(t *testing.T, mounter *mount.SafeFormatAn
 
 func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, metaService metadataservice.MetadataService) *GCEDriver {
 	gceDriver := GetGCEDriver()
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter))
+	enableDataCache := false
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter), enableDataCache)
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
@@ -61,8 +62,9 @@ func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, dev
 
 func getTestBlockingMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
+	enableDataCache := false
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter))
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), enableDataCache)
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
@@ -72,8 +74,9 @@ func getTestBlockingMountGCEDriver(t *testing.T, readyToExecute chan chan struct
 
 func getTestBlockingFormatAndMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
+	enableDataCache := false
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter)).WithSerializedFormatAndMount(5*time.Second, 1)
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), enableDataCache).WithSerializedFormatAndMount(5*time.Second, 1)
 
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
diff --git a/test/e2e/tests/multi_zone_e2e_test.go b/test/e2e/tests/multi_zone_e2e_test.go
index 0e3eb48b..ffb15503 100644
--- a/test/e2e/tests/multi_zone_e2e_test.go
+++ b/test/e2e/tests/multi_zone_e2e_test.go
@@ -238,7 +238,7 @@ var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 			if i >= 1 {
 				readOnly = true
 			}
-			err = testAttachWriteReadDetach(volume.VolumeId, volName, testContext.Instance, testContext.Client, readOnly)
+			err = testAttachWriteReadDetach(volume.VolumeId, volName, testContext.Instance, testContext.Client, readOnly, false /* detachAndReattach */, false /* setupDataCache */)
 			Expect(err).To(BeNil(), "failed volume lifecycle checks")
 			i = i + 1
 		}
@@ -325,7 +325,7 @@ var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 
 		// Attach disk to instance in the first zone.
 		tc0 := zoneToContext[zones[0]]
-		err, detacher, args := testAttachAndMount(volume.VolumeId, volName, tc0.Instance, tc0.Client, false /* useBlock */, false /* forceAttach */)
+		err, detacher, args := testAttachAndMount(volume.VolumeId, volName, tc0.Instance, tc0.Client, false /* useBlock */, false /* forceAttach */, false /* setupDataCache */)
 		detachers = append(detachers, detacher)
 		Expect(err).To(BeNil(), "failed attach in zone 0")
 		testFileName := filepath.Join(args.publishDir, "force-attach-test")
@@ -341,7 +341,7 @@ var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 
 		// Now force attach to the second instance without detaching.
 		tc1 := zoneToContext[zones[1]]
-		err, detacher, args = testAttachAndMount(volume.VolumeId, volName, tc1.Instance, tc1.Client, false /* useBlock */, true /* forceAttach */)
+		err, detacher, args = testAttachAndMount(volume.VolumeId, volName, tc1.Instance, tc1.Client, false /* useBlock */, true /* forceAttach */, false /* setupDataCache */)
 		detachers = append(detachers, detacher)
 		Expect(err).To(BeNil(), "failed force attach in zone 1")
 		readContents, err = testutils.ReadFile(tc1.Instance, testFileName)
@@ -360,7 +360,12 @@ func deleteDisk(controllerClient *remote.CsiClient, p, zone, volID, volName stri
 	Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
 }
 
-func testAttachWriteReadDetach(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, readOnly bool) error {
+func testAttachWriteReadDetach(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, readOnly bool, detachAndReattach bool, setupDataCache bool) error {
+	writeFile, verifyReadFile := testWriteAndReadFile(instance, readOnly)
+	return testLifecycleWithVerify(volID, volName, instance, client, readOnly, false /* fs */, writeFile, verifyReadFile, detachAndReattach, setupDataCache)
+}
+
+func testWriteAndReadFile(instance *remote.InstanceInfo, readOnly bool) (verifyFunc, verifyFunc) {
 	var testFileContents = "test"
 	writeFile := func(a *verifyArgs) error {
 		if !readOnly {
@@ -386,35 +391,44 @@ func testAttachWriteReadDetach(volID string, volName string, instance *remote.In
 		}
 		return nil
 	}
-	return testLifecycleWithVerify(volID, volName, instance, client, readOnly, false /* fs */, writeFile, verifyReadFile)
+	return writeFile, verifyReadFile
 }
 
-func testAttachAndMount(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, useBlock, forceAttach bool) (error, func(), *verifyArgs) {
-	// Attach Disk
-	err := client.ControllerPublishVolumeReadWrite(volID, instance.GetNodeID(), forceAttach)
+func testAttachAndMount(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, useBlock, forceAttach bool, setupDataCache bool) (error, func(), *verifyArgs) {
+	klog.Infof("Starting testAttachAndMount with volume %v node %v \n", volID, instance.GetNodeID())
+	err, unstageAndDetach, stageDir := testAttach(volID, volName, instance, client, useBlock, forceAttach, setupDataCache)
 	if err != nil {
-		return fmt.Errorf("ControllerPublishVolume failed with error for disk %v on node %v: %v", volID, instance.GetNodeID(), err.Error()), nil, nil
+		return err, nil, nil
 	}
+	// Mount Disk
+	err, args := testMount(volID, volName, instance, client, useBlock, stageDir)
+	if err != nil {
+		unstageAndDetach()
+		return err, nil, nil
+	}
+	return nil, unstageAndDetach, args
+}
 
-	detach := func() {
-		// Detach Disk
-		err = client.ControllerUnpublishVolume(volID, instance.GetNodeID())
-		if err != nil {
-			klog.Errorf("Failed to detach disk: %v", err)
-		}
+func testAttach(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, useBlock, forceAttach bool, setupDataCache bool) (error, func(), string) {
+	klog.Infof("Starting testAttach with volume %v node %v \n", volID, instance.GetNodeID())
+	// Attach Disk
+	var stageDir string
+	err := client.ControllerPublishVolumeReadWrite(volID, instance.GetNodeID(), forceAttach)
+	if err != nil {
+		return fmt.Errorf("ControllerPublishVolume failed with error for disk %v on node %v: %v", volID, instance.GetNodeID(), err.Error()), nil, stageDir
 	}
 
 	// Stage Disk
-	stageDir := filepath.Join("/tmp/", volName, "stage")
+	stageDir = filepath.Join("/tmp/", volName, "stage")
 	if useBlock {
-		err = client.NodeStageBlockVolume(volID, stageDir)
+		err = client.NodeStageBlockVolume(volID, stageDir, setupDataCache)
 	} else {
-		err = client.NodeStageExt4Volume(volID, stageDir)
+		err = client.NodeStageExt4Volume(volID, stageDir, setupDataCache)
 	}
 
 	if err != nil {
-		detach()
-		return fmt.Errorf("NodeStageExt4Volume failed with error: %w", err), nil, nil
+		_ = detach(volID, instance, client)
+		return fmt.Errorf("NodeStageExt4Volume failed with error: %w for node: %v", err, instance.GetNodeID()), nil, stageDir
 	}
 
 	unstageAndDetach := func() {
@@ -429,9 +443,23 @@ func testAttachAndMount(volID string, volName string, instance *remote.InstanceI
 			klog.Errorf("Failed to rm file path %s: %v", fp, err)
 		}
 
-		detach()
+		detach(volID, instance, client)
 	}
+	return nil, unstageAndDetach, stageDir
+}
 
+func detach(volID string, instance *remote.InstanceInfo, client *remote.CsiClient) error {
+	// Detach Disk
+	err := client.ControllerUnpublishVolume(volID, instance.GetNodeID())
+	if err != nil {
+		klog.Errorf("Failed to detach disk  %v", err)
+		return fmt.Errorf("Failed to detach disk: %v", err)
+	}
+	return nil
+}
+
+func testMount(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, useBlock bool, stageDir string) (error, *verifyArgs) {
+	var err error
 	// Mount Disk
 	publishDir := filepath.Join("/tmp/", volName, "mount")
 
@@ -442,26 +470,22 @@ func testAttachAndMount(volID string, volName string, instance *remote.InstanceI
 	}
 
 	if err != nil {
-		unstageAndDetach()
-		return fmt.Errorf("NodePublishVolume failed with error: %v", err.Error()), nil, nil
+		return fmt.Errorf("NodePublishVolume failed with error: %v", err.Error()), nil
 	}
 	err = testutils.ForceChmod(instance, filepath.Join("/tmp/", volName), "777")
 	if err != nil {
-		unstageAndDetach()
-		return fmt.Errorf("Chmod failed with error: %v", err.Error()), nil, nil
+		return fmt.Errorf("Chmod failed with error: %v", err.Error()), nil
 	}
 
-	args := &verifyArgs{
+	return nil, &verifyArgs{
 		publishDir: publishDir,
 		stageDir:   stageDir,
 	}
-
-	return nil, unstageAndDetach, args
 }
 
-func testLifecycleWithVerify(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, readOnly, useBlock bool, firstMountVerify, secondMountVerify verifyFunc) error {
+func testLifecycleWithVerify(volID string, volName string, instance *remote.InstanceInfo, client *remote.CsiClient, readOnly, useBlock bool, firstMountVerify, secondMountVerify verifyFunc, detachAndReattach bool, setupDataCache bool) error {
 	klog.Infof("Starting testAttachWriteReadDetach with volume %v node %v with readonly %v\n", volID, instance.GetNodeID(), readOnly)
-	err, detacher, args := testAttachAndMount(volID, volName, instance, client, useBlock, false /* forceAttach */)
+	err, detacher, args := testAttachAndMount(volID, volName, instance, client, useBlock, false /* forceAttach */, setupDataCache)
 	if err != nil {
 		return fmt.Errorf("failed to attach and mount: %w", err)
 	}
@@ -478,13 +502,28 @@ func testLifecycleWithVerify(volID string, volName string, instance *remote.Inst
 		return fmt.Errorf("NodeUnpublishVolume failed with error: %v", err.Error())
 	}
 
+	stageDir := args.stageDir
+	if detachAndReattach {
+		// Unstage and detach
+		err = client.NodeUnstageVolume(volID, stageDir)
+		if err != nil {
+			klog.Errorf("Failed to unstage volume: %v", err)
+		}
+		detach(volID, instance, client)
+		// Reattach the volume
+		err, _, stageDir = testAttach(volID, volName, instance, client, useBlock, false /* forceAttach */, setupDataCache)
+		if err != nil {
+			return err
+		}
+	}
+
 	if secondMountVerify != nil {
 		// Mount disk somewhere else
 		secondPublishDir := filepath.Join("/tmp/", volName, "secondmount")
 		if useBlock {
-			err = client.NodePublishBlockVolume(volID, args.stageDir, secondPublishDir)
+			err = client.NodePublishBlockVolume(volID, stageDir, secondPublishDir)
 		} else {
-			err = client.NodePublishVolume(volID, args.stageDir, secondPublishDir)
+			err = client.NodePublishVolume(volID, stageDir, secondPublishDir)
 		}
 		if err != nil {
 			return fmt.Errorf("NodePublishVolume failed with error: %v", err.Error())
diff --git a/test/e2e/tests/resize_e2e_test.go b/test/e2e/tests/resize_e2e_test.go
index fbe8af86..fc4d8755 100644
--- a/test/e2e/tests/resize_e2e_test.go
+++ b/test/e2e/tests/resize_e2e_test.go
@@ -80,7 +80,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		// Stage Disk
 		stageDir := filepath.Join("/tmp/", volName, "stage")
-		err = client.NodeStageExt4Volume(volume.VolumeId, stageDir)
+		err = client.NodeStageExt4Volume(volume.VolumeId, stageDir, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Node Stage volume failed")
 
 		defer func() {
@@ -174,7 +174,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}()
 
 		// Volume should be attached/formatted/mounted/unmounted/detached
-		err = testAttachWriteReadDetach(volume.VolumeId, volName, instance, client, false /* readOnly */)
+		err = testAttachWriteReadDetach(volume.VolumeId, volName, instance, client, false /* readOnly */, false /* detachAndReattach */, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 
 		// Resize controller
@@ -203,7 +203,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		// Stage Disk
 		stageDir := filepath.Join("/tmp/", volName, "stage")
-		err = client.NodeStageExt4Volume(volume.VolumeId, stageDir)
+		err = client.NodeStageExt4Volume(volume.VolumeId, stageDir, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Node Stage volume failed")
 
 		defer func() {
@@ -295,7 +295,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		// Stage Disk
 		stageDir := filepath.Join("/tmp/", volName, "stage")
-		err = client.NodeStageBlockVolume(volume.VolumeId, stageDir)
+		err = client.NodeStageBlockVolume(volume.VolumeId, stageDir, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Node Stage volume failed")
 
 		defer func() {
diff --git a/test/e2e/tests/setup_e2e_test.go b/test/e2e/tests/setup_e2e_test.go
index 5eda4b2f..a4fcc490 100644
--- a/test/e2e/tests/setup_e2e_test.go
+++ b/test/e2e/tests/setup_e2e_test.go
@@ -19,6 +19,7 @@ import (
 	"flag"
 	"fmt"
 	"math/rand"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -52,6 +53,8 @@ var (
 	kmsClient           *cloudkms.KeyManagementClient
 )
 
+const localSSDCount = 2
+
 func init() {
 	klog.InitFlags(flag.CommandLine)
 }
@@ -68,6 +71,7 @@ var _ = BeforeSuite(func() {
 	defer close(tcc)
 
 	zones := strings.Split(*zones, ",")
+	// Create 2 instances for each zone as we need 2 instances each zone for certain test cases
 
 	rand.Seed(time.Now().UnixNano())
 
@@ -93,14 +97,21 @@ var _ = BeforeSuite(func() {
 
 	klog.Infof("Running in project %v with service account %v", *project, *serviceAccount)
 
-	for _, zone := range zones {
-		go func(curZone string) {
-			defer GinkgoRecover()
-			tcc <- NewTestContext(curZone)
-		}(zone)
+	numberOfInstancesPerZone := 2
+
+	setupContext := func(zones []string, randInt int) {
+		for _, zone := range zones {
+			go func(curZone string) {
+				defer GinkgoRecover()
+				tcc <- NewTestContext(curZone, strconv.Itoa(randInt))
+			}(zone)
+		}
+	}
+	for j := 0; j < numberOfInstancesPerZone; j++ {
+		setupContext(zones, j)
 	}
 
-	for i := 0; i < len(zones); i++ {
+	for i := 0; i < len(zones)*numberOfInstancesPerZone; i++ {
 		tc := <-tcc
 		testContexts = append(testContexts, tc)
 		klog.Infof("Added TestContext for node %s", tc.Instance.GetName())
@@ -127,11 +138,10 @@ func getRemoteInstanceConfig() *remote.InstanceConfig {
 		CloudtopHost:   *cloudtopHost}
 }
 
-func NewTestContext(zone string) *remote.TestContext {
-	nodeID := fmt.Sprintf("gce-pd-csi-e2e-%s", zone)
+func NewTestContext(zone string, instanceNumber string) *remote.TestContext {
+	nodeID := fmt.Sprintf("gce-pd-csi-e2e-%s-%v", zone, instanceNumber)
 	klog.Infof("Setting up node %s", nodeID)
-
-	i, err := remote.SetupInstance(getRemoteInstanceConfig(), zone, nodeID, computeService)
+	i, err := remote.SetupInstance(getRemoteInstanceConfig(), zone, nodeID, computeService, localSSDCount)
 	if err != nil {
 		klog.Fatalf("Failed to setup instance %v: %v", nodeID, err)
 	}
@@ -150,7 +160,16 @@ func NewTestContext(zone string) *remote.TestContext {
 	if err != nil {
 		klog.Fatalf("Failed to copy google_nvme_id to containerized directory: %v", err)
 	}
+	pkgs := []string{"lvm2", "mdadm", "grep", "coreutils"}
+	err = testutils.InstallDependencies(i, pkgs)
+	if err != nil {
+		klog.Errorf("Failed to install dependency package on node %v: error : %v", i.GetNodeID(), err)
+	}
 
+	err = testutils.SetupDataCachingConfig(i)
+	if err != nil {
+		klog.Errorf("Failed to setup data cache required config error %v", err)
+	}
 	klog.Infof("Creating new driver and client for node %s", i.GetName())
 	tc, err := testutils.GCEClientAndDriverSetup(i, "")
 	if err != nil {
diff --git a/test/e2e/tests/single_zone_e2e_test.go b/test/e2e/tests/single_zone_e2e_test.go
index 6b3cfea7..46cee539 100644
--- a/test/e2e/tests/single_zone_e2e_test.go
+++ b/test/e2e/tests/single_zone_e2e_test.go
@@ -94,7 +94,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}()
 
 		// Attach Disk
-		err := testAttachWriteReadDetach(volID, volName, instance, client, false /* readOnly */)
+		err := testAttachWriteReadDetach(volID, volName, instance, client, false /* readOnly */, false /* detachAndReattach */, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 	})
 
@@ -142,7 +142,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		// Stage Disk
 		stageDir := filepath.Join("/tmp/", volName, "stage")
-		err = client.NodeStageExt4Volume(volID, stageDir)
+		err = client.NodeStageExt4Volume(volID, stageDir, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "failed to repair /dev/by-id symlink and stage volume")
 
 		// Validate that the link is correct
@@ -212,7 +212,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		// Stage Disk
 		stageDir := filepath.Join("/tmp/", volName, "stage")
-		err = client.NodeStageExt4Volume(volID, stageDir)
+		err = client.NodeStageExt4Volume(volID, stageDir, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "failed to repair /dev/by-id symlink and stage volume")
 
 		// Validate that the link is correct
@@ -292,7 +292,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			}()
 
 			// Attach Disk
-			err := testAttachWriteReadDetach(underSpecifiedID, volName, instance, client, false /* readOnly */)
+			err := testAttachWriteReadDetach(underSpecifiedID, volName, instance, client, false /* readOnly */, false /* detachAndReattach */, false /* setupDataCache*/)
 			Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 		},
 		Entry("on pd-standard", standardDiskType),
@@ -636,7 +636,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			}()
 
 			// Test disk works
-			err = testAttachWriteReadDetach(volume.VolumeId, volName, controllerInstance, controllerClient, false /* readOnly */)
+			err = testAttachWriteReadDetach(volume.VolumeId, volName, controllerInstance, controllerClient, false /* readOnly */, false /* detachAndReattach */, false /* setupDataCache */)
 			Expect(err).To(BeNil(), "Failed to go through volume lifecycle before revoking CMEK key")
 
 			// Revoke CMEK key
@@ -657,7 +657,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			}
 
 			// Make sure attach of PD fails
-			err = testAttachWriteReadDetach(volume.VolumeId, volName, controllerInstance, controllerClient, false /* readOnly */)
+			err = testAttachWriteReadDetach(volume.VolumeId, volName, controllerInstance, controllerClient, false /* readOnly */, false /* detachAndReattach */, false /* setupDataCache */)
 			Expect(err).ToNot(BeNil(), "Volume lifecycle should have failed, but succeeded")
 
 			// Restore CMEK key
@@ -678,7 +678,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			// The controller publish failure in above step would set a backoff condition on the node. Wait suffcient amount of time for the driver to accept new controller publish requests.
 			time.Sleep(time.Second)
 			// Make sure attach of PD succeeds
-			err = testAttachWriteReadDetach(volume.VolumeId, volName, controllerInstance, controllerClient, false /* readOnly */)
+			err = testAttachWriteReadDetach(volume.VolumeId, volName, controllerInstance, controllerClient, false /* readOnly */, false /* detachAndReattach */, false /* setupDataCache */)
 			Expect(err).To(BeNil(), "Failed to go through volume lifecycle after restoring CMEK key")
 		},
 		Entry("on pd-standard", standardDiskType),
@@ -821,7 +821,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}
 
 		// Attach Disk
-		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, true /* block */, verifyVolumeStats, nil)
+		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, true /* block */, verifyVolumeStats, nil, false /* detachAndReattach */, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 	})
 
@@ -858,7 +858,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}
 
 		// Attach Disk
-		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, false /* fs */, verifyVolumeStats, nil)
+		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, false /* fs */, verifyVolumeStats, nil, false /* detachAndReattach */, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 	})
 
@@ -927,7 +927,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			}
 			return nil
 		}
-		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, true /* block */, writeFunc, verifyReadFunc)
+		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, true /* block */, writeFunc, verifyReadFunc, false /* detachAndReattach */, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 	})
 
@@ -1289,7 +1289,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		zone := "us-central1-c"
 		nodeID := fmt.Sprintf("gce-pd-csi-e2e-%s", zone)
-		i, err := remote.SetupInstance(getRemoteInstanceConfig(), zone, nodeID, computeService)
+		i, err := remote.SetupInstance(getRemoteInstanceConfig(), zone, nodeID, computeService, 0 /* localSSDCount */)
 
 		if err != nil {
 			klog.Fatalf("Failed to setup instance %v: %v", nodeID, err)
@@ -1363,7 +1363,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 				Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 			},
 		}
-		err = client.NodeStageVolume(volID, stageDir, volCap)
+		err = client.NodeStageVolume(volID, stageDir, volCap, false /* setupDataCache */)
 		Expect(err).To(BeNil(), "failed to stage volume: %v", err)
 
 		// Validate that the link is correct
@@ -1400,6 +1400,106 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			}
 		}()
 	})
+	It("Should create disks, attach them to instance with local ssd, setup caching between LSSD->detach->reattach to same instance", func() {
+		Expect(testContexts).ToNot(BeEmpty())
+		testContext := getRandomTestContext()
+
+		p, z, _ := testContext.Instance.GetIdentity()
+		client := testContext.Client
+		instance := testContext.Instance
+		volName, volID := createAndValidateUniqueZonalDisk(client, p, z, standardDiskType)
+		defer deleteVolumeOrError(client, volID)
+
+		// Attach Disk
+		err := testAttachWriteReadDetach(volID, volName, instance, client, false /* readOnly */, true /* detachAndReattach */, true /* setupDataCache */)
+		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
+
+	})
+	It("Should create->attach->setup caching->write->detach->attach to different node->mount->read", func() {
+		Expect(testContexts).ToNot(BeEmpty())
+		zoneToContext := map[string][]*remote.TestContext{}
+		testZoneContexts := []*remote.TestContext{}
+		for _, tc := range testContexts {
+			_, z, _ := tc.Instance.GetIdentity()
+			// Zone hasn't been seen before
+			if _, ok := zoneToContext[z]; !ok {
+				zoneToContext[z] = []*remote.TestContext{tc}
+			} else {
+				zoneToContext[z] = append(zoneToContext[z], tc)
+			}
+			if len(zoneToContext[z]) >= 2 {
+				testZoneContexts = zoneToContext[z]
+				break
+			}
+		}
+		if len(testZoneContexts) < 2 {
+			klog.Fatalf("No test contexts setup %v", testZoneContexts)
+		}
+		testContextForVm1 := testZoneContexts[0]
+		p, z, _ := testContextForVm1.Instance.GetIdentity()
+
+		client := testContextForVm1.Client
+		firstInstance := testContextForVm1.Instance
+
+		volName, volID := createAndValidateUniqueZonalDisk(client, p, z, standardDiskType)
+		defer deleteVolumeOrError(client, volID)
+
+		testContextForVm2 := testZoneContexts[1]
+		secondClient := testContextForVm2.Client
+		secondInstance := testContextForVm2.Instance
+		unmountDisk := func(client *remote.CsiClient, volID string, args *verifyArgs) {
+			err := client.NodeUnpublishVolume(volID, args.publishDir)
+			if err != nil {
+				klog.Errorf("NodeUnpublishVolume failed with error: %v", err)
+			}
+		}
+		// Controller Publish (Attach) - Node Stage - Node Publish(Mount) Volume
+		err, _, args := testAttachAndMount(volID, volName, firstInstance, client, false /* useBlock */, false /* forceAttach */, true)
+		if err != nil {
+			klog.Errorf("Failed to attach and mount: %v", err.Error())
+		}
+		// Write file in the volume
+		firstMountVerify, _ := testWriteAndReadFile(firstInstance, false /* readOnly */)
+		err = firstMountVerify(args)
+		if err != nil {
+			klog.Errorf("failed to verify after first mount to %s: %v", args, err)
+		}
+		Expect(err).To(BeNil(), "Failed to write data to volume %s on instance %s", volName, firstInstance.GetName())
+		// Unmount Disk
+		unmountDisk(client, volID, args)
+
+		// Node Unstage
+		err = client.NodeUnstageVolume(volID, args.stageDir)
+		if err != nil {
+			klog.Errorf("Failed to unstage volume: %v", err)
+		}
+		detach(volID, firstInstance, client)
+
+		// Attach Disk to secondInstance
+		err, detacher, stageDir := testAttach(volID, volName, secondInstance, secondClient, false /* useBlock */, false /* forceAttach */, true /* setupDataCache */)
+		if err != nil {
+			klog.Errorf("Failed to attach disk %v", err)
+		}
+		defer func() {
+			detacher()
+			deleteVolumeOrError(secondClient, volID)
+		}()
+
+		// Mount disk
+		err, args = testMount(volID, volName, secondInstance, secondClient, false /* useBlock */, stageDir)
+		if err != nil {
+			klog.Fatalf("Failed to mount disk %v", err)
+		}
+		_, secondMountRead := testWriteAndReadFile(secondInstance, false /* readOnly */)
+		err = secondMountRead(args)
+		if err != nil {
+			klog.Errorf("failed to verify after second mount to %s: %v", args, err)
+		}
+		// Unmount disk for cleanup
+		unmountDisk(secondClient, volID, args)
+		Expect(err).To(BeNil(), "Failed to read data from volume %s on instance %s", volName, secondInstance.GetName())
+
+	})
 
 	It("Should block unstage if filesystem mounted", func() {
 		testContext := getRandomTestContext()
@@ -1435,7 +1535,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 
 		// Stage Disk
 		stageDir := filepath.Join("/tmp/", volName, "stage")
-		err = client.NodeStageExt4Volume(volID, stageDir)
+		err = client.NodeStageExt4Volume(volID, stageDir, false)
 		Expect(err).To(BeNil(), "failed to stage volume: %v", err)
 
 		// Create private bind mount
diff --git a/test/e2e/utils/utils.go b/test/e2e/utils/utils.go
index a044b689..94c18b3e 100644
--- a/test/e2e/utils/utils.go
+++ b/test/e2e/utils/utils.go
@@ -60,6 +60,7 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, computeEndpoint stri
 		"--multi-zone-volume-handle-disk-types=pd-standard",
 		"--use-instance-api-to-poll-attachment-disk-types=pd-ssd",
 		"--use-instance-api-to-list-volumes-published-nodes",
+		"--enable-data-cache",
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", computeEndpoint))
 
@@ -260,6 +261,32 @@ func CopyFile(instance *remote.InstanceInfo, src, dest string) error {
 	return nil
 }
 
+func InstallDependencies(instance *remote.InstanceInfo, pkgs []string) error {
+	for _, pkg := range pkgs {
+		output, err := instance.SSH("apt-get", "install", "-y", pkg)
+		if err != nil {
+			return fmt.Errorf("failed to install package %s. Output: %v, errror: %v", pkg, output, err.Error())
+		}
+	}
+	return nil
+}
+
+func SetupDataCachingConfig(instance *remote.InstanceInfo) error {
+	output, err := instance.SSH("/bin/sed", "-i", "-e", "\"s/.*allow_mixed_block_sizes = 0.*/	allow_mixed_block_sizes = 1/\"", "/etc/lvm/lvm.conf")
+	if err != nil {
+		return fmt.Errorf("failed to update field allow_mixed_block_sizes, error:%v; output: %v", err, output)
+	}
+	output, err = instance.SSH("/bin/sed", "-i", "-e", "\"s/.*udev_sync = 1.*/ udev_sync = 0/\"", "/etc/lvm/lvm.conf")
+	if err != nil {
+		return fmt.Errorf("failed to update field udev_sync, error:%v; output: %v", err, output)
+	}
+	output, err = instance.SSH("/bin/sed", "-i", "-e", "\"s/.*udev_rules = 1.*/ udev_rules = 0/\"", "/etc/lvm/lvm.conf")
+	if err != nil {
+		return fmt.Errorf("failed to update field udev_rules, error:%v; output: %v", err, output)
+	}
+	return nil
+}
+
 // ValidateLogicalLinkIsDisk takes a symlink location at "link" and finds the
 // link location - it then finds the backing PD using either scsi_id or
 // google_nvme_id (depending on the /dev path) and validates that it is the
diff --git a/test/k8s-integration/config/data-cache-sc.yaml b/test/k8s-integration/config/data-cache-sc.yaml
new file mode 100644
index 00000000..7e911eb3
--- /dev/null
+++ b/test/k8s-integration/config/data-cache-sc.yaml
@@ -0,0 +1,11 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: csi-gcepd-balanced
+provisioner: pd.csi.storage.gke.io
+parameters:
+  type: pd-balanced
+  data-cache-mode: writeback
+  data-cache-size: "50Gi"
+volumeBindingMode: WaitForFirstConsumer
+allowVolumeExpansion: true
\ No newline at end of file
diff --git a/test/remote/client-wrappers.go b/test/remote/client-wrappers.go
index 671cbb1d..efb0035f 100644
--- a/test/remote/client-wrappers.go
+++ b/test/remote/client-wrappers.go
@@ -52,6 +52,19 @@ var (
 	}
 )
 
+const (
+	// Keys in the volume context.
+	contextForceAttach   = "force-attach"
+	contextDataCacheSize = "data-cache-size"
+	contextDataCacheMode = "data-cache-mode"
+
+	// Keys in the publish context
+	contexLocalSsdCacheSize = "local-ssd-cache-size"
+
+	defaultLocalSsdCacheSize = "200Gi"
+	defaultDataCacheMode     = common.DataCacheModeWriteThrough
+)
+
 type CsiClient struct {
 	conn       *grpc.ClientConn
 	idClient   csipb.IdentityClient
@@ -179,19 +192,25 @@ func (c *CsiClient) ControllerUnpublishVolume(volId, nodeId string) error {
 	return err
 }
 
-func (c *CsiClient) NodeStageExt4Volume(volId, stageDir string) error {
-	return c.NodeStageVolume(volId, stageDir, stdVolCap)
+func (c *CsiClient) NodeStageExt4Volume(volId, stageDir string, setupDataCache bool) error {
+	return c.NodeStageVolume(volId, stageDir, stdVolCap, setupDataCache)
 }
 
-func (c *CsiClient) NodeStageBlockVolume(volId, stageDir string) error {
-	return c.NodeStageVolume(volId, stageDir, blockVolCap)
+func (c *CsiClient) NodeStageBlockVolume(volId, stageDir string, setupDataCache bool) error {
+	return c.NodeStageVolume(volId, stageDir, blockVolCap, setupDataCache)
 }
 
-func (c *CsiClient) NodeStageVolume(volId, stageDir string, volumeCap *csipb.VolumeCapability) error {
+func (c *CsiClient) NodeStageVolume(volId string, stageDir string, volumeCap *csipb.VolumeCapability, setupDataCache bool) error {
+	publishContext := map[string]string{}
+	if setupDataCache {
+		publishContext[contexLocalSsdCacheSize] = defaultLocalSsdCacheSize
+		publishContext[contextDataCacheMode] = defaultDataCacheMode
+	}
 	nodeStageReq := &csipb.NodeStageVolumeRequest{
 		VolumeId:          volId,
 		StagingTargetPath: stageDir,
 		VolumeCapability:  volumeCap,
+		PublishContext:    publishContext,
 	}
 	_, err := c.nodeClient.NodeStageVolume(context.Background(), nodeStageReq)
 	return err
diff --git a/test/remote/instance.go b/test/remote/instance.go
index 6fb1b523..f1666ac8 100644
--- a/test/remote/instance.go
+++ b/test/remote/instance.go
@@ -98,7 +98,7 @@ func CreateInstanceInfo(config *InstanceConfig, zone, name string, cs *compute.S
 }
 
 // Provision a gce instance using image
-func (i *InstanceInfo) CreateOrGetInstance() error {
+func (i *InstanceInfo) CreateOrGetInstance(localSSDCount int) error {
 	var err error
 	var instance *compute.Instance
 	klog.V(4).Infof("Creating instance: %v", i.name)
@@ -134,7 +134,20 @@ func (i *InstanceInfo) CreateOrGetInstance() error {
 			},
 		},
 	}
+	klog.Infof("=======Adding LocalSSD %v=============", localSSDCount)
 
+	localSSDConfig := &compute.AttachedDisk{
+		Type: "SCRATCH",
+		InitializeParams: &compute.AttachedDiskInitializeParams{
+			DiskType: fmt.Sprintf("zones/%s/diskTypes/local-ssd", i.zone),
+		},
+		AutoDelete: true,
+		Interface:  "NVME",
+	}
+
+	for i := 0; i < localSSDCount; i++ {
+		newInst.Disks = append(newInst.Disks, localSSDConfig)
+	}
 	saObj := &compute.ServiceAccount{
 		Email:  i.serviceAccount,
 		Scopes: []string{"https://www.googleapis.com/auth/cloud-platform"},
diff --git a/test/remote/setup-teardown.go b/test/remote/setup-teardown.go
index 3db9fcf1..a7762307 100644
--- a/test/remote/setup-teardown.go
+++ b/test/remote/setup-teardown.go
@@ -54,14 +54,14 @@ type processes struct {
 }
 
 // SetupInstance sets up the specified GCE Instance for E2E testing and returns a handle to the instance object for future use.
-func SetupInstance(config *InstanceConfig, instanceZone, instanceName string, cs *compute.Service) (*InstanceInfo, error) {
+func SetupInstance(config *InstanceConfig, instanceZone, instanceName string, cs *compute.Service, localSSDCount int) (*InstanceInfo, error) {
 	// Create the instance in the requisite zone
 	instance, err := CreateInstanceInfo(config, instanceZone, instanceName, cs)
 	if err != nil {
 		return nil, err
 	}
 
-	err = instance.CreateOrGetInstance()
+	err = instance.CreateOrGetInstance(localSSDCount)
 	if err != nil {
 		return nil, err
 	}
@@ -84,6 +84,12 @@ func SetupNewDriverAndClient(instance *InstanceInfo, config *ClientConfig) (*Tes
 		}
 	}()
 
+	// Copy dependencies
+	output, err := instance.SSH("apt-get", "install", "-y", "mdadm", "lvm2")
+	if err != nil {
+		return nil, fmt.Errorf("failed to install dependencis. Output: %v, errror: %v", output, err.Error())
+	}
+
 	// Upload archive to instance and run binaries
 	driverPID, err := instance.UploadAndRun(archivePath, config.WorkspaceDir, config.RunDriverCmd)
 	if err != nil {
diff --git a/test/run-e2e-local.sh b/test/run-e2e-local.sh
index 330ca37f..b62881f6 100755
--- a/test/run-e2e-local.sh
+++ b/test/run-e2e-local.sh
@@ -16,4 +16,4 @@ if hostname | grep -q c.googlers.com ; then
   CLOUDTOP_HOST=--cloudtop-host
 fi
 
-ginkgo --v "test/e2e/tests" -- --project "${PROJECT}" --service-account "${IAM_NAME}" "${CLOUDTOP_HOST}" --v=6 --logtostderr
+ginkgo --v --focus "Should create->attach->setup caching->write->detach->attach to different node->mount->read" "test/e2e/tests" -- --project "${PROJECT}" --service-account "${IAM_NAME}" "${CLOUDTOP_HOST}" --v=6 --logtostderr
diff --git a/test/sanity/sanity_test.go b/test/sanity/sanity_test.go
index f09c64ef..133d3e88 100644
--- a/test/sanity/sanity_test.go
+++ b/test/sanity/sanity_test.go
@@ -63,6 +63,7 @@ func TestSanity(t *testing.T) {
 
 	fallbackRequisiteZones := []string{}
 	enableStoragePools := false
+	enableDataCache := false
 	multiZoneVolumeHandleConfig := driver.MultiZoneVolumeHandleConfig{}
 	listVolumesConfig := driver.ListVolumesConfig{}
 
@@ -71,9 +72,9 @@ func TestSanity(t *testing.T) {
 
 	//Initialize GCE Driver
 	identityServer := driver.NewIdentityServer(gceDriver)
-	controllerServer := driver.NewControllerServer(gceDriver, cloudProvider, 0, 5*time.Minute, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig)
+	controllerServer := driver.NewControllerServer(gceDriver, cloudProvider, 0, 5*time.Minute, fallbackRequisiteZones, enableStoragePools, enableDataCache, multiZoneVolumeHandleConfig, listVolumesConfig)
 	fakeStatter := mountmanager.NewFakeStatterWithOptions(mounter, mountmanager.FakeStatterOptions{IsBlock: false})
-	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter)
+	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter, enableDataCache)
 	err = gceDriver.SetupGCEDriver(driverName, vendorVersion, extraLabels, nil, identityServer, controllerServer, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to initialize GCE CSI Driver: %v", err.Error())