feat: move folder to pkg

.github/workflows/test.yml

@@ -57,7 +57,7 @@ jobs:
           route -n
 
       - name: Test
-        run: go test -v ./test/function_test.go
+        run: go test -v ./pkg/test/function_test.go
 
   macos:
     runs-on: macos-latest
@@ -116,7 +116,7 @@ jobs:
           netstat -anr
 
       - name: Test
-        run: go test -v ./test/function_test.go
+        run: go test -v ./pkg/test/function_test.go
 
 #  windows:
 #    runs-on: windows-latest

.github/workflows/upload_release.yml

@@ -39,6 +39,6 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
           upload_url: ${{ github.event.client_payload.url }}
-          asset_path: ${{ matrix.os-arch }}
+          asset_path: ./bin/${{ matrix.os-arch }}
           asset_name: ${{ matrix.os-arch }}
           asset_content_type: application/octet-stream

.gitignore

@@ -19,4 +19,4 @@
 .vscode/
 
 # Build artifacts
-kubevpn
+bin

Makefile

@@ -10,7 +10,9 @@ OS_ARCH := ${GOOS}/${GOARCH}
 
 BASE := github.com/wencaiwulue/kubevpn
 FOLDER := ${BASE}/cmd/kubevpn
-CONTROL_PLANE_FOLDER := ${BASE}/cmd/mesh
+BUILD_DIR := ./build
+OUTPUT_DIR := ./bin
+REGISTRY ?= naison
 
 # Setup the -ldflags option for go build here, interpolate the variable values
 LDFLAGS=--ldflags "\
@@ -35,70 +37,60 @@ all-image: image image-mesh image-control-plane
 # ---------darwin-----------
 .PHONY: kubevpn-darwin-amd64
 kubevpn-darwin-amd64:
-	CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build ${LDFLAGS} -o kubevpn-darwin-amd64 ${FOLDER}
+	CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-darwin-amd64 ${FOLDER}
-	chmod +x kubevpn-darwin-amd64
+	chmod +x $(OUTPUT_DIR)/kubevpn-darwin-amd64
-	cp kubevpn-darwin-amd64 /usr/local/bin/kubevpn
 .PHONY: kubevpn-darwin-arm64
 kubevpn-darwin-arm64:
-	CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build ${LDFLAGS} -o kubevpn-darwin-arm64 ${FOLDER}
+	CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-darwin-arm64 ${FOLDER}
-	chmod +x kubevpn-darwin-arm64
+	chmod +x $(OUTPUT_DIR)/kubevpn-darwin-arm64
-	cp kubevpn-darwin-arm64 /usr/local/bin/kubevpn
 # ---------darwin-----------
 
 # ---------windows-----------
 .PHONY: kubevpn-windows-amd64
 kubevpn-windows-amd64:
-	CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build ${LDFLAGS} -o kubevpn-windows-amd64.exe ${FOLDER}
+	CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-windows-amd64.exe ${FOLDER}
 .PHONY: kubevpn-windows-arm64
 kubevpn-windows-arm64:
-	CGO_ENABLED=0 GOOS=windows GOARCH=arm64 go build ${LDFLAGS} -o kubevpn-windows-arm64.exe ${FOLDER}
+	CGO_ENABLED=0 GOOS=windows GOARCH=arm64 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-windows-arm64.exe ${FOLDER}
 .PHONY: kubevpn-windows-386
 kubevpn-windows-386:
-	CGO_ENABLED=0 GOOS=windows GOARCH=386 go build ${LDFLAGS} -o kubevpn-windows-386.exe ${FOLDER}
+	CGO_ENABLED=0 GOOS=windows GOARCH=386 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-windows-386.exe ${FOLDER}
 # ---------windows-----------
 
 # ---------linux-----------
 .PHONY: kubevpn-linux-amd64
 kubevpn-linux-amd64:
-	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build ${LDFLAGS} -o kubevpn-linux-amd64 ${FOLDER}
+	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-linux-amd64 ${FOLDER}
-	chmod +x kubevpn-linux-amd64
+	chmod +x $(OUTPUT_DIR)/kubevpn-linux-amd64
-	cp kubevpn-linux-amd64 /usr/local/bin/kubevpn
 .PHONY: kubevpn-linux-arm64
 kubevpn-linux-arm64:
-	CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build ${LDFLAGS} -o kubevpn-linux-arm64 ${FOLDER}
+	CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-linux-arm64 ${FOLDER}
-	chmod +x kubevpn-linux-arm64
+	chmod +x $(OUTPUT_DIR)/kubevpn-linux-arm64
-	cp kubevpn-linux-arm64 /usr/local/bin/kubevpn
 .PHONY: kubevpn-linux-386
 kubevpn-linux-386:
-	CGO_ENABLED=0 GOOS=linux GOARCH=386 go build ${LDFLAGS} -o kubevpn-linux-386 ${FOLDER}
+	CGO_ENABLED=0 GOOS=linux GOARCH=386 go build ${LDFLAGS} -o $(OUTPUT_DIR)/kubevpn-linux-386 ${FOLDER}
-	chmod +x kubevpn-linux-386
+	chmod +x $(OUTPUT_DIR)/kubevpn-linux-386
-	cp kubevpn-linux-386 /usr/local/bin/kubevpn
 # ---------linux-----------
 
 .PHONY: image
-image: kubevpn-linux-amd64
+image:
-	mv kubevpn-linux-amd64 kubevpn
+	docker build -t $(REGISTRY)/kubevpn:${VERSION} -f $(BUILD_DIR)/server/Dockerfile .
-	docker build -t naison/kubevpn:${VERSION} -f ./dockerfile/server/Dockerfile .
+	docker tag $(REGISTRY)/kubevpn:${VERSION} $(REGISTRY)/kubevpn:latest
-	rm -fr kubevpn
+	docker push $(REGISTRY)/kubevpn:${VERSION}
-	docker tag naison/kubevpn:${VERSION} naison/kubevpn:latest
+	docker push $(REGISTRY)/kubevpn:latest
-	docker push naison/kubevpn:${VERSION}
-	docker push naison/kubevpn:latest
 
 .PHONY: image-mesh
 image-mesh:
