add tcp/udp proxy feature

pkg/context/httpcontext.go

@@ -21,6 +21,7 @@ import (
 	stdcontext "context"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"runtime/debug"
 	"strings"
@@ -42,6 +43,37 @@ type (
 	// HandlerCaller is a helper function to call the handler
 	HandlerCaller func(lastResult string) string
 
+	// Layer4Context is all context of an TCP processing.
+	// It is not goroutine-safe, callers must use Lock/Unlock
+	// to protect it by themselves.
+	Layer4Context interface {
+		Lock()
+		Unlock()
+
+		stdcontext.Context
+		Cancel(err error)
+		Cancelled() bool
+		ClientDisconnected() bool
+
+		ClientConn() *net.TCPConn
+
+		Duration() time.Duration // For log, sample, etc.
+		OnFinish(func())         // For setting final client statistics, etc.
+		AddTag(tag string)       // For debug, log, etc.
+
+		Finish()
+
+		Host() string
+		SetHost(host string)
+		Port() uint16
+		SetPort(port uint16)
+
+		ClientIP() string
+
+		CallNextHandler(lastResult string) string
+		SetHandlerCaller(caller HandlerCaller)
+	}
+
 	// HTTPContext is all context of an HTTP processing.
 	// It is not goroutine-safe, callers must use Lock/Unlock
 	// to protect it by themselves.
@@ -151,6 +183,32 @@ type (
 		cancelFunc     stdcontext.CancelFunc
 		err            error
 	}
+
+	tcpContext struct {
+		mutex sync.Mutex
+
+		startTime *time.Time
+		endTime   *time.Time
+
+		tags []string
+
+		clientConn  *net.TCPConn
+		backendConn *net.TCPConn
+		stdctx      stdcontext.Context
+		err         error
+	}
+
+	udpContext struct {
+		mutex sync.Mutex
+
+		startTime *time.Time
+		endTime   *time.Time
+
+		tags []string
+
+		stdctx stdcontext.Context
+		err    error
+	}
 )
 
 // New creates an HTTPContext.
@@ -336,3 +394,7 @@ func (ctx *httpContext) SaveReqToTemplate(filterName string) error {
 func (ctx *httpContext) SaveRspToTemplate(filterName string) error {
 	return ctx.ht.SaveResponse(filterName, ctx)
 }
+
+func (ctx *tcpContext) saveBackendConn(conn *net.TCPConn) {
+	ctx.backendConn = conn
+}

pkg/filter/layer4proxy/masterslavereader.go

@@ -0,0 +1,103 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package layer4proxy
+
+import (
+	"bytes"
+	"io"
+)
+
+type (
+	// masterSlaveReader reads bytes to master,
+	// and synchronize them to slave.
+	// Currently, only support one slave.
+	masterSlaveReader struct {
+		masterReader io.Reader
+		slaveReader  io.Reader
+	}
+
+	masterReader struct {
+		r        io.Reader
+		buffChan chan []byte
+	}
+
+	slaveReader struct {
+		unreadBuff *bytes.Buffer
+		buffChan   chan []byte
+	}
+)
+
+func newMasterSlaveReader(r io.Reader) (io.ReadCloser, io.Reader) {
+	buffChan := make(chan []byte, 10)
+	mr := &masterReader{
+		r:        r,
+		buffChan: buffChan,
+	}
+	sr := &slaveReader{
+		unreadBuff: bytes.NewBuffer(nil),
+		buffChan:   buffChan,
+	}
+
+	return mr, sr
+}
+
+func (mr *masterReader) Read(p []byte) (n int, err error) {
+	buff := bytes.NewBuffer(nil)
+	tee := io.TeeReader(mr.r, buff)
+	n, err = tee.Read(p)
+
+	if n != 0 {
+		mr.buffChan <- buff.Bytes()
+	}
+
+	if err == io.EOF {
+		close(mr.buffChan)
+	}
+
+	return n, err
+}
+
+func (mr *masterReader) Close() error {
+	if closer, ok := mr.r.(io.ReadCloser); ok {
+		return closer.Close()
+	}
+
+	return nil
+}
+
+func (sr *slaveReader) Read(p []byte) (int, error) {
+	buff, ok := <-sr.buffChan
+
+	if !ok {
+		return 0, io.EOF
+	}
+
+	var n int
+	// NOTE: This if-branch is defensive programming,
+	// Because the callers of Read of both master and slave
+	// are the same, so it never happens that len(p) < len(buff).
+	// else-branch is faster because it is one less copy operation than if-branch.
+	if sr.unreadBuff.Len() > 0 || len(p) < len(buff) {
+		sr.unreadBuff.Write(buff)
+		n, _ = sr.unreadBuff.Read(p)
+	} else {
+		n = copy(p, buff)
+	}
+
+	return n, nil
+}

pkg/filter/layer4proxy/pool.go

@@ -0,0 +1,163 @@
+package layer4proxy
+
+import (
+	"bufio"
+	"fmt"
+	"github.com/google/martian/log"
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/supervisor"
+	"github.com/megaease/easegress/pkg/util/layer4stat"
+	"github.com/megaease/easegress/pkg/util/memorycache"
+	"github.com/megaease/easegress/pkg/util/stringtool"
+	"io"
+	"net"
+	"time"
+)
+
+type (
+	protocol string
+
+	pool struct {
+		spec *PoolSpec
+
+		tagPrefix     string
+		writeResponse bool
+
+		servers     *servers
+		layer4stat  *layer4stat.Layer4Stat
+		memoryCache *memorycache.MemoryCache
+	}
+
+	// PoolSpec describes a pool of servers.
+	PoolSpec struct {
+		Protocol        protocol          `yaml:"protocol" jsonschema:"required" `
+		SpanName        string            `yaml:"spanName" jsonschema:"omitempty"`
+		ServersTags     []string          `yaml:"serversTags" jsonschema:"omitempty,uniqueItems=true"`
+		Servers         []*Server         `yaml:"servers" jsonschema:"omitempty"`
+		ServiceRegistry string            `yaml:"serviceRegistry" jsonschema:"omitempty"`
+		ServiceName     string            `yaml:"serviceName" jsonschema:"omitempty"`
+		LoadBalance     *LoadBalance      `yaml:"loadBalance" jsonschema:"required"`
+		MemoryCache     *memorycache.Spec `yaml:"memoryCache,omitempty" jsonschema:"omitempty"`
+	}
+
+	// PoolStatus is the status of Pool.
+	PoolStatus struct {
+		Stat *layer4stat.Status `yaml:"stat"`
+	}
+)
+
+// Validate validates poolSpec.
+func (s PoolSpec) Validate() error {
+	if s.ServiceName == "" && len(s.Servers) == 0 {
+		return fmt.Errorf("both serviceName and servers are empty")
+	}
+
+	serversGotWeight := 0
+	for _, server := range s.Servers {
+		if server.Weight > 0 {
+			serversGotWeight++
+		}
+	}
+	if serversGotWeight > 0 && serversGotWeight < len(s.Servers) {
+		return fmt.Errorf("not all servers have weight(%d/%d)",
+			serversGotWeight, len(s.Servers))
+	}
+
+	if s.ServiceName == "" {
+		servers := newStaticServers(s.Servers, s.ServersTags, s.LoadBalance)
+		if servers.len() == 0 {
+			return fmt.Errorf("serversTags picks none of servers")
+		}
+	}
+
+	return nil
+}
+
+func newPool(super *supervisor.Supervisor, spec *PoolSpec, tagPrefix string, writeResponse bool) *pool {
+
+	var memoryCache *memorycache.MemoryCache
+	if spec.MemoryCache != nil {
+		memoryCache = memorycache.New(spec.MemoryCache)
+	}
+
+	return &pool{
+		spec: spec,
+
+		tagPrefix:     tagPrefix,
+		writeResponse: writeResponse,
+
+		servers:     newServers(super, spec),
+		layer4stat:  layer4stat.New(),
+		memoryCache: memoryCache,
+	}
+}
+
+func (p *pool) status() *PoolStatus {
+	s := &PoolStatus{Stat: p.layer4stat.Status()}
+	return s
+}
+
+func (p *pool) handle(ctx context.Layer4Context) string {
+
+	addTag := func(subPrefix, msg string) {
+		tag := stringtool.Cat(p.tagPrefix, "#", subPrefix, ": ", msg)
+		ctx.Lock()
+		ctx.AddTag(tag)
+		ctx.Unlock()
+	}
+
+	server, err := p.servers.next(ctx)
+	if err != nil {
+		addTag("serverErr", err.Error())
+		return resultInternalError
+	}
+	addTag("addr", server.HostPort)
+
+	rawConn, err := net.DialTimeout("tcp", server.HostPort, 1000*time.Millisecond)
+	if err != nil {
+		log.Errorf("dial tcp for addr: % failed, err: %v", server.HostPort, err)
+	}
+	backendConn := rawConn.(*net.TCPConn)
+
+	defer func(backendConn *net.TCPConn) {
+		closeErr := backendConn.Close()
+		if closeErr != nil {
+			logger.Warnf("close backend conn for %v failed, err: %v", server.HostPort, err)
+		}
+	}(backendConn)
+
+	errChan := make(chan error)
+	go p.connCopy(backendConn, ctx.ClientConn(), errChan)
+	go p.connCopy(ctx.ClientConn(), backendConn, errChan)
+
+	err = <-errChan
+	if err != nil {
+		logger.Errorf("Error during connection: %v", err)
+	}
+
+	err = <-errChan // TODO export tcp config for backend conn, watch client/backend error
+
+	ctx.Lock()
+	defer ctx.Unlock()
+	// NOTE: The code below can't use addTag and setStatusCode in case of deadlock.
+
+	return ""
+}
+
+func (p *pool) close() {
+	p.servers.close()
+}
+
+func (p *pool) connCopy(dst *net.TCPConn, src *net.TCPConn, errCh chan error) {
+	writer := bufio.NewWriter(dst)
+	reader := bufio.NewReader(src)
+	_, err := io.Copy(writer, reader)
+	_ = writer.Flush() // need flush bytes in buffer
+	errCh <- err
+
+	errClose := dst.CloseWrite()
+	if errClose != nil {
+		logger.Debugf("Error while terminating connection: %v", errClose)
+	}
+}