[fix] : add k8s node cache and prefer providerID for node search (#353)

* add k8s node cache and prefer providerID for node search

* fix failing unittests

* add e2e test for updating instance label and making sure route still exists

* update docs and add comments

* fix capacity of map initialized

* fix linting error

Author: rahulait

cloud/linode/node_controller.go

@@ -24,10 +24,13 @@ import (
 )
 
 const (
-	informerResyncPeriod = 1 * time.Minute
-	defaultMetadataTTL   = 300 * time.Second
+	informerResyncPeriod   = 1 * time.Minute
+	defaultMetadataTTL     = 300 * time.Second
+	defaultK8sNodeCacheTTL = 300 * time.Second
 )
 
+var registeredK8sNodeCache *k8sNodeCache = newK8sNodeCache()
+
 type nodeRequest struct {
 	node      *v1.Node
 	timestamp time.Time
@@ -48,6 +51,104 @@ type nodeController struct {
 	nodeLastAdded map[string]time.Time
 }
 
+// k8sNodeCache stores node related info as registered in k8s
+type k8sNodeCache struct {
+	sync.RWMutex
+	nodes       map[string]*v1.Node
+	providerIDs map[string]string
+	lastUpdate  time.Time
+	ttl         time.Duration
+}
+
+// updateCache updates the k8s node cache with the latest nodes from the k8s API server.
+func (c *k8sNodeCache) updateCache(kubeclient kubernetes.Interface) {
+	c.Lock()
+	defer c.Unlock()
+	if time.Since(c.lastUpdate) < c.ttl {
+		return
+	}
+
+	nodeList, err := kubeclient.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		klog.Errorf("failed to list nodes, cannot create/update k8s node cache: %s", err)
+		return
+	}
+
+	nodes := make(map[string]*v1.Node, len(nodeList.Items))
+	providerIDs := make(map[string]string, len(nodeList.Items))
+	for _, node := range nodeList.Items {
+		if node.Spec.ProviderID == "" {
+			klog.Errorf("Empty providerID [%s] for node %s, skipping it", node.Spec.ProviderID, node.Name)
+			continue
+		}
+		nodes[node.Name] = &node
+		providerIDs[node.Spec.ProviderID] = node.Name
+	}
+
+	c.nodes = nodes
+	c.providerIDs = providerIDs
+	c.lastUpdate = time.Now()
+}
+
+// addNodeToCache stores the specified node in k8s node cache
+func (c *k8sNodeCache) addNodeToCache(node *v1.Node) {
+	c.Lock()
+	defer c.Unlock()
+	if node.Spec.ProviderID == "" {
+		klog.Errorf("Empty providerID [%s] for node %s, skipping it", node.Spec.ProviderID, node.Name)
+		return
+	}
+	c.nodes[node.Name] = node
+	c.providerIDs[node.Spec.ProviderID] = node.Name
+}
+
+// getNodeLabel returns the k8s node label for the given provider ID or instance label.
+// If the provider ID or label is not found in the cache, it returns an empty string and false.
+func (c *k8sNodeCache) getNodeLabel(providerID string, instanceLabel string) (string, bool) {
+	c.RLock()
+	defer c.RUnlock()
+
+	// check if instance label matches with the registered k8s node
+	if _, exists := c.nodes[instanceLabel]; exists {
+		return instanceLabel, true
+	}
+
+	// check if provider id matches with the registered k8s node
+	if label, exists := c.providerIDs[providerID]; exists {
+		return label, true
+	}
+
+	return "", false
+}
+
+// getProviderID returns linode specific providerID for given k8s node name
+func (c *k8sNodeCache) getProviderID(nodeName string) (string, bool) {
+	c.RLock()
+	defer c.RUnlock()
+
+	if node, exists := c.nodes[nodeName]; exists {
+		return node.Spec.ProviderID, true
+	}
+
+	return "", false
+}
+
+// newK8sNodeCache returns new k8s node cache instance
+func newK8sNodeCache() *k8sNodeCache {
+	timeout := defaultK8sNodeCacheTTL
+	if raw, ok := os.LookupEnv("K8S_NODECACHE_TTL"); ok {
+		if t, err := strconv.Atoi(raw); t > 0 && err == nil {
+			timeout = time.Duration(t) * time.Second
+		}
+	}
+
+	return &k8sNodeCache{
+		nodes:       make(map[string]*v1.Node, 0),
+		providerIDs: make(map[string]string, 0),
+		ttl:         timeout,
+	}
+}
+
 func newNodeController(kubeclient kubernetes.Interface, client client.Client, informer v1informers.NodeInformer, instanceCache *instances) *nodeController {
 	timeout := defaultMetadataTTL
 	if raw, ok := os.LookupEnv("LINODE_METADATA_TTL"); ok {
@@ -145,6 +246,7 @@ func (s *nodeController) processNext() bool {
 		klog.Errorf("failed to add metadata for node (%s); will not retry: %s", request.node.Name, err)
 	}
 
+	registeredK8sNodeCache.updateCache(s.kubeclient)
 	return true
 }
 
@@ -187,9 +289,7 @@ func (s *nodeController) handleNode(ctx context.Context, node *v1.Node) error {
 
 	expectedPrivateIP := ""
 	// linode API response for linode will contain only one private ip
-	// if any private ip is configured. If it changes in future or linode
-	// supports other subnets with nodebalancer, this logic needs to be updated.
-	// https://www.linode.com/docs/api/linode-instances/#linode-view
+	// if any private ip is configured.
 	for _, addr := range linode.IPv4 {
 		if isPrivate(addr) {
 			expectedPrivateIP = addr.String()
@@ -202,6 +302,7 @@ func (s *nodeController) handleNode(ctx context.Context, node *v1.Node) error {
 		return nil
 	}
 
+	var updatedNode *v1.Node
 	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
 		// Get a fresh copy of the node so the resource version is up-to-date
 		nodeResult, err := s.kubeclient.CoreV1().Nodes().Get(ctx, node.Name, metav1.GetOptions{})
@@ -223,13 +324,16 @@ func (s *nodeController) handleNode(ctx context.Context, node *v1.Node) error {
 		if nodeResult.Annotations[annotations.AnnLinodeNodePrivateIP] != expectedPrivateIP && expectedPrivateIP != "" {
 			nodeResult.Annotations[annotations.AnnLinodeNodePrivateIP] = expectedPrivateIP
 		}
-		_, err = s.kubeclient.CoreV1().Nodes().Update(ctx, nodeResult, metav1.UpdateOptions{})
+		updatedNode, err = s.kubeclient.CoreV1().Nodes().Update(ctx, nodeResult, metav1.UpdateOptions{})
 		return err
 	}); err != nil {
 		klog.V(1).ErrorS(err, "Node update error")
 		return err
 	}
 
+	if updatedNode != nil {
+		registeredK8sNodeCache.addNodeToCache(updatedNode)
+	}
 	s.SetLastMetadataUpdate(node.Name)
 
 	return nil

cloud/linode/node_controller_test.go

@@ -125,6 +125,34 @@ func TestNodeController_processNext(t *testing.T) {
 		}
 	})
 
+	t.Run("should return no error if node has providerID set", func(t *testing.T) {
+		node2 := &v1.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "test-node2",
+				Labels:      map[string]string{},
+				Annotations: map[string]string{},
+			},
+			Spec: v1.NodeSpec{ProviderID: "linode://112"},
+		}
+		currInstances := controller.instances
+		defer func() {
+			controller.instances = currInstances
+		}()
+		controller.instances = newInstances(client)
+		registeredK8sNodeCache.lastUpdate = time.Now().Add(-15 * time.Minute)
+		controller.addNodeToQueue(node2)
+		publicIP := net.ParseIP("172.234.31.123")
+		privateIP := net.ParseIP("192.168.159.135")
+		client.EXPECT().ListInstances(gomock.Any(), nil).Times(1).Return([]linodego.Instance{
+			{ID: 112, Label: "test-node2", IPv4: []*net.IP{&publicIP, &privateIP}, HostUUID: "112"},
+		}, nil)
+		result := controller.processNext()
+		assert.True(t, result, "processNext should return true")
+		if queue.Len() != 0 {
+			t.Errorf("expected queue to be empty, got %d items", queue.Len())
+		}
+	})
+
 	t.Run("should return no error if node in k8s doesn't exist", func(t *testing.T) {
 		controller.addNodeToQueue(node)
 		controller.kubeclient = fake.NewSimpleClientset()
@@ -192,6 +220,10 @@ func TestNodeController_handleNode(t *testing.T) {
 	nodeCtrl := newNodeController(kubeClient, client, nil, instCache)
 	assert.Equal(t, 30*time.Second, nodeCtrl.ttl, "expected ttl to be 30 seconds")
 
+	t.Setenv("K8S_NODECACHE_TTL", "60")
+	currK8sNodeCache := newK8sNodeCache()
+	assert.Equal(t, 60*time.Second, currK8sNodeCache.ttl, "expected ttl to be 60 seconds")
+
 	// Test: Successful metadata update
 	publicIP := net.ParseIP("172.234.31.123")
 	privateIP := net.ParseIP("192.168.159.135")
@@ -231,3 +263,39 @@ func TestNodeController_handleNode(t *testing.T) {
 	err = nodeCtrl.handleNode(context.TODO(), node)
 	assert.NoError(t, err, "expected no error during handleNode")
 }
+
+func Test_k8sNodeCache_addNodeToCache(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	// Add node with providerID set
+	node := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "test-node",
+			Labels:      map[string]string{},
+			Annotations: map[string]string{},
+		},
+		Spec: v1.NodeSpec{ProviderID: "linode://123"},
+	}
+
+	currK8sNodeCache := newK8sNodeCache()
+	currK8sNodeCache.addNodeToCache(node)
+
+	if _, exists := currK8sNodeCache.nodes[node.Name]; !exists {
+		t.Errorf("expected node to be added to cache")
+	}
+
+	// Add node without providerID set
+	node2 := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "test-node2",
+			Labels:      map[string]string{},
+			Annotations: map[string]string{},
+		},
+	}
+
+	currK8sNodeCache.addNodeToCache(node2)
+	if _, exists := currK8sNodeCache.nodes[node2.Name]; exists {
+		t.Errorf("expected node to not be added to cache")
+	}
+}