-	docker build -t naison/kubevpn-mesh:${VERSION} -f ./dockerfile/mesh/Dockerfile .
+	docker build -t $(REGISTRY)/kubevpn-mesh:${VERSION} -f $(BUILD_DIR)/mesh/Dockerfile .
-	docker tag naison/kubevpn-mesh:${VERSION} naison/kubevpn-mesh:latest
+	docker tag $(REGISTRY)/kubevpn-mesh:${VERSION} $(REGISTRY)/kubevpn-mesh:latest
-	docker push naison/kubevpn-mesh:${VERSION}
+	docker push $(REGISTRY)/kubevpn-mesh:${VERSION}
-	docker push naison/kubevpn-mesh:latest
+	docker push $(REGISTRY)/kubevpn-mesh:latest
 
 
 .PHONY: image-control-plane
 image-control-plane:
-	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o envoy-xds-server ${CONTROL_PLANE_FOLDER}
+	docker build -t $(REGISTRY)/envoy-xds-server:${VERSION} -f $(BUILD_DIR)/control_plane/Dockerfile .
-	chmod +x envoy-xds-server
+	docker tag $(REGISTRY)/envoy-xds-server:${VERSION} $(REGISTRY)/envoy-xds-server:latest
-	docker build -t naison/envoy-xds-server:${VERSION} -f ./dockerfile/control_plane/Dockerfile .
+	docker push $(REGISTRY)/envoy-xds-server:${VERSION}
-	rm -fr envoy-xds-server
+	docker push $(REGISTRY)/envoy-xds-server:latest
-	docker tag naison/envoy-xds-server:${VERSION} naison/envoy-xds-server:latest
-	docker push naison/envoy-xds-server:${VERSION}
-	docker push naison/envoy-xds-server:latest
 

build/control_plane/Dockerfile

@@ -0,0 +1,20 @@
+ARG BASE=github.com/wencaiwulue/kubevpn
+
+FROM golang:1.18 AS builder
+
+COPY . /go/src/$BASE
+
+WORKDIR /go/src/$BASE
+
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o envoy-xds-server /go/src/$BASE/cmd/mesh
+
+FROM ubuntu:latest
+
+RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
+    && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
+RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl  \
+    net-tools iptables iputils-ping lsof iproute2 tcpdump
+
+WORKDIR /app
+
+COPY --from=builder /go/src/$BASE/envoy-xds-server /bin/envoy-xds-server

build/mesh/Dockerfile

@@ -3,4 +3,5 @@ FROM envoyproxy/envoy:v1.21.1
 WORKDIR /app
 RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
     && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
-RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl net-tools iptables iputils-ping lsof iproute2 tcpdump
+RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl  \
+    net-tools iptables iputils-ping lsof iproute2 tcpdump

build/server/Dockerfile

@@ -0,0 +1,20 @@
+ARG BASE=github.com/wencaiwulue/kubevpn
+
+FROM golang:1.18 AS builder
+
+COPY . /go/src/$BASE
+
+WORKDIR /go/src/$BASE
+
+RUN make kubevpn-linux-amd64
+
+FROM ubuntu:latest
+
+RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
+    && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
+RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl  \
+    net-tools iptables iputils-ping lsof iproute2 tcpdump
+
+WORKDIR /app
+
+COPY --from=builder /go/src/$BASE/bin/kubevpn-linux-amd64 /usr/local/bin/kubevpn

cmd/kubevpn/cmds/connect.go

