Remove dependency on apiserver for IPAMD startup (#3243)

* remove apiserver dependency for ipamd startup

* fix format issue in UT

* wait apiserver connectivty for pod annotate feature

* return maxPods value directly when parsing the local file

Author: oliviassss

cmd/aws-k8s-agent/main.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"os"
+	"time"
 
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/ipamd"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/k8sapi"
@@ -24,6 +25,7 @@ import (
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/version"
 	"github.com/aws/amazon-vpc-cni-k8s/utils"
 	metrics "github.com/aws/amazon-vpc-cni-k8s/utils/prometheusmetrics"
+	"k8s.io/client-go/kubernetes"
 )
 
 const (
@@ -36,44 +38,108 @@ const (
 
 	// Environment variable to disable the IPAMD introspection endpoint on 61679
 	envDisableIntrospection = "DISABLE_INTROSPECTION"
+
+	restCfgTimeout = 5 * time.Second
+	pollInterval   = 5 * time.Second
+	pollTimeout    = 30 * time.Second
 )
 
 func main() {
 	os.Exit(_main())
 }
 
+// startBackgroundAPIServerCheck checks API connectivity in the background
+func startBackgroundAPIServerCheck(ipamContext *ipamd.IPAMContext) {
+	go func() {
+		log := logger.Get()
+		log.Info("Starting background API server connectivity check...")
+
+		// Create a new client for API server check
+		restCfg, err := k8sapi.GetRestConfig()
+		if err != nil {
+			log.Errorf("Failed to get REST config for background API check: %v", err)
+			return
+		}
+		restCfg.Timeout = restCfgTimeout
+		clientSet, err := kubernetes.NewForConfig(restCfg)
+		if err != nil {
+			log.Errorf("Failed to create k8s client for background API check: %v", err)
+			return
+		}
+
+		// Keep checking until connection is established
+		for {
+			version, err := clientSet.Discovery().ServerVersion()
+			if err == nil {
+				log.Infof("API server connectivity established in background! Cluster Version is: %s", version.GitVersion)
+
+				// Update IPAM context with new API server connectivity
+				ipamContext.SetAPIServerConnectivity(true)
+
+				// Exit the goroutine after successful connection
+				log.Info("Background API server check completed successfully")
+				return
+			}
+
+			log.Debugf("Still waiting for API server connectivity in background: %v", err)
+			time.Sleep(pollInterval)
+		}
+	}()
+}
+
 func _main() int {
 	// Do not add anything before initializing logger
 	log := logger.Get()
 
 	log.Infof("Starting L-IPAMD %s  ...", version.Version)
 	version.RegisterMetric()
 
+	enabledPodEni := ipamd.EnablePodENI()
+	enabledCustomNetwork := ipamd.UseCustomNetworkCfg()
+	enabledPodAnnotation := ipamd.EnablePodIPAnnotation()
+	withApiServer := false
 	// Check API Server Connectivity
-	if err := k8sapi.CheckAPIServerConnectivity(); err != nil {
-		log.Errorf("Failed to check API server connectivity: %s", err)
-		return 1
+	if enabledPodEni || enabledCustomNetwork || enabledPodAnnotation {
+		log.Info("SGP, custom networking or pod annotation feature is in use, waiting for API server connectivity to start IPAMD")
+		if err := k8sapi.CheckAPIServerConnectivity(); err != nil {
+			log.Errorf("Failed to check API server connectivity: %s", err)
+			return 1
+		} else {
+			log.Info("API server connectivity established.")
+			withApiServer = true
+		}
+	} else {
+		log.Infof("Waiting to connect API server for upto %s...", pollTimeout)
+		// Try a quick check first
+		if err := k8sapi.CheckAPIServerConnectivityWithTimeout(pollInterval, pollTimeout); err != nil {
+			log.Warn("Proceeding without API server connectivity, will run background API server connectivity check")
+			withApiServer = false
+		} else {
+			log.Info("API server connectivity established.")
+			withApiServer = true
+		}
 	}
-
 	// Create Kubernetes client for API server requests
 	k8sClient, err := k8sapi.CreateKubeClient(appName)
 	if err != nil {
 		log.Errorf("Failed to create kube client: %s", err)
-		return 1
 	}
-
 	// Create EventRecorder for use by IPAMD
-	if err := eventrecorder.Init(k8sClient); err != nil {
+	if err := eventrecorder.Init(k8sClient, withApiServer); err != nil {
 		log.Errorf("Failed to create event recorder: %s", err)
-		return 1
+		log.Warn("Skipping event recorder initialization")
 	}
-
-	ipamContext, err := ipamd.New(k8sClient)
+	ipamContext, err := ipamd.New(k8sClient, withApiServer)
 	if err != nil {
 		log.Errorf("Initialization failure: %v", err)
 		return 1
 	}
 
+	// If not connected to API server yet, start background checks
+	if !withApiServer {
+		startBackgroundAPIServerCheck(ipamContext)
+	}
+
 	// Pool manager
 	go ipamContext.StartNodeIPPoolManager()
 

pkg/ipamd/ipamd.go

@@ -24,6 +24,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/aws/amazon-vpc-cni-k8s/pkg/k8sapi"
+
 	"github.com/aws/smithy-go"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -42,7 +44,6 @@ import (
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/awsutils"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/eniconfig"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/ipamd/datastore"
-	"github.com/aws/amazon-vpc-cni-k8s/pkg/k8sapi"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/networkutils"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/utils/cniutils"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/utils/logger"
@@ -185,6 +186,13 @@ const (
 	// envEnableNetworkPolicy is used to enable IPAMD/CNI to send pod create events to network policy agent.
 	envNetworkPolicyMode     = "NETWORK_POLICY_ENFORCING_MODE"
 	defaultNetworkPolicyMode = "standard"
+
+	defaultMaxPodsFromKubelet = 110
+	kubeletConfigPath         = "/host/etc/kubernetes/kubelet/kubelet-config.json"
+	eniMaxPodsFilePath        = "/app/eni-max-pods.txt"
+
+	// Application name for k8s client
+	appName = "aws-node"
 )
 
 var log = logger.Get()
@@ -230,6 +238,11 @@ type IPAMContext struct {
 	enablePodIPAnnotation     bool
 	maxPods                   int // maximum number of pods that can be scheduled on the node
 	networkPolicyMode         string
+	withApiServer             bool
+}
+
+type kubeletConfig struct {
+	MaxPods *int64 `json:"maxPods"`
 }
 
 // setUnmanagedENIs will rebuild the set of ENI IDs for ENIs tagged as "no_manage"
@@ -335,7 +348,7 @@ func (c *IPAMContext) inInsufficientCidrCoolingPeriod() bool {
 
 // New retrieves IP address usage information from Instance MetaData service and Kubelet
 // then initializes IP address pool data store
-func New(k8sClient client.Client) (*IPAMContext, error) {
+func New(k8sClient client.Client, withApiServer bool) (*IPAMContext, error) {
 	prometheusRegister()
 	c := &IPAMContext{}
 	c.k8sClient = k8sClient
@@ -360,9 +373,9 @@ func New(k8sClient client.Client) (*IPAMContext, error) {
 	c.warmIPTarget = getWarmIPTarget()
 	c.minimumIPTarget = getMinimumIPTarget()
 	c.warmPrefixTarget = getWarmPrefixTarget()
-	c.enablePodENI = enablePodENI()
+	c.enablePodENI = EnablePodENI()
 	c.enableManageUntaggedMode = enableManageUntaggedMode()
-	c.enablePodIPAnnotation = enablePodIPAnnotation()
+	c.enablePodIPAnnotation = EnablePodIPAnnotation()
 	c.numNetworkCards = len(c.awsClient.GetNetworkCards())
 
 	c.networkPolicyMode, err = getNetworkPolicyMode()
@@ -385,7 +398,7 @@ func New(k8sClient client.Client) (*IPAMContext, error) {
 	c.myNodeName = os.Getenv(envNodeName)
 	checkpointer := datastore.NewJSONFile(dsBackingStorePath())
 	c.dataStore = datastore.NewDataStore(log, checkpointer, c.enablePrefixDelegation)
-
+	c.withApiServer = withApiServer
 	if err := c.nodeInit(); err != nil {
 		return nil, err
 	}
@@ -523,21 +536,32 @@ func (c *IPAMContext) nodeInit() error {
 		}, 30*time.Second)
 	}
 
-	// Make a k8s client request for the current node so that max pods can be derived
-	node, err := k8sapi.GetNode(ctx, c.k8sClient)
-	if err != nil {
-		log.Errorf("Failed to get node", err)
-		podENIErrInc("nodeInit")
-		return err
-	}
-
-	maxPods, isInt64 := node.Status.Capacity.Pods().AsInt64()
-	if !isInt64 {
-		log.Errorf("Failed to parse max pods: %s", node.Status.Capacity.Pods().String)
-		podENIErrInc("nodeInit")
-		return errors.New("error while trying to determine max pods")
+	// if apiserver is connected, get the maxPods from node
+	var node corev1.Node
+	if c.withApiServer {
+		node, err := k8sapi.GetNode(ctx, c.k8sClient)
+		if err != nil {
+			log.Errorf("Failed to get node, %s", err)
+			podENIErrInc("nodeInit")
+			return err
+		} else {
+			maxPods, isInt64 := node.Status.Capacity.Pods().AsInt64()
+			if !isInt64 {
+				log.Errorf("Failed to parse max pods: %s", node.Status.Capacity.Pods().String)
+				podENIErrInc("nodeInit")
+				return errors.New("error while trying to determine max pods")
+			}
+			c.maxPods = int(maxPods)
+		}
+	} else {
+		maxPods, err := c.getMaxPodsFromFile()
+		if err != nil {
+			log.Warnf("Using default maxPods as %d because reading from file failed: %v", defaultMaxPodsFromKubelet, err)
+			c.maxPods = defaultMaxPodsFromKubelet
+		} else {
+			c.maxPods = int(maxPods)
+		}
 	}
-	c.maxPods = int(maxPods)
 
 	if c.useCustomNetworking {
 		// When custom networking is enabled and a valid ENIConfig is found, IPAMD patches the CNINode
@@ -1691,6 +1715,57 @@ func (c *IPAMContext) warmIPTargetsDefined() bool {
 	return c.warmIPTarget != noWarmIPTarget || c.minimumIPTarget != noMinimumIPTarget
 }
 
+// max pods from instance type mapping file
+type instanceTypeMaxPodsMapping map[string]int64
+
+// getMaxPodsFromFile reads the max pods value from the eni-max-pods.txt file
+// based on the instance type
+func (c *IPAMContext) getMaxPodsFromFile() (int64, error) {
+	instanceType := c.awsClient.GetInstanceType()
+	if instanceType == "" {
+		return 0, fmt.Errorf("failed to get instance type")
+	}
+
+	data, err := os.ReadFile(eniMaxPodsFilePath)
+	if err != nil {
+		return 0, fmt.Errorf("failed to read ENI max pods file: %w", err)
+	}
+
+	maxPods, err := parseMaxPodsForInstanceFromFile(string(data), instanceType)
+	return maxPods, err
+}
+
+// parseMaxPodsFile parses the eni-max-pods.txt file content and returns a mapping
+// of instance type to max pods
+func parseMaxPodsForInstanceFromFile(content string, instanceType string) (int64, error) {
+	lines := strings.Split(content, "\n")
+
+	for _, line := range lines {
+		// Skip comments and empty lines
+		line = strings.TrimSpace(line)
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+
+		// Split the line into instance type and max pods
+		parts := strings.Fields(line)
+		if len(parts) != 2 {
+			continue
+		}
+
+		currInstanceType := parts[0]
+		if currInstanceType == instanceType {
+			maxPods, err := strconv.ParseInt(parts[1], 10, 64)
+			if err != nil {
+				return 0, fmt.Errorf("failed to parse maxPods for instance type %q: %v", instanceType, err)
+			}
+			return maxPods, nil
+		}
+	}
+
+	return 0, fmt.Errorf("instance type %q not found in ENI max pods file", instanceType)
+}
+
 // UseCustomNetworkCfg returns whether Pods needs to use pod specific configuration or not.
 func UseCustomNetworkCfg() bool {
 	return parseBoolEnvVar(envCustomNetworkCfg, false)
@@ -1766,7 +1841,7 @@ func disableLeakedENICleanup() bool {
 	return isIPv6Enabled() || disableENIProvisioning() || utils.GetBoolAsStringEnvVar(envDisableLeakedENICleanup, false)
 }
 
-func enablePodENI() bool {
+func EnablePodENI() bool {
 	return utils.GetBoolAsStringEnvVar(envEnablePodENI, false)
 }
 
@@ -1796,7 +1871,7 @@ func enableManageUntaggedMode() bool {
 	return utils.GetBoolAsStringEnvVar(envManageUntaggedENI, true)
 }
 
-func enablePodIPAnnotation() bool {
+func EnablePodIPAnnotation() bool {
 	return utils.GetBoolAsStringEnvVar(envAnnotatePodIP, false)
 }
 
@@ -2355,3 +2430,49 @@ func (c *IPAMContext) AddFeatureToCNINode(ctx context.Context, featureName rcv1a
 	newCNINode.Spec.Features = append(newCNINode.Spec.Features, newFeature)
 	return c.k8sClient.Patch(ctx, newCNINode, client.MergeFromWithOptions(cniNode, client.MergeFromWithOptimisticLock{}))
 }
+
+// SetAPIServerConnectivity updates the API server connectivity status and reconfigures
+// components that depend on API server access
+func (c *IPAMContext) SetAPIServerConnectivity(connected bool) {
+	if c.withApiServer == connected {
+		// Status didn't change
+		return
+	}
+
+	log.Infof("Updating API server connectivity status from %v to %v", c.withApiServer, connected)
+	c.withApiServer = connected
+
+	if connected {
+		// API server is now available - update maxPods from node object
+		// First, try to recreate the client with caching enabled
+		newClient, err := k8sapi.CreateKubeClient(appName)
+		if err != nil {
+			log.Errorf("Failed to recreate k8s client with cache when API server became available: %v", err)
+		} else {
+			log.Info("Successfully recreated k8s client with cache after API server became available")
+			c.k8sClient = newClient
+		}
+
+		// Now get the node to update maxPods
+		ctx := context.TODO()
+		node, err := k8sapi.GetNode(ctx, c.k8sClient)
+		if err != nil {
+			log.Errorf("Failed to get node after API server connection established: %s", err)
+		} else {
+			maxPods, isInt64 := node.Status.Capacity.Pods().AsInt64()
+			if !isInt64 {
+				log.Errorf("Failed to parse max pods from node: %s", node.Status.Capacity.Pods().String())
+			} else {
+				// Update maxPods with the value from the node
+				oldMaxPods := c.maxPods
+				c.maxPods = int(maxPods)
+				log.Infof("Updated maxPods from %d to %d based on node capacity", oldMaxPods, c.maxPods)
+			}
+		}
+	} else {
+		// API server connection was lost
+		// No action needed here as we already have maxPods from file or kubelet
+		// and we want to keep working with that value
+		log.Info("API server connection lost, continuing with current maxPods value")
+	}
+}