Repository: kubernetes/kubernetes (language: go)

Issue:
Graceful node shutdown doesn't wait for volume teardown
### What happened?

We have observed that after a node is preempted, pods using persistent volumes take 6+ minutes to restart. This happens when a volume was not able to be cleanly unmounted (indicated by updating `Node.Status.VolumesInUse`), which causes the Attach/Detach controller (in kube-controller-manager) to wait for 6 minutes before issuing a force detach. In addition to a long pod restart period, an unclean mount could also cause data corruption.

From the logs we've observed the following non-ideal behaviors, which I think reduces to the same issue that we only check for containers' being terminated and not their volumes.

1. The CSI driver running on the node gets terminated before the workload's volumes get unmounted. This causes unmount to fail. In our case, the CSI driver is part of the `system-node-critical" class, but since we don't wait for volumes to be unmounted, we move on to shutting down the critical pods.

3. The OS gets shutdown before volume teardown is completed, even if there is still time left in the `shutdownGracePeriod`. In our case, the `shutdownGracePeriod` is set to 30 seconds, however we see that the node got shutdown in 10 seconds, even while unmount has not succeeded.

kubelet.log
```
Jan 17 21:48:18 ip-10-0-1-208 kubelet[950]: I0117 21:48:18.460409     950 setters.go:548] "Node became not ready" node="ip-10-0-1-208" condition={Type:Ready Status:False LastHeartbeatTime:2023-01-17 21:48:18.46035065 +0000 UTC m=+1235.025577092 LastTransitionTime:2023-01-17 21:48:18.46035065 +0000 UTC m=+1235.025577092 Reason:KubeletNotReady Message:node is shutting down}
...
Jan 17 21:48:27 ip-10-0-1-208 kubelet[950]: E0117 21:48:27.126414     950 nestedpendingoperations.go:348] Operation for "{volumeName:kubernetes.io/csi/ebs.csi.aws.com^vol-00477959568e1db98 podName: nodeName:}" failed. No retries permitted until 2023-01-17 21:48:35.126400526 +0000 UTC m=+1251.691626967 (durationBeforeRetry 8s). Error: UnmountDevice failed for volume "pvc-513e457c-86c1-4564-911c-6c0670cb839c" (UniqueName: "kubernetes.io/csi/ebs.csi.aws.com^vol-00477959568e1db98") on node "ip-10-0-1-208" : kubernetes.io/csi: attacher.UnmountDevice failed to create newCsiDriverClient: driver name ebs.csi.aws.com not found in the list of registered CSI drivers
...
Jan 17 21:48:29 ip-10-0-1-208 kubelet[950]: E0117 21:48:29.006761     950 kubelet.go:2475] "Container runtime network not ready" networkReady="NetworkReady=false reason:NetworkPluginNotReady message:Network plugin returns error: cni plugin not initialized"
Connection to 10.0.1.208 closed by remote host.
Connection to 10.0.1.208 closed.
```

Some code tracing:

1. The logic for releasing the inhibitor lock waits for pods to be terminated within the grace period. 

https://github.com/kubernetes/kubernetes/blob/37e73b419e455db34f5fe3e8d815418680ab23df/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go#L326

2. It depends on `IsTerminated` to indicate that the pod has been killed: https://github.com/kubernetes/kubernetes/blob/37e73b419e455db34f5fe3e8d815418680ab23df/pkg/kubelet/pod_workers.go#L676

3. `IsTerminated` depends on `terminatedAt`, which is set when all containers have stopped:
https://github.com/kubernetes/kubernetes/blob/37e73b419e455db34f5fe3e8d815418680ab23df/pkg/kubelet/pod_workers.go#L273

4. However, volume termination for a Pod only begins when all containers have stopped:
https://github.com/kubernetes/kubernetes/blob/37e73b419e455db34f5fe3e8d815418680ab23df/pkg/kubelet/volumemanager/populator/desired_state_of_world_populator.go#L249


### What did you expect to happen?

Node graceful shutdown to also wait for volumes to be torn down.

### How can we reproduce it (as minimally and precisely as possible)?

Deploy a StatefulSet pod using an attachable PVC, and then trigger node shutdown. 

### Anything else we need to know?

_No response_

### Kubernetes version

<details>

Tested on 1.25, 1.26

</details>


### Cloud provider

<details>
Observed on GCP, AWS
</details>


### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>



Discussion / hints:
/sig node
/sig storage
/cc
/cc @sonasingh46 
For this CSI driver, what do the `preStop` hooks do?

