Process annotation marking pod as not safe to evict first

By processing this annotation first, pods that would match also other,
subsequent rules (e.g. pods without a controller), will generate the
`NotSafeToEvictAnnotation` instead of e.g. `NotReplicated`.

This allows marking pods as not safe to evict to change the warning to
one that can be more effectively ignored, e.g. for ephemeral build pods
that are known to not be backed by a controller, and should not be evicted.

cluster-autoscaler/utils/drain/drain.go

@@ -127,6 +127,9 @@ func GetPodsForDeletionOnNodeDrain(
 		}
 
 		if !safeToEvict && !terminal {
+			if hasNotSafeToEvictAnnotation(pod) {
+				return []*apiv1.Pod{}, []*apiv1.Pod{}, &BlockingPod{Pod: pod, Reason: NotSafeToEvictAnnotation}, fmt.Errorf("pod annotated as not safe to evict present: %s", pod.Name)
+			}
 			if !replicated {
 				return []*apiv1.Pod{}, []*apiv1.Pod{}, &BlockingPod{Pod: pod, Reason: NotReplicated}, fmt.Errorf("%s/%s is not replicated", pod.Namespace, pod.Name)
 			}
@@ -142,9 +145,6 @@ func GetPodsForDeletionOnNodeDrain(
 			if HasBlockingLocalStorage(pod) && skipNodesWithLocalStorage {
 				return []*apiv1.Pod{}, []*apiv1.Pod{}, &BlockingPod{Pod: pod, Reason: LocalStorageRequested}, fmt.Errorf("pod with local storage present: %s", pod.Name)
 			}
-			if hasNotSafeToEvictAnnotation(pod) {
-				return []*apiv1.Pod{}, []*apiv1.Pod{}, &BlockingPod{Pod: pod, Reason: NotSafeToEvictAnnotation}, fmt.Errorf("pod annotated as not safe to evict present: %s", pod.Name)
-			}
 		}
 		pods = append(pods, pod)
 	}

cluster-autoscaler/utils/drain/drain_test.go

@@ -528,6 +528,19 @@ func TestDrain(t *testing.T) {
 		},
 	}
 
+	unsafeNakedPod := &apiv1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "bar",
+			Namespace: "default",
+			Annotations: map[string]string{
+				PodSafeToEvictKey: "false",
+			},
+		},
+		Spec: apiv1.PodSpec{
+			NodeName: "node",
+		},
+	}
+
 	kubeSystemSafePod := &apiv1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "bar",
@@ -838,6 +851,15 @@ func TestDrain(t *testing.T) {
 			expectBlockingPod:   &BlockingPod{Pod: unsafeJobPod, Reason: NotSafeToEvictAnnotation},
 			expectDaemonSetPods: []*apiv1.Pod{},
 		},
+		{
+			description:         "naked pod with PodSafeToEvict=false annotation",
+			pods:                []*apiv1.Pod{unsafeNakedPod},
+			pdbs:                []*policyv1.PodDisruptionBudget{},
+			expectFatal:         true,
+			expectPods:          []*apiv1.Pod{},
+			expectBlockingPod:   &BlockingPod{Pod: unsafeNakedPod, Reason: NotSafeToEvictAnnotation},
+			expectDaemonSetPods: []*apiv1.Pod{},
+		},
 		{
 			description:         "empty PDB with RC-managed pod",
 			pods:                []*apiv1.Pod{rcPod},