feat: complete gvisor

build/Dockerfile

@@ -1,5 +1,5 @@
 FROM envoyproxy/envoy:v1.25.0 AS envoy
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 ARG BASE=github.com/wencaiwulue/kubevpn
 
 COPY . /go/src/$BASE

build/local.Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.19 AS builder
+FROM golang:1.20 AS builder
 RUN go env -w GO111MODULE=on && go env -w GOPROXY=https://goproxy.cn,direct
 RUN go install github.com/go-delve/delve/cmd/dlv@latest
 

pkg/core/chain.go

@@ -3,7 +3,6 @@ package core
 import (
 	"context"
 	"errors"
-	"fmt"
 	"math"
 	"net"
 )
@@ -61,7 +60,7 @@ func (c *Chain) dial(ctx context.Context) (net.Conn, error) {
 func (*Chain) resolve(addr string) string {
 	if host, port, err := net.SplitHostPort(addr); err == nil {
 		if ips, err := net.LookupIP(host); err == nil && len(ips) > 0 {
-			return fmt.Sprintf("%s:%s", ips[0].String(), port)
+			return net.JoinHostPort(ips[0].String(), port)
 		}
 	}
 	return addr

pkg/core/gvisortcpforwarder.go

@@ -2,7 +2,9 @@ package core
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
+	"errors"
 	"io"
 	"net"
 
@@ -16,33 +18,63 @@ import (
 	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
+var GvisorTCPForwardAddr string
+
 func TCPForwarder(s *stack.Stack) func(stack.TransportEndpointID, *stack.PacketBuffer) bool {
 	rcvWnd := 1000 << 20 // 1000MB
 	return tcp.NewForwarder(s, rcvWnd, 100000, func(request *tcp.ForwarderRequest) {
 		defer request.Complete(false)
 		id := request.ID()
-		log.Debugf("[TUN-TCP] Info: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
+		log.Debugf("[TUN-TCP] Debug: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
 			id.LocalPort, id.LocalAddress.String(), id.RemotePort, id.RemoteAddress.String(),
 		)
-		remote, err := net.Dial("tcp", "localhost:10801")
+
+		node, err := ParseNode(GvisorTCPForwardAddr)
 		if err != nil {
-			log.Warningln(err)
+			log.Errorf("[TUN-TCP] Error: can not parse gvisor tcp forward addr %s: %v", GvisorTCPForwardAddr, err)
+			return
+		}
+		node.Client = &Client{
+			Connector:   GvisorTCPTunnelConnector(),
+			Transporter: TCPTransporter(),
+		}
+		forwardChain := NewChain(5, node)
+
+		remote, err := forwardChain.dial(context.Background())
+		if err != nil {
+			log.Errorf("[TUN-TCP] Error: failed to dial remote conn: %v", err)
 			return
 		}
 		if err = WriteProxyInfo(remote, id); err != nil {
-			log.Warningln(err)
+			log.Errorf("[TUN-TCP] Error: failed to write proxy info: %v", err)
 			return
 		}
 
 		w := &waiter.Queue{}
-		endpoint, t := request.CreateEndpoint(w)
-		if t != nil {
-			log.Warningln(t)
+		endpoint, tErr := request.CreateEndpoint(w)
+		if tErr != nil {
+			log.Errorf("[TUN-TCP] Error: can not create endpoint: %v", tErr)
 			return
 		}
 		conn := gonet.NewTCPConn(w, endpoint)
-		go io.Copy(remote, conn)
-		io.Copy(conn, remote)
+
+		defer conn.Close()
+		defer remote.Close()
+		errChan := make(chan error, 2)
+		go func() {
+			written, err2 := io.Copy(remote, conn)
+			log.Errorf("[TUN-TCP] Debug: write length %d data to remote", written)
+			errChan <- err2
+		}()
+		go func() {
+			written, err2 := io.Copy(conn, remote)
+			log.Errorf("[TUN-TCP] Debug: read length %d data from remote", written)
+			errChan <- err2
+		}()
+		err = <-errChan
+		if err != nil && !errors.Is(err, io.EOF) {
+			log.Errorf("[TUN-TCP] Error: dsiconnect: %s >-<: %s: %v", conn.LocalAddr(), remote.RemoteAddr(), err)
+		}
 	}).HandlePacket
 }
 

pkg/core/gvisortcphandler.go

@@ -2,14 +2,41 @@ package core
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"io"
 	"net"
-	"strconv"
 	"time"
 
 	log "github.com/sirupsen/logrus"
 )
 
+type gvisorTCPTunnelConnector struct {
+}
+
+func GvisorTCPTunnelConnector() Connector {
+	return &gvisorTCPTunnelConnector{}
+}
+
+func (c *gvisorTCPTunnelConnector) ConnectContext(ctx context.Context, conn net.Conn) (net.Conn, error) {
+	switch con := conn.(type) {
+	case *net.TCPConn:
+		err := con.SetNoDelay(true)
+		if err != nil {
+			return nil, err
+		}
+		err = con.SetKeepAlive(true)
+		if err != nil {
+			return nil, err
+		}
+		err = con.SetKeepAlivePeriod(15 * time.Second)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return conn, nil
+}
+
 type gvisorTCPHandler struct{}
 
 func GvisorTCPHandler() Handler {
@@ -18,30 +45,41 @@ func GvisorTCPHandler() Handler {
 
 func (h *gvisorTCPHandler) Handle(ctx context.Context, tcpConn net.Conn) {
 	defer tcpConn.Close()
-	log.Debugf("[GvisorTCPServer] %s -> %s\n", tcpConn.RemoteAddr(), tcpConn.LocalAddr())
-	func(conn net.Conn) {
-		defer conn.Close()
+	log.Debugf("[TUN-TCP] %s -> %s", tcpConn.RemoteAddr(), tcpConn.LocalAddr())
 	// 1, get proxy info
-		endpointID, err := ParseProxyInfo(conn)
+	endpointID, err := ParseProxyInfo(tcpConn)
 	if err != nil {
-			log.Warning("failed to parse proxy info", "err: ", err)
+		log.Errorf("[TUN-TCP] Error: failed to parse proxy info: %v", err)
 		return
 	}
-		log.Debugf("[TUN-TCP] Info: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
+	log.Debugf("[TUN-TCP] Debug: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
 		endpointID.LocalPort, endpointID.LocalAddress.String(), endpointID.RemotePort, endpointID.RemoteAddress.String(),
 	)
 	// 2, dial proxy
-		s := net.ParseIP(endpointID.LocalAddress.String()).String()
-		port := strconv.FormatUint(uint64(endpointID.LocalPort), 10)
-		var dial net.Conn
-		dial, err = net.DialTimeout("tcp", net.JoinHostPort(s, port), time.Second*5)
+	host := endpointID.LocalAddress.String()
+	port := fmt.Sprintf("%d", endpointID.LocalPort)
+	var remote net.Conn
+	remote, err = net.DialTimeout("tcp", net.JoinHostPort(host, port), time.Second*5)
 	if err != nil {
-			log.Warningln(err)
+		log.Errorf("[TUN-TCP] Error: failed to connect addr %s: %v", net.JoinHostPort(host, port), err)
 		return
 	}
-		go io.Copy(conn, dial)
-		io.Copy(dial, conn)
-	}(tcpConn)
+
+	errChan := make(chan error, 2)
+	go func() {
+		written, err2 := io.Copy(remote, tcpConn)
+		log.Errorf("[TUN-TCP] Debug: write length %d data to remote", written)
+		errChan <- err2
+	}()
+	go func() {
+		written, err2 := io.Copy(tcpConn, remote)
+		log.Errorf("[TUN-TCP] Debug: read length %d data from remote", written)
+		errChan <- err2
+	}()
+	err = <-errChan
+	if err != nil && !errors.Is(err, io.EOF) {
+		log.Errorf("[TUN-TCP] Error: dsiconnect: %s >-<: %s: %v", tcpConn.LocalAddr(), remote.RemoteAddr(), err)
+	}
 }
 
 func GvisorTCPListener(addr string) (net.Listener, error) {

pkg/core/gvisorudpforwarder.go

@@ -2,8 +2,8 @@ package core
 
 import (
 	"context"
+	"errors"
 	"io"
-	"net"
 
 	log "github.com/sirupsen/logrus"
 	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
@@ -12,36 +12,65 @@ import (
 	"gvisor.dev/gvisor/pkg/waiter"
 )
 
+var GvisorUDPForwardAddr string
+
 func UDPForwarder(s *stack.Stack) func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool {
 	return udp.NewForwarder(s, func(request *udp.ForwarderRequest) {
 		endpointID := request.ID()
-		log.Infof("[TUN-UDP] Info: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
+		log.Debugf("[TUN-UDP] Debug: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
 			endpointID.LocalPort, endpointID.LocalAddress.String(), endpointID.RemotePort, endpointID.RemoteAddress.String(),
 		)
 		w := &waiter.Queue{}
-		endpoint, t := request.CreateEndpoint(w)
-		if t != nil {
-			log.Warningln(t)
+		endpoint, tErr := request.CreateEndpoint(w)
+		if tErr != nil {
+			log.Errorf("[TUN-UDP] Error: can not create endpoint: %v", tErr)
 			return
 		}
-		c, err2 := net.Dial("tcp", "127.0.0.1:10802")
-		if err2 != nil {
-			log.Error(err2)
-		}
-		if err := WriteProxyInfo(c, endpointID); err != nil {
-			log.Warningln(err)
-			return
-		}
-		ctx := context.Background()
-		dial, err := GvisorUDPOverTCPTunnelConnector(endpointID).ConnectContext(ctx, c)
+
+		node, err := ParseNode(GvisorUDPForwardAddr)
 		if err != nil {
-			log.Warningln(err)
+			log.Errorf("[TUN-UDP] Error: parse gviosr udp forward addr %s: %v", GvisorUDPForwardAddr, err)
+			return
+		}
+		node.Client = &Client{
+			Connector:   GvisorUDPOverTCPTunnelConnector(endpointID),
+			Transporter: TCPTransporter(),
+		}
+		forwardChain := NewChain(5, node)
+
+		ctx := context.Background()
+		c, err := forwardChain.getConn(ctx)
+		if err != nil {
+			log.Errorf("[TUN-UDP] Error: can not get conn: %v", err)
+		}
+		if err = WriteProxyInfo(c, endpointID); err != nil {
+			log.Errorf("[TUN-UDP] Error: can not write proxy info: %v", err)
+			return
+		}
+		remote, err := node.Client.ConnectContext(ctx, c)
+		if err != nil {
+			log.Errorf("[TUN-UDP] Error: can not connect: %v", err)
 			return
 		}
 		conn := gonet.NewUDPConn(s, w, endpoint)
 		go func() {
-			go io.Copy(dial, conn)
-			io.Copy(conn, dial)
+			defer conn.Close()
+			defer remote.Close()
+			errChan := make(chan error, 2)
+			go func() {
+				written, err2 := io.Copy(remote, conn)
+				log.Errorf("[TUN-UDP] Debug: write length %d data to remote", written)
+				errChan <- err2
+			}()
+			go func() {
+				written, err2 := io.Copy(conn, remote)
+				log.Errorf("[TUN-UDP] Debug: read length %d data from remote", written)
+				errChan <- err2
+			}()
+			err = <-errChan
+			if err != nil && !errors.Is(err, io.EOF) {
+				log.Errorf("[TUN-UDP] Error: dsiconnect: %s >-<: %s: %v", conn.LocalAddr(), remote.RemoteAddr(), err)
+			}
 		}()
 	}).HandlePacket
 }

pkg/core/gvisorudphandler.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -14,17 +13,17 @@ import (
 	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
-type GvisorFakeUDPTunnelConnector struct {
+type gvisorUDPOverTCPTunnelConnector struct {
 	Id stack.TransportEndpointID
 }
 
 func GvisorUDPOverTCPTunnelConnector(endpointID stack.TransportEndpointID) Connector {
-	return &GvisorFakeUDPTunnelConnector{
+	return &gvisorUDPOverTCPTunnelConnector{
 		Id: endpointID,
 	}
 }
 
-func (c *GvisorFakeUDPTunnelConnector) ConnectContext(ctx context.Context, conn net.Conn) (net.Conn, error) {
+func (c *gvisorUDPOverTCPTunnelConnector) ConnectContext(ctx context.Context, conn net.Conn) (net.Conn, error) {
 	switch con := conn.(type) {
 	case *net.TCPConn:
 		err := con.SetNoDelay(true)
@@ -52,31 +51,27 @@ func GvisorUDPHandler() Handler {
 func (h *gvisorUDPHandler) Handle(ctx context.Context, tcpConn net.Conn) {
 	defer tcpConn.Close()
 	log.Debugf("[TUN-UDP] %s -> %s", tcpConn.RemoteAddr(), tcpConn.LocalAddr())
-	func(conn net.Conn) {
-		defer conn.Close()
 	// 1, get proxy info
-		endpointID, err := ParseProxyInfo(conn)
+	endpointID, err := ParseProxyInfo(tcpConn)
 	if err != nil {
-			log.Warningf("[TUN-UDP] ERROR Failed to parse proxy info: %v", err)
+		log.Warningf("[TUN-UDP] Error: Failed to parse proxy info: %v", err)
 		return
 	}
-		log.Infof("[TUN-UDP] Info: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
+	log.Debugf("[TUN-UDP] Debug: LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
 		endpointID.LocalPort, endpointID.LocalAddress.String(), endpointID.RemotePort, endpointID.RemoteAddress.String(),
 	)
 	// 2, dial proxy
-		var dial *net.UDPConn
-		// todo 发送到 localhost:8422 ? 🤔
 	addr := &net.UDPAddr{
 		IP:   endpointID.LocalAddress.AsSlice(),
 		Port: int(endpointID.LocalPort),
 	}
-		dial, err = net.DialUDP("udp", nil, addr)
+	var remote *net.UDPConn
+	remote, err = net.DialUDP("udp", nil, addr)
 	if err != nil {
-			log.Warningln(err)
+		log.Errorf("[TUN-UDP] Error: failed to connect addr %s: %v", addr.String(), err)
 		return
 	}
-		h.HandleInner(ctx, tcpConn, dial)
-	}(tcpConn)
+	handle(ctx, tcpConn, remote)
 }
 
 // fake udp connect over tcp
@@ -93,7 +88,7 @@ func newGvisorFakeUDPTunnelConnOverTCP(ctx context.Context, conn net.Conn) (net.
 func (c *gvisorFakeUDPTunnelConn) Read(b []byte) (int, error) {
 	select {
 	case <-c.ctx.Done():
-		return 0, errors.New("closed connection")
+		return 0, c.ctx.Err()
 	default:
 		dgram, err := readDatagramPacket(c.Conn, b)
 		if err != nil {
@@ -156,10 +151,9 @@ func copyPacketData(dst, src net.PacketConn, to net.Addr, timeout time.Duration)
 	}
 }
 
-func (h *gvisorUDPHandler) HandleInner(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
+func handle(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
 	defer udpConn.Close()
-	log.Debugf("[TUN-UDP] %s -> %s", tcpConn.RemoteAddr(), tcpConn.LocalAddr())
-	log.Debugf("[TUN-UDP] udp-tun %s <-> %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
+	log.Debugf("[TUN-UDP] Debug: %s <-> %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
 	errChan := make(chan error, 2)
 	go func() {
 		b := config.LPool.Get().([]byte)
@@ -168,22 +162,22 @@ func (h *gvisorUDPHandler) HandleInner(ctx context.Context, tcpConn net.Conn, ud
 		for {
 			dgram, err := readDatagramPacket(tcpConn, b[:])
 			if err != nil {
-				log.Debugf("[TUN-UDP] %s -> 0 : %v", tcpConn.RemoteAddr(), err)
+				log.Debugf("[TUN-UDP] Debug: %s -> 0 : %v", tcpConn.RemoteAddr(), err)
 				errChan <- err
 				return
 			}
 			if dgram.DataLength == 0 {
-				log.Debugf("[TUN-UDP] length is zero")
+				log.Debugf("[TUN-UDP] Error: length is zero")
 				errChan <- fmt.Errorf("length of read packet is zero")
 				return
 			}
 
 			if _, err = udpConn.Write(dgram.Data); err != nil {
-				log.Debugf("[TUN-UDP] udp-tun %s -> %s : %s", tcpConn.RemoteAddr(), Server8422, err)
+				log.Debugf("[TUN-UDP] Error: %s -> %s : %s", tcpConn.RemoteAddr(), Server8422, err)
 				errChan <- err
 				return
 			}
-			log.Debugf("[TUN-UDP] udp-tun %s >>> %s length: %d", tcpConn.RemoteAddr(), Server8422, dgram.DataLength)
+			log.Debugf("[TUN-UDP] Debug: %s >>> %s length: %d", tcpConn.RemoteAddr(), Server8422, dgram.DataLength)
 		}
 	}()
 
@@ -194,12 +188,12 @@ func (h *gvisorUDPHandler) HandleInner(ctx context.Context, tcpConn net.Conn, ud
 		for {
 			n, _, err := udpConn.ReadFrom(b[:])
 			if err != nil {
-				log.Debugf("[TUN-UDP] %s : %s", tcpConn.RemoteAddr(), err)
+				log.Debugf("[TUN-UDP] Error: %s : %s", tcpConn.RemoteAddr(), err)
 				errChan <- err
 				return
 			}
 			if n == 0 {
-				log.Debugf("[TUN-UDP] length is zero")
+				log.Debugf("[TUN-UDP] Error: length is zero")
 				errChan <- fmt.Errorf("length of read packet is zero")
 				return
 			}
@@ -207,17 +201,17 @@ func (h *gvisorUDPHandler) HandleInner(ctx context.Context, tcpConn net.Conn, ud
 			// pipe from peer to tunnel
 			dgram := newDatagramPacket(b[:n])
 			if err = dgram.Write(tcpConn); err != nil {
-				log.Debugf("[TUN-UDP] udp-tun %s <- %s : %s", tcpConn.RemoteAddr(), dgram.Addr(), err)
+				log.Debugf("[TUN-UDP] Error: %s <- %s : %s", tcpConn.RemoteAddr(), dgram.Addr(), err)
 				errChan <- err
 				return
 			}
-			log.Debugf("[TUN-UDP] udp-tun %s <<< %s length: %d", tcpConn.RemoteAddr(), dgram.Addr(), len(dgram.Data))
+			log.Debugf("[TUN-UDP] Debug: %s <<< %s length: %d", tcpConn.RemoteAddr(), dgram.Addr(), len(dgram.Data))
 		}
 	}()
 	err := <-errChan
 	if err != nil {
-		log.Error(err)
+		log.Debugf("[TUN-UDP] Error: %v", err)
 	}
-	log.Debugf("[TUN-UDP] udp-tun %s >-< %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
+	log.Debugf("[TUN-UDP] Debug: %s >-< %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
 	return
 }

pkg/core/tcphandler.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"context"
-	"errors"
 	"net"
 	"sync"
 	"time"
@@ -119,7 +118,7 @@ func newFakeUDPTunnelConnOverTCP(ctx context.Context, conn net.Conn) (net.Conn,
 func (c *fakeUDPTunnelConn) ReadFrom(b []byte) (int, net.Addr, error) {
 	select {
 	case <-c.ctx.Done():
-		return 0, nil, errors.New("closed connection")
+		return 0, nil, c.ctx.Err()
 	default:
 		dgram, err := readDatagramPacket(c.Conn, b)
 		if err != nil {

pkg/core/tunhandler.go

@@ -229,7 +229,7 @@ func (d *Device) parseIPHeader() {
 			continue
 		}
 
-		log.Debugf("[tun] %s --> %s", e.src, e.dst)
+		log.Debugf("[tun] %s --> %s, length: %d", e.src, e.dst, e.length)
 		d.tunInbound <- e
 	}
 }
@@ -370,7 +370,7 @@ func (d *Device) Start(ctx context.Context) {
 	}
 	go d.tunInboundHandler(d.tunInbound, d.tunOutbound)
 	go d.writeToTun()
-	//go heartbeats(d.tunInbound)
+	go heartbeats(d.tunInbound)
 
 	select {
 	case err := <-d.chExit:

pkg/core/tunhandlerclient.go

@@ -117,7 +117,7 @@ type ClientDevice struct {
 
 func (d *ClientDevice) Start(ctx context.Context) {
 	go d.tunInboundHandler(d.tunInbound, d.tunOutbound)
-	//go heartbeats(d.tunInbound)
+	go heartbeats(d.tunInbound)
 
 	select {
 	case err := <-d.chExit:

pkg/dns/dns_server.go

@@ -4,9 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"math"
 	"math/rand"
+	"net"
 	"os"
 	"strings"
 	"sync"
@@ -102,7 +102,7 @@ func (s *server) ServeDNS(w miekgdns.ResponseWriter, r *miekgdns.Msg) {
 				msg.Ns = nil
 				msg.Extra = nil
 				msg.Id = uint16(rand.Intn(math.MaxUint16 + 1))
-				answer, _, err := s.client.ExchangeContext(context.Background(), &msg, fmt.Sprintf("%s:%s", dnsAddr, s.forwardDNS.Port))
+				answer, _, err := s.client.ExchangeContext(context.Background(), &msg, net.JoinHostPort(dnsAddr, s.forwardDNS.Port))
 
 				if err == nil && len(answer.Answer) != 0 {
 					s.dnsCache.Add(originName, name, time.Hour*24*365*100) // never expire

pkg/handler/connect.go

@@ -158,16 +158,20 @@ func (c *ConnectOptions) DoConnect() (err error) {
 	if err = c.portForward(ctx, fmt.Sprintf("%d:10800", port)); err != nil {
 		return
 	}
-	if err = c.portForward(ctx, "10801:10801"); err != nil {
+	tcpForwardPort := util.GetAvailableTCPPortOrDie()
+	if err = c.portForward(ctx, fmt.Sprintf("%d:10801", tcpForwardPort)); err != nil {
 		return
 	}
-	if err = c.portForward(ctx, "10802:10802"); err != nil {
+	udpForwardPort := util.GetAvailableTCPPortOrDie()
+	if err = c.portForward(ctx, fmt.Sprintf("%d:10802", udpForwardPort)); err != nil {
 		return
 	}
 	if util.IsWindows() {
 		driver.InstallWireGuardTunDriver()
 	}
 	forward := fmt.Sprintf("tcp://127.0.0.1:%d", port)
+	core.GvisorTCPForwardAddr = fmt.Sprintf("tcp://127.0.0.1:%d", tcpForwardPort)
+	core.GvisorUDPForwardAddr = fmt.Sprintf("tcp://127.0.0.1:%d", udpForwardPort)
 	if err = c.startLocalTunServe(ctx, forward); err != nil {
 		return
 	}
@@ -181,13 +185,13 @@ func (c *ConnectOptions) DoConnect() (err error) {
 	if err = c.setupDNS(); err != nil {
 		return
 	}
-	//go c.heartbeats()
+	go c.heartbeats()
 	log.Info("dns service ok")
 	return
 }
 
 // detect pod is delete event, if pod is deleted, needs to redo port-forward immediately
-func (c *ConnectOptions) portForward(ctx context.Context, port string) error {
+func (c *ConnectOptions) portForward(ctx context.Context, portPair string) error {
 	var readyChan = make(chan struct{}, 1)
 	var errChan = make(chan error, 1)
 	podInterface := c.clientset.CoreV1().Pods(c.Namespace)
@@ -218,7 +222,7 @@ func (c *ConnectOptions) portForward(ctx context.Context, port string) error {
 					c.restclient,
 					podName,
 					c.Namespace,
-					port,
+					portPair,
 					readyChan,
 					childCtx.Done(),
 				)
@@ -232,7 +236,7 @@ func (c *ConnectOptions) portForward(ctx context.Context, port string) error {
 				}
 				if strings.Contains(err.Error(), "unable to listen on any of the requested ports") ||
 					strings.Contains(err.Error(), "address already in use") {
-					log.Errorf("port %s already in use, needs to release it manually", port)
+					log.Errorf("port %s already in use, needs to release it manually", portPair)
 					time.Sleep(time.Second * 5)
 				} else {
 					log.Debugf("port-forward occurs error, err: %v, retrying", err)
@@ -247,7 +251,7 @@ func (c *ConnectOptions) portForward(ctx context.Context, port string) error {
 	case err := <-errChan:
 		return err
 	case <-readyChan:
-		log.Info("port forward ready")
+		log.Infof("port forward %s ready", portPair)
 		return nil
 	}
 }
@@ -576,7 +580,7 @@ func Run(ctx context.Context, servers []core.Server) error {
 			for {
 				select {
 				case <-ctx.Done():
-					return nil
+					return ctx.Err()
 				default:
 				}
 

pkg/util/ssh.go

@@ -245,7 +245,7 @@ func getBastion(name string) *SshConfig {
 			config.Keyfile = value
 		}
 	}
-	config.Addr = fmt.Sprintf("%s:%s", host, port)
+	config.Addr = net.JoinHostPort(host, port)
 	return &config
 }
 

pkg/util/util.go

@@ -102,7 +102,7 @@ func WaitPod(podInterface v12.PodInterface, list metav1.ListOptions, checker fun
 	}
 }
 
-func PortForwardPod(config *rest.Config, clientset *rest.RESTClient, podName, namespace, port string, readyChan chan struct{}, stopChan <-chan struct{}) error {
+func PortForwardPod(config *rest.Config, clientset *rest.RESTClient, podName, namespace, portPair string, readyChan chan struct{}, stopChan <-chan struct{}) error {
 	url := clientset.
 		Post().
 		Resource("pods").
@@ -116,7 +116,7 @@ func PortForwardPod(config *rest.Config, clientset *rest.RESTClient, podName, na
 		return err
 	}
 	dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", url)
-	p := []string{port}
+	p := []string{portPair}
 	forwarder, err := portforward.NewOnAddresses(dialer, []string{"0.0.0.0"}, p, stopChan, readyChan, nil, os.Stderr)
 	if err != nil {
 		log.Error(err)