Fix konnectivity issue

Fix that context was cancelled before the proxy connection setup
completely which cause by GH-229

pkg/csi/client/grpc.go

@@ -87,10 +87,14 @@ func (c *workerConnection) Close() error {
 	return c.conn.Close()
 }
 
-func connect(address string, timeout time.Duration, proxyOpts GrpcProxyClientOptions) (*grpc.ClientConn, error) {
-	log.V(6).Infof("New Connecting to %s", address)
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+// connect create new connection to the `address`
+// if proxyOpts.ProxyUDSName or proxyOpts.ProxyHost is not empty, a proxied connection returned
+// currently, the proxy server can only be konnnectivity
+func connect(address string, timeout time.Duration, proxyOpts GrpcProxyClientOptions) (conn *grpc.ClientConn, err error) {
+	log.V(6).Infof("New Connecting to remote address %s ...", address)
+	connectCtx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
+
 	// only for unit test
 	var bufDialerFunc func(context.Context, string) (net.Conn, error)
 	dialOptions := []grpc.DialOption{
@@ -103,18 +107,22 @@ func connect(address string, timeout time.Duration, proxyOpts GrpcProxyClientOpt
 		}
 		dialOptions = append(dialOptions, grpc.WithContextDialer(bufDialerFunc))
 	}
+	// end for unit test
 	// setup konnectivity proxy connection
+	var proxyAddress string
 	if proxyOpts.ProxyHost != "" || proxyOpts.ProxyUDSName != "" {
-		var err error
 		var proxyDialer proxyFunc
 		if proxyOpts.ProxyUDSName == "" {
-			proxyDialer, err = getKonnectivityMTLSDialer(ctx, address, timeout, proxyOpts)
+			proxyAddress = proxyOpts.ProxyHost
+			proxyDialer, err = getKonnectivityMTLSDialer(connectCtx, address, proxyOpts)
 		} else {
-			proxyDialer, err = getKonnectivityUDSDialer(ctx, address, timeout, proxyOpts)
+			proxyAddress = proxyOpts.ProxyUDSName
+			proxyDialer, err = getKonnectivityUDSDialer(connectCtx, address, proxyOpts)
 		}
 		if err != nil {
-			return nil, fmt.Errorf("failed to setup konnectivity dialer: %w", err)
+			return nil, fmt.Errorf("failed to setup konnectivity dialer（%q）: %w", address, err)
 		}
+		log.Infof("connected to proxy server %q.", proxyAddress)
 		dialOptions = append(dialOptions, grpc.WithContextDialer(proxyDialer))
 	}
 	// if strings.HasPrefix(address, "/") {
@@ -129,14 +137,14 @@ func connect(address string, timeout time.Duration, proxyOpts GrpcProxyClientOpt
 				return net.DialTimeout("unix", addr, timeout)
 			}))
 	}
