Merge pull request #10 from anfernee/master

Cleanup and unit test

Author: k8s-ci-robot

.gitignore

@@ -0,0 +1 @@
+/bin

cmd/agent/main.go

@@ -20,16 +20,17 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
-	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentclient"
+	"io/ioutil"
+	"net/http"
+	"os"
+
 	"github.com/golang/glog"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
-	"io/ioutil"
-	"net/http"
-	"os"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentclient"
 )
 
 func main() {

cmd/client/main.go

@@ -20,16 +20,16 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
-	"github.com/spf13/pflag"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials"
 	"io"
 	"io/ioutil"
 	"os"
 
-	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/client"
 	"github.com/golang/glog"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/client"
 )
 
 func main() {

cmd/proxy/main.go

@@ -17,23 +17,22 @@ limitations under the License.
 package main
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
-	"github.com/prometheus/client_golang/prometheus"
-	"google.golang.org/grpc/credentials"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
 
-	"google.golang.org/grpc"
-
-	"crypto/tls"
-	"crypto/x509"
-	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentserver"
-	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 	"github.com/golang/glog"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
-	"io/ioutil"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/agent/agentserver"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 )
 
 func main() {

examples/legacy/client/main.go

@@ -23,9 +23,9 @@ import (
 	"sync"
 	"sync/atomic"
 
-	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 	"github.com/golang/glog"
 	"google.golang.org/grpc"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 )
 
 type AgentClient struct {

examples/legacy/server/main.go

@@ -0,0 +1,112 @@
+package agentclient
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"google.golang.org/grpc"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
+)
+
+func TestServeData_HTTP(t *testing.T) {
+	client := NewAgentClient("")
+	client.stream = newStream()
+	stopCh := make(chan struct{})
+
+	// Start agnet
+	go client.Serve(stopCh)
+	defer close(stopCh)
+
+	// Start test http server as remote service
+	expectedBody := "Hello, client"
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, expectedBody)
+	}))
+	defer ts.Close()
+
+	// Stimulate sending KAS DIAL_REQ to AgentClient
+	dialPacket := &agent.Packet{
+		Type: agent.PacketType_DIAL_REQ,
+		Payload: &agent.Packet_DialRequest{
+			DialRequest: &agent.DialRequest{
+				Protocol: "tcp",
+				Address:  ts.URL[len("http://"):],
+				Random:   111,
+			},
+		},
+	}
+	client.stream.Send(dialPacket)
+
+	// Expect receiving DIAL_RSP packet from AgentClient
+	pkg, _ := client.stream.Recv()
+	if pkg == nil {
+		t.Error("unexpected nil packet")
+	}
+	if pkg.Type != agent.PacketType_DIAL_RSP {
+		t.Errorf("expect PacketType_DIAL_RSP; got %v", pkg.Type)
+	}
+	dialRsp := pkg.Payload.(*agent.Packet_DialResponse)
+	connID := dialRsp.DialResponse.ConnectID
+	if dialRsp.DialResponse.Random != 111 {
+		t.Errorf("expect random=111; got %v", dialRsp.DialResponse.Random)
+	}
+
+	// Send Data (HTTP Request) via AgentClient to the test http server
+	dataPacket := &agent.Packet{
+		Type: agent.PacketType_DATA,
+		Payload: &agent.Packet_Data{
+			Data: &agent.Data{
+				ConnectID: connID,
+				Data:      []byte("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"), // HTTP request
+			},
+		},
+	}
+	client.stream.Send(dataPacket)
+
+	// Expect receiving http response via AgentClient
+	pkg, _ = client.stream.Recv()
+	if pkg == nil {
+		t.Error("unexpected nil packet")
+	}
+	if pkg.Type != agent.PacketType_DATA {
+		t.Errorf("expect PacketType_DIAL_RSP; got %v", pkg.Type)
+	}
+	data := pkg.Payload.(*agent.Packet_Data).Data.Data
+
+	// Verify response data
+	//
+	// HTTP/1.1 200 OK\r\n
+	// Date: Tue, 07 May 2019 06:44:57 GMT\r\n
+	// Content-Length: 14\r\n
+	// Content-Type: text/plain; charset=utf-8\r\n
+	// \r\n
+	// Hello, client
+	headAndBody := strings.Split(string(data), "\r\n")
+	if body := headAndBody[len(headAndBody)-1]; body != expectedBody {
+		t.Errorf("expect body %v; got %v", expectedBody, body)
+	}
+}
+
+// fakeStream implements AgentService_ConnectClient
+type fakeStream struct {
+	grpc.ClientStream
+	ch chan *agent.Packet
+}
+
+func newStream() agent.AgentService_ConnectClient {
+	s := &fakeStream{}
+	s.ch = make(chan *agent.Packet)
+	return s
+}
+
+func (s *fakeStream) Send(packet *agent.Packet) error {
+	s.ch <- packet
+	return nil
+}
+
+func (s *fakeStream) Recv() (*agent.Packet, error) {
+	return <-s.ch, nil
+}

pkg/agent/agentclient/client.go

@@ -17,19 +17,18 @@ limitations under the License.
 package agentserver
 
 import (
+	"fmt"
 	"io"
+	"net"
 
-	"fmt"
-	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 	"github.com/golang/glog"
-	"net"
+	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 )
 
 // ProxyClientConnection...
 type ProxyClientConnection struct {
-	Mode string
-	Grpc agent.ProxyService_ProxyServer
-	// Http http.ResponseWriter
+	Mode      string
+	Grpc      agent.ProxyService_ProxyServer
 	Http      net.Conn
 	connected chan struct{}
 	connectID int64
@@ -74,14 +73,14 @@ func NewProxyServer() *ProxyServer {
 	}
 }
 
-// Agent ...
+// Proxy handles incoming streams from gRPC frontend.
 func (s *ProxyServer) Proxy(stream agent.ProxyService_ProxyServer) error {
 	glog.Info("proxy request from client")
 
 	recvCh := make(chan *agent.Packet, 10)
 	stopCh := make(chan error)
 
-	go s.serveRecv(stream, recvCh)
+	go s.serveRecvFrontend(stream, recvCh)
 
 	defer func() {
 		close(recvCh)
@@ -106,7 +105,7 @@ func (s *ProxyServer) Proxy(stream agent.ProxyService_ProxyServer) error {
 	return <-stopCh
 }
 
-func (s *ProxyServer) serveRecv(stream agent.ProxyService_ProxyServer, recvCh <-chan *agent.Packet) {
+func (s *ProxyServer) serveRecvFrontend(stream agent.ProxyService_ProxyServer, recvCh <-chan *agent.Packet) {
 	glog.Info("start serve recv ...")
 	for pkt := range recvCh {
 		switch pkt.Type {