Repository: kubernetes-sigs/aws-ebs-csi-driver

Issue:
Investigate Volume Cloning
**Is your feature request related to a problem? Please describe.**
Volume Cloning and creating a new volume from existing volume is a feature supported by [CSI 1.0](https://github.com/container-storage-interface/spec/blob/master/spec.md#controller-service-rpc). We should investigate whether it is beneficial to EBS

**Describe the solution you'd like in detail**
TBD



Hints from issue comments:
@fejta-bot: Issues go stale after 90d of inactivity.
Mark the issue as fresh with `/remove-lifecycle stale`.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with `/close`.

Send feedback to sig-testing, kubernetes/test-infra and/or [fejta](https://github.com/fejta).
/lifecycle stale@leakingtapan: /remove-lifecycle stale@fejta-bot: Issues go stale after 90d of inactivity.
Mark the issue as fresh with `/remove-lifecycle stale`.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with `/close`.

Send feedback to sig-testing, kubernetes/test-infra and/or [fejta](https://github.com/fejta).
/lifecycle stale@fejta-bot: Stale issues rot after 30d of inactivity.
Mark the issue as fresh with `/remove-lifecycle rotten`.
Rotten issues close after an additional 30d of inactivity.

If this issue is safe to close now please do so with `/close`.

Send feedback to sig-testing, kubernetes/test-infra and/or [fejta](https://github.com/fejta).
/lifecycle rotten@leakingtapan: /remove-lifecycle rotten@leakingtapan: /remove-lifecycle stale@fejta-bot: Issues go stale after 90d of inactivity.
Mark the issue as fresh with `/remove-lifecycle stale`.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with `/close`.

Send feedback to sig-testing, kubernetes/test-infra and/or [fejta](https://github.com/fejta).
/lifecycle stale@leakingtapan: /remove-lifecycle stale@MrNorm: +1@fejta-bot: Issues go stale after 90d of inactivity.
Mark the issue as fresh with `/remove-lifecycle stale`.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with `/close`.

Send feedback to sig-testing, kubernetes/test-infra and/or [fejta](https://github.com/fejta).
/lifecycle stale@frittentheke: +1@ArseniiPetrovich: +1@frittentheke: /remove-lifecycle stale

@ognots: +1@fejta-bot: Issues go stale after 90d of inactivity.
Mark the issue as fresh with `/remove-lifecycle stale`.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with `/close`.

Send feedback to sig-testing, kubernetes/test-infra and/or [fejta](https://github.com/fejta).
/lifecycle stale@ayberk: /remove-lifecycle stale
/lifecycle frozen@Jaybe-bdk: +1@phewitt: +1@hdiass: Is VolumeCloning something you guys support or not ?@Tishka17: I am also interested in `CSI Volume Cloning` feature. Is it supported or not?@nariman-maad: +@vishal-swarankar-sdl: 
I am also interested in CSI Volume Cloning feature. Is it supported or not?

@torredil: Currently, EBS doesn't support volume cloning. The driver can add support for cloning if/when EBS adds support for this.@HarryWeppner: @torredil, do you know the state of EBS volume cloning in 2023?@ksliu58: Thank you for your interest. For copying volumes, we recommend that customers take a snapshot of their volume and create a new volume (the copy) from that snapshot. Can you provide details on your use case for volume cloning?@FriedCircuits: @ksliu58  I could of using this feature today to restore a PV for our Jenkins deployment. I was redoing some stuff on the Jenkins helm chart and storageClass and wanted it to reattach to the previous volume. I could of quickly use this to create a new PVC from the existing one so it would be attached to this helm deployment. I used the work around and created a volume snapshot of the existing volume and use that with dataSource restore, but takes longer having to wait for AWS to create the snapshot etc. @AndrewSirenko: /priority important-longterm@roma-glushko: It would be nice to support this to simplify the EBS volume cloning workflow. 

Imagine there is some user-facing functionality where there is an entity that has a dedicated volume. Users may want to duplicate an entity which would entail duplication of the underlying volume. 

Doing the duplication via a temporary snapshot would definitely work but it's not the most simple way I guess. Also, in terms of cloning latency, if EBS CSI supported the K8S CSI cloning, would it work any faster than the "take snapshot -> create a new volume with a dataSource pointed to the snapshot" workflow?@JobRamos: Tried to perform this feature on EKS v1.29 but it is still unavailable, receiveing the message ProvisioningFailed "CSI driver does not support clone operations: controller CLONE_VOLUME capability is not reported".@ksliu58: > [@ksliu58](https://github.com/ksliu58) I could of using this feature today to restore a PV for our Jenkins deployment. I was redoing some stuff on the Jenkins helm chart and storageClass and wanted it to reattach to the previous volume. I could of quickly use this to create a new PVC from the existing one so it would be attached to this helm deployment. I used the work around and created a volume snapshot of the existing volume and use that with dataSource restore, but takes longer having to wait for AWS to create the snapshot etc.

Thanks for the background @FriedCircuits, While the clone conversation continues, I want to better understand you PV re-use scenario. Is it for unplanned DR, planned migration, test/dev environment, or something else? Some customers take snapshots before needing to restore a volume. For unplanned PV restores, you can also reduce the snapshot create time by taking snapshots more frequently. Do you have a specific RPO/RTO target?@billyboyballin: So I went through the code and I can see why adding the clone feature to ebs csi driver is little bit more unconventional but not impossible. AWS EBS does not have a direct "clone volume" feature. To make a copy of an ebs volume you must first take a snapshot and then restore that snapshot. Looking at [google cloud provider](https://cloud.google.com/compute/docs/disks/clone-duplicate-disks) you can directly clone volumes. 

So when user wants to clone a PVC, we first have to grab the sourceVolumeID from VolumeContentSource object, then create a temporary snapshot of that volume, wait for that snapshot to be created, then restore that snapshot into a new volume, wait for that restoration process to complete, and then finally we have a cloned volume that can be used. 

Essentially there's a timing factor. I only briefly looked over the code in this [file](https://github.com/kubernetes-sigs/aws-ebs-csi-driver/blob/master/pkg/driver/controller.go) and this [file](https://github.com/kubernetes-csi/external-provisioner/blob/master/pkg/controller/controller.go). I still think the step of creating temporary snapshot and restoring it into a new volume is possible but would need look more into how and where that happens. And the flow of it. 

As for why this cloning feature would be desirable, we're trying to implement blue green solution for stateful sets and the ability to simply create a new PVC from an existing PVC would simplify our flow. Right now it seems like we have to rely on the [volume snapshotter](https://github.com/kubernetes-sigs/aws-ebs-csi-driver/blob/master/examples/kubernetes/snapshot/README.md)@billyboyballin: I just read posts from earlier and seems like the juice is not worth squeeze for implementing this feature. It's not impossible, maybe the logic can go here [here](https://github.com/kubernetes-sigs/aws-ebs-csi-driver/tree/master/pkg/cloud). Anyways hopefully this helps 

PS = If AWS wants to reach to me with a job offer in return for adding this feature let me know. @RicoToothless: Even AWS supports snapshots, but it would be nice to integrate Kubernetes better.

Wait this feature +1

Solution patch:
diff --git a/docs/install.md b/docs/install.md
index 4cefadcb..5b23f2fd 100644
--- a/docs/install.md
+++ b/docs/install.md
@@ -80,6 +80,57 @@ Modification of tags of existing volumes can, in some configurations, allow the
 </pre>
 </details>
 
+<details>
+<summary>Creating a clone of a volume</summary>
+<br>
+The following statements give the EBS CSI Driver access to clone volumes:
+<pre>
+{ 
+  "Effect": "Allow",
+  "Action": [
+    "ec2:CopyVolumes"
+  ],
+  "Resource": "arn:aws:ec2:*:*:volume/vol-*"
+},
+{
+  "Effect": "Allow",
+  "Action": [
+    "ec2:CopyVolumes"
+  ],
+  "Resource": "arn:aws:ec2:*:*:volume/*",
+  "Condition": {
+    "StringLike": {
+      "aws:RequestTag/ebs.csi.aws.com/cluster": "true"
+    }
+  }
+},
+{
+  "Effect": "Allow",
+  "Action": [
+    "ec2:CopyVolumes"
+  ],
+  "Resource": "arn:aws:ec2:*:*:volume/*",
+  "Condition": {
+    "StringLike": {
+      "aws:RequestTag/CSIVolumeName": "*"
+    }
+  }
+},
+{
+  "Effect": "Allow",
+  "Action": [
+    "ec2:CopyVolumes"
+  ],
+  "Resource": "arn:aws:ec2:*:*:volume/*",
+  "Condition": {
+    "StringEquals": {
+      "ec2:CreateAction": "CopyVolumes"
+    }
+  }
+}
+</pre>
+</details>
+
 There are several options to pass credentials to the EBS CSI Driver, each documented below:
 
 #### (EKS Only) EKS Pod Identity
diff --git a/go.mod b/go.mod
index f44d3d23..98e41ccb 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,7 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.39.2
 	github.com/aws/aws-sdk-go-v2/config v1.31.12
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.9
-	github.com/aws/aws-sdk-go-v2/service/ec2 v1.254.1
+	github.com/aws/aws-sdk-go-v2/service/ec2 v1.256.0
 	github.com/aws/aws-sdk-go-v2/service/sagemaker v1.215.3
 	github.com/aws/aws-sdk-go-v2/service/sts v1.38.6
 	github.com/aws/smithy-go v1.23.0
diff --git a/go.sum b/go.sum
index 25445ef3..9770c724 100644
--- a/go.sum
+++ b/go.sum
@@ -19,8 +19,8 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.9 h1:6RBnKZLkJM4hQ+kN6E7
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.9/go.mod h1:V9rQKRmK7AWuEsOMnHzKj8WyrIir1yUJbZxDuZLFvXI=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
-github.com/aws/aws-sdk-go-v2/service/ec2 v1.254.1 h1:7p9bJCZ/b3EJXXARW7JMEs2IhsnI4YFHpfXQfgMh0eg=
-github.com/aws/aws-sdk-go-v2/service/ec2 v1.254.1/go.mod h1:M8WWWIfXmxA4RgTXcI/5cSByxRqjgne32Sh0VIbrn0A=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.256.0 h1:PLIiJRPfUKlgRazuO+hFuNGU2E2IHRqRW/nq4WzG9r4=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.256.0/go.mod h1:M8WWWIfXmxA4RgTXcI/5cSByxRqjgne32Sh0VIbrn0A=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1 h1:oegbebPEMA/1Jny7kvwejowCaHz1FWZAQ94WXFNCyTM=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1/go.mod h1:kemo5Myr9ac0U9JfSjMo9yHLtw+pECEHsFtJ9tqCEI8=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.9 h1:5r34CgVOD4WZudeEKZ9/iKpiT6cM1JyEROpXjOcdWv8=
diff --git a/hack/e2e/config.sh b/hack/e2e/config.sh
index 284f44b3..dbbbf343 100644
--- a/hack/e2e/config.sh
+++ b/hack/e2e/config.sh
@@ -71,5 +71,5 @@ COLLECT_METRICS=${COLLECT_METRICS:-"false"}
 
 TEST_PATH=${TEST_PATH:-"./tests/e2e-kubernetes/..."}
 GINKGO_FOCUS=${GINKGO_FOCUS:-"External.Storage"}
-GINKGO_SKIP=${GINKGO_SKIP:-"\[Disruptive\]|\[Serial\]|\[Flaky\]"}
+GINKGO_SKIP=${GINKGO_SKIP:-"\[Disruptive\]|\[Serial\]|\[Flaky\]|should provision storage with pvc data source in parallel"}
 GINKGO_PARALLEL=${GINKGO_PARALLEL:-25}
diff --git a/hack/e2e/eksctl/cluster.yaml b/hack/e2e/eksctl/cluster.yaml
index e64b5792..2c7f42ae 100644
--- a/hack/e2e/eksctl/cluster.yaml
+++ b/hack/e2e/eksctl/cluster.yaml
@@ -28,6 +28,35 @@ iam:
         namespace: kube-system
       wellKnownPolicies:
         ebsCSIController: true
+      attachPolicy:
+        Version: '2012-10-17'
+        Statement:
+          - Effect: Allow
+            Action:
+              - ec2:CopyVolumes
+            Resource: "arn:aws:ec2:*:*:volume/vol-*"
+          - Effect: Allow
+            Action:
+              - ec2:CopyVolumes
+            Resource: "arn:aws:ec2:*:*:volume/*"
+            Condition:
+              StringLike:
+                "aws:RequestTag/ebs.csi.aws.com/cluster": "true"
+          - Effect: Allow
+            Action:
+              - ec2:CopyVolumes
+            Resource: "arn:aws:ec2:*:*:volume/*"
+            Condition:
+              StringLike:
+                "aws:RequestTag/CSIVolumeName": "*"
+          - Effect: Allow
+            Action:
+              - ec2:CreateTags
+            Resource:
+              - "arn:aws:ec2:*:*:volume/*"
+            Condition:
+              StringEquals:
+                "ec2:CreateAction": "CopyVolumes"
 managedNodeGroups:
   - name: ng-linux
     amiFamily: {{ .Env.AMI_FAMILY }}
diff --git a/hack/e2e/kops/patch-cluster.yaml b/hack/e2e/kops/patch-cluster.yaml
index 32f3ccbe..9557cdbc 100644
--- a/hack/e2e/kops/patch-cluster.yaml
+++ b/hack/e2e/kops/patch-cluster.yaml
@@ -48,6 +48,15 @@ spec:
           ],
           "Resource": "arn:aws:ec2:*:*:volume/*"
         },
+        {
+          "Effect": "Allow",
+          "Action": [
+            "ec2:CopyVolumes"
+          ],
+          "Resource": [
+            "arn:aws:ec2:*:*:volume/vol-*"
+          ]
+        },
         {
           "Effect": "Allow",
           "Action": [
@@ -80,7 +89,8 @@ spec:
             "StringEquals": {
               "ec2:CreateAction": [
                 "CreateVolume",
-                "CreateSnapshot"
+                "CreateSnapshot",
+                "CopyVolumes"
               ]
             }
           }
@@ -98,7 +108,8 @@ spec:
         {
           "Effect": "Allow",
           "Action": [
-            "ec2:CreateVolume"
+            "ec2:CreateVolume",
+            "ec2:CopyVolumes"
           ],
           "Resource": "arn:aws:ec2:*:*:volume/*",
           "Condition": {
@@ -110,7 +121,8 @@ spec:
         {
           "Effect": "Allow",
           "Action": [
-            "ec2:CreateVolume"
+            "ec2:CreateVolume",
+            "ec2:CopyVolumes"
           ],
           "Resource": "arn:aws:ec2:*:*:volume/*",
           "Condition": {
diff --git a/hack/e2e/run.sh b/hack/e2e/run.sh
index c1a02713..482aa8d4 100755
--- a/hack/e2e/run.sh
+++ b/hack/e2e/run.sh
@@ -135,7 +135,7 @@ else
     packageVersion=$(echo $(cut -d '.' -f 1,2 <<<$K8S_VERSION))
 
     # TODO: Always skip broken upstream test - remove after fix released
-    GINKGO_SKIP="(should be protected by vac\\-protection finalizer)|${GINKGO_SKIP}"
+    GINKGO_SKIP="(should be protected by vac\\-protection finalizer)|should provision storage with pvc data source in parallel|${GINKGO_SKIP}"
     GINKGO_SKIP="${GINKGO_SKIP%|}" # Strip trailing | if needed - remove with above TODO
     set -x
     set +e
diff --git a/pkg/cloud/cloud.go b/pkg/cloud/cloud.go
index 2b2a0ddd..d1b2eca8 100644
--- a/pkg/cloud/cloud.go
+++ b/pkg/cloud/cloud.go
@@ -219,8 +219,10 @@ type Disk struct {
 	CapacityGiB        int32
 	AvailabilityZone   string
 	AvailabilityZoneID string
+	SourceVolumeID     string
 	SnapshotID         string
 	OutpostArn         string
+	KmsKeyID           string
 	Attachments        []string
 }
 
@@ -242,6 +244,7 @@ type DiskOptions struct {
 	// example: arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef
 	KmsKeyID                 string
 	SnapshotID               string
+	SourceVolumeID           string
 	VolumeInitializationRate int32
 }
 
@@ -605,8 +608,13 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 		throughput    int32
 		err           error
 		requestedIops int32
+		size          int32
+		outpostArn    string
+		volumeID      string
 	)
 
+	isClone := diskOptions.SourceVolumeID != ""
+
 	capacityGiB := util.BytesToGiB(diskOptions.CapacityBytes)
 
 	if diskOptions.IOPS > 0 && diskOptions.IOPSPerGB > 0 {
@@ -659,26 +667,6 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 	// We use a sha256 hash to guarantee the token that is less than or equal to 64 characters
 	clientToken := sha256.Sum256([]byte(tokenBase))
 
-	requestInput := &ec2.CreateVolumeInput{
-		ClientToken:        aws.String(hex.EncodeToString(clientToken[:])),
-		Size:               aws.Int32(capacityGiB),
-		VolumeType:         types.VolumeType(createType),
-		Encrypted:          aws.Bool(diskOptions.Encrypted),
-		MultiAttachEnabled: aws.Bool(diskOptions.MultiAttachEnabled),
-		TagSpecifications:  []types.TagSpecification{tagSpec},
-	}
-
-	if len(zone) > 0 {
-		requestInput.AvailabilityZone = aws.String(zone)
-	}
-	if len(zoneID) > 0 {
-		requestInput.AvailabilityZoneId = aws.String(zoneID)
-	}
-	// EBS doesn't handle empty outpost arn, so we have to include it only when it's non-empty
-	if len(diskOptions.OutpostArn) > 0 {
-		requestInput.OutpostArn = aws.String(diskOptions.OutpostArn)
-	}
-
 	azParams := getVolumeLimitsParams{
 		availabilityZone:   zone,
 		availabilityZoneId: zoneID,
@@ -699,27 +687,27 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 		iops = capIOPS(createType, capacityGiB, requestedIops, iopsLimit, diskOptions.AllowIOPSPerGBIncrease)
 	}
 
-	if len(diskOptions.KmsKeyID) > 0 {
-		requestInput.KmsKeyId = aws.String(diskOptions.KmsKeyID)
-		requestInput.Encrypted = aws.Bool(true)
-	}
-	if iops > 0 {
-		requestInput.Iops = aws.Int32(iops)
-	}
-	if throughput > 0 {
-		requestInput.Throughput = aws.Int32(throughput)
-	}
-	snapshotID := diskOptions.SnapshotID
-	if len(snapshotID) > 0 {
-		requestInput.SnapshotId = aws.String(snapshotID)
-	}
-	if diskOptions.VolumeInitializationRate > 0 {
-		requestInput.VolumeInitializationRate = aws.Int32(diskOptions.VolumeInitializationRate)
+	if isClone {
+		copyRequestInput := &ec2.CopyVolumesInput{
+			SourceVolumeId:     aws.String(diskOptions.SourceVolumeID),
+			ClientToken:        aws.String(hex.EncodeToString(clientToken[:])),
+			Size:               aws.Int32(capacityGiB),
+			VolumeType:         types.VolumeType(createType),
+			MultiAttachEnabled: aws.Bool(diskOptions.MultiAttachEnabled),
+			TagSpecifications:  []types.TagSpecification{tagSpec},
+		}
+		size, outpostArn, volumeID, err = c.createCloneHelper(ctx, copyRequestInput, iops, throughput)
+	} else {
+		createRequestInput := &ec2.CreateVolumeInput{
+			ClientToken:        aws.String(hex.EncodeToString(clientToken[:])),
+			Size:               aws.Int32(capacityGiB),
+			VolumeType:         types.VolumeType(createType),
+			Encrypted:          aws.Bool(diskOptions.Encrypted),
+			MultiAttachEnabled: aws.Bool(diskOptions.MultiAttachEnabled),
+			TagSpecifications:  []types.TagSpecification{tagSpec},
+		}
+		size, outpostArn, volumeID, err = c.createVolumeHelper(ctx, diskOptions, createRequestInput, iops, throughput, zone, zoneID)
 	}
-
-	response, err := c.ec2.CreateVolume(ctx, requestInput, func(o *ec2.Options) {
-		o.Retryer = c.rm.createVolumeRetryer
-	})
 	if err != nil {
 		if isAWSErrorSnapshotNotFound(err) {
 			return nil, ErrNotFound
@@ -732,24 +720,22 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 			c.latestClientTokens.Set(volumeName, &nextTokenNumber)
 			return nil, ErrIdempotentParameterMismatch
 		}
-		if isAWSErrorInvalidParameterCombination(err) {
-			return nil, ErrInvalidArgument
-		}
 		if isAWSErrorVolumeLimitExceeded(err) {
 			return nil, fmt.Errorf("%w: %w", ErrLimitExceeded, err)
 		}
+		if isAWSErrorInvalidParameterCombination(err) {
+			return nil, ErrInvalidArgument
+		}
 		if isAwsErrorMaxIOPSLimitExceeded(err) {
 			return nil, fmt.Errorf("%w: %w", ErrLimitExceeded, err)
 		}
 		return nil, fmt.Errorf("could not create volume in EC2: %w", err)
 	}
 
-	volumeID := aws.ToString(response.VolumeId)
 	if len(volumeID) == 0 {
 		return nil, errors.New("volume ID was not returned by CreateVolume")
 	}
 
-	size := *response.Size
 	if size == 0 {
 		return nil, errors.New("disk size was not returned by CreateVolume")
 	}
@@ -757,8 +743,66 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 	if err := c.waitForVolume(ctx, volumeID); err != nil {
 		return nil, fmt.Errorf("timed out waiting for volume to create: %w", err)
 	}
+	return &Disk{CapacityGiB: size, VolumeID: volumeID, AvailabilityZone: zone, SnapshotID: diskOptions.SnapshotID, SourceVolumeID: diskOptions.SourceVolumeID, OutpostArn: outpostArn}, nil
+}
 
-	return &Disk{CapacityGiB: size, VolumeID: volumeID, AvailabilityZone: zone, SnapshotID: snapshotID, OutpostArn: aws.ToString(response.OutpostArn)}, nil
+func (c *cloud) createCloneHelper(ctx context.Context, input *ec2.CopyVolumesInput, iops int32, throughput int32) (int32, string, string, error) {
+	if iops > 0 {
+		input.Iops = aws.Int32(iops)
+	}
+	if throughput > 0 {
+		input.Throughput = aws.Int32(throughput)
+	}
+	copyResponse, err := c.ec2.CopyVolumes(ctx, input, func(o *ec2.Options) {
+		o.Retryer = c.rm.copyVolumeRetryer
+	})
+	if err != nil {
+		return 0, "", "", err
+	}
+	if len(copyResponse.Volumes) != 1 {
+		return 0, "", "", errors.New("copyResponse does not contain volume information")
+	}
+	return *copyResponse.Volumes[0].Size, aws.ToString(copyResponse.Volumes[0].OutpostArn), aws.ToString(copyResponse.Volumes[0].VolumeId), nil
+}
+
+func (c *cloud) createVolumeHelper(ctx context.Context, diskOptions *DiskOptions, input *ec2.CreateVolumeInput, iops int32, throughput int32, zone string, zoneID string) (int32, string, string, error) {
+	if len(zone) > 0 {
+		input.AvailabilityZone = aws.String(zone)
+	}
+	if len(zoneID) > 0 {
+		input.AvailabilityZoneId = aws.String(zoneID)
+	}
+	if len(diskOptions.OutpostArn) > 0 {
+		input.OutpostArn = aws.String(diskOptions.OutpostArn)
+	}
+
+	if len(diskOptions.KmsKeyID) > 0 {
+		input.KmsKeyId = aws.String(diskOptions.KmsKeyID)
+		input.Encrypted = aws.Bool(true)
+	}
+
+	if iops > 0 {
+		input.Iops = aws.Int32(iops)
+	}
+
+	if diskOptions.Throughput > 0 {
+		input.Throughput = aws.Int32(throughput)
+	}
+	snapshotID := diskOptions.SnapshotID
+	if len(snapshotID) > 0 {
+		input.SnapshotId = aws.String(snapshotID)
+	}
+	if diskOptions.VolumeInitializationRate > 0 {
+		input.VolumeInitializationRate = aws.Int32(diskOptions.VolumeInitializationRate)
+	}
+
+	createResponse, err := c.ec2.CreateVolume(ctx, input, func(o *ec2.Options) {
+		o.Retryer = c.rm.createVolumeRetryer
+	})
+	if err != nil {
+		return 0, "", "", err
+	}
+	return *createResponse.Size, aws.ToString(createResponse.OutpostArn), aws.ToString(createResponse.VolumeId), nil
 }
 
 // execBatchDescribeVolumesModifications executes a batched DescribeVolumesModifications API call.
@@ -1488,6 +1532,7 @@ func (c *cloud) GetDiskByID(ctx context.Context, volumeID string) (*Disk, error)
 		AvailabilityZone: aws.ToString(volume.AvailabilityZone),
 		OutpostArn:       aws.ToString(volume.OutpostArn),
 		Attachments:      getVolumeAttachmentsList(*volume),
+		KmsKeyID:         aws.ToString(volume.KmsKeyId),
 	}
 
 	if volume.Size != nil {
diff --git a/pkg/cloud/ec2_interface.go b/pkg/cloud/ec2_interface.go
index 31246ce8..236affa3 100644
--- a/pkg/cloud/ec2_interface.go
+++ b/pkg/cloud/ec2_interface.go
@@ -24,6 +24,7 @@ type EC2API interface {
 	DescribeVolumes(ctx context.Context, params *ec2.DescribeVolumesInput, optFns ...func(*ec2.Options)) (*ec2.DescribeVolumesOutput, error)
 	DescribeVolumeStatus(ctx context.Context, params *ec2.DescribeVolumeStatusInput, optFns ...func(options *ec2.Options)) (*ec2.DescribeVolumeStatusOutput, error)
 	CreateVolume(ctx context.Context, params *ec2.CreateVolumeInput, optFns ...func(*ec2.Options)) (*ec2.CreateVolumeOutput, error)
+	CopyVolumes(ctx context.Context, params *ec2.CopyVolumesInput, optFns ...func(*ec2.Options)) (*ec2.CopyVolumesOutput, error)
 	DeleteVolume(ctx context.Context, params *ec2.DeleteVolumeInput, optFns ...func(*ec2.Options)) (*ec2.DeleteVolumeOutput, error)
 	AttachVolume(ctx context.Context, params *ec2.AttachVolumeInput, optFns ...func(*ec2.Options)) (*ec2.AttachVolumeOutput, error)
 	DetachVolume(ctx context.Context, params *ec2.DetachVolumeInput, optFns ...func(*ec2.Options)) (*ec2.DetachVolumeOutput, error)
diff --git a/pkg/cloud/mock_ec2.go b/pkg/cloud/mock_ec2.go
index 7beb3f3a..c12d9f5e 100644
--- a/pkg/cloud/mock_ec2.go
+++ b/pkg/cloud/mock_ec2.go
@@ -55,6 +55,26 @@ func (mr *MockEC2APIMockRecorder) AttachVolume(ctx, params interface{}, optFns .
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AttachVolume", reflect.TypeOf((*MockEC2API)(nil).AttachVolume), varargs...)
 }
 
+// CopyVolumes mocks base method.
+func (m *MockEC2API) CopyVolumes(ctx context.Context, params *ec2.CopyVolumesInput, optFns ...func(*ec2.Options)) (*ec2.CopyVolumesOutput, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, params}
+	for _, a := range optFns {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "CopyVolumes", varargs...)
+	ret0, _ := ret[0].(*ec2.CopyVolumesOutput)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CopyVolumes indicates an expected call of CopyVolumes.
+func (mr *MockEC2APIMockRecorder) CopyVolumes(ctx, params interface{}, optFns ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, params}, optFns...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CopyVolumes", reflect.TypeOf((*MockEC2API)(nil).CopyVolumes), varargs...)
+}
+
 // CreateSnapshot mocks base method.
 func (m *MockEC2API) CreateSnapshot(ctx context.Context, params *ec2.CreateSnapshotInput, optFns ...func(*ec2.Options)) (*ec2.CreateSnapshotOutput, error) {
 	m.ctrl.T.Helper()
diff --git a/pkg/cloud/retry_manager.go b/pkg/cloud/retry_manager.go
index 077991d7..2c7ee7d6 100644
--- a/pkg/cloud/retry_manager.go
+++ b/pkg/cloud/retry_manager.go
@@ -33,6 +33,7 @@ const (
 // Separate retryers ensures that throttling one API doesn't unintentionally throttle others with separate token buckets.
 type retryManager struct {
 	createVolumeRetryer                            aws.Retryer
+	copyVolumeRetryer                              aws.Retryer
 	deleteVolumeRetryer                            aws.Retryer
 	attachVolumeRetryer                            aws.Retryer
 	detachVolumeRetryer                            aws.Retryer
@@ -46,6 +47,7 @@ type retryManager struct {
 func newRetryManager() *retryManager {
 	return &retryManager{
 		createVolumeRetryer:                            newAdaptiveRetryer(),
+		copyVolumeRetryer:                              newAdaptiveRetryer(),
 		attachVolumeRetryer:                            newAdaptiveRetryer(),
 		deleteVolumeRetryer:                            newAdaptiveRetryer(),
 		detachVolumeRetryer:                            newAdaptiveRetryer(),
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 2cd4603d..24a61ebd 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -47,6 +47,7 @@ var (
 	// controllerCaps represents the capability of controller service.
 	controllerCaps = []csi.ControllerServiceCapability_RPC_Type{
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+		csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
 		csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 		csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
@@ -113,6 +114,7 @@ func (d *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 		throughput               int32
 		volumeInitializationRate int32
 		isEncrypted              bool
+		encryptedKey             string
 		kmsKeyID                 string
 		tagsToEvaluate           = make([]string, 0)
 		volumeTags               = map[string]string{
@@ -164,6 +166,7 @@ func (d *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 			throughput = int32(parseThroughput)
 		case EncryptedKey:
 			isEncrypted = isTrue(value)
+			encryptedKey = value
 		case KmsKeyIDKey:
 			kmsKeyID = value
 		case PVCNameKey:
@@ -324,22 +327,58 @@ func (d *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 	}
 
 	snapshotID := ""
+	volumeID := ""
 	volumeSource := req.GetVolumeContentSource()
 	if volumeSource != nil {
-		if _, ok := volumeSource.GetType().(*csi.VolumeContentSource_Snapshot); !ok {
+		sourceSnapshot := volumeSource.GetSnapshot()
+		sourceVolume := volumeSource.GetVolume()
+
+		if sourceSnapshot != nil && sourceVolume != nil {
+			return nil, status.Error(codes.InvalidArgument, "Cannot have more than one volume source")
+		}
+
+		if sourceSnapshot == nil && sourceVolume == nil {
 			return nil, status.Error(codes.InvalidArgument, "Unsupported volumeContentSource type")
 		}
-		sourceSnapshot := volumeSource.GetSnapshot()
-		if sourceSnapshot == nil {
-			return nil, status.Error(codes.InvalidArgument, "Error retrieving snapshot from the volumeContentSource")
+
+		if sourceSnapshot != nil {
+			snapshotID = sourceSnapshot.GetSnapshotId()
+		}
+
+		if sourceVolume != nil {
+			if encryptedKey != "" && !isEncrypted {
+				return nil, status.Error(codes.InvalidArgument, "Cannot make an unencrypted clone")
+			}
+			volumeID = sourceVolume.GetVolumeId()
 		}
-		snapshotID = sourceSnapshot.GetSnapshotId()
 	}
+	var zone string
+	var zoneID string
+	var outpostArn string
+	// create or clone a new volume
+	if volumeID != "" {
+		sourceVolume, err := d.cloud.GetDiskByID(ctx, volumeID)
+
+		if kmsKeyID != "" && sourceVolume.KmsKeyID != kmsKeyID {
+			return nil, status.Errorf(codes.InvalidArgument, "Cannot provision clone with different KMS key than source volume")
+		}
 
-	// create a new volume
-	zone := pickAvailabilityZone(req.GetAccessibilityRequirements())
-	zoneID := pickAvailabilityZoneID(req.GetAccessibilityRequirements())
-	outpostArn := getOutpostArn(req.GetAccessibilityRequirements())
+		if err != nil {
+			return nil, status.Errorf(codes.NotFound, "Error source volume with volumeID %v not found: %v", volumeID, err)
+		}
+
+		err = checkSourceTopology(req.GetAccessibilityRequirements(), sourceVolume.AvailabilityZone, sourceVolume.OutpostArn, sourceVolume.AvailabilityZoneID)
+		if err != nil {
+			return nil, err
+		}
+		zone = sourceVolume.AvailabilityZone
+		zoneID = sourceVolume.AvailabilityZoneID
+		outpostArn = sourceVolume.OutpostArn
+	} else {
+		zone = pickAvailabilityZone(req.GetAccessibilityRequirements())
+		zoneID = pickAvailabilityZoneID(req.GetAccessibilityRequirements())
+		outpostArn = getOutpostArn(req.GetAccessibilityRequirements())
+	}
 
 	opts := &cloud.DiskOptions{
 		CapacityBytes:            volSizeBytes,
@@ -355,6 +394,7 @@ func (d *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 		Encrypted:                isEncrypted,
 		KmsKeyID:                 kmsKeyID,
 		SnapshotID:               snapshotID,
+		SourceVolumeID:           volumeID,
 		MultiAttachEnabled:       multiAttach,
 		VolumeInitializationRate: volumeInitializationRate,
 	}
@@ -990,6 +1030,51 @@ func getOutpostArn(requirement *csi.TopologyRequirement) string {
 	return ""
 }
 
+// Check if source volumes topology matches with clones requisite topology requirements.
+func checkSourceTopology(requirement *csi.TopologyRequirement, sourceVolumeZone string, sourceVolumeOutpostArn string, sourceVolumeZoneID string) error {
+	if requirement.GetRequisite() == nil || requirement == nil {
+		return nil
+	}
+	for _, toplogy := range requirement.GetRequisite() {
+		zone, hasZone := toplogy.GetSegments()[WellKnownZoneTopologyKey]
+		if hasZone && zone == sourceVolumeZone {
+			_, hasOutpost := toplogy.GetSegments()[AwsOutpostIDKey]
+			if hasOutpost {
+				if BuildOutpostArn(toplogy.GetSegments()) == sourceVolumeOutpostArn {
+					return nil
+				}
+			} else {
+				return nil
+			}
+		}
+
+		zone, hasZone = toplogy.GetSegments()[ZoneTopologyKey]
+		if hasZone && zone == sourceVolumeZone {
+			_, hasOutpost := toplogy.GetSegments()[AwsOutpostIDKey]
+			if hasOutpost {
+				if BuildOutpostArn(toplogy.GetSegments()) == sourceVolumeOutpostArn {
+					return nil
+				}
+			} else {
+				return nil
+			}
+		}
+
+		zoneid, hasZoneID := toplogy.GetSegments()[ZoneIDTopologyKey]
+		if hasZoneID && zoneid == sourceVolumeZoneID {
+			_, hasOutpost := toplogy.GetSegments()[AwsOutpostIDKey]
+			if hasOutpost {
+				if BuildOutpostArn(toplogy.GetSegments()) == sourceVolumeOutpostArn {
+					return nil
+				}
+			} else {
+				return nil
+			}
+		}
+	}
+	return status.Errorf(codes.ResourceExhausted, "Cannot provision clone with the specified topology constraints")
+}
+
 func newCreateVolumeResponse(disk *cloud.Disk, ctx map[string]string) *csi.CreateVolumeResponse {
 	var src *csi.VolumeContentSource
 	if disk.SnapshotID != "" {
@@ -1002,6 +1087,16 @@ func newCreateVolumeResponse(disk *cloud.Disk, ctx map[string]string) *csi.Creat
 		}
 	}
 
+	if disk.SourceVolumeID != "" {
+		src = &csi.VolumeContentSource{
+			Type: &csi.VolumeContentSource_Volume{
+				Volume: &csi.VolumeContentSource_VolumeSource{
+					VolumeId: disk.SourceVolumeID,
+				},
+			},
+		}
+	}
+
 	segments := map[string]string{WellKnownZoneTopologyKey: disk.AvailabilityZone}
 
 	arn, err := arn.Parse(disk.OutpostArn)



Test patch:
diff --git a/pkg/cloud/cloud_test.go b/pkg/cloud/cloud_test.go
index 10c88c6c..ea316528 100644
--- a/pkg/cloud/cloud_test.go
+++ b/pkg/cloud/cloud_test.go
@@ -128,7 +128,6 @@ func TestNewCloud(t *testing.T) {
 		}
 	}
 }
-
 func TestBatchDescribeVolumes(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {
@@ -833,9 +832,11 @@ func TestCreateDisk(t *testing.T) {
 		expDisk              *Disk
 		expErr               error
 		expCreateVolumeErr   error
+		expCopyVolumesErr    error
 		expDescVolumeErr     error
 		expCreateTagsErr     error
 		expCreateVolumeInput *ec2.CreateVolumeInput
+		expCopyVolumesInput  *ec2.CopyVolumesInput
 	}{
 		{
 			name:       "success: normal",
@@ -852,6 +853,23 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: normal clone",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: normal with iops",
 			volumeName: "vol-test-name",
@@ -870,6 +888,26 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone with iops",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(500),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				IOPS:             6000,
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      500,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(6000),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: normal with gp2 options",
 			volumeName: "vol-test-name",
@@ -886,6 +924,24 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone with gp2 options",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test"},
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+				VolumeType:       VolumeTypeGP2,
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              nil,
+		},
 		{
 			name:       "success: normal with io2 options",
 			volumeName: "vol-test-name",
@@ -905,6 +961,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone with io2 options",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+				VolumeType:       VolumeTypeIO2,
+				IOPSPerGB:        100,
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(100),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: io1 with IOPS parameter",
 			volumeName: "vol-test-name",
@@ -924,6 +1001,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone io1 with IOPS parameter",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(200),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+				VolumeType:       VolumeTypeIO1,
+				IOPSPerGB:        100,
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      200,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(100),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: io2 with IOPS parameter",
 			volumeName: "vol-test-name",
@@ -943,6 +1041,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone io2 with IOPS parameter",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+				VolumeType:       VolumeTypeIO2,
+				IOPSPerGB:        100,
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(100),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: normal with gp3 options",
 			volumeName: "vol-test-name",
@@ -964,6 +1083,29 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone gp3 options",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(400),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				SourceVolumeID:   "test-vol-id",
+				AvailabilityZone: defaultZone,
+				VolumeType:       VolumeTypeGP3,
+				IOPS:             3000,
+				Throughput:       125,
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      400,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops:       aws.Int32(3000),
+				Throughput: aws.Int32(125),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: normal with provided zone",
 			volumeName: "vol-test-name",
@@ -1014,6 +1156,25 @@ func TestCreateDisk(t *testing.T) {
 			expCreateVolumeInput: &ec2.CreateVolumeInput{},
 			expErr:               nil,
 		},
+		{
+			name:       "success: clone with encrypted volume",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				Encrypted:        true,
+				KmsKeyID:         "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+				SourceVolumeID:   "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: expZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              nil,
+		},
 		{
 			name:       "success: outpost volume",
 			volumeName: "vol-test-name",
@@ -1032,6 +1193,25 @@ func TestCreateDisk(t *testing.T) {
 			expCreateVolumeInput: &ec2.CreateVolumeInput{},
 			expErr:               nil,
 		},
+		{
+			name:       "success: outpost clone",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				OutpostArn:       "arn:aws:outposts:us-west-2:111111111111:outpost/op-0aaa000a0aaaa00a0",
+				SourceVolumeID:   "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: expZone,
+				OutpostArn:       "arn:aws:outposts:us-west-2:111111111111:outpost/op-0aaa000a0aaaa00a0",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              nil,
+		},
 		{
 			name:       "success: empty outpost arn",
 			volumeName: "vol-test-name",
@@ -1049,6 +1229,24 @@ func TestCreateDisk(t *testing.T) {
 			expCreateVolumeInput: &ec2.CreateVolumeInput{},
 			expErr:               nil,
 		},
+		{
+			name:       "success: empty outpost arn clone",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: expZone,
+				OutpostArn:       "",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              nil,
+		},
 		{
 			name:       "fail: ec2.CreateVolume returned CreateVolume error",
 			volumeName: "vol-test-name-error",
@@ -1061,6 +1259,19 @@ func TestCreateDisk(t *testing.T) {
 			expErr:               errors.New("could not create volume in EC2: CreateVolume generic error"),
 			expCreateVolumeErr:   errors.New("CreateVolume generic error"),
 		},
+		{
+			name:       "fail: ec2.CopyVolume returned CopyVolume error",
+			volumeName: "vol-test-name-error",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              errors.New("could not create volume in EC2: CopyVolumes generic error"),
+			expCopyVolumesErr:   errors.New("CopyVolumes generic error"),
+		},
 		{
 			name:       "fail: ec2.CreateVolume returned snapshot not found error",
 			volumeName: "vol-test-name-error",
@@ -1076,6 +1287,19 @@ func TestCreateDisk(t *testing.T) {
 				Message: "Snapshot not found",
 			},
 		},
+		{
+			name:       "fail: ec2.CopyVolumes returned source volume not found error",
+			volumeName: "vol-test-name-error",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              fmt.Errorf("could not create volume in EC2: %w", errors.New("copyResponse does not contain volume information")),
+			expCopyVolumesErr:   errors.New("copyResponse does not contain volume information"),
+		},
 		{
 			name:       "fail: ec2.CreateVolume returned Idempotent Parameter Mismatch error",
 			volumeName: "vol-test-name-error",
@@ -1088,6 +1312,19 @@ func TestCreateDisk(t *testing.T) {
 			expErr:               fmt.Errorf("could not create volume in EC2: %w", errors.New("an error occurred: IdempotentParameterMismatch")),
 			expCreateVolumeErr:   fmt.Errorf("an error occurred: %w", errors.New("IdempotentParameterMismatch")),
 		},
+		{
+			name:       "fail: ec2.CopyVolumes returned Idempotent Parameter Mismatch error",
+			volumeName: "vol-test-name-error",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              fmt.Errorf("could not create volume in EC2: %w", errors.New("an error occurred: IdempotentParameterMismatch")),
+			expCopyVolumesErr:   fmt.Errorf("an error occurred: %w", errors.New("IdempotentParameterMismatch")),
+		},
 		{
 			name:       "fail: ec2.DescribeVolumes error after volume created",
 			volumeName: "vol-test-name-error",
@@ -1101,6 +1338,20 @@ func TestCreateDisk(t *testing.T) {
 			expErr:               errors.New("timed out waiting for volume to create: DescribeVolumes generic error"),
 			expDescVolumeErr:     errors.New("DescribeVolumes generic error"),
 		},
+		{
+			name:       "fail: clone ec2.DescribeVolumes error after volume created",
+			volumeName: "vol-test-name-error",
+			volState:   "creating",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: expZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              errors.New("timed out waiting for volume to create: DescribeVolumes generic error"),
+			expDescVolumeErr:    errors.New("DescribeVolumes generic error"),
+		},
 		{
 			name:       "fail: Volume is not ready to use, volume stuck in creating status and controller context deadline exceeded",
 			volumeName: "vol-test-name-error",
@@ -1113,6 +1364,19 @@ func TestCreateDisk(t *testing.T) {
 			expCreateVolumeInput: &ec2.CreateVolumeInput{},
 			expErr:               errors.New("timed out waiting for volume to create: timed out waiting for the condition"),
 		},
+		{
+			name:       "fail: Clone Volume is not ready to use, volume stuck in creating status and controller context deadline exceeded",
+			volumeName: "vol-test-name-error",
+			volState:   "creating",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				AvailabilityZone: defaultZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{},
+			expErr:              errors.New("timed out waiting for volume to create: timed out waiting for the condition"),
+		},
 		{
 			name:       "success: normal from snapshot",
 			volumeName: "vol-test-name",
@@ -1150,6 +1414,28 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone io1 with too low iopsPerGB and AllowIOPSPerGBIncrease",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:          util.GiBToBytes(4),
+				Tags:                   map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:             VolumeTypeIO1,
+				IOPSPerGB:              1,
+				AllowIOPSPerGBIncrease: true,
+				AvailabilityZone:       defaultZone,
+				SourceVolumeID:         "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(100),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "fail: invalid StorageClass parameters; specified both IOPS and IOPSPerGb",
 			volumeName: "vol-test-name",
@@ -1187,6 +1473,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone small io1 with too high iopsPerGB",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(4),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:       VolumeTypeIO1,
+				IOPSPerGB:        10000,
+				AvailabilityZone: defaultZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(200),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: large io1 with too high iopsPerGB",
 			volumeName: "vol-test-name",
@@ -1206,6 +1513,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone large io1 with too high iopsPerGB",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(4000),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:       VolumeTypeIO1,
+				IOPSPerGB:        10000,
+				AvailabilityZone: defaultZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4000,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(64000),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: io2 with too low iopsPerGB and AllowIOPSPerGBIncrease",
 			volumeName: "vol-test-name",
@@ -1226,6 +1554,28 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone io2 with too low iopsPerGB and AllowIOPSPerGBIncrease",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:          util.GiBToBytes(4),
+				Tags:                   map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:             VolumeTypeIO2,
+				IOPSPerGB:              1,
+				AllowIOPSPerGBIncrease: true,
+				AvailabilityZone:       defaultZone,
+				SourceVolumeID:         "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(100),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: small io2 with too high iopsPerGB",
 			volumeName: "vol-test-name",
@@ -1245,6 +1595,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone small io2 with too high iopsPerGB",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(4),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:       VolumeTypeIO2,
+				IOPSPerGB:        10000,
+				AvailabilityZone: defaultZone,
+				SourceVolumeID:   "test-vol-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(2000),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: large io2 with too high iopsPerGB",
 			volumeName: "vol-test-name",
@@ -1264,6 +1635,27 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone large io2 with too high iopsPerGB",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(4000),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:       VolumeTypeIO2,
+				IOPSPerGB:        100000,
+				AvailabilityZone: defaultZone,
+				SourceVolumeID:   "vol-test-name",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4000,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(64000),
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: large io2 with too high iopsPerGB",
 			volumeName: "vol-test-name",
@@ -1301,6 +1693,26 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone create default volume with throughput",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				Throughput:       250,
+				AvailabilityZone: defaultZone,
+				SourceVolumeID:   "vol-test-id",
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Throughput: aws.Int32(250),
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
+			expErr: nil,
+		},
 		{
 			name:       "success: multi-attach with IO2",
 			volumeName: "vol-test-name",
@@ -1321,6 +1733,29 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: nil,
 		},
+		{
+			name:       "success: clone multi-attach with IO2",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:      util.GiBToBytes(4),
+				Tags:               map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+				VolumeType:         VolumeTypeIO2,
+				MultiAttachEnabled: true,
+				IOPSPerGB:          10000,
+				AvailabilityZone:   defaultZone,
+				SourceVolumeID:     "vol-test-id",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      4,
+				AvailabilityZone: defaultZone,
+			},
+			expCopyVolumesInput: &ec2.CopyVolumesInput{
+				Iops: aws.Int32(2000),
+			},
+
+			expErr: nil,
+		},
 		{
 			name:       "success: create volume from snapshot with initialization rate",
 			volumeName: "vol-test-name",
@@ -1464,6 +1899,29 @@ func TestCreateDisk(t *testing.T) {
 						},
 					}, nil)
 				}
+			} else if tc.expCopyVolumesInput != nil {
+				mockEC2.EXPECT().CopyVolumes(gomock.Any(), gomock.Any(), gomock.Any()).Return(&ec2.CopyVolumesOutput{
+					Volumes: []types.Volume{
+						{
+							VolumeId:         aws.String(tc.diskOptions.Tags[VolumeNameTagKey]),
+							Size:             aws.Int32(util.BytesToGiB(tc.diskOptions.CapacityBytes)),
+							State:            types.VolumeState(volState),
+							AvailabilityZone: aws.String(tc.diskOptions.AvailabilityZone),
+							OutpostArn:       aws.String(tc.diskOptions.OutpostArn),
+						},
+					},
+				}, tc.expCopyVolumesErr)
+				mockEC2.EXPECT().DescribeVolumes(gomock.Any(), gomock.Any()).Return(&ec2.DescribeVolumesOutput{
+					Volumes: []types.Volume{
+						{
+							VolumeId:         aws.String(tc.diskOptions.Tags[VolumeNameTagKey]),
+							Size:             aws.Int32(util.BytesToGiB(tc.diskOptions.CapacityBytes)),
+							State:            types.VolumeState(volState),
+							AvailabilityZone: aws.String(tc.diskOptions.AvailabilityZone),
+							OutpostArn:       aws.String(tc.diskOptions.OutpostArn),
+						},
+					},
+				}, tc.expDescVolumeErr).AnyTimes()
 			}
 
 			if tc.expCreateVolumeInput == nil {
diff --git a/pkg/driver/controller_test.go b/pkg/driver/controller_test.go
index 06b6ee11..1dc4124d 100644
--- a/pkg/driver/controller_test.go
+++ b/pkg/driver/controller_test.go
@@ -47,7 +47,8 @@ const (
 	expInstanceID = "i-123456789abcdef01"
 	expDevicePath = "/dev/xvda"
 
-	testOutpostARN = "arn:aws:outposts:us-west-2:111111111111:outpost/op-0aaa000a0aaaa00a0"
+	testOutpostARN  = "arn:aws:outposts:us-west-2:111111111111:outpost/op-0aaa000a0aaaa00a0"
+	testSourceVolID = "source-volume-id"
 )
 
 func TestCreateVolume(t *testing.T) {
@@ -228,6 +229,368 @@ func TestCreateVolume(t *testing.T) {
 				}
 			},
 		},
+		{
+			name: "clone success KMS key id",
+			testFunc: func(t *testing.T) {
+				t.Helper()
+				req := &csi.CreateVolumeRequest{
+					Name:               "random-vol-name",
+					CapacityRange:      stdCapRange,
+					VolumeCapabilities: stdVolCap,
+					Parameters: map[string]string{
+						EncryptedKey: "true",
+						KmsKeyIDKey:  "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+					},
+					VolumeContentSource: &csi.VolumeContentSource{
+						Type: &csi.VolumeContentSource_Volume{
+							Volume: &csi.VolumeContentSource_VolumeSource{
+								VolumeId: "volume-id",
+							},
+						},
+					},
+				}
+
+				ctx := t.Context()
+
+				mockDisk := &cloud.Disk{
+					VolumeID:         req.GetName(),
+					AvailabilityZone: expZone,
+					CapacityGiB:      util.BytesToGiB(stdVolSize),
+					SourceVolumeID:   testSourceVolID,
+					KmsKeyID:         "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+				}
+
+				mockSourceDisk := &cloud.Disk{
+					VolumeID:         testSourceVolID,
+					AvailabilityZone: expZone,
+					OutpostArn:       "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+					KmsKeyID:         "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+				}
+
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockCloud := cloud.NewMockCloud(mockCtl)
+				mockCloud.EXPECT().CreateDisk(gomock.Eq(ctx), gomock.Eq(req.GetName()), gomock.Any()).Return(mockDisk, nil)
+				mockCloud.EXPECT().GetDiskByID(gomock.Eq(ctx), gomock.Any()).Return(mockSourceDisk, nil)
+				awsDriver := ControllerService{
+					cloud:    mockCloud,
+					inFlight: internal.NewInFlight(),
+					options:  &Options{},
+				}
+
+				rsp, err := awsDriver.CreateVolume(ctx, req)
+				if err != nil {
+					srvErr, ok := status.FromError(err)
+					if !ok {
+						t.Fatalf("Could not get error status code from error: %v", srvErr)
+					}
+					t.Fatalf("Unexpected error: %v", srvErr.Code())
+				}
+
+				sourceVolumeID := ""
+				if rsp.GetVolume() != nil && rsp.GetVolume().GetContentSource() != nil && rsp.GetVolume().GetContentSource().GetVolume() != nil {
+					sourceVolumeID = rsp.GetVolume().GetContentSource().GetVolume().GetVolumeId()
+				}
+				if rsp.GetVolume().GetContentSource().GetVolume().GetVolumeId() != testSourceVolID {
+					t.Errorf("Unexpected source volume: %q", sourceVolumeID)
+				}
+			},
+		},
+		{
+			name: "clone success",
+			testFunc: func(t *testing.T) {
+				t.Helper()
+				req := &csi.CreateVolumeRequest{
+					Name:               "random-vol-name",
+					CapacityRange:      stdCapRange,
+					VolumeCapabilities: stdVolCap,
+					Parameters:         nil,
+					VolumeContentSource: &csi.VolumeContentSource{
+						Type: &csi.VolumeContentSource_Volume{
+							Volume: &csi.VolumeContentSource_VolumeSource{
+								VolumeId: "volume-id",
+							},
+						},
+					},
+				}
+
+				ctx := t.Context()
+
+				mockDisk := &cloud.Disk{
+					VolumeID:         req.GetName(),
+					AvailabilityZone: expZone,
+					CapacityGiB:      util.BytesToGiB(stdVolSize),
+					SourceVolumeID:   testSourceVolID,
+				}
+
+				mockSourceDisk := &cloud.Disk{
+					VolumeID:         testSourceVolID,
+					AvailabilityZone: expZone,
+					OutpostArn:       "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+				}
+
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockCloud := cloud.NewMockCloud(mockCtl)
+				mockCloud.EXPECT().CreateDisk(gomock.Eq(ctx), gomock.Eq(req.GetName()), gomock.Any()).Return(mockDisk, nil)
+				mockCloud.EXPECT().GetDiskByID(gomock.Eq(ctx), gomock.Any()).Return(mockSourceDisk, nil)
+				awsDriver := ControllerService{
+					cloud:    mockCloud,
+					inFlight: internal.NewInFlight(),
+					options:  &Options{},
+				}
+
+				rsp, err := awsDriver.CreateVolume(ctx, req)
+				if err != nil {
+					srvErr, ok := status.FromError(err)
+					if !ok {
+						t.Fatalf("Could not get error status code from error: %v", srvErr)
+					}
+					t.Fatalf("Unexpected error: %v", srvErr.Code())
+				}
+
+				sourceVolumeID := ""
+				if rsp.GetVolume() != nil && rsp.GetVolume().GetContentSource() != nil && rsp.GetVolume().GetContentSource().GetVolume() != nil {
+					sourceVolumeID = rsp.GetVolume().GetContentSource().GetVolume().GetVolumeId()
+				}
+				if rsp.GetVolume().GetContentSource().GetVolume().GetVolumeId() != testSourceVolID {
+					t.Errorf("Unexpected source volume: %q", sourceVolumeID)
+				}
+			},
+		},
+		{
+			name: "clone success AZ and ARN",
+			testFunc: func(t *testing.T) {
+				t.Helper()
+				req := &csi.CreateVolumeRequest{
+					Name:               "random-vol-name",
+					CapacityRange:      stdCapRange,
+					VolumeCapabilities: stdVolCap,
+					Parameters:         nil,
+					AccessibilityRequirements: &csi.TopologyRequirement{
+						Preferred: []*csi.Topology{{}},
+						Requisite: []*csi.Topology{
+							{
+								Segments: map[string]string{
+									ZoneTopologyKey:          expZone,
+									WellKnownZoneTopologyKey: expZone,
+									AwsAccountIDKey:          "222222222222",
+									AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+									AwsRegionKey:             "us-west-2",
+									AwsPartitionKey:          "aws",
+								},
+							},
+						},
+					},
+					VolumeContentSource: &csi.VolumeContentSource{
+						Type: &csi.VolumeContentSource_Volume{
+							Volume: &csi.VolumeContentSource_VolumeSource{
+								VolumeId: "volume-id",
+							},
+						},
+					},
+				}
+
+				ctx := t.Context()
+
+				mockDisk := &cloud.Disk{
+					VolumeID:         req.GetName(),
+					AvailabilityZone: expZone,
+					CapacityGiB:      util.BytesToGiB(stdVolSize),
+					SourceVolumeID:   testSourceVolID,
+				}
+
+				mockSourceDisk := &cloud.Disk{
+					VolumeID:         testSourceVolID,
+					AvailabilityZone: expZone,
+					OutpostArn:       "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+				}
+
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockCloud := cloud.NewMockCloud(mockCtl)
+				mockCloud.EXPECT().CreateDisk(gomock.Eq(ctx), gomock.Eq(req.GetName()), gomock.Any()).Return(mockDisk, nil)
+				mockCloud.EXPECT().GetDiskByID(gomock.Eq(ctx), gomock.Any()).Return(mockSourceDisk, nil)
+				awsDriver := ControllerService{
+					cloud:    mockCloud,
+					inFlight: internal.NewInFlight(),
+					options:  &Options{},
+				}
+
+				rsp, err := awsDriver.CreateVolume(ctx, req)
+				if err != nil {
+					srvErr, ok := status.FromError(err)
+					if !ok {
+						t.Fatalf("Could not get error status code from error: %v", srvErr)
+					}
+					t.Fatalf("Unexpected error: %v", srvErr.Code())
+				}
+
+				sourceVolumeID := ""
+				if rsp.GetVolume() != nil && rsp.GetVolume().GetContentSource() != nil && rsp.GetVolume().GetContentSource().GetVolume() != nil {
+					sourceVolumeID = rsp.GetVolume().GetContentSource().GetVolume().GetVolumeId()
+				}
+				if rsp.GetVolume().GetContentSource().GetVolume().GetVolumeId() != testSourceVolID {
+					t.Errorf("Unexpected source volume: %q", sourceVolumeID)
+				}
+			},
+		},
+		{
+			name: "clone fail: different AZ than source",
+			testFunc: func(t *testing.T) {
+				t.Helper()
+				req := &csi.CreateVolumeRequest{
+					Name:               "random-vol-name",
+					CapacityRange:      stdCapRange,
+					VolumeCapabilities: stdVolCap,
+					Parameters:         nil,
+					AccessibilityRequirements: &csi.TopologyRequirement{
+						Preferred: []*csi.Topology{{}},
+						Requisite: []*csi.Topology{
+							{
+								Segments: map[string]string{ZoneTopologyKey: "us-west-1c", WellKnownZoneTopologyKey: "us-west-1b"},
+							},
+						},
+					},
+					VolumeContentSource: &csi.VolumeContentSource{
+						Type: &csi.VolumeContentSource_Volume{
+							Volume: &csi.VolumeContentSource_VolumeSource{
+								VolumeId: "volume-id",
+							},
+						},
+					},
+				}
+
+				ctx := t.Context()
+
+				mockSourceDisk := &cloud.Disk{
+					VolumeID:         testSourceVolID,
+					AvailabilityZone: "us-east-1a",
+					OutpostArn:       "arn:aws:outposts:us-east-1:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+				}
+
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockCloud := cloud.NewMockCloud(mockCtl)
+				mockCloud.EXPECT().GetDiskByID(gomock.Eq(ctx), gomock.Any()).Return(mockSourceDisk, nil)
+				awsDriver := ControllerService{
+					cloud:    mockCloud,
+					inFlight: internal.NewInFlight(),
+					options:  &Options{},
+				}
+				_, err := awsDriver.CreateVolume(ctx, req)
+				if status.Code(err) != codes.ResourceExhausted {
+					t.Fatalf("failed expected ResourceExhausted error but got %v", err)
+				}
+			},
+		},
+		{
+			name: "clone fail: correct AZ different outpost",
+			testFunc: func(t *testing.T) {
+				t.Helper()
+				req := &csi.CreateVolumeRequest{
+					Name:               "random-vol-name",
+					CapacityRange:      stdCapRange,
+					VolumeCapabilities: stdVolCap,
+					Parameters:         nil,
+					AccessibilityRequirements: &csi.TopologyRequirement{
+						Preferred: []*csi.Topology{{}},
+						Requisite: []*csi.Topology{
+							{
+								Segments: map[string]string{
+									ZoneTopologyKey:          expZone,
+									WellKnownZoneTopologyKey: expZone,
+									AwsAccountIDKey:          "222222222222",
+									AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+									AwsRegionKey:             "us-west-2",
+									AwsPartitionKey:          "diff",
+								},
+							},
+						},
+					},
+					VolumeContentSource: &csi.VolumeContentSource{
+						Type: &csi.VolumeContentSource_Volume{
+							Volume: &csi.VolumeContentSource_VolumeSource{
+								VolumeId: "volume-id",
+							},
+						},
+					},
+				}
+
+				ctx := t.Context()
+
+				mockSourceDisk := &cloud.Disk{
+					VolumeID:         testSourceVolID,
+					AvailabilityZone: expZone,
+					OutpostArn:       "arn:aws:outposts:us-east-1:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+				}
+
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockCloud := cloud.NewMockCloud(mockCtl)
+				mockCloud.EXPECT().GetDiskByID(gomock.Eq(ctx), gomock.Any()).Return(mockSourceDisk, nil)
+				awsDriver := ControllerService{
+					cloud:    mockCloud,
+					inFlight: internal.NewInFlight(),
+					options:  &Options{},
+				}
+				_, err := awsDriver.CreateVolume(ctx, req)
+				if status.Code(err) != codes.ResourceExhausted {
+					t.Fatalf("failed expected ResourceExhausted error but got %v", err)
+				}
+			},
+		},
+		{
+			name: "clone fail: different KMS key than source",
+			testFunc: func(t *testing.T) {
+				t.Helper()
+				req := &csi.CreateVolumeRequest{
+					Name:               "random-vol-name",
+					CapacityRange:      stdCapRange,
+					VolumeCapabilities: stdVolCap,
+					Parameters: map[string]string{
+						EncryptedKey: "true",
+						KmsKeyIDKey:  "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+					},
+					VolumeContentSource: &csi.VolumeContentSource{
+						Type: &csi.VolumeContentSource_Volume{
+							Volume: &csi.VolumeContentSource_VolumeSource{
+								VolumeId: "volume-id",
+							},
+						},
+					},
+				}
+
+				ctx := t.Context()
+
+				mockSourceDisk := &cloud.Disk{
+					VolumeID:         testSourceVolID,
+					AvailabilityZone: expZone,
+					OutpostArn:       "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+					KmsKeyID:         "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cddiff",
+				}
+
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockCloud := cloud.NewMockCloud(mockCtl)
+				mockCloud.EXPECT().GetDiskByID(gomock.Eq(ctx), gomock.Any()).Return(mockSourceDisk, nil)
+				awsDriver := ControllerService{
+					cloud:    mockCloud,
+					inFlight: internal.NewInFlight(),
+					options:  &Options{},
+				}
+				_, err := awsDriver.CreateVolume(ctx, req)
+				if status.Code(err) != codes.InvalidArgument {
+					t.Fatalf("failed expected InvalidArgument error but got %v", err)
+				}
+			},
+		},
 		{
 			name: "success AZ-ID",
 			testFunc: func(t *testing.T) {
@@ -2521,6 +2884,263 @@ func TestDeleteVolume(t *testing.T) {
 	}
 }
 
+func TestCheckSourceTopology(t *testing.T) {
+	testCases := []struct {
+		name                   string
+		requirement            *csi.TopologyRequirement
+		sourceVolumeZone       string
+		sourceVolumeOutpostArn string
+		sourceVolumeZoneID     string
+		expErr                 bool
+	}{
+		{
+			name:                   "no requirement",
+			requirement:            &csi.TopologyRequirement{},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 false,
+		},
+		{
+			name: "matching AZ and outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          expZone,
+							WellKnownZoneTopologyKey: expZone,
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "aws",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 false,
+		},
+		{
+			name: "matching AZ matching outpostARN only on WellKnownZoneTopologyKey",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          "us-east-1b",
+							WellKnownZoneTopologyKey: expZone,
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "aws",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 false,
+		},
+		{
+			name: "matching AZ matching outpostARN only on ZoneTopologyKey",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          expZone,
+							WellKnownZoneTopologyKey: "us-east-1a",
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "aws",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 false,
+		},
+		{
+			name: "matching AZ wrong outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          expZone,
+							WellKnownZoneTopologyKey: expZone,
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "diff",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 true,
+		},
+		{
+			name: "wrong AZ matching outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          "us-east-1b",
+							WellKnownZoneTopologyKey: "us-east-1a",
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "aws",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 true,
+		},
+		{
+			name: "wrong AZ wrong outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          "us-east-1b",
+							WellKnownZoneTopologyKey: "us-east-1a",
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "diff",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 true,
+		},
+		{
+			name: "matching AZ wrong outpostARN but only preferred",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneTopologyKey:          expZone,
+							WellKnownZoneTopologyKey: expZone,
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "diff",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZone,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 false,
+		},
+		{
+			name: "matching AZ-ID and outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneIDTopologyKey: expZoneID,
+							AwsAccountIDKey:   "222222222222",
+							AwsOutpostIDKey:   "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:      "us-west-2",
+							AwsPartitionKey:   "aws",
+						},
+					},
+				},
+			},
+			sourceVolumeZoneID:     expZoneID,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 false,
+		},
+		{
+			name: "matching AZ-ID wrong outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneIDTopologyKey: expZoneID,
+							AwsAccountIDKey:   "222222222222",
+							AwsOutpostIDKey:   "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:      "us-west-2",
+							AwsPartitionKey:   "diff",
+						},
+					},
+				},
+			},
+			sourceVolumeZoneID:     expZoneID,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 true,
+		},
+		{
+			name: "wrong AZ-ID matching outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneIDTopologyKey: "usw2-az1",
+							AwsAccountIDKey:   "222222222222",
+							AwsOutpostIDKey:   "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:      "us-west-2",
+							AwsPartitionKey:   "aws",
+						},
+					},
+				},
+			},
+			sourceVolumeZone:       expZoneID,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 true,
+		},
+		{
+			name: "wrong AZ-ID wrong outpostARN",
+			requirement: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{{}},
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							ZoneIDTopologyKey:        "usw2-az1",
+							WellKnownZoneTopologyKey: "us-east-1a",
+							AwsAccountIDKey:          "222222222222",
+							AwsOutpostIDKey:          "aa-aaaaaaaaaaaaaaaaa",
+							AwsRegionKey:             "us-west-2",
+							AwsPartitionKey:          "diff",
+						},
+					},
+				},
+			},
+			sourceVolumeZoneID:     expZoneID,
+			sourceVolumeOutpostArn: "arn:aws:outposts:us-west-2:222222222222:outpost/aa-aaaaaaaaaaaaaaaaa",
+			expErr:                 true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := checkSourceTopology(tc.requirement, tc.sourceVolumeZone, tc.sourceVolumeOutpostArn, tc.sourceVolumeZoneID)
+			if err != nil && !tc.expErr {
+				t.Fatalf("Unexpected error: %v", err)
+			} else if tc.expErr && status.Code(err) != codes.ResourceExhausted {
+				t.Fatalf("Incorrect error code expected ResourceExhausted (8) but got : %v", err)
+			}
+		})
+	}
+}
+
 func TestPickAvailabilityZone(t *testing.T) {
 	testCases := []struct {
 		name        string
diff --git a/tests/e2e-kubernetes/manifests.yaml b/tests/e2e-kubernetes/manifests.yaml
index a1edb9f7..56b59a62 100644
--- a/tests/e2e-kubernetes/manifests.yaml
+++ b/tests/e2e-kubernetes/manifests.yaml
@@ -41,7 +41,7 @@ DriverInfo:
     block: true
     exec: true
     snapshotDataSource: true
-    pvcDataSource: false
+    pvcDataSource: true
     multipods: true
     controllerExpansion: true
     nodeExpansion: true
diff --git a/tests/e2e-kubernetes/storageclass.yaml b/tests/e2e-kubernetes/storageclass.yaml
index 332da571..02c8457b 100644
--- a/tests/e2e-kubernetes/storageclass.yaml
+++ b/tests/e2e-kubernetes/storageclass.yaml
@@ -18,3 +18,5 @@ metadata:
   name: ebs.csi.aws.com
 provisioner: ebs.csi.aws.com
 volumeBindingMode: WaitForFirstConsumer
+parameters:
+  encrypted: "true"
\ No newline at end of file
diff --git a/tests/e2e/go.mod b/tests/e2e/go.mod
index 66dfab76..d0a113e9 100644
--- a/tests/e2e/go.mod
+++ b/tests/e2e/go.mod
@@ -3,7 +3,7 @@ module github.com/kubernetes-sigs/aws-ebs-csi-driver/tests/e2e
 require (
 	github.com/aws/aws-sdk-go-v2 v1.39.2
 	github.com/aws/aws-sdk-go-v2/config v1.31.12
-	github.com/aws/aws-sdk-go-v2/service/ec2 v1.254.1
+	github.com/aws/aws-sdk-go-v2/service/ec2 v1.256.0
 	github.com/google/uuid v1.6.0
 	github.com/kubernetes-csi/external-snapshotter/client/v4 v4.2.0
 	github.com/kubernetes-sigs/aws-ebs-csi-driver v1.50.1
@@ -74,7 +74,7 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kubernetes-csi/csi-proxy/client v1.3.0 // indirect
 	github.com/kubernetes-csi/csi-proxy/v2 v2.0.0-alpha.2 // indirect
-	github.com/microsoft/wmi v0.37.0 // indirect
+	github.com/microsoft/wmi v0.38.1 // indirect
 	github.com/moby/spdystream v0.5.0 // indirect
 	github.com/moby/sys/mountinfo v0.7.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -117,8 +117,8 @@ require (
 	golang.org/x/text v0.30.0 // indirect
 	golang.org/x/time v0.14.0 // indirect
 	golang.org/x/tools v0.38.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20251007200510-49b9836ed3ff // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20251014184007-4626949a642f // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251014184007-4626949a642f // indirect
 	google.golang.org/grpc v1.76.0 // indirect
 	google.golang.org/protobuf v1.36.10 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
diff --git a/tests/e2e/go.sum b/tests/e2e/go.sum
index 0ab3ed3a..9587973f 100644
--- a/tests/e2e/go.sum
+++ b/tests/e2e/go.sum
@@ -28,8 +28,8 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.9 h1:6RBnKZLkJM4hQ+kN6E7
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.9/go.mod h1:V9rQKRmK7AWuEsOMnHzKj8WyrIir1yUJbZxDuZLFvXI=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
-github.com/aws/aws-sdk-go-v2/service/ec2 v1.254.1 h1:7p9bJCZ/b3EJXXARW7JMEs2IhsnI4YFHpfXQfgMh0eg=
-github.com/aws/aws-sdk-go-v2/service/ec2 v1.254.1/go.mod h1:M8WWWIfXmxA4RgTXcI/5cSByxRqjgne32Sh0VIbrn0A=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.256.0 h1:PLIiJRPfUKlgRazuO+hFuNGU2E2IHRqRW/nq4WzG9r4=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.256.0/go.mod h1:M8WWWIfXmxA4RgTXcI/5cSByxRqjgne32Sh0VIbrn0A=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1 h1:oegbebPEMA/1Jny7kvwejowCaHz1FWZAQ94WXFNCyTM=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1/go.mod h1:kemo5Myr9ac0U9JfSjMo9yHLtw+pECEHsFtJ9tqCEI8=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.9 h1:5r34CgVOD4WZudeEKZ9/iKpiT6cM1JyEROpXjOcdWv8=
@@ -246,8 +246,8 @@ github.com/maruel/natural v1.1.1 h1:Hja7XhhmvEFhcByqDoHz9QZbkWey+COd9xWfCfn1ioo=
 github.com/maruel/natural v1.1.1/go.mod h1:v+Rfd79xlw1AgVBjbO0BEQmptqb5HvL/k9GRHB7ZKEg=
 github.com/mfridman/tparse v0.18.0 h1:wh6dzOKaIwkUGyKgOntDW4liXSo37qg5AXbIhkMV3vE=
 github.com/mfridman/tparse v0.18.0/go.mod h1:gEvqZTuCgEhPbYk/2lS3Kcxg1GmTxxU7kTC8DvP0i/A=
-github.com/microsoft/wmi v0.37.0 h1:TQ0+JHz8ogtl6twPOOQXA2ON2ErUS8WyInSq+yY4AT4=
-github.com/microsoft/wmi v0.37.0/go.mod h1:XF+cfluA15xGnSCYkJIYuj2vWzdm2YrNuvqlC+baWY0=
+github.com/microsoft/wmi v0.38.1 h1:j+pZjiVnbDEnVqLa3f0VxFouQGe3SMYgi1BAUhI9PXE=
+github.com/microsoft/wmi v0.38.1/go.mod h1:XF+cfluA15xGnSCYkJIYuj2vWzdm2YrNuvqlC+baWY0=
 github.com/moby/spdystream v0.5.0 h1:7r0J1Si3QO/kjRitvSLVVFUjxMEb/YLj6S9FF62JBCU=
 github.com/moby/spdystream v0.5.0/go.mod h1:xBAYlnt/ay+11ShkdFKNAG7LsyK/tmNBVvVOwrfMgdI=
 github.com/moby/sys/mountinfo v0.7.2 h1:1shs6aH5s4o5H2zQLn796ADW1wMrIwHsyJ2v9KouLrg=
@@ -683,10 +683,10 @@ google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
-google.golang.org/genproto/googleapis/api v0.0.0-20251007200510-49b9836ed3ff h1:8Zg5TdmcbU8A7CXGjGXF1Slqu/nIFCRaR3S5gT2plIA=
-google.golang.org/genproto/googleapis/api v0.0.0-20251007200510-49b9836ed3ff/go.mod h1:dbWfpVPvW/RqafStmRWBUpMN14puDezDMHxNYiRfQu0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff h1:A90eA31Wq6HOMIQlLfzFwzqGKBTuaVztYu/g8sn+8Zc=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+google.golang.org/genproto/googleapis/api v0.0.0-20251014184007-4626949a642f h1:OiFuztEyBivVKDvguQJYWq1yDcfAHIID/FVrPR4oiI0=
+google.golang.org/genproto/googleapis/api v0.0.0-20251014184007-4626949a642f/go.mod h1:kprOiu9Tr0JYyD6DORrc4Hfyk3RFXqkQ3ctHEum3ZbM=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251014184007-4626949a642f h1:1FTH6cpXFsENbPR5Bu8NQddPSaUUE6NA2XdZdDSAJK4=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251014184007-4626949a642f/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
diff --git a/tests/e2e/requires_aws_api.go b/tests/e2e/requires_aws_api.go
index 2ffc6248..18c6db6e 100644
--- a/tests/e2e/requires_aws_api.go
+++ b/tests/e2e/requires_aws_api.go
@@ -351,6 +351,163 @@ var _ = Describe("[ebs-csi-e2e] [single-az] [requires-aws-api] Dynamic Provision
 		}
 		test.Run(cs, snapshotrcs, ns)
 	})
+	It("should copy a volume with different volume parameters", func() {
+		pod := testsuites.PodDetails{
+			Cmd: testsuites.PodCmdWriteToVolume("/mnt/test-1"),
+			Volumes: []testsuites.VolumeDetails{
+				{
+					CreateVolumeParameters: map[string]string{
+						ebscsidriver.EncryptedKey:  "true",
+						ebscsidriver.VolumeTypeKey: awscloud.VolumeTypeGP2,
+						ebscsidriver.FSTypeKey:     ebscsidriver.FSTypeExt4,
+					},
+					ClaimSize:   driver.MinimumSizeForVolumeType(awscloud.VolumeTypeIO2),
+					VolumeMount: testsuites.DefaultGeneratedVolumeMount,
+				},
+			},
+		}
+		clonedPod := testsuites.PodDetails{
+			Cmd: testsuites.PodCmdWriteToVolume("/mnt/test-1"),
+			Volumes: []testsuites.VolumeDetails{
+				{
+					CreateVolumeParameters: map[string]string{
+						ebscsidriver.EncryptedKey:  "true",
+						ebscsidriver.VolumeTypeKey: awscloud.VolumeTypeIO2,
+						ebscsidriver.IopsKey:       testsuites.DefaultIopsIoVolumes,
+					},
+					ClaimSize:   driver.MinimumSizeForVolumeType(awscloud.VolumeTypeIO2),
+					VolumeMount: testsuites.DefaultGeneratedVolumeMount,
+				},
+			},
+		}
+		test := testsuites.DynamicallyProvisionedCopyVolumeTest{
+			CSIDriver: ebsDriver,
+			Pod:       pod,
+			ClonedPod: clonedPod,
+			ValidateFunc: func() {
+				result, err := ec2Client.DescribeVolumes(context.Background(), &ec2.DescribeVolumesInput{
+					VolumeIds: []string{clonedPod.Volumes[0].VolumeID},
+				})
+				if err != nil {
+					Fail(fmt.Sprintf("failed to describe volume: %v", err))
+				}
+
+				if len(result.Volumes) != 1 {
+					Fail(fmt.Sprintf("expected 1 volume, got %d", len(result.Volumes)))
+				}
+
+				if result.Volumes[0].VolumeType != awscloud.VolumeTypeIO1 {
+					Fail(fmt.Sprintf("expected volume type %s, got %s", awscloud.VolumeTypeIO1, result.Volumes[0].VolumeType))
+				}
+			},
+		}
+		test.Run(cs, ns)
+	})
+	// This is a test to ensure driver logic is handled correctly when all params are omitted.
+	It("should copy a volume with all omitted params and not get same IOPS as source ", func() {
+		sourceIops := "3012"
+		pod := testsuites.PodDetails{
+			Cmd: testsuites.PodCmdWriteToVolume("/mnt/test-1"),
+			Volumes: []testsuites.VolumeDetails{
+				{
+					CreateVolumeParameters: map[string]string{
+						ebscsidriver.EncryptedKey:  "true",
+						ebscsidriver.VolumeTypeKey: awscloud.VolumeTypeGP2,
+						ebscsidriver.IopsKey:       sourceIops,
+						ebscsidriver.FSTypeKey:     ebscsidriver.FSTypeExt4,
+					},
+					ClaimSize:   driver.MinimumSizeForVolumeType(awscloud.VolumeTypeGP3),
+					VolumeMount: testsuites.DefaultGeneratedVolumeMount,
+				},
+			},
+		}
+		clonedPod := testsuites.PodDetails{
+			Cmd: testsuites.PodCmdWriteToVolume("/mnt/test-1"),
+			Volumes: []testsuites.VolumeDetails{
+				{
+					CreateVolumeParameters: map[string]string{
+						ebscsidriver.VolumeTypeKey: awscloud.VolumeTypeGP3,
+					},
+					ClaimSize:   driver.MinimumSizeForVolumeType(awscloud.VolumeTypeGP3),
+					VolumeMount: testsuites.DefaultGeneratedVolumeMount,
+				},
+			},
+		}
+		test := testsuites.DynamicallyProvisionedCopyVolumeTest{
+			CSIDriver: ebsDriver,
+			Pod:       pod,
+			ClonedPod: clonedPod,
+			ValidateFunc: func() {
+				result, err := ec2Client.DescribeVolumes(context.Background(), &ec2.DescribeVolumesInput{
+					VolumeIds: []string{clonedPod.Volumes[0].VolumeID},
+				})
+				if err != nil {
+					Fail(fmt.Sprintf("failed to describe volume: %v", err))
+				}
+
+				if len(result.Volumes) != 1 {
+					Fail(fmt.Sprintf("expected 1 volume, got %d", len(result.Volumes)))
+				}
+
+				if fmt.Sprintf("%d", result.Volumes[0].Iops) == sourceIops {
+					Fail(fmt.Sprintf("expected volume iops %d, to not be same as source %s", result.Volumes[0].Iops, sourceIops))
+				}
+			},
+		}
+		test.Run(cs, ns)
+	})
+	It("should copy a volume to a bigger volume", func() {
+		cloneClaimSize := "3Gi"
+		pod := testsuites.PodDetails{
+			Cmd: testsuites.PodCmdWriteToVolume("/mnt/test-1"),
+			Volumes: []testsuites.VolumeDetails{
+				{
+					CreateVolumeParameters: map[string]string{
+						ebscsidriver.EncryptedKey:  "true",
+						ebscsidriver.VolumeTypeKey: awscloud.VolumeTypeGP2,
+						ebscsidriver.FSTypeKey:     ebscsidriver.FSTypeExt4,
+					},
+					ClaimSize:   driver.MinimumSizeForVolumeType(awscloud.VolumeTypeGP2),
+					VolumeMount: testsuites.DefaultGeneratedVolumeMount,
+				},
+			},
+		}
+		clonedPod := testsuites.PodDetails{
+			Cmd: testsuites.PodCmdWriteToVolume("/mnt/test-1"),
+			Volumes: []testsuites.VolumeDetails{
+				{
+					CreateVolumeParameters: map[string]string{
+						ebscsidriver.VolumeTypeKey: awscloud.VolumeTypeGP2,
+						ebscsidriver.FSTypeKey:     ebscsidriver.FSTypeExt4,
+					},
+					ClaimSize:   cloneClaimSize,
+					VolumeMount: testsuites.DefaultGeneratedVolumeMount,
+				},
+			},
+		}
+		test := testsuites.DynamicallyProvisionedCopyVolumeTest{
+			CSIDriver: ebsDriver,
+			Pod:       pod,
+			ClonedPod: clonedPod,
+			ValidateFunc: func() {
+				result, err := ec2Client.DescribeVolumes(context.Background(), &ec2.DescribeVolumesInput{
+					VolumeIds: []string{clonedPod.Volumes[0].VolumeID},
+				})
+				if err != nil {
+					Fail(fmt.Sprintf("failed to describe volume: %v", err))
+				}
+
+				if len(result.Volumes) != 1 {
+					Fail(fmt.Sprintf("expected 1 volume, got %d", len(result.Volumes)))
+				}
+
+				if fmt.Sprintf("%dGi", result.Volumes[0].Size) != cloneClaimSize {
+					Fail(fmt.Sprintf("expected volume size to be %s, got %d", cloneClaimSize, result.Volumes[0].Size))
+				}
+			},
+		}
+		test.Run(cs, ns)
+	})
 
 	It("should validate GP3 volume IOPS are not capped at 16000 when requesting 16001", func() {
 		testTag := generateTagName()
diff --git a/tests/e2e/testsuites/dynamically_provisioned_copy_volume_tester.go b/tests/e2e/testsuites/dynamically_provisioned_copy_volume_tester.go
new file mode 100644
index 00000000..3a43e994
--- /dev/null
+++ b/tests/e2e/testsuites/dynamically_provisioned_copy_volume_tester.go
@@ -0,0 +1,71 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+   http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testsuites
+
+import (
+	"github.com/kubernetes-sigs/aws-ebs-csi-driver/tests/e2e/driver"
+	. "github.com/onsi/ginkgo/v2"
+	v1 "k8s.io/api/core/v1"
+	clientset "k8s.io/client-go/kubernetes"
+)
+
+// DynamicallyProvisionedCopyVolumeTest will provision required StorageClass(es), PVC(s) and Pod(s)
+// Waiting for the PV provisioner to create a new PV
+// Testing if the Pod(s) can write and read to mounted volumes
+// Create a copy of the volume using PVC as data source, validate the data is copied, and then write and read to it again
+// And finally delete the snapshot
+// This test only supports a single volume.
+type DynamicallyProvisionedCopyVolumeTest struct {
+	CSIDriver    driver.PVTestDriver
+	Pod          PodDetails
+	ClonedPod    PodDetails
+	ValidateFunc func()
+}
+
+func (t *DynamicallyProvisionedCopyVolumeTest) Run(client clientset.Interface, namespace *v1.Namespace) {
+	tpod := NewTestPod(client, namespace, t.Pod.Cmd)
+	volume := t.Pod.Volumes[0]
+	tpvc, pvcCleanup := volume.SetupDynamicPersistentVolumeClaim(client, namespace, t.CSIDriver)
+	for i := range pvcCleanup {
+		defer pvcCleanup[i]()
+	}
+	tpod.SetupVolume(tpvc.persistentVolumeClaim, volume.VolumeMount.NameGenerate+"1", volume.VolumeMount.MountPathGenerate+"1", volume.VolumeMount.ReadOnly)
+
+	By("deploying the pod")
+	tpod.Create()
+	defer tpod.Cleanup()
+	By("checking that the pods command exits with no error")
+	tpod.WaitForSuccess()
+
+	By("creating a clone of the source volume")
+	t.ClonedPod.Volumes[0].DataSource = &DataSource{
+		Name: tpvc.persistentVolumeClaim.Name,
+		Kind: PersistentVolumeClaimKind,
+	}
+
+	tcpod := NewTestPod(client, namespace, t.ClonedPod.Cmd)
+	cvolume := t.ClonedPod.Volumes[0]
+	tcpvc, cpvcCleanup := cvolume.SetupDynamicPersistentVolumeClaim(client, namespace, t.CSIDriver)
+	for i := range cpvcCleanup {
+		defer cpvcCleanup[i]()
+	}
+	tcpod.SetupVolume(tcpvc.persistentVolumeClaim, cvolume.VolumeMount.NameGenerate+"1", cvolume.VolumeMount.MountPathGenerate+"1", cvolume.VolumeMount.ReadOnly)
+
+	By("deploying a second pod with a volume cloned from the original")
+	tcpod.Create()
+	defer tcpod.Cleanup()
+	By("checking that the pods command exits with no error")
+	tcpod.WaitForSuccess()
+}
diff --git a/tests/e2e/testsuites/dynamically_provisioned_volume_snapshot_tester.go b/tests/e2e/testsuites/dynamically_provisioned_volume_snapshot_tester.go
index 1906d7a3..7df68c45 100644
--- a/tests/e2e/testsuites/dynamically_provisioned_volume_snapshot_tester.go
+++ b/tests/e2e/testsuites/dynamically_provisioned_volume_snapshot_tester.go
@@ -60,7 +60,7 @@ func (t *DynamicallyProvisionedVolumeSnapshotTest) Run(client clientset.Interfac
 	defer tvsc.DeleteSnapshot(snapshot)
 	tvsc.ReadyToUse(snapshot)
 
-	t.RestoredPod.Volumes[0].DataSource = &DataSource{Name: snapshot.Name}
+	t.RestoredPod.Volumes[0].DataSource = &DataSource{Name: snapshot.Name, Kind: VolumeSnapshotKind}
 	trpod := NewTestPod(client, namespace, t.RestoredPod.Cmd)
 	rvolume := t.RestoredPod.Volumes[0]
 	trpvc, rpvcCleanup := rvolume.SetupDynamicPersistentVolumeClaim(client, namespace, t.CSIDriver)
diff --git a/tests/e2e/testsuites/specs.go b/tests/e2e/testsuites/specs.go
index 8aa03d72..234e93bb 100644
--- a/tests/e2e/testsuites/specs.go
+++ b/tests/e2e/testsuites/specs.go
@@ -56,6 +56,7 @@ const (
 
 const (
 	VolumeSnapshotKind        = "VolumeSnapshot"
+	PersistentVolumeClaimKind = "PersistentVolumeClaim"
 	VolumeSnapshotContentKind = "VolumeSnapshotContent"
 	SnapshotAPIVersion        = "snapshot.storage.k8s.io/v1"
 	APIVersionv1              = "v1"
@@ -78,6 +79,7 @@ type VolumeDeviceDetails struct {
 
 type DataSource struct {
 	Name string
+	Kind string
 }
 
 func (pod *PodDetails) SetupWithDynamicVolumes(client clientset.Interface, namespace *v1.Namespace, csiDriver driver.DynamicPVTestDriver) (*TestPod, []func()) {
@@ -145,9 +147,11 @@ func (volume *VolumeDetails) SetupDynamicPersistentVolumeClaim(client clientset.
 	var tpvc *TestPersistentVolumeClaim
 	if volume.DataSource != nil {
 		dataSource := &v1.TypedLocalObjectReference{
-			Name:     volume.DataSource.Name,
-			Kind:     VolumeSnapshotKind,
-			APIGroup: &SnapshotAPIGroup,
+			Name: volume.DataSource.Name,
+			Kind: volume.DataSource.Kind,
+		}
+		if volume.DataSource.Kind == VolumeSnapshotKind {
+			dataSource.APIGroup = &SnapshotAPIGroup
 		}
 		tpvc = NewTestPersistentVolumeClaimWithDataSource(client, namespace, volume.ClaimSize, volume.VolumeMode, &createdStorageClass, dataSource, volume.AccessMode)
 	} else {