Repository: kubernetes-sigs/apiserver-network-proxy (go)

Issue:
Make sure proxy-agent reconnects to the proxy-server.
Make sure we handle cases like the proxy-agent coming up before the proxy-server. Also it should reconnect if the connection is broken. We should also have tests for this.

Pull request description:
Agent supports reconnect
When the dialback is interrupted, the agent connects back to the proxy
server and continue to serve the proxy requests.

Fixes #21

The change also adds integration test for basic proxy functionality
    - The test covers both grpc and http-connect
    - The remote server is http based, so it won't send traffic volunteerly
      unless there is a request
    - The test traffic is not over TLS



Patch:
diff --git a/.gitignore b/.gitignore
index 5e56e040..6580405c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,1 +1,7 @@
 /bin
+/certs/
+/cfssl
+/cfssljson
+/easy-rsa-master/
+/easy-rsa.tar.gz
+
diff --git a/Makefile b/Makefile
index 204b27b2..a6ac2989 100644
--- a/Makefile
+++ b/Makefile
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-.PHONY: gen clean certs build docker/proxy-server docker/proxy-agent push-images
+.PHONY: gen clean certs build docker/proxy-server docker/proxy-agent push-images test
 proto/agent/agent.pb.go: proto/agent/agent.proto
 	protoc -I proto proto/agent/agent.proto --go_out=plugins=grpc:proto
 
@@ -104,3 +104,6 @@ push-images: docker/proxy-agent docker/proxy-server
 
 clean:
 	rm -rf proto/agent/agent.pb.go proto/proxy.pb.go easy-rsa.tar.gz easy-rsa-master cfssl cfssljson certs bin
