UPSTREAM: <carry>: kubelet: fix readiness probes with pod termination

This patch merges both input channels into one.

cmd/kube-controller-manager/app/controllermanager.go

@@ -197,7 +197,7 @@ func ResyncPeriod(c *config.CompletedConfig) func() time.Duration {
 // Run runs the KubeControllerManagerOptions.
 func Run(ctx context.Context, c *config.CompletedConfig, stopCh2 <-chan struct{}) error {
 	logger := klog.FromContext(ctx)
-	stopCh := ctx.Done()
+	stopCh := mergeCh(ctx.Done(), stopCh2)
 
 	// To help debugging, immediately log version
 	logger.Info("Starting", "version", version.Get())
@@ -220,12 +220,8 @@ func Run(ctx context.Context, c *config.CompletedConfig, stopCh2 <-chan struct{}
 		hmCtx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 		go func() {
-			select {
-			case <-stopCh:
-				cancel()
-			case <-stopCh2:
-				cancel()
-			}
+			<-stopCh
+			cancel()
 		}()
 		go c.OpenShiftContext.PreferredHostHealthMonitor.Run(hmCtx)
 	}
@@ -333,10 +329,6 @@ func Run(ctx context.Context, c *config.CompletedConfig, stopCh2 <-chan struct{}
 					// We were asked to terminate. Exit 0.
 					klog.Info("Requested to terminate. Exiting.")
 					os.Exit(0)
-				case <-stopCh2:
-					// We were asked to terminate. Exit 0.
-					klog.Info("Requested to terminate. Exiting.")
-					os.Exit(0)
 				default:
 					// We lost the lock.
 					logger.Error(nil, "leaderelection lost")
@@ -369,10 +361,6 @@ func Run(ctx context.Context, c *config.CompletedConfig, stopCh2 <-chan struct{}
 						// We were asked to terminate. Exit 0.
 						klog.Info("Requested to terminate. Exiting.")
 						os.Exit(0)
-					case <-stopCh2:
-						// We were asked to terminate. Exit 0.
-						klog.Info("Requested to terminate. Exiting.")
-						os.Exit(0)
 					default:
 						// We lost the lock.
 						logger.Error(nil, "migration leaderelection lost")

cmd/kube-controller-manager/app/patch.go

@@ -164,3 +164,17 @@ func (rt *rejectIfNotReadyHeaderRT) RoundTrip(r *http.Request) (*http.Response,
 	}
 	return rt.baseRT.RoundTrip(r)
 }
+
+// mergeCh takes two stop channels and return a single one that
+// closes as soon as one of the inputs closes or receives data.
+func mergeCh(stopCh1, stopCh2 <-chan struct{}) <-chan struct{} {
+	merged := make(chan struct{})
+	go func() {
+		defer close(merged)
+		select {
+		case <-stopCh1:
+		case <-stopCh2:
+		}
+	}()
+	return merged
+}

cmd/kube-controller-manager/app/patch_test.go

@@ -72,3 +72,56 @@ type fakeRTFunc func(r *http.Request) (*http.Response, error)
 func (rt fakeRTFunc) RoundTrip(r *http.Request) (*http.Response, error) {
 	return rt(r)
 }
+
+func TestMergeCh(t *testing.T) {
+	testCases := []struct {
+		name    string
+		chan1   chan struct{}
+		chan2   chan struct{}
+		closeFn func(chan struct{}, chan struct{})
+	}{
+		{
+			name:  "chan1 gets closed",
+			chan1: make(chan struct{}),
+			chan2: make(chan struct{}),
+			closeFn: func(a, b chan struct{}) {
+				close(a)
+			},
+		},
+		{
+			name:  "chan2 gets closed",
+			chan1: make(chan struct{}),
+			chan2: make(chan struct{}),
+			closeFn: func(a, b chan struct{}) {
+				close(b)
+			},
+		},
+		{
+			name:  "both channels get closed",
+			chan1: make(chan struct{}),
+			chan2: make(chan struct{}),
+			closeFn: func(a, b chan struct{}) {
+				close(a)
+				close(b)
+			},
+		},
+		{
+			name:  "channel receives data and returned channel is closed",
+			chan1: make(chan struct{}),
+			chan2: make(chan struct{}),
+			closeFn: func(a, b chan struct{}) {
+				a <- struct{}{}
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			go tc.closeFn(tc.chan1, tc.chan2)
+			merged := mergeCh(tc.chan1, tc.chan2)
+			if _, ok := <-merged; ok {
+				t.Fatalf("expected closed channel, got data")
+			}
+		})
+	}
+}