Repository: kubernetes/kubernetes (language: go)

Issue:
Pod stuck at Running state due to unexpected skip of taint manager work
### What happened?

1、We shutdown the worker node that our pods is running on
2、A pod stays in Running state and never change again



### What did you expect to happen?

We expected the pod to be update to Terminating.

### How can we reproduce it (as minimally and precisely as possible)?

It is unlikely to be reproduced since we found the 'only possible' cause to be of ridiculously low probability.

### Anything else we need to know?

The pod is defined with tolerations as below, but doesn't seem to be a problem:
```
tolerations:
  - effect: NoExecute
    key: node.kubernetes.io/not-ready
    operator: Exists
    tolerationSeconds: 2
  - effect: NoExecute
    key: node.kubernetes.io/unreachable
    operator: Exists
    tolerationSeconds: 2 
```
Strange log sequence was found in controller-manager's log
```
I0604 15:28:49.239218       1 timed_workers.go:113] Adding TimedWorkerQueue item dpa/infrastructure-0 at 2024-06-04 15:28:49.239118259 +0800 CST m=+252387.993120493 to be fired at 2024-06-04 15:28:51.239118259 +0800 CST m=+252389.993120493
......
I0604 15:28:51.240319       1 taint_manager.go:106] "NoExecuteTaintManager is deleting pod" pod="dpa/infrastructure-0"
I0604 15:28:51.259631       1 taint_manager.go:416] "Noticed pod deletion" pod="dpa/infrastructure-0"
I0604 15:28:51.259642       1 timed_workers.go:132] Cancelling TimedWorkerQueue item dpa/infrastructure-0 at 2024-06-04 15:28:51.259639403 +0800 CST m=+252390.013641637
......
I0604 17:38:00.725378       1 timed_workers.go:113] Adding TimedWorkerQueue item dpa/infrastructure-0 at 2024-06-04 17:38:00.725348125 +0800 CST m=+260139.479350359 to be fired at 2024-06-04 17:38:02.725348125 +0800 CST m=+260141.479350359
W0604 17:38:00.725393       1 timed_workers.go:118] Trying to add already existing work for &{NamespacedName:dpa/infrastructure-0}. Skipping.
......
```
We noticed that we had restarted the node once at arround 15:28, and that trigger the eviction of pod dpa/infrastructure-0 as expected. But when we shutdown the node again at arround 17:38, the taint manager skipped the work due to the same key found in workers map.
The controller manager was never restarted or changed leader.
The log is ridiculous, but after look at the code, we did find a way that a key stays in workers map after Cancelling:
```
func (q *TimedWorkerQueue) getWrappedWorkerFunc(key string) func(ctx context.Context, args *WorkArgs) error {
	return func(ctx context.Context, args *WorkArgs) error {
		err := q.workFunc(ctx, args)
		q.Lock()
		defer q.Unlock()
		if err == nil {
			// To avoid duplicated calls we keep the key in the queue, to prevent
			// subsequent additions.
			q.workers[key] = nil
		} else {
			delete(q.workers, key)
		}
		return err
	}
}
```
In this part of code generates a WorkerFunc that would be called in a work, taint manager will run q.workFunc to delete pod from apiserver, which would trigger a podUpdate and lead to CancelWork:
```
func (q *TimedWorkerQueue) CancelWork(key string) bool {
	q.Lock()
	defer q.Unlock()
	worker, found := q.workers[key]
	result := false
	if found {
		klog.V(4).Infof("Cancelling TimedWorkerQueue item %v at %v", key, time.Now())
		if worker != nil {
			result = true
			worker.Cancel()
		}
		delete(q.workers, key)
	}
	return result
}
```
If and only if CancelWork acquired the lock before WorkerFunc, the `q.workers[key]` won't be deleted as expected, but `q.workers[key] = nil` instead, and lead to later skip of work.

### Kubernetes version

<details>

```console
$ kubectl version
Client Version: version.Info{Major:"1", Minor:"25+", GitVersion:"v1.25.3+77d8f59014e1f3-dirty", GitCommit:"77d8f59014e1f3ca907c1b4a4f57900539b88dc8", GitTreeState:"dirty", BuildDate:"2024-04-29T13:26:15Z", GoVersion:"go1.19.6", Compiler:"gc", Platform:"linux/arm64"}
Kustomize Version: v4.5.7
Server Version: version.Info{Major:"1", Minor:"25+", GitVersion:"v1.25.3+77d8f59014e1f3-dirty", GitCommit:"77d8f59014e1f3ca907c1b4a4f57900539b88dc8", GitTreeState:"dirty", BuildDate:"2024-04-29T13:23:44Z", GoVersion:"go1.19.6", Compiler:"gc", Platform:"linux/arm64"}
```

</details>


### Cloud provider

<details>
None
</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 scheduling
/sig node
this seems to be a pretty old version of kube. is this reproducable on 1.28 or later?

