Added multiple clusters, routes, and refactorings.

Author: Reto Lehmann

api/endpoints.go

@@ -1,2 +1 @@
 package api
-

api/server.go

@@ -1,9 +1,9 @@
 package api
 
 import (
+	"github.com/gin-contrib/cors"
 	"github.com/gin-gonic/gin"
 	"github.com/golang/glog"
-	"github.com/gin-contrib/cors"
 )
 
 func init() {

balancer/cluster.go

@@ -0,0 +1,17 @@
+package core
+
+import "net"
+
+type Context struct {
+	Hostname string
+	Conn     net.Conn
+}
+
+type ReadWriteCount struct {
+	CountRead uint
+	CountWrite uint
+}
+
+func (rwc ReadWriteCount) IsZero() bool {
+	return rwc.CountRead == 0 && rwc.CountWrite == 0
+}

balancer/core.go

@@ -0,0 +1,74 @@
+package core
+
+import (
+	"bufio"
+	"bytes"
+	"net/http"
+)
+
+// HttpHostHeader returns the HTTP Host header from br without
+// consuming any of its bytes. It returns "" if it can't find one.
+func HttpHostHeader(br *bufio.Reader) string {
+	const maxPeek = 4 << 10
+	peekSize := 0
+	for {
+		peekSize++
+		if peekSize > maxPeek {
+			b, _ := br.Peek(br.Buffered())
+			return httpHostHeaderFromBytes(b)
+		}
+		b, err := br.Peek(peekSize)
+		if n := br.Buffered(); n > peekSize {
+			b, _ = br.Peek(n)
+			peekSize = n
+		}
+		if len(b) > 0 {
+			if b[0] < 'A' || b[0] > 'Z' {
+				// Doesn't look like an HTTP verb
+				// (GET, POST, etc).
+				return ""
+			}
+			if bytes.Index(b, crlfcrlf) != -1 || bytes.Index(b, lflf) != -1 {
+				req, err := http.ReadRequest(bufio.NewReader(bytes.NewReader(b)))
+				if err != nil {
+					return ""
+				}
+				if len(req.Header["Host"]) > 1 {
+					return ""
+				}
+				return req.Host
+			}
+		}
+		if err != nil {
+			return httpHostHeaderFromBytes(b)
+		}
+	}
+}
+
+var (
+	lfHostColon = []byte("\nHost:")
+	lfhostColon = []byte("\nhost:")
+	crlf        = []byte("\r\n")
+	lf          = []byte("\n")
+	crlfcrlf    = []byte("\r\n\r\n")
+	lflf        = []byte("\n\n")
+)
+
+func httpHostHeaderFromBytes(b []byte) string {
+	if i := bytes.Index(b, lfHostColon); i != -1 {
+		return string(bytes.TrimSpace(untilEOL(b[i+len(lfHostColon):])))
+	}
+	if i := bytes.Index(b, lfhostColon); i != -1 {
+		return string(bytes.TrimSpace(untilEOL(b[i+len(lfhostColon):])))
+	}
+	return ""
+}
+
+// untilEOL returns v, truncated before the first '\n' byte, if any.
+// The returned slice may include a '\r' at the end.
+func untilEOL(v []byte) []byte {
+	if i := bytes.IndexByte(v, '\n'); i != -1 {
+		return v[:i]
+	}
+	return v
+}

balancer/core/core.go

@@ -1,28 +1,23 @@
-package balancer
+package core
 
 import (
 	"io"
 	"net"
+
 	"time"
 
 	log "github.com/sirupsen/logrus"
 )
 
 const (
-
 	/* Buffer size to handle data from socket */
 	BUFFER_SIZE = 16 * 1024
 
-	/* Interval of pushing aggregated read/write stats */
+	/* Interval of pushing r/w stats */
 	PROXY_STATS_PUSH_INTERVAL = 1 * time.Second
 )
 
-/**
- * Perform copy/proxy data from 'from' to 'to' socket, counting r/w stats and
- * dropping connection if timeout exceeded
- */
-func proxy(to net.Conn, from net.Conn, timeout time.Duration) <-chan ReadWriteCount {
-
+func Proxy(to net.Conn, from net.Conn, timeout time.Duration) <-chan ReadWriteCount {
 	stats := make(chan ReadWriteCount)
 	outStats := make(chan ReadWriteCount)
 
@@ -74,8 +69,7 @@ func proxy(to net.Conn, from net.Conn, timeout time.Duration) <-chan ReadWriteCo
 
 	// Run proxy copier
 	go func() {
-		err := Copy(to, from, stats)
-		// hack to determine normal close. TODO: fix when it will be exposed in golang
+		err := copy(to, from, stats)
 		e, ok := err.(*net.OpError)
 		if err != nil && (!ok || e.Err.Error() != "use of closed network connection") {
 			log.Warn(err)
@@ -84,18 +78,13 @@ func proxy(to net.Conn, from net.Conn, timeout time.Duration) <-chan ReadWriteCo
 		to.Close()
 		from.Close()
 
-		// Stop stats collecting goroutine
 		close(stats)
 	}()
 
 	return outStats
 }
 
-/**
- * It's build by analogy of io.Copy
- */
-func Copy(to io.Writer, from io.Reader, ch chan<- ReadWriteCount) error {
-
+func copy(to io.Writer, from io.Reader, ch chan<- ReadWriteCount) error {
 	buf := make([]byte, BUFFER_SIZE)
 	var err error = nil
 

balancer/core/httpreader.go

@@ -30,7 +30,7 @@ func NewHealthCheck(ip string, status chan HealthCheckResult, checkInterval time
 }
 
 func (hc *HealthCheck) Start() {
-	log.Infof("Starting health checks for %v", hc.routerHostIP)
+	log.Infof("Starting health checks for router host %v", hc.routerHostIP)
 
 	hc.ticker = *time.NewTicker(hc.interval)
 
@@ -52,7 +52,7 @@ func (hc *HealthCheck) Start() {
 }
 
 func (hc *HealthCheck) Stop() {
-	log.Infof("Stopping health checks for %v", hc.routerHostIP)
+	log.Infof("Stopping health checks for router host %v", hc.routerHostIP)
 	hc.stop <- true
 }
 
@@ -67,7 +67,7 @@ func checkRouterHost(routerHostIp string, result chan<- HealthCheckResult) {
 		conn.Close()
 	}
 
-	// Tell the cluster about the health result
+	// Tell the balancer about the health result
 	result <- HealthCheckResult{
 		routerHostIP: routerHostIp,
 		healthy:      healthy,