frp/client/proxy/proxy_wrapper.go

280 lines
6.6 KiB
Go
Raw Normal View History

// 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.
2018-12-09 22:06:22 +08:00
package proxy
2018-12-07 17:05:36 +08:00
import (
2019-10-12 20:13:12 +08:00
"context"
2018-12-07 17:05:36 +08:00
"fmt"
2019-10-12 20:13:12 +08:00
"net"
"strconv"
2018-12-07 17:05:36 +08:00
"sync"
"sync/atomic"
"time"
2022-08-29 01:02:53 +08:00
"github.com/fatedier/golib/errors"
2018-12-09 22:06:22 +08:00
"github.com/fatedier/frp/client/event"
"github.com/fatedier/frp/client/health"
v1 "github.com/fatedier/frp/pkg/config/v1"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/msg"
"github.com/fatedier/frp/pkg/transport"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/util/xlog"
2018-12-07 17:05:36 +08:00
)
const (
2020-05-24 17:48:37 +08:00
ProxyPhaseNew = "new"
ProxyPhaseWaitStart = "wait start"
ProxyPhaseStartErr = "start error"
ProxyPhaseRunning = "running"
ProxyPhaseCheckFailed = "check failed"
ProxyPhaseClosed = "closed"
2018-12-07 17:05:36 +08:00
)
var (
2022-08-29 01:02:53 +08:00
statusCheckInterval = 3 * time.Second
waitResponseTimeout = 20 * time.Second
startErrTimeout = 30 * time.Second
2018-12-07 17:05:36 +08:00
)
2020-05-24 17:48:37 +08:00
type WorkingStatus struct {
Name string `json:"name"`
Type string `json:"type"`
Phase string `json:"status"`
Err string `json:"err"`
Cfg v1.ProxyConfigurer `json:"cfg"`
2018-12-07 17:05:36 +08:00
// Got from server.
RemoteAddr string `json:"remote_addr"`
}
2020-05-24 17:48:37 +08:00
type Wrapper struct {
WorkingStatus
2018-12-07 17:05:36 +08:00
// underlying proxy
pxy Proxy
// if ProxyConf has healcheck config
// monitor will watch if it is alive
2020-05-24 17:48:37 +08:00
monitor *health.Monitor
2018-12-07 17:05:36 +08:00
// event handler
2020-05-24 17:48:37 +08:00
handler event.Handler
2018-12-07 17:05:36 +08:00
msgTransporter transport.MessageTransporter
2018-12-07 17:05:36 +08:00
health uint32
lastSendStartMsg time.Time
lastStartErr time.Time
closeCh chan struct{}
2018-12-09 21:56:46 +08:00
healthNotifyCh chan struct{}
2018-12-07 17:05:36 +08:00
mu sync.RWMutex
2019-10-12 20:13:12 +08:00
xl *xlog.Logger
ctx context.Context
2018-12-07 17:05:36 +08:00
}
func NewWrapper(
ctx context.Context,
cfg v1.ProxyConfigurer,
clientCfg *v1.ClientCommonConfig,
eventHandler event.Handler,
msgTransporter transport.MessageTransporter,
) *Wrapper {
2023-05-30 20:25:22 +08:00
baseInfo := cfg.GetBaseConfig()
xl := xlog.FromContextSafe(ctx).Spawn().AppendPrefix(baseInfo.Name)
2020-05-24 17:48:37 +08:00
pw := &Wrapper{
WorkingStatus: WorkingStatus{
Name: baseInfo.Name,
Type: baseInfo.Type,
2020-05-24 17:48:37 +08:00
Phase: ProxyPhaseNew,
Cfg: cfg,
2018-12-07 17:05:36 +08:00
},
2018-12-09 21:56:46 +08:00
closeCh: make(chan struct{}),
healthNotifyCh: make(chan struct{}),
handler: eventHandler,
msgTransporter: msgTransporter,
2019-10-12 20:13:12 +08:00
xl: xl,
ctx: xlog.NewContext(ctx, xl),
2018-12-07 17:05:36 +08:00
}
if baseInfo.HealthCheck.Type != "" && baseInfo.LocalPort > 0 {
2018-12-07 17:05:36 +08:00
pw.health = 1 // means failed
addr := net.JoinHostPort(baseInfo.LocalIP, strconv.Itoa(baseInfo.LocalPort))
pw.monitor = health.NewMonitor(pw.ctx, baseInfo.HealthCheck, addr,
pw.statusNormalCallback, pw.statusFailedCallback)
2019-10-12 20:13:12 +08:00
xl.Trace("enable health check monitor")
2018-12-07 17:05:36 +08:00
}
pw.pxy = NewProxy(pw.ctx, pw.Cfg, clientCfg, pw.msgTransporter)
2018-12-07 17:05:36 +08:00
return pw
}
2023-11-21 11:19:35 +08:00
func (pw *Wrapper) SetInWorkConnCallback(cb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool) {
pw.pxy.SetInWorkConnCallback(cb)
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) SetRunningStatus(remoteAddr string, respErr string) error {
2018-12-07 17:05:36 +08:00
pw.mu.Lock()
defer pw.mu.Unlock()
2020-05-24 17:48:37 +08:00
if pw.Phase != ProxyPhaseWaitStart {
2018-12-07 17:05:36 +08:00
return fmt.Errorf("status not wait start, ignore start message")
}
pw.RemoteAddr = remoteAddr
if respErr != "" {
2020-05-24 17:48:37 +08:00
pw.Phase = ProxyPhaseStartErr
2018-12-07 17:05:36 +08:00
pw.Err = respErr
pw.lastStartErr = time.Now()
return fmt.Errorf(pw.Err)
}
if err := pw.pxy.Run(); err != nil {
pw.close()
2020-05-24 17:48:37 +08:00
pw.Phase = ProxyPhaseStartErr
2018-12-07 17:05:36 +08:00
pw.Err = err.Error()
pw.lastStartErr = time.Now()
return err
}
2020-05-24 17:48:37 +08:00
pw.Phase = ProxyPhaseRunning
2018-12-07 17:05:36 +08:00
pw.Err = ""
return nil
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) Start() {
2018-12-07 17:05:36 +08:00
go pw.checkWorker()
if pw.monitor != nil {
go pw.monitor.Start()
}
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) Stop() {
2018-12-07 17:05:36 +08:00
pw.mu.Lock()
defer pw.mu.Unlock()
2018-12-09 21:56:46 +08:00
close(pw.closeCh)
close(pw.healthNotifyCh)
2018-12-07 17:05:36 +08:00
pw.pxy.Close()
if pw.monitor != nil {
pw.monitor.Stop()
}
2020-05-24 17:48:37 +08:00
pw.Phase = ProxyPhaseClosed
pw.close()
}
2018-12-07 17:05:36 +08:00
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) close() {
2022-08-29 01:02:53 +08:00
_ = pw.handler(&event.CloseProxyPayload{
2018-12-07 17:05:36 +08:00
CloseProxyMsg: &msg.CloseProxy{
ProxyName: pw.Name,
},
})
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) checkWorker() {
2019-10-12 20:13:12 +08:00
xl := pw.xl
2018-12-09 21:56:46 +08:00
if pw.monitor != nil {
// let monitor do check request first
time.Sleep(500 * time.Millisecond)
}
2018-12-07 17:05:36 +08:00
for {
// check proxy status
now := time.Now()
if atomic.LoadUint32(&pw.health) == 0 {
pw.mu.Lock()
2020-05-24 17:48:37 +08:00
if pw.Phase == ProxyPhaseNew ||
pw.Phase == ProxyPhaseCheckFailed ||
(pw.Phase == ProxyPhaseWaitStart && now.After(pw.lastSendStartMsg.Add(waitResponseTimeout))) ||
(pw.Phase == ProxyPhaseStartErr && now.After(pw.lastStartErr.Add(startErrTimeout))) {
2018-12-07 17:05:36 +08:00
2020-05-24 17:48:37 +08:00
xl.Trace("change status from [%s] to [%s]", pw.Phase, ProxyPhaseWaitStart)
pw.Phase = ProxyPhaseWaitStart
2018-12-07 17:05:36 +08:00
var newProxyMsg msg.NewProxy
pw.Cfg.MarshalToMsg(&newProxyMsg)
pw.lastSendStartMsg = now
2022-08-29 01:02:53 +08:00
_ = pw.handler(&event.StartProxyPayload{
2018-12-07 17:05:36 +08:00
NewProxyMsg: &newProxyMsg,
})
}
pw.mu.Unlock()
} else {
pw.mu.Lock()
2020-05-24 17:48:37 +08:00
if pw.Phase == ProxyPhaseRunning || pw.Phase == ProxyPhaseWaitStart {
pw.close()
2020-05-24 17:48:37 +08:00
xl.Trace("change status from [%s] to [%s]", pw.Phase, ProxyPhaseCheckFailed)
pw.Phase = ProxyPhaseCheckFailed
2018-12-07 17:05:36 +08:00
}
pw.mu.Unlock()
}
select {
case <-pw.closeCh:
return
case <-time.After(statusCheckInterval):
2018-12-09 21:56:46 +08:00
case <-pw.healthNotifyCh:
2018-12-07 17:05:36 +08:00
}
}
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) statusNormalCallback() {
2019-10-12 20:13:12 +08:00
xl := pw.xl
2018-12-07 17:05:36 +08:00
atomic.StoreUint32(&pw.health, 0)
2022-08-29 01:02:53 +08:00
_ = errors.PanicToError(func() {
2018-12-09 21:56:46 +08:00
select {
case pw.healthNotifyCh <- struct{}{}:
default:
}
})
2019-10-12 20:13:12 +08:00
xl.Info("health check success")
2018-12-07 17:05:36 +08:00
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) statusFailedCallback() {
2019-10-12 20:13:12 +08:00
xl := pw.xl
2018-12-07 17:05:36 +08:00
atomic.StoreUint32(&pw.health, 1)
2022-08-29 01:02:53 +08:00
_ = errors.PanicToError(func() {
2018-12-09 21:56:46 +08:00
select {
case pw.healthNotifyCh <- struct{}{}:
default:
}
})
2019-10-12 20:13:12 +08:00
xl.Info("health check failed")
2018-12-07 17:05:36 +08:00
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) InWorkConn(workConn net.Conn, m *msg.StartWorkConn) {
2019-10-12 20:13:12 +08:00
xl := pw.xl
2018-12-07 17:05:36 +08:00
pw.mu.RLock()
pxy := pw.pxy
pw.mu.RUnlock()
2021-03-31 16:57:39 +08:00
if pxy != nil && pw.Phase == ProxyPhaseRunning {
2019-10-12 20:13:12 +08:00
xl.Debug("start a new work connection, localAddr: %s remoteAddr: %s", workConn.LocalAddr().String(), workConn.RemoteAddr().String())
2019-03-29 19:01:18 +08:00
go pxy.InWorkConn(workConn, m)
2018-12-07 17:05:36 +08:00
} else {
workConn.Close()
}
}
2020-05-24 17:48:37 +08:00
func (pw *Wrapper) GetStatus() *WorkingStatus {
2018-12-07 17:05:36 +08:00
pw.mu.RLock()
defer pw.mu.RUnlock()
2020-05-24 17:48:37 +08:00
ps := &WorkingStatus{
2018-12-07 17:05:36 +08:00
Name: pw.Name,
Type: pw.Type,
2020-05-24 17:48:37 +08:00
Phase: pw.Phase,
2018-12-07 17:05:36 +08:00
Err: pw.Err,
Cfg: pw.Cfg,
RemoteAddr: pw.RemoteAddr,
}
return ps
}