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

Issue:
EBS volume created but not linked to PV/PVC
/kind bug

**What happened?**

Provisioning an io2 volume close to the limit for `Storage for Provisioned IOPS SSD (io2) volumes, in TiB` (`L-09BD8365`) causes 

```
E0605 06:26:48.679755       1 handlers.go:85] "Error from AWS API" err="api
error VolumeLimitExceeded: You have exceeded your maximum io2 storage limit of
30 TiB in this region. Please contact AWS Support to request an Elastic Block
Store service limit increase."
```

We got this error for 2 days while waiting for a quota increase. Once it was raised to 60TiB, the controller didn't fail anymore and could link the EBS volume to the PV/PVC correctly.

**What you expected to happen?**

The controller should calculate the limits correctly and fail robustly. 

**How to reproduce it (as minimally and precisely as possible)?**

Since I cannot arbitrarily set the quota for L-09BD8365, I will just list what we had done:

- Have the io2 TiB limit at `30TiB` for the account
- Provision the following io2 volumes: 
  - 1x 900GiB
  - 3x 1100GiB
  - 2x 7700GiB
  This leads to a utilization of ~19TiB
- Try to provision another 7700GiB io2 volume
- Find the error in the events of the PVC or the controller logs

**Anything else we need to know?**:

<details>

<summary>Relevant logs:</summary>

```
2025-06-05 08:26:06.431 I0605 06:26:06.431598       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Normal" reason="Provisioning" message="External provisioner is provisioning volume for claim \"history-infra/history-cnpg-15\""
2025-06-05 08:26:06.449 I0605 06:26:06.449257       1 controller.go:951] "Retrying syncing claim" key="c8f34cdd-edc1-4f77-8d9f-59bf73365069" failures=0
2025-06-05 08:26:06.449 E0605 06:26:06.449306       1 controller.go:974] "Unhandled Error" err="error syncing claim \"c8f34cdd-edc1-4f77-8d9f-59bf73365069\": failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal" logger="UnhandledError"
2025-06-05 08:26:06.449 I0605 06:26:06.449327       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Warning" reason="ProvisioningFailed" message="failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal"
2025-06-05 08:26:07.450 I0605 06:26:07.450324       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Normal" reason="Provisioning" message="External provisioner is provisioning volume for claim \"history-infra/history-cnpg-15\""
2025-06-05 08:26:07.462 I0605 06:26:07.462566       1 controller.go:951] "Retrying syncing claim" key="c8f34cdd-edc1-4f77-8d9f-59bf73365069" failures=1
2025-06-05 08:26:07.462 E0605 06:26:07.462594       1 controller.go:974] "Unhandled Error" err="error syncing claim \"c8f34cdd-edc1-4f77-8d9f-59bf73365069\": failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal" logger="UnhandledError"
2025-06-05 08:26:07.462 I0605 06:26:07.462639       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Warning" reason="ProvisioningFailed" message="failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal"
2025-06-05 08:26:09.463 I0605 06:26:09.463468       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Normal" reason="Provisioning" message="External provisioner is provisioning volume for claim \"history-infra/history-cnpg-15\""
2025-06-05 08:26:09.477 I0605 06:26:09.477309       1 controller.go:951] "Retrying syncing claim" key="c8f34cdd-edc1-4f77-8d9f-59bf73365069" failures=2
2025-06-05 08:26:09.477 E0605 06:26:09.477342       1 controller.go:974] "Unhandled Error" err="error syncing claim \"c8f34cdd-edc1-4f77-8d9f-59bf73365069\": failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal" logger="UnhandledError"
2025-06-05 08:26:09.477 I0605 06:26:09.477366       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Warning" reason="ProvisioningFailed" message="failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal"
2025-06-05 08:26:13.478 I0605 06:26:13.478427       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Normal" reason="Provisioning" message="External provisioner is provisioning volume for claim \"history-infra/history-cnpg-15\""
2025-06-05 08:26:13.489 I0605 06:26:13.489602       1 controller.go:951] "Retrying syncing claim" key="c8f34cdd-edc1-4f77-8d9f-59bf73365069" failures=3
2025-06-05 08:26:13.489 E0605 06:26:13.489633       1 controller.go:974] "Unhandled Error" err="error syncing claim \"c8f34cdd-edc1-4f77-8d9f-59bf73365069\": failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal" logger="UnhandledError"
2025-06-05 08:26:13.489 I0605 06:26:13.489685       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Warning" reason="ProvisioningFailed" message="failed to provision volume with StorageClass \"cnpg-io2\": error generating accessibility requirements: no topology key found on CSINode ip-10-252-48-139.eu-central-1.compute.internal"
2025-06-05 08:26:21.490 I0605 06:26:21.490638       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Normal" reason="Provisioning" message="External provisioner is provisioning volume for claim \"history-infra/history-cnpg-15\""
2025-06-05 08:26:32.594 E0605 06:26:32.594502       1 driver.go:108] "GRPC error" err="rpc error: code = Internal desc = Could not create volume \"pvc-c8f34cdd-edc1-4f77-8d9f-59bf73365069\": timed out waiting for volume to create: timed out waiting for the condition"
2025-06-05 08:26:32.594 I0605 06:26:32.594847       1 controller.go:1094] "Final error received, removing PVC from claims in progress" claimUID="c8f34cdd-edc1-4f77-8d9f-59bf73365069"
2025-06-05 08:26:32.594 I0605 06:26:32.594867       1 controller.go:951] "Retrying syncing claim" key="c8f34cdd-edc1-4f77-8d9f-59bf73365069" failures=4
2025-06-05 08:26:32.594 E0605 06:26:32.594894       1 controller.go:974] "Unhandled Error" err="error syncing claim \"c8f34cdd-edc1-4f77-8d9f-59bf73365069\": failed to provision volume with StorageClass \"cnpg-io2\": rpc error: code = Internal desc = Could not create volume \"pvc-c8f34cdd-edc1-4f77-8d9f-59bf73365069\": timed out waiting for volume to create: timed out waiting for the condition" logger="UnhandledError"
2025-06-05 08:26:32.594 I0605 06:26:32.594946       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Warning" reason="ProvisioningFailed" message="failed to provision volume with StorageClass \"cnpg-io2\": rpc error: code = Internal desc = Could not create volume \"pvc-c8f34cdd-edc1-4f77-8d9f-59bf73365069\": timed out waiting for volume to create: timed out waiting for the condition"
2025-06-05 08:26:48.595 I0605 06:26:48.595544       1 event.go:389] "Event occurred" object="history-infra/history-cnpg-15" fieldPath="" kind="PersistentVolumeClaim" apiVersion="v1" type="Normal" reason="Provisioning" message="External provisioner is provisioning volume for claim \"history-infra/history-cnpg-15\""
2025-06-05 08:26:48.679 E0605 06:26:48.679755       1 handlers.go:85] "Error from AWS API" err="api error VolumeLimitExceeded: You have exceeded your maximum io2 storage limit of 30 TiB in this region. Please contact AWS Support to request an Elastic Block Store service limit increase."
2025-06-05 08:26:48.679 E0605 06:26:48.679865       1 driver.go:108] "GRPC error" err="rpc error: code = Internal desc = Could not create volume \"pvc-c8f34cdd-edc1-4f77-8d9f-59bf73365069\": could not create volume in EC2: operation error EC2: CreateVolume, https response error StatusCode: 400, RequestID: 183d1dbb-1564-4ff1-9fc6-16340eb2cbf6, api error VolumeLimitExceeded: You have exceeded your maximum io2 storage limit of 30 TiB in this region. Please contact AWS Support to request an Elastic Block Store service limit increase."
2025-06-05 08:26:48.680 I0605 06:26:48.680236       1 controller.go:1094] "Final error received, removing PVC from claims in progress" claimUID="c8f34cdd-edc1-4f77-8d9f-59bf73365069"
2025-06-05 08:26:48.680 I0605 06:26:48.680258       1 controller.go:951] "Retrying syncing claim" key="c8f34cdd-edc1-4f77-8d9f-59bf73365069" failures=5
2025-06-05 08:26:48.680 E0605 06:26:48.680298       1 controller.go:974] "Unhandled Error" err="error syncing claim \"c8f34cdd-edc1-4f77-8d9f-59bf73365069\": failed to provision volume with StorageClass \"cnpg-io2\": rpc error: code = Internal desc = Could not create volume \"pvc-c8f34cdd-edc1-4f77-8d9f-59bf73365069\": could not create volume in EC2: operation error EC2: CreateVolume, https response error StatusCode: 400, RequestID: 183d1dbb-1564-4ff1-9fc6-16340eb2cbf6, api error VolumeLimitExceeded: You have exceeded your maximum io2 storage limit of 30 TiB in this region. Please contact AWS Support to request an Elastic Block Store service limit increase." logger="UnhandledError"
```

