Use Connection() from util package

Gopkg.toml

@@ -44,7 +44,7 @@
 
 [[constraint]]
   name = "github.com/kubernetes-csi/csi-lib-utils"
-  version = "0.1.0"
+  version = "0.3.0"
 
 [prune]
    non-go = true

cmd/csi-attacher/main.go

@@ -50,7 +50,7 @@ const (
 var (
 	kubeconfig        = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
 	resync            = flag.Duration("resync", 10*time.Minute, "Resync interval of the controller.")
-	connectionTimeout = flag.Duration("connection-timeout", 1*time.Minute, "Timeout for waiting for CSI driver socket.")
+	connectionTimeout = flag.Duration("connection-timeout", 0, "This option is deprecated.")
 	csiAddress        = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
 	dummy             = flag.Bool("dummy", false, "Run in dummy mode, i.e. not connecting to CSI driver and marking everything as attached. Expected CSI driver name is \"csi/dummy\".")
 	showVersion       = flag.Bool("version", false, "Show version.")
@@ -76,6 +76,10 @@ func main() {
 	}
 	klog.Infof("Version: %s", version)
 
+	if *connectionTimeout != 0 {
+		klog.Warningf("Warning: option -connection-timeout is deprecated and has no effect")
+	}
+
 	// Create the client config. Use kubeconfig if given, otherwise assume in-cluster.
 	config, err := buildConfig(*kubeconfig)
 	if err != nil {
@@ -106,18 +110,12 @@ func main() {
 		attacher = dummyAttacherName
 	} else {
 		// Connect to CSI.
-		csiConn, err := connection.New(*csiAddress, *connectionTimeout)
+		csiConn, err := connection.New(*csiAddress)
 		if err != nil {
 			klog.Error(err.Error())
 			os.Exit(1)
 		}
 
-		// Check it's ready
-		if err = waitForDriverReady(csiConn, *connectionTimeout); err != nil {
-			klog.Error(err.Error())
-			os.Exit(1)
-		}
-
 		// Find driver name.
 		ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 		defer cancel()

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"
 )
@@ -76,8 +73,8 @@ var (
 )
 
 // New provides a new CSI connection object.
-func New(address string, timeout time.Duration) (CSIConnection, error) {
-	conn, err := connect(address, timeout)
+func New(address string) (CSIConnection, error) {
+	conn, err := connection.Connect(address, connection.OnConnectionLoss(onReconnect))
 	if err != nil {
 		return nil, err
 	}
@@ -86,36 +83,9 @@ func New(address string, timeout time.Duration) (CSIConnection, error) {
 	}, 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 onReconnect() bool {
+	klog.Fatalf("Lost connection to CSI driver, exiting")
+	return false
 }
 
 func (c *csiConnection) GetDriverName(ctx context.Context) (string, error) {