Fix CSINodeInfo startup

To speed up unit tests and add more observability when things go wrong.

Author: jsafrane

pkg/volume/csi/csi_plugin.go

@@ -17,15 +17,14 @@ limitations under the License.
 package csi
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
-	"context"
-
 	"k8s.io/klog"
 
 	api "k8s.io/api/core/v1"
@@ -256,7 +255,8 @@ func initializeCSINode(host volume.VolumeHost) error {
 		defer utilruntime.HandleCrash()
 
 		// First wait indefinitely to talk to Kube APIServer
-		err := waitForAPIServerForever(kubeClient)
+		nodeName := host.GetNodeName()
+		err := waitForAPIServerForever(kubeClient, nodeName)
 		if err != nil {
 			klog.Fatalf("Failed to initialize CSINode while waiting for API server to report ok: %v", err)
 		}
@@ -921,20 +921,25 @@ func highestSupportedVersion(versions []string) (*utilversion.Version, error) {
 	return highestSupportedVersion, nil
 }
 
-// waitForAPIServerForever waits forever to get the APIServer Version as a proxy
-// for a healthy APIServer.
-func waitForAPIServerForever(client clientset.Interface) error {
+// waitForAPIServerForever waits forever to get a CSINode instance as a proxy
+// for a healthy APIServer
+func waitForAPIServerForever(client clientset.Interface, nodeName types.NodeName) error {
 	var lastErr error
-	err := wait.PollInfinite(time.Second, func() (bool, error) {
-		_, lastErr = client.Discovery().ServerVersion()
-		if lastErr != nil {
-			lastErr = fmt.Errorf("failed to get apiserver version: %v", lastErr)
-			return false, nil
+	err := wait.PollImmediateInfinite(time.Second, func() (bool, error) {
+		// Get a CSINode from API server to make sure 1) kubelet can reach API server
+		// and 2) it has enough permissions. Kubelet may have restricted permissions
+		// when it's bootstrapping TLS.
+		// https://kubernetes.io/docs/reference/command-line-tools-reference/kubelet-tls-bootstrapping/
+		_, lastErr = client.StorageV1().CSINodes().Get(context.TODO(), string(nodeName), meta.GetOptions{})
+		if lastErr == nil || apierrors.IsNotFound(lastErr) {
+			// API server contacted
+			return true, nil
 		}
-
-		return true, nil
+		klog.V(2).Infof("Failed to contact API server when waiting for CSINode publishing: %s", lastErr)
+		return false, nil
 	})
 	if err != nil {
+		// In theory this is unreachable, but just in case:
 		return fmt.Errorf("%v: %v", err, lastErr)
 	}
 

pkg/volume/csi/nodeinfomanager/nodeinfomanager.go

@@ -397,16 +397,16 @@ func (nim *nodeInfoManager) InitializeCSINodeWithAnnotation() error {
 		return goerrors.New("error getting CSI client")
 	}
 
-	var updateErrs []error
+	var lastErr error
 	err := wait.ExponentialBackoff(updateBackoff, func() (bool, error) {
-		if err := nim.tryInitializeCSINodeWithAnnotation(csiKubeClient); err != nil {
-			updateErrs = append(updateErrs, err)
+		if lastErr = nim.tryInitializeCSINodeWithAnnotation(csiKubeClient); lastErr != nil {
+			klog.V(2).Infof("Failed to publish CSINode: %v", lastErr)
 			return false, nil
 		}
 		return true, nil
 	})
 	if err != nil {
-		return fmt.Errorf("error updating CSINode annotation: %v; caused by: %v", err, utilerrors.NewAggregate(updateErrs))
+		return fmt.Errorf("error updating CSINode annotation: %v; caused by: %v", err, lastErr)
 	}
 
 	return nil

pkg/volume/testing/testing.go

@@ -1870,7 +1870,7 @@ func (f *fakeVolumeHost) WaitForCacheSync() error {
 }
 
 func (f *fakeVolumeHost) WaitForKubeletErrNil() error {
-	return wait.PollImmediate(100*time.Millisecond, 10*time.Second, func() (bool, error) {
+	return wait.PollImmediate(10*time.Millisecond, 10*time.Second, func() (bool, error) {
 		f.mux.Lock()
 		defer f.mux.Unlock()
 		return f.kubeletErr == nil, nil