Adding retry with exponential backoff to fetch NodeMetadata during node driver startup

Author: dhananjay-ng

pkg/csi-util/utils.go

@@ -27,6 +27,8 @@ import (
 	"sync"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/wait"
+
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/oracle/oci-go-sdk/v65/core"
 	"go.uber.org/zap"
@@ -104,15 +106,19 @@ type FSSVolumeHandler struct {
 	FsExportPath         string
 }
 
-type NodeIpFamily struct {
-	PreferredNodeIpFamily string
-	Ipv4Enabled           bool
-	Ipv6Enabled           bool
+type NodeMetadata struct {
+	PreferredNodeIpFamily  string
+	Ipv4Enabled            bool
+	Ipv6Enabled            bool
+	AvailabilityDomain     string
+	FullAvailabilityDomain string
+	IsNodeMetadataLoaded   bool
 }
 
 // CSIConfig represents the structure of the ConfigMap data.
 type CSIConfig struct {
 	Lustre *DriverConfig `yaml:"lustre"`
+	IsLoaded bool
 }
 
 // DriverConfig represents driver-specific configurations.
@@ -136,25 +142,75 @@ func (u *Util) LookupNodeID(k kubernetes.Interface, nodeName string) (string, er
 	return n.Spec.ProviderID, nil
 }
 
-func (u *Util) LookupNodeAvailableDomain(k kubernetes.Interface, nodeID string) (string, string, error) {
-	n, err := k.CoreV1().Nodes().Get(context.Background(), nodeID, metav1.GetOptions{})
+func (u *Util) WaitForKubeApiServerToBeReachableWithContext(ctx context.Context, k kubernetes.Interface, backOffCap time.Duration) {
+
+	waitForKubeApiServerCtx, waitForKubeApiServerCtxCancel := context.WithTimeout(ctx, time.Second * 45)
+	defer waitForKubeApiServerCtxCancel()
+
+	backoff := wait.Backoff{
+		Duration: 1 * time.Second,
+		Factor:   2.0,
+		Steps:    5,
+		Cap: backOffCap,
+	}
+
+	wait.ExponentialBackoffWithContext(
+		waitForKubeApiServerCtx,
+		backoff,
+		func(waitForKubeApiServerCtx context.Context) (bool, error) {
+			attemptCtx, attemptCancel := context.WithTimeout(waitForKubeApiServerCtx, backoff.Step())
+			defer attemptCancel()
+			_, err := k.CoreV1().RESTClient().Get().AbsPath("/version").Do(attemptCtx).Raw()
+			if err != nil {
+				u.Logger.With(zap.Error(err)).Errorf("Waiting for kube api server to be reachable, Retrying..")
+				return false, nil
+			}
+			u.Logger.Infof("Kube Api Server is Reachable")
+			return true, nil
+		},
+	)
+}
+
+func (u *Util) LoadNodeMetadataFromApiServer(ctx context.Context, k kubernetes.Interface, nodeID string, nodeMetadata *NodeMetadata) (error) {
+
+	u.WaitForKubeApiServerToBeReachableWithContext(ctx, k, time.Second * 30)
+
+	node, err := k.CoreV1().Nodes().Get(ctx, nodeID, metav1.GetOptions{})
+
 	if err != nil {
-		u.Logger.With(zap.Error(err)).With("nodeId", nodeID).Error("Failed to get Node by name.")
-		return "", "",fmt.Errorf("failed to get node %s", nodeID)
+		u.Logger.With(zap.Error(err)).With("nodeId", nodeID).Error("Failed to get Node information from kube api server, Please check if kube api server is accessible.")
+		return fmt.Errorf("Failed to get node information from kube api server, please check if kube api server is accessible.")
 	}
-	if n.Labels != nil {
-		ad, ok := n.Labels[kubeAPI.LabelTopologyZone]
+
+	var ok bool
+	if node.Labels != nil {
+		nodeMetadata.AvailabilityDomain, ok = node.Labels[kubeAPI.LabelTopologyZone]
 		if !ok {
-			ad, ok = n.Labels[kubeAPI.LabelZoneFailureDomain]
+			nodeMetadata.AvailabilityDomain, ok = node.Labels[kubeAPI.LabelZoneFailureDomain]
 		}
 		if ok {
-			fullAdName, _ := n.Labels[AvailabilityDomainLabel]
-			return ad, fullAdName, nil
+			nodeMetadata.FullAvailabilityDomain, _ = node.Labels[AvailabilityDomainLabel]
+		}
+
+		if preferredIpFamily, ok := node.Labels[LabelIpFamilyPreferred]; ok {
+			nodeMetadata.PreferredNodeIpFamily = FormatValidIpStackInK8SConvention(preferredIpFamily)
+		}
+		if ipv4Enabled, ok := node.Labels[LabelIpFamilyIpv4]; ok && strings.EqualFold(ipv4Enabled, "true") {
+			nodeMetadata.Ipv4Enabled = true
+		}
+		if ipv6Enabled, ok := node.Labels[LabelIpFamilyIpv6]; ok && strings.EqualFold(ipv6Enabled, "true") {
+			nodeMetadata.Ipv6Enabled = true
 		}
 	}
-	errMsg := fmt.Sprintf("Did not find the label for the fault domain. Checked Topology Labels: %s, %s", kubeAPI.LabelTopologyZone, kubeAPI.LabelZoneFailureDomain)
-	u.Logger.With("nodeId", nodeID).Error(errMsg)
-	return "","", fmt.Errorf(errMsg)
+	if !nodeMetadata.Ipv4Enabled && !nodeMetadata.Ipv6Enabled {
+		nodeMetadata.PreferredNodeIpFamily = Ipv4Stack
+		nodeMetadata.Ipv4Enabled = true
+		u.Logger.With("nodeId", nodeID, "nodeMetadata", nodeMetadata).Info("No IP family labels identified on node, defaulting to ipv4.")
+	} else {
+		u.Logger.With("nodeId", nodeID, "nodeMetadata", nodeMetadata).Info("Node IP family identified.")
+	}
+	nodeMetadata.IsNodeMetadataLoaded = true
+	return  nil
 }
 
 // waitForPathToExist waits for for a given filesystem path to exist.
@@ -524,37 +580,6 @@ func GetIsFeatureEnabledFromEnv(logger *zap.SugaredLogger, featureName string, d
 	return enableFeature
 }
 
-func GetNodeIpFamily(k kubernetes.Interface, nodeID string, logger *zap.SugaredLogger) (*NodeIpFamily, error) {
-	n, err := k.CoreV1().Nodes().Get(context.Background(), nodeID, metav1.GetOptions{})
-
-	if err != nil {
-		logger.With(zap.Error(err)).With("nodeId", nodeID).Error("Failed to get Node information from kube api server, Please check if kube api server is accessible.")
-		return nil, fmt.Errorf("Failed to get node information from kube api server, please check if kube api server is accessible.")
-	}
-
-	nodeIpFamily := &NodeIpFamily{}
-
-	if n.Labels != nil {
-		if preferredIpFamily, ok := n.Labels[LabelIpFamilyPreferred]; ok {
-			nodeIpFamily.PreferredNodeIpFamily = FormatValidIpStackInK8SConvention(preferredIpFamily)
-		}
-		if ipv4Enabled, ok := n.Labels[LabelIpFamilyIpv4]; ok && strings.EqualFold(ipv4Enabled, "true") {
-			nodeIpFamily.Ipv4Enabled = true
-		}
-		if ipv6Enabled, ok := n.Labels[LabelIpFamilyIpv6]; ok && strings.EqualFold(ipv6Enabled, "true") {
-			nodeIpFamily.Ipv6Enabled = true
-		}
-	}
-	if !nodeIpFamily.Ipv4Enabled && !nodeIpFamily.Ipv6Enabled {
-		nodeIpFamily.PreferredNodeIpFamily = Ipv4Stack
-		nodeIpFamily.Ipv4Enabled = true
-		logger.With("nodeId", nodeID, "nodeIpFamily", *nodeIpFamily).Info("No IP family labels identified on node, defaulting to ipv4.")
-	} else {
-		logger.With("nodeId", nodeID, "nodeIpFamily", *nodeIpFamily).Info("Node IP family identified.")
-	}
-
-	return nodeIpFamily, nil
-}
 func ConvertIscsiIpFromIpv4ToIpv6(ipv4IscsiIp string) (string, error) {
 	ipv4IscsiIP := net.ParseIP(ipv4IscsiIp).To4()
 	if ipv4IscsiIP == nil {
@@ -609,30 +634,27 @@ func IsValidIpFamilyPresentInClusterIpFamily(clusterIpFamily string) bool {
 	return len(clusterIpFamily) > 0 && (strings.Contains(clusterIpFamily, Ipv4Stack) || strings.Contains(clusterIpFamily, Ipv6Stack))
 }
 
-func IsIpv6SingleStackNode(nodeIpFamily *NodeIpFamily) bool {
-	if nodeIpFamily == nil {
+func IsIpv6SingleStackNode(nodeMetadata *NodeMetadata) bool {
+	if nodeMetadata == nil {
 		return false
 	}
-	return nodeIpFamily.Ipv6Enabled == true && nodeIpFamily.Ipv4Enabled == false
+	return nodeMetadata.Ipv6Enabled == true && nodeMetadata.Ipv4Enabled == false
 }
 
-func LoadCSIConfigFromConfigMap(k kubernetes.Interface, configMapName string, logger *zap.SugaredLogger) (*CSIConfig) {
+func LoadCSIConfigFromConfigMap(csiConfig *CSIConfig, k kubernetes.Interface, configMapName string, logger *zap.SugaredLogger) {
 	// Get the ConfigMap
 	// Parse the configuration for each driver
-	config := &CSIConfig{}
 	cm, err := k.CoreV1().ConfigMaps("kube-system").Get(context.Background(), configMapName, metav1.GetOptions{})
 	if err != nil {
 		logger.Debugf("Failed to load ConfigMap %v due to error %v. Using default configuration.", configMapName, err)
-		return config
+		return
 	}
 
 	if lustreConfig, exists := cm.Data["lustre"]; exists {
-		if err := yaml.Unmarshal([]byte(lustreConfig), &config.Lustre); err != nil {
+		if err := yaml.Unmarshal([]byte(lustreConfig), &csiConfig.Lustre); err != nil {
 			logger.Debugf("Failed to parse lustre key in config map %v. Error: %v",configMapName,  err)
-			return config
+			return
 		}
 		logger.Infof("Successfully loaded ConfigMap %v. Using customized configuration for csi driver.", configMapName)
 	}
-
-	return config
 }

pkg/csi-util/utils_test.go

@@ -15,15 +15,20 @@
 package csi_util
 
 import (
+	"context"
 	"fmt"
+	"log"
+	"os"
 	"reflect"
 	"regexp"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/oracle/oci-cloud-controller-manager/pkg/util"
 	"github.com/oracle/oci-go-sdk/v65/core"
 	"go.uber.org/zap"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/pointer"
 )
 
@@ -344,18 +349,21 @@ func Test_DiskByPathPatternForPV(t *testing.T) {
 	}
 }
 
-func Test_GetNodeIpFamily(t *testing.T) {
+func Test_LoadNodeMetadataFromApiServer(t *testing.T) {
 
 	tests := []struct {
 		name     string
 		nodeName string
-		want     *NodeIpFamily
+		want     *NodeMetadata
+		kubeclient 		 kubernetes.Interface
 		err      error
 	}{
 		{
 			name:     "should return ipv6 for ipv6 preferred node",
 			nodeName: "ipv6Preferred",
-			want: &NodeIpFamily{
+			want: &NodeMetadata{
+				FullAvailabilityDomain: "xyz:PHX-AD-3",
+				AvailabilityDomain: "PHX-AD-3",
 				PreferredNodeIpFamily: Ipv6Stack,
 				Ipv4Enabled:           true,
 				Ipv6Enabled:           true,
@@ -364,16 +372,18 @@ func Test_GetNodeIpFamily(t *testing.T) {
 		{
 			name:     "should return ipv4 for ipv4 preferred node",
 			nodeName: "ipv4Preferred",
-			want: &NodeIpFamily{
+			want: &NodeMetadata{
 				PreferredNodeIpFamily: Ipv4Stack,
+				AvailabilityDomain: "PHX-AD-3",
 				Ipv4Enabled:           true,
 				Ipv6Enabled:           true,
 			},
 		},
 		{
 			name:     "should return default IPv4 family for no ip preference",
 			nodeName: "noIpPreference",
-			want: &NodeIpFamily{
+			want: &NodeMetadata{
+				AvailabilityDomain: "PHX-AD-3",
 				PreferredNodeIpFamily: Ipv4Stack,
 				Ipv4Enabled:           true,
 				Ipv6Enabled:           false,
@@ -382,21 +392,73 @@ func Test_GetNodeIpFamily(t *testing.T) {
 		{
 			name:     "should return error for invalid node",
 			nodeName: "InvalidNode",
-			want:     nil,
+			want:     &NodeMetadata{},
 			err:      fmt.Errorf("Failed to get node information from kube api server, please check if kube api server is accessible."),
 		},
+		{
+			name:     "should return error for node with any ad labels",
+			nodeName: "nodeWithMissingAdLabels",
+			want: &NodeMetadata{
+				PreferredNodeIpFamily: Ipv4Stack,
+				Ipv4Enabled:           true,
+				Ipv6Enabled:           false,
+			},
+			err: fmt.Errorf("Failed to get node information from kube api server, please check if kube api server is accessible."),
+		},
+		{
+			name:     "Call to get node info is done  even if health check fails",
+			nodeName: "ipv4Preferred",
+			want: &NodeMetadata{
+				PreferredNodeIpFamily: Ipv4Stack,
+				AvailabilityDomain: "PHX-AD-3",
+				Ipv4Enabled:           true,
+				Ipv6Enabled:           true,
+			},
+			kubeclient: &util.MockKubeClientWithFailingRestClient{
+				CoreClient: &util.MockCoreClientWithFailingRestClient{},
+			},
+		},
+		{
+			name:     "should return error for invalid node and failing health check",
+			nodeName: "InvalidNode",
+			want:     &NodeMetadata{},
+			err:      fmt.Errorf("Failed to get node information from kube api server, please check if kube api server is accessible."),
+			kubeclient: &util.MockKubeClientWithFailingRestClient{
+			CoreClient: &util.MockCoreClientWithFailingRestClient{},
+		},
+		},
+	}
+
+	logger, _ := zap.NewDevelopment()
+	sugar := logger.Sugar()
+	u := &Util{
+		Logger: sugar,
 	}
+
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := GetNodeIpFamily(&util.MockKubeClient{
-				CoreClient: &util.MockCoreClient{},
-			}, tt.nodeName, zap.S())
-			if (tt.want != got) && (tt.want.PreferredNodeIpFamily != got.PreferredNodeIpFamily ||
-				tt.want.Ipv6Enabled != got.Ipv6Enabled || tt.want.Ipv4Enabled != got.Ipv4Enabled) {
-				t.Errorf("GetNodeIpFamily() = %v, want %v", got, tt.want)
+
+
+			log.SetOutput(os.Stdout)
+			nodeMetadata := &NodeMetadata{}
+			ctx, cancel := context.WithTimeout(context.Background(), 10 * time.Second)
+			defer cancel()
+
+
+			var k kubernetes.Interface
+			if tt.kubeclient != nil {
+				k = tt.kubeclient
+			} else {
+				k = &util.MockKubeClient{CoreClient: &util.MockCoreClient{}}
+			}
+
+			err := u.LoadNodeMetadataFromApiServer(ctx, k, tt.nodeName, nodeMetadata)
+			if (tt.want != nodeMetadata) && (tt.want.PreferredNodeIpFamily != nodeMetadata.PreferredNodeIpFamily ||
+				tt.want.Ipv6Enabled != nodeMetadata.Ipv6Enabled || tt.want.Ipv4Enabled != nodeMetadata.Ipv4Enabled) {
+				t.Errorf("LoadNodeMetadataFromApiServer() = %v, want %v", nodeMetadata, tt.want)
 			}
 			if err != nil && !strings.EqualFold(tt.err.Error(), err.Error()) {
-				t.Errorf("GetNodeIpFamily() = %v, want %v", err, tt.err)
+				t.Errorf("LoadNodeMetadataFromApiServer() = %v, want %v", err, tt.err)
 			}
 
 		})
@@ -764,18 +826,20 @@ func Test_LoadCSIConfigFromConfigMap(t *testing.T) {
 		{
 			name:     "Return default config if config map is not present",
 			configMapName: "invalid",
-			want: &CSIConfig{},
+			want: &CSIConfig{
+			},
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := LoadCSIConfigFromConfigMap(&util.MockKubeClient{
+			csiConfig := &CSIConfig{}
+			LoadCSIConfigFromConfigMap(csiConfig, &util.MockKubeClient{
 				CoreClient: &util.MockCoreClient{},
 			}, tt.configMapName, zap.S())
 
-			if !reflect.DeepEqual(tt.want, got) {
-				t.Errorf("LoadCSIConfigFromConfigMap() => got : %v, want :  %v", got, tt.want)
+			if !reflect.DeepEqual(tt.want, csiConfig) {
+				t.Errorf("LoadCSIConfigFromConfigMap() => got : %v, want :  %v", csiConfig, tt.want)
 			}
 		})
 	}