Proxy the gateway using TCP

There appeared to be an issue with logs appearing #98 and #68

@LucasRoesler spent a considerable amount of time looking into
this and concluded that the faas-provider and approach we are
taking to stream logs from journalctl as a process was
working as expected.

The issue appears to have been with the proxy code and its
use of a HTTP connection. Somewhere within the code, a buffer
was holding onto the data before flushing it 20-30 seconds later

This appeared to users as if the logs were not working at all.

Before fixing, the gateway container was tested by exposing
it over an SSH tunnel and inlets tunnel, both worked as
expected. The updates have been tested on multipass with
Ubuntu 18.04 and a binary built locally.

Signed-off-by: Alex Ellis (OpenFaaS Ltd) <alexellis2@gmail.com>

pkg/proxy.go

@@ -1,11 +1,10 @@
 package pkg
 
 import (
-	"context"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"log"
+	"net"
 	"net/http"
 
 	"time"
@@ -27,6 +26,10 @@ type Proxy struct {
 	Port    int
 }
 
+type proxyState struct {
+	Host string
+}
+
 // Start listening and forwarding HTTP to the host
 func (p *Proxy) Start(gatewayChan chan string, done chan bool) error {
 	tcp := p.Port
@@ -44,83 +47,37 @@ func (p *Proxy) Start(gatewayChan chan string, done chan bool) error {
 
 	fmt.Printf("Gateway: %s\n", ps.Host)
 
-	s := &http.Server{
+	l, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%d", tcp))
-		Addr:           fmt.Sprintf(":%d", tcp),
+	if err != nil {
-		ReadTimeout:    p.Timeout,
+		log.Printf("Error: %s", err.Error())
-		WriteTimeout:   p.Timeout,
+		return err
-		MaxHeaderBytes: 1 << 20, // Max header of 1MB
-		Handler:        http.HandlerFunc(makeProxy(&ps)),
 	}
 
-	go func() {
+	defer l.Close()
-		log.Printf("[proxy] Begin listen on %d\n", p.Port)
+
-		if err := s.ListenAndServe(); err != http.ErrServerClosed {
+	for {
-			log.Printf("Error ListenAndServe: %v", err)
+		// Wait for a connection.
+		conn, err := l.Accept()
+		if err != nil {
+			acceptErr := fmt.Errorf("Unable to accept on %d, error: %s", tcp, err.Error())
+			log.Printf("%s", acceptErr.Error())
+			return acceptErr
 		}
-	}()
 
-	log.Println("[proxy] Wait for done")
+		upstream, err := net.Dial("tcp", fmt.Sprintf("%s", ps.Host))
-	<-done
-	log.Println("[proxy] Done received")
-	if err := s.Shutdown(context.Background()); err != nil {
-		log.Printf("[proxy] Error in Shutdown: %v", err)
-	}
 
-	return nil
+		if err != nil {
-}
+			log.Printf("unable to dial to %s, error: %s", ps.Host, err.Error())
+			return err
+		}
+
+		go pipe(conn, upstream)
+		go pipe(upstream, conn)
 
-// copyHeaders clones the header values from the source into the destination.
-func copyHeaders(destination http.Header, source *http.Header) {
-	for k, v := range *source {
-		vClone := make([]string, len(v))
-		copy(vClone, v)
-		destination[k] = vClone
 	}
 }
 
-type proxyState struct {
+func pipe(from net.Conn, to net.Conn) {
-	Host string
+	defer from.Close()
-}
+	io.Copy(from, to)
-
-func makeProxy(ps *proxyState) func(w http.ResponseWriter, r *http.Request) {
-	return func(w http.ResponseWriter, r *http.Request) {
-
-		query := ""
-		if len(r.URL.RawQuery) > 0 {
-			query = "?" + r.URL.RawQuery
-		}
-
-		upstream := fmt.Sprintf("http://%s%s%s", ps.Host, r.URL.Path, query)
-		fmt.Printf("[faasd] proxy: %s\n", upstream)
-
-		if r.Body != nil {
-			defer r.Body.Close()
-		}
-
-		wrapper := ioutil.NopCloser(r.Body)
-		upReq, upErr := http.NewRequest(r.Method, upstream, wrapper)
-
-		copyHeaders(upReq.Header, &r.Header)
-
-		if upErr != nil {
-			log.Println(upErr)
-
-			http.Error(w, upErr.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		upRes, upResErr := http.DefaultClient.Do(upReq)
-
-		if upResErr != nil {
-			log.Println(upResErr)
-
-			http.Error(w, upResErr.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		copyHeaders(w.Header(), &upRes.Header)
-
-		w.WriteHeader(upRes.StatusCode)
-		io.Copy(w, upRes.Body)
-	}
 }