-	conn, err := grpc.Dial(address, dialOptions...)
+	conn, err = grpc.Dial(address, dialOptions...)
 
 	if err != nil {
 		return nil, err
 	}
 	for {
-		if !conn.WaitForStateChange(ctx, conn.GetState()) {
-			log.Warningf("Connection to %s timed out", address)
+		if !conn.WaitForStateChange(connectCtx, conn.GetState()) {
+			log.Warningf("Connection to %s timed out, subsequent calls might fail due to this.", address)
 			return conn, nil // return nil, subsequent GetPluginInfo will show the real connection error
 		}
 		if conn.GetState() == connectivity.Ready {

pkg/csi/client/konnectivity.go

@@ -26,7 +26,6 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
-	"time"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -47,7 +46,7 @@ type GrpcProxyClientOptions struct {
 
 type proxyFunc func(ctx context.Context, addr string) (net.Conn, error)
 
-func getKonnectivityUDSDialer(ctx context.Context, address string, timeout time.Duration, o GrpcProxyClientOptions) (func(ctx context.Context, addr string) (net.Conn, error), error) {
+func getKonnectivityUDSDialer(ctx context.Context, address string,o GrpcProxyClientOptions) (func(ctx context.Context, addr string) (net.Conn, error), error) {
 	log.Infof("using konnectivity UDS dialer")
 
 	var proxyConn net.Conn
@@ -57,15 +56,15 @@ func getKonnectivityUDSDialer(ctx context.Context, address string, timeout time.
 	switch o.Mode {
 	case "grpc":
 		dialOption := grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
-			c, err := net.DialTimeout("unix", o.ProxyUDSName, timeout)
+			c, err := net.Dial("unix", o.ProxyUDSName)
 			if err != nil {
 				log.ErrorS(err, "failed to create connection to uds", "name", o.ProxyUDSName)
 			}
 			return c, err
 		})
 		tunnel, err := client.CreateSingleUseGrpcTunnelWithContext(
-			context.TODO(),
-			ctx,
+			ctx,// create context should follow grpc timeout configuration
+			context.TODO(), // tunnel context use context.TODO()
 			o.ProxyUDSName,
 			dialOption,
 			grpc.WithTransportCredentials(insecure.NewCredentials()),
@@ -112,7 +111,7 @@ func getKonnectivityUDSDialer(ctx context.Context, address string, timeout time.
 	}, nil
 }
 
-func getKonnectivityMTLSDialer(ctx context.Context, address string, _ time.Duration, o GrpcProxyClientOptions) (func(ctx context.Context, addr string) (net.Conn, error), error) {
+func getKonnectivityMTLSDialer(ctx context.Context, address string, o GrpcProxyClientOptions) (func(ctx context.Context, addr string) (net.Conn, error), error) {
 	log.Infof("using konnectivity mTLS dialer")
 
 	tlsConfig, err := getClientTLSConfig(o.CACert, o.ClientCert, o.ClientKey, o.ProxyHost, nil)
@@ -126,7 +125,7 @@ func getKonnectivityMTLSDialer(ctx context.Context, address string, _ time.Durat
 		transportCreds := credentials.NewTLS(tlsConfig)
 		dialOption := grpc.WithTransportCredentials(transportCreds)
 		serverAddress := fmt.Sprintf("%s:%d", o.ProxyHost, o.ProxyPort)
-		tunnel, err := client.CreateSingleUseGrpcTunnelWithContext(context.TODO(), ctx, serverAddress, dialOption)
+		tunnel, err := client.CreateSingleUseGrpcTunnelWithContext(ctx, context.TODO(), serverAddress, dialOption)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create tunnel %s, got %v", serverAddress, err)
 		}

pkg/csi/controllerserver.go

@@ -210,7 +210,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 				return nil, status.Errorf(codes.Internal, "CreateVolume: fail to get lv %s from node %s: %s", req.Name, nodeName, err.Error())
 			} else {
 				if lvName == "" {
-					log.Info("CreateVolume: volume %s not found, creating volume on node %s", volumeID, nodeName)
+					log.Infof("CreateVolume: volume %s is not found, creating volume on node %s", volumeID, nodeName)
 					outstr, err := conn.CreateVolume(ctx, options)
 					if err != nil {
 						return nil, status.Errorf(codes.Internal, "CreateVolume: fail to create lv %s(options: %v): %s", utils.GetNameKey(vgName, volumeID), options, err.Error())
@@ -767,6 +767,7 @@ func (cs *controllerServer) newCreateSnapshotResponse(snapshotId, sourceVolumeId
 	}, nil
 }
 
+// getNodeConn creates a new connection to the lvmd for the `nodeSelected` node
 func (cs *controllerServer) getNodeConn(nodeSelected string) (client.Connection, error) {
 	node, err := cs.nodeLister.Get(nodeSelected)
 	if err != nil {

pkg/csi/nodeutils.go

@@ -476,15 +476,15 @@ func (ns *nodeServer) createLvm(vgName, volumeID, lvmType, unit string, pvSize i
 			log.Errorf("createVolume:: lvcreate command %s error: %v", cmd, err)
 			return err
 		}
-		log.Infof("Successful Create Striping LVM volume: %s, with command: %s", volumeID, cmd)
+		log.Infof("Successfully Create Striping LVM volume: %s, with command: %s", volumeID, cmd)
 	} else if lvmType == LinearType {
 		cmd := fmt.Sprintf("%s lvcreate -n %s -L %d%s -Wy -y %s", localtype.NsenterCmd, volumeID, pvSize, unit, vgName)
 		_, err := ns.osTool.RunCommand(cmd)
 		if err != nil {
 			log.Errorf("createVolume:: lvcreate linear command %s error: %v", cmd, err)
 			return err
 		}
-		log.Infof("Successful Create Linear LVM volume: %s, with command: %s", volumeID, cmd)
+		log.Infof("Successfully Create Linear LVM volume: %s, with command: %s", volumeID, cmd)
 	}
 	return nil
 }