I'm wondering if there's an opportunity to detect the shutdown and perform cleanup, or at least a more graceful detach.
CSI drivers depend on kubelet (volume manager) to tell it when it's safe to do an unmount. The key signal for volume manager is that the Pod's containers have terminated. We do not want to unmount a volume while a container could still be writing to it. 

Here's a diagram I made that illustrates the race.  The main challenge is that node shutdown manager and volume manager are operating off of the same signals (pod terminated) concurrently.

![Blank diagram (1)](https://user-images.githubusercontent.com/24448061/213246295-af0c53e0-e443-4da8-87ff-60dd794cc6be.png)

There's two possible solutions I can see:
1. preStop or sig-term handler in the CSI driver to detect and wait for all mounts to be unmounted. We would have to recommend that every CSI driver implement this.
2. node shutdown manager also considers volume cleanup as a signal
I'd imagined that the CSI driver could be aware of all the mounted volumes for that particular driver, and complete its `preStop` once those are all unpublished / unstaged. Is that viable?
Many CSI drivers don't keep state of what's mounted. It potentially could do that by scanning mounts on the host.
/triage accepted
Adding on top of https://github.com/kubernetes/kubernetes/issues/115148#issuecomment-1387425026, I think the "wait for volumes unmounted" workaround (in either preStop hook or CSI Driver SIGTERM handler impls) might not work in the upgrade scenario:

- Conditions
  - node has CSI Driver v1 with a "wait for volumes unmounted" impl
  - there's a Pod p1 running with a volume attached
- Daemonset is updated to CSI Driver v2
- kubelet attempts to delete the CSI Driver v1 pod, because of the "wait for volumes unmounted" logic it'll block until it all volumes are unmounted, in reality Pod p1 is still running fine and doesn't need its volume to be unmounted, therefore the CSI Driver v1 pod termination will be delayed until it's forcefully terminated

This could be a reason why we should pursue the solution: "node shutdown manager also considers volume cleanup as a signal"
If normal-priority Pod is _still running_, then the `preStop` hook I'm imagining has no cleanup to do there. I'm picturing a situation where a CSI driver is shutting down, there are now _no_ running containers that require storage from the driver, but the local cleanup for the mounted PersistentVolumes / supporting infrastructure has not completed.

In other words, this cleanup should happen after _kill normal pods_ as illustrated in https://github.com/kubernetes/kubernetes/issues/115148#issuecomment-1387425026, and before the kubelet considers that all Pods have terminated.

As @mauriciopoppe pointed out, the challenge with adding a prestop/sigterm hook is that there are situations when the driver pod is shutting down but user's pods + mounts are expected to stay up and running. The most common scenario would be when you're doing a rolling update of the csi driver daemonset.
@msau42 are you also saying that it's difficult for a CSI driver to tell if it's in use? I'm suggesting that a `preStop` hook could use that information (whether or not any running containers are still using the driver) and adjust its behavior accordingly.

For example, a Pod shutdown due to a DaemonSet rolling update could do an early exit during `preStop` after detecting that a running Pod (or container - either is fine) depends on the storage that this driver Pod is providing.
> are you also saying that it's difficult for a CSI driver to tell if it's in use? I'm suggesting that a `preStop` hook could use that information (whether or not any running containers are still using the driver) and adjust its behavior accordingly.
> 
> For example, a Pod shutdown due to a DaemonSet rolling update could do an early exit during `preStop` after detecting that a running Pod (or container - either is fine) depends on the storage that this driver Pod is providing.

Yes, this is difficult because CSI drivers are designed to be Kubernetes-agnostic. So things like listing and inspecting Pods are not things we require CSI drivers to do. In addition, the check would be racy as pods comes and go, and also would be difficult to scale (we try to avoid per-node components making expensive List calls) 


:thought_balloon: is there a way that the kubelet or host OS can provide extra context to the DaemonSet Pods?

Analogy: `/etc/nologin` which marks that a POSIX-like machine shouldn't allow logins. If either a systemd unit or the kubelet were to create a file as soon as the node begins a graceful shutdown (possibly also other actions, like a full drain), then we could mount that path into CSI driver Pods and allow them to customize their behavior.

Other parts of the Kubernetes ecosystem, not just storage, might find it useful to know that the kubelet is shutting down all Pods in sequence. For example, a network driver might [unarp](https://datatracker.ietf.org/doc/html/rfc1868) or release IP addresses if the driver Pod termination is triggered by a node shutdown. Lots of other examples.

Separately, but also relevant: if you want to set up a node so that stopping the kubelet does trigger a local drain, this change I'm imagining could be part of that design - and only needs minimal or no changes to the kubelet itself.
Having the kubelet advertise some sort of file-based signal could be an option.

From the CSI perspective, I think a vast majority of drivers would all want the same behavior and would end up implementing the same logic. Given that there are [over 100 CSI drivers](https://kubernetes-csi.github.io/docs/drivers.html), I would prefer to pursue an option that makes it easier to maintain consistent behavior.
I'm sharing an idea that I posted internally, tldr: it's about adding a flag to the KubeletConfiguration in `shutdownGracePeriodByPodPriority` to respect the shutdownGracePeriod time, we can configure workloads to be in a group that node shutdown manager will terminate (and wait for the respective shutdownGracePeriod) and critical workloads like CSI Drivers to be terminated later in a different group, with that we give the CSI Driver the chance to complete the volume teardown CSI calls

---

[processShutdownEvent](https://github.com/kubernetes/kubernetes/blob/cc60df95959bc4e6422ff05dc3d7c6ba3817bff0/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go#L321) does the following:
- Get pods and group them by priority
  - Grouping configuration is set through the [KubeletConfiguration shutdown* keys](https://kubernetes.io/docs/reference/config-api/kubelet-config.v1beta1/#:~:text=pods.%20Default%3A%20%220s%22-,shutdownGracePeriodByPodPriority,-%5B%5DShutdownGracePeriodByPodPriority), we can set (shutdownGracePeriod and shutdownGracePeriodCriticalPods) or (shutdownGracePeriodByPodPriority), if (shutdownGracePeriod and shutdownGracePeriodCriticalPods) are set then in the code these are migrated to the (shutdownGracePeriodByPodPriority) config anyways. 
- Let's assume that we have 2 groups (workload group where the StatefulSet is at and critical group where the CSI Driver is at), https://github.com/kubernetes/kubernetes/blob/cc60df95959bc4e6422ff05dc3d7c6ba3817bff0/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go#L355-L365 is iterating over all the Pods in a group and in a goroutine for each one is terminating them, the overall group termination completion is controlled with a sync.WaitGroup, at the end we'll either timeout with the config timeout (i.e. the shutdownGracePeriod for this group) or when all the pods are terminated https://github.com/kubernetes/kubernetes/blob/cc60df95959bc4e6422ff05dc3d7c6ba3817bff0/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go#L404-L414.
- The idea is to add an additional parameter to ShutdownGracePeriodByPodPriority https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/apis/config/types.go#L642 that enables respecting the shutdownGracePeriod time even if Pods in this group completed, if that's set then in https://github.com/kubernetes/kubernetes/blob/cc60df95959bc4e6422ff05dc3d7c6ba3817bff0/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go#L404-L414 we'd no longer react to the sync.WaitGroup completion and only the timeout

KubeletConfiguration example:

```yaml
kind: KubeletConfiguration
shutdownGracePeriodByPodPriority:
  - priority: 0
    shutdownGracePeriodSeconds: 10s
    respectShutdownGracePeriod: true  #new
  - priority: inf
    shutdownGracePeriodSeconds: 20s
```

```golang
 loop:
  for {
                select {
                case <-doneCh:
                        if group.RespectShutdownGracePeriod {
                           // we don't react to the sync.WaitGroup finish
                           // instead wait for the timer to signal
                           break
                        }
                        timer.Stop()
                        break loop                       
                case <-timer.C():
                        m.logger.V(1).Info("Shutdown manager pod killing time out", "gracePeriod", group.ShutdownGracePeriodSeconds, "priority", group.Priority)
                         break loop
                }
  }
```

Once we add context cancellation, we should be much closer to terminating pods in deterministic time (right now a long syncPod can take longer than graceful shutdown), which means we could and should also allocate some time for other subsystems of the node to reach terminal state.  I'll keep that in mind when we look at components downstream of pod worker pointing at pod worker state instead of pod manager.
This issue has not been updated in over 1 year, and should be re-triaged.

You can:
- Confirm that this issue is still relevant with `/triage accepted` (org members only)
- Close this issue with `/close`

For more details on the triage process, see https://www.kubernetes.dev/docs/guide/issue-triage/

/remove-triage accepted
/triage accepted

Fix patch:
diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go
index 6f602eb425c77..4b591aef33037 100644
--- a/pkg/kubelet/kubelet.go
+++ b/pkg/kubelet/kubelet.go
@@ -931,6 +931,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	shutdownManager, shutdownAdmitHandler := nodeshutdown.NewManager(&nodeshutdown.Config{
 		Logger:                           logger,
 		ProbeManager:                     klet.probeManager,
+		VolumeManager:                    klet.volumeManager,
 		Recorder:                         kubeDeps.Recorder,
 		NodeRef:                          nodeRef,
 		GetPodsFunc:                      klet.GetActivePods,
diff --git a/pkg/kubelet/nodeshutdown/nodeshutdown_manager.go b/pkg/kubelet/nodeshutdown/nodeshutdown_manager.go
index aac590f967dab..0b5a5a244c62e 100644
--- a/pkg/kubelet/nodeshutdown/nodeshutdown_manager.go
+++ b/pkg/kubelet/nodeshutdown/nodeshutdown_manager.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/eviction"
 	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/prober"
+	"k8s.io/kubernetes/pkg/kubelet/volumemanager"
 	"k8s.io/utils/clock"
 )
 
@@ -40,6 +41,7 @@ type Manager interface {
 type Config struct {
 	Logger                           klog.Logger
 	ProbeManager                     prober.Manager
+	VolumeManager                    volumemanager.VolumeManager
 	Recorder                         record.EventRecorder
 	NodeRef                          *v1.ObjectReference
 	GetPodsFunc                      eviction.ActivePodsFunc
diff --git a/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go b/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go
index a5fc6f9583f6e..e56d03155c28b 100644
--- a/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go
+++ b/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux.go
@@ -21,6 +21,7 @@ limitations under the License.
 package nodeshutdown
 
 import (
+	"context"
 	"fmt"
 	"path/filepath"
 	"sort"
@@ -41,6 +42,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	"k8s.io/kubernetes/pkg/kubelet/nodeshutdown/systemd"
 	"k8s.io/kubernetes/pkg/kubelet/prober"
+	"k8s.io/kubernetes/pkg/kubelet/volumemanager"
 	"k8s.io/utils/clock"
 )
 
@@ -73,6 +75,8 @@ type managerImpl struct {
 	nodeRef      *v1.ObjectReference
 	probeManager prober.Manager
 
+	volumeManager volumemanager.VolumeManager
+
 	shutdownGracePeriodByPodPriority []kubeletconfig.ShutdownGracePeriodByPodPriority
 
 	getPods        eviction.ActivePodsFunc
@@ -123,6 +127,7 @@ func NewManager(conf *Config) (Manager, lifecycle.PodAdmitHandler) {
 		logger:                           conf.Logger,
 		probeManager:                     conf.ProbeManager,
 		recorder:                         conf.Recorder,
+		volumeManager:                    conf.VolumeManager,
 		nodeRef:                          conf.NodeRef,
 		getPods:                          conf.GetPodsFunc,
 		killPodFunc:                      conf.KillPodFunc,
@@ -395,19 +400,44 @@ func (m *managerImpl) processShutdownEvent() error {
 			}(pod, group)
 		}
 
+		// This duration determines how long the shutdown manager will wait for the pods in this group
+		// to terminate before proceeding to the next group.
+		var groupTerminationWaitDuration = time.Duration(group.ShutdownGracePeriodSeconds) * time.Second
 		var (
-			doneCh = make(chan struct{})
-			timer  = m.clock.NewTimer(time.Duration(group.ShutdownGracePeriodSeconds) * time.Second)
+			doneCh         = make(chan struct{})
+			timer          = m.clock.NewTimer(groupTerminationWaitDuration)
+			ctx, ctxCancel = context.WithTimeout(context.Background(), groupTerminationWaitDuration)
 		)
 		go func() {
 			defer close(doneCh)
+			defer ctxCancel()
 			wg.Wait()
+			// The signal to kill a Pod was sent successfully to all the pods,
+			// let's wait until all the volumes are unmounted from all the pods before
+			// continuing to the next group. This is done so that the CSI Driver (assuming
+			// that it's part of the highest group) has a chance to perform unmounts.
+			if err := m.volumeManager.WaitForAllPodsUnmount(ctx, group.Pods); err != nil {
+				var podIdentifiers []string
+				for _, pod := range group.Pods {
+					podIdentifiers = append(podIdentifiers, fmt.Sprintf("%s/%s", pod.Namespace, pod.Name))
+				}
+
+				// Waiting for volume teardown is done on a best basis effort,
+				// report an error and continue.
+				//
+				// Depending on the user provided kubelet configuration value
+				// either the `timer` will tick and we'll continue to shutdown the next group, or,
+				// WaitForAllPodsUnmount will timeout, therefore this goroutine
+				// will close doneCh and we'll continue to shutdown the next group.
+				m.logger.Error(err, "Failed while waiting for all the volumes belonging to Pods in this group to unmount", "pods", podIdentifiers)
+			}
 		}()
 
 		select {
 		case <-doneCh:
 			timer.Stop()
 		case <-timer.C():
+			ctxCancel()
 			m.logger.V(1).Info("Shutdown manager pod killing time out", "gracePeriod", group.ShutdownGracePeriodSeconds, "priority", group.Priority)
 		}
 	}
diff --git a/pkg/kubelet/volumemanager/volume_manager.go b/pkg/kubelet/volumemanager/volume_manager.go
index 4c2f8eedb9c78..abe2fd14e5b9f 100644
--- a/pkg/kubelet/volumemanager/volume_manager.go
+++ b/pkg/kubelet/volumemanager/volume_manager.go
@@ -23,6 +23,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
@@ -31,6 +32,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	k8stypes "k8s.io/apimachinery/pkg/types"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -106,6 +108,12 @@ type VolumeManager interface {
 	// the duration defined in podAttachAndMountTimeout.
 	WaitForUnmount(ctx context.Context, pod *v1.Pod) error
 
+	// WaitForAllPodsUnmount is a version of WaitForUnmount that blocks and
+	// waits until all the volumes belonging to all the pods are unmounted.
+	// An error is returned if there's at least one Pod with volumes not unmounted
+	// within the duration defined in podAttachAndMountTimeout.
+	WaitForAllPodsUnmount(ctx context.Context, pods []*v1.Pod) error
+
 	// GetMountedVolumesForPod returns a VolumeMap containing the volumes
 	// referenced by the specified pod that are successfully attached and
 	// mounted. The key in the map is the OuterVolumeSpecName (i.e.
@@ -479,6 +487,24 @@ func (vm *volumeManager) WaitForUnmount(ctx context.Context, pod *v1.Pod) error
 	return nil
 }
 
+func (vm *volumeManager) WaitForAllPodsUnmount(ctx context.Context, pods []*v1.Pod) error {
+	var (
+		errors []error
+		wg     sync.WaitGroup
+	)
+	wg.Add(len(pods))
+	for _, pod := range pods {
+		go func(pod *v1.Pod) {
+			defer wg.Done()
+			if err := vm.WaitForUnmount(ctx, pod); err != nil {
+				errors = append(errors, err)
+			}
+		}(pod)
+	}
+	wg.Wait()
+	return utilerrors.NewAggregate(errors)
+}
+
 func (vm *volumeManager) getVolumesNotInDSW(uniquePodName types.UniquePodName, expectedVolumes []string) []string {
 	volumesNotInDSW := sets.New[string](expectedVolumes...)
 
diff --git a/pkg/kubelet/volumemanager/volume_manager_fake.go b/pkg/kubelet/volumemanager/volume_manager_fake.go
index d7d4e1b22cac8..02c424c380154 100644
--- a/pkg/kubelet/volumemanager/volume_manager_fake.go
+++ b/pkg/kubelet/volumemanager/volume_manager_fake.go
@@ -18,6 +18,7 @@ package volumemanager
 
 import (
 	"context"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/kubernetes/pkg/kubelet/config"
@@ -29,10 +30,14 @@ import (
 type FakeVolumeManager struct {
 	volumes       map[v1.UniqueVolumeName]bool
 	reportedInUse map[v1.UniqueVolumeName]bool
+	unmountDelay  time.Duration
+	unmountError  error
 }
 
+var _ VolumeManager = &FakeVolumeManager{}
+
 // NewFakeVolumeManager creates a new VolumeManager test instance
-func NewFakeVolumeManager(initialVolumes []v1.UniqueVolumeName) *FakeVolumeManager {
+func NewFakeVolumeManager(initialVolumes []v1.UniqueVolumeName, unmountDelay time.Duration, unmountError error) *FakeVolumeManager {
 	volumes := map[v1.UniqueVolumeName]bool{}
 	for _, v := range initialVolumes {
 		volumes[v] = true
@@ -40,6 +45,8 @@ func NewFakeVolumeManager(initialVolumes []v1.UniqueVolumeName) *FakeVolumeManag
 	return &FakeVolumeManager{
 		volumes:       volumes,
 		reportedInUse: map[v1.UniqueVolumeName]bool{},
+		unmountDelay:  unmountDelay,
+		unmountError:  unmountError,
 	}
 }
 
@@ -57,6 +64,15 @@ func (f *FakeVolumeManager) WaitForUnmount(ctx context.Context, pod *v1.Pod) err
 	return nil
 }
 
+func (f *FakeVolumeManager) WaitForAllPodsUnmount(ctx context.Context, pods []*v1.Pod) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-time.After(f.unmountDelay):
+		return f.unmountError
+	}
+}
+
 // GetMountedVolumesForPod is not implemented
 func (f *FakeVolumeManager) GetMountedVolumesForPod(podName types.UniquePodName) container.VolumeMap {
 	return nil


Test patch:
diff --git a/pkg/kubelet/kubelet_node_status_test.go b/pkg/kubelet/kubelet_node_status_test.go
index 2cee198ecf756..af09cc9d34118 100644
--- a/pkg/kubelet/kubelet_node_status_test.go
+++ b/pkg/kubelet/kubelet_node_status_test.go
@@ -1124,7 +1124,7 @@ func TestUpdateNodeStatusAndVolumesInUseWithNodeLease(t *testing.T) {
 			kubelet.setCachedMachineInfo(&cadvisorapi.MachineInfo{})
 
 			// override test volumeManager
-			fakeVolumeManager := kubeletvolume.NewFakeVolumeManager(tc.existingVolumes)
+			fakeVolumeManager := kubeletvolume.NewFakeVolumeManager(tc.existingVolumes, 0, nil)
 			kubelet.volumeManager = fakeVolumeManager
 
 			// Only test VolumesInUse setter
diff --git a/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux_test.go b/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux_test.go
index 65bfe014c840e..1566153b9e894 100644
--- a/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux_test.go
+++ b/pkg/kubelet/nodeshutdown/nodeshutdown_manager_linux_test.go
@@ -30,6 +30,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -45,6 +46,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/nodeshutdown/systemd"
 	"k8s.io/kubernetes/pkg/kubelet/prober"
 	probetest "k8s.io/kubernetes/pkg/kubelet/prober/testing"
+	"k8s.io/kubernetes/pkg/kubelet/volumemanager"
 	"k8s.io/utils/clock"
 	testingclock "k8s.io/utils/clock/testing"
 )
@@ -348,10 +350,12 @@ func TestManager(t *testing.T) {
 
 			proberManager := probetest.FakeManager{}
 			fakeRecorder := &record.FakeRecorder{}
+			fakeVolumeManager := volumemanager.NewFakeVolumeManager([]v1.UniqueVolumeName{}, 0, nil)
 			nodeRef := &v1.ObjectReference{Kind: "Node", Name: "test", UID: types.UID("test"), Namespace: ""}
 			manager, _ := NewManager(&Config{
 				Logger:                          logger,
 				ProbeManager:                    proberManager,
+				VolumeManager:                   fakeVolumeManager,
 				Recorder:                        fakeRecorder,
 				NodeRef:                         nodeRef,
 				GetPodsFunc:                     activePodsFunc,
@@ -452,11 +456,13 @@ func TestFeatureEnabled(t *testing.T) {
 
 			proberManager := probetest.FakeManager{}
 			fakeRecorder := &record.FakeRecorder{}
+			fakeVolumeManager := volumemanager.NewFakeVolumeManager([]v1.UniqueVolumeName{}, 0, nil)
 			nodeRef := &v1.ObjectReference{Kind: "Node", Name: "test", UID: types.UID("test"), Namespace: ""}
 
 			manager, _ := NewManager(&Config{
 				Logger:                          logger,
 				ProbeManager:                    proberManager,
+				VolumeManager:                   fakeVolumeManager,
 				Recorder:                        fakeRecorder,
 				NodeRef:                         nodeRef,
 				GetPodsFunc:                     activePodsFunc,
@@ -509,10 +515,12 @@ func TestRestart(t *testing.T) {
 
 	proberManager := probetest.FakeManager{}
 	fakeRecorder := &record.FakeRecorder{}
+	fakeVolumeManager := volumemanager.NewFakeVolumeManager([]v1.UniqueVolumeName{}, 0, nil)
 	nodeRef := &v1.ObjectReference{Kind: "Node", Name: "test", UID: types.UID("test"), Namespace: ""}
 	manager, _ := NewManager(&Config{
 		Logger:                          logger,
 		ProbeManager:                    proberManager,
+		VolumeManager:                   fakeVolumeManager,
 		Recorder:                        fakeRecorder,
 		NodeRef:                         nodeRef,
 		GetPodsFunc:                     activePodsFunc,
@@ -738,17 +746,19 @@ func Test_groupByPriority(t *testing.T) {
 
 func Test_managerImpl_processShutdownEvent(t *testing.T) {
 	var (
-		probeManager   = probetest.FakeManager{}
-		fakeRecorder   = &record.FakeRecorder{}
-		syncNodeStatus = func() {}
-		nodeRef        = &v1.ObjectReference{Kind: "Node", Name: "test", UID: types.UID("test"), Namespace: ""}
-		fakeclock      = testingclock.NewFakeClock(time.Now())
+		probeManager      = probetest.FakeManager{}
+		fakeRecorder      = &record.FakeRecorder{}
+		fakeVolumeManager = volumemanager.NewFakeVolumeManager([]v1.UniqueVolumeName{}, 0, nil)
+		syncNodeStatus    = func() {}
+		nodeRef           = &v1.ObjectReference{Kind: "Node", Name: "test", UID: types.UID("test"), Namespace: ""}
+		fakeclock         = testingclock.NewFakeClock(time.Now())
 	)
 
 	type fields struct {
 		recorder                         record.EventRecorder
 		nodeRef                          *v1.ObjectReference
 		probeManager                     prober.Manager
+		volumeManager                    volumemanager.VolumeManager
 		shutdownGracePeriodByPodPriority []kubeletconfig.ShutdownGracePeriodByPodPriority
 		getPods                          eviction.ActivePodsFunc
 		killPodFunc                      eviction.KillPodFunc
@@ -767,9 +777,10 @@ func Test_managerImpl_processShutdownEvent(t *testing.T) {
 		{
 			name: "kill pod func take too long",
 			fields: fields{
-				recorder:     fakeRecorder,
-				nodeRef:      nodeRef,
-				probeManager: probeManager,
+				recorder:      fakeRecorder,
+				nodeRef:       nodeRef,
+				probeManager:  probeManager,
+				volumeManager: fakeVolumeManager,
 				shutdownGracePeriodByPodPriority: []kubeletconfig.ShutdownGracePeriodByPodPriority{
 					{
 						Priority:                   1,
@@ -808,6 +819,7 @@ func Test_managerImpl_processShutdownEvent(t *testing.T) {
 			)
 			m := &managerImpl{
 				logger:                           logger,
+				volumeManager:                    tt.fields.volumeManager,
 				recorder:                         tt.fields.recorder,
 				nodeRef:                          tt.fields.nodeRef,
 				probeManager:                     tt.fields.probeManager,
@@ -839,3 +851,65 @@ func Test_managerImpl_processShutdownEvent(t *testing.T) {
 		})
 	}
 }
+
+func Test_processShutdownEvent_VolumeUnmountTimeout(t *testing.T) {
+	var (
+		probeManager               = probetest.FakeManager{}
+		fakeRecorder               = &record.FakeRecorder{}
+		syncNodeStatus             = func() {}
+		nodeRef                    = &v1.ObjectReference{Kind: "Node", Name: "test", UID: types.UID("test"), Namespace: ""}
+		fakeclock                  = testingclock.NewFakeClock(time.Now())
+		shutdownGracePeriodSeconds = 2
+	)
+
+	fakeVolumeManager := volumemanager.NewFakeVolumeManager(
+		[]v1.UniqueVolumeName{},
+		3*time.Second, // This value is intentionally longer than the shutdownGracePeriodSeconds (2s) to test the behavior
+		// for volume unmount operations that take longer than the allowed grace period.
+		fmt.Errorf("unmount timeout"),
+	)
+	logger := ktesting.NewLogger(t, ktesting.NewConfig(ktesting.BufferLogs(true)))
+	m := &managerImpl{
+		logger:        logger,
+		volumeManager: fakeVolumeManager,
+		recorder:      fakeRecorder,
+		nodeRef:       nodeRef,
+		probeManager:  probeManager,
+		shutdownGracePeriodByPodPriority: []kubeletconfig.ShutdownGracePeriodByPodPriority{
+			{
+				Priority:                   1,
+				ShutdownGracePeriodSeconds: int64(shutdownGracePeriodSeconds),
+			},
+		},
+		getPods: func() []*v1.Pod {
+			return []*v1.Pod{
+				makePod("test-pod", 1, nil),
+			}
+		},
+		killPodFunc: func(pod *v1.Pod, isEvicted bool, gracePeriodOverride *int64, fn func(*v1.PodStatus)) error {
+			return nil
+		},
+		syncNodeStatus: syncNodeStatus,
+		dbusCon:        &fakeDbus{},
+		clock:          fakeclock,
+	}
+
+	start := fakeclock.Now()
+	err := m.processShutdownEvent()
+	end := fakeclock.Now()
+
+	require.NoError(t, err, "managerImpl.processShutdownEvent() should not return an error")
+
+	// Check if processShutdownEvent completed within the expected time
+	actualDuration := int(end.Sub(start).Seconds())
+	assert.LessOrEqual(t, actualDuration, shutdownGracePeriodSeconds, "processShutdownEvent took too long")
+
+	underlier, ok := logger.GetSink().(ktesting.Underlier)
+	if !ok {
+		t.Fatalf("Should have had a ktesting LogSink, got %T", logger.GetSink())
+	}
+
+	log := underlier.GetBuffer().String()
+	expectedLogMessage := "Failed while waiting for all the volumes belonging to Pods in this group to unmount"
+	assert.Contains(t, log, expectedLogMessage, "Expected log message not found")
+}
diff --git a/pkg/kubelet/volumemanager/volume_manager_test.go b/pkg/kubelet/volumemanager/volume_manager_test.go
index 8c8a4372936b8..e46649b430725 100644
--- a/pkg/kubelet/volumemanager/volume_manager_test.go
+++ b/pkg/kubelet/volumemanager/volume_manager_test.go
@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kubetypes "k8s.io/apimachinery/pkg/types"
@@ -86,7 +87,11 @@ func TestGetMountedVolumesForPodAndGetVolumesInUse(t *testing.T) {
 			if err != nil {
 				t.Fatalf("can't make a temp dir: %v", err)
 			}
-			defer os.RemoveAll(tmpDir)
+			defer func() {
+				if err := os.RemoveAll(tmpDir); err != nil {
+					t.Fatalf("failed to remove temp dir: %v", err)
+				}
+			}()
 			podManager := kubepod.NewBasicPodManager()
 
 			node, pod, pv, claim := createObjects(test.pvMode, test.podMode)
@@ -545,3 +550,65 @@ func delayClaimBecomesBound(
 	}
 	kubeClient.CoreV1().PersistentVolumeClaims(namespace).Update(context.TODO(), volumeClaim, metav1.UpdateOptions{})
 }
+
+func TestWaitForAllPodsUnmount(t *testing.T) {
+	tmpDir, err := utiltesting.MkTmpdir("volumeManagerTest")
+	require.NoError(t, err, "Failed to create temp directory")
+	defer func() {
+		if err := os.RemoveAll(tmpDir); err != nil {
+			t.Errorf("Failed to remove temp directory: %v", err)
+		}
+	}()
+
+	tests := []struct {
+		name          string
+		podMode       v1.PersistentVolumeMode
+		expectedError bool
+	}{
+		{
+			name:          "successful unmount",
+			podMode:       "",
+			expectedError: false,
+		},
+		{
+			name:          "timeout waiting for unmount",
+			podMode:       v1.PersistentVolumeFilesystem,
+			expectedError: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			podManager := kubepod.NewBasicPodManager()
+
+			node, pod, pv, claim := createObjects(test.podMode, test.podMode)
+			kubeClient := fake.NewSimpleClientset(node, pod, pv, claim)
+
+			manager := newTestVolumeManager(t, tmpDir, podManager, kubeClient, node)
+
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+			defer cancel()
+			sourcesReady := config.NewSourcesReady(func(_ sets.Set[string]) bool { return true })
+			go manager.Run(ctx, sourcesReady)
+
+			podManager.SetPods([]*v1.Pod{pod})
+
+			go simulateVolumeInUseUpdate(
+				v1.UniqueVolumeName(node.Status.VolumesAttached[0].Name),
+				ctx.Done(),
+				manager)
+
+			err := manager.WaitForAttachAndMount(context.Background(), pod)
+			require.NoError(t, err, "Failed to wait for attach and mount")
+
+			err = manager.WaitForAllPodsUnmount(ctx, []*v1.Pod{pod})
+
+			if test.expectedError {
+				require.Error(t, err, "Expected error due to timeout")
+				require.Contains(t, err.Error(), "context deadline exceeded", "Expected deadline exceeded error")
+			} else {
+				require.NoError(t, err, "Expected no error")
+			}
+		})
+	}
+}