Add 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

Author: anfernee

.gitignore

@@ -1 +1,7 @@
 /bin
+/certs/
+/cfssl
+/cfssljson
+/easy-rsa-master/
+/easy-rsa.tar.gz
+

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 ./...

pkg/agent/agentclient/client.go

@@ -81,10 +81,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
@@ -106,7 +112,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
@@ -130,7 +136,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)
 					}
 
@@ -140,7 +146,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
 			}
@@ -213,7 +219,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)
 			}
 		}

pkg/agent/agentclient/stream.go

@@ -3,20 +3,35 @@ 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
 
@@ -25,7 +40,7 @@ type RedialableAgentClient struct {
 	opts          []grpc.DialOption
 	conn          *grpc.ClientConn
 	stopCh        chan struct{}
-	reconnTrigger chan struct{}
+	reconnTrigger chan error
 
 	// locks
 	sendLock   sync.Mutex
@@ -48,7 +63,7 @@ func NewRedialableAgentClient(address string, opts ...grpc.DialOption) *Redialab
 		stopCh:   make(chan struct{}),
 	}
 
-	c.reconnect()
+	_ = <-c.triggerReconnect()
 
 	go c.probe()
 
@@ -69,7 +84,10 @@ func (c *RedialableAgentClient) probe() {
 			}
 		}
 
-		c.reconnect()
+		klog.Info("probe failure: reconnect")
+		if err := <-c.triggerReconnect(); err != nil {
+			klog.Infof("probe reconnect failed: %v", err)
+		}
 	}
 }
 
@@ -78,46 +96,75 @@ func (c *RedialableAgentClient) Send(pkt *agent.Packet) error {
 	defer c.sendLock.Unlock()
 
 	if err := c.stream.Send(pkt); err != nil {
-		if c.conn.GetState() == connectivity.Ready {
+		if err == io.EOF {
 			return err
 		}
-
-		if err2 := c.reconnect(); err2 != nil {
-			return err2
+		return &ReconnectError{
+			internalErr: err,
+			errChan:     c.triggerReconnect(),
 		}
 	}
 
-	return c.stream.Send(pkt)
+	return nil
 }
 
-func (c *RedialableAgentClient) Recv() (*agent.Packet, error) {
-	c.recvLock.Lock()
-	defer c.recvLock.Unlock()
+func (c *RedialableAgentClient) RetrySend(pkt *agent.Packet) error {
+	err := c.Send(pkt)
+	if err == nil {
+		return nil
+	} else if err == io.EOF {
+		return err
+	}
 
-	// this just get block..
-	if pkt, err := c.stream.Recv(); err != nil {
-		klog.Infof("error recving: %v", err)
-		klog.Info("start reconnecting")
+	if err2, ok := err.(*ReconnectError); ok {
+		err = err2.Wait()
+	}
+	if err != nil {
+		return err
+	}
+	return c.RetrySend(pkt)
+}
 
-		if err2 := c.reconnect(); err2 != nil {
-			klog.Infof("reconnect failed: %v", err2)
-			return pkt, err2
-		}
+func (c *RedialableAgentClient) triggerReconnect() <-chan error {
+	c.reconnLock.Lock()
+	defer c.reconnLock.Unlock()
 
-		return c.stream.Recv()
-	} else {
-		return pkt, err
+	if c.reconnTrigger != nil {
+		return c.reconnTrigger
 	}
+
+	c.reconnTrigger = make(chan error)
+	go c.reconnect()
+
+	return c.reconnTrigger
 }
 
-func (c *RedialableAgentClient) reconnect() error {
+func (c *RedialableAgentClient) doneReconnect(err error) {
 	c.reconnLock.Lock()
 	defer c.reconnLock.Unlock()
 
-	if c.conn != nil && c.conn.GetState() == connectivity.Ready {
-		return nil
+	c.reconnTrigger <- err
+	c.reconnTrigger = 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) reconnect() {
 	klog.Info("start to connect...")
 
 	var err error
@@ -126,14 +173,16 @@ func (c *RedialableAgentClient) reconnect() error {
 	for retry < c.Retry {
 		if err = c.tryConnect(); err == nil {
 			klog.Info("connected")
-			return nil
+			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)
 	}
 
-	return fmt.Errorf("Failed to connect to proxy server: %v", err)
+	c.doneReconnect(fmt.Errorf("Failed to connect to proxy server: %v", err))
 }
 
 func (c *RedialableAgentClient) tryConnect() error {

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),

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