+
+test:
+	go test ./...
diff --git a/cmd/agent/main.go b/cmd/agent/main.go
index c3c96a1a..5d6fe3d1 100644
--- a/cmd/agent/main.go
+++ b/cmd/agent/main.go
@@ -30,9 +30,9 @@ import (
 	"github.com/spf13/pflag"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
+	"k8s.io/klog"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentclient"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/util"
-	"k8s.io/klog"
 )
 
 func main() {
@@ -113,9 +113,9 @@ func (o *GrpcProxyAgentOptions) Validate() error {
 
 func newGrpcProxyAgentOptions() *GrpcProxyAgentOptions {
 	o := GrpcProxyAgentOptions{
-		agentCert: "",
-		agentKey:  "",
-		caCert:    "",
+		agentCert:       "",
+		agentKey:        "",
+		caCert:          "",
 		proxyServerHost: "127.0.0.1",
 		proxyServerPort: 8091,
 	}
@@ -178,10 +178,9 @@ func (p *Agent) runProxyConnection(o *GrpcProxyAgentOptions) error {
 		RootCAs:      certPool,
 	})
 	dialOption := grpc.WithTransportCredentials(transportCreds)
-	client := agentclient.NewAgentClient(fmt.Sprintf("%s:%d", o.proxyServerHost, o.proxyServerPort))
-
-	if err := client.Connect(dialOption); err != nil {
-		return fmt.Errorf("failed to connect to proxy-server: %v", err)
+	client, err := agentclient.NewAgentClient(fmt.Sprintf("%s:%d", o.proxyServerHost, o.proxyServerPort), dialOption)
+	if err != nil {
+		return err
 	}
 
 	stopCh := make(chan struct{})
diff --git a/pkg/agent/agentclient/client.go b/pkg/agent/agentclient/client.go
index 2873fbe6..37bff045 100644
--- a/pkg/agent/agentclient/client.go
+++ b/pkg/agent/agentclient/client.go
@@ -17,7 +17,6 @@ limitations under the License.
 package agentclient
 
 import (
-	"context"
 	"io"
 	"net"
 	"sync"
@@ -33,19 +32,22 @@ import (
 type AgentClient struct {
 	nextConnID  int64
 	connContext map[int64]*connContext
-	address     string
 
-	stream agent.AgentService_ConnectClient
+	stream *RedialableAgentClient
 }
 
 // NewAgentClient creates an AgentClient
-func NewAgentClient(address string) *AgentClient {
+func NewAgentClient(address string, opts ...grpc.DialOption) (*AgentClient, error) {
+	stream, err := NewRedialableAgentClient(address, opts...)
+	if err != nil {
+		return nil, err
+	}
+
 	a := &AgentClient{
 		connContext: make(map[int64]*connContext),
-		address:     address,
+		stream:      stream,
 	}
-
-	return a
+	return a, nil
 }
 
 // connContext tracks a connection from agent to node network.
@@ -68,21 +70,6 @@ func (c *connContext) cleanup() {
 //
 // The caller needs to call Serve to start serving proxy requests
 // coming from proxy server.
-func (a *AgentClient) Connect(opts ...grpc.DialOption) error {
-	c, err := grpc.Dial(a.address, opts...)
-	if err != nil {
-		return err
-	}
-
-	client := agent.NewAgentServiceClient(c)
-
-	a.stream, err = client.Connect(context.Background())
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
 
 // Serve starts to serve proxied requests from proxy server over the
 // gRPC stream. Successful Connect is required before Serve. The
@@ -98,10 +85,16 @@ func (a *AgentClient) Serve(stopCh <-chan struct{}) {
 		}
 
 		pkt, err := a.stream.Recv()
-		if err == io.EOF {
-			klog.Info("received EOF, exit")
-			return
+		if err != nil {
+			if err2, ok := err.(*ReconnectError); ok {
+				err = err2.Wait()
+				continue
+			} else if err == io.EOF {
+				klog.Info("received EOF, exit")
+				return
+			}
 		}
+
 		if err != nil {
 			klog.Warningf("stream read error: %v", err)
 			return
@@ -123,7 +116,7 @@ func (a *AgentClient) Serve(stopCh <-chan struct{}) {
 			conn, err := net.Dial(dialReq.Protocol, dialReq.Address)
 			if err != nil {
 				resp.GetDialResponse().Error = err.Error()
-				if err := a.stream.Send(resp); err != nil {
+				if err := a.stream.RetrySend(resp); err != nil {
 					klog.Warningf("stream send error: %v", err)
 				}
 				continue
@@ -147,7 +140,7 @@ func (a *AgentClient) Serve(stopCh <-chan struct{}) {
 						resp.GetCloseResponse().Error = err.Error()
 					}
 
-					if err := a.stream.Send(resp); err != nil {
+					if err := a.stream.RetrySend(resp); err != nil {
 						klog.Warningf("close response send error: %v", err)
 					}
 
@@ -157,7 +150,7 @@ func (a *AgentClient) Serve(stopCh <-chan struct{}) {
 			}
 
 			resp.GetDialResponse().ConnectID = connID
-			if err := a.stream.Send(resp); err != nil {
+			if err := a.stream.RetrySend(resp); err != nil {
 				klog.Warningf("stream send error: %v", err)
 				continue
 			}
@@ -230,7 +223,7 @@ func (a *AgentClient) remoteToProxy(conn net.Conn, connID int64) {
 				Data:      buf[:n],
 				ConnectID: connID,
 			}}
-			if err := a.stream.Send(resp); err != nil {
+			if err := a.stream.RetrySend(resp); err != nil {
 				klog.Warningf("stream send error: %v", err)
 			}
 		}
diff --git a/pkg/agent/agentclient/stream.go b/pkg/agent/agentclient/stream.go
new file mode 100644
index 00000000..743827b2
--- /dev/null
+++ b/pkg/agent/agentclient/stream.go
@@ -0,0 +1,213 @@
+package agentclient
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"k8s.io/klog"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
+	// "k8s.io/client-go/util/retry"
+)
+
+const (
+	defaultRetry    = 20
+	defaultInterval = 5 * time.Second
+)
+
+type ReconnectError struct {
+	internalErr error
+	errChan     <-chan error
+}
+
+func (e *ReconnectError) Error() string {
+	return "transient error: " + e.internalErr.Error()
+}
+
+func (e *ReconnectError) Wait() error {
+	return <-e.errChan
+}
+
+type RedialableAgentClient struct {
+	stream agent.AgentService_ConnectClient
+
+	// connect opts
+	address       string
+	opts          []grpc.DialOption
+	conn          *grpc.ClientConn
+	stopCh        chan struct{}
+	reconnOngoing bool
+	reconnWaiters []chan error
+
+	// locks
+	sendLock   sync.Mutex
+	recvLock   sync.Mutex
+	reconnLock sync.Mutex
+
+	// Retry times to reconnect to proxy server
+	Retry int
+
+	// Interval between every reconnect
+	Interval time.Duration
+}
+
+func NewRedialableAgentClient(address string, opts ...grpc.DialOption) (*RedialableAgentClient, error) {
+	c := &RedialableAgentClient{
+		address:  address,
+		opts:     opts,
+		Retry:    defaultRetry,
+		Interval: defaultInterval,
+		stopCh:   make(chan struct{}),
+	}
+
+	return c, c.Connect()
+}
+
+func (c *RedialableAgentClient) probe() {
+	for {
+		select {
+		case <-c.stopCh:
+			return
+		case <-time.After(c.Interval):
+			// health check
+			if c.conn != nil && c.conn.GetState() == connectivity.Ready {
+				continue
+			} else {
+				klog.Infof("Connection state %v", c.conn.GetState())
+			}
+		}
+
+		klog.Info("probe failure: reconnect")
+		if err := <-c.triggerReconnect(); err != nil {
+			klog.Infof("probe reconnect failed: %v", err)
+		}
+	}
+}
+
+func (c *RedialableAgentClient) Send(pkt *agent.Packet) error {
+	c.sendLock.Lock()
+	defer c.sendLock.Unlock()
+
+	if err := c.stream.Send(pkt); err != nil {
+		if err == io.EOF {
+			return err
+		}
+		return &ReconnectError{
+			internalErr: err,
+			errChan:     c.triggerReconnect(),
+		}
+	}
+
+	return nil
+}
+
+func (c *RedialableAgentClient) RetrySend(pkt *agent.Packet) error {
+	err := c.Send(pkt)
+	if err == nil {
+		return nil
+	} else if err == io.EOF {
+		return err
+	}
+
+	if err2, ok := err.(*ReconnectError); ok {
+		err = err2.Wait()
+	}
+	if err != nil {
+		return err
+	}
+	return c.RetrySend(pkt)
+}
+
+func (c *RedialableAgentClient) triggerReconnect() <-chan error {
+	c.reconnLock.Lock()
+	defer c.reconnLock.Unlock()
+
+	errch := make(chan error)
+	c.reconnWaiters = append(c.reconnWaiters, errch)
+
+	if !c.reconnOngoing {
+		go c.reconnect()
+		c.reconnOngoing = true
+	}
+
+	return errch
+}
+
+func (c *RedialableAgentClient) doneReconnect(err error) {
+	c.reconnLock.Lock()
+	defer c.reconnLock.Unlock()
+
+	for _, ch := range c.reconnWaiters {
+		ch <- err
+	}
+	c.reconnOngoing = false
+	c.reconnWaiters = nil
+}
+
+func (c *RedialableAgentClient) Recv() (*agent.Packet, error) {
+	c.recvLock.Lock()
+	defer c.recvLock.Unlock()
+
+	if pkt, err := c.stream.Recv(); err != nil {
+		if err == io.EOF {
+			return pkt, err
+		}
+		return pkt, &ReconnectError{
+			internalErr: err,
+			errChan:     c.triggerReconnect(),
+		}
+	} else {
+		return pkt, nil
+	}
+}
+
+func (c *RedialableAgentClient) Connect() error {
+	if err := c.tryConnect(); err != nil {
+		return err
+	}
+
+	go c.probe()
+	return nil
+}
+
+func (c *RedialableAgentClient) reconnect() {
+	klog.Info("start to connect...")
+
+	var err error
+	var retry int
+
+	for retry < c.Retry {
+		if err = c.tryConnect(); err == nil {
+			klog.Info("connected")
+			c.doneReconnect(nil)
+			return
+		}
+		retry++
+		klog.V(5).Infof("Failed to connect to proxy server, retry %d in %v: %v", retry, c.Interval, err)
+		klog.Infof("Failed to connect to proxy server, retry %d in %v: %v", retry, c.Interval, err)
+		time.Sleep(c.Interval)
+	}
+
+	c.doneReconnect(fmt.Errorf("Failed to connect to proxy server: %v", err))
+}
+
+func (c *RedialableAgentClient) tryConnect() error {
+	var err error
+
+	c.conn, err = grpc.Dial(c.address, c.opts...)
+	if err != nil {
+		return err
+	}
+
+	c.stream, err = agent.NewAgentServiceClient(c.conn).Connect(context.Background())
+	return err
+}
+
+// interrupt interrupt the stream connection. (For testing purpose)
+func (c *RedialableAgentClient) interrupt() {
+	c.conn.Close()
+}
diff --git a/pkg/agent/agentserver/server.go b/pkg/agent/agentserver/server.go
index 6c785ea9..9fa9b4d0 100644
--- a/pkg/agent/agentserver/server.go
+++ b/pkg/agent/agentserver/server.go
@@ -54,7 +54,7 @@ func (c *ProxyClientConnection) send(pkt *agent.Packet) error {
 	}
 }
 
-// ProxyServer ...
+// ProxyServer 
 type ProxyServer struct {
 	Backend agent.AgentService_ConnectServer
 
@@ -67,7 +67,7 @@ var _ agent.AgentServiceServer = &ProxyServer{}
 
 var _ agent.ProxyServiceServer = &ProxyServer{}
 
-// NewProxyServer ...
+// NewProxyServer creates a new ProxyServer instance
 func NewProxyServer() *ProxyServer {
 	return &ProxyServer{
 		Frontends:   make(map[int64]*ProxyClientConnection),
diff --git a/pkg/agent/agentserver/tunnel.go b/pkg/agent/agentserver/tunnel.go
index 7953ddbb..3ba94cb1 100644
--- a/pkg/agent/agentserver/tunnel.go
+++ b/pkg/agent/agentserver/tunnel.go
@@ -32,10 +32,10 @@ type Tunnel struct {
 }
 
 func (t *Tunnel) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	klog.Infof("Received %s request to %q from %v",
-		r.Method,
-		r.Host,
-		r.TLS.PeerCertificates[0].Subject.CommonName) // can do authz with certs
+	klog.Infof("Received %s request to %q", r.Method, r.Host)
+	if r.TLS != nil {
+		klog.Infof("TLS CommonName: %v", r.TLS.PeerCertificates[0].Subject.CommonName)
+	}
 	if r.Method != http.MethodConnect {
 		http.Error(w, "this proxy only supports CONNECT passthrough", http.StatusMethodNotAllowed)
 		return


Test patch:
diff --git a/pkg/agent/agentclient/client_test.go b/pkg/agent/agentclient/client_test.go
index 54a8b593..4641223a 100644
--- a/pkg/agent/agentclient/client_test.go
+++ b/pkg/agent/agentclient/client_test.go
@@ -16,8 +16,10 @@ import (
 
 func TestServeData_HTTP(t *testing.T) {
 	var stream agent.AgentService_ConnectClient
-	client := NewAgentClient("")
-	client.stream, stream = pipe()
+	client := &AgentClient{
+		connContext: make(map[int64]*connContext),
+	}
+	client.stream, stream = pipe2()
 	stopCh := make(chan struct{})
 
 	// Start agent
@@ -100,8 +102,10 @@ func TestServeData_HTTP(t *testing.T) {
 
 func TestClose_Client(t *testing.T) {
 	var stream agent.AgentService_ConnectClient
-	client := NewAgentClient("")
-	client.stream, stream = pipe()
+	client := &AgentClient{
+		connContext: make(map[int64]*connContext),
+	}
+	client.stream, stream = pipe2()
 	stopCh := make(chan struct{})
 
 	// Start agent
@@ -193,6 +197,11 @@ func pipe() (agent.AgentService_ConnectClient, agent.AgentService_ConnectClient)
 	return s1, s2
 }
 
+func pipe2() (*RedialableAgentClient, agent.AgentService_ConnectClient) {
+	s1, s2 := pipe()
+	return &RedialableAgentClient{stream: s1}, s2
+}
+
 func (s *fakeStream) Send(packet *agent.Packet) error {
 	klog.Infof("[DEBUG] send packet %+v", packet)
 	s.w <- packet
diff --git a/pkg/agent/agentclient/stream_test.go b/pkg/agent/agentclient/stream_test.go
new file mode 100644
index 00000000..79e999f9
--- /dev/null
+++ b/pkg/agent/agentclient/stream_test.go
@@ -0,0 +1,124 @@
+package agentclient
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"k8s.io/klog"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
+)
+
+func TestReconnectExits(t *testing.T) {
+	server := newTestServer("localhost:8899") // random addr
+	server.Start()
+	defer server.Stop()
+
+	time.Sleep(time.Millisecond)
+
+	client, err := NewRedialableAgentClient("localhost:8899", grpc.WithInsecure())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = client.Send(&agent.Packet{
+		Type: agent.PacketType_DIAL_REQ,
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	client1 := make(chan bool)
+	go func() {
+		_, err := client.Recv()
+		if err != nil {
+			if err2, ok := err.(*ReconnectError); ok {
+				err2.Wait()
+				client1 <- true
+			}
+		}
+	}()
+
+	client2 := make(chan bool)
+	go func() {
+		_, err := client.Recv()
+		if err != nil {
+			if err2, ok := err.(*ReconnectError); ok {
+				err2.Wait()
+				client2 <- true
+			}
+		}
+	}()
+
+	client.interrupt()
+
+	var got1 bool
+	var got2 bool
+	select {
+	case got1 = <-client1:
+	case <-time.After(time.Second):
+	}
+	select {
+	case got2 = <-client2:
+	case <-time.After(time.Second):
+	}
+
+	if !got1 || !got2 {
+		t.Errorf("expect both clients get unblocked; not they don't (%t %t)", got1, got2)
+	}
+}
+
+type testServer struct {
+	addr       string
+	grpcServer *grpc.Server
+}
+
+func newTestServer(addr string) *testServer {
+	return &testServer{addr: addr}
+}
+
+func (s *testServer) Connect(stream agent.AgentService_ConnectServer) error {
+	stopCh := make(chan error)
+
+	// Recv only
+	go func() {
+		for {
+			_, err := stream.Recv()
+			if err == io.EOF {
+				close(stopCh)
+				return
+			}
+			if err != nil {
+				klog.Warningf(">>> Stream read from frontend error: %v", err)
+				close(stopCh)
+				return
+			}
+		}
+	}()
+
+	return <-stopCh
+}
+
+func (s *testServer) Start() error {
+	s.grpcServer = grpc.NewServer()
+	agent.RegisterAgentServiceServer(s.grpcServer, s)
+	lis, err := net.Listen("tcp", s.addr)
+	if err != nil {
+		return fmt.Errorf("failed to listen on %s: %v", s.addr, err)
+	}
+	go s.grpcServer.Serve(lis)
+	return nil
+}
+
+func (s *testServer) Stop() {
+	if s.grpcServer != nil {
+		s.grpcServer.Stop()
+	}
+}
+
+func (s *testServer) Addr() string {
+	return s.addr
+}
diff --git a/tests/proxy_test.go b/tests/proxy_test.go
new file mode 100644
index 00000000..2dd2e5a6
--- /dev/null
+++ b/tests/proxy_test.go
@@ -0,0 +1,242 @@
+package tests
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentclient"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentserver"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/client"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
+)
+
+// test remote server
+type testServer struct {
+}
+
+func (s *testServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	w.Write([]byte("hello"))
+}
+
+func TestBasicProxy_GRPC(t *testing.T) {
+	server := httptest.NewServer(&testServer{})
+	defer server.Close()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	proxy, cleanup, err := runGRPCProxyServer()
+	defer cleanup()
+
+	if err := runAgent(proxy.agent, stopCh); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for agent to register on proxy server
+	time.Sleep(time.Second)
+
+	// run test client
+	tunnel, err := client.CreateGrpcTunnel(proxy.front, grpc.WithInsecure())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	c := &http.Client{
+		Transport: &http.Transport{
+			Dial: tunnel.Dial,
+		},
+	}
+
+	r, err := c.Get(server.URL)
+	if err != nil {
+		t.Error(err)
+	}
+
+	data, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if string(data) != "hello" {
+		t.Errorf("expect %v; got %v", "hello", string(data))
+	}
+}
+
+func TestBasicProxy_HTTPCONN(t *testing.T) {
+	server := httptest.NewServer(&testServer{})
+	defer server.Close()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	proxy, cleanup, err := runHTTPConnProxyServer()
+	defer cleanup()
+
+	if err := runAgent(proxy.agent, stopCh); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for agent to register on proxy server
+	time.Sleep(time.Second)
+
+	conn, err := net.Dial("tcp", proxy.front)
+	if err != nil {
+		t.Error(err)
+	}
+
+	serverURL, _ := url.Parse(server.URL)
+
+	// Send HTTP-Connect request
+	_, err = fmt.Fprintf(conn, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", serverURL.Host, "127.0.0.1")
+	if err != nil {
+		t.Error(err)
+	}
+
+	// Parse the HTTP response for Connect
+	br := bufio.NewReader(conn)
+	res, err := http.ReadResponse(br, nil)
+	if err != nil {
+		t.Errorf("reading HTTP response from CONNECT: %v", err)
+	}
+	if res.StatusCode != 200 {
+		t.Errorf("expect 200; got %d", res.StatusCode)
+	}
+	if br.Buffered() > 0 {
+		t.Error("unexpected extra buffer")
+	}
+
+	dialer := func(network, addr string) (net.Conn, error) {
+		return conn, nil
+	}
+
+	c := &http.Client{
+		Transport: &http.Transport{
+			Dial: dialer,
+		},
+	}
+
+	r, err := c.Get(server.URL)
+	if err != nil {
+		t.Error(err)
+	}
+
+	data, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if string(data) != "hello" {
+		t.Errorf("expect %v; got %v", "hello", string(data))
+	}
+
+}
+
+func localAddr(addr net.Addr) string {
+	return addr.String()
+}
+
+type proxy struct {
+	front string
+	agent string
+}
+
+func runGRPCProxyServer() (proxy, func(), error) {
+	var proxy proxy
+	var err error
+	var lis, lis2 net.Listener
+
+	server := agentserver.NewProxyServer()
+	grpcServer := grpc.NewServer()
+	agentServer := grpc.NewServer()
+	cleanup := func() {
+		if lis != nil {
+			lis.Close()
+		}
+		if lis2 != nil {
+			lis2.Close()
+		}
+	}
+
+	agent.RegisterProxyServiceServer(grpcServer, server)
+	lis, err = net.Listen("tcp", "")
+	if err != nil {
+		return proxy, cleanup, err
+	}
+	go grpcServer.Serve(lis)
+	proxy.front = localAddr(lis.Addr())
+
+	agent.RegisterAgentServiceServer(agentServer, server)
+	lis2, err = net.Listen("tcp", "")
+	if err != nil {
+		return proxy, cleanup, err
+	}
+	go func() {
+		agentServer.Serve(lis2)
+	}()
+	proxy.agent = localAddr(lis2.Addr())
+
+	return proxy, cleanup, nil
+}
+
+func runHTTPConnProxyServer() (proxy, func(), error) {
+	var proxy proxy
+	server := agentserver.NewProxyServer()
+	agentServer := grpc.NewServer()
+
+	agent.RegisterAgentServiceServer(agentServer, server)
+	lis, err := net.Listen("tcp", "")
+	if err != nil {
+		return proxy, func() {}, err
+	}
+	go func() {
+		agentServer.Serve(lis)
+	}()
+	proxy.agent = localAddr(lis.Addr())
+
+	// http-connect
+	httpServer := &http.Server{
+		Handler: &agentserver.Tunnel{
+			Server: server,
+		},
+	}
+	lis2, err := net.Listen("tcp", "")
+	if err != nil {
+		return proxy, func() {}, err
+	}
+	proxy.front = localAddr(lis2.Addr())
+
+	go func() {
+		err := httpServer.Serve(lis2)
+		if err != nil {
+			fmt.Println("http connect server error: ", err)
+		}
+	}()
+
+	cleanup := func() {
+		lis.Close()
+		lis2.Close()
+		httpServer.Shutdown(context.Background())
+	}
+
+	return proxy, cleanup, nil
+}
+
+func runAgent(addr string, stopCh <-chan struct{}) error {
+	client, err := agentclient.NewAgentClient(addr, grpc.WithInsecure())
+	if err != nil {
+		return err
+	}
+
+	go client.Serve(stopCh)
+
+	return nil
+}