</details>

**Environment**
- Kubernetes version (use `kubectl version`):
  ```
  Client Version: v1.33.1
  Kustomize Version: v5.6.0
  Server Version: v1.32.5-eks-5d4a308
  ```
- Driver version: v1.38.1
- Helm chart version 2.38.1
- Helm chart values:
  ```
  controller:
    serviceAccount:
      create: false
      name: ********
    logLevel: 3
    replicaCount: 1
    region: eu-central-1
  node:
    tolerateAllTaints: false
    tolerations: {}
    volumeAttachLimit: 24
    enableMetrics: true
  ```

We don't tolerate taints due to some Fargate nodes we have in our clusters.

Hints from issue comments:
@ConnorJC3: Hi @samox73 - that limit is imposed by AWS and not the EBS CSI Driver. The EBS CSI Driver is just returning the error we received from the AWS API, it does not perform any limit calculations.

If you believe there is a bug/mistake in the calculation of EBS limits, please reach out to AWS Support to report that issue.@AndrewSirenko: +1 to @ConnorJC3's comment

> and fail robustly.

What do you mean by fail robustly? The driver does fail the CreateVolume request and leaves an event on the PVC resource. 

If you delete the PVC resource the driver will stop trying to create the volume and you will no longer get the error.

The reason the driver keeps retrying is because [Kubernetes controllers]([Controllers | Kubernetes](https://kubernetes.io/docs/concepts/architecture/controller/)) continually attempt to move the cluster to the desired state (in this case, towards creating an io2 ebs volume). The benefit here is that the volume you have requested by creating the PVC resource will eventually get provisioned when your account limit is raised or other io2 volumes are deleted. @samox73: Oh sorry, I left out an important piece of information: The EBS volume **was** created but the controller failed to link it to the PVC somehow and then started to throw the "limit reached" errors. That's what I meant by "fail robustly". I haven't taken a look into the code, but I feel more logging would be useful here.

It does not log anything that would indicate the creation of the EBS volume succeeded. But if the controller doesn't do any limit calculations then that may as well be a bug on AWS's side?@AndrewSirenko: Ah thank you for that additional piece of information. This is unexpected behavior and we will look into it. Thank you. @AndrewSirenko: For folks following, this looks like an EC2/EBS issue. 

You can easily reproduce this by creating a volume that will put you just under your regional max storage limit, and then making the same request with the same client-token:

```

$ aws ec2 --region us-east-2 create-volume --volume-type io2 --size 4000 --availability-zone us-east-2a --iops 10000 --client-token abc123       
{
    "Iops": 10000,
    "Tags": [],
    "VolumeType": "io2",
    "MultiAttachEnabled": false,
    "VolumeId": "vol-02240b7a5159dd8df",
    "Size": 4000,
    "SnapshotId": "",
    "AvailabilityZone": "us-east-2a",
    "State": "creating",
    "CreateTime": "2025-06-06T14:00:33+00:00",
    "Encrypted": false
}

$ aws ec2 --region us-east-2 create-volume --volume-type io2 --size 4000 --availability-zone us-east-2a --iops 10000 --client-token abc123

An error occurred (VolumeLimitExceeded) when calling the CreateVolume operation: You have exceeded your maximum io2 storage limit of 20 TiB in this region. Please contact AWS Support to request an Elastic Block Store service limit increase.
```

The driver relies upon the fact that create-volume requests made with the same client-token (which we choose by hashing the PV's UID) will lead to the same response, which is violated in this case above. 

We have brought this up internally. You can consider also making an AWS customer support case. If EC2/EBS does not change this behavior, then we will need to add a workaround in our driver.

Thanks!@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@ConnorJC3: /lifecycle frozen

Solution patch:
diff --git a/charts/aws-ebs-csi-driver/templates/_node-windows.tpl b/charts/aws-ebs-csi-driver/templates/_node-windows.tpl
index ae0db8ca..ea49a149 100644
--- a/charts/aws-ebs-csi-driver/templates/_node-windows.tpl
+++ b/charts/aws-ebs-csi-driver/templates/_node-windows.tpl
@@ -123,6 +123,10 @@ spec:
             - name: AWS_USE_FIPS_ENDPOINT
               value: "true"
             {{- end }}
+            {{- if .Values.node.serviceAccount.disableMutation }}
+            - name: DISABLE_TAINT_WATCHER
+              value: "true"
+            {{- end }}
             {{- with .Values.node.env }}
             {{- . | toYaml | nindent 12 }}
             {{- end }}
diff --git a/charts/aws-ebs-csi-driver/templates/_node.tpl b/charts/aws-ebs-csi-driver/templates/_node.tpl
index 1cae74a1..ed020543 100644
--- a/charts/aws-ebs-csi-driver/templates/_node.tpl
+++ b/charts/aws-ebs-csi-driver/templates/_node.tpl
@@ -123,6 +123,10 @@ spec:
             - name: AWS_USE_FIPS_ENDPOINT
               value: "true"
             {{- end }}
+            {{- if .Values.node.serviceAccount.disableMutation }}
+            - name: DISABLE_TAINT_WATCHER
+              value: "true"
+            {{- end }}
             {{- with .Values.node.env }}
             {{- . | toYaml | nindent 12 }}
             {{- end }}
diff --git a/charts/aws-ebs-csi-driver/templates/clusterrole-csi-node.yaml b/charts/aws-ebs-csi-driver/templates/clusterrole-csi-node.yaml
index c04310f6..c80866d8 100644
--- a/charts/aws-ebs-csi-driver/templates/clusterrole-csi-node.yaml
+++ b/charts/aws-ebs-csi-driver/templates/clusterrole-csi-node.yaml
@@ -6,12 +6,19 @@ metadata:
   labels:
     {{- include "aws-ebs-csi-driver.labels" . | nindent 4 }}
 rules:
-  - apiGroups: [""]
-    resources: ["nodes"]
-    verbs: ["get", "patch", "list", "watch"]
   - apiGroups: ["storage.k8s.io"]
     resources: ["volumeattachments"]
     verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get"]
+  {{- if not .Values.node.serviceAccount.disableMutation }}
+  {{- /* We also disable some read APIs here as they are only used in the taint removal feature */}}
+  {{- /* But those are not the important part and could be re-added in the future if needed */}}
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["patch", "list", "watch"]
   - apiGroups: ["storage.k8s.io"]
     resources: ["csinodes"]
     verbs: ["get"]
+  {{- end }}
diff --git a/charts/aws-ebs-csi-driver/values.schema.json b/charts/aws-ebs-csi-driver/values.schema.json
index 9db42f3e..c3aad4e0 100644
--- a/charts/aws-ebs-csi-driver/values.schema.json
+++ b/charts/aws-ebs-csi-driver/values.schema.json
@@ -717,6 +717,11 @@
               "type": ["object", "null"],
               "description": "Additional annotations added to the ebs-csi-node-sa service account",
               "default": null
+            },
+            "disableMutation": {
+              "type": ["boolean", "null"],
+              "description": "Disable mutating permissions for the node service account. When enabled, some features of the EBS CSI Driver node pods will not function, such as taint removal. Primarily useful in particularly security-sensitive environments, or on multi-tenant clusters that isolate tenants by node.",
+              "default": false
             }
           }
         },