cloud/linode/route_controller.go

@@ -112,6 +112,11 @@ func (r *routes) getInstanceFromName(ctx context.Context, name string) (*linodeg
 		},
 	}
 
+	// fetch providerID from k8s node cache if it exists
+	if id, ok := registeredK8sNodeCache.getProviderID(name); ok {
+		node.Spec.ProviderID = id
+	}
+
 	// fetch instance with specified node name
 	instance, err := r.instances.lookupLinode(ctx, node)
 	if err != nil {
@@ -235,6 +240,15 @@ func (r *routes) ListRoutes(ctx context.Context, clusterName string) ([]*cloudpr
 
 	var configuredRoutes []*cloudprovider.Route
 	for _, instance := range instances {
+		providerID := providerIDPrefix + strconv.Itoa(instance.ID)
+		label, found := registeredK8sNodeCache.getNodeLabel(providerID, instance.Label)
+		if !found {
+			klog.V(4).Infof("Node %s not found in k8s node cache, skipping listing its routes", instance.Label)
+			continue
+		}
+		// Update label to match with k8s registered label
+		instance.Label = label
+
 		instanceRoutes, err := r.getInstanceRoutes(ctx, instance.ID)
 		if err != nil {
 			klog.Errorf("Failed finding routes for instance id %d. Error: %v", instance.ID, err)