Merge pull request #64896 from rphillips/fixes/kubectl_eviction

Automatic merge from submit-queue (batch tested with PRs 65776, 64896). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

kubectl: wait for all errors and successes on podEviction

**What this PR does / why we need it**: This fixes `kubectl drain` to wait until all errors and successes are processed, instead of returning the first error. It also tweaks the behavior of the cleanup to check to see if the pod is already terminating, and if it is to not reissue the pod terminate which leads to an error getting thrown. This fix will allow `kubectl drain` to complete successfully when a node is draining.

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #

**Special notes for your reviewer**:
/cc @sjenning 

**Release note**:
```release-note
NONE
```
#### Reproduction steps
### sleep.yml
```yaml
apiVersion: v1
kind: Pod
metadata:
  name: bash
spec: 
  containers:
  - name: bash
    image: bash
    resources:
      limits:
        cpu: 500m
        memory: 500Mi
    command:
    - bash
    - -c
    - "nothing() { sleep 1; } ; trap nothing 15 ; while true; do echo \"hello\"; sleep 10; done"
  terminationGracePeriodSeconds: 3000
  restartPolicy: Never
```

```
$ kubectl create ns testing
$ kubectl create -f sleep.yml
$ kubectl delete ns testing
$ kubectl drain 127.0.0.1 --force
```

pkg/kubectl/cmd/drain.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
@@ -572,38 +573,39 @@ func (o *DrainOptions) deleteOrEvictPods(pods []corev1.Pod) error {
 }
 
 func (o *DrainOptions) evictPods(pods []corev1.Pod, policyGroupVersion string, getPodFn func(namespace, name string) (*corev1.Pod, error)) error {
-	doneCh := make(chan bool, len(pods))
-	errCh := make(chan error, 1)
+	returnCh := make(chan error, 1)
 
 	for _, pod := range pods {
-		go func(pod corev1.Pod, doneCh chan bool, errCh chan error) {
+		go func(pod corev1.Pod, returnCh chan error) {
 			var err error
 			for {
 				err = o.evictPod(pod, policyGroupVersion)
 				if err == nil {
 					break
 				} else if apierrors.IsNotFound(err) {
-					doneCh <- true
+					returnCh <- nil
 					return
 				} else if apierrors.IsTooManyRequests(err) {
 					fmt.Fprintf(o.ErrOut, "error when evicting pod %q (will retry after 5s): %v\n", pod.Name, err)
 					time.Sleep(5 * time.Second)
 				} else {
-					errCh <- fmt.Errorf("error when evicting pod %q: %v", pod.Name, err)
+					returnCh <- fmt.Errorf("error when evicting pod %q: %v", pod.Name, err)
 					return
 				}
 			}
 			podArray := []corev1.Pod{pod}
 			_, err = o.waitForDelete(podArray, 1*time.Second, time.Duration(math.MaxInt64), true, getPodFn)
 			if err == nil {
-				doneCh <- true
+				returnCh <- nil
 			} else {
-				errCh <- fmt.Errorf("error when waiting for pod %q terminating: %v", pod.Name, err)
+				returnCh <- fmt.Errorf("error when waiting for pod %q terminating: %v", pod.Name, err)
 			}
-		}(pod, doneCh, errCh)
+		}(pod, returnCh)
 	}
 
 	doneCount := 0
+	var errors []error
+
 	// 0 timeout means infinite, we use MaxInt64 to represent it.
 	var globalTimeout time.Duration
 	if o.Timeout == 0 {
@@ -612,19 +614,19 @@ func (o *DrainOptions) evictPods(pods []corev1.Pod, policyGroupVersion string, g
 		globalTimeout = o.Timeout
 	}
 	globalTimeoutCh := time.After(globalTimeout)
-	for {
+	numPods := len(pods)
+	for doneCount < numPods {
 		select {
-		case err := <-errCh:
-			return err
-		case <-doneCh:
+		case err := <-returnCh:
 			doneCount++
-			if doneCount == len(pods) {
-				return nil
+			if err != nil {
+				errors = append(errors, err)
 			}
 		case <-globalTimeoutCh:
 			return fmt.Errorf("Drain did not complete within %v", globalTimeout)
 		}
 	}
+	return utilerrors.NewAggregate(errors)
 }
 
 func (o *DrainOptions) deletePods(pods []corev1.Pod, getPodFn func(namespace, name string) (*corev1.Pod, error)) error {