Allow using kube-apiserver cache for pod list requests (#1018)

* Allow using kube-apiserver cache for pod list requests

* When UseAPIServerCacheToListPods is true, don't call kubectl's RunNodeDrain

Author: hieurender

pkg/config/config.go

@@ -88,6 +88,7 @@ const (
 	uptimeFromFileDefault                   = ""
 	workersConfigKey                        = "WORKERS"
 	workersDefault                          = 10
+	useAPIServerCache                       = "USE_APISERVER_CACHE"
 	// prometheus
 	enablePrometheusDefault   = false
 	enablePrometheusConfigKey = "ENABLE_PROMETHEUS_SERVER"
@@ -161,6 +162,7 @@ type Config struct {
 	UseProviderId                       bool
 	CompleteLifecycleActionDelaySeconds int
 	DeleteSqsMsgIfNodeNotFound          bool
+	UseAPIServerCacheToListPods         bool
 }
 
 // ParseCliArgs parses cli arguments and uses environment variables as fallback values
@@ -223,6 +225,7 @@ func ParseCliArgs() (config Config, err error) {
 	flag.BoolVar(&config.UseProviderId, "use-provider-id", getBoolEnv(useProviderIdConfigKey, useProviderIdDefault), "If true, fetch node name through Kubernetes node spec ProviderID instead of AWS event PrivateDnsHostname.")
 	flag.IntVar(&config.CompleteLifecycleActionDelaySeconds, "complete-lifecycle-action-delay-seconds", getIntEnv(completeLifecycleActionDelaySecondsKey, -1), "Delay completing the Autoscaling lifecycle action after a node has been drained.")
 	flag.BoolVar(&config.DeleteSqsMsgIfNodeNotFound, "delete-sqs-msg-if-node-not-found", getBoolEnv(deleteSqsMsgIfNodeNotFoundKey, false), "If true, delete SQS Messages from the SQS Queue if the targeted node(s) are not found.")
+	flag.BoolVar(&config.UseAPIServerCacheToListPods, "use-apiserver-cache", getBoolEnv(useAPIServerCache, false), "If true, leverage the k8s apiserver's index on pod's spec.nodeName to list pods on a node, instead of doing an etcd quorum read.")
 	flag.Parse()
 
 	if isConfigProvided("pod-termination-grace-period", podTerminationGracePeriodConfigKey) && isConfigProvided("grace-period", gracePeriodConfigKey) {
@@ -324,6 +327,7 @@ func (c Config) PrintJsonConfigArgs() {
 		Bool("check_tag_before_draining", c.CheckTagBeforeDraining).
 		Str("ManagedTag", c.ManagedTag).
 		Bool("use_provider_id", c.UseProviderId).
+		Bool("use_apiserver_cache", c.UseAPIServerCacheToListPods).
 		Msg("aws-node-termination-handler arguments")
 }
 
@@ -374,7 +378,8 @@ func (c Config) PrintHumanConfigArgs() {
 			"\tcheck-tag-before-draining: %t,\n"+
 			"\tmanaged-tag: %s,\n"+
 			"\tuse-provider-id: %t,\n"+
-			"\taws-endpoint: %s,\n",
+			"\taws-endpoint: %s,\n"+
+			"\tuse-apiserver-cache: %t,\n",
 		c.DryRun,
 		c.NodeName,
 		c.PodName,
@@ -414,6 +419,7 @@ func (c Config) PrintHumanConfigArgs() {
 		c.ManagedTag,
 		c.UseProviderId,
 		c.AWSEndpoint,
+		c.UseAPIServerCacheToListPods,
 	)
 }
 

pkg/config/config_test.go

@@ -52,6 +52,7 @@ func TestParseCliArgsEnvSuccess(t *testing.T) {
 	t.Setenv("WEBHOOK_TEMPLATE", "WEBHOOK_TEMPLATE")
 	t.Setenv("METADATA_TRIES", "100")
 	t.Setenv("CORDON_ONLY", "false")
+	t.Setenv("USE_APISERVER_CACHE", "true")
 	nthConfig, err := config.ParseCliArgs()
 	h.Ok(t, err)
 
@@ -76,6 +77,7 @@ func TestParseCliArgsEnvSuccess(t *testing.T) {
 	h.Equals(t, "WEBHOOK_TEMPLATE", nthConfig.WebhookTemplate)
 	h.Equals(t, 100, nthConfig.MetadataTries)
 	h.Equals(t, false, nthConfig.CordonOnly)
+	h.Equals(t, true, nthConfig.UseAPIServerCacheToListPods)
 
 	// Check that env vars were set
 	value, ok := os.LookupEnv("KUBERNETES_SERVICE_HOST")
@@ -111,6 +113,7 @@ func TestParseCliArgsSuccess(t *testing.T) {
 		"--webhook-template=WEBHOOK_TEMPLATE",
 		"--metadata-tries=100",
 		"--cordon-only=false",
+		"--use-apiserver-cache=true",
 	}
 	nthConfig, err := config.ParseCliArgs()
 	h.Ok(t, err)
@@ -137,6 +140,7 @@ func TestParseCliArgsSuccess(t *testing.T) {
 	h.Equals(t, 100, nthConfig.MetadataTries)
 	h.Equals(t, false, nthConfig.CordonOnly)
 	h.Equals(t, false, nthConfig.EnablePrometheus)
+	h.Equals(t, true, nthConfig.UseAPIServerCacheToListPods)
 
 	// Check that env vars were set
 	value, ok := os.LookupEnv("KUBERNETES_SERVICE_HOST")

pkg/node/node.go

@@ -114,15 +114,18 @@ func (n Node) CordonAndDrain(nodeName string, reason string, recorder recorderIn
 	if err != nil {
 		return err
 	}
-	// Delete all pods on the node
-	log.Info().Msg("Draining the node")
+	// Be very careful here: in tests, nodeName and node.Name can be different, as
+	// fetchKubernetesNode does some translation using the kubernetes.io/hostname label
 	node, err := n.fetchKubernetesNode(nodeName)
 	if err != nil {
 		return err
 	}
+	var pods *corev1.PodList
+	// Delete all pods on the node
+	log.Info().Msg("Draining the node")
 	// Emit events for all pods that will be evicted
 	if recorder != nil {
-		pods, err := n.fetchAllPods(nodeName)
+		pods, err = n.fetchAllPods(node.Name)
 		if err == nil {
 			for _, pod := range pods.Items {
 				podRef := &corev1.ObjectReference{
@@ -139,7 +142,14 @@ func (n Node) CordonAndDrain(nodeName string, reason string, recorder recorderIn
 			}
 		}
 	}
-	err = drain.RunNodeDrain(n.drainHelper, node.Name)
+	if n.nthConfig.UseAPIServerCacheToListPods {
+		if pods != nil {
+			err = n.drainHelper.DeleteOrEvictPods(pods.Items)
+		}
+	} else {
+		// RunNodeDrain does an etcd quorum-read to list all pods on this node
+		err = drain.RunNodeDrain(n.drainHelper, node.Name)
+	}
 	if err != nil {
 		return err
 	}
@@ -628,9 +638,13 @@ func (n Node) fetchAllPods(nodeName string) (*corev1.PodList, error) {
 		log.Info().Msgf("Would have retrieved running pod list on node %s, but dry-run flag was set", nodeName)
 		return &corev1.PodList{}, nil
 	}
-	return n.drainHelper.Client.CoreV1().Pods("").List(context.TODO(), metav1.ListOptions{
+	listOptions := metav1.ListOptions{
 		FieldSelector: "spec.nodeName=" + nodeName,
-	})
+	}
+	if n.nthConfig.UseAPIServerCacheToListPods {
+		listOptions.ResourceVersion = "0"
+	}
+	return n.drainHelper.Client.CoreV1().Pods("").List(context.TODO(), listOptions)
 }
 
 func getDrainHelper(nthConfig config.Config, clientset *kubernetes.Clientset) (*drain.Helper, error) {