diff --git a/charts/aws-ebs-csi-driver/values.yaml b/charts/aws-ebs-csi-driver/values.yaml
index b2f0cff8..fecab6e4 100644
--- a/charts/aws-ebs-csi-driver/values.yaml
+++ b/charts/aws-ebs-csi-driver/values.yaml
@@ -418,6 +418,10 @@ node:
     ## Enable if EKS IAM for SA is used
     # eks.amazonaws.com/role-arn: arn:<partition>:iam::<account>:role/ebs-csi-role
     automountServiceAccountToken: true
+    # Disable mutating permissions for the node service account.
+    # When enabled, some features of the EBS CSI Driver node pods will not function, such as taint removal.
+    # Primarily useful in particularly security-sensitive environments, or on multi-tenant clusters that isolate tenants by node.
+    disableMutation: false
   # Enable the linux daemonset creation
   enableLinux: true
   enableWindows: true
diff --git a/deploy/kubernetes/base/clusterrole-csi-node.yaml b/deploy/kubernetes/base/clusterrole-csi-node.yaml
index 77bac09e..93540a4f 100644
--- a/deploy/kubernetes/base/clusterrole-csi-node.yaml
+++ b/deploy/kubernetes/base/clusterrole-csi-node.yaml
@@ -7,12 +7,15 @@ metadata:
   labels:
     app.kubernetes.io/name: aws-ebs-csi-driver
 rules:
-  - apiGroups: [""]
-    resources: ["nodes"]
-    verbs: ["get", "patch", "list", "watch"]
   - apiGroups: ["storage.k8s.io"]
     resources: ["volumeattachments"]
     verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["patch", "list", "watch"]
   - apiGroups: ["storage.k8s.io"]
     resources: ["csinodes"]
     verbs: ["get"]
