From c33b5152e722a50d65cd6925ba498f3016c7c19c Mon Sep 17 00:00:00 2001 From: fatedier Date: Mon, 25 Jun 2018 18:22:35 +0800 Subject: [PATCH 01/10] split visitors from proxies and add health check config --- client/admin_api.go | 2 +- client/control.go | 17 ++- client/health.go | 32 +++++ client/proxy_manager.go | 87 +++--------- client/service.go | 2 +- client/visitor.go | 14 +- client/visitor_manager.go | 111 +++++++++++++++ cmd/frpc/sub/root.go | 4 +- cmd/frpc/sub/stcp.go | 75 +++++----- cmd/frpc/sub/xtcp.go | 75 +++++----- conf/frpc_full.ini | 9 ++ models/config/proxy.go | 287 ++++++++++++++++++++------------------ models/config/visitor.go | 213 ++++++++++++++++++++++++++++ 13 files changed, 641 insertions(+), 287 deletions(-) create mode 100644 client/health.go create mode 100644 client/visitor_manager.go create mode 100644 models/config/visitor.go diff --git a/client/admin_api.go b/client/admin_api.go index 854e970..4eafa10 100644 --- a/client/admin_api.go +++ b/client/admin_api.go @@ -77,7 +77,7 @@ func (svr *Service) apiReload(w http.ResponseWriter, r *http.Request) { return } - pxyCfgs, visitorCfgs, err := config.LoadProxyConfFromIni(g.GlbClientCfg.User, conf, newCommonCfg.Start) + pxyCfgs, visitorCfgs, err := config.LoadAllConfFromIni(g.GlbClientCfg.User, conf, newCommonCfg.Start) if err != nil { res.Code = 3 res.Msg = err.Error() diff --git a/client/control.go b/client/control.go index 53669ae..04be13c 100644 --- a/client/control.go +++ b/client/control.go @@ -47,8 +47,12 @@ type Control struct { // login message to server, only used loginMsg *msg.Login + // manage all proxies pm *ProxyManager + // manage all visitors + vm *VisitorManager + // control connection conn frpNet.Conn @@ -82,7 +86,7 @@ type Control struct { log.Logger } -func NewControl(svr *Service, pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.ProxyConf) *Control { +func NewControl(svr *Service, pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) *Control { loginMsg := &msg.Login{ Arch: runtime.GOARCH, Os: runtime.GOOS, @@ -102,7 +106,9 @@ func NewControl(svr *Service, pxyCfgs map[string]config.ProxyConf, visitorCfgs m Logger: log.NewPrefixLogger(""), } ctl.pm = NewProxyManager(ctl, ctl.sendCh, "") - ctl.pm.Reload(pxyCfgs, visitorCfgs, false) + ctl.pm.Reload(pxyCfgs, false) + ctl.vm = NewVisitorManager(ctl) + ctl.vm.Reload(visitorCfgs) return ctl } @@ -129,6 +135,8 @@ func (ctl *Control) Run() (err error) { // start all local visitors and send NewProxy message for all configured proxies ctl.pm.Reset(ctl.sendCh, ctl.runId) ctl.pm.CheckAndStartProxy([]string{ProxyStatusNew}) + + go ctl.vm.Run() return nil } @@ -444,7 +452,8 @@ func (ctl *Control) worker() { } } -func (ctl *Control) reloadConf(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.ProxyConf) error { - err := ctl.pm.Reload(pxyCfgs, visitorCfgs, true) +func (ctl *Control) reloadConf(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) error { + ctl.vm.Reload(visitorCfgs) + err := ctl.pm.Reload(pxyCfgs, true) return err } diff --git a/client/health.go b/client/health.go new file mode 100644 index 0000000..ad58554 --- /dev/null +++ b/client/health.go @@ -0,0 +1,32 @@ +// Copyright 2018 fatedier, fatedier@gmail.com +// +// 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 client + +import ( + "github.com/fatedier/frp/models/config" +) + +type HealthCheckMonitor struct { + cfg config.HealthCheckConf +} + +func NewHealthCheckMonitor(cfg *config.HealthCheckConf) *HealthCheckMonitor { + return &HealthCheckMonitor{ + cfg: *cfg, + } +} + +func (monitor *HealthCheckMonitor) Start() { +} diff --git a/client/proxy_manager.go b/client/proxy_manager.go index 6702972..cfa56fc 100644 --- a/client/proxy_manager.go +++ b/client/proxy_manager.go @@ -13,25 +13,21 @@ import ( ) const ( - ProxyStatusNew = "new" - ProxyStatusStartErr = "start error" - ProxyStatusWaitStart = "wait start" - ProxyStatusRunning = "running" - ProxyStatusClosed = "closed" + ProxyStatusNew = "new" + ProxyStatusStartErr = "start error" + ProxyStatusWaitStart = "wait start" + ProxyStatusRunning = "running" + ProxyStatusCheckFailed = "check failed" + ProxyStatusCheckSuccess = "check success" + ProxyStatusClosed = "closed" ) type ProxyManager struct { - ctl *Control - + ctl *Control + sendCh chan (msg.Message) proxies map[string]*ProxyWrapper - - visitorCfgs map[string]config.ProxyConf - visitors map[string]Visitor - - sendCh chan (msg.Message) - - closed bool - mu sync.RWMutex + closed bool + mu sync.RWMutex log.Logger } @@ -151,13 +147,11 @@ func (pw *ProxyWrapper) Close() { func NewProxyManager(ctl *Control, msgSendCh chan (msg.Message), logPrefix string) *ProxyManager { return &ProxyManager{ - ctl: ctl, - proxies: make(map[string]*ProxyWrapper), - visitorCfgs: make(map[string]config.ProxyConf), - visitors: make(map[string]Visitor), - sendCh: msgSendCh, - closed: false, - Logger: log.NewPrefixLogger(logPrefix), + ctl: ctl, + proxies: make(map[string]*ProxyWrapper), + sendCh: msgSendCh, + closed: false, + Logger: log.NewPrefixLogger(logPrefix), } } @@ -239,24 +233,9 @@ func (pm *ProxyManager) CheckAndStartProxy(pxyStatus []string) { } } } - - for _, cfg := range pm.visitorCfgs { - name := cfg.GetBaseInfo().ProxyName - if _, exist := pm.visitors[name]; !exist { - pm.Info("try to start visitor [%s]", name) - visitor := NewVisitor(pm.ctl, cfg) - err := visitor.Run() - if err != nil { - visitor.Warn("start error: %v", err) - continue - } - pm.visitors[name] = visitor - visitor.Info("start visitor success") - } - } } -func (pm *ProxyManager) Reload(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.ProxyConf, startNow bool) error { +func (pm *ProxyManager) Reload(pxyCfgs map[string]config.ProxyConf, startNow bool) error { pm.mu.Lock() defer func() { pm.mu.Unlock() @@ -308,38 +287,6 @@ func (pm *ProxyManager) Reload(pxyCfgs map[string]config.ProxyConf, visitorCfgs } } pm.Info("proxy added: %v", addPxyNames) - - delVisitorName := make([]string, 0) - for name, oldVisitorCfg := range pm.visitorCfgs { - del := false - cfg, ok := visitorCfgs[name] - if !ok { - del = true - } else { - if !oldVisitorCfg.Compare(cfg) { - del = true - } - } - - if del { - delVisitorName = append(delVisitorName, name) - delete(pm.visitorCfgs, name) - if visitor, ok := pm.visitors[name]; ok { - visitor.Close() - } - delete(pm.visitors, name) - } - } - pm.Info("visitor removed: %v", delVisitorName) - - addVisitorName := make([]string, 0) - for name, visitorCfg := range visitorCfgs { - if _, ok := pm.visitorCfgs[name]; !ok { - pm.visitorCfgs[name] = visitorCfg - addVisitorName = append(addVisitorName, name) - } - } - pm.Info("visitor added: %v", addVisitorName) return nil } diff --git a/client/service.go b/client/service.go index 5fbf33c..2589f52 100644 --- a/client/service.go +++ b/client/service.go @@ -27,7 +27,7 @@ type Service struct { closedCh chan int } -func NewService(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.ProxyConf) (svr *Service) { +func NewService(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) (svr *Service) { svr = &Service{ closedCh: make(chan int), } diff --git a/client/visitor.go b/client/visitor.go index 44e384d..6e1e1c8 100644 --- a/client/visitor.go +++ b/client/visitor.go @@ -44,18 +44,18 @@ type Visitor interface { log.Logger } -func NewVisitor(ctl *Control, pxyConf config.ProxyConf) (visitor Visitor) { +func NewVisitor(ctl *Control, cfg config.VisitorConf) (visitor Visitor) { baseVisitor := BaseVisitor{ ctl: ctl, - Logger: log.NewPrefixLogger(pxyConf.GetBaseInfo().ProxyName), + Logger: log.NewPrefixLogger(cfg.GetBaseInfo().ProxyName), } - switch cfg := pxyConf.(type) { - case *config.StcpProxyConf: + switch cfg := cfg.(type) { + case *config.StcpVisitorConf: visitor = &StcpVisitor{ BaseVisitor: baseVisitor, cfg: cfg, } - case *config.XtcpProxyConf: + case *config.XtcpVisitorConf: visitor = &XtcpVisitor{ BaseVisitor: baseVisitor, cfg: cfg, @@ -75,7 +75,7 @@ type BaseVisitor struct { type StcpVisitor struct { BaseVisitor - cfg *config.StcpProxyConf + cfg *config.StcpVisitorConf } func (sv *StcpVisitor) Run() (err error) { @@ -162,7 +162,7 @@ func (sv *StcpVisitor) handleConn(userConn frpNet.Conn) { type XtcpVisitor struct { BaseVisitor - cfg *config.XtcpProxyConf + cfg *config.XtcpVisitorConf } func (sv *XtcpVisitor) Run() (err error) { diff --git a/client/visitor_manager.go b/client/visitor_manager.go new file mode 100644 index 0000000..3e0aa80 --- /dev/null +++ b/client/visitor_manager.go @@ -0,0 +1,111 @@ +// Copyright 2018 fatedier, fatedier@gmail.com +// +// 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 client + +import ( + "sync" + "time" + + "github.com/fatedier/frp/models/config" + "github.com/fatedier/frp/utils/log" +) + +type VisitorManager struct { + ctl *Control + + cfgs map[string]config.VisitorConf + visitors map[string]Visitor + + checkInterval time.Duration + + mu sync.Mutex +} + +func NewVisitorManager(ctl *Control) *VisitorManager { + return &VisitorManager{ + ctl: ctl, + cfgs: make(map[string]config.VisitorConf), + visitors: make(map[string]Visitor), + checkInterval: 10 * time.Second, + } +} + +func (vm *VisitorManager) Run() { + for { + time.Sleep(vm.checkInterval) + vm.mu.Lock() + for _, cfg := range vm.cfgs { + name := cfg.GetBaseInfo().ProxyName + if _, exist := vm.visitors[name]; !exist { + log.Info("try to start visitor [%s]", name) + vm.startVisitor(cfg) + } + } + vm.mu.Unlock() + } +} + +// Hold lock before calling this function. +func (vm *VisitorManager) startVisitor(cfg config.VisitorConf) (err error) { + name := cfg.GetBaseInfo().ProxyName + visitor := NewVisitor(vm.ctl, cfg) + err = visitor.Run() + if err != nil { + visitor.Warn("start error: %v", err) + } else { + vm.visitors[name] = visitor + visitor.Info("start visitor success") + } + return +} + +func (vm *VisitorManager) Reload(cfgs map[string]config.VisitorConf) { + vm.mu.Lock() + defer vm.mu.Unlock() + + delNames := make([]string, 0) + for name, oldCfg := range vm.cfgs { + del := false + cfg, ok := cfgs[name] + if !ok { + del = true + } else { + if !oldCfg.Compare(cfg) { + del = true + } + } + + if del { + delNames = append(delNames, name) + delete(vm.cfgs, name) + if visitor, ok := vm.visitors[name]; ok { + visitor.Close() + } + delete(vm.visitors, name) + } + } + log.Info("visitor removed: %v", delNames) + + addNames := make([]string, 0) + for name, cfg := range cfgs { + if _, ok := vm.cfgs[name]; !ok { + vm.cfgs[name] = cfg + addNames = append(addNames, name) + vm.startVisitor(cfg) + } + } + log.Info("visitor added: %v", addNames) + return +} diff --git a/cmd/frpc/sub/root.go b/cmd/frpc/sub/root.go index 9b9a226..23b3bf9 100644 --- a/cmd/frpc/sub/root.go +++ b/cmd/frpc/sub/root.go @@ -180,7 +180,7 @@ func runClient(cfgFilePath string) (err error) { return err } - pxyCfgs, visitorCfgs, err := config.LoadProxyConfFromIni(g.GlbClientCfg.User, conf, g.GlbClientCfg.Start) + pxyCfgs, visitorCfgs, err := config.LoadAllConfFromIni(g.GlbClientCfg.User, conf, g.GlbClientCfg.Start) if err != nil { return err } @@ -189,7 +189,7 @@ func runClient(cfgFilePath string) (err error) { return } -func startService(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.ProxyConf) (err error) { +func startService(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) (err error) { log.InitLog(g.GlbClientCfg.LogWay, g.GlbClientCfg.LogFile, g.GlbClientCfg.LogLevel, g.GlbClientCfg.LogMaxDays) if g.GlbClientCfg.DnsServer != "" { s := g.GlbClientCfg.DnsServer diff --git a/cmd/frpc/sub/stcp.go b/cmd/frpc/sub/stcp.go index 4915e52..0920927 100644 --- a/cmd/frpc/sub/stcp.go +++ b/cmd/frpc/sub/stcp.go @@ -57,48 +57,57 @@ var stcpCmd = &cobra.Command{ os.Exit(1) } - cfg := &config.StcpProxyConf{} + proxyConfs := make(map[string]config.ProxyConf) + visitorConfs := make(map[string]config.VisitorConf) + var prefix string if user != "" { prefix = user + "." } - cfg.ProxyName = prefix + proxyName - cfg.ProxyType = consts.StcpProxy - cfg.Role = role - cfg.Sk = sk - cfg.ServerName = serverName - cfg.LocalIp = localIp - cfg.LocalPort = localPort - cfg.BindAddr = bindAddr - cfg.BindPort = bindPort - cfg.UseEncryption = useEncryption - cfg.UseCompression = useCompression - err = cfg.CheckForCli() + if role == "server" { + cfg := &config.StcpProxyConf{} + cfg.ProxyName = prefix + proxyName + cfg.ProxyType = consts.StcpProxy + cfg.UseEncryption = useEncryption + cfg.UseCompression = useCompression + cfg.Role = role + cfg.Sk = sk + cfg.LocalIp = localIp + cfg.LocalPort = localPort + err = cfg.CheckForCli() + if err != nil { + fmt.Println(err) + os.Exit(1) + } + proxyConfs[cfg.ProxyName] = cfg + } else if role == "visitor" { + cfg := &config.StcpVisitorConf{} + cfg.ProxyName = prefix + proxyName + cfg.ProxyType = consts.StcpProxy + cfg.UseEncryption = useEncryption + cfg.UseCompression = useCompression + cfg.Role = role + cfg.Sk = sk + cfg.ServerName = serverName + cfg.BindAddr = bindAddr + cfg.BindPort = bindPort + err = cfg.Check() + if err != nil { + fmt.Println(err) + os.Exit(1) + } + visitorConfs[cfg.ProxyName] = cfg + } else { + fmt.Println("invalid role") + os.Exit(1) + } + + err = startService(proxyConfs, visitorConfs) if err != nil { fmt.Println(err) os.Exit(1) } - - if cfg.Role == "server" { - proxyConfs := map[string]config.ProxyConf{ - cfg.ProxyName: cfg, - } - err = startService(proxyConfs, nil) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - } else { - visitorConfs := map[string]config.ProxyConf{ - cfg.ProxyName: cfg, - } - err = startService(nil, visitorConfs) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - } return nil }, } diff --git a/cmd/frpc/sub/xtcp.go b/cmd/frpc/sub/xtcp.go index 8c18a85..b6ae541 100644 --- a/cmd/frpc/sub/xtcp.go +++ b/cmd/frpc/sub/xtcp.go @@ -57,48 +57,57 @@ var xtcpCmd = &cobra.Command{ os.Exit(1) } - cfg := &config.XtcpProxyConf{} + proxyConfs := make(map[string]config.ProxyConf) + visitorConfs := make(map[string]config.VisitorConf) + var prefix string if user != "" { prefix = user + "." } - cfg.ProxyName = prefix + proxyName - cfg.ProxyType = consts.XtcpProxy - cfg.Role = role - cfg.Sk = sk - cfg.ServerName = serverName - cfg.LocalIp = localIp - cfg.LocalPort = localPort - cfg.BindAddr = bindAddr - cfg.BindPort = bindPort - cfg.UseEncryption = useEncryption - cfg.UseCompression = useCompression - err = cfg.CheckForCli() + if role == "server" { + cfg := &config.XtcpProxyConf{} + cfg.ProxyName = prefix + proxyName + cfg.ProxyType = consts.StcpProxy + cfg.UseEncryption = useEncryption + cfg.UseCompression = useCompression + cfg.Role = role + cfg.Sk = sk + cfg.LocalIp = localIp + cfg.LocalPort = localPort + err = cfg.CheckForCli() + if err != nil { + fmt.Println(err) + os.Exit(1) + } + proxyConfs[cfg.ProxyName] = cfg + } else if role == "visitor" { + cfg := &config.XtcpVisitorConf{} + cfg.ProxyName = prefix + proxyName + cfg.ProxyType = consts.StcpProxy + cfg.UseEncryption = useEncryption + cfg.UseCompression = useCompression + cfg.Role = role + cfg.Sk = sk + cfg.ServerName = serverName + cfg.BindAddr = bindAddr + cfg.BindPort = bindPort + err = cfg.Check() + if err != nil { + fmt.Println(err) + os.Exit(1) + } + visitorConfs[cfg.ProxyName] = cfg + } else { + fmt.Println("invalid role") + os.Exit(1) + } + + err = startService(proxyConfs, visitorConfs) if err != nil { fmt.Println(err) os.Exit(1) } - - if cfg.Role == "server" { - proxyConfs := map[string]config.ProxyConf{ - cfg.ProxyName: cfg, - } - err = startService(proxyConfs, nil) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - } else { - visitorConfs := map[string]config.ProxyConf{ - cfg.ProxyName: cfg, - } - err = startService(nil, visitorConfs) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - } return nil }, } diff --git a/conf/frpc_full.ini b/conf/frpc_full.ini index c3d13ff..2307eeb 100644 --- a/conf/frpc_full.ini +++ b/conf/frpc_full.ini @@ -73,6 +73,10 @@ remote_port = 6001 group = test_group # group should have same group key group_key = 123456 +# enable health check for the backend service, it support 'tcp' and 'http' now +# frpc will connect local service's port to detect it's healthy status +health_check_type = tcp +health_check_interval_s = 10 [ssh_random] type = tcp @@ -126,6 +130,11 @@ locations = /,/pic host_header_rewrite = example.com # params with prefix "header_" will be used to update http request headers header_X-From-Where = frp +health_check_type = http +# frpc will send a GET http request '/status' to local http service +# http service is alive when it return 2xx http response code +health_check_url = /status +health_check_interval_s = 10 [web02] type = https diff --git a/models/config/proxy.go b/models/config/proxy.go index 0270c1c..b600be5 100644 --- a/models/config/proxy.go +++ b/models/config/proxy.go @@ -91,7 +91,9 @@ func NewProxyConfFromIni(prefix string, name string, section ini.Section) (cfg P if err = cfg.UnmarshalFromIni(prefix, name, section); err != nil { return } - err = cfg.CheckForCli() + if err = cfg.CheckForCli(); err != nil { + return + } return } @@ -104,6 +106,9 @@ type BaseProxyConf struct { UseCompression bool `json:"use_compression"` Group string `json:"group"` GroupKey string `json:"group_key"` + + LocalSvrConf + HealthCheckConf // only used for client } func (cfg *BaseProxyConf) GetBaseInfo() *BaseProxyConf { @@ -119,6 +124,12 @@ func (cfg *BaseProxyConf) compare(cmp *BaseProxyConf) bool { cfg.GroupKey != cmp.GroupKey { return false } + if !cfg.LocalSvrConf.compare(&cmp.LocalSvrConf) { + return false + } + if !cfg.HealthCheckConf.compare(&cmp.HealthCheckConf) { + return false + } return true } @@ -151,6 +162,14 @@ func (cfg *BaseProxyConf) UnmarshalFromIni(prefix string, name string, section i cfg.Group = section["group"] cfg.GroupKey = section["group_key"] + + if err := cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { + return err + } + + if err := cfg.HealthCheckConf.UnmarshalFromIni(prefix, name, section); err != nil { + return err + } return nil } @@ -163,6 +182,16 @@ func (cfg *BaseProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { pMsg.GroupKey = cfg.GroupKey } +func (cfg *BaseProxyConf) checkForCli() (err error) { + if err = cfg.LocalSvrConf.checkForCli(); err != nil { + return + } + if err = cfg.HealthCheckConf.checkForCli(); err != nil { + return + } + return nil +} + // Bind info type BindInfoConf struct { RemotePort int `json:"remote_port"` @@ -335,12 +364,70 @@ func (cfg *LocalSvrConf) UnmarshalFromIni(prefix string, name string, section in return } +func (cfg *LocalSvrConf) checkForCli() (err error) { + if cfg.Plugin == "" { + if cfg.LocalIp == "" { + err = fmt.Errorf("local ip or plugin is required") + return + } + if cfg.LocalPort <= 0 { + err = fmt.Errorf("error local_port") + return + } + } + return +} + +// Health check info +type HealthCheckConf struct { + HealthCheckType string `json:"health_check_type"` // tcp | http + HealthCheckIntervalS int `json:"health_check_interval_s"` + HealthCheckUrl string `json:"health_check_url"` + + // local_ip + local_port + HealthCheckAddr string `json:"-"` +} + +func (cfg *HealthCheckConf) compare(cmp *HealthCheckConf) bool { + if cfg.HealthCheckType != cmp.HealthCheckType || + cfg.HealthCheckUrl != cmp.HealthCheckUrl || + cfg.HealthCheckIntervalS != cmp.HealthCheckIntervalS { + return false + } + return true +} + +func (cfg *HealthCheckConf) UnmarshalFromIni(prefix string, name string, section ini.Section) (err error) { + cfg.HealthCheckType = section["health_check_type"] + cfg.HealthCheckUrl = section["health_check_url"] + + if tmpStr, ok := section["health_check_interval_s"]; ok { + if cfg.HealthCheckIntervalS, err = strconv.Atoi(tmpStr); err != nil { + return fmt.Errorf("Parse conf error: proxy [%s] health_check_interval_s error", name) + } + } + return +} + +func (cfg *HealthCheckConf) checkForCli() error { + if cfg.HealthCheckType != "" && cfg.HealthCheckType != "tcp" && cfg.HealthCheckType != "http" { + return fmt.Errorf("unsupport health check type") + } + if cfg.HealthCheckType != "" { + if cfg.HealthCheckType == "http" && cfg.HealthCheckUrl == "" { + return fmt.Errorf("health_check_url is required for health check type 'http'") + } + if cfg.HealthCheckIntervalS <= 0 { + return fmt.Errorf("health_check_interval_s is required and should greater than 0") + } + } + return nil +} + // TCP type TcpProxyConf struct { BaseProxyConf BindInfoConf - - LocalSvrConf } func (cfg *TcpProxyConf) Compare(cmp ProxyConf) bool { @@ -350,8 +437,7 @@ func (cfg *TcpProxyConf) Compare(cmp ProxyConf) bool { } if !cfg.BaseProxyConf.compare(&cmpConf.BaseProxyConf) || - !cfg.BindInfoConf.compare(&cmpConf.BindInfoConf) || - !cfg.LocalSvrConf.compare(&cmpConf.LocalSvrConf) { + !cfg.BindInfoConf.compare(&cmpConf.BindInfoConf) { return false } return true @@ -369,9 +455,6 @@ func (cfg *TcpProxyConf) UnmarshalFromIni(prefix string, name string, section in if err = cfg.BindInfoConf.UnmarshalFromIni(prefix, name, section); err != nil { return } - if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { - return - } return } @@ -380,7 +463,12 @@ func (cfg *TcpProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { cfg.BindInfoConf.MarshalToMsg(pMsg) } -func (cfg *TcpProxyConf) CheckForCli() error { return nil } +func (cfg *TcpProxyConf) CheckForCli() (err error) { + if err = cfg.BaseProxyConf.checkForCli(); err != nil { + return err + } + return +} func (cfg *TcpProxyConf) CheckForSvr() error { return nil } @@ -388,8 +476,6 @@ func (cfg *TcpProxyConf) CheckForSvr() error { return nil } type UdpProxyConf struct { BaseProxyConf BindInfoConf - - LocalSvrConf } func (cfg *UdpProxyConf) Compare(cmp ProxyConf) bool { @@ -399,8 +485,7 @@ func (cfg *UdpProxyConf) Compare(cmp ProxyConf) bool { } if !cfg.BaseProxyConf.compare(&cmpConf.BaseProxyConf) || - !cfg.BindInfoConf.compare(&cmpConf.BindInfoConf) || - !cfg.LocalSvrConf.compare(&cmpConf.LocalSvrConf) { + !cfg.BindInfoConf.compare(&cmpConf.BindInfoConf) { return false } return true @@ -418,9 +503,6 @@ func (cfg *UdpProxyConf) UnmarshalFromIni(prefix string, name string, section in if err = cfg.BindInfoConf.UnmarshalFromIni(prefix, name, section); err != nil { return } - if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { - return - } return } @@ -429,7 +511,12 @@ func (cfg *UdpProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { cfg.BindInfoConf.MarshalToMsg(pMsg) } -func (cfg *UdpProxyConf) CheckForCli() error { return nil } +func (cfg *UdpProxyConf) CheckForCli() (err error) { + if err = cfg.BaseProxyConf.checkForCli(); err != nil { + return + } + return +} func (cfg *UdpProxyConf) CheckForSvr() error { return nil } @@ -438,8 +525,6 @@ type HttpProxyConf struct { BaseProxyConf DomainConf - LocalSvrConf - Locations []string `json:"locations"` HttpUser string `json:"http_user"` HttpPwd string `json:"http_pwd"` @@ -455,7 +540,6 @@ func (cfg *HttpProxyConf) Compare(cmp ProxyConf) bool { if !cfg.BaseProxyConf.compare(&cmpConf.BaseProxyConf) || !cfg.DomainConf.compare(&cmpConf.DomainConf) || - !cfg.LocalSvrConf.compare(&cmpConf.LocalSvrConf) || strings.Join(cfg.Locations, " ") != strings.Join(cmpConf.Locations, " ") || cfg.HostHeaderRewrite != cmpConf.HostHeaderRewrite || cfg.HttpUser != cmpConf.HttpUser || @@ -494,9 +578,6 @@ func (cfg *HttpProxyConf) UnmarshalFromIni(prefix string, name string, section i if err = cfg.DomainConf.UnmarshalFromIni(prefix, name, section); err != nil { return } - if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { - return - } var ( tmpStr string @@ -533,6 +614,9 @@ func (cfg *HttpProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { } func (cfg *HttpProxyConf) CheckForCli() (err error) { + if err = cfg.BaseProxyConf.checkForCli(); err != nil { + return + } if err = cfg.DomainConf.checkForCli(); err != nil { return } @@ -554,8 +638,6 @@ func (cfg *HttpProxyConf) CheckForSvr() (err error) { type HttpsProxyConf struct { BaseProxyConf DomainConf - - LocalSvrConf } func (cfg *HttpsProxyConf) Compare(cmp ProxyConf) bool { @@ -565,8 +647,7 @@ func (cfg *HttpsProxyConf) Compare(cmp ProxyConf) bool { } if !cfg.BaseProxyConf.compare(&cmpConf.BaseProxyConf) || - !cfg.DomainConf.compare(&cmpConf.DomainConf) || - !cfg.LocalSvrConf.compare(&cmpConf.LocalSvrConf) { + !cfg.DomainConf.compare(&cmpConf.DomainConf) { return false } return true @@ -584,9 +665,6 @@ func (cfg *HttpsProxyConf) UnmarshalFromIni(prefix string, name string, section if err = cfg.DomainConf.UnmarshalFromIni(prefix, name, section); err != nil { return } - if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { - return - } return } @@ -596,6 +674,9 @@ func (cfg *HttpsProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { } func (cfg *HttpsProxyConf) CheckForCli() (err error) { + if err = cfg.BaseProxyConf.checkForCli(); err != nil { + return + } if err = cfg.DomainConf.checkForCli(); err != nil { return } @@ -619,14 +700,6 @@ type StcpProxyConf struct { Role string `json:"role"` Sk string `json:"sk"` - - // used in role server - LocalSvrConf - - // used in role visitor - ServerName string `json:"server_name"` - BindAddr string `json:"bind_addr"` - BindPort int `json:"bind_port"` } func (cfg *StcpProxyConf) Compare(cmp ProxyConf) bool { @@ -636,12 +709,8 @@ func (cfg *StcpProxyConf) Compare(cmp ProxyConf) bool { } if !cfg.BaseProxyConf.compare(&cmpConf.BaseProxyConf) || - !cfg.LocalSvrConf.compare(&cmpConf.LocalSvrConf) || cfg.Role != cmpConf.Role || - cfg.Sk != cmpConf.Sk || - cfg.ServerName != cmpConf.ServerName || - cfg.BindAddr != cmpConf.BindAddr || - cfg.BindPort != cmpConf.BindPort { + cfg.Sk != cmpConf.Sk { return false } return true @@ -658,35 +727,15 @@ func (cfg *StcpProxyConf) UnmarshalFromIni(prefix string, name string, section i return } - tmpStr := section["role"] - if tmpStr == "" { - tmpStr = "server" - } - if tmpStr == "server" || tmpStr == "visitor" { - cfg.Role = tmpStr - } else { - return fmt.Errorf("Parse conf error: proxy [%s] incorrect role [%s]", name, tmpStr) + cfg.Role = section["role"] + if cfg.Role != "server" { + return fmt.Errorf("Parse conf error: proxy [%s] incorrect role [%s]", name, cfg.Role) } cfg.Sk = section["sk"] - if tmpStr == "visitor" { - cfg.ServerName = prefix + section["server_name"] - if cfg.BindAddr = section["bind_addr"]; cfg.BindAddr == "" { - cfg.BindAddr = "127.0.0.1" - } - - if tmpStr, ok := section["bind_port"]; ok { - if cfg.BindPort, err = strconv.Atoi(tmpStr); err != nil { - return fmt.Errorf("Parse conf error: proxy [%s] bind_port error", name) - } - } else { - return fmt.Errorf("Parse conf error: proxy [%s] bind_port not found", name) - } - } else { - if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { - return - } + if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { + return } return } @@ -697,19 +746,12 @@ func (cfg *StcpProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { } func (cfg *StcpProxyConf) CheckForCli() (err error) { - if cfg.Role != "server" && cfg.Role != "visitor" { - err = fmt.Errorf("role should be 'server' or 'visitor'") + if err = cfg.BaseProxyConf.checkForCli(); err != nil { return } - if cfg.Role == "visitor" { - if cfg.BindAddr == "" { - err = fmt.Errorf("bind_addr shouldn't be empty") - return - } - if cfg.BindPort == 0 { - err = fmt.Errorf("bind_port should be set") - return - } + if cfg.Role != "server" { + err = fmt.Errorf("role should be 'server'") + return } return } @@ -724,14 +766,6 @@ type XtcpProxyConf struct { Role string `json:"role"` Sk string `json:"sk"` - - // used in role server - LocalSvrConf - - // used in role visitor - ServerName string `json:"server_name"` - BindAddr string `json:"bind_addr"` - BindPort int `json:"bind_port"` } func (cfg *XtcpProxyConf) Compare(cmp ProxyConf) bool { @@ -743,10 +777,7 @@ func (cfg *XtcpProxyConf) Compare(cmp ProxyConf) bool { if !cfg.BaseProxyConf.compare(&cmpConf.BaseProxyConf) || !cfg.LocalSvrConf.compare(&cmpConf.LocalSvrConf) || cfg.Role != cmpConf.Role || - cfg.Sk != cmpConf.Sk || - cfg.ServerName != cmpConf.ServerName || - cfg.BindAddr != cmpConf.BindAddr || - cfg.BindPort != cmpConf.BindPort { + cfg.Sk != cmpConf.Sk { return false } return true @@ -763,35 +794,15 @@ func (cfg *XtcpProxyConf) UnmarshalFromIni(prefix string, name string, section i return } - tmpStr := section["role"] - if tmpStr == "" { - tmpStr = "server" - } - if tmpStr == "server" || tmpStr == "visitor" { - cfg.Role = tmpStr - } else { - return fmt.Errorf("Parse conf error: proxy [%s] incorrect role [%s]", name, tmpStr) + cfg.Role = section["role"] + if cfg.Role != "server" { + return fmt.Errorf("Parse conf error: proxy [%s] incorrect role [%s]", name, cfg.Role) } cfg.Sk = section["sk"] - if tmpStr == "visitor" { - cfg.ServerName = prefix + section["server_name"] - if cfg.BindAddr = section["bind_addr"]; cfg.BindAddr == "" { - cfg.BindAddr = "127.0.0.1" - } - - if tmpStr, ok := section["bind_port"]; ok { - if cfg.BindPort, err = strconv.Atoi(tmpStr); err != nil { - return fmt.Errorf("Parse conf error: proxy [%s] bind_port error", name) - } - } else { - return fmt.Errorf("Parse conf error: proxy [%s] bind_port not found", name) - } - } else { - if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { - return - } + if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil { + return } return } @@ -802,19 +813,12 @@ func (cfg *XtcpProxyConf) MarshalToMsg(pMsg *msg.NewProxy) { } func (cfg *XtcpProxyConf) CheckForCli() (err error) { - if cfg.Role != "server" && cfg.Role != "visitor" { - err = fmt.Errorf("role should be 'server' or 'visitor'") + if err = cfg.BaseProxyConf.checkForCli(); err != nil { return } - if cfg.Role == "visitor" { - if cfg.BindAddr == "" { - err = fmt.Errorf("bind_addr shouldn't be empty") - return - } - if cfg.BindPort == 0 { - err = fmt.Errorf("bind_port should be set") - return - } + if cfg.Role != "server" { + err = fmt.Errorf("role should be 'server'") + return } return } @@ -857,8 +861,8 @@ func ParseRangeSection(name string, section ini.Section) (sections map[string]in // if len(startProxy) is 0, start all // otherwise just start proxies in startProxy map -func LoadProxyConfFromIni(prefix string, conf ini.File, startProxy map[string]struct{}) ( - proxyConfs map[string]ProxyConf, visitorConfs map[string]ProxyConf, err error) { +func LoadAllConfFromIni(prefix string, conf ini.File, startProxy map[string]struct{}) ( + proxyConfs map[string]ProxyConf, visitorConfs map[string]VisitorConf, err error) { if prefix != "" { prefix += "." @@ -869,7 +873,7 @@ func LoadProxyConfFromIni(prefix string, conf ini.File, startProxy map[string]st startAll = false } proxyConfs = make(map[string]ProxyConf) - visitorConfs = make(map[string]ProxyConf) + visitorConfs = make(map[string]VisitorConf) for name, section := range conf { if name == "common" { continue @@ -894,16 +898,27 @@ func LoadProxyConfFromIni(prefix string, conf ini.File, startProxy map[string]st } for subName, subSection := range subSections { - cfg, err := NewProxyConfFromIni(prefix, subName, subSection) - if err != nil { - return proxyConfs, visitorConfs, err + if subSection["role"] == "" { + subSection["role"] = "server" } - role := subSection["role"] - if role == "visitor" { + if role == "server" { + cfg, errRet := NewProxyConfFromIni(prefix, subName, subSection) + if errRet != nil { + err = errRet + return + } + proxyConfs[prefix+subName] = cfg + } else if role == "visitor" { + cfg, errRet := NewVisitorConfFromIni(prefix, subName, subSection) + if errRet != nil { + err = errRet + return + } visitorConfs[prefix+subName] = cfg } else { - proxyConfs[prefix+subName] = cfg + err = fmt.Errorf("role should be 'server' or 'visitor'") + return } } } diff --git a/models/config/visitor.go b/models/config/visitor.go new file mode 100644 index 0000000..4233375 --- /dev/null +++ b/models/config/visitor.go @@ -0,0 +1,213 @@ +// Copyright 2018 fatedier, fatedier@gmail.com +// +// 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 config + +import ( + "fmt" + "reflect" + "strconv" + + "github.com/fatedier/frp/models/consts" + + ini "github.com/vaughan0/go-ini" +) + +var ( + visitorConfTypeMap map[string]reflect.Type +) + +func init() { + visitorConfTypeMap = make(map[string]reflect.Type) + visitorConfTypeMap[consts.StcpProxy] = reflect.TypeOf(StcpVisitorConf{}) + visitorConfTypeMap[consts.XtcpProxy] = reflect.TypeOf(XtcpVisitorConf{}) +} + +type VisitorConf interface { + GetBaseInfo() *BaseVisitorConf + Compare(cmp VisitorConf) bool + UnmarshalFromIni(prefix string, name string, section ini.Section) error + Check() error +} + +func NewVisitorConfByType(cfgType string) VisitorConf { + v, ok := visitorConfTypeMap[cfgType] + if !ok { + return nil + } + cfg := reflect.New(v).Interface().(VisitorConf) + return cfg +} + +func NewVisitorConfFromIni(prefix string, name string, section ini.Section) (cfg VisitorConf, err error) { + cfgType := section["type"] + if cfgType == "" { + err = fmt.Errorf("visitor [%s] type shouldn't be empty", name) + return + } + cfg = NewVisitorConfByType(cfgType) + if cfg == nil { + err = fmt.Errorf("visitor [%s] type [%s] error", name, cfgType) + return + } + if err = cfg.UnmarshalFromIni(prefix, name, section); err != nil { + return + } + if err = cfg.Check(); err != nil { + return + } + return +} + +type BaseVisitorConf struct { + ProxyName string `json:"proxy_name"` + ProxyType string `json:"proxy_type"` + UseEncryption bool `json:"use_encryption"` + UseCompression bool `json:"use_compression"` + Role string `json:"role"` + Sk string `json:"sk"` + ServerName string `json:"server_name"` + BindAddr string `json:"bind_addr"` + BindPort int `json:"bind_port"` +} + +func (cfg *BaseVisitorConf) GetBaseInfo() *BaseVisitorConf { + return cfg +} + +func (cfg *BaseVisitorConf) compare(cmp *BaseVisitorConf) bool { + if cfg.ProxyName != cmp.ProxyName || + cfg.ProxyType != cmp.ProxyType || + cfg.UseEncryption != cmp.UseEncryption || + cfg.UseCompression != cmp.UseCompression || + cfg.Role != cmp.Role || + cfg.Sk != cmp.Sk || + cfg.ServerName != cmp.ServerName || + cfg.BindAddr != cmp.BindAddr || + cfg.BindPort != cmp.BindPort { + return false + } + return true +} + +func (cfg *BaseVisitorConf) check() (err error) { + if cfg.Role != "visitor" { + err = fmt.Errorf("invalid role") + return + } + if cfg.BindAddr == "" { + err = fmt.Errorf("bind_addr shouldn't be empty") + return + } + if cfg.BindPort <= 0 { + err = fmt.Errorf("bind_port is required") + return + } + return +} + +func (cfg *BaseVisitorConf) UnmarshalFromIni(prefix string, name string, section ini.Section) (err error) { + var ( + tmpStr string + ok bool + ) + cfg.ProxyName = prefix + name + cfg.ProxyType = section["type"] + + if tmpStr, ok = section["use_encryption"]; ok && tmpStr == "true" { + cfg.UseEncryption = true + } + if tmpStr, ok = section["use_compression"]; ok && tmpStr == "true" { + cfg.UseCompression = true + } + + cfg.Role = section["role"] + if cfg.Role != "visitor" { + return fmt.Errorf("Parse conf error: proxy [%s] incorrect role [%s]", name, cfg.Role) + } + cfg.Sk = section["sk"] + cfg.ServerName = prefix + section["server_name"] + if cfg.BindAddr = section["bind_addr"]; cfg.BindAddr == "" { + cfg.BindAddr = "127.0.0.1" + } + + if tmpStr, ok = section["bind_port"]; ok { + if cfg.BindPort, err = strconv.Atoi(tmpStr); err != nil { + return fmt.Errorf("Parse conf error: proxy [%s] bind_port incorrect", name) + } + } else { + return fmt.Errorf("Parse conf error: proxy [%s] bind_port not found", name) + } + return nil +} + +type StcpVisitorConf struct { + BaseVisitorConf +} + +func (cfg *StcpVisitorConf) Compare(cmp VisitorConf) bool { + cmpConf, ok := cmp.(*StcpVisitorConf) + if !ok { + return false + } + + if !cfg.BaseVisitorConf.compare(&cmpConf.BaseVisitorConf) { + return false + } + return true +} + +func (cfg *StcpVisitorConf) UnmarshalFromIni(prefix string, name string, section ini.Section) (err error) { + if err = cfg.BaseVisitorConf.UnmarshalFromIni(prefix, name, section); err != nil { + return + } + return +} + +func (cfg *StcpVisitorConf) Check() (err error) { + if err = cfg.BaseVisitorConf.check(); err != nil { + return + } + return +} + +type XtcpVisitorConf struct { + BaseVisitorConf +} + +func (cfg *XtcpVisitorConf) Compare(cmp VisitorConf) bool { + cmpConf, ok := cmp.(*XtcpVisitorConf) + if !ok { + return false + } + + if !cfg.BaseVisitorConf.compare(&cmpConf.BaseVisitorConf) { + return false + } + return true +} + +func (cfg *XtcpVisitorConf) UnmarshalFromIni(prefix string, name string, section ini.Section) (err error) { + if err = cfg.BaseVisitorConf.UnmarshalFromIni(prefix, name, section); err != nil { + return + } + return +} + +func (cfg *XtcpVisitorConf) Check() (err error) { + if err = cfg.BaseVisitorConf.check(); err != nil { + return + } + return +} From 42ee536dae2ace3d423de6c5262819b71e82312c Mon Sep 17 00:00:00 2001 From: fatedier Date: Wed, 27 Jun 2018 11:46:53 +0800 Subject: [PATCH 02/10] add module comments for vgo --- cmd/frpc/main.go | 2 +- cmd/frps/main.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/cmd/frpc/main.go b/cmd/frpc/main.go index 215325d..3c6f8e6 100644 --- a/cmd/frpc/main.go +++ b/cmd/frpc/main.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package main +package main // "github.com/fatedier/frp/cmd/frpc" import ( "github.com/fatedier/frp/cmd/frpc/sub" diff --git a/cmd/frps/main.go b/cmd/frps/main.go index 4e235ac..34f90d1 100644 --- a/cmd/frps/main.go +++ b/cmd/frps/main.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package main +package main // "github.com/fatedier/frp/cmd/frps" import ( "github.com/fatedier/golib/crypto" From 0d02f291e323625bbb6ef6efc5f30374cb92e129 Mon Sep 17 00:00:00 2001 From: fatedier Date: Wed, 11 Jul 2018 23:27:47 +0800 Subject: [PATCH 03/10] refactor ci test --- Makefile | 20 +- tests/clean_test.sh | 20 -- tests/{conf => config}/auto_test_frpc.ini | 2 +- .../auto_test_frpc_visitor.ini | 2 +- tests/{conf => config}/auto_test_frps.ini | 2 +- tests/config/config.go | 9 + tests/consts/consts.go | 65 ++++++ tests/func_test.go | 215 ++++++++---------- tests/{ => mock}/echo_server.go | 17 +- tests/{ => mock}/http_server.go | 16 +- tests/run_test.sh | 9 - tests/util/process.go | 29 +++ tests/{ => util}/util.go | 20 +- 13 files changed, 240 insertions(+), 186 deletions(-) delete mode 100755 tests/clean_test.sh rename tests/{conf => config}/auto_test_frpc.ini (99%) rename tests/{conf => config}/auto_test_frpc_visitor.ini (92%) rename tests/{conf => config}/auto_test_frps.ini (88%) create mode 100644 tests/config/config.go create mode 100644 tests/consts/consts.go rename tests/{ => mock}/echo_server.go (83%) rename tests/{ => mock}/http_server.go (80%) delete mode 100755 tests/run_test.sh create mode 100644 tests/util/process.go rename tests/{ => util}/util.go (86%) diff --git a/Makefile b/Makefile index 488e57a..bc450af 100644 --- a/Makefile +++ b/Makefile @@ -26,24 +26,18 @@ frpc: test: gotest gotest: - go test -v ./assets/... - go test -v ./client/... - go test -v ./cmd/... - go test -v ./models/... - go test -v ./server/... - go test -v ./utils/... + go test -v --cover ./assets/... + go test -v --cover ./client/... + go test -v --cover ./cmd/... + go test -v --cover ./models/... + go test -v --cover ./server/... + go test -v --cover ./utils/... ci: - cd ./tests && ./run_test.sh && cd - - go test -v ./tests/... - cd ./tests && ./clean_test.sh && cd - - -cic: - cd ./tests && ./clean_test.sh && cd - + go test -count=1 -v ./tests/... alltest: gotest ci clean: rm -f ./bin/frpc rm -f ./bin/frps - cd ./tests && ./clean_test.sh && cd - diff --git a/tests/clean_test.sh b/tests/clean_test.sh deleted file mode 100755 index b0b3763..0000000 --- a/tests/clean_test.sh +++ /dev/null @@ -1,20 +0,0 @@ -#!/bin/bash - -pid=`ps aux|grep './../bin/frps -c ./conf/auto_test_frps.ini'|grep -v grep|awk {'print $2'}` -if [ -n "${pid}" ]; then - kill ${pid} -fi - -pid=`ps aux|grep './../bin/frpc -c ./conf/auto_test_frpc.ini'|grep -v grep|awk {'print $2'}` -if [ -n "${pid}" ]; then - kill ${pid} -fi - -pid=`ps aux|grep './../bin/frpc -c ./conf/auto_test_frpc_visitor.ini'|grep -v grep|awk {'print $2'}` -if [ -n "${pid}" ]; then - kill ${pid} -fi - -rm -f ./frps.log -rm -f ./frpc.log -rm -f ./frpc_visitor.log diff --git a/tests/conf/auto_test_frpc.ini b/tests/config/auto_test_frpc.ini similarity index 99% rename from tests/conf/auto_test_frpc.ini rename to tests/config/auto_test_frpc.ini index 14f2e85..407d679 100644 --- a/tests/conf/auto_test_frpc.ini +++ b/tests/config/auto_test_frpc.ini @@ -1,7 +1,7 @@ [common] server_addr = 127.0.0.1 server_port = 10700 -log_file = ./frpc.log +log_file = console # debug, info, warn, error log_level = debug token = 123456 diff --git a/tests/conf/auto_test_frpc_visitor.ini b/tests/config/auto_test_frpc_visitor.ini similarity index 92% rename from tests/conf/auto_test_frpc_visitor.ini rename to tests/config/auto_test_frpc_visitor.ini index e06d160..660c793 100644 --- a/tests/conf/auto_test_frpc_visitor.ini +++ b/tests/config/auto_test_frpc_visitor.ini @@ -1,7 +1,7 @@ [common] server_addr = 0.0.0.0 server_port = 10700 -log_file = ./frpc_visitor.log +log_file = console # debug, info, warn, error log_level = debug token = 123456 diff --git a/tests/conf/auto_test_frps.ini b/tests/config/auto_test_frps.ini similarity index 88% rename from tests/conf/auto_test_frps.ini rename to tests/config/auto_test_frps.ini index fc62c39..9300b55 100644 --- a/tests/conf/auto_test_frps.ini +++ b/tests/config/auto_test_frps.ini @@ -2,7 +2,7 @@ bind_addr = 0.0.0.0 bind_port = 10700 vhost_http_port = 10804 -log_file = ./frps.log +log_file = console log_level = debug token = 123456 allow_ports = 10000-20000,20002,30000-50000 diff --git a/tests/config/config.go b/tests/config/config.go new file mode 100644 index 0000000..80e096b --- /dev/null +++ b/tests/config/config.go @@ -0,0 +1,9 @@ +package util + +import ( + "io/ioutil" +) + +func GenerateConfigFile(path string, content string) error { + return ioutil.WriteFile(path, []byte(content), 0666) +} diff --git a/tests/consts/consts.go b/tests/consts/consts.go new file mode 100644 index 0000000..7e70c89 --- /dev/null +++ b/tests/consts/consts.go @@ -0,0 +1,65 @@ +package consts + +import "path/filepath" + +var ( + FRPS_BIN_PATH = "../bin/frps" + FRPC_BIN_PATH = "../bin/frpc" + + SERVER_ADDR = "127.0.0.1" + ADMIN_ADDR = "127.0.0.1:10600" + ADMIN_USER = "abc" + ADMIN_PWD = "abc" + + TEST_STR = "frp is a fast reverse proxy to help you expose a local server behind a NAT or firewall to the internet." + TEST_TCP_PORT int = 10701 + TEST_TCP2_PORT int = 10702 + TEST_TCP_FRP_PORT int = 10801 + TEST_TCP2_FRP_PORT int = 10802 + TEST_TCP_EC_FRP_PORT int = 10901 + TEST_TCP_ECHO_STR string = "tcp type:" + TEST_STR + + TEST_UDP_PORT int = 10702 + TEST_UDP_FRP_PORT int = 10802 + TEST_UDP_EC_FRP_PORT int = 10902 + TEST_UDP_ECHO_STR string = "udp type:" + TEST_STR + + TEST_UNIX_DOMAIN_ADDR string = "/tmp/frp_echo_server.sock" + TEST_UNIX_DOMAIN_FRP_PORT int = 10803 + TEST_UNIX_DOMAIN_STR string = "unix domain type:" + TEST_STR + + TEST_HTTP_PORT int = 10704 + TEST_HTTP_FRP_PORT int = 10804 + TEST_HTTP_NORMAL_STR string = "http normal string: " + TEST_STR + TEST_HTTP_FOO_STR string = "http foo string: " + TEST_STR + TEST_HTTP_BAR_STR string = "http bar string: " + TEST_STR + + TEST_STCP_FRP_PORT int = 10805 + TEST_STCP_EC_FRP_PORT int = 10905 + TEST_STCP_ECHO_STR string = "stcp type:" + TEST_STR + + ProxyTcpPortNotAllowed string = "tcp_port_not_allowed" + ProxyTcpPortUnavailable string = "tcp_port_unavailable" + ProxyTcpPortNormal string = "tcp_port_normal" + ProxyTcpRandomPort string = "tcp_random_port" + ProxyUdpPortNotAllowed string = "udp_port_not_allowed" + ProxyUdpPortNormal string = "udp_port_normal" + ProxyUdpRandomPort string = "udp_random_port" + ProxyHttpProxy string = "http_proxy" + + ProxyRangeTcpPrefix string = "range_tcp" +) + +func init() { + if path, err := filepath.Abs(FRPS_BIN_PATH); err != nil { + panic(err) + } else { + FRPS_BIN_PATH = path + } + + if path, err := filepath.Abs(FRPC_BIN_PATH); err != nil { + panic(err) + } else { + FRPC_BIN_PATH = path + } +} diff --git a/tests/func_test.go b/tests/func_test.go index 1d0cd37..a83a6dc 100644 --- a/tests/func_test.go +++ b/tests/func_test.go @@ -13,194 +13,179 @@ import ( "github.com/fatedier/frp/client" "github.com/fatedier/frp/server/ports" + "github.com/fatedier/frp/tests/consts" + "github.com/fatedier/frp/tests/mock" + "github.com/fatedier/frp/tests/util" gnet "github.com/fatedier/golib/net" ) -var ( - SERVER_ADDR = "127.0.0.1" - ADMIN_ADDR = "127.0.0.1:10600" - ADMIN_USER = "abc" - ADMIN_PWD = "abc" - - TEST_STR = "frp is a fast reverse proxy to help you expose a local server behind a NAT or firewall to the internet." - TEST_TCP_PORT int = 10701 - TEST_TCP2_PORT int = 10702 - TEST_TCP_FRP_PORT int = 10801 - TEST_TCP2_FRP_PORT int = 10802 - TEST_TCP_EC_FRP_PORT int = 10901 - TEST_TCP_ECHO_STR string = "tcp type:" + TEST_STR - - TEST_UDP_PORT int = 10702 - TEST_UDP_FRP_PORT int = 10802 - TEST_UDP_EC_FRP_PORT int = 10902 - TEST_UDP_ECHO_STR string = "udp type:" + TEST_STR - - TEST_UNIX_DOMAIN_ADDR string = "/tmp/frp_echo_server.sock" - TEST_UNIX_DOMAIN_FRP_PORT int = 10803 - TEST_UNIX_DOMAIN_STR string = "unix domain type:" + TEST_STR - - TEST_HTTP_PORT int = 10704 - TEST_HTTP_FRP_PORT int = 10804 - TEST_HTTP_NORMAL_STR string = "http normal string: " + TEST_STR - TEST_HTTP_FOO_STR string = "http foo string: " + TEST_STR - TEST_HTTP_BAR_STR string = "http bar string: " + TEST_STR - - TEST_STCP_FRP_PORT int = 10805 - TEST_STCP_EC_FRP_PORT int = 10905 - TEST_STCP_ECHO_STR string = "stcp type:" + TEST_STR - - ProxyTcpPortNotAllowed string = "tcp_port_not_allowed" - ProxyTcpPortUnavailable string = "tcp_port_unavailable" - ProxyTcpPortNormal string = "tcp_port_normal" - ProxyTcpRandomPort string = "tcp_random_port" - ProxyUdpPortNotAllowed string = "udp_port_not_allowed" - ProxyUdpPortNormal string = "udp_port_normal" - ProxyUdpRandomPort string = "udp_random_port" - ProxyHttpProxy string = "http_proxy" - - ProxyRangeTcpPrefix string = "range_tcp" -) - func init() { - go StartTcpEchoServer() - go StartTcpEchoServer2() - go StartUdpEchoServer() - go StartUnixDomainServer() - go StartHttpServer() + go mock.StartTcpEchoServer(consts.TEST_TCP_PORT) + go mock.StartTcpEchoServer2(consts.TEST_TCP2_PORT) + go mock.StartUdpEchoServer(consts.TEST_UDP_PORT) + go mock.StartUnixDomainServer(consts.TEST_UNIX_DOMAIN_ADDR) + go mock.StartHttpServer(consts.TEST_HTTP_PORT) + + if err := runFrps(); err != nil { + panic(err) + } + time.Sleep(200 * time.Millisecond) + + if err := runFrpc(); err != nil { + panic(err) + } + if err := runFrpcVisitor(); err != nil { + panic(err) + } time.Sleep(500 * time.Millisecond) } +func runFrps() error { + p := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", "./config/auto_test_frps.ini"}) + return p.Start() +} + +func runFrpc() error { + p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./config/auto_test_frpc.ini"}) + return p.Start() +} + +func runFrpcVisitor() error { + p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./config/auto_test_frpc_visitor.ini"}) + return p.Start() +} + func TestTcp(t *testing.T) { assert := assert.New(t) // Normal - addr := fmt.Sprintf("127.0.0.1:%d", TEST_TCP_FRP_PORT) - res, err := sendTcpMsg(addr, TEST_TCP_ECHO_STR) + addr := fmt.Sprintf("127.0.0.1:%d", consts.TEST_TCP_FRP_PORT) + res, err := util.SendTcpMsg(addr, consts.TEST_TCP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_TCP_ECHO_STR, res) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) // Encrytion and compression - addr = fmt.Sprintf("127.0.0.1:%d", TEST_TCP_EC_FRP_PORT) - res, err = sendTcpMsg(addr, TEST_TCP_ECHO_STR) + addr = fmt.Sprintf("127.0.0.1:%d", consts.TEST_TCP_EC_FRP_PORT) + res, err = util.SendTcpMsg(addr, consts.TEST_TCP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_TCP_ECHO_STR, res) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) } func TestUdp(t *testing.T) { assert := assert.New(t) // Normal - addr := fmt.Sprintf("127.0.0.1:%d", TEST_UDP_FRP_PORT) - res, err := sendUdpMsg(addr, TEST_UDP_ECHO_STR) + addr := fmt.Sprintf("127.0.0.1:%d", consts.TEST_UDP_FRP_PORT) + res, err := util.SendUdpMsg(addr, consts.TEST_UDP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_UDP_ECHO_STR, res) + assert.Equal(consts.TEST_UDP_ECHO_STR, res) // Encrytion and compression - addr = fmt.Sprintf("127.0.0.1:%d", TEST_UDP_EC_FRP_PORT) - res, err = sendUdpMsg(addr, TEST_UDP_ECHO_STR) + addr = fmt.Sprintf("127.0.0.1:%d", consts.TEST_UDP_EC_FRP_PORT) + res, err = util.SendUdpMsg(addr, consts.TEST_UDP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_UDP_ECHO_STR, res) + assert.Equal(consts.TEST_UDP_ECHO_STR, res) } func TestUnixDomain(t *testing.T) { assert := assert.New(t) // Normal - addr := fmt.Sprintf("127.0.0.1:%d", TEST_UNIX_DOMAIN_FRP_PORT) - res, err := sendTcpMsg(addr, TEST_UNIX_DOMAIN_STR) + addr := fmt.Sprintf("127.0.0.1:%d", consts.TEST_UNIX_DOMAIN_FRP_PORT) + res, err := util.SendTcpMsg(addr, consts.TEST_UNIX_DOMAIN_STR) if assert.NoError(err) { - assert.Equal(TEST_UNIX_DOMAIN_STR, res) + assert.Equal(consts.TEST_UNIX_DOMAIN_STR, res) } } func TestStcp(t *testing.T) { assert := assert.New(t) // Normal - addr := fmt.Sprintf("127.0.0.1:%d", TEST_STCP_FRP_PORT) - res, err := sendTcpMsg(addr, TEST_STCP_ECHO_STR) + addr := fmt.Sprintf("127.0.0.1:%d", consts.TEST_STCP_FRP_PORT) + res, err := util.SendTcpMsg(addr, consts.TEST_STCP_ECHO_STR) if assert.NoError(err) { - assert.Equal(TEST_STCP_ECHO_STR, res) + assert.Equal(consts.TEST_STCP_ECHO_STR, res) } // Encrytion and compression - addr = fmt.Sprintf("127.0.0.1:%d", TEST_STCP_EC_FRP_PORT) - res, err = sendTcpMsg(addr, TEST_STCP_ECHO_STR) + addr = fmt.Sprintf("127.0.0.1:%d", consts.TEST_STCP_EC_FRP_PORT) + res, err = util.SendTcpMsg(addr, consts.TEST_STCP_ECHO_STR) if assert.NoError(err) { - assert.Equal(TEST_STCP_ECHO_STR, res) + assert.Equal(consts.TEST_STCP_ECHO_STR, res) } } func TestHttp(t *testing.T) { assert := assert.New(t) // web01 - code, body, _, err := sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "", nil, "") + code, body, _, err := util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "", nil, "") if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_NORMAL_STR, body) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, body) } // web02 - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test2.frp.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test2.frp.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_NORMAL_STR, body) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, body) } // error host header - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "errorhost.frp.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "errorhost.frp.com", nil, "") if assert.NoError(err) { assert.Equal(404, code) } // web03 - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test3.frp.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test3.frp.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_NORMAL_STR, body) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, body) } - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d/foo", TEST_HTTP_FRP_PORT), "test3.frp.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d/foo", consts.TEST_HTTP_FRP_PORT), "test3.frp.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_FOO_STR, body) + assert.Equal(consts.TEST_HTTP_FOO_STR, body) } // web04 - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d/bar", TEST_HTTP_FRP_PORT), "test3.frp.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d/bar", consts.TEST_HTTP_FRP_PORT), "test3.frp.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_BAR_STR, body) + assert.Equal(consts.TEST_HTTP_BAR_STR, body) } // web05 - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test5.frp.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test5.frp.com", nil, "") if assert.NoError(err) { assert.Equal(401, code) } headers := make(map[string]string) - headers["Authorization"] = basicAuth("test", "test") - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test5.frp.com", headers, "") + headers["Authorization"] = util.BasicAuth("test", "test") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test5.frp.com", headers, "") if assert.NoError(err) { assert.Equal(401, code) } // web06 var header http.Header - code, body, header, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test6.frp.com", nil, "") + code, body, header, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test6.frp.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_NORMAL_STR, body) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, body) assert.Equal("true", header.Get("X-Header-Set")) } // subhost01 - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test01.sub.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test01.sub.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) assert.Equal("test01.sub.com", body) } // subhost02 - code, body, _, err = sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), "test02.sub.com", nil, "") + code, body, _, err = util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "test02.sub.com", nil, "") if assert.NoError(err) { assert.Equal(200, code) assert.Equal("test02.sub.com", body) @@ -210,47 +195,47 @@ func TestHttp(t *testing.T) { func TestWebSocket(t *testing.T) { assert := assert.New(t) - u := url.URL{Scheme: "ws", Host: fmt.Sprintf("%s:%d", "127.0.0.1", TEST_HTTP_FRP_PORT), Path: "/ws"} + u := url.URL{Scheme: "ws", Host: fmt.Sprintf("%s:%d", "127.0.0.1", consts.TEST_HTTP_FRP_PORT), Path: "/ws"} c, _, err := websocket.DefaultDialer.Dial(u.String(), nil) assert.NoError(err) defer c.Close() - err = c.WriteMessage(websocket.TextMessage, []byte(TEST_HTTP_NORMAL_STR)) + err = c.WriteMessage(websocket.TextMessage, []byte(consts.TEST_HTTP_NORMAL_STR)) assert.NoError(err) _, msg, err := c.ReadMessage() assert.NoError(err) - assert.Equal(TEST_HTTP_NORMAL_STR, string(msg)) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, string(msg)) } func TestAllowPorts(t *testing.T) { assert := assert.New(t) // Port not allowed - status, err := getProxyStatus(ProxyTcpPortNotAllowed) + status, err := util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyTcpPortNotAllowed) if assert.NoError(err) { assert.Equal(client.ProxyStatusStartErr, status.Status) assert.True(strings.Contains(status.Err, ports.ErrPortNotAllowed.Error())) } - status, err = getProxyStatus(ProxyUdpPortNotAllowed) + status, err = util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyUdpPortNotAllowed) if assert.NoError(err) { assert.Equal(client.ProxyStatusStartErr, status.Status) assert.True(strings.Contains(status.Err, ports.ErrPortNotAllowed.Error())) } - status, err = getProxyStatus(ProxyTcpPortUnavailable) + status, err = util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyTcpPortUnavailable) if assert.NoError(err) { assert.Equal(client.ProxyStatusStartErr, status.Status) assert.True(strings.Contains(status.Err, ports.ErrPortUnAvailable.Error())) } // Port normal - status, err = getProxyStatus(ProxyTcpPortNormal) + status, err = util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyTcpPortNormal) if assert.NoError(err) { assert.Equal(client.ProxyStatusRunning, status.Status) } - status, err = getProxyStatus(ProxyUdpPortNormal) + status, err = util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyUdpPortNormal) if assert.NoError(err) { assert.Equal(client.ProxyStatusRunning, status.Status) } @@ -259,45 +244,45 @@ func TestAllowPorts(t *testing.T) { func TestRandomPort(t *testing.T) { assert := assert.New(t) // tcp - status, err := getProxyStatus(ProxyTcpRandomPort) + status, err := util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyTcpRandomPort) if assert.NoError(err) { addr := status.RemoteAddr - res, err := sendTcpMsg(addr, TEST_TCP_ECHO_STR) + res, err := util.SendTcpMsg(addr, consts.TEST_TCP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_TCP_ECHO_STR, res) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) } // udp - status, err = getProxyStatus(ProxyUdpRandomPort) + status, err = util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyUdpRandomPort) if assert.NoError(err) { addr := status.RemoteAddr - res, err := sendUdpMsg(addr, TEST_UDP_ECHO_STR) + res, err := util.SendUdpMsg(addr, consts.TEST_UDP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_UDP_ECHO_STR, res) + assert.Equal(consts.TEST_UDP_ECHO_STR, res) } } func TestPluginHttpProxy(t *testing.T) { assert := assert.New(t) - status, err := getProxyStatus(ProxyHttpProxy) + status, err := util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, consts.ProxyHttpProxy) if assert.NoError(err) { assert.Equal(client.ProxyStatusRunning, status.Status) // http proxy addr := status.RemoteAddr - code, body, _, err := sendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", TEST_HTTP_FRP_PORT), + code, body, _, err := util.SendHttpMsg("GET", fmt.Sprintf("http://127.0.0.1:%d", consts.TEST_HTTP_FRP_PORT), "", nil, "http://"+addr) if assert.NoError(err) { assert.Equal(200, code) - assert.Equal(TEST_HTTP_NORMAL_STR, body) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, body) } // connect method - conn, err := gnet.DialTcpByProxy("http://"+addr, fmt.Sprintf("127.0.0.1:%d", TEST_TCP_FRP_PORT)) + conn, err := gnet.DialTcpByProxy("http://"+addr, fmt.Sprintf("127.0.0.1:%d", consts.TEST_TCP_FRP_PORT)) if assert.NoError(err) { - res, err := sendTcpMsgByConn(conn, TEST_TCP_ECHO_STR) + res, err := util.SendTcpMsgByConn(conn, consts.TEST_TCP_ECHO_STR) assert.NoError(err) - assert.Equal(TEST_TCP_ECHO_STR, res) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) } } } @@ -306,8 +291,8 @@ func TestRangePortsMapping(t *testing.T) { assert := assert.New(t) for i := 0; i < 3; i++ { - name := fmt.Sprintf("%s_%d", ProxyRangeTcpPrefix, i) - status, err := getProxyStatus(name) + name := fmt.Sprintf("%s_%d", consts.ProxyRangeTcpPrefix, i) + status, err := util.GetProxyStatus(consts.ADMIN_ADDR, consts.ADMIN_USER, consts.ADMIN_PWD, name) if assert.NoError(err) { assert.Equal(client.ProxyStatusRunning, status.Status) } @@ -321,15 +306,15 @@ func TestGroup(t *testing.T) { p1 int p2 int ) - addr := fmt.Sprintf("127.0.0.1:%d", TEST_TCP2_FRP_PORT) + addr := fmt.Sprintf("127.0.0.1:%d", consts.TEST_TCP2_FRP_PORT) for i := 0; i < 6; i++ { - res, err := sendTcpMsg(addr, TEST_TCP_ECHO_STR) + res, err := util.SendTcpMsg(addr, consts.TEST_TCP_ECHO_STR) assert.NoError(err) switch res { - case TEST_TCP_ECHO_STR: + case consts.TEST_TCP_ECHO_STR: p1++ - case TEST_TCP_ECHO_STR + TEST_TCP_ECHO_STR: + case consts.TEST_TCP_ECHO_STR + consts.TEST_TCP_ECHO_STR: p2++ } } diff --git a/tests/echo_server.go b/tests/mock/echo_server.go similarity index 83% rename from tests/echo_server.go rename to tests/mock/echo_server.go index 380c036..a24947f 100644 --- a/tests/echo_server.go +++ b/tests/mock/echo_server.go @@ -1,4 +1,4 @@ -package tests +package mock import ( "fmt" @@ -10,8 +10,8 @@ import ( frpNet "github.com/fatedier/frp/utils/net" ) -func StartTcpEchoServer() { - l, err := frpNet.ListenTcp("127.0.0.1", TEST_TCP_PORT) +func StartTcpEchoServer(port int) { + l, err := frpNet.ListenTcp("127.0.0.1", port) if err != nil { fmt.Printf("echo server listen error: %v\n", err) return @@ -28,8 +28,8 @@ func StartTcpEchoServer() { } } -func StartTcpEchoServer2() { - l, err := frpNet.ListenTcp("127.0.0.1", TEST_TCP2_PORT) +func StartTcpEchoServer2(port int) { + l, err := frpNet.ListenTcp("127.0.0.1", port) if err != nil { fmt.Printf("echo server2 listen error: %v\n", err) return @@ -46,8 +46,8 @@ func StartTcpEchoServer2() { } } -func StartUdpEchoServer() { - l, err := frpNet.ListenUDP("127.0.0.1", TEST_UDP_PORT) +func StartUdpEchoServer(port int) { + l, err := frpNet.ListenUDP("127.0.0.1", port) if err != nil { fmt.Printf("udp echo server listen error: %v\n", err) return @@ -64,8 +64,7 @@ func StartUdpEchoServer() { } } -func StartUnixDomainServer() { - unixPath := TEST_UNIX_DOMAIN_ADDR +func StartUnixDomainServer(unixPath string) { os.Remove(unixPath) syscall.Umask(0) l, err := net.Listen("unix", unixPath) diff --git a/tests/http_server.go b/tests/mock/http_server.go similarity index 80% rename from tests/http_server.go rename to tests/mock/http_server.go index 640ae9a..7e97ad6 100644 --- a/tests/http_server.go +++ b/tests/mock/http_server.go @@ -1,4 +1,4 @@ -package tests +package mock import ( "fmt" @@ -7,15 +7,17 @@ import ( "regexp" "strings" + "github.com/fatedier/frp/tests/consts" + "github.com/gorilla/websocket" ) var upgrader = websocket.Upgrader{} -func StartHttpServer() { +func StartHttpServer(port int) { http.HandleFunc("/", handleHttp) http.HandleFunc("/ws", handleWebSocket) - http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", TEST_HTTP_PORT), nil) + http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", port), nil) } func handleWebSocket(w http.ResponseWriter, r *http.Request) { @@ -58,15 +60,15 @@ func handleHttp(w http.ResponseWriter, r *http.Request) { if strings.Contains(r.Host, "127.0.0.1") || strings.Contains(r.Host, "test2.frp.com") || strings.Contains(r.Host, "test5.frp.com") || strings.Contains(r.Host, "test6.frp.com") { w.WriteHeader(200) - w.Write([]byte(TEST_HTTP_NORMAL_STR)) + w.Write([]byte(consts.TEST_HTTP_NORMAL_STR)) } else if strings.Contains(r.Host, "test3.frp.com") { w.WriteHeader(200) if strings.Contains(r.URL.Path, "foo") { - w.Write([]byte(TEST_HTTP_FOO_STR)) + w.Write([]byte(consts.TEST_HTTP_FOO_STR)) } else if strings.Contains(r.URL.Path, "bar") { - w.Write([]byte(TEST_HTTP_BAR_STR)) + w.Write([]byte(consts.TEST_HTTP_BAR_STR)) } else { - w.Write([]byte(TEST_HTTP_NORMAL_STR)) + w.Write([]byte(consts.TEST_HTTP_NORMAL_STR)) } } else { w.WriteHeader(404) diff --git a/tests/run_test.sh b/tests/run_test.sh deleted file mode 100755 index a852a3d..0000000 --- a/tests/run_test.sh +++ /dev/null @@ -1,9 +0,0 @@ -#!/bin/bash - -./../bin/frps -c ./conf/auto_test_frps.ini & -sleep 1 -./../bin/frpc -c ./conf/auto_test_frpc.ini & -./../bin/frpc -c ./conf/auto_test_frpc_visitor.ini & - -# wait until proxies are connected -sleep 2 diff --git a/tests/util/process.go b/tests/util/process.go new file mode 100644 index 0000000..1e34040 --- /dev/null +++ b/tests/util/process.go @@ -0,0 +1,29 @@ +package util + +import ( + "context" + "os/exec" +) + +type Process struct { + cmd *exec.Cmd + cancel context.CancelFunc +} + +func NewProcess(path string, params []string) *Process { + ctx, cancel := context.WithCancel(context.Background()) + cmd := exec.CommandContext(ctx, path, params...) + return &Process{ + cmd: cmd, + cancel: cancel, + } +} + +func (p *Process) Start() error { + return p.cmd.Start() +} + +func (p *Process) Stop() error { + p.cancel() + return p.cmd.Wait() +} diff --git a/tests/util.go b/tests/util/util.go similarity index 86% rename from tests/util.go rename to tests/util/util.go index 5bc3e22..ac314ec 100644 --- a/tests/util.go +++ b/tests/util/util.go @@ -1,4 +1,4 @@ -package tests +package util import ( "encoding/base64" @@ -16,13 +16,13 @@ import ( frpNet "github.com/fatedier/frp/utils/net" ) -func getProxyStatus(name string) (status *client.ProxyStatusResp, err error) { - req, err := http.NewRequest("GET", "http://"+ADMIN_ADDR+"/api/status", nil) +func GetProxyStatus(statusAddr string, user string, passwd string, name string) (status *client.ProxyStatusResp, err error) { + req, err := http.NewRequest("GET", "http://"+statusAddr+"/api/status", nil) if err != nil { return status, err } - authStr := "Basic " + base64.StdEncoding.EncodeToString([]byte(ADMIN_USER+":"+ADMIN_PWD)) + authStr := "Basic " + base64.StdEncoding.EncodeToString([]byte(user+":"+passwd)) req.Header.Add("Authorization", authStr) resp, err := http.DefaultClient.Do(req) if err != nil { @@ -75,17 +75,17 @@ func getProxyStatus(name string) (status *client.ProxyStatusResp, err error) { return status, errors.New("no proxy status found") } -func sendTcpMsg(addr string, msg string) (res string, err error) { +func SendTcpMsg(addr string, msg string) (res string, err error) { c, err := frpNet.ConnectTcpServer(addr) if err != nil { err = fmt.Errorf("connect to tcp server error: %v", err) return } defer c.Close() - return sendTcpMsgByConn(c, msg) + return SendTcpMsgByConn(c, msg) } -func sendTcpMsgByConn(c net.Conn, msg string) (res string, err error) { +func SendTcpMsgByConn(c net.Conn, msg string) (res string, err error) { timer := time.Now().Add(5 * time.Second) c.SetDeadline(timer) c.Write([]byte(msg)) @@ -99,7 +99,7 @@ func sendTcpMsgByConn(c net.Conn, msg string) (res string, err error) { return string(buf[:n]), nil } -func sendUdpMsg(addr string, msg string) (res string, err error) { +func SendUdpMsg(addr string, msg string) (res string, err error) { udpAddr, errRet := net.ResolveUDPAddr("udp", addr) if errRet != nil { err = fmt.Errorf("resolve udp addr error: %v", err) @@ -126,7 +126,7 @@ func sendUdpMsg(addr string, msg string) (res string, err error) { return string(buf[:n]), nil } -func sendHttpMsg(method, urlStr string, host string, headers map[string]string, proxy string) (code int, body string, header http.Header, err error) { +func SendHttpMsg(method, urlStr string, host string, headers map[string]string, proxy string) (code int, body string, header http.Header, err error) { req, errRet := http.NewRequest(method, urlStr, nil) if errRet != nil { err = errRet @@ -177,7 +177,7 @@ func sendHttpMsg(method, urlStr string, host string, headers map[string]string, return } -func basicAuth(username, passwd string) string { +func BasicAuth(username, passwd string) string { auth := username + ":" + passwd return "Basic " + base64.StdEncoding.EncodeToString([]byte(auth)) } From d74b45be5d037b58d18fec3be05064c3ec68e404 Mon Sep 17 00:00:00 2001 From: fatedier Date: Thu, 12 Jul 2018 00:31:21 +0800 Subject: [PATCH 04/10] more ci --- tests/{config => ci}/auto_test_frpc.ini | 0 .../{config => ci}/auto_test_frpc_visitor.ini | 0 tests/{config => ci}/auto_test_frps.ini | 0 tests/ci/cmd_test.go | 1 + tests/{func_test.go => ci/normal_test.go} | 8 +- tests/ci/reconnect_test.go | 1 + tests/ci/reload_test.go | 108 ++++++++++++++++++ tests/config/config.go | 10 +- tests/consts/consts.go | 4 +- 9 files changed, 123 insertions(+), 9 deletions(-) rename tests/{config => ci}/auto_test_frpc.ini (100%) rename tests/{config => ci}/auto_test_frpc_visitor.ini (100%) rename tests/{config => ci}/auto_test_frps.ini (100%) create mode 100644 tests/ci/cmd_test.go rename tests/{func_test.go => ci/normal_test.go} (97%) create mode 100644 tests/ci/reconnect_test.go create mode 100644 tests/ci/reload_test.go diff --git a/tests/config/auto_test_frpc.ini b/tests/ci/auto_test_frpc.ini similarity index 100% rename from tests/config/auto_test_frpc.ini rename to tests/ci/auto_test_frpc.ini diff --git a/tests/config/auto_test_frpc_visitor.ini b/tests/ci/auto_test_frpc_visitor.ini similarity index 100% rename from tests/config/auto_test_frpc_visitor.ini rename to tests/ci/auto_test_frpc_visitor.ini diff --git a/tests/config/auto_test_frps.ini b/tests/ci/auto_test_frps.ini similarity index 100% rename from tests/config/auto_test_frps.ini rename to tests/ci/auto_test_frps.ini diff --git a/tests/ci/cmd_test.go b/tests/ci/cmd_test.go new file mode 100644 index 0000000..006e04c --- /dev/null +++ b/tests/ci/cmd_test.go @@ -0,0 +1 @@ +package ci diff --git a/tests/func_test.go b/tests/ci/normal_test.go similarity index 97% rename from tests/func_test.go rename to tests/ci/normal_test.go index a83a6dc..1ee0f58 100644 --- a/tests/func_test.go +++ b/tests/ci/normal_test.go @@ -1,4 +1,4 @@ -package tests +package ci import ( "fmt" @@ -42,17 +42,17 @@ func init() { } func runFrps() error { - p := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", "./config/auto_test_frps.ini"}) + p := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", "./auto_test_frps.ini"}) return p.Start() } func runFrpc() error { - p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./config/auto_test_frpc.ini"}) + p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./auto_test_frpc.ini"}) return p.Start() } func runFrpcVisitor() error { - p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./config/auto_test_frpc_visitor.ini"}) + p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./auto_test_frpc_visitor.ini"}) return p.Start() } diff --git a/tests/ci/reconnect_test.go b/tests/ci/reconnect_test.go new file mode 100644 index 0000000..006e04c --- /dev/null +++ b/tests/ci/reconnect_test.go @@ -0,0 +1 @@ +package ci diff --git a/tests/ci/reload_test.go b/tests/ci/reload_test.go new file mode 100644 index 0000000..0a89f89 --- /dev/null +++ b/tests/ci/reload_test.go @@ -0,0 +1,108 @@ +package ci + +import ( + "os" + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/fatedier/frp/tests/config" + "github.com/fatedier/frp/tests/consts" + "github.com/fatedier/frp/tests/util" +) + +const FRPS_CONF = ` +[common] +server_addr = 127.0.0.1 +server_port = 10700 +log_file = console +# debug, info, warn, error +log_level = debug +token = 123456 +admin_port = 10600 +admin_user = abc +admin_pwd = abc +` + +const FRPC_CONF_1 = ` +[common] +server_addr = 127.0.0.1 +server_port = 20000 +log_file = console +# debug, info, warn, error +log_level = debug +token = 123456 +admin_port = 21000 +admin_user = abc +admin_pwd = abc + +[tcp] +type = tcp +local_port = 10701 +remote_port = 20801 + +# change remote port +[tcp2] +type = tcp +local_port = 10701 +remote_port = 20802 + +# delete +[tcp3] +type = tcp +local_port = 10701 +remote_port = 20803 +` + +const FRPC_CONF_2 = ` +[common] +server_addr = 127.0.0.1 +server_port = 20000 +log_file = console +# debug, info, warn, error +log_level = debug +token = 123456 +admin_port = 21000 +admin_user = abc +admin_pwd = abc + +[tcp] +type = tcp +local_port = 10701 +remote_port = 20801 + +[tcp2] +type = tcp +local_port = 10701 +remote_port = 20902 +` + +func TestReload(t *testing.T) { + assert := assert.New(t) + frpsCfgPath, err := config.GenerateConfigFile("./auto_test_frps.ini", FRPS_CONF) + if assert.NoError(err) { + defer os.Remove(frpsCfgPath) + } + + frpcCfgPath, err := config.GenerateConfigFile("./auto_test_frpc.ini", FRPC_CONF_1) + if assert.NoError(err) { + defer os.Remove(frpcCfgPath) + } + + frpsProcess := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", frpsCfgPath}) + err = frpsProcess.Start() + if assert.NoError(err) { + defer frpsProcess.Stop() + } + + time.Sleep(200 * time.Millisecond) + + frpcProcess := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", frpcCfgPath}) + err = frpcProcess.Start() + if assert.NoError(err) { + defer frpcProcess.Stop() + } + + // TODO +} diff --git a/tests/config/config.go b/tests/config/config.go index 80e096b..ac09446 100644 --- a/tests/config/config.go +++ b/tests/config/config.go @@ -1,9 +1,13 @@ -package util +package config import ( "io/ioutil" + "os" + "path/filepath" ) -func GenerateConfigFile(path string, content string) error { - return ioutil.WriteFile(path, []byte(content), 0666) +func GenerateConfigFile(path string, content string) (realPath string, err error) { + realPath = filepath.Join(os.TempDir(), path) + err = ioutil.WriteFile(realPath, []byte(content), 0666) + return realPath, err } diff --git a/tests/consts/consts.go b/tests/consts/consts.go index 7e70c89..9deae3a 100644 --- a/tests/consts/consts.go +++ b/tests/consts/consts.go @@ -3,8 +3,8 @@ package consts import "path/filepath" var ( - FRPS_BIN_PATH = "../bin/frps" - FRPC_BIN_PATH = "../bin/frpc" + FRPS_BIN_PATH = "../../bin/frps" + FRPC_BIN_PATH = "../../bin/frpc" SERVER_ADDR = "127.0.0.1" ADMIN_ADDR = "127.0.0.1:10600" From 57417c83ae8695e43e8b0cb2f7ba88660a3eb8e5 Mon Sep 17 00:00:00 2001 From: fatedier Date: Thu, 12 Jul 2018 15:23:34 +0800 Subject: [PATCH 05/10] add ci case of reload and reconnect --- tests/ci/cmd_test.go | 2 + tests/ci/normal_test.go | 35 +++++------ tests/ci/reconnect_test.go | 116 +++++++++++++++++++++++++++++++++++++ tests/ci/reload_test.go | 60 +++++++++++++++---- tests/consts/consts.go | 3 + tests/util/util.go | 22 +++++++ 6 files changed, 208 insertions(+), 30 deletions(-) diff --git a/tests/ci/cmd_test.go b/tests/ci/cmd_test.go index 006e04c..7780397 100644 --- a/tests/ci/cmd_test.go +++ b/tests/ci/cmd_test.go @@ -1 +1,3 @@ package ci + +// TODO diff --git a/tests/ci/normal_test.go b/tests/ci/normal_test.go index 1ee0f58..c528346 100644 --- a/tests/ci/normal_test.go +++ b/tests/ci/normal_test.go @@ -4,6 +4,7 @@ import ( "fmt" "net/http" "net/url" + "os" "strings" "testing" "time" @@ -20,40 +21,36 @@ import ( gnet "github.com/fatedier/golib/net" ) -func init() { +func TestMain(m *testing.M) { go mock.StartTcpEchoServer(consts.TEST_TCP_PORT) go mock.StartTcpEchoServer2(consts.TEST_TCP2_PORT) go mock.StartUdpEchoServer(consts.TEST_UDP_PORT) go mock.StartUnixDomainServer(consts.TEST_UNIX_DOMAIN_ADDR) go mock.StartHttpServer(consts.TEST_HTTP_PORT) - if err := runFrps(); err != nil { + var err error + p1 := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", "./auto_test_frps.ini"}) + if err = p1.Start(); err != nil { panic(err) } - time.Sleep(200 * time.Millisecond) - if err := runFrpc(); err != nil { + time.Sleep(200 * time.Millisecond) + p2 := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./auto_test_frpc.ini"}) + if err = p2.Start(); err != nil { panic(err) } - if err := runFrpcVisitor(); err != nil { + + p3 := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./auto_test_frpc_visitor.ini"}) + if err = p3.Start(); err != nil { panic(err) } time.Sleep(500 * time.Millisecond) -} -func runFrps() error { - p := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", "./auto_test_frps.ini"}) - return p.Start() -} - -func runFrpc() error { - p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./auto_test_frpc.ini"}) - return p.Start() -} - -func runFrpcVisitor() error { - p := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", "./auto_test_frpc_visitor.ini"}) - return p.Start() + exitCode := m.Run() + p1.Stop() + p2.Stop() + p3.Stop() + os.Exit(exitCode) } func TestTcp(t *testing.T) { diff --git a/tests/ci/reconnect_test.go b/tests/ci/reconnect_test.go index 006e04c..2378a08 100644 --- a/tests/ci/reconnect_test.go +++ b/tests/ci/reconnect_test.go @@ -1 +1,117 @@ package ci + +import ( + "os" + "testing" + "time" + + "github.com/fatedier/frp/tests/config" + "github.com/fatedier/frp/tests/consts" + "github.com/fatedier/frp/tests/util" + + "github.com/stretchr/testify/assert" +) + +const FRPS_RECONNECT_CONF = ` +[common] +bind_addr = 0.0.0.0 +bind_port = 20000 +log_file = console +# debug, info, warn, error +log_level = debug +token = 123456 +` + +const FRPC_RECONNECT_CONF = ` +[common] +server_addr = 127.0.0.1 +server_port = 20000 +log_file = console +# debug, info, warn, error +log_level = debug +token = 123456 +admin_port = 21000 +admin_user = abc +admin_pwd = abc + +[tcp] +type = tcp +local_port = 10701 +remote_port = 20801 +` + +func TestReconnect(t *testing.T) { + assert := assert.New(t) + frpsCfgPath, err := config.GenerateConfigFile(consts.FRPS_NORMAL_CONFIG, FRPS_RECONNECT_CONF) + if assert.NoError(err) { + defer os.Remove(frpsCfgPath) + } + + frpcCfgPath, err := config.GenerateConfigFile(consts.FRPC_NORMAL_CONFIG, FRPC_RECONNECT_CONF) + if assert.NoError(err) { + defer os.Remove(frpcCfgPath) + } + + frpsProcess := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", frpsCfgPath}) + err = frpsProcess.Start() + if assert.NoError(err) { + defer frpsProcess.Stop() + } + + time.Sleep(200 * time.Millisecond) + + frpcProcess := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", frpcCfgPath}) + err = frpcProcess.Start() + if assert.NoError(err) { + defer frpcProcess.Stop() + } + time.Sleep(250 * time.Millisecond) + + // test tcp + res, err := util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // stop frpc + frpcProcess.Stop() + time.Sleep(100 * time.Millisecond) + + // test tcp, expect failed + _, err = util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.Error(err) + + // restart frpc + newFrpcProcess := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", frpcCfgPath}) + err = newFrpcProcess.Start() + if assert.NoError(err) { + defer newFrpcProcess.Stop() + } + time.Sleep(250 * time.Millisecond) + + // test tcp + res, err = util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // stop frps + frpsProcess.Stop() + time.Sleep(100 * time.Millisecond) + + // test tcp, expect failed + _, err = util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.Error(err) + + // restart frps + newFrpsProcess := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-c", frpsCfgPath}) + err = newFrpsProcess.Start() + if assert.NoError(err) { + defer newFrpsProcess.Stop() + } + + time.Sleep(2 * time.Second) + + // test tcp + res, err = util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) +} diff --git a/tests/ci/reload_test.go b/tests/ci/reload_test.go index 0a89f89..5ba358a 100644 --- a/tests/ci/reload_test.go +++ b/tests/ci/reload_test.go @@ -12,20 +12,17 @@ import ( "github.com/fatedier/frp/tests/util" ) -const FRPS_CONF = ` +const FRPS_RELOAD_CONF = ` [common] -server_addr = 127.0.0.1 -server_port = 10700 +bind_addr = 0.0.0.0 +bind_port = 20000 log_file = console # debug, info, warn, error log_level = debug token = 123456 -admin_port = 10600 -admin_user = abc -admin_pwd = abc ` -const FRPC_CONF_1 = ` +const FRPC_RELOAD_CONF_1 = ` [common] server_addr = 127.0.0.1 server_port = 20000 @@ -55,7 +52,7 @@ local_port = 10701 remote_port = 20803 ` -const FRPC_CONF_2 = ` +const FRPC_RELOAD_CONF_2 = ` [common] server_addr = 127.0.0.1 server_port = 20000 @@ -80,12 +77,12 @@ remote_port = 20902 func TestReload(t *testing.T) { assert := assert.New(t) - frpsCfgPath, err := config.GenerateConfigFile("./auto_test_frps.ini", FRPS_CONF) + frpsCfgPath, err := config.GenerateConfigFile(consts.FRPS_NORMAL_CONFIG, FRPS_RELOAD_CONF) if assert.NoError(err) { defer os.Remove(frpsCfgPath) } - frpcCfgPath, err := config.GenerateConfigFile("./auto_test_frpc.ini", FRPC_CONF_1) + frpcCfgPath, err := config.GenerateConfigFile(consts.FRPC_NORMAL_CONFIG, FRPC_RELOAD_CONF_1) if assert.NoError(err) { defer os.Remove(frpcCfgPath) } @@ -104,5 +101,46 @@ func TestReload(t *testing.T) { defer frpcProcess.Stop() } - // TODO + time.Sleep(250 * time.Millisecond) + + // test tcp1 + res, err := util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // test tcp2 + res, err = util.SendTcpMsg("127.0.0.1:20802", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // test tcp3 + res, err = util.SendTcpMsg("127.0.0.1:20803", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // reload frpc config + frpcCfgPath, err = config.GenerateConfigFile(consts.FRPC_NORMAL_CONFIG, FRPC_RELOAD_CONF_2) + assert.NoError(err) + err = util.ReloadConf("127.0.0.1:21000", "abc", "abc") + assert.NoError(err) + + time.Sleep(time.Second) + + // test tcp1 + res, err = util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // test origin tcp2, expect failed + res, err = util.SendTcpMsg("127.0.0.1:20802", consts.TEST_TCP_ECHO_STR) + assert.Error(err) + + // test new origin tcp2 with different port + res, err = util.SendTcpMsg("127.0.0.1:20902", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) + + // test tcp3, expect failed + res, err = util.SendTcpMsg("127.0.0.1:20803", consts.TEST_TCP_ECHO_STR) + assert.Error(err) } diff --git a/tests/consts/consts.go b/tests/consts/consts.go index 9deae3a..60dcffe 100644 --- a/tests/consts/consts.go +++ b/tests/consts/consts.go @@ -6,6 +6,9 @@ var ( FRPS_BIN_PATH = "../../bin/frps" FRPC_BIN_PATH = "../../bin/frpc" + FRPS_NORMAL_CONFIG = "./auto_test_frps.ini" + FRPC_NORMAL_CONFIG = "./auto_test_frpc.ini" + SERVER_ADDR = "127.0.0.1" ADMIN_ADDR = "127.0.0.1:10600" ADMIN_USER = "abc" diff --git a/tests/util/util.go b/tests/util/util.go index ac314ec..2070ce3 100644 --- a/tests/util/util.go +++ b/tests/util/util.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "io/ioutil" "net" "net/http" @@ -75,6 +76,27 @@ func GetProxyStatus(statusAddr string, user string, passwd string, name string) return status, errors.New("no proxy status found") } +func ReloadConf(reloadAddr string, user string, passwd string) error { + req, err := http.NewRequest("GET", "http://"+reloadAddr+"/api/reload", nil) + if err != nil { + return err + } + + authStr := "Basic " + base64.StdEncoding.EncodeToString([]byte(user+":"+passwd)) + req.Header.Add("Authorization", authStr) + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } else { + if resp.StatusCode != 200 { + return fmt.Errorf("admin api status code [%d]", resp.StatusCode) + } + defer resp.Body.Close() + io.Copy(ioutil.Discard, resp.Body) + } + return nil +} + func SendTcpMsg(addr string, msg string) (res string, err error) { c, err := frpNet.ConnectTcpServer(addr) if err != nil { From cc6486addbafaed64274d898e2d1f3c9e06e124b Mon Sep 17 00:00:00 2001 From: fatedier Date: Thu, 12 Jul 2018 16:49:16 +0800 Subject: [PATCH 06/10] add more cmd test --- tests/ci/cmd_test.go | 84 +++++++++++++++++++++++++++++++++++++- tests/ci/reconnect_test.go | 2 +- tests/ci/reload_test.go | 2 +- 3 files changed, 85 insertions(+), 3 deletions(-) diff --git a/tests/ci/cmd_test.go b/tests/ci/cmd_test.go index 7780397..ea76e6d 100644 --- a/tests/ci/cmd_test.go +++ b/tests/ci/cmd_test.go @@ -1,3 +1,85 @@ package ci -// TODO +import ( + "testing" + "time" + + "github.com/fatedier/frp/tests/consts" + "github.com/fatedier/frp/tests/util" + + "github.com/stretchr/testify/assert" +) + +func TestCmdTcp(t *testing.T) { + assert := assert.New(t) + + var err error + s := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-t", "123", "-p", "20000"}) + err = s.Start() + if assert.NoError(err) { + defer s.Stop() + } + time.Sleep(100 * time.Millisecond) + + c := util.NewProcess(consts.FRPC_BIN_PATH, []string{"tcp", "-s", "127.0.0.1:20000", "-t", "123", "-u", "test", + "-l", "10701", "-r", "20801", "-n", "tcp_test"}) + err = c.Start() + if assert.NoError(err) { + defer c.Stop() + } + time.Sleep(250 * time.Millisecond) + + res, err := util.SendTcpMsg("127.0.0.1:20801", consts.TEST_TCP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_TCP_ECHO_STR, res) +} + +func TestCmdUdp(t *testing.T) { + assert := assert.New(t) + + var err error + s := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-t", "123", "-p", "20000"}) + err = s.Start() + if assert.NoError(err) { + defer s.Stop() + } + time.Sleep(100 * time.Millisecond) + + c := util.NewProcess(consts.FRPC_BIN_PATH, []string{"udp", "-s", "127.0.0.1:20000", "-t", "123", "-u", "test", + "-l", "10702", "-r", "20802", "-n", "udp_test"}) + err = c.Start() + if assert.NoError(err) { + defer c.Stop() + } + time.Sleep(250 * time.Millisecond) + + res, err := util.SendUdpMsg("127.0.0.1:20802", consts.TEST_UDP_ECHO_STR) + assert.NoError(err) + assert.Equal(consts.TEST_UDP_ECHO_STR, res) +} + +func TestCmdHttp(t *testing.T) { + assert := assert.New(t) + + var err error + s := util.NewProcess(consts.FRPS_BIN_PATH, []string{"-t", "123", "-p", "20000", "--vhost_http_port", "20001"}) + err = s.Start() + if assert.NoError(err) { + defer s.Stop() + } + time.Sleep(100 * time.Millisecond) + + c := util.NewProcess(consts.FRPC_BIN_PATH, []string{"http", "-s", "127.0.0.1:20000", "-t", "123", "-u", "test", + "-n", "udp_test", "-l", "10704", "--custom_domain", "127.0.0.1"}) + err = c.Start() + if assert.NoError(err) { + defer c.Stop() + } + time.Sleep(250 * time.Millisecond) + + code, body, _, err := util.SendHttpMsg("GET", "http://127.0.0.1:20001", "", nil, "") + if assert.NoError(err) { + assert.Equal(200, code) + assert.Equal(consts.TEST_HTTP_NORMAL_STR, body) + } +} diff --git a/tests/ci/reconnect_test.go b/tests/ci/reconnect_test.go index 2378a08..7974c2c 100644 --- a/tests/ci/reconnect_test.go +++ b/tests/ci/reconnect_test.go @@ -58,7 +58,7 @@ func TestReconnect(t *testing.T) { defer frpsProcess.Stop() } - time.Sleep(200 * time.Millisecond) + time.Sleep(100 * time.Millisecond) frpcProcess := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", frpcCfgPath}) err = frpcProcess.Start() diff --git a/tests/ci/reload_test.go b/tests/ci/reload_test.go index 5ba358a..9811db9 100644 --- a/tests/ci/reload_test.go +++ b/tests/ci/reload_test.go @@ -93,7 +93,7 @@ func TestReload(t *testing.T) { defer frpsProcess.Stop() } - time.Sleep(200 * time.Millisecond) + time.Sleep(100 * time.Millisecond) frpcProcess := util.NewProcess(consts.FRPC_BIN_PATH, []string{"-c", frpcCfgPath}) err = frpcProcess.Start() From 082447f5170445e0bc69865fa37d756646c17511 Mon Sep 17 00:00:00 2001 From: fatedier Date: Mon, 16 Jul 2018 01:21:29 +0800 Subject: [PATCH 07/10] frpc: support health check --- client/health.go | 123 +++++++++++++++++++++++++++++++++++++++-- models/config/proxy.go | 2 + 2 files changed, 121 insertions(+), 4 deletions(-) diff --git a/client/health.go b/client/health.go index ad58554..8e84a6f 100644 --- a/client/health.go +++ b/client/health.go @@ -15,18 +15,133 @@ package client import ( - "github.com/fatedier/frp/models/config" + "context" + "net" + "net/http" + "time" ) type HealthCheckMonitor struct { - cfg config.HealthCheckConf + checkType string + interval time.Duration + timeout time.Duration + maxFailedTimes int + + // For tcp + addr string + + // For http + url string + + failedTimes uint64 + statusOK bool + statusNormalFn func() + statusFailedFn func() + + ctx context.Context + cancel context.CancelFunc } -func NewHealthCheckMonitor(cfg *config.HealthCheckConf) *HealthCheckMonitor { +func NewHealthCheckMonitor(checkType string, intervalS int, timeoutS int, maxFailedTimes int, addr string, url string, + statusNormalFn func(), statusFailedFn func()) *HealthCheckMonitor { + + if intervalS <= 0 { + intervalS = 10 + } + if timeoutS <= 0 { + timeoutS = 3 + } + if maxFailedTimes <= 0 { + maxFailedTimes = 1 + } + ctx, cancel := context.WithCancel(context.Background()) return &HealthCheckMonitor{ - cfg: *cfg, + checkType: checkType, + interval: time.Duration(intervalS) * time.Second, + timeout: time.Duration(timeoutS) * time.Second, + maxFailedTimes: maxFailedTimes, + addr: addr, + url: url, + statusOK: false, + statusNormalFn: statusNormalFn, + statusFailedFn: statusFailedFn, + ctx: ctx, + cancel: cancel, } } func (monitor *HealthCheckMonitor) Start() { + go monitor.checkWorker() +} + +func (monitor *HealthCheckMonitor) Stop() { + monitor.cancel() +} + +func (monitor *HealthCheckMonitor) checkWorker() { + for { + ctx, cancel := context.WithDeadline(monitor.ctx, time.Now().Add(monitor.timeout)) + ok := monitor.doCheck(ctx) + + // check if this monitor has been closed + select { + case <-ctx.Done(): + cancel() + return + default: + cancel() + } + + if ok { + if !monitor.statusOK && monitor.statusNormalFn != nil { + monitor.statusOK = true + monitor.statusNormalFn() + } + } else { + monitor.failedTimes++ + if monitor.statusOK && int(monitor.failedTimes) >= monitor.maxFailedTimes && monitor.statusFailedFn != nil { + monitor.statusOK = false + monitor.statusFailedFn() + } + } + + time.Sleep(monitor.interval) + } +} + +func (monitor *HealthCheckMonitor) doCheck(ctx context.Context) bool { + switch monitor.checkType { + case "tcp": + return monitor.doTcpCheck(ctx) + case "http": + return monitor.doHttpCheck(ctx) + default: + return false + } +} + +func (monitor *HealthCheckMonitor) doTcpCheck(ctx context.Context) bool { + var d net.Dialer + conn, err := d.DialContext(ctx, "tcp", monitor.addr) + if err != nil { + return false + } + conn.Close() + return true +} + +func (monitor *HealthCheckMonitor) doHttpCheck(ctx context.Context) bool { + req, err := http.NewRequest("GET", monitor.url, nil) + if err != nil { + return false + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return false + } + + if resp.StatusCode/100 != 2 { + return false + } + return true } diff --git a/models/config/proxy.go b/models/config/proxy.go index b600be5..9ea680a 100644 --- a/models/config/proxy.go +++ b/models/config/proxy.go @@ -381,6 +381,8 @@ func (cfg *LocalSvrConf) checkForCli() (err error) { // Health check info type HealthCheckConf struct { HealthCheckType string `json:"health_check_type"` // tcp | http + HealthCheckTimeout int `json:"health_check_timeout"` + HealthCheckMaxFailed int `json:"health_check_max_failed"` HealthCheckIntervalS int `json:"health_check_interval_s"` HealthCheckUrl string `json:"health_check_url"` From 698219b6218897241aef7e148e2e611f9b8c6ccf Mon Sep 17 00:00:00 2001 From: fatedier Date: Mon, 16 Jul 2018 01:21:29 +0800 Subject: [PATCH 08/10] frpc: support health check --- client/control.go | 1 + client/health.go | 123 ++++++++++++++++++++++++++++++++++++++-- client/proxy_manager.go | 1 - models/config/proxy.go | 2 + 4 files changed, 122 insertions(+), 5 deletions(-) diff --git a/client/control.go b/client/control.go index 04be13c..90d2d4a 100644 --- a/client/control.go +++ b/client/control.go @@ -255,6 +255,7 @@ func (ctl *Control) login() (err error) { return nil } +// connectServer return a new connection to frps func (ctl *Control) connectServer() (conn frpNet.Conn, err error) { if g.GlbClientCfg.TcpMux { stream, errRet := ctl.session.OpenStream() diff --git a/client/health.go b/client/health.go index ad58554..8e84a6f 100644 --- a/client/health.go +++ b/client/health.go @@ -15,18 +15,133 @@ package client import ( - "github.com/fatedier/frp/models/config" + "context" + "net" + "net/http" + "time" ) type HealthCheckMonitor struct { - cfg config.HealthCheckConf + checkType string + interval time.Duration + timeout time.Duration + maxFailedTimes int + + // For tcp + addr string + + // For http + url string + + failedTimes uint64 + statusOK bool + statusNormalFn func() + statusFailedFn func() + + ctx context.Context + cancel context.CancelFunc } -func NewHealthCheckMonitor(cfg *config.HealthCheckConf) *HealthCheckMonitor { +func NewHealthCheckMonitor(checkType string, intervalS int, timeoutS int, maxFailedTimes int, addr string, url string, + statusNormalFn func(), statusFailedFn func()) *HealthCheckMonitor { + + if intervalS <= 0 { + intervalS = 10 + } + if timeoutS <= 0 { + timeoutS = 3 + } + if maxFailedTimes <= 0 { + maxFailedTimes = 1 + } + ctx, cancel := context.WithCancel(context.Background()) return &HealthCheckMonitor{ - cfg: *cfg, + checkType: checkType, + interval: time.Duration(intervalS) * time.Second, + timeout: time.Duration(timeoutS) * time.Second, + maxFailedTimes: maxFailedTimes, + addr: addr, + url: url, + statusOK: false, + statusNormalFn: statusNormalFn, + statusFailedFn: statusFailedFn, + ctx: ctx, + cancel: cancel, } } func (monitor *HealthCheckMonitor) Start() { + go monitor.checkWorker() +} + +func (monitor *HealthCheckMonitor) Stop() { + monitor.cancel() +} + +func (monitor *HealthCheckMonitor) checkWorker() { + for { + ctx, cancel := context.WithDeadline(monitor.ctx, time.Now().Add(monitor.timeout)) + ok := monitor.doCheck(ctx) + + // check if this monitor has been closed + select { + case <-ctx.Done(): + cancel() + return + default: + cancel() + } + + if ok { + if !monitor.statusOK && monitor.statusNormalFn != nil { + monitor.statusOK = true + monitor.statusNormalFn() + } + } else { + monitor.failedTimes++ + if monitor.statusOK && int(monitor.failedTimes) >= monitor.maxFailedTimes && monitor.statusFailedFn != nil { + monitor.statusOK = false + monitor.statusFailedFn() + } + } + + time.Sleep(monitor.interval) + } +} + +func (monitor *HealthCheckMonitor) doCheck(ctx context.Context) bool { + switch monitor.checkType { + case "tcp": + return monitor.doTcpCheck(ctx) + case "http": + return monitor.doHttpCheck(ctx) + default: + return false + } +} + +func (monitor *HealthCheckMonitor) doTcpCheck(ctx context.Context) bool { + var d net.Dialer + conn, err := d.DialContext(ctx, "tcp", monitor.addr) + if err != nil { + return false + } + conn.Close() + return true +} + +func (monitor *HealthCheckMonitor) doHttpCheck(ctx context.Context) bool { + req, err := http.NewRequest("GET", monitor.url, nil) + if err != nil { + return false + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return false + } + + if resp.StatusCode/100 != 2 { + return false + } + return true } diff --git a/client/proxy_manager.go b/client/proxy_manager.go index cfa56fc..bd193bb 100644 --- a/client/proxy_manager.go +++ b/client/proxy_manager.go @@ -18,7 +18,6 @@ const ( ProxyStatusWaitStart = "wait start" ProxyStatusRunning = "running" ProxyStatusCheckFailed = "check failed" - ProxyStatusCheckSuccess = "check success" ProxyStatusClosed = "closed" ) diff --git a/models/config/proxy.go b/models/config/proxy.go index b600be5..9ea680a 100644 --- a/models/config/proxy.go +++ b/models/config/proxy.go @@ -381,6 +381,8 @@ func (cfg *LocalSvrConf) checkForCli() (err error) { // Health check info type HealthCheckConf struct { HealthCheckType string `json:"health_check_type"` // tcp | http + HealthCheckTimeout int `json:"health_check_timeout"` + HealthCheckMaxFailed int `json:"health_check_max_failed"` HealthCheckIntervalS int `json:"health_check_interval_s"` HealthCheckUrl string `json:"health_check_url"` From 1a8ac148ca8ff25b72b388e563b37fbbdfa4bffc Mon Sep 17 00:00:00 2001 From: fatedier Date: Thu, 18 Oct 2018 13:55:51 +0800 Subject: [PATCH 09/10] fix xtcp visitor panic --- client/proxy_manager.go | 12 ++++++------ client/visitor.go | 9 +++++++++ 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/client/proxy_manager.go b/client/proxy_manager.go index bd193bb..fe175a0 100644 --- a/client/proxy_manager.go +++ b/client/proxy_manager.go @@ -13,12 +13,12 @@ import ( ) const ( - ProxyStatusNew = "new" - ProxyStatusStartErr = "start error" - ProxyStatusWaitStart = "wait start" - ProxyStatusRunning = "running" - ProxyStatusCheckFailed = "check failed" - ProxyStatusClosed = "closed" + ProxyStatusNew = "new" + ProxyStatusStartErr = "start error" + ProxyStatusWaitStart = "wait start" + ProxyStatusRunning = "running" + ProxyStatusCheckFailed = "check failed" + ProxyStatusClosed = "closed" ) type ProxyManager struct { diff --git a/client/visitor.go b/client/visitor.go index 6e1e1c8..6634401 100644 --- a/client/visitor.go +++ b/client/visitor.go @@ -202,7 +202,16 @@ func (sv *XtcpVisitor) handleConn(userConn frpNet.Conn) { raddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", g.GlbClientCfg.ServerAddr, g.GlbClientCfg.ServerUdpPort)) + if err != nil { + sv.Error("resolve server UDP addr error") + return + } + visitorConn, err := net.DialUDP("udp", nil, raddr) + if err != nil { + sv.Warn("dial server udp addr error: %v", err) + return + } defer visitorConn.Close() now := time.Now().Unix() From b33ea9274cdd263dd0a2ff9cf8f068606171d42a Mon Sep 17 00:00:00 2001 From: fatedier Date: Tue, 6 Nov 2018 18:35:05 +0800 Subject: [PATCH 10/10] client/control: refactor code --- client/admin_api.go | 2 +- client/control.go | 213 ++++++------------------------------ client/proxy.go | 2 +- client/proxy_manager.go | 232 ++++++++++++++++++++-------------------- client/service.go | 185 ++++++++++++++++++++++++++++++-- conf/frpc_full.ini | 2 +- 6 files changed, 323 insertions(+), 313 deletions(-) diff --git a/client/admin_api.go b/client/admin_api.go index 4eafa10..5074540 100644 --- a/client/admin_api.go +++ b/client/admin_api.go @@ -85,7 +85,7 @@ func (svr *Service) apiReload(w http.ResponseWriter, r *http.Request) { return } - err = svr.ctl.reloadConf(pxyCfgs, visitorCfgs) + err = svr.ctl.ReloadConf(pxyCfgs, visitorCfgs) if err != nil { res.Code = 4 res.Msg = err.Error() diff --git a/client/control.go b/client/control.go index 90d2d4a..09ca2a0 100644 --- a/client/control.go +++ b/client/control.go @@ -17,8 +17,6 @@ package client import ( "fmt" "io" - "io/ioutil" - "runtime" "runtime/debug" "sync" "time" @@ -28,24 +26,15 @@ import ( "github.com/fatedier/frp/models/msg" "github.com/fatedier/frp/utils/log" frpNet "github.com/fatedier/frp/utils/net" - "github.com/fatedier/frp/utils/util" - "github.com/fatedier/frp/utils/version" "github.com/fatedier/golib/control/shutdown" "github.com/fatedier/golib/crypto" fmux "github.com/hashicorp/yamux" ) -const ( - connReadTimeout time.Duration = 10 * time.Second -) - type Control struct { - // frpc service - svr *Service - - // login message to server, only used - loginMsg *msg.Login + // uniq id got from frps, attach it in loginMsg + runId string // manage all proxies pm *ProxyManager @@ -65,14 +54,10 @@ type Control struct { // read from this channel to get the next message sent by server readCh chan (msg.Message) - // run id got from server - runId string - - // if we call close() in control, do not reconnect to server - exit bool - // goroutines can block by reading from this channel, it will be closed only in reader() when control connection is closed - closedCh chan int + closedCh chan struct{} + + closedDoneCh chan struct{} // last time got the Pong message lastPong time.Time @@ -86,50 +71,28 @@ type Control struct { log.Logger } -func NewControl(svr *Service, pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) *Control { - loginMsg := &msg.Login{ - Arch: runtime.GOARCH, - Os: runtime.GOOS, - PoolCount: g.GlbClientCfg.PoolCount, - User: g.GlbClientCfg.User, - Version: version.Full(), - } +func NewControl(runId string, conn frpNet.Conn, session *fmux.Session, pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) *Control { ctl := &Control{ - svr: svr, - loginMsg: loginMsg, + runId: runId, + conn: conn, + session: session, sendCh: make(chan msg.Message, 100), readCh: make(chan msg.Message, 100), - closedCh: make(chan int), + closedCh: make(chan struct{}), + closedDoneCh: make(chan struct{}), readerShutdown: shutdown.New(), writerShutdown: shutdown.New(), msgHandlerShutdown: shutdown.New(), Logger: log.NewPrefixLogger(""), } - ctl.pm = NewProxyManager(ctl, ctl.sendCh, "") + ctl.pm = NewProxyManager(ctl.sendCh, "") ctl.pm.Reload(pxyCfgs, false) ctl.vm = NewVisitorManager(ctl) ctl.vm.Reload(visitorCfgs) return ctl } -func (ctl *Control) Run() (err error) { - for { - err = ctl.login() - if err != nil { - ctl.Warn("login to server failed: %v", err) - - // if login_fail_exit is true, just exit this program - // otherwise sleep a while and continues relogin to server - if g.GlbClientCfg.LoginFailExit { - return - } else { - time.Sleep(10 * time.Second) - } - } else { - break - } - } - +func (ctl *Control) Run() { go ctl.worker() // start all local visitors and send NewProxy message for all configured proxies @@ -137,7 +100,7 @@ func (ctl *Control) Run() (err error) { ctl.pm.CheckAndStartProxy([]string{ProxyStatusNew}) go ctl.vm.Run() - return nil + return } func (ctl *Control) HandleReqWorkConn(inMsg *msg.ReqWorkConn) { @@ -179,80 +142,13 @@ func (ctl *Control) HandleNewProxyResp(inMsg *msg.NewProxyResp) { } func (ctl *Control) Close() error { - ctl.mu.Lock() - defer ctl.mu.Unlock() - ctl.exit = true ctl.pm.CloseProxies() return nil } -// login send a login message to server and wait for a loginResp message. -func (ctl *Control) login() (err error) { - if ctl.conn != nil { - ctl.conn.Close() - } - if ctl.session != nil { - ctl.session.Close() - } - - conn, err := frpNet.ConnectServerByProxy(g.GlbClientCfg.HttpProxy, g.GlbClientCfg.Protocol, - fmt.Sprintf("%s:%d", g.GlbClientCfg.ServerAddr, g.GlbClientCfg.ServerPort)) - if err != nil { - return err - } - - defer func() { - if err != nil { - conn.Close() - } - }() - - if g.GlbClientCfg.TcpMux { - fmuxCfg := fmux.DefaultConfig() - fmuxCfg.LogOutput = ioutil.Discard - session, errRet := fmux.Client(conn, fmuxCfg) - if errRet != nil { - return errRet - } - stream, errRet := session.OpenStream() - if errRet != nil { - session.Close() - return errRet - } - conn = frpNet.WrapConn(stream) - ctl.session = session - } - - now := time.Now().Unix() - ctl.loginMsg.PrivilegeKey = util.GetAuthKey(g.GlbClientCfg.Token, now) - ctl.loginMsg.Timestamp = now - ctl.loginMsg.RunId = ctl.runId - - if err = msg.WriteMsg(conn, ctl.loginMsg); err != nil { - return err - } - - var loginRespMsg msg.LoginResp - conn.SetReadDeadline(time.Now().Add(connReadTimeout)) - if err = msg.ReadMsgInto(conn, &loginRespMsg); err != nil { - return err - } - conn.SetReadDeadline(time.Time{}) - - if loginRespMsg.Error != "" { - err = fmt.Errorf("%s", loginRespMsg.Error) - ctl.Error("%s", loginRespMsg.Error) - return err - } - - ctl.conn = conn - // update runId got from server - ctl.runId = loginRespMsg.RunId - g.GlbClientCfg.ServerUdpPort = loginRespMsg.ServerUdpPort - ctl.ClearLogPrefix() - ctl.AddLogPrefix(loginRespMsg.RunId) - ctl.Info("login to server success, get run id [%s], server udp port [%d]", loginRespMsg.RunId, loginRespMsg.ServerUdpPort) - return nil +// ClosedDoneCh returns a channel which will be closed after all resources are released +func (ctl *Control) ClosedDoneCh() <-chan struct{} { + return ctl.closedDoneCh } // connectServer return a new connection to frps @@ -373,87 +269,38 @@ func (ctl *Control) msgHandler() { } } -// controler keep watching closedCh, start a new connection if previous control connection is closed. -// If controler is notified by closedCh, reader and writer and handler will exit, then recall these functions. +// If controler is notified by closedCh, reader and writer and handler will exit func (ctl *Control) worker() { go ctl.msgHandler() go ctl.reader() go ctl.writer() - var err error - maxDelayTime := 20 * time.Second - delayTime := time.Second - checkInterval := 60 * time.Second checkProxyTicker := time.NewTicker(checkInterval) + for { select { case <-checkProxyTicker.C: // check which proxy registered failed and reregister it to server ctl.pm.CheckAndStartProxy([]string{ProxyStatusStartErr, ProxyStatusClosed}) - case _, ok := <-ctl.closedCh: - // we won't get any variable from this channel - if !ok { - // close related channels and wait until other goroutines done - close(ctl.readCh) - ctl.readerShutdown.WaitDone() - ctl.msgHandlerShutdown.WaitDone() + case <-ctl.closedCh: + // close related channels and wait until other goroutines done + close(ctl.readCh) + ctl.readerShutdown.WaitDone() + ctl.msgHandlerShutdown.WaitDone() - close(ctl.sendCh) - ctl.writerShutdown.WaitDone() + close(ctl.sendCh) + ctl.writerShutdown.WaitDone() - ctl.pm.CloseProxies() - // if ctl.exit is true, just exit - ctl.mu.RLock() - exit := ctl.exit - ctl.mu.RUnlock() - if exit { - return - } + ctl.pm.CloseProxies() - // loop util reconnecting to server success - for { - ctl.Info("try to reconnect to server...") - err = ctl.login() - if err != nil { - ctl.Warn("reconnect to server error: %v", err) - time.Sleep(delayTime) - delayTime = delayTime * 2 - if delayTime > maxDelayTime { - delayTime = maxDelayTime - } - continue - } - // reconnect success, init delayTime - delayTime = time.Second - break - } - - // init related channels and variables - ctl.sendCh = make(chan msg.Message, 100) - ctl.readCh = make(chan msg.Message, 100) - ctl.closedCh = make(chan int) - ctl.readerShutdown = shutdown.New() - ctl.writerShutdown = shutdown.New() - ctl.msgHandlerShutdown = shutdown.New() - ctl.pm.Reset(ctl.sendCh, ctl.runId) - - // previous work goroutines should be closed and start them here - go ctl.msgHandler() - go ctl.writer() - go ctl.reader() - - // start all configured proxies - ctl.pm.CheckAndStartProxy([]string{ProxyStatusNew, ProxyStatusClosed}) - - checkProxyTicker.Stop() - checkProxyTicker = time.NewTicker(checkInterval) - } + close(ctl.closedDoneCh) + return } } } -func (ctl *Control) reloadConf(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) error { +func (ctl *Control) ReloadConf(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) error { ctl.vm.Reload(visitorCfgs) err := ctl.pm.Reload(pxyCfgs, true) return err diff --git a/client/proxy.go b/client/proxy.go index 26c9a66..a89921d 100644 --- a/client/proxy.go +++ b/client/proxy.go @@ -35,7 +35,7 @@ import ( "github.com/fatedier/golib/pool" ) -// Proxy defines how to deal with work connections for different proxy type. +// Proxy defines how to handle work connections for different proxy type. type Proxy interface { Run() error diff --git a/client/proxy_manager.go b/client/proxy_manager.go index fe175a0..dc9f350 100644 --- a/client/proxy_manager.go +++ b/client/proxy_manager.go @@ -22,7 +22,6 @@ const ( ) type ProxyManager struct { - ctl *Control sendCh chan (msg.Message) proxies map[string]*ProxyWrapper closed bool @@ -31,122 +30,8 @@ type ProxyManager struct { log.Logger } -type ProxyWrapper struct { - Name string - Type string - Status string - Err string - Cfg config.ProxyConf - - RemoteAddr string - - pxy Proxy - - mu sync.RWMutex -} - -type ProxyStatus struct { - Name string `json:"name"` - Type string `json:"type"` - Status string `json:"status"` - Err string `json:"err"` - Cfg config.ProxyConf `json:"cfg"` - - // Got from server. - RemoteAddr string `json:"remote_addr"` -} - -func NewProxyWrapper(cfg config.ProxyConf) *ProxyWrapper { - return &ProxyWrapper{ - Name: cfg.GetBaseInfo().ProxyName, - Type: cfg.GetBaseInfo().ProxyType, - Status: ProxyStatusNew, - Cfg: cfg, - pxy: nil, - } -} - -func (pw *ProxyWrapper) GetStatusStr() string { - pw.mu.RLock() - defer pw.mu.RUnlock() - return pw.Status -} - -func (pw *ProxyWrapper) GetStatus() *ProxyStatus { - pw.mu.RLock() - defer pw.mu.RUnlock() - ps := &ProxyStatus{ - Name: pw.Name, - Type: pw.Type, - Status: pw.Status, - Err: pw.Err, - Cfg: pw.Cfg, - RemoteAddr: pw.RemoteAddr, - } - return ps -} - -func (pw *ProxyWrapper) WaitStart() { - pw.mu.Lock() - defer pw.mu.Unlock() - pw.Status = ProxyStatusWaitStart -} - -func (pw *ProxyWrapper) Start(remoteAddr string, serverRespErr string) error { - if pw.pxy != nil { - pw.pxy.Close() - pw.pxy = nil - } - - if serverRespErr != "" { - pw.mu.Lock() - pw.Status = ProxyStatusStartErr - pw.RemoteAddr = remoteAddr - pw.Err = serverRespErr - pw.mu.Unlock() - return fmt.Errorf(serverRespErr) - } - - pxy := NewProxy(pw.Cfg) - pw.mu.Lock() - defer pw.mu.Unlock() - pw.RemoteAddr = remoteAddr - if err := pxy.Run(); err != nil { - pw.Status = ProxyStatusStartErr - pw.Err = err.Error() - return err - } - pw.Status = ProxyStatusRunning - pw.Err = "" - pw.pxy = pxy - return nil -} - -func (pw *ProxyWrapper) InWorkConn(workConn frpNet.Conn) { - pw.mu.RLock() - pxy := pw.pxy - pw.mu.RUnlock() - if pxy != nil { - workConn.Debug("start a new work connection, localAddr: %s remoteAddr: %s", workConn.LocalAddr().String(), workConn.RemoteAddr().String()) - go pxy.InWorkConn(workConn) - } else { - workConn.Close() - } -} - -func (pw *ProxyWrapper) Close() { - pw.mu.Lock() - defer pw.mu.Unlock() - if pw.pxy != nil { - pw.pxy.Close() - pw.pxy = nil - } - pw.Status = ProxyStatusClosed -} - -func NewProxyManager(ctl *Control, msgSendCh chan (msg.Message), logPrefix string) *ProxyManager { +func NewProxyManager(msgSendCh chan (msg.Message), logPrefix string) *ProxyManager { return &ProxyManager{ - ctl: ctl, proxies: make(map[string]*ProxyWrapper), sendCh: msgSendCh, closed: false, @@ -309,3 +194,118 @@ func (pm *ProxyManager) GetAllProxyStatus() []*ProxyStatus { } return ps } + +type ProxyStatus struct { + Name string `json:"name"` + Type string `json:"type"` + Status string `json:"status"` + Err string `json:"err"` + Cfg config.ProxyConf `json:"cfg"` + + // Got from server. + RemoteAddr string `json:"remote_addr"` +} + +// ProxyWrapper is a wrapper of Proxy interface only used in ProxyManager +// Add additional proxy status info +type ProxyWrapper struct { + Name string + Type string + Status string + Err string + Cfg config.ProxyConf + + RemoteAddr string + + pxy Proxy + + mu sync.RWMutex +} + +func NewProxyWrapper(cfg config.ProxyConf) *ProxyWrapper { + return &ProxyWrapper{ + Name: cfg.GetBaseInfo().ProxyName, + Type: cfg.GetBaseInfo().ProxyType, + Status: ProxyStatusNew, + Cfg: cfg, + pxy: nil, + } +} + +func (pw *ProxyWrapper) GetStatusStr() string { + pw.mu.RLock() + defer pw.mu.RUnlock() + return pw.Status +} + +func (pw *ProxyWrapper) GetStatus() *ProxyStatus { + pw.mu.RLock() + defer pw.mu.RUnlock() + ps := &ProxyStatus{ + Name: pw.Name, + Type: pw.Type, + Status: pw.Status, + Err: pw.Err, + Cfg: pw.Cfg, + RemoteAddr: pw.RemoteAddr, + } + return ps +} + +func (pw *ProxyWrapper) WaitStart() { + pw.mu.Lock() + defer pw.mu.Unlock() + pw.Status = ProxyStatusWaitStart +} + +func (pw *ProxyWrapper) Start(remoteAddr string, serverRespErr string) error { + if pw.pxy != nil { + pw.pxy.Close() + pw.pxy = nil + } + + if serverRespErr != "" { + pw.mu.Lock() + pw.Status = ProxyStatusStartErr + pw.RemoteAddr = remoteAddr + pw.Err = serverRespErr + pw.mu.Unlock() + return fmt.Errorf(serverRespErr) + } + + pxy := NewProxy(pw.Cfg) + pw.mu.Lock() + defer pw.mu.Unlock() + pw.RemoteAddr = remoteAddr + if err := pxy.Run(); err != nil { + pw.Status = ProxyStatusStartErr + pw.Err = err.Error() + return err + } + pw.Status = ProxyStatusRunning + pw.Err = "" + pw.pxy = pxy + return nil +} + +func (pw *ProxyWrapper) InWorkConn(workConn frpNet.Conn) { + pw.mu.RLock() + pxy := pw.pxy + pw.mu.RUnlock() + if pxy != nil { + workConn.Debug("start a new work connection, localAddr: %s remoteAddr: %s", workConn.LocalAddr().String(), workConn.RemoteAddr().String()) + go pxy.InWorkConn(workConn) + } else { + workConn.Close() + } +} + +func (pw *ProxyWrapper) Close() { + pw.mu.Lock() + defer pw.mu.Unlock() + if pw.pxy != nil { + pw.pxy.Close() + pw.pxy = nil + } + pw.Status = ProxyStatusClosed +} diff --git a/client/service.go b/client/service.go index 2589f52..62cf151 100644 --- a/client/service.go +++ b/client/service.go @@ -15,35 +15,85 @@ package client import ( + "fmt" + "io/ioutil" + "runtime" + "sync" + "sync/atomic" + "time" + "github.com/fatedier/frp/g" "github.com/fatedier/frp/models/config" + "github.com/fatedier/frp/models/msg" "github.com/fatedier/frp/utils/log" + frpNet "github.com/fatedier/frp/utils/net" + "github.com/fatedier/frp/utils/util" + "github.com/fatedier/frp/utils/version" + + fmux "github.com/hashicorp/yamux" ) type Service struct { - // manager control connection with server - ctl *Control + // uniq id got from frps, attach it in loginMsg + runId string + // manager control connection with server + ctl *Control + ctlMu sync.RWMutex + + pxyCfgs map[string]config.ProxyConf + visitorCfgs map[string]config.VisitorConf + cfgMu sync.RWMutex + + exit uint32 // 0 means not exit closedCh chan int } func NewService(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) (svr *Service) { svr = &Service{ - closedCh: make(chan int), + pxyCfgs: pxyCfgs, + visitorCfgs: visitorCfgs, + exit: 0, + closedCh: make(chan int), } - ctl := NewControl(svr, pxyCfgs, visitorCfgs) - svr.ctl = ctl return } +func (svr *Service) GetController() *Control { + svr.ctlMu.RLock() + defer svr.ctlMu.RUnlock() + return svr.ctl +} + func (svr *Service) Run() error { - err := svr.ctl.Run() - if err != nil { - return err + // first login + for { + conn, session, err := svr.login() + if err != nil { + log.Warn("login to server failed: %v", err) + + // if login_fail_exit is true, just exit this program + // otherwise sleep a while and try again to connect to server + if g.GlbClientCfg.LoginFailExit { + return err + } else { + time.Sleep(10 * time.Second) + } + } else { + // login success + ctl := NewControl(svr.runId, conn, session, svr.pxyCfgs, svr.visitorCfgs) + ctl.Run() + svr.ctlMu.Lock() + svr.ctl = ctl + svr.ctlMu.Unlock() + break + } } + go svr.keepControllerWorking() + if g.GlbClientCfg.AdminPort != 0 { - err = svr.RunAdminServer(g.GlbClientCfg.AdminAddr, g.GlbClientCfg.AdminPort) + err := svr.RunAdminServer(g.GlbClientCfg.AdminAddr, g.GlbClientCfg.AdminPort) if err != nil { log.Warn("run admin server error: %v", err) } @@ -54,6 +104,119 @@ func (svr *Service) Run() error { return nil } -func (svr *Service) Close() { - svr.ctl.Close() +func (svr *Service) keepControllerWorking() { + maxDelayTime := 20 * time.Second + delayTime := time.Second + + for { + <-svr.ctl.ClosedDoneCh() + if atomic.LoadUint32(&svr.exit) != 0 { + return + } + + for { + log.Info("try to reconnect to server...") + conn, session, err := svr.login() + if err != nil { + log.Warn("reconnect to server error: %v", err) + time.Sleep(delayTime) + delayTime = delayTime * 2 + if delayTime > maxDelayTime { + delayTime = maxDelayTime + } + continue + } + // reconnect success, init delayTime + delayTime = time.Second + + ctl := NewControl(svr.runId, conn, session, svr.pxyCfgs, svr.visitorCfgs) + ctl.Run() + svr.ctlMu.Lock() + svr.ctl = ctl + svr.ctlMu.Unlock() + break + } + } +} + +// login creates a connection to frps and registers it self as a client +// conn: control connection +// session: if it's not nil, using tcp mux +func (svr *Service) login() (conn frpNet.Conn, session *fmux.Session, err error) { + conn, err = frpNet.ConnectServerByProxy(g.GlbClientCfg.HttpProxy, g.GlbClientCfg.Protocol, + fmt.Sprintf("%s:%d", g.GlbClientCfg.ServerAddr, g.GlbClientCfg.ServerPort)) + if err != nil { + return + } + + defer func() { + if err != nil { + conn.Close() + } + }() + + if g.GlbClientCfg.TcpMux { + fmuxCfg := fmux.DefaultConfig() + fmuxCfg.LogOutput = ioutil.Discard + session, err = fmux.Client(conn, fmuxCfg) + if err != nil { + return + } + stream, errRet := session.OpenStream() + if errRet != nil { + session.Close() + err = errRet + return + } + conn = frpNet.WrapConn(stream) + } + + now := time.Now().Unix() + loginMsg := &msg.Login{ + Arch: runtime.GOARCH, + Os: runtime.GOOS, + PoolCount: g.GlbClientCfg.PoolCount, + User: g.GlbClientCfg.User, + Version: version.Full(), + PrivilegeKey: util.GetAuthKey(g.GlbClientCfg.Token, now), + Timestamp: now, + RunId: svr.runId, + } + + if err = msg.WriteMsg(conn, loginMsg); err != nil { + return + } + + var loginRespMsg msg.LoginResp + conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + if err = msg.ReadMsgInto(conn, &loginRespMsg); err != nil { + return + } + conn.SetReadDeadline(time.Time{}) + + if loginRespMsg.Error != "" { + err = fmt.Errorf("%s", loginRespMsg.Error) + log.Error("%s", loginRespMsg.Error) + return + } + + svr.runId = loginRespMsg.RunId + g.GlbClientCfg.ServerUdpPort = loginRespMsg.ServerUdpPort + log.Info("login to server success, get run id [%s], server udp port [%d]", loginRespMsg.RunId, loginRespMsg.ServerUdpPort) + return +} + +func (svr *Service) ReloadConf(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) error { + svr.cfgMu.Lock() + svr.pxyCfgs = pxyCfgs + svr.visitorCfgs = visitorCfgs + svr.cfgMu.Unlock() + + return svr.ctl.ReloadConf(pxyCfgs, visitorCfgs) +} + +func (svr *Service) Close() { + atomic.StoreUint32(&svr.exit, 1) + svr.ctl.Close() + close(svr.closedCh) } diff --git a/conf/frpc_full.ini b/conf/frpc_full.ini index 2307eeb..d9892f5 100644 --- a/conf/frpc_full.ini +++ b/conf/frpc_full.ini @@ -25,7 +25,7 @@ token = 12345678 admin_addr = 127.0.0.1 admin_port = 7400 admin_user = admin -admin_passwd = admin +admin_pwd = admin # connections will be established in advance, default value is zero pool_count = 5