@@ -12,19 +12,19 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/retry"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/driver"
+	"github.com/wencaiwulue/kubevpn/pkg/driver"
-	"github.com/wencaiwulue/kubevpn/pkg"
+	"github.com/wencaiwulue/kubevpn/pkg/handler"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
-var connect = pkg.ConnectOptions{}
+var connect = handler.ConnectOptions{}
 
 func init() {
 	connectCmd.Flags().StringVar(&connect.KubeconfigPath, "kubeconfig", clientcmd.RecommendedHomeFile, "kubeconfig")
 	connectCmd.Flags().StringVarP(&connect.Namespace, "namespace", "n", "", "namespace")
 	connectCmd.PersistentFlags().StringArrayVar(&connect.Workloads, "workloads", []string{}, "workloads, like: pods/tomcat, deployment/nginx, replicaset/tomcat...")
-	connectCmd.Flags().StringVar((*string)(&connect.Mode), "mode", string(pkg.Reverse), "default mode is reverse")
+	connectCmd.Flags().StringVar((*string)(&connect.Mode), "mode", string(handler.Reverse), "default mode is reverse")
 	connectCmd.Flags().StringToStringVarP(&connect.Headers, "headers", "H", map[string]string{}, "headers, format is k=v, like: k1=v1,k2=v2")
 	connectCmd.Flags().BoolVar(&config.Debug, "debug", false, "true/false")
 	RootCmd.AddCommand(connectCmd)
@@ -55,7 +55,7 @@ var connectCmd = &cobra.Command{
 		connect.PreCheckResource()
 		if err := connect.DoConnect(); err != nil {
 			log.Errorln(err)
-			pkg.Cleanup(syscall.SIGQUIT)
+			handler.Cleanup(syscall.SIGQUIT)
 			return
 		}
 		fmt.Println(`

cmd/kubevpn/cmds/serve.go

@@ -7,12 +7,12 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/pkg"
+	"github.com/wencaiwulue/kubevpn/pkg/handler"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
-var route pkg.Route
+var route handler.Route
 
 func init() {
 	ServerCmd.Flags().StringArrayVarP(&route.ServeNodes, "nodeCommand", "L", []string{}, "command needs to be executed")
@@ -30,7 +30,7 @@ var ServerCmd = &cobra.Command{
 		go func() { log.Info(http.ListenAndServe("localhost:6060", nil)) }()
 	},
 	Run: func(cmd *cobra.Command, args []string) {
-		if err := pkg.Start(context.TODO(), route); err != nil {
+		if err := handler.Start(context.TODO(), route); err != nil {
 			log.Fatal(err)
 		}
 		select {}

cmd/kubevpn/cmds/version.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 // --ldflags -X

cmd/mesh/main.go

@@ -9,7 +9,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/wencaiwulue/kubevpn/pkg/control_plane"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 var (

dockerfile/control_plane/Dockerfile

@@ -1,6 +0,0 @@
-FROM ubuntu:latest
-WORKDIR /app
-RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
-    && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
-RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl net-tools iptables iputils-ping lsof iproute2 tcpdump
-COPY envoy-xds-server /bin/envoy-xds-server

dockerfile/server/Dockerfile

@@ -1,6 +0,0 @@
-FROM ubuntu:latest
-WORKDIR /app
-RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
-    && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
-RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl net-tools iptables iputils-ping lsof iproute2 tcpdump
-COPY ./kubevpn /usr/local/bin/kubevpn

pkg/config/tls_test.go

@@ -8,7 +8,7 @@ import (
 	"testing"
 
 	log "github.com/sirupsen/logrus"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 func init() {

pkg/core/pool.go

@@ -3,7 +3,7 @@ package core
 import (
 	"sync"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 var (

pkg/core/tcp.go

@@ -4,7 +4,7 @@ import (
 	"crypto/tls"
 	"net"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 type tcpTransporter struct{}

pkg/core/tcphandler.go

@@ -8,7 +8,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 type fakeUDPTunnelConnector struct {

pkg/core/tunhandler.go

@@ -13,7 +13,7 @@ import (
 	"golang.org/x/net/ipv4"
 	"golang.org/x/net/ipv6"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 func ipToTunRouteKey(ip net.IP) string {

pkg/dns/dns.go

@@ -12,7 +12,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 func GetDNSServiceIPFromPod(clientset *kubernetes.Clientset, restclient *rest.RESTClient, config *rest.Config, podName, namespace string) (*miekgdns.ClientConfig, error) {

pkg/dns/dns_server_test.go

@@ -8,7 +8,7 @@ import (
 
 	miekgdns "github.com/miekg/dns"
 
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 func TestSetupDnsServer(t *testing.T) {

pkg/dns/dns_unix.go

@@ -20,7 +20,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/util/sets"
 
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 var cancel context.CancelFunc

pkg/driver/driver.go

@@ -8,8 +8,8 @@ import (
 	log "github.com/sirupsen/logrus"
 	"k8s.io/client-go/util/retry"
 
-	"github.com/wencaiwulue/kubevpn/driver/openvpn"
+	"github.com/wencaiwulue/kubevpn/pkg/driver/openvpn"
-	"github.com/wencaiwulue/kubevpn/driver/wintun"
+	"github.com/wencaiwulue/kubevpn/pkg/driver/wintun"
 )
 
 func InstallTunTapDriver() {

pkg/driver/wintun/amd64.go

@@ -5,9 +5,6 @@ package wintun
 
 import (
 	"embed"
-	"io/ioutil"
-	"os"
-	"path/filepath"
 )
 
 //go:embed bin/amd64/wintun.dll
@@ -18,12 +15,5 @@ func InstallWintunDriver() error {
 	if err != nil {
 		return err
 	}
-	executable, err := os.Executable()
+	return copyDriver(bytes)
-	if err != nil {
-		return err
-	}
-	filename := filepath.Join(filepath.Dir(executable), "wintun.dll")
-	_ = os.Remove(filename)
-	err = ioutil.WriteFile(filename, bytes, 644)
-	return err
 }

pkg/driver/wintun/arm.go

@@ -5,9 +5,6 @@ package wintun
 
 import (
 	"embed"
-	"io/ioutil"
-	"os"
-	"path/filepath"
 )
 
 //go:embed bin/arm/wintun.dll
@@ -18,12 +15,5 @@ func InstallWintunDriver() error {
 	if err != nil {
 		return err
 	}
-	executable, err := os.Executable()
+	return copyDriver(bytes)
-	if err != nil {
-		return err
-	}
-	filename := filepath.Join(filepath.Dir(executable), "wintun.dll")
-	_ = os.Remove(filename)
-	err = ioutil.WriteFile(filename, bytes, 644)
-	return err
 }

pkg/driver/wintun/arm64.go

@@ -5,9 +5,6 @@ package wintun
 
 import (
 	"embed"
-	"io/ioutil"
-	"os"
-	"path/filepath"
 )
 
 //go:embed bin/arm64/wintun.dll
@@ -18,12 +15,5 @@ func InstallWintunDriver() error {
 	if err != nil {
 		return err
 	}
-	executable, err := os.Executable()
+	return copyDriver(bytes)
-	if err != nil {
-		return err
-	}
-	filename := filepath.Join(filepath.Dir(executable), "wintun.dll")
-	_ = os.Remove(filename)
-	err = ioutil.WriteFile(filename, bytes, 644)
-	return err
 }

pkg/driver/wintun/func.go

@@ -0,0 +1,28 @@
+package wintun
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+)
+
+// driver download from: https://www.wintun.net/builds/wintun-0.14.1.zip
+func copyDriver(b []byte) error {
+	executable, err := os.Executable()
+	if err != nil {
+		return err
+	}
+	filename := filepath.Join(filepath.Dir(executable), "wintun.dll")
+	var content []byte
+	content, err = ioutil.ReadFile(filename)
+	if err == nil {
+		// already exists and content are same, not need to copy this file
+		if bytes.Compare(b, content) == 0 {
+			return nil
+		}
+		_ = os.Remove(filename)
+	}
+	err = ioutil.WriteFile(filename, b, 644)
+	return err
+}

pkg/driver/wintun/others.go

@@ -3,7 +3,9 @@
 
 package wintun
 
-import "github.com/pkg/errors"
+import (
+	"github.com/pkg/errors"
+)
 
 func InstallWintunDriver() error {
 	return errors.New("not implement")

pkg/driver/wintun/x86.go

@@ -6,9 +6,6 @@ package wintun
 
 import (
 	"embed"
-	"io/ioutil"
-	"os"
-	"path/filepath"
 )
 
 //go:embed bin/x86/wintun.dll
@@ -19,12 +16,5 @@ func InstallWintunDriver() error {
 	if err != nil {
 		return err
 	}
-	executable, err := os.Executable()
+	return copyDriver(bytes)
-	if err != nil {
-		return err
-	}
-	filename := filepath.Join(filepath.Dir(executable), "wintun.dll")
-	_ = os.Remove(filename)
-	err = ioutil.WriteFile(filename, bytes, 644)
-	return err
 }

pkg/exchange/controller.go

@@ -4,8 +4,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 func RemoveContainer(spec *corev1.PodSpec) {

pkg/handler/cleaner.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -16,12 +16,12 @@ import (
 	v12 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/util/retry"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/dns"
+	"github.com/wencaiwulue/kubevpn/pkg/dns"
 )
 
 var stopChan = make(chan os.Signal)
-var rollbackFuncList = make([]func(), 2)
+var RollbackFuncList = make([]func(), 2)
 var ctx, cancel = context.WithCancel(context.TODO())
 
 func (c *ConnectOptions) addCleanUpResourceHandler(clientset *kubernetes.Clientset, namespace string) {
@@ -34,7 +34,7 @@ func (c *ConnectOptions) addCleanUpResourceHandler(clientset *kubernetes.Clients
 			log.Errorf("failed to release ip to dhcp, err: %v", err)
 		}
 		cancel()
-		for _, function := range rollbackFuncList {
+		for _, function := range RollbackFuncList {
 			if function != nil {
 				function()
 			}

pkg/handler/connect.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -12,10 +12,6 @@ import (
 
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
-	"github.com/wencaiwulue/kubevpn/config"
-	"github.com/wencaiwulue/kubevpn/core"
-	"github.com/wencaiwulue/kubevpn/dns"
-	"github.com/wencaiwulue/kubevpn/util"
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -28,6 +24,12 @@ import (
 	"k8s.io/client-go/rest"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 	"k8s.io/kubectl/pkg/polymorphichelpers"
+
+	"github.com/wencaiwulue/kubevpn/pkg/config"
+	"github.com/wencaiwulue/kubevpn/pkg/core"
+	"github.com/wencaiwulue/kubevpn/pkg/dns"
+	"github.com/wencaiwulue/kubevpn/pkg/route"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 type Mode string
@@ -277,7 +279,7 @@ func (c *ConnectOptions) detectConflictDevice() {
 	if len(tun) == 0 {
 		return
 	}
-	if err := DetectAndDisableConflictDevice(tun); err != nil {
+	if err := route.DetectAndDisableConflictDevice(tun); err != nil {
 		log.Warnf("error occours while disable conflict devices, err: %v", err)
 	}
 }

pkg/handler/connect_test.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -21,7 +21,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 var (

pkg/handler/dhcp.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -13,7 +13,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 type DHCPManager struct {

pkg/handler/envoy.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -20,10 +20,10 @@ import (
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 	"sigs.k8s.io/yaml"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 	"github.com/wencaiwulue/kubevpn/pkg/control_plane"
 	"github.com/wencaiwulue/kubevpn/pkg/mesh"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 // https://istio.io/latest/docs/ops/deployment/requirements/#ports-used-by-istio
@@ -64,7 +64,7 @@ func InjectVPNAndEnvoySidecar(factory cmdutil.Factory, clientset v12.ConfigMapIn
 	}
 	if containerNames.HasAll(config.SidecarVPN, config.SidecarEnvoyProxy) {
 		// add rollback func to remove envoy config
-		rollbackFuncList = append(rollbackFuncList, func() {
+		RollbackFuncList = append(RollbackFuncList, func() {
 			err = removeEnvoyConfig(clientset, nodeID, headers)
 			if err != nil {
 				log.Warnln(err)
@@ -99,7 +99,7 @@ func InjectVPNAndEnvoySidecar(factory cmdutil.Factory, clientset v12.ConfigMapIn
 			object.Mapping.GroupVersionKind.GroupKind().String(), object.Name, err)
 	}
 
-	rollbackFuncList = append(rollbackFuncList, func() {
+	RollbackFuncList = append(RollbackFuncList, func() {
 		if err = UnPatchContainer(factory, clientset, namespace, workloads, headers); err != nil {
 			log.Error(err)
 		}

pkg/handler/remote.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -26,9 +26,9 @@ import (
 	"k8s.io/client-go/util/retry"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 	"github.com/wencaiwulue/kubevpn/pkg/exchange"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 func CreateOutboundPod(clientset *kubernetes.Clientset, namespace string, trafficManagerIP string, nodeCIDR []*net.IPNet) (net.IP, error) {
@@ -251,7 +251,7 @@ func InjectVPNSidecar(factory cmdutil.Factory, namespace, workloads string, conf
 			return err
 		}
 
-		rollbackFuncList = append(rollbackFuncList, func() {
+		RollbackFuncList = append(RollbackFuncList, func() {
 			p2 := &v1.Pod{ObjectMeta: origin.ObjectMeta, Spec: origin.Spec}
 			CleanupUselessInfo(p2)
 			if err = createAfterDeletePod(factory, p2, helper); err != nil {
@@ -281,7 +281,7 @@ func InjectVPNSidecar(factory cmdutil.Factory, namespace, workloads string, conf
 			log.Warnf("error while remove probe of resource: %s %s, ignore, err: %v",
 				object.Mapping.GroupVersionKind.GroupKind().String(), object.Name, err)
 		}
-		rollbackFuncList = append(rollbackFuncList, func() {
+		RollbackFuncList = append(RollbackFuncList, func() {
 			if err = removeInboundContainer(factory, namespace, workloads); err != nil {
 				log.Error(err)
 			}

pkg/handler/remote_test.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"context"
@@ -27,8 +27,8 @@ import (
 	"k8s.io/client-go/util/retry"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 //func TestCreateServer(t *testing.T) {

pkg/handler/route.go

@@ -1,4 +1,4 @@
-package pkg
+package handler
 
 import (
 	"crypto/tls"
@@ -7,9 +7,9 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/core"
+	"github.com/wencaiwulue/kubevpn/pkg/core"
-	"github.com/wencaiwulue/kubevpn/tun"
+	"github.com/wencaiwulue/kubevpn/pkg/tun"
 )
 
 type Route struct {

pkg/mesh/controller.go

@@ -7,8 +7,8 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/sets"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 func RemoveContainers(spec *v1.PodTemplateSpec) {

pkg/route/network_interface.go

@@ -1,4 +1,4 @@
-package pkg
+package route
 
 import (
 	"bytes"
@@ -14,7 +14,7 @@ import (
 // 2, detect conflict
 // 3, disable device
 func DetectAndDisableConflictDevice(origin string) error {
-	routeTable, err := getRouteTable()
+	routeTable, err := GetRouteTable()
 	if err != nil {
 		return err
 	}
@@ -22,7 +22,7 @@ func DetectAndDisableConflictDevice(origin string) error {
 	if len(conflict) != 0 {
 		log.Infof("those device: %s will to be disabled because of route conflict with %s", strings.Join(conflict, ","), origin)
 	}
-	err = disableDevice(conflict)
+	err = DisableDevice(conflict)
 	return err
 }
 

pkg/route/network_interface_test.go

@@ -1,4 +1,4 @@
-package pkg
+package route
 
 import (
 	"fmt"

pkg/route/route_table_by_net.go

@@ -1,7 +1,7 @@
 //go:build !amd64 && !arm64 && !x86 && !386
 // +build !amd64,!arm64,!x86,!386
 
-package pkg
+package route
 
 import (
 	"net"

pkg/route/route_table_darwin.go

@@ -1,7 +1,7 @@
 //go:build darwin
 // +build darwin
 
-package pkg
+package route
 
 import (
 	"net"
@@ -13,21 +13,22 @@ import (
 )
 
 // sudo ifconfig utun3 down
-func disableDevice(conflict []string) error {
+func DisableDevice(conflict []string) error {
 	for _, dev := range conflict {
 		if err := exec.Command("sudo", "ifconfig", dev, "down").Run(); err != nil {
 			log.Errorf("can not disable interface: %s, err: %v", dev, err)
 			return err
 		} else {
-			rollbackFuncList = append(rollbackFuncList, func() {
+			// todo: optimize it
-				_ = exec.Command("sudo", "ifconfig", dev, "up").Run()
+			//handler.RollbackFuncList = append(handler.RollbackFuncList, func() {
-			})
+			//	_ = exec.Command("sudo", "ifconfig", dev, "up").Run()
+			//})
 		}
 	}
 	return nil
 }
 
-func getRouteTable() (map[string][]*net.IPNet, error) {
+func GetRouteTable() (map[string][]*net.IPNet, error) {
 	output, err := exec.Command("netstat", "-anr").CombinedOutput()
 	if err != nil {
 		return nil, err

pkg/route/route_table_linux.go

@@ -1,7 +1,7 @@
 //go:build !windows && !darwin
 // +build !windows,!darwin
 
-package pkg
+package route
 
 import (
 	"net"
@@ -9,7 +9,7 @@ import (
 	"strings"
 )
 
-func getRouteTable() (map[string][]*net.IPNet, error) {
+func GetRouteTable() (map[string][]*net.IPNet, error) {
 	output, err := exec.Command("route", "-n").CombinedOutput()
 	if err != nil {
 		return nil, err
@@ -33,14 +33,15 @@ func getRouteTable() (map[string][]*net.IPNet, error) {
 	return routeTable, nil
 }
 
-func disableDevice(list []string) error {
+func DisableDevice(list []string) error {
 	for _, dev := range list {
 		if err := exec.Command("sudo", "ifconfig", dev, "down").Run(); err != nil {
 			return err
 		} else {
-			rollbackFuncList = append(rollbackFuncList, func() {
+			// todo: optimize code
-				_ = exec.Command("sudo", "ifconfig", dev, "up").Run()
+			//rollbackFuncList = append(rollbackFuncList, func() {
-			})
+			//	_ = exec.Command("sudo", "ifconfig", dev, "up").Run()
+			//})
 		}
 
 	}

pkg/route/route_table_windows.go

@@ -1,16 +1,16 @@
 //go:build windows
 // +build windows
 
-package pkg
+package route
 
 import (
 	"net"
 )
 
-func getRouteTable() (map[string][]*net.IPNet, error) {
+func GetRouteTable() (map[string][]*net.IPNet, error) {
 	return make(map[string][]*net.IPNet), nil
 }
 
-func disableDevice(list []string) error {
+func DisableDevice(list []string) error {
 	return nil
 }

pkg/test/function_test.go

@@ -25,7 +25,7 @@ import (
 	"k8s.io/client-go/util/retry"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
-	"github.com/wencaiwulue/kubevpn/util"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 var (

pkg/tun/tun_darwin.go

@@ -9,7 +9,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/songgao/water"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 func createTun(cfg Config) (conn net.Conn, itf *net.Interface, err error) {

pkg/tun/tun_linux.go

@@ -11,7 +11,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/songgao/water"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 func createTun(cfg Config) (conn net.Conn, itf *net.Interface, err error) {

pkg/tun/tun_unix.go

@@ -12,7 +12,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/songgao/water"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 func createTun(cfg Config) (conn net.Conn, itf *net.Interface, err error) {

pkg/util/networkpolicy.go

@@ -6,7 +6,7 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 // DeleteWindowsFirewallRule Delete all action block firewall rule

pkg/util/portforward.go

@@ -0,0 +1,477 @@
+package util
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/portforward"
+)
+
+// PortForwarder knows how to listen for local connections and forward them to
+// a remote pod via an upgraded HTTP request.
+type PortForwarder struct {
+	addresses     []listenAddress
+	ports         []ForwardedPort
+	stopChan      <-chan struct{}
+	innerStopChan chan struct{}
+	// if failed to find socat, send error
+	// if pod is not found, send error
+	errChan chan error
+
+	dialer        httpstream.Dialer
+	streamConn    httpstream.Connection
+	listeners     []io.Closer
+	Ready         chan struct{}
+	requestIDLock sync.Mutex
+	requestID     int
+	out           io.Writer
+	errOut        io.Writer
+}
+
+// ForwardedPort contains a Local:Remote port pairing.
+type ForwardedPort struct {
+	Local  uint16
+	Remote uint16
+}
+
+/*
+	valid port specifications:
+
+	5000
+	- forwards from localhost:5000 to pod:5000
+
+	8888:5000
+	- forwards from localhost:8888 to pod:5000
+
+	0:5000
+	:5000
+	- selects a random available local port,
+	  forwards from localhost:<random port> to pod:5000
+*/
+func parsePorts(ports []string) ([]ForwardedPort, error) {
+	var forwards []ForwardedPort
+	for _, portString := range ports {
+		parts := strings.Split(portString, ":")
+		var localString, remoteString string
+		if len(parts) == 1 {
+			localString = parts[0]
+			remoteString = parts[0]
+		} else if len(parts) == 2 {
+			localString = parts[0]
+			if localString == "" {
+				// support :5000
+				localString = "0"
+			}
+			remoteString = parts[1]
+		} else {
+			return nil, fmt.Errorf("invalid port format '%s'", portString)
+		}
+
+		localPort, err := strconv.ParseUint(localString, 10, 16)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing local port '%s': %s", localString, err)
+		}
+
+		remotePort, err := strconv.ParseUint(remoteString, 10, 16)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing remote port '%s': %s", remoteString, err)
+		}
+		if remotePort == 0 {
+			return nil, fmt.Errorf("remote port must be > 0")
+		}
+
+		forwards = append(forwards, ForwardedPort{uint16(localPort), uint16(remotePort)})
+	}
+
+	return forwards, nil
+}
+
+type listenAddress struct {
+	address     string
+	protocol    string
+	failureMode string
+}
+
+func parseAddresses(addressesToParse []string) ([]listenAddress, error) {
+	var addresses []listenAddress
+	parsed := make(map[string]listenAddress)
+	for _, address := range addressesToParse {
+		if address == "localhost" {
+			if _, exists := parsed["127.0.0.1"]; !exists {
+				ip := listenAddress{address: "127.0.0.1", protocol: "tcp4", failureMode: "all"}
+				parsed[ip.address] = ip
+			}
+			if _, exists := parsed["::1"]; !exists {
+				ip := listenAddress{address: "::1", protocol: "tcp6", failureMode: "all"}
+				parsed[ip.address] = ip
+			}
+		} else if net.ParseIP(address).To4() != nil {
+			parsed[address] = listenAddress{address: address, protocol: "tcp4", failureMode: "any"}
+		} else if net.ParseIP(address) != nil {
+			parsed[address] = listenAddress{address: address, protocol: "tcp6", failureMode: "any"}
+		} else {
+			return nil, fmt.Errorf("%s is not a valid IP", address)
+		}
+	}
+	addresses = make([]listenAddress, len(parsed))
+	id := 0
+	for _, v := range parsed {
+		addresses[id] = v
+		id++
+	}
+	// Sort addresses before returning to get a stable order
+	sort.Slice(addresses, func(i, j int) bool { return addresses[i].address < addresses[j].address })
+
+	return addresses, nil
+}
+
+// NewOnAddresses creates a new PortForwarder with custom listen addresses.
+func NewOnAddresses(dialer httpstream.Dialer, addresses []string, ports []string, stopChan <-chan struct{}, readyChan chan struct{}, out, errOut io.Writer) (*PortForwarder, error) {
+	if len(addresses) == 0 {
+		return nil, errors.New("you must specify at least 1 address")
+	}
+	parsedAddresses, err := parseAddresses(addresses)
+	if err != nil {
+		return nil, err
+	}
+	if len(ports) == 0 {
+		return nil, errors.New("you must specify at least 1 port")
+	}
+	parsedPorts, err := parsePorts(ports)
+	if err != nil {
+		return nil, err
+	}
+	return &PortForwarder{
+		dialer:        dialer,
+		addresses:     parsedAddresses,
+		ports:         parsedPorts,
+		stopChan:      stopChan,
+		innerStopChan: make(chan struct{}, 1),
+		errChan:       make(chan error, 1),
+		Ready:         readyChan,
+		out:           out,
+		errOut:        errOut,
+	}, nil
+}
+
+// ForwardPorts formats and executes a port forwarding request. The connection will remain
+// open until stopChan is closed.
+func (pf *PortForwarder) ForwardPorts() error {
+	defer pf.Close()
+
+	var err error
+	pf.streamConn, _, err = pf.dialer.Dial(portforward.PortForwardProtocolV1Name)
+	if err != nil {
+		return fmt.Errorf("error upgrading connection: %s", err)
+	}
+	defer pf.streamConn.Close()
+
+	return pf.forward()
+}
+
+// forward dials the remote host specific in req, upgrades the request, starts
+// listeners for each port specified in ports, and forwards local connections
+// to the remote host via streams.
+func (pf *PortForwarder) forward() error {
+	var err error
+
+	listenSuccess := false
+	for i := range pf.ports {
+		port := &pf.ports[i]
+		err = pf.listenOnPort(port)
+		switch {
+		case err == nil:
+			listenSuccess = true
+		default:
+			if pf.errOut != nil {
+				fmt.Fprintf(pf.errOut, "Unable to listen on port %d: %v\n", port.Local, err)
+			}
+		}
+	}
+
+	if !listenSuccess {
+		return fmt.Errorf("unable to listen on any of the requested ports: %v", pf.ports)
+	}
+
+	if pf.Ready != nil {
+		close(pf.Ready)
+	}
+
+	// wait for interrupt or conn closure
+	select {
+	case <-pf.stopChan:
+	case <-pf.innerStopChan:
+		runtime.HandleError(errors.New("lost connection to pod"))
+	}
+	select {
+	case errs, ok := <-pf.errChan:
+		if ok {
+			return errs
+		}
+		return nil
+	default:
+		return nil
+	}
+}
+
+// listenOnPort delegates listener creation and waits for connections on requested bind addresses.
+// An error is raised based on address groups (default and localhost) and their failure modes
+func (pf *PortForwarder) listenOnPort(port *ForwardedPort) error {
+	var errors []error
+	failCounters := make(map[string]int, 2)
+	successCounters := make(map[string]int, 2)
+	for _, addr := range pf.addresses {
+		err := pf.listenOnPortAndAddress(port, addr.protocol, addr.address)
+		if err != nil {
+			errors = append(errors, err)
+			failCounters[addr.failureMode]++
+		} else {
+			successCounters[addr.failureMode]++
+		}
+	}
+	if successCounters["all"] == 0 && failCounters["all"] > 0 {
+		return fmt.Errorf("%s: %v", "Listeners failed to create with the following errors", errors)
+	}
+	if failCounters["any"] > 0 {
+		return fmt.Errorf("%s: %v", "Listeners failed to create with the following errors", errors)
+	}
+	return nil
+}
+
+// listenOnPortAndAddress delegates listener creation and waits for new connections
+// in the background f
+func (pf *PortForwarder) listenOnPortAndAddress(port *ForwardedPort, protocol string, address string) error {
+	listener, err := pf.getListener(protocol, address, port)
+	if err != nil {
+		return err
+	}
+	pf.listeners = append(pf.listeners, listener)
+	go pf.waitForConnection(listener, *port)
+	return nil
+}
+
+// getListener creates a listener on the interface targeted by the given hostname on the given port with
+// the given protocol. protocol is in net.Listen style which basically admits values like tcp, tcp4, tcp6
+func (pf *PortForwarder) getListener(protocol string, hostname string, port *ForwardedPort) (net.Listener, error) {
+	listener, err := net.Listen(protocol, net.JoinHostPort(hostname, strconv.Itoa(int(port.Local))))
+	if err != nil {
+		return nil, fmt.Errorf("unable to create listener: Error %s", err)
+	}
+	listenerAddress := listener.Addr().String()
+	host, localPort, _ := net.SplitHostPort(listenerAddress)
+	localPortUInt, err := strconv.ParseUint(localPort, 10, 16)
+
+	if err != nil {
+		fmt.Fprintf(pf.out, "Failed to forward from %s:%d -> %d\n", hostname, localPortUInt, port.Remote)
+		return nil, fmt.Errorf("error parsing local port: %s from %s (%s)", err, listenerAddress, host)
+	}
+	port.Local = uint16(localPortUInt)
+	if pf.out != nil {
+		fmt.Fprintf(pf.out, "Forwarding from %s -> %d\n", net.JoinHostPort(hostname, strconv.Itoa(int(localPortUInt))), port.Remote)
+	}
+
+	return listener, nil
+}
+
+// waitForConnection waits for new connections to listener and handles them in
+// the background.
+func (pf *PortForwarder) waitForConnection(listener net.Listener, port ForwardedPort) {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			// TODO consider using something like https://github.com/hydrogen18/stoppableListener?
+			if !strings.Contains(strings.ToLower(err.Error()), "use of closed network connection") {
+				runtime.HandleError(fmt.Errorf("error accepting connection on port %d: %v", port.Local, err))
+			}
+			return
+		}
+		go pf.handleConnection(conn, port)
+	}
+}
+
+func (pf *PortForwarder) nextRequestID() int {
+	pf.requestIDLock.Lock()
+	defer pf.requestIDLock.Unlock()
+	id := pf.requestID
+	pf.requestID++
+	return id
+}
+
+// handleConnection copies data between the local connection and the stream to
+// the remote server.
+func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) {
+	defer conn.Close()
+
+	if pf.out != nil {
+		fmt.Fprintf(pf.out, "Handling connection for %d\n", port.Local)
+	}
+
+	defaultRetry := 5
+
+firstCreateStream:
+	requestID := pf.nextRequestID()
+	// create error stream
+	headers := http.Header{}
+	headers.Set(v1.StreamType, v1.StreamTypeError)
+	headers.Set(v1.PortHeader, fmt.Sprintf("%d", port.Remote))
+	headers.Set(v1.PortForwardRequestIDHeader, strconv.Itoa(requestID))
+	var err error
+	errorStream, err := pf.tryToCreateStream(&headers)
+	if err != nil {
+		runtime.HandleError(fmt.Errorf("error creating error stream for port %d -> %d: %v", port.Local, port.Remote, err))
+		return
+	}
+	// we're not writing to this stream
+	errorStream.Close()
+
+	errorChan := make(chan error)
+	go func() {
+		message, err := ioutil.ReadAll(errorStream)
+		switch {
+		case err != nil:
+			errorChan <- fmt.Errorf("error reading from error stream for port %d -> %d: %v", port.Local, port.Remote, err)
+		case len(message) > 0:
+			errorChan <- fmt.Errorf("an error occurred forwarding %d -> %d: %v", port.Local, port.Remote, string(message))
+		}
+		close(errorChan)
+	}()
+
+	// create data stream
+	headers.Set(v1.StreamType, v1.StreamTypeData)
+	dataStream, err := pf.streamConn.CreateStream(headers)
+	if err != nil {
+		defaultRetry--
+		if defaultRetry > 0 {
+			goto firstCreateStream
+		}
+		runtime.HandleError(fmt.Errorf("error creating forwarding stream for port %d -> %d: %v", port.Local, port.Remote, err))
+		return
+	}
+
+	localError := make(chan struct{})
+	remoteDone := make(chan struct{})
+
+	go func() {
+		// Copy from the remote side to the local port.
+		if _, err := io.Copy(conn, dataStream); err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			runtime.HandleError(fmt.Errorf("error copying from remote stream to local connection: %v", err))
+		}
+
+		// inform the select below that the remote copy is done
+		close(remoteDone)
+	}()
+
+	go func() {
+		// inform server we're not sending any more data after copy unblocks
+		defer dataStream.Close()
+
+		// Copy from the local port to the remote side.
+		if _, err := io.Copy(dataStream, conn); err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			runtime.HandleError(fmt.Errorf("error copying from local connection to remote stream: %v", err))
+			// break out of the select below without waiting for the other copy to finish
+			close(localError)
+		}
+	}()
+
+	// wait for either a local->remote error or for copying from remote->local to finish
+	select {
+	case <-remoteDone:
+	case <-localError:
+	}
+
+	// always expect something on errorChan (it may be nil)
+	err = <-errorChan
+	if err != nil {
+		if strings.Contains(err.Error(), "failed to find socat") {
+			select {
+			case pf.errChan <- err:
+			default:
+			}
+			close(pf.innerStopChan)
+		}
+		runtime.HandleError(err)
+	}
+}
+
+// Close stops all listeners of PortForwarder.
+func (pf *PortForwarder) Close() {
+	// stop all listeners
+	for _, l := range pf.listeners {
+		if err := l.Close(); err != nil {
+			runtime.HandleError(fmt.Errorf("error closing listener: %v", err))
+		}
+	}
+}
+
+// GetPorts will return the ports that were forwarded; this can be used to
+// retrieve the locally-bound port in cases where the input was port 0. This
+// function will signal an error if the Ready channel is nil or if the
+// listeners are not ready yet; this function will succeed after the Ready
+// channel has been closed.
+func (pf *PortForwarder) GetPorts() ([]ForwardedPort, error) {
+	if pf.Ready == nil {
+		return nil, fmt.Errorf("no Ready channel provided")
+	}
+	select {
+	case <-pf.Ready:
+		return pf.ports, nil
+	default:
+		return nil, fmt.Errorf("listeners not ready")
+	}
+}
+
+func (pf *PortForwarder) tryToCreateStream(header *http.Header) (httpstream.Stream, error) {
+	errorChan := make(chan error, 2)
+	var resultChan atomic.Value
+	time.AfterFunc(time.Second*1, func() {
+		errorChan <- errors.New("timeout")
+	})
+	go func() {
+		if pf.streamConn != nil {
+			if stream, err := pf.streamConn.CreateStream(*header); err == nil && stream != nil {
+				errorChan <- nil
+				resultChan.Store(stream)
+				return
+			}
+		}
+		errorChan <- errors.New("")
+	}()
+	if err := <-errorChan; err == nil && resultChan.Load() != nil {
+		return resultChan.Load().(httpstream.Stream), nil
+	}
+	// close old connection in case of resource leak
+	if pf.streamConn != nil {
+		_ = pf.streamConn.Close()
+	}
+	var err error
+	pf.streamConn, _, err = pf.dialer.Dial(portforward.PortForwardProtocolV1Name)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			runtime.HandleError(fmt.Errorf("pod not found: %s", err))
+			select {
+			case pf.errChan <- err:
+			default:
+			}
+			close(pf.innerStopChan)
+		} else {
+			runtime.HandleError(fmt.Errorf("error upgrading connection: %s", err))
+		}
+		return nil, err
+	}
+	header.Set(v1.PortForwardRequestIDHeader, strconv.Itoa(pf.nextRequestID()))
+	return pf.streamConn.CreateStream(*header)
+}

pkg/util/util.go

@@ -35,7 +35,6 @@ import (
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/tools/portforward"
 	"k8s.io/client-go/tools/remotecommand"
 	watchtools "k8s.io/client-go/tools/watch"
 	"k8s.io/client-go/transport/spdy"
@@ -44,7 +43,7 @@ import (
 	"k8s.io/kubectl/pkg/polymorphichelpers"
 	"k8s.io/kubectl/pkg/util/interrupt"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 func GetAvailableUDPPortOrDie() int {
@@ -97,7 +96,7 @@ func PortForwardPod(config *rest.Config, clientset *rest.RESTClient, podName, na
 	}
 	dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", url)
 	p := []string{port}
-	forwarder, err := portforward.NewOnAddresses(dialer, []string{"0.0.0.0"}, p, stopChan, readyChan, nil, os.Stderr)
+	forwarder, err := NewOnAddresses(dialer, []string{"0.0.0.0"}, p, stopChan, readyChan, nil, os.Stderr)
 	if err != nil {
 		log.Error(err)
 		return err

pkg/util/util_test.go

@@ -15,7 +15,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/kubectl/pkg/cmd/util"
 
-	"github.com/wencaiwulue/kubevpn/config"
+	"github.com/wencaiwulue/kubevpn/pkg/config"
 )
 
 var (