diff --git a/docs/parameters.md b/docs/parameters.md
index be9ff0d2..466ffc65 100644
--- a/docs/parameters.md
+++ b/docs/parameters.md
@@ -9,8 +9,9 @@ The AWS EBS CSI Driver supports [tagging](tagging.md) through `StorageClass.para
 |------------------------------|-------------------------------------------------|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | "csi.storage.k8s.io/fstype"  | xfs, ext3, ext4                            | ext4    | File system type that will be formatted during volume creation. This parameter is case sensitive!                                                                                                                                                                                                                                                                                             |
 | "type"                       | io1, io2, gp2, gp3, sc1, st1, standard | gp3*    | EBS volume type.                                                                                                                                                                                                                                                                                                                                                                              |
-| "iopsPerGB"                  |                                                 |         | I/O operations per second per GiB. Can be specified for IO1, IO2, and GP3 volumes.                                                                                                                                                                                                                                                                                                            |
+| "iopsPerGB"                  |                                                 |         | I/O operations per second per GiB. Can be specified for IO1, IO2, and GP3. If `iopsPerGB * <volume size>` exceeds volume limits, the IOPS will be capped at the maximum allowed for that volume type and the call will succeed.                                                                                                                                                                                                                                                                                                      |
 | "allowAutoIOPSPerGBIncrease" | true, false                                     | false   | When `"true"`, the CSI driver increases IOPS for a volume when `iopsPerGB * <volume size>` is too low to fit into IOPS range supported by AWS. This allows dynamic provisioning to always succeed, even when user specifies too small PVC capacity or `iopsPerGB` value. On the other hand, it may introduce additional costs, as such volumes have higher IOPS than requested in `iopsPerGB`. |
+| "allowAutoIOPSIncreaseOnModify" | true, false                                  | false   | When `"true"`, the CSI Driver adjusts IOPS for a volume during resizing if `iopsPerGB` is set. This ensures that the volume maintains the desired IOPS/GiB ratio.
 | "iops"                       |                                                 |         | I/O operations per second. Can be specified for IO1, IO2, and GP3 volumes.                                                                                                                                                                                                                                                                                                                    |
 | "throughput"                 |                                                 | 125     | Throughput in MiB/s. Only effective when gp3 volume type is specified. If empty, it will set to 125MiB/s as documented [here](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ebs-volume-types.html).                                                                                                                                                                                     |
 | "encrypted"                  | true, false                                     | false   | Whether the volume should be encrypted or not. Valid values are "true" or "false".                                                                                                                                                                                                                                                                                                            |
diff --git a/hack/e2e/run.sh b/hack/e2e/run.sh
index 482aa8d4..8a568b71 100755
--- a/hack/e2e/run.sh
+++ b/hack/e2e/run.sh
@@ -105,10 +105,8 @@ if [[ "${HELM_CT_TEST}" == true ]]; then
   (
     while true; do
       if kubectl get pod ebs-csi-driver-test -n kube-system --kubeconfig "${KUBECONFIG}" &>/dev/null; then
-        echo "Pod found, waiting for it to become ready..."
         if kubectl wait --for=condition=ready pod ebs-csi-driver-test -n kube-system --timeout=300s --kubeconfig "${KUBECONFIG}"; then
-          echo "Pod is ready, fetching logs..."
-          kubectl logs -f ebs-csi-driver-test -n kube-system -c kubetest2 --kubeconfig "${KUBECONFIG}"
+          kubectl logs -f ebs-csi-driver-test -n kube-system -c kubetest2 --kubeconfig "${KUBECONFIG}" >"${REPORT_DIR}/helm-test-pod.txt"
         fi
       fi
       sleep 30
@@ -156,7 +154,7 @@ else
   else
     set -x
     set +e
-    "${BIN}/ginkgo" -p -nodes="${GINKGO_PARALLEL}" -v \
+    "${BIN}/ginkgo" -p -nodes="${GINKGO_PARALLEL}" \
       --focus="${GINKGO_FOCUS}" \
       --skip="${GINKGO_SKIP}" \
       --junit-report="${REPORT_DIR}/junit.xml" \
@@ -172,10 +170,7 @@ else
   PODS=$(kubectl get pod -n kube-system -l "app.kubernetes.io/name=aws-ebs-csi-driver,app.kubernetes.io/instance=aws-ebs-csi-driver" -o json --kubeconfig "${KUBECONFIG}" | jq -r .items[].metadata.name)
 
   while IFS= read -r POD; do
-    loudecho "Printing pod ${POD} container logs"
-    set +e
-    kubectl logs "${POD}" -n kube-system --all-containers --ignore-errors --kubeconfig "${KUBECONFIG}"
-    set -e
+    kubectl logs "${POD}" -n kube-system --all-containers --ignore-errors --kubeconfig "${KUBECONFIG}" >"${REPORT_DIR}/${POD}.txt"
   done <<<"${PODS}"
 fi
 
diff --git a/pkg/cloud/cloud.go b/pkg/cloud/cloud.go
index d1b2eca8..6e2ffcec 100644
--- a/pkg/cloud/cloud.go
+++ b/pkg/cloud/cloud.go
@@ -123,6 +123,10 @@ const (
 	KubernetesTagKeyPrefix = "kubernetes.io"
 	// AwsEbsDriverTagKey is the tag to identify if a volume/snapshot is managed by ebs csi driver.
 	AwsEbsDriverTagKey = util.DriverName + "/cluster"
+	// AllowAutoIOPSIncreaseOnModifyKey is the tag key for allowing IOPS increase on resizing if IOPSPerGB is set to ensure desired ratio is maintained.
+	AllowAutoIOPSIncreaseOnModifyKey = util.DriverName + "/AllowAutoIOPSIncreaseOnModify"
+	// IOPSPerGBKey represents the tag key for IOPS per GB.
+	IOPSPerGBKey = util.DriverName + "/IOPSPerGb"
 )
 
 // Batcher.
@@ -250,9 +254,11 @@ type DiskOptions struct {
 
 // ModifyDiskOptions represents parameters to modify an EBS volume.
 type ModifyDiskOptions struct {
-	VolumeType string
-	IOPS       int32
-	Throughput int32
+	VolumeType                string
+	IOPS                      int32
+	Throughput                int32
+	IOPSPerGB                 int32
+	AllowIopsIncreaseOnResize bool
 }
 
 // iopsLimits represents the IOPS limits set by EBS of a volume dependent on the volume type.
@@ -709,27 +715,58 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 		size, outpostArn, volumeID, err = c.createVolumeHelper(ctx, diskOptions, createRequestInput, iops, throughput, zone, zoneID)
 	}
 	if err != nil {
-		if isAWSErrorSnapshotNotFound(err) {
+		switch {
+		case isAWSErrorSnapshotNotFound(err):
 			return nil, ErrNotFound
-		}
-		if isAWSErrorIdempotentParameterMismatch(err) {
+		case isAWSErrorIdempotentParameterMismatch(err):
 			nextTokenNumber := 2
 			if tokenNumber, ok := c.latestClientTokens.Get(volumeName); ok {
 				nextTokenNumber = *tokenNumber + 1
 			}
 			c.latestClientTokens.Set(volumeName, &nextTokenNumber)
 			return nil, ErrIdempotentParameterMismatch
-		}
-		if isAWSErrorVolumeLimitExceeded(err) {
-			return nil, fmt.Errorf("%w: %w", ErrLimitExceeded, err)
-		}
-		if isAWSErrorInvalidParameterCombination(err) {
+		case isAWSErrorInvalidParameterCombination(err):
 			return nil, ErrInvalidArgument
-		}
-		if isAwsErrorMaxIOPSLimitExceeded(err) {
+		case isAWSErrorVolumeLimitExceeded(err):
+			// EC2 API does NOT handle idempotency correctly when a theoretical volume
+			// would put the caller over a limit for their account
+			//
+			// To avoid leaking volumes, make a DescribeVolumes call here
+			request := &ec2.DescribeVolumesInput{
+				Filters: []types.Filter{
+					{
+						Name:   aws.String("tag:" + VolumeNameTagKey),
+						Values: []string{volumeName},
+					},
+				},
+			}
+			// Call DescribeVolumes directly as there is a high chance this volume
+			// will return a NotFound error and would poison a batch call
+			volumes, describeErr := describeVolumes(ctx, c.ec2, request)
+			if describeErr != nil {
+				if isAWSErrorVolumeNotFound(describeErr) {
+					return nil, fmt.Errorf("%w: %w", ErrLimitExceeded, err)
+				} else {
+					return nil, describeErr
+				}
+			}
+
+			// Volume with requested name exists, continue with it
+			if l := len(volumes); l > 1 {
+				return nil, ErrMultiDisks
+			} else if l < 1 {
+				// This should in theory be impossible, but if the API
+				// changes or breaks it would cause a panic, so handle it
+				return nil, fmt.Errorf("%w: %w", ErrLimitExceeded, err)
+			}
+			volumeID = aws.ToString(volumes[0].VolumeId)
+			size = aws.ToInt32(volumes[0].Size)
+			outpostArn = aws.ToString(volumes[0].OutpostArn)
+		case isAwsErrorMaxIOPSLimitExceeded(err):
 			return nil, fmt.Errorf("%w: %w", ErrLimitExceeded, err)
+		default:
+			return nil, fmt.Errorf("could not create volume in EC2: %w", err)
 		}
-		return nil, fmt.Errorf("could not create volume in EC2: %w", err)
 	}
 
 	if len(volumeID) == 0 {
@@ -905,27 +942,89 @@ func (c *cloud) ResizeOrModifyDisk(ctx context.Context, volumeID string, newSize
 	if err != nil {
 		return 0, err
 	}
-	needsModification, volumeSize, err := c.validateModifyVolume(ctx, volumeID, newSizeGiB, options)
+	request := &ec2.DescribeVolumesInput{
+		VolumeIds: []string{volumeID},
+	}
+	volume, err := c.getVolume(ctx, request)
+	if err != nil {
+		return 0, err
+	}
 
+	needsModification, volumeSize, err := c.validateVolumeState(ctx, volumeID, newSizeGiB, *volume.Size, options)
 	if err != nil || !needsModification {
 		return volumeSize, err
 	}
 
+	if options.IOPS > 0 && options.IOPSPerGB > 0 {
+		return 0, errors.New("invalid VAC parameter; specify either IOPS or IOPSPerGb, not both")
+	}
+
 	req := &ec2.ModifyVolumeInput{
 		VolumeId: aws.String(volumeID),
 	}
 	if newSizeBytes != 0 {
 		req.Size = aws.Int32(newSizeGiB)
 	}
-	if options.IOPS != 0 {
-		req.Iops = aws.Int32(options.IOPS)
-	}
+	volTypeToUse := volume.VolumeType
 	if options.VolumeType != "" {
 		req.VolumeType = types.VolumeType(options.VolumeType)
+		volTypeToUse = req.VolumeType
 	}
 	if options.Throughput != 0 {
 		req.Throughput = aws.Int32(options.Throughput)
 	}
+
+	var sizeToUse int32
+	if req.Size != nil {
+		sizeToUse = *req.Size
+	} else {
+		sizeToUse = *volume.Size
+	}
+
+	allowAutoIncreaseIsSet, iopsPerGbVal, err := c.checkIfIopsIncreaseOnExpansion(volume.Tags)
+	if err != nil {
+		return 0, err
+	}
+	var iopsForModify int32
+	switch {
+	case options.IOPS != 0:
+		iopsForModify = options.IOPS
+	case options.IOPSPerGB != 0 && (options.AllowIopsIncreaseOnResize || allowAutoIncreaseIsSet):
+		iopsForModify = sizeToUse * options.IOPSPerGB
+	case iopsPerGbVal > 0 && (options.AllowIopsIncreaseOnResize || allowAutoIncreaseIsSet):
+		iopsForModify = sizeToUse * iopsPerGbVal
+	}
+	if iopsForModify != 0 {
+		azParams := getVolumeLimitsParams{}
+
+		if volume.AvailabilityZone != nil {
+			azParams.availabilityZone = *volume.AvailabilityZone
+		}
+		if volume.AvailabilityZoneId != nil {
+			azParams.availabilityZoneId = *volume.AvailabilityZoneId
+		}
+		// EBS doesn't handle empty outpost arn, so we have to include it only when it's non-empty
+		if volume.OutpostArn != nil {
+			azParams.outpostArn = *volume.OutpostArn
+		}
+		iopsLimit, err := c.getVolumeLimits(ctx, string(volTypeToUse), azParams)
+		if err != nil {
+			return 0, err
+		}
+		// Even if volume type does not support IOPS, still pass value and let EC2 handle error.
+		if iopsLimit.maxIops == 0 {
+			req.Iops = aws.Int32(iopsForModify)
+		} else {
+			req.Iops = aws.Int32(capIOPS(string(volTypeToUse), sizeToUse, iopsForModify, iopsLimit, true))
+		}
+		options.IOPS = *req.Iops
+	}
+
+	needsModification, volumeSize, err = c.validateModifyVolume(ctx, volumeID, newSizeGiB, options, *volume)
+	if err != nil || !needsModification {
+		return volumeSize, err
+	}
+
 	response, err := c.ec2.ModifyVolume(ctx, req, func(o *ec2.Options) {
 		o.Retryer = c.rm.modifyVolumeRetryer
 	})
@@ -2404,7 +2503,7 @@ func (c *cloud) AvailabilityZones(ctx context.Context) (map[string]struct{}, err
 	return zones, nil
 }
 
-func needsVolumeModification(volume types.Volume, newSizeGiB int32, options *ModifyDiskOptions) bool {
+func needsVolumeModification(volume types.Volume, newSizeGiB int32, req *ModifyDiskOptions) bool {
 	oldSizeGiB := *volume.Size
 	//nolint:staticcheck // staticcheck suggests merging all of the below conditionals into one line,
 	// but that would be extremely difficult to read
@@ -2414,66 +2513,94 @@ func needsVolumeModification(volume types.Volume, newSizeGiB int32, options *Mod
 		needsModification = true
 	}
 
-	if options.IOPS != 0 && (volume.Iops == nil || *volume.Iops != options.IOPS) {
+	if req.IOPS != 0 && (volume.Iops == nil || *volume.Iops != req.IOPS) {
 		needsModification = true
 	}
 
-	if options.VolumeType != "" && !strings.EqualFold(string(volume.VolumeType), options.VolumeType) {
+	if req.VolumeType != "" && !strings.EqualFold(string(volume.VolumeType), req.VolumeType) {
 		needsModification = true
 	}
 
-	if options.Throughput != 0 && (volume.Throughput == nil || *volume.Throughput != options.Throughput) {
+	if req.Throughput != 0 && (volume.Throughput == nil || *volume.Throughput != req.Throughput) {
 		needsModification = true
 	}
 
 	return needsModification
 }
 
-func (c *cloud) validateModifyVolume(ctx context.Context, volumeID string, newSizeGiB int32, options *ModifyDiskOptions) (bool, int32, error) {
-	request := &ec2.DescribeVolumesInput{
-		VolumeIds: []string{volumeID},
+func getVolumeAttachmentsList(volume types.Volume) []string {
+	var volumeAttachmentList []string
+	for _, attachment := range volume.Attachments {
+		if attachment.State == types.VolumeAttachmentStateAttached {
+			volumeAttachmentList = append(volumeAttachmentList, aws.ToString(attachment.InstanceId))
+		}
 	}
 
-	volume, err := c.getVolume(ctx, request)
-	if err != nil {
-		return true, 0, err
-	}
+	return volumeAttachmentList
+}
 
-	if volume.Size == nil {
-		return true, 0, fmt.Errorf("volume %q has no size", volumeID)
+// Checks if a volume's IOPS can be increased on expansion to adhere to IopsPerGB ratio.
+func (c *cloud) checkIfIopsIncreaseOnExpansion(existingTags []types.Tag) (allowAutoIncreaseIsSet bool, iopsPerGbVal int32, err error) {
+	for _, tag := range existingTags {
+		switch *tag.Key {
+		case IOPSPerGBKey:
+			iopsPerGbVal64, err := strconv.ParseInt(*tag.Value, 10, 32)
+			if err != nil {
+				return false, 0, fmt.Errorf("%w: %w", ErrInvalidArgument, err)
+			}
+			iopsPerGbVal = int32(iopsPerGbVal64)
+		case AllowAutoIOPSIncreaseOnModifyKey:
+			allowAutoIncreaseIsSet, err = strconv.ParseBool(*tag.Value)
+			if err != nil {
+				return false, 0, fmt.Errorf("%w: %w", ErrInvalidArgument, err)
+			}
+		}
 	}
-	oldSizeGiB := *volume.Size
+	return allowAutoIncreaseIsSet, iopsPerGbVal, nil
+}
 
-	// This call must NOT be batched because a missing volume modification will return client error
+func (c *cloud) getVolumeModificationState(ctx context.Context, volumeID string) (*types.VolumeModification, error) {
 	latestMod, err := c.getLatestVolumeModification(ctx, volumeID, false)
 	if err != nil && !errors.Is(err, ErrVolumeNotBeingModified) {
-		return true, oldSizeGiB, fmt.Errorf("error fetching volume modifications for %q: %w", volumeID, err)
+		return nil, fmt.Errorf("error fetching volume modifications for %q: %w", volumeID, err)
+	}
+	return latestMod, nil
+}
+
+func (c *cloud) validateVolumeState(ctx context.Context, volumeID string, newSizeGiB int32, oldSizeGiB int32, options *ModifyDiskOptions) (bool, int32, error) {
+	latestMod, err := c.getVolumeModificationState(ctx, volumeID)
+	if err != nil {
+		return false, 0, err
 	}
 
-	state := ""
 	// latestMod can be nil if the volume has never been modified
-	if latestMod != nil {
-		state = string(latestMod.ModificationState)
-		if state == string(types.VolumeModificationStateModifying) {
-			// If volume is already modifying, detour to waiting for it to modify
-			klog.V(5).InfoS("[Debug] Watching ongoing modification", "volumeID", volumeID)
-			err = c.waitForVolumeModification(ctx, volumeID)
-			if err != nil {
-				return true, oldSizeGiB, err
-			}
-			returnGiB, returnErr := c.checkDesiredState(ctx, volumeID, newSizeGiB, options)
-			return false, returnGiB, returnErr
+	if latestMod != nil && string(latestMod.ModificationState) == string(types.VolumeModificationStateModifying) {
+		// If volume is already modifying, detour to waiting for it to modify
+		klog.V(5).InfoS("[Debug] Watching ongoing modification", "volumeID", volumeID)
+		err = c.waitForVolumeModification(ctx, volumeID)
+		if err != nil {
+			return false, oldSizeGiB, err
 		}
+		returnGiB, returnErr := c.checkDesiredState(ctx, volumeID, newSizeGiB, options)
+		return false, returnGiB, returnErr
 	}
+	return true, 0, nil
+}
+
+func (c *cloud) validateModifyVolume(ctx context.Context, volumeID string, newSizeGiB int32, options *ModifyDiskOptions, volume types.Volume) (bool, int32, error) {
+	if volume.Size == nil {
+		return true, 0, fmt.Errorf("volume %q has no size", volumeID)
+	}
+	oldSizeGiB := *volume.Size
 
 	// At this point, we know we are starting a new volume modification
 	// If we're asked to modify a volume to its current state, ignore the request and immediately return a success
 	// This is because as of March 2024, EC2 ModifyVolume calls that don't change any parameters still modify the volume
-	if !needsVolumeModification(*volume, newSizeGiB, options) {
+	if !needsVolumeModification(volume, newSizeGiB, options) {
 		klog.V(5).InfoS("[Debug] Skipping modification for volume due to matching stats", "volumeID", volumeID)
 		// Wait for any existing modifications to prevent race conditions where DescribeVolume(s) returns the new
 		// state before the volume is actually finished modifying
-		err = c.waitForVolumeModification(ctx, volumeID)
+		err := c.waitForVolumeModification(ctx, volumeID)
 		if err != nil {
 			return true, oldSizeGiB, err
 		}
@@ -2481,7 +2608,12 @@ func (c *cloud) validateModifyVolume(ctx context.Context, volumeID string, newSi
 		return false, returnGiB, returnErr
 	}
 
-	if state == string(types.VolumeModificationStateOptimizing) {
+	latestMod, err := c.getVolumeModificationState(ctx, volumeID)
+	if err != nil {
+		return true, 0, err
+	}
+
+	if latestMod != nil && string(latestMod.ModificationState) == string(types.VolumeModificationStateOptimizing) {
 		return true, 0, fmt.Errorf("volume %q in OPTIMIZING state, cannot currently modify", volumeID)
 	}
 
@@ -2492,17 +2624,6 @@ func volumeModificationDone(state string) bool {
 	return state == string(types.VolumeModificationStateCompleted) || state == string(types.VolumeModificationStateOptimizing)
 }
 
-func getVolumeAttachmentsList(volume types.Volume) []string {
-	var volumeAttachmentList []string
-	for _, attachment := range volume.Attachments {
-		if attachment.State == types.VolumeAttachmentStateAttached {
-			volumeAttachmentList = append(volumeAttachmentList, aws.ToString(attachment.InstanceId))
-		}
-	}
-
-	return volumeAttachmentList
-}
-
 // Calculate actual IOPS for a volume and cap it at supported AWS limits.
 func capIOPS(volumeType string, requestedCapacityGiB int32, requestedIops int32, iopsLimits iopsLimits, allowIncrease bool) int32 {
 	// If requestedIops is zero the user did not request a specific amount, and the default will be used instead
diff --git a/pkg/driver/constants.go b/pkg/driver/constants.go
index d86ea9ef..8a9c61ba 100644
--- a/pkg/driver/constants.go
+++ b/pkg/driver/constants.go
@@ -47,6 +47,9 @@ const (
 	// AllowAutoIOPSPerGBIncreaseKey represents key for allowing automatic increase of IOPS.
 	AllowAutoIOPSPerGBIncreaseKey = "allowautoiopspergbincrease"
 
+	// AllowAutoIOPSIncreaseOnModifyKey represents key for allowing IOPS increase on resizing if IopsPerGB is set to ensure desired ratio is maintained.
+	AllowAutoIOPSIncreaseOnModifyKey = "allowautoiopsincreaseonmodify"
+
 	// VolumeInitializationRateKey represents key for volume initialization rate when creating volumes from snapshots.
 	VolumeInitializationRateKey = "volumeinitializationrate"
 
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 24a61ebd..f0e8ddbc 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -144,6 +144,9 @@ func (d *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 				return nil, status.Errorf(codes.InvalidArgument, "Could not parse invalid iopsPerGB: %v", parseIopsPerGBKeyErr)
 			}
 			iopsPerGB = int32(parseIopsPerGBKey)
+			volumeTags[cloud.IOPSPerGBKey] = strconv.Itoa(int(iopsPerGB))
+		case AllowAutoIOPSIncreaseOnModifyKey:
+			volumeTags[cloud.AllowAutoIOPSIncreaseOnModifyKey] = strconv.FormatBool(isTrue(value))
 		case AllowAutoIOPSPerGBIncreaseKey:
 			allowIOPSPerGBIncrease = isTrue(value)
 		case IopsKey:
diff --git a/pkg/driver/controller_modify_volume.go b/pkg/driver/controller_modify_volume.go
index 5890e5b3..d403765a 100644
--- a/pkg/driver/controller_modify_volume.go
+++ b/pkg/driver/controller_modify_volume.go
@@ -146,7 +146,8 @@ func executeModifyVolumeRequest(c cloud.Cloud) func(string, modifyVolumeRequest)
 		if err != nil {
 			return 0, err
 		}
-		if (req.modifyDiskOptions.IOPS != 0) || (req.modifyDiskOptions.Throughput != 0) || (req.modifyDiskOptions.VolumeType != "") || (req.newSize != 0) {
+
+		if (req.modifyDiskOptions.IOPS != 0) || (req.modifyDiskOptions.Throughput != 0) || (req.modifyDiskOptions.VolumeType != "") || (req.newSize != 0) || (req.modifyDiskOptions.IOPSPerGB != 0) {
 			actualSizeGiB, err := c.ResizeOrModifyDisk(ctx, volumeID, req.newSize, &req.modifyDiskOptions)
 			if err != nil {
 				switch {
@@ -181,6 +182,7 @@ func parseModifyVolumeParameters(params map[string]string) (*modifyVolumeRequest
 		},
 	}
 	var rawTagsToAdd []string
+	var noValidationTags = make(map[string]string)
 	tProps := new(template.PVProps)
 	for key, value := range params {
 		switch key {
@@ -205,6 +207,16 @@ func parseModifyVolumeParameters(params map[string]string) (*modifyVolumeRequest
 			options.modifyDiskOptions.VolumeType = value
 		case ModificationKeyVolumeType:
 			options.modifyDiskOptions.VolumeType = value
+		case IopsPerGBKey:
+			noValidationTags[cloud.IOPSPerGBKey] = value
+			iopsPerGb, err := strconv.ParseInt(value, 10, 32)
+			if err != nil {
+				return nil, status.Errorf(codes.InvalidArgument, "Could not parse IopsPerGb: %q", value)
+			}
+			options.modifyDiskOptions.IOPSPerGB = int32(iopsPerGb)
+		case AllowAutoIOPSIncreaseOnModifyKey:
+			noValidationTags[cloud.AllowAutoIOPSIncreaseOnModifyKey] = value
+			options.modifyDiskOptions.AllowIopsIncreaseOnResize = isTrue(value)
 		case PVCNameKey:
 			tProps.PVCName = value
 		case PVCNamespaceKey:
@@ -229,6 +241,9 @@ func parseModifyVolumeParameters(params map[string]string) (*modifyVolumeRequest
 	if err := validateExtraTags(addTags, false); err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "Invalid tag value: %v", err)
 	}
+	for k, v := range noValidationTags {
+		addTags[k] = v
+	}
 	options.modifyTagsOptions.TagsToAdd = addTags
 	return &options, nil
 }
diff --git a/pkg/driver/node.go b/pkg/driver/node.go
index 2cec8490..1a1521cd 100644
--- a/pkg/driver/node.go
+++ b/pkg/driver/node.go
@@ -842,6 +842,10 @@ func collectMountOptions(fsType string, mntFlags []string) []string {
 // startNotReadyTaintWatcher launches a short‑lived Node informer that removes the
 // ebs.csi.aws.com/agent‑not‑ready taint. The informer is stopped after maxWatchDuration.
 func startNotReadyTaintWatcher(clientset kubernetes.Interface, maxWatchDuration time.Duration) {
+	if os.Getenv("DISABLE_TAINT_WATCHER") != "" {
+		klog.V(4).InfoS("DISABLE_TAINT_WATCHER set, skipping taint watcher")
+		return
+	}
 	nodeName := os.Getenv("CSI_NODE_NAME")
 	if nodeName == "" {
 		klog.V(4).InfoS("CSI_NODE_NAME missing, skipping taint watcher")
diff --git a/pkg/driver/validation.go b/pkg/driver/validation.go
index 87a96467..49b7fdd2 100644
--- a/pkg/driver/validation.go
+++ b/pkg/driver/validation.go
@@ -25,6 +25,16 @@ import (
 	"k8s.io/klog/v2"
 )
 
+var (
+	reservedTagKeys = map[string]string{
+		cloud.VolumeNameTagKey:           "",
+		cloud.AwsEbsDriverTagKey:         "",
+		cloud.SnapshotNameTagKey:         "",
+		IopsPerGBKey:                     "",
+		AllowAutoIOPSIncreaseOnModifyKey: "",
+	}
+)
+
 func ValidateDriverOptions(options *Options) error {
 	if err := validateExtraTags(options.ExtraTags, false); err != nil {
 		return fmt.Errorf("invalid extra tags: %w", err)
@@ -43,14 +53,8 @@ func ValidateDriverOptions(options *Options) error {
 
 func validateExtraTags(tags map[string]string, warnOnly bool) error {
 	validate := func(k, _ string) error {
-		if k == cloud.VolumeNameTagKey {
-			return fmt.Errorf("tag key '%s' is reserved", cloud.VolumeNameTagKey)
-		}
-		if k == cloud.AwsEbsDriverTagKey {
-			return fmt.Errorf("tag key '%s' is reserved", cloud.AwsEbsDriverTagKey)
-		}
-		if k == cloud.SnapshotNameTagKey {
-			return fmt.Errorf("tag key '%s' is reserved", cloud.SnapshotNameTagKey)
+		if _, ok := reservedTagKeys[k]; ok {
+			return fmt.Errorf("tag key '%s' is reserved", k)
 		}
 		if strings.HasPrefix(k, cloud.KubernetesTagKeyPrefix) {
 			return fmt.Errorf("tag key prefix '%s' is reserved", cloud.KubernetesTagKeyPrefix)



Test patch:
diff --git a/pkg/cloud/cloud_test.go b/pkg/cloud/cloud_test.go
index ea316528..3aedf889 100644
--- a/pkg/cloud/cloud_test.go
+++ b/pkg/cloud/cloud_test.go
@@ -1823,16 +1823,61 @@ func TestCreateDisk(t *testing.T) {
 			expErr: fmt.Errorf("invalid AWS VolumeType %q", "invalidVolumeType"),
 		},
 		{
-			name:       "failure: create volume returned volume limit exceeded error",
+			name:       "success: create volume returned volume limit exceeded error, but volume exists",
 			volumeName: "vol-test-name",
 			diskOptions: &DiskOptions{
 				CapacityBytes: util.GiBToBytes(1),
 				Tags:          map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
 			},
-			expDisk:              nil,
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
 			expCreateVolumeInput: &ec2.CreateVolumeInput{},
-			expCreateVolumeErr:   errors.New("VolumeLimitExceeded"),
-			expErr:               fmt.Errorf("could not create volume in EC2: %w", errors.New("VolumeLimitExceeded")),
+			expCreateVolumeErr: &smithy.GenericAPIError{
+				Code:    "VolumeLimitExceeded",
+				Message: "Volume limit exceeded",
+			},
+		},
+		{
+			name:       "failure: create volume returned volume limit exceeded error, describe returns not found",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes: util.GiBToBytes(1),
+				Tags:          map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+			},
+			expCreateVolumeInput: &ec2.CreateVolumeInput{},
+			expCreateVolumeErr: &smithy.GenericAPIError{
+				Code:    "VolumeLimitExceeded",
+				Message: "Volume limit exceeded",
+			},
+			expDescVolumeErr: &smithy.GenericAPIError{
+				Code:    "InvalidVolume.NotFound",
+				Message: "Volume not found",
+			},
+			expErr: fmt.Errorf("%w: %w", ErrLimitExceeded, &smithy.GenericAPIError{
+				Code:    "VolumeLimitExceeded",
+				Message: "Volume limit exceeded",
+			}),
+		},
+		{
+			name:       "failure: create volume returned volume limit exceeded error, describe returns existing volume",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes: util.GiBToBytes(1),
+				Tags:          map[string]string{VolumeNameTagKey: "vol-test", AwsEbsDriverTagKey: "true"},
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: defaultZone,
+			},
+			expCreateVolumeInput: &ec2.CreateVolumeInput{},
+			expCreateVolumeErr: &smithy.GenericAPIError{
+				Code:    "VolumeLimitExceeded",
+				Message: "Volume limit exceeded",
+			},
 		},
 		{
 			name:       "failure: create volume returned max iops limit exceeded error",
@@ -3338,6 +3383,7 @@ func TestResizeOrModifyDisk(t *testing.T) {
 				VolumeId:         aws.String("vol-test"),
 				Size:             aws.Int32(1),
 				AvailabilityZone: aws.String(defaultZone),
+				VolumeType:       types.VolumeTypeGp3,
 			},
 			modifiedVolume: &ec2.ModifyVolumeOutput{
 				VolumeModification: &types.VolumeModification{
@@ -3358,6 +3404,7 @@ func TestResizeOrModifyDisk(t *testing.T) {
 				VolumeId:         aws.String("vol-test"),
 				Size:             aws.Int32(1),
 				AvailabilityZone: aws.String(defaultZone),
+				VolumeType:       types.VolumeTypeGp3,
 			},
 			modifiedVolume: &ec2.ModifyVolumeOutput{
 				VolumeModification: &types.VolumeModification{
@@ -3387,6 +3434,7 @@ func TestResizeOrModifyDisk(t *testing.T) {
 				VolumeId:         aws.String("vol-test"),
 				Size:             aws.Int32(2),
 				AvailabilityZone: aws.String(defaultZone),
+				VolumeType:       types.VolumeTypeGp3,
 			},
 			descModVolume: &ec2.DescribeVolumesModificationsOutput{
 				VolumesModifications: []types.VolumeModification{
@@ -3472,6 +3520,7 @@ func TestResizeOrModifyDisk(t *testing.T) {
 				VolumeId:         aws.String("vol-test"),
 				Size:             aws.Int32(1),
 				AvailabilityZone: aws.String(defaultZone),
+				VolumeType:       types.VolumeTypeGp3,
 			},
 			descModVolume: &ec2.DescribeVolumesModificationsOutput{
 				VolumesModifications: []types.VolumeModification{
@@ -3496,6 +3545,7 @@ func TestResizeOrModifyDisk(t *testing.T) {
 				VolumeId:         aws.String("vol-test"),
 				AvailabilityZone: aws.String(defaultZone),
 				VolumeType:       types.VolumeTypeGp2,
+				Size:             aws.Int32(1),
 			},
 			modifiedVolumeError: errors.New("GenericErr"),
 			expErr:              errors.New("GenericErr"),
@@ -3572,6 +3622,7 @@ func TestResizeOrModifyDisk(t *testing.T) {
 				AvailabilityZone: aws.String(defaultZone),
 				Size:             aws.Int32(13),
 				Iops:             aws.Int32(3000),
+				VolumeType:       types.VolumeTypeGp3,
 			},
 			reqSizeGiB: 13,
 			modifyDiskOptions: &ModifyDiskOptions{
@@ -3620,6 +3671,12 @@ func TestResizeOrModifyDisk(t *testing.T) {
 						}, tc.existingVolumeError)
 				}
 			}
+
+			mockEC2.EXPECT().DescribeTags(gomock.Any(), gomock.Any()).Return(&ec2.DescribeTagsOutput{}, nil).AnyTimes()
+
+			// Mock CreateVolume for getVolumeLimits dry-run calls
+			mockEC2.EXPECT().CreateVolume(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, &smithy.GenericAPIError{Code: "DryRunOperation"}).AnyTimes()
+
 			if tc.modifiedVolume != nil || tc.modifiedVolumeError != nil {
 				mockEC2.EXPECT().ModifyVolume(gomock.Any(), gomock.Any(), gomock.Any()).Return(tc.modifiedVolume, tc.modifiedVolumeError).AnyTimes()
 			}
@@ -5082,3 +5139,84 @@ func createDescribeVolumesOutput(volumeIDs []*string, nodeID, path, state string
 		Volumes: volumes,
 	}
 }
+
+func TestCheckIfIopsIncreaseOnExpansion(t *testing.T) {
+	testCases := []struct {
+		name                      string
+		tags                      []types.Tag
+		expectedAllowAutoIncrease bool
+		expectedIopsPerGb         int32
+		expectedError             bool
+	}{
+		{
+			name: "both tags present with valid values",
+			tags: []types.Tag{
+				{Key: aws.String(IOPSPerGBKey), Value: aws.String("10")},
+				{Key: aws.String(AllowAutoIOPSIncreaseOnModifyKey), Value: aws.String("true")},
+			},
+			expectedAllowAutoIncrease: true,
+			expectedIopsPerGb:         10,
+			expectedError:             false,
+		},
+		{
+			name: "only IOPSPerGB tag present",
+			tags: []types.Tag{
+				{Key: aws.String(IOPSPerGBKey), Value: aws.String("5")},
+			},
+			expectedAllowAutoIncrease: false,
+			expectedIopsPerGb:         5,
+			expectedError:             false,
+		},
+		{
+			name: "only AllowAutoIOPSIncrease tag present",
+			tags: []types.Tag{
+				{Key: aws.String(AllowAutoIOPSIncreaseOnModifyKey), Value: aws.String("false")},
+			},
+			expectedAllowAutoIncrease: false,
+			expectedIopsPerGb:         0,
+			expectedError:             false,
+		},
+		{
+			name:                      "no relevant tags",
+			tags:                      []types.Tag{},
+			expectedAllowAutoIncrease: false,
+			expectedIopsPerGb:         0,
+			expectedError:             false,
+		},
+		{
+			name: "invalid IOPSPerGB value",
+			tags: []types.Tag{
+				{Key: aws.String(IOPSPerGBKey), Value: aws.String("invalid")},
+			},
+			expectedAllowAutoIncrease: false,
+			expectedIopsPerGb:         0,
+			expectedError:             true,
+		},
+		{
+			name: "invalid AllowAutoIOPSIncrease value",
+			tags: []types.Tag{
+				{Key: aws.String(AllowAutoIOPSIncreaseOnModifyKey), Value: aws.String("invalid")},
+			},
+			expectedAllowAutoIncrease: false,
+			expectedIopsPerGb:         0,
+			expectedError:             true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			c := &cloud{}
+
+			allowAutoIncrease, iopsPerGb, err := c.checkIfIopsIncreaseOnExpansion(tc.tags)
+
+			if tc.expectedError {
+				require.Error(t, err)
+				require.ErrorIs(t, err, ErrInvalidArgument)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expectedAllowAutoIncrease, allowAutoIncrease)
+				assert.Equal(t, tc.expectedIopsPerGb, iopsPerGb)
+			}
+		})
+	}
+}
diff --git a/pkg/driver/controller_modify_volume_test.go b/pkg/driver/controller_modify_volume_test.go
index 6d2d930f..ccf11494 100644
--- a/pkg/driver/controller_modify_volume_test.go
+++ b/pkg/driver/controller_modify_volume_test.go
@@ -151,9 +151,12 @@ func TestParseModifyVolumeParameters(t *testing.T) {
 		{
 			name: "basic params",
 			params: map[string]string{
-				ModificationKeyVolumeType: validType,
-				ModificationKeyIOPS:       validIops,
-				ModificationKeyThroughput: validThroughput,
+				ModificationKeyVolumeType:        validType,
+				ModificationKeyIOPS:              validIops,
+				ModificationKeyThroughput:        validThroughput,
+				AllowAutoIOPSIncreaseOnModifyKey: "true",
+				// IopsPerGB would not be actually allowed if IOPS is set but that is not checked in this function. Just testing that it properly parses it and adds the tag.
+				IopsPerGBKey:              "1000",
 				ModificationAddTag + "_1": validTagSpecificationInput,
 				ModificationAddTag + "_2": "key2={{ .PVCName }}",
 				ModificationAddTag + "_3": "key3={{ .PVCNamespace }}",
@@ -165,9 +168,11 @@ func TestParseModifyVolumeParameters(t *testing.T) {
 			},
 			expectedOptions: &modifyVolumeRequest{
 				modifyDiskOptions: cloud.ModifyDiskOptions{
-					VolumeType: validType,
-					IOPS:       validIopsInt,
-					Throughput: validThroughputInt,
+					VolumeType:                validType,
+					IOPS:                      validIopsInt,
+					Throughput:                validThroughputInt,
+					AllowIopsIncreaseOnResize: true,
+					IOPSPerGB:                 1000,
 				},
 				modifyTagsOptions: cloud.ModifyTagsOptions{
 					TagsToAdd: map[string]string{
@@ -175,6 +180,8 @@ func TestParseModifyVolumeParameters(t *testing.T) {
 						"key2": "ebs-claim",
 						"key3": "test-namespace",
 						"key4": "testPV-Name",
+						"ebs.csi.aws.com/AllowAutoIOPSIncreaseOnModify": "true",
+						"ebs.csi.aws.com/IOPSPerGb":                     "1000",
 					},
 					TagsToDelete: []string{
 						"key2",
diff --git a/pkg/driver/controller_test.go b/pkg/driver/controller_test.go
index 1dc4124d..ef970eeb 100644
--- a/pkg/driver/controller_test.go
+++ b/pkg/driver/controller_test.go
@@ -4911,6 +4911,7 @@ func TestControllerExpandVolume(t *testing.T) {
 			}
 
 			mockCloud := cloud.NewMockCloud(mockCtl)
+			mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(tc.req.GetVolumeId())).AnyTimes()
 			mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(tc.req.GetVolumeId()), gomock.Any(), gomock.Any()).Return(retSizeGiB, nil).AnyTimes()
 
 			awsDriver := ControllerService{
diff --git a/pkg/driver/request_coalescing_test.go b/pkg/driver/request_coalescing_test.go
index 330436a2..ad2147f1 100644
--- a/pkg/driver/request_coalescing_test.go
+++ b/pkg/driver/request_coalescing_test.go
@@ -105,6 +105,7 @@ func testBasicRequestCoalescingSuccess(t *testing.T, executor modifyVolumeExecut
 	defer mockCtl.Finish()
 
 	mockCloud := cloud.NewMockCloud(mockCtl)
+	mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(volumeID)).AnyTimes()
 	mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(volumeID), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, volumeID string, newSize int64, options *cloud.ModifyDiskOptions) (int64, error) {
 		klog.InfoS("ResizeOrModifyDisk called", "volumeID", volumeID, "newSize", newSize, "options", options)
 		if newSize != NewSize {
@@ -167,6 +168,7 @@ func testRequestFail(t *testing.T, executor modifyVolumeExecutor) {
 	defer mockCtl.Finish()
 
 	mockCloud := cloud.NewMockCloud(mockCtl)
+	mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(volumeID)).AnyTimes()
 	mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(volumeID), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, volumeID string, newSize int64, options *cloud.ModifyDiskOptions) (int64, error) {
 		klog.InfoS("ResizeOrModifyDisk called", "volumeID", volumeID, "newSize", newSize, "options", options)
 		return 0, errors.New("ResizeOrModifyDisk failed")
@@ -230,6 +232,7 @@ func testPartialFail(t *testing.T, executor modifyVolumeExecutor) {
 	volumeTypeChosen := ""
 
 	mockCloud := cloud.NewMockCloud(mockCtl)
+	mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(volumeID)).AnyTimes()
 	mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(volumeID), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, volumeID string, newSize int64, options *cloud.ModifyDiskOptions) (int64, error) {
 		klog.InfoS("ResizeOrModifyDisk called", "volumeID", volumeID, "newSize", newSize, "options", options)
 		if newSize != NewSize {
@@ -322,6 +325,7 @@ func testSequentialRequests(t *testing.T, executor modifyVolumeExecutor) {
 	defer mockCtl.Finish()
 
 	mockCloud := cloud.NewMockCloud(mockCtl)
+	mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(volumeID)).AnyTimes()
 	mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(volumeID), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, volumeID string, newSize int64, options *cloud.ModifyDiskOptions) (int64, error) {
 		klog.InfoS("ResizeOrModifyDisk", "volumeID", volumeID, "newSize", newSize, "options", options)
 		return newSize, nil
@@ -380,6 +384,7 @@ func testDuplicateRequest(t *testing.T, executor modifyVolumeExecutor) {
 	defer mockCtl.Finish()
 
 	mockCloud := cloud.NewMockCloud(mockCtl)
+	mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(volumeID)).AnyTimes()
 	mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(volumeID), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, volumeID string, newSize int64, options *cloud.ModifyDiskOptions) (int64, error) {
 		klog.InfoS("ResizeOrModifyDisk called", "volumeID", volumeID, "newSize", newSize, "options", options)
 		return newSize, nil
@@ -438,6 +443,7 @@ func testResponseReturnTiming(t *testing.T, executor modifyVolumeExecutor) {
 	defer mockCtl.Finish()
 
 	mockCloud := cloud.NewMockCloud(mockCtl)
+	mockCloud.EXPECT().GetDiskByID(gomock.Any(), gomock.Eq(volumeID)).AnyTimes()
 	mockCloud.EXPECT().ResizeOrModifyDisk(gomock.Any(), gomock.Eq(volumeID), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, volumeID string, newSize int64, options *cloud.ModifyDiskOptions) (int64, error) {
 		klog.InfoS("ResizeOrModifyDisk called", "volumeID", volumeID, "newSize", newSize, "options", options)