add tcp/udp proxy feature

pkg/object/tcpserver/context.go

@@ -0,0 +1,15 @@
+package tcpserver
+
+import (
+ stdcontext "context"
+ "time"
+)
+
+const (
+ serverShutdownTimeout = 30 * time.Second
+)
+
+func serverShutdownContext() (stdcontext.Context, stdcontext.CancelFunc) {
+ ctx, cancelFunc := stdcontext.WithTimeout(stdcontext.Background(), serverShutdownTimeout)
+ return ctx, cancelFunc
+}

pkg/object/tcpserver/spec.go

@@ -0,0 +1,47 @@
+package tcpserver
+
+import (
+ "fmt"
+
+ "github.com/megaease/easegress/pkg/tracing"
+ "github.com/megaease/easegress/pkg/util/ipfilter"
+)
+
+type (
+ // Spec describes the TcpServer.
+ Spec struct {
+ Port uint16 `yaml:"port" jsonschema:"required,minimum=1"`
+ MaxConnections uint32 `yaml:"maxConnections" jsonschema:"omitempty,minimum=1"`
+
+ // By default, backlog is set to -1 on FreeBSD, DragonFly BSD, and macOS, and to 511 on other platforms.
+ Backlog int32 `yaml:"backlog" jsonschema:"omitempty,minimum=-1"`
+ SendBuf uint32 `yaml:"sendBuf" jsonschema:"omitempty"`
+ RecvBuf uint32 `yaml:"recvBuf" jsonschema:"omitempty"`
+ Reuseport bool `yaml:"reuseport" jsonschema:"omitempty"`
+ KeepAlive bool `yaml:"keepAlive" jsonschema:"required"`
+ TcpNodelay bool `yaml:"tcpNodelay" jsonschema:"omitempty"`
+ ProxyConnectTimeout uint32 `yaml:"proxyConnectTimeout" jsonschema:"omitempty"`
+ ProxyTimeout uint32 `yaml:"proxyTimeout" jsonschema:"omitempty"`
+
+ // Support multiple certs, preserve the certbase64 and keybase64
+ // for backward compatibility
+ SSL bool `yaml:"ssl" jsonschema:"omitempty"`
+ CertBase64 string `yaml:"certBase64" jsonschema:"omitempty,format=base64"`
+ KeyBase64 string `yaml:"keyBase64" jsonschema:"omitempty,format=base64"`
+
+ IPFilter *ipfilter.Spec `yaml:"ipFilter,omitempty" jsonschema:"omitempty"`
+
+ Tracing *tracing.Spec `yaml:"tracing" jsonschema:"omitempty"`
+ }
+)
+
+// Validate validates TcpServerSpec.
+func (spec *Spec) Validate() error {
+ if spec.SSL {
+ if spec.CertBase64 == "" || spec.KeyBase64 == "" {
+ return fmt.Errorf("tcp proxy ssl parameters is incomplete")
+ }
+ // TODO need check ssl parameters
+ }
+ return nil
+}

pkg/object/tcpserver/tcpserver.go

