use CSI connection lib

cmd/csi-cluster-driver-registrar/main.go

@@ -53,7 +53,7 @@ var (
 			"- csi.storage.k8s.io/pod.namespace: pod.Namespace\n"+
 			"- csi.storage.k8s.io/pod.uid: string(pod.UID)",
 	)
-	connectionTimeout = flag.Duration("connection-timeout", 1*time.Minute, "Timeout for waiting for CSI driver socket.")
+	connectionTimeout = flag.Duration("connection-timeout", 0, "The --connection-timeout flag is deprecated")
 	csiAddress        = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
 	showVersion       = flag.Bool("version", false, "Show version.")
 	version           = "unknown"
@@ -72,9 +72,13 @@ func main() {
 	}
 	klog.Infof("Version: %s", version)
 
+	if *connectionTimeout != 0 {
+		klog.Warning("--connection-timeout is deprecated and will have no effect")
+	}
+
 	// Connect to CSI.
 	klog.V(1).Infof("Attempting to open a gRPC connection with: %q", *csiAddress)
-	csiConn, err := connection.NewConnection(*csiAddress, *connectionTimeout)
+	csiConn, err := connection.NewConnection(*csiAddress)
 	if err != nil {
 		klog.Error(err.Error())
 		os.Exit(1)

pkg/connection/connection.go

@@ -19,15 +19,12 @@ package connection
 import (
 	"context"
 	"fmt"
-	"net"
-	"strings"
-	"time"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-csi/csi-lib-utils/connection"
 	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/status"
 	"k8s.io/klog"
 )
@@ -59,8 +56,8 @@ var (
 )
 
 func NewConnection(
-	address string, timeout time.Duration) (CSIConnection, error) {
-	conn, err := connect(address, timeout)
+	address string) (CSIConnection, error) {
+	conn, err := connection.Connect(address)
 	if err != nil {
 		return nil, err
 	}
@@ -69,38 +66,6 @@ func NewConnection(
 	}, nil
 }
 
-func connect(address string, timeout time.Duration) (*grpc.ClientConn, error) {
-	klog.V(2).Infof("Connecting to %s", address)
-	dialOptions := []grpc.DialOption{
-		grpc.WithInsecure(),
-		grpc.WithBackoffMaxDelay(time.Second),
-		grpc.WithUnaryInterceptor(logGRPC),
-	}
-	if strings.HasPrefix(address, "/") {
-		dialOptions = append(dialOptions, grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
-			return net.DialTimeout("unix", addr, timeout)
-		}))
-	}
-	conn, err := grpc.Dial(address, dialOptions...)
-
-	if err != nil {
-		return nil, err
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-	for {
-		if !conn.WaitForStateChange(ctx, conn.GetState()) {
-			klog.V(4).Infof("Connection timed out")
-			return conn, nil // return nil, subsequent GetPluginInfo will show the real connection error
-		}
-		if conn.GetState() == connectivity.Ready {
-			klog.V(3).Infof("Connected")
-			return conn, nil
-		}
-		klog.V(4).Infof("Still trying, connection is %s", conn.GetState())
-	}
-}
-
 func (c *csiConnection) GetDriverName(ctx context.Context) (string, error) {
 	client := csi.NewIdentityClient(c.conn)
 

pkg/connection/connection_test.go

@@ -48,7 +48,7 @@ func createMockServer(t *testing.T) (
 
 	// Create a client connection to it
 	addr := drv.Address()
-	csiConn, err := NewConnection(addr, 10)
+	csiConn, err := NewConnection(addr)
 	if err != nil {
 		return nil, nil, nil, nil, nil, nil, err
 	}