improve e2e port allocator (#3882)

modify ext func, specify whether exit immediately

.gitignore

@@ -34,6 +34,8 @@ dist/
 .idea/
 .vscode/
 .autogen_ssh_key
+client.crt
+client.key
 
 # Cache
 *.swp

.golangci.yml

@@ -132,6 +132,9 @@ issues:
   - linters:
     - revive
     text: "unused-parameter"
+  - linters:
+    - unparam
+    text: "is always false"
 
   # Independently from option `exclude` we use default exclude patterns,
   # it can be disabled by this option. To list all

Release.md

@@ -1,11 +1,3 @@
-### Features
-
-* The new command line parameter `--strict_config` has been added to enable strict configuration validation mode. It will throw an error for unknown fields instead of ignoring them. In future versions, we will set the default value of this parameter to true to avoid misconfigurations.
-* Support `SSH reverse tunneling`. With this feature, you can expose your local service without running frpc, only using SSH. The SSH reverse tunnel agent has many functional limitations compared to the frpc agent. The currently supported proxy types are tcp, http, https, tcpmux, and stcp.
-* The frpc tcpmux command line parameters have been updated to support configuring `http_user` and `http_pwd`.
-* The frpc stcp/sudp/xtcp command line parameters have been updated to support configuring `allow_users`.
-
 ### Fixes
 
-* frpc: Return code 1 when the first login attempt fails and exits.
-* When auth.method is `oidc` and auth.additionalScopes contains `HeartBeats`, if obtaining AccessToken fails, the application will be unresponsive.
+* frpc has a certain chance to panic when login: close of closed channel.

client/control.go

@@ -239,15 +239,15 @@ func (ctl *Control) heartbeatWorker() {
 	// Users can still enable heartbeat feature by setting HeartbeatInterval to a positive value.
 	if ctl.sessionCtx.Common.Transport.HeartbeatInterval > 0 {
 		// send heartbeat to server
-		sendHeartBeat := func() error {
+		sendHeartBeat := func() (bool, error) {
 			xl.Debug("send heartbeat to server")
 			pingMsg := &msg.Ping{}
 			if err := ctl.sessionCtx.AuthSetter.SetPing(pingMsg); err != nil {
 				xl.Warn("error during ping authentication: %v, skip sending ping message", err)
-				return err
+				return false, err
 			}
 			_ = ctl.msgDispatcher.Send(pingMsg)
-			return nil
+			return false, nil
 		}
 
 		go wait.BackoffUntil(sendHeartBeat,

client/service.go

@@ -192,16 +192,16 @@ func (svr *Service) keepControllerWorking() {
 	// the control immediately exits. It is necessary to limit the frequency of reconnection in this case.
 	// The interval for the first three retries in 1 minute will be very short, and then it will increase exponentially.
 	// The maximum interval is 20 seconds.
-	wait.BackoffUntil(func() error {
+	wait.BackoffUntil(func() (bool, error) {
 		// loopLoginUntilSuccess is another layer of loop that will continuously attempt to
 		// login to the server until successful.
 		svr.loopLoginUntilSuccess(20*time.Second, false)
 		if svr.ctl != nil {
 			<-svr.ctl.Done()
-			return errors.New("control is closed and try another loop")
+			return false, errors.New("control is closed and try another loop")
 		}
 		// If the control is nil, it means that the login failed and the service is also closed.
-		return nil
+		return false, nil
 	}, wait.NewFastBackoffManager(
 		wait.FastBackoffOptions{
 			Duration:        time.Second,
@@ -282,9 +282,8 @@ func (svr *Service) login() (conn net.Conn, connector Connector, err error) {
 
 func (svr *Service) loopLoginUntilSuccess(maxInterval time.Duration, firstLoginExit bool) {
 	xl := xlog.FromContextSafe(svr.ctx)
-	successCh := make(chan struct{})
 
-	loginFunc := func() error {
+	loginFunc := func() (bool, error) {
 		xl.Info("try to connect to server...")
 		conn, connector, err := svr.login()
 		if err != nil {
@@ -292,7 +291,7 @@ func (svr *Service) loopLoginUntilSuccess(maxInterval time.Duration, firstLoginE
 			if firstLoginExit {
 				svr.cancel(cancelErr{Err: err})
 			}
-			return err
+			return false, err
 		}
 
 		svr.cfgMu.RLock()
@@ -315,7 +314,7 @@ func (svr *Service) loopLoginUntilSuccess(maxInterval time.Duration, firstLoginE
 		if err != nil {
 			conn.Close()
 			xl.Error("NewControl error: %v", err)
-			return err
+			return false, err
 		}
 		ctl.SetInWorkConnCallback(svr.handleWorkConnCb)
 
@@ -327,9 +326,7 @@ func (svr *Service) loopLoginUntilSuccess(maxInterval time.Duration, firstLoginE
 		}
 		svr.ctl = ctl
 		svr.ctlMu.Unlock()
-
-		close(successCh)
-		return nil
+		return true, nil
 	}
 
 	// try to reconnect to server until success
@@ -339,9 +336,7 @@ func (svr *Service) loopLoginUntilSuccess(maxInterval time.Duration, firstLoginE
 			Factor:      2,
 			Jitter:      0.1,
 			MaxDuration: maxInterval,
-		}),
-		true,
-		wait.MergeAndCloseOnAnyStopChannel(svr.ctx.Done(), successCh))
+		}), true, svr.ctx.Done())
 }
 
 func (svr *Service) UpdateAllConfigurer(proxyCfgs []v1.ProxyConfigurer, visitorCfgs []v1.VisitorConfigurer) error {

hack/run-e2e.sh

@@ -26,5 +26,9 @@ frpsPath=${ROOT}/bin/frps
 if [ "${FRPS_PATH}" ]; then
     frpsPath="${FRPS_PATH}"
 fi
+concurrency="16"
+if [ "${CONCURRENCY}" ]; then
+    concurrency="${CONCURRENCY}"
+fi
 
-ginkgo -nodes=8 --poll-progress-after=60s ${ROOT}/test/e2e -- -frpc-path=${frpcPath} -frps-path=${frpsPath} -log-level=${logLevel} -debug=${debug}
+ginkgo -nodes=${concurrency} --poll-progress-after=60s ${ROOT}/test/e2e -- -frpc-path=${frpcPath} -frps-path=${frpsPath} -log-level=${logLevel} -debug=${debug}

pkg/util/version/version.go

@@ -19,7 +19,7 @@ import (
 	"strings"
 )
 
-var version = "0.53.0"
+var version = "0.53.2"
 
 func Full() string {
 	return version

pkg/util/wait/backoff.go

@@ -113,7 +113,7 @@ func (f *fastBackoffImpl) Backoff(previousDuration time.Duration, previousCondit
 	return f.options.Duration
 }
 
-func BackoffUntil(f func() error, backoff BackoffManager, sliding bool, stopCh <-chan struct{}) {
+func BackoffUntil(f func() (bool, error), backoff BackoffManager, sliding bool, stopCh <-chan struct{}) {
 	var delay time.Duration
 	previousError := false
 
@@ -131,7 +131,9 @@ func BackoffUntil(f func() error, backoff BackoffManager, sliding bool, stopCh <
 			delay = backoff.Backoff(delay, previousError)
 		}
 
-		if err := f(); err != nil {
+		if done, err := f(); done {
+			return
+		} else if err != nil {
 			previousError = true
 		} else {
 			previousError = false
@@ -142,12 +144,6 @@ func BackoffUntil(f func() error, backoff BackoffManager, sliding bool, stopCh <
 		}
 
 		ticker.Reset(delay)
-		select {
-		case <-stopCh:
-			return
-		default:
-		}
-
 		select {
 		case <-stopCh:
 			return
@@ -170,9 +166,9 @@ func Jitter(duration time.Duration, maxFactor float64) time.Duration {
 }
 
 func Until(f func(), period time.Duration, stopCh <-chan struct{}) {
-	ff := func() error {
+	ff := func() (bool, error) {
 		f()
-		return nil
+		return false, nil
 	}
 	BackoffUntil(ff, BackoffFunc(func(time.Duration, bool) time.Duration {
 		return period

test/e2e/framework/framework.go

@@ -67,7 +67,7 @@ func NewDefaultFramework() *Framework {
 		TotalParallelNode: suiteConfig.ParallelTotal,
 		CurrentNodeIndex:  suiteConfig.ParallelProcess,
 		FromPortIndex:     10000,
-		ToPortIndex:       60000,
+		ToPortIndex:       30000,
 	}
 	return NewFramework(options)
 }

test/e2e/framework/process.go

@@ -42,7 +42,7 @@ func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []str
 		ExpectNoError(err)
 		time.Sleep(500 * time.Millisecond)
 	}
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 
 	currentClientProcesses := make([]*process.Process, 0, len(clientTemplates))
 	for i := range clientTemplates {
@@ -76,7 +76,7 @@ func (f *Framework) RunFrps(args ...string) (*process.Process, string, error) {
 		return p, p.StdOutput(), err
 	}
 	// sleep for a while to get std output
-	time.Sleep(time.Second)
+	time.Sleep(2 * time.Second)
 	return p, p.StdOutput(), nil
 }
 
@@ -87,7 +87,7 @@ func (f *Framework) RunFrpc(args ...string) (*process.Process, string, error) {
 	if err != nil {
 		return p, p.StdOutput(), err
 	}
-	time.Sleep(time.Second)
+	time.Sleep(2 * time.Second)
 	return p, p.StdOutput(), nil
 }
 

test/e2e/legacy/basic/server.go

@@ -22,11 +22,11 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 		clientConf := consts.LegacyDefaultClientConfig
 
 		serverConf += `
-			allow_ports = 20000-25000,25002,30000-50000
+			allow_ports = 10000-11000,11002,12000-13000
 		`
 
-		tcpPortName := port.GenName("TCP", port.WithRangePorts(20000, 25000))
-		udpPortName := port.GenName("UDP", port.WithRangePorts(30000, 50000))
+		tcpPortName := port.GenName("TCP", port.WithRangePorts(10000, 11000))
+		udpPortName := port.GenName("UDP", port.WithRangePorts(12000, 13000))
 		clientConf += fmt.Sprintf(`
 			[tcp-allowded-in-range]
 			type = tcp
@@ -37,7 +37,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 			[tcp-port-not-allowed]
 			type = tcp
 			local_port = {{ .%s }}
-			remote_port = 25001
+			remote_port = 11001
 			`, framework.TCPEchoServerPort)
 		clientConf += fmt.Sprintf(`
 			[tcp-port-unavailable]
@@ -55,7 +55,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 			[udp-port-not-allowed]
 			type = udp
 			local_port = {{ .%s }}
-			remote_port = 25003
+			remote_port = 11003
 			`, framework.UDPEchoServerPort)
 
 		f.RunProcesses([]string{serverConf}, []string{clientConf})
@@ -65,7 +65,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 		framework.NewRequestExpect(f).PortName(tcpPortName).Ensure()
 
 		// Not Allowed
-		framework.NewRequestExpect(f).Port(25001).ExpectError(true).Ensure()
+		framework.NewRequestExpect(f).Port(11001).ExpectError(true).Ensure()
 
 		// Unavailable, already bind by frps
 		framework.NewRequestExpect(f).PortName(consts.PortServerName).ExpectError(true).Ensure()
@@ -76,7 +76,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 
 		// Not Allowed
 		framework.NewRequestExpect(f).RequestModify(func(r *request.Request) {
-			r.UDP().Port(25003)
+			r.UDP().Port(11003)
 		}).ExpectError(true).Ensure()
 	})
 

test/e2e/pkg/port/port.go

@@ -79,6 +79,7 @@ func (pa *Allocator) GetByName(portName string) int {
 		udpConn.Close()
 
 		pa.used.Insert(port)
+		pa.reserved.Delete(port)
 		return port
 	}
 	return 0

test/e2e/v1/basic/server.go

@@ -23,14 +23,14 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 
 		serverConf += `
 		allowPorts = [
-		  { start = 20000, end = 25000 },
-		  { single = 25002 },
-		  { start = 30000, end = 50000 },
+		  { start = 10000, end = 11000 },
+		  { single = 11002 },
+		  { start = 12000, end = 13000 },
 		]
 		`
 
-		tcpPortName := port.GenName("TCP", port.WithRangePorts(20000, 25000))
-		udpPortName := port.GenName("UDP", port.WithRangePorts(30000, 50000))
+		tcpPortName := port.GenName("TCP", port.WithRangePorts(10000, 11000))
+		udpPortName := port.GenName("UDP", port.WithRangePorts(12000, 13000))
 		clientConf += fmt.Sprintf(`
 			[[proxies]]
 			name = "tcp-allowded-in-range"
@@ -43,7 +43,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 			name = "tcp-port-not-allowed"
 			type = "tcp"
 			localPort = {{ .%s }}
-			remotePort = 25001
+			remotePort = 11001
 			`, framework.TCPEchoServerPort)
 		clientConf += fmt.Sprintf(`
 			[[proxies]]
@@ -64,7 +64,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 			name = "udp-port-not-allowed"
 			type = "udp"
 			localPort = {{ .%s }}
-			remotePort = 25003
+			remotePort = 11003
 			`, framework.UDPEchoServerPort)
 
 		f.RunProcesses([]string{serverConf}, []string{clientConf})
@@ -74,7 +74,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 		framework.NewRequestExpect(f).PortName(tcpPortName).Ensure()
 
 		// Not Allowed
-		framework.NewRequestExpect(f).Port(25001).ExpectError(true).Ensure()
+		framework.NewRequestExpect(f).Port(11001).ExpectError(true).Ensure()
 
 		// Unavailable, already bind by frps
 		framework.NewRequestExpect(f).PortName(consts.PortServerName).ExpectError(true).Ensure()
@@ -85,7 +85,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
 
 		// Not Allowed
 		framework.NewRequestExpect(f).RequestModify(func(r *request.Request) {
-			r.UDP().Port(25003)
+			r.UDP().Port(11003)
 		}).ExpectError(true).Ensure()
 	})