func connect(
	address string,
	metricsManager metrics.CSIMetricsManager,
	dialOptions []grpc.DialOption, connectOptions []Option) (*grpc.ClientConn, error) {
	var o options
	for _, option := range connectOptions {
		option(&o)
	}
	dialOptions = append(dialOptions,
		grpc.WithInsecure(),                    
		grpc.WithBackoffMaxDelay(time.Second),  
		grpc.WithBlock(),                       
		grpc.WithChainUnaryInterceptor(
			LogGRPC,  
			ExtendedCSIMetricsManager{metricsManager}.RecordMetricsClientInterceptor,  
		),
	)
	unixPrefix := "unix://"
	if strings.HasPrefix(address, "/") {
		address = unixPrefix + address
	}
	if strings.HasPrefix(address, unixPrefix) {
		haveConnected := false
		lostConnection := false
		reconnect := true
		dialOptions = append(dialOptions, grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
			if haveConnected && !lostConnection {
				klog.Errorf("Lost connection to %s.", address)
				if o.reconnect != nil {
					reconnect = o.reconnect()
				}
				lostConnection = true
			}
			if !reconnect {
				return nil, errors.New("connection lost, reconnecting disabled")
			}
			conn, err := net.DialTimeout("unix", address[len(unixPrefix):], timeout)
			if err == nil {
				haveConnected = true
				lostConnection = false
			}
			return conn, err
		}))
	} else if o.reconnect != nil {
		return nil, errors.New("OnConnectionLoss callback only supported for unix:// addresses")
	}
	klog.V(5).Infof("Connecting to %s", address)
	var conn *grpc.ClientConn
	var err error
	ready := make(chan bool)
	go func() {
		conn, err = grpc.Dial(address, dialOptions...)
		close(ready)
	}()
	ticker := time.NewTicker(connectionLoggingInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			klog.Warningf("Still connecting to %s", address)
		case <-ready:
			return conn, err
		}
	}
}
connect is the internal implementation of Connect. It has more options to enable testing.
package connection
import (
	"context"
	"errors"
	"io/ioutil"
	"net"
	"strings"
	"time"
	"github.com/kubernetes-csi/csi-lib-utils/metrics"
	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
	"google.golang.org/grpc"
	"k8s.io/klog/v2"
)
const (
	connectionLoggingInterval = 10 * time.Second
	probeInterval = 1 * time.Second
)
const terminationLogPath = "/dev/termination-log"
func Connect(address string, metricsManager metrics.CSIMetricsManager, options ...Option) (*grpc.ClientConn, error) {
	return connect(address, metricsManager, []grpc.DialOption{}, options)
}
type Option func(o *options)
func OnConnectionLoss(reconnect func() bool) Option {
	return func(o *options) {
		o.reconnect = reconnect
	}
}
func ExitOnConnectionLoss() func() bool {
	return func() bool {
		terminationMsg := "Lost connection to CSI driver, exiting"
		if err := ioutil.WriteFile(terminationLogPath, []byte(terminationMsg), 0644); err != nil {
			klog.Errorf("%s: %s", terminationLogPath, err)
		}
		klog.Exit(terminationMsg)
		return false
	}
}
type options struct {
	reconnect func() bool
}