frpc: support stop command (#3511)

Release.md

@@ -1,6 +1,11 @@
 ### Features
 
 * frpc supports connecting to frps via the wss protocol by enabling the configuration `protocol = wss`.
+* frpc supports stopping the service through the stop command.
+
+### Improvements
+
+* service.Run supports passing in context.
 
 ### Fixes
 

client/admin.go

@@ -52,6 +52,7 @@ func (svr *Service) RunAdminServer(address string) (err error) {
 
 	// api, see admin_api.go
 	subRouter.HandleFunc("/api/reload", svr.apiReload).Methods("GET")
+	subRouter.HandleFunc("/api/stop", svr.apiStop).Methods("POST")
 	subRouter.HandleFunc("/api/status", svr.apiStatus).Methods("GET")
 	subRouter.HandleFunc("/api/config", svr.apiGetConfig).Methods("GET")
 	subRouter.HandleFunc("/api/config", svr.apiPutConfig).Methods("PUT")

client/admin_api.go

@@ -24,6 +24,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/samber/lo"
 
@@ -42,7 +43,7 @@ func (svr *Service) healthz(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(200)
 }
 
-// GET api/reload
+// GET /api/reload
 func (svr *Service) apiReload(w http.ResponseWriter, r *http.Request) {
 	res := GeneralResponse{Code: 200}
 
@@ -72,6 +73,22 @@ func (svr *Service) apiReload(w http.ResponseWriter, r *http.Request) {
 	log.Info("success reload conf")
 }
 
+// POST /api/stop
+func (svr *Service) apiStop(w http.ResponseWriter, r *http.Request) {
+	res := GeneralResponse{Code: 200}
+
+	log.Info("api request [/api/stop]")
+	defer func() {
+		log.Info("api response [/api/stop], code [%d]", res.Code)
+		w.WriteHeader(res.Code)
+		if len(res.Msg) > 0 {
+			_, _ = w.Write([]byte(res.Msg))
+		}
+	}()
+
+	go svr.GracefulClose(100 * time.Millisecond)
+}
+
 type StatusResp map[string][]ProxyStatusResp
 
 type ProxyStatusResp struct {
@@ -106,7 +123,7 @@ func NewProxyStatusResp(status *proxy.WorkingStatus, serverAddr string) ProxySta
 	return psr
 }
 
-// GET api/status
+// GET /api/status
 func (svr *Service) apiStatus(w http.ResponseWriter, r *http.Request) {
 	var (
 		buf []byte
@@ -135,7 +152,7 @@ func (svr *Service) apiStatus(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-// GET api/config
+// GET /api/config
 func (svr *Service) apiGetConfig(w http.ResponseWriter, r *http.Request) {
 	res := GeneralResponse{Code: 200}
 
@@ -175,7 +192,7 @@ func (svr *Service) apiGetConfig(w http.ResponseWriter, r *http.Request) {
 	res.Msg = strings.Join(newRows, "\n")
 }
 
-// PUT api/config
+// PUT /api/config
 func (svr *Service) apiPutConfig(w http.ResponseWriter, r *http.Request) {
 	res := GeneralResponse{Code: 200}
 

cmd/frpc/sub/root.go

@@ -153,12 +153,11 @@ func Execute() {
 	}
 }
 
-func handleSignal(svr *client.Service, doneCh chan struct{}) {
+func handleTermSignal(svr *client.Service) {
 	ch := make(chan os.Signal, 1)
 	signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
 	<-ch
 	svr.GracefulClose(500 * time.Millisecond)
-	close(doneCh)
 }
 
 func parseClientCommonCfgFromCmd() (cfg config.ClientCommonConf, err error) {
@@ -227,16 +226,12 @@ func startService(
 		return
 	}
 
-	closedDoneCh := make(chan struct{})
 	shouldGracefulClose := cfg.Protocol == "kcp" || cfg.Protocol == "quic"
 	// Capture the exit signal if we use kcp or quic.
 	if shouldGracefulClose {
-		go handleSignal(svr, closedDoneCh)
+		go handleTermSignal(svr)
 	}
 
-	err = svr.Run(context.Background())
-	if err == nil && shouldGracefulClose {
-		<-closedDoneCh
-	}
+	_ = svr.Run(context.Background())
 	return
 }

cmd/frpc/sub/stop.go

@@ -0,0 +1,84 @@
+// Copyright 2023 The frp Authors
+//
+// 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 sub
+
+import (
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+
+	"github.com/spf13/cobra"
+
+	"github.com/fatedier/frp/pkg/config"
+)
+
+func init() {
+	rootCmd.AddCommand(stopCmd)
+}
+
+var stopCmd = &cobra.Command{
+	Use:   "stop",
+	Short: "Stop the running frpc",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cfg, _, _, err := config.ParseClientConfig(cfgFile)
+		if err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+
+		err = stopClient(cfg)
+		if err != nil {
+			fmt.Printf("frpc stop error: %v\n", err)
+			os.Exit(1)
+		}
+		fmt.Printf("stop success\n")
+		return nil
+	},
+}
+
+func stopClient(clientCfg config.ClientCommonConf) error {
+	if clientCfg.AdminPort == 0 {
+		return fmt.Errorf("admin_port shoud be set if you want to use stop feature")
+	}
+
+	req, err := http.NewRequest("POST", "http://"+
+		clientCfg.AdminAddr+":"+fmt.Sprintf("%d", clientCfg.AdminPort)+"/api/stop", nil)
+	if err != nil {
+		return err
+	}
+
+	authStr := "Basic " + base64.StdEncoding.EncodeToString([]byte(clientCfg.AdminUser+":"+
+		clientCfg.AdminPwd))
+
+	req.Header.Add("Authorization", authStr)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == 200 {
+		return nil
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+	return fmt.Errorf("code [%d], %s", resp.StatusCode, strings.TrimSpace(string(body)))
+}

server/dashboard_api.go

@@ -59,7 +59,7 @@ func (svr *Service) Healthz(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(200)
 }
 
-// api/serverinfo
+// /api/serverinfo
 func (svr *Service) APIServerInfo(w http.ResponseWriter, r *http.Request) {
 	res := GeneralResponse{Code: 200}
 	defer func() {
@@ -176,7 +176,7 @@ type GetProxyInfoResp struct {
 	Proxies []*ProxyStatsInfo `json:"proxies"`
 }
 
-// api/proxy/:type
+// /api/proxy/:type
 func (svr *Service) APIProxyByType(w http.ResponseWriter, r *http.Request) {
 	res := GeneralResponse{Code: 200}
 	params := mux.Vars(r)
@@ -244,7 +244,7 @@ type GetProxyStatsResp struct {
 	Status          string      `json:"status"`
 }
 
-// api/proxy/:type/:name
+// /api/proxy/:type/:name
 func (svr *Service) APIProxyByTypeAndName(w http.ResponseWriter, r *http.Request) {
 	res := GeneralResponse{Code: 200}
 	params := mux.Vars(r)
@@ -307,7 +307,7 @@ func (svr *Service) getProxyStatsByTypeAndName(proxyType string, proxyName strin
 	return
 }
 
-// api/traffic/:name
+// /api/traffic/:name
 type GetProxyTrafficResp struct {
 	Name       string  `json:"name"`
 	TrafficIn  []int64 `json:"traffic_in"`

test/e2e/basic/client.go

@@ -101,4 +101,32 @@ var _ = ginkgo.Describe("[Feature: ClientManage]", func() {
 		}).Port(dashboardPort).
 			Ensure(framework.ExpectResponseCode(401))
 	})
+
+	ginkgo.It("stop", func() {
+		serverConf := consts.DefaultServerConfig
+
+		adminPort := f.AllocPort()
+		testPort := f.AllocPort()
+		clientConf := consts.DefaultClientConfig + fmt.Sprintf(`
+		admin_port = %d
+
+		[test]
+		type = tcp
+		local_port = {{ .%s }}
+		remote_port = %d
+		`, adminPort, framework.TCPEchoServerPort, testPort)
+
+		f.RunProcesses([]string{serverConf}, []string{clientConf})
+
+		framework.NewRequestExpect(f).Port(testPort).Ensure()
+
+		client := clientsdk.New("127.0.0.1", adminPort)
+		err := client.Stop()
+		framework.ExpectNoError(err)
+
+		time.Sleep(3 * time.Second)
+
+		// frpc stopped so the port is not listened, expect error
+		framework.NewRequestExpect(f).Port(testPort).ExpectError(true).Ensure()
+	})
 })

test/e2e/pkg/sdk/client/client.go

@@ -62,6 +62,15 @@ func (c *Client) Reload() error {
 	return err
 }
 
+func (c *Client) Stop() error {
+	req, err := http.NewRequest("POST", "http://"+c.address+"/api/stop", nil)
+	if err != nil {
+		return err
+	}
+	_, err = c.do(req)
+	return err
+}
+
 func (c *Client) GetConfig() (string, error) {
 	req, err := http.NewRequest("GET", "http://"+c.address+"/api/config", nil)
 	if err != nil {