@@ -0,0 +1,93 @@
+package tcpserver
+
+import (
+ "github.com/megaease/easegress/pkg/protocol"
+ "sync/atomic"
+
+ "github.com/megaease/easegress/pkg/supervisor"
+ "github.com/megaease/easegress/pkg/util/limitlistener"
+ "github.com/megaease/easegress/pkg/util/tcpstat"
+)
+
+const (
+ // Category is the category of HTTPServer.
+ Category = supervisor.CategoryTrafficGate
+
+ // Kind is the kind of HTTPServer.
+ Kind = "TCPServer"
+)
+
+func init() {
+ supervisor.Register(&TCPServer{})
+}
+
+type (
+ stateType string
+
+ eventCheckFailed struct{}
+ eventServeFailed struct {
+ startNum uint64
+ err error
+ }
+ eventReload struct {
+ nextSuperSpec *supervisor.Spec
+ }
+ eventClose struct{ done chan struct{} }
+
+ TCPServer struct {
+ superSpec *supervisor.Spec
+ spec *Spec
+ startNum uint64
+ eventChan chan interface{}
+
+ // status
+ state atomic.Value // stateType
+ err atomic.Value // error
+
+ tcpstat *tcpstat.TcpStat
+ limitListener *limitlistener.LimitListener
+ }
+
+ // Status contains all status generated by runtime, for displaying to users.
+ Status struct {
+ Health string `yaml:"health"`
+
+ State stateType `yaml:"state"`
+ Error string `yaml:"error,omitempty"`
+
+ *tcpstat.Status
+ }
+)
+
+func (T *TCPServer) Category() supervisor.ObjectCategory {
+ return Category
+}
+
+func (T *TCPServer) Kind() string {
+ return Kind
+}
+
+func (T *TCPServer) DefaultSpec() interface{} {
+ return &Spec{
+ KeepAlive: true,
+ MaxConnections: 10240,
+ }
+}
+
+func (T *TCPServer) Status() *supervisor.Status {
+ panic("implement me")
+}
+
+func (T *TCPServer) Close() {
+ panic("implement me")
+}
+
+// Init initializes HTTPServer.
+func (T *TCPServer) Init(superSpec *supervisor.Spec, muxMapper protocol.MuxMapper) {
+ panic("implement me")
+}
+
+// Inherit inherits previous generation of HTTPServer.
+func (T *TCPServer) Inherit(superSpec *supervisor.Spec, previousGeneration supervisor.Object, muxMapper protocol.MuxMapper) {
+ panic("implement me")
+}

pkg/util/tcpstat/tcpstat.go

@@ -0,0 +1,73 @@
+package tcpstat
+
+import (
+ "sync"
+
+ "github.com/megaease/easegress/pkg/util/sampler"
+ "github.com/rcrowley/go-metrics"
+)
+
+type (
+ // TcpStat is the statistics tool for TCP traffic.
+ TcpStat struct {
+ mutex sync.Mutex
+
+ count uint64 // for tcp connection
+ rate1 metrics.EWMA
+ rate5 metrics.EWMA
+ rate15 metrics.EWMA
+
+ errCount uint64
+ errRate1 metrics.EWMA
+ errRate5 metrics.EWMA
+ errRate15 metrics.EWMA
+
+ m1ErrPercent float64
+ m5ErrPercent float64
+ m15ErrPercent float64
+
+ total uint64
+ min uint64
+ mean uint64
+ max uint64
+
+ durationSampler *sampler.DurationSampler
+
+ reqSize uint64
+ respSize uint64
+ }
+
+ // Status contains all status generated by HTTPStat.
+ Status struct {
+ Count uint64 `yaml:"count"`
+ M1 float64 `yaml:"m1"`
+ M5 float64 `yaml:"m5"`
+ M15 float64 `yaml:"m15"`
+
+ ErrCount uint64 `yaml:"errCount"`
+ M1Err float64 `yaml:"m1Err"`
+ M5Err float64 `yaml:"m5Err"`
+ M15Err float64 `yaml:"m15Err"`
+
+ M1ErrPercent float64 `yaml:"m1ErrPercent"`
+ M5ErrPercent float64 `yaml:"m5ErrPercent"`
+ M15ErrPercent float64 `yaml:"m15ErrPercent"`
+
+ Min uint64 `yaml:"min"`
+ Max uint64 `yaml:"max"`
+ Mean uint64 `yaml:"mean"`
+
+ P25 float64 `yaml:"p25"`
+ P50 float64 `yaml:"p50"`
+ P75 float64 `yaml:"p75"`
+ P95 float64 `yaml:"p95"`
+ P98 float64 `yaml:"p98"`
+ P99 float64 `yaml:"p99"`
+ P999 float64 `yaml:"p999"`
+
+ ReqSize uint64 `yaml:"reqSize"`
+ RespSize uint64 `yaml:"respSize"`
+
+ Codes map[int]uint64 `yaml:"codes"`
+ }
+)