Merge pull request #61 from yue9944882/bugfix/cluster-proxy-client-port-release

Bugfix: singleton dialer tunnel TIMED_WAIT client port release

Author: Somefive

pkg/apis/cluster/v1alpha1/clustergateway_proxy.go

@@ -224,7 +224,7 @@ func (p *proxyHandler) ServeHTTP(writer http.ResponseWriter, request *http.Reque
 	newReq.URL.RawQuery = request.URL.RawQuery
 	newReq.RequestURI = newReq.URL.RequestURI()
 
-	cfg, err := NewConfigFromCluster(cluster)
+	cfg, err := NewConfigFromCluster(request.Context(), cluster)
 	if err != nil {
 		responsewriters.InternalError(writer, request, errors.Wrapf(err, "failed creating cluster proxy client config %s", cluster.Name))
 		return

pkg/apis/cluster/v1alpha1/transport.go

@@ -5,18 +5,20 @@ import (
 	"net"
 	"net/url"
 	"strconv"
+	"time"
 
 	"github.com/oam-dev/cluster-gateway/pkg/config"
 	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 	grpccredentials "google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
 	k8snet "k8s.io/apimachinery/pkg/util/net"
 	restclient "k8s.io/client-go/rest"
 	konnectivity "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/util"
 )
 
-var DialerGetter = func() (k8snet.DialFunc, error) {
+var DialerGetter = func(ctx context.Context) (k8snet.DialFunc, error) {
 	tlsCfg, err := util.GetClientTLSConfig(
 		config.ClusterProxyCAFile,
 		config.ClusterProxyCertFile,
@@ -26,18 +28,21 @@ var DialerGetter = func() (k8snet.DialFunc, error) {
 	if err != nil {
 		return nil, err
 	}
-	tunnel, err := konnectivity.CreateSingleUseGrpcTunnel(
-		context.TODO(),
+	dialerTunnel, err := konnectivity.CreateSingleUseGrpcTunnel(
+		ctx,
 		net.JoinHostPort(config.ClusterProxyHost, strconv.Itoa(config.ClusterProxyPort)),
 		grpc.WithTransportCredentials(grpccredentials.NewTLS(tlsCfg)),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			Time: time.Second * 5,
+		}),
 	)
 	if err != nil {
 		return nil, err
 	}
-	return tunnel.DialContext, nil
+	return dialerTunnel.DialContext, nil
 }
 
-func NewConfigFromCluster(c *ClusterGateway) (*restclient.Config, error) {
+func NewConfigFromCluster(ctx context.Context, c *ClusterGateway) (*restclient.Config, error) {
 	cfg := &restclient.Config{}
 	// setting up endpoint
 	switch c.Spec.Access.Endpoint.Type {
@@ -69,7 +74,7 @@ func NewConfigFromCluster(c *ClusterGateway) (*restclient.Config, error) {
 		cfg.Host = c.Name // the same as the cluster name
 		cfg.Insecure = true
 		cfg.CAData = nil
-		dail, err := DialerGetter()
+		dail, err := DialerGetter(ctx)
 		if err != nil {
 			return nil, err
 		}

pkg/apis/cluster/v1alpha1/transport_test.go

@@ -21,7 +21,7 @@ func TestClusterRestConfigConversion(t *testing.T) {
 	testDialFunc := func(ctx context.Context, net, addr string) (net.Conn, error) {
 		return nil, nil
 	}
-	DialerGetter = func() (k8snet.DialFunc, error) {
+	DialerGetter = func(ctx context.Context) (k8snet.DialFunc, error) {
 		return testDialFunc, nil
 	}
 	cases := []struct {
@@ -176,7 +176,7 @@ func TestClusterRestConfigConversion(t *testing.T) {
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			cfg, err := NewConfigFromCluster(c.clusterGateway)
+			cfg, err := NewConfigFromCluster(context.TODO(), c.clusterGateway)
 			if err != nil {
 				if c.expectFailure {
 					return