Merge pull request #93 from Jefftree/grpc-close-conn

Close UDS socket connection on completion in GRPC mode

Author: k8s-ci-robot

cmd/client/main.go

@@ -264,7 +264,7 @@ func (c *Client) getUDSDialer(o *GrpcProxyClientOptions) (func(ctx context.Conte
 			}
 			return c, err
 		})
-		tunnel, err := client.CreateGrpcTunnel(o.proxyUdsName, dialOption, grpc.WithInsecure(), grpc.WithUserAgent(o.userAgent))
+		tunnel, err := client.CreateSingleUseGrpcTunnel(o.proxyUdsName, dialOption, grpc.WithInsecure(), grpc.WithUserAgent(o.userAgent))
 		if err != nil {
 			return nil, fmt.Errorf("failed to create tunnel %s, got %v", o.proxyUdsName, err)
 		}
@@ -334,7 +334,7 @@ func (c *Client) getMTLSDialer(o *GrpcProxyClientOptions) (func(ctx context.Cont
 		transportCreds := credentials.NewTLS(tlsConfig)
 		dialOption := grpc.WithTransportCredentials(transportCreds)
 		serverAddress := fmt.Sprintf("%s:%d", o.proxyHost, o.proxyPort)
-		tunnel, err := client.CreateGrpcTunnel(serverAddress, dialOption)
+		tunnel, err := client.CreateSingleUseGrpcTunnel(serverAddress, dialOption)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create tunnel %s, got %v", serverAddress, err)
 		}

konnectivity-client/pkg/client/client.go

@@ -51,9 +51,11 @@ type grpcTunnel struct {
 	connsLock       sync.RWMutex
 }
 
-// CreateGrpcTunnel creates a Tunnel to dial to a remote server through a
+// CreateSingleUseGrpcTunnel creates a Tunnel to dial to a remote server through a
 // gRPC based proxy service.
-func CreateGrpcTunnel(address string, opts ...grpc.DialOption) (Tunnel, error) {
+// Currently, a single tunnel supports a single connection, and the tunnel is closed when the connection is terminated
+// The Dial() method of the returned tunnel should only be called once
+func CreateSingleUseGrpcTunnel(address string, opts ...grpc.DialOption) (Tunnel, error) {
 	c, err := grpc.Dial(address, opts...)
 	if err != nil {
 		return nil, err
@@ -72,12 +74,14 @@ func CreateGrpcTunnel(address string, opts ...grpc.DialOption) (Tunnel, error) {
 		conns:       make(map[int64]*conn),
 	}
 
-	go tunnel.serve()
+	go tunnel.serve(c)
 
 	return tunnel, nil
 }
 
-func (t *grpcTunnel) serve() {
+func (t *grpcTunnel) serve(c *grpc.ClientConn) {
+	defer c.Close()
+
 	for {
 		pkt, err := t.stream.Recv()
 		if err == io.EOF {
@@ -130,9 +134,9 @@ func (t *grpcTunnel) serve() {
 				t.connsLock.Lock()
 				delete(t.conns, resp.ConnectID)
 				t.connsLock.Unlock()
-			} else {
-				klog.Warningf("connection id %d not recognized", resp.ConnectID)
+				return
 			}
+			klog.Warningf("connection id %d not recognized", resp.ConnectID)
 		}
 	}
 }

tests/concurrent_client_request_test.go

@@ -39,7 +39,7 @@ func (s *simpleServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 
 // TODO: test http-connect as well.
 func getTestClient(front string, t *testing.T) *http.Client {
-	tunnel, err := client.CreateGrpcTunnel(front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}

tests/concurrent_test.go

@@ -33,7 +33,7 @@ func TestProxy_Concurrency(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// run test client
-	tunnel, err := client.CreateGrpcTunnel(proxy.front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(proxy.front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}

tests/ha_proxy_server_test.go

@@ -183,7 +183,7 @@ func TestBasicHAProxyServer_GRPC(t *testing.T) {
 }
 
 func testProxyServer(t *testing.T, front string, target string) {
-	tunnel, err := client.CreateGrpcTunnel(front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}

tests/proxy_test.go

@@ -67,7 +67,7 @@ func TestBasicProxy_GRPC(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// run test client
-	tunnel, err := client.CreateGrpcTunnel(proxy.front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(proxy.front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -111,7 +111,7 @@ func TestProxyHandleDialError_GRPC(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// run test client
-	tunnel, err := client.CreateGrpcTunnel(proxy.front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(proxy.front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -152,7 +152,7 @@ func TestProxy_LargeResponse(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// run test client
-	tunnel, err := client.CreateGrpcTunnel(proxy.front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(proxy.front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}

tests/tcp_server_test.go

@@ -60,7 +60,7 @@ func TestEchoServer(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// run test client
-	tunnel, err := client.CreateGrpcTunnel(proxy.front, grpc.WithInsecure())
+	tunnel, err := client.CreateSingleUseGrpcTunnel(proxy.front, grpc.WithInsecure())
 	if err != nil {
 		t.Fatal(err)
 	}