@T-Lakshmi can I ask why you added sig node here?
We've reviewed the code, taint_manager.go and timed_worker.go has been moved to pkg/controller/tainteviction since 1.29, but the code content hasn't been changed

> this seems to be a pretty old version of kube. is this reproducable on 1.28 or later?
> 
> @T-Lakshmi can I ask why you added sig node here?


We found a same issue and a related PR #110402 which seems to be a good fix, can we reopen the PR?
/triage accepted

Reopened https://github.com/kubernetes/kubernetes/pull/110402
/priority backlog

doesn't look like a regression

Fix patch:
diff --git a/pkg/controller/tainteviction/timed_workers.go b/pkg/controller/tainteviction/timed_workers.go
index 8260c86ff360d..fa7615d9b9a19 100644
--- a/pkg/controller/tainteviction/timed_workers.go
+++ b/pkg/controller/tainteviction/timed_workers.go
@@ -105,13 +105,7 @@ func (q *TimedWorkerQueue) getWrappedWorkerFunc(key string) func(ctx context.Con
 		err := q.workFunc(ctx, fireAt, args)
 		q.Lock()
 		defer q.Unlock()
-		if err == nil {
-			// To avoid duplicated calls we keep the key in the queue, to prevent
-			// subsequent additions.
-			q.workers[key] = nil
-		} else {
-			delete(q.workers, key)
-		}
+		delete(q.workers, key)
 		return err
 	}
 }


Test patch:
diff --git a/pkg/controller/tainteviction/timed_workers_test.go b/pkg/controller/tainteviction/timed_workers_test.go
index 389ccd3ceb265..d39acaed44351 100644
--- a/pkg/controller/tainteviction/timed_workers_test.go
+++ b/pkg/controller/tainteviction/timed_workers_test.go
@@ -28,6 +28,7 @@ import (
 )
 
 func TestExecute(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
 	testVal := int32(0)
 	wg := sync.WaitGroup{}
 	wg.Add(5)
@@ -37,17 +38,11 @@ func TestExecute(t *testing.T) {
 		return nil
 	})
 	now := time.Now()
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, now)
-	// Adding the same thing second time should be no-op
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, now)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, now)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, now)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, now)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, now)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, now)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, now)
 	wg.Wait()
 	lastVal := atomic.LoadInt32(&testVal)
 	if lastVal != 5 {
@@ -56,6 +51,7 @@ func TestExecute(t *testing.T) {
 }
 
 func TestExecuteDelayed(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
 	testVal := int32(0)
 	wg := sync.WaitGroup{}
 	wg.Add(5)
@@ -68,16 +64,16 @@ func TestExecuteDelayed(t *testing.T) {
 	then := now.Add(10 * time.Second)
 	fakeClock := testingclock.NewFakeClock(now)
 	queue.clock = fakeClock
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, then)
 	fakeClock.Step(11 * time.Second)
 	wg.Wait()
 	lastVal := atomic.LoadInt32(&testVal)
@@ -87,6 +83,7 @@ func TestExecuteDelayed(t *testing.T) {
 }
 
 func TestCancel(t *testing.T) {
+	logger, ctx := ktesting.NewTestContext(t)
 	testVal := int32(0)
 	wg := sync.WaitGroup{}
 	wg.Add(3)
@@ -99,17 +96,16 @@ func TestCancel(t *testing.T) {
 	then := now.Add(10 * time.Second)
 	fakeClock := testingclock.NewFakeClock(now)
 	queue.clock = fakeClock
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, then)
-	logger, _ := ktesting.NewTestContext(t)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, then)
 	queue.CancelWork(logger, NewWorkArgs("2", "2").KeyFromWorkArgs())
 	queue.CancelWork(logger, NewWorkArgs("4", "4").KeyFromWorkArgs())
 	fakeClock.Step(11 * time.Second)
@@ -121,6 +117,7 @@ func TestCancel(t *testing.T) {
 }
 
 func TestCancelAndReadd(t *testing.T) {
+	logger, ctx := ktesting.NewTestContext(t)
 	testVal := int32(0)
 	wg := sync.WaitGroup{}
 	wg.Add(4)
@@ -133,20 +130,19 @@ func TestCancelAndReadd(t *testing.T) {
 	then := now.Add(10 * time.Second)
 	fakeClock := testingclock.NewFakeClock(now)
 	queue.clock = fakeClock
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("1", "1"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("3", "3"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("4", "4"), now, then)
-	queue.AddWork(context.TODO(), NewWorkArgs("5", "5"), now, then)
-	logger, _ := ktesting.NewTestContext(t)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("5", "5"), now, then)
 	queue.CancelWork(logger, NewWorkArgs("2", "2").KeyFromWorkArgs())
 	queue.CancelWork(logger, NewWorkArgs("4", "4").KeyFromWorkArgs())
-	queue.AddWork(context.TODO(), NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(ctx, NewWorkArgs("2", "2"), now, then)
 	fakeClock.Step(11 * time.Second)
 	wg.Wait()
 	lastVal := atomic.LoadInt32(&testVal)