XiaoMo/frp
1
0
Fork 0
mirror of https://github.com/fatedier/frp.git synced 2026-01-11 22:23:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

use new log package (#4054)

Browse Source
This commit is contained in:
fatedier
2024-03-12 13:58:53 +08:00
committed by GitHub
parent 49443cb2c6
commit 43ba7bd338
51 changed files with 389 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
client/visitor/stcp.go
View File

@@ -56,7 +56,7 @@ func (sv *STCPVisitor) worker() {
for {
conn, err := sv.l.Accept()
if err != nil {
xl.Warn("stcp local listener closed")
xl.Warnf("stcp local listener closed")
return
}
go sv.handleConn(conn)
@@ -68,7 +68,7 @@ func (sv *STCPVisitor) internalConnWorker() {
for {
conn, err := sv.internalLn.Accept()
if err != nil {
xl.Warn("stcp internal listener closed")
xl.Warnf("stcp internal listener closed")
return
}
go sv.handleConn(conn)
@@ -79,7 +79,7 @@ func (sv *STCPVisitor) handleConn(userConn net.Conn) {
xl := xlog.FromContextSafe(sv.ctx)
defer userConn.Close()
xl.Debug("get a new stcp user connection")
xl.Debugf("get a new stcp user connection")
visitorConn, err := sv.helper.ConnectServer()
if err != nil {
return
@@ -97,7 +97,7 @@ func (sv *STCPVisitor) handleConn(userConn net.Conn) {
}
err = msg.WriteMsg(visitorConn, newVisitorConnMsg)
if err != nil {
xl.Warn("send newVisitorConnMsg to server error: %v", err)
xl.Warnf("send newVisitorConnMsg to server error: %v", err)
return
}
@@ -105,13 +105,13 @@ func (sv *STCPVisitor) handleConn(userConn net.Conn) {
_ = visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
err = msg.ReadMsgInto(visitorConn, &newVisitorConnRespMsg)
if err != nil {
xl.Warn("get newVisitorConnRespMsg error: %v", err)
xl.Warnf("get newVisitorConnRespMsg error: %v", err)
return
}
_ = visitorConn.SetReadDeadline(time.Time{})
if newVisitorConnRespMsg.Error != "" {
xl.Warn("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
xl.Warnf("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
return
}
@@ -120,7 +120,7 @@ func (sv *STCPVisitor) handleConn(userConn net.Conn) {
if sv.cfg.Transport.UseEncryption {
remote, err = libio.WithEncryption(remote, []byte(sv.cfg.SecretKey))
if err != nil {
xl.Error("create encryption stream error: %v", err)
xl.Errorf("create encryption stream error: %v", err)
return
}
}

32
client/visitor/sudp.go
View File

@@ -62,7 +62,7 @@ func (sv *SUDPVisitor) Run() (err error) {
sv.sendCh = make(chan *msg.UDPPacket, 1024)
sv.readCh = make(chan *msg.UDPPacket, 1024)
xl.Info("sudp start to work, listen on %s", addr)
xl.Infof("sudp start to work, listen on %s", addr)
go sv.dispatcher()
go udp.ForwardUserConn(sv.udpConn, sv.readCh, sv.sendCh, int(sv.clientCfg.UDPPacketSize))
@@ -84,17 +84,17 @@ func (sv *SUDPVisitor) dispatcher() {
select {
case firstPacket = <-sv.sendCh:
if firstPacket == nil {
xl.Info("frpc sudp visitor proxy is closed")
xl.Infof("frpc sudp visitor proxy is closed")
return
}
case <-sv.checkCloseCh:
xl.Info("frpc sudp visitor proxy is closed")
xl.Infof("frpc sudp visitor proxy is closed")
return
}
visitorConn, err = sv.getNewVisitorConn()
if err != nil {
xl.Warn("newVisitorConn to frps error: %v, try to reconnect", err)
xl.Warnf("newVisitorConn to frps error: %v, try to reconnect", err)
continue
}
@@ -111,7 +111,7 @@ func (sv *SUDPVisitor) dispatcher() {
func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) {
xl := xlog.FromContextSafe(sv.ctx)
xl.Debug("starting sudp proxy worker")
xl.Debugf("starting sudp proxy worker")
wg := &sync.WaitGroup{}
wg.Add(2)
@@ -134,21 +134,21 @@ func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) {
// frpc will send heartbeat in workConn to frpc visitor for keeping alive
_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
if rawMsg, errRet = msg.ReadMsg(conn); errRet != nil {
xl.Warn("read from workconn for user udp conn error: %v", errRet)
xl.Warnf("read from workconn for user udp conn error: %v", errRet)
return
}
_ = conn.SetReadDeadline(time.Time{})
switch m := rawMsg.(type) {
case *msg.Ping:
xl.Debug("frpc visitor get ping message from frpc")
xl.Debugf("frpc visitor get ping message from frpc")
continue
case *msg.UDPPacket:
if errRet := errors.PanicToError(func() {
sv.readCh <- m
xl.Trace("frpc visitor get udp packet from workConn: %s", m.Content)
xl.Tracef("frpc visitor get udp packet from workConn: %s", m.Content)
}); errRet != nil {
xl.Info("reader goroutine for udp work connection closed")
xl.Infof("reader goroutine for udp work connection closed")
return
}
}
@@ -165,25 +165,25 @@ func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) {
var errRet error
if firstPacket != nil {
if errRet = msg.WriteMsg(conn, firstPacket); errRet != nil {
xl.Warn("sender goroutine for udp work connection closed: %v", errRet)
xl.Warnf("sender goroutine for udp work connection closed: %v", errRet)
return
}
xl.Trace("send udp package to workConn: %s", firstPacket.Content)
xl.Tracef("send udp package to workConn: %s", firstPacket.Content)
}
for {
select {
case udpMsg, ok := <-sv.sendCh:
if !ok {
xl.Info("sender goroutine for udp work connection closed")
xl.Infof("sender goroutine for udp work connection closed")
return
}
if errRet = msg.WriteMsg(conn, udpMsg); errRet != nil {
xl.Warn("sender goroutine for udp work connection closed: %v", errRet)
xl.Warnf("sender goroutine for udp work connection closed: %v", errRet)
return
}
xl.Trace("send udp package to workConn: %s", udpMsg.Content)
xl.Tracef("send udp package to workConn: %s", udpMsg.Content)
case <-closeCh:
return
}
@@ -194,7 +194,7 @@ func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) {
go workConnSenderFn(workConn)
wg.Wait()
xl.Info("sudp worker is closed")
xl.Infof("sudp worker is closed")
}
func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, error) {
@@ -235,7 +235,7 @@ func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, error) {
if sv.cfg.Transport.UseEncryption {
remote, err = libio.WithEncryption(remote, []byte(sv.cfg.SecretKey))
if err != nil {
xl.Error("create encryption stream error: %v", err)
xl.Errorf("create encryption stream error: %v", err)
return nil, err
}
}

12
client/visitor/visitor_manager.go
View File

@@ -79,14 +79,14 @@ func (vm *Manager) keepVisitorsRunning() {
for {
select {
case <-vm.stopCh:
xl.Trace("gracefully shutdown visitor manager")
xl.Tracef("gracefully shutdown visitor manager")
return
case <-ticker.C:
vm.mu.Lock()
for _, cfg := range vm.cfgs {
name := cfg.GetBaseConfig().Name
if _, exist := vm.visitors[name]; !exist {
xl.Info("try to start visitor [%s]", name)
xl.Infof("try to start visitor [%s]", name)
_ = vm.startVisitor(cfg)
}
}
@@ -115,10 +115,10 @@ func (vm *Manager) startVisitor(cfg v1.VisitorConfigurer) (err error) {
visitor := NewVisitor(vm.ctx, cfg, vm.clientCfg, vm.helper)
err = visitor.Run()
if err != nil {
xl.Warn("start error: %v", err)
xl.Warnf("start error: %v", err)
} else {
vm.visitors[name] = visitor
xl.Info("start visitor success")
xl.Infof("start visitor success")
}
return
}
@@ -156,7 +156,7 @@ func (vm *Manager) UpdateAll(cfgs []v1.VisitorConfigurer) {
}
}
if len(delNames) > 0 {
xl.Info("visitor removed: %v", delNames)
xl.Infof("visitor removed: %v", delNames)
}
addNames := make([]string, 0)
@@ -169,7 +169,7 @@ func (vm *Manager) UpdateAll(cfgs []v1.VisitorConfigurer) {
}
}
if len(addNames) > 0 {
xl.Info("visitor added: %v", addNames)
xl.Infof("visitor added: %v", addNames)
}
}

48
client/visitor/xtcp.go
View File

@@ -93,7 +93,7 @@ func (sv *XTCPVisitor) worker() {
for {
conn, err := sv.l.Accept()
if err != nil {
xl.Warn("xtcp local listener closed")
xl.Warnf("xtcp local listener closed")
return
}
go sv.handleConn(conn)
@@ -105,7 +105,7 @@ func (sv *XTCPVisitor) internalConnWorker() {
for {
conn, err := sv.internalLn.Accept()
if err != nil {
xl.Warn("xtcp internal listener closed")
xl.Warnf("xtcp internal listener closed")
return
}
go sv.handleConn(conn)
@@ -140,14 +140,14 @@ func (sv *XTCPVisitor) keepTunnelOpenWorker() {
case <-sv.ctx.Done():
return
case <-ticker.C:
xl.Debug("keepTunnelOpenWorker try to check tunnel...")
xl.Debugf("keepTunnelOpenWorker try to check tunnel...")
conn, err := sv.getTunnelConn()
if err != nil {
xl.Warn("keepTunnelOpenWorker get tunnel connection error: %v", err)
xl.Warnf("keepTunnelOpenWorker get tunnel connection error: %v", err)
_ = sv.retryLimiter.Wait(sv.ctx)
continue
}
xl.Debug("keepTunnelOpenWorker check success")
xl.Debugf("keepTunnelOpenWorker check success")
if conn != nil {
conn.Close()
}
@@ -164,7 +164,7 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
}
}()
xl.Debug("get a new xtcp user connection")
xl.Debugf("get a new xtcp user connection")
// Open a tunnel connection to the server. If there is already a successful hole-punching connection,
// it will be reused. Otherwise, it will block and wait for a successful hole-punching connection until timeout.
@@ -176,15 +176,15 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
}
tunnelConn, err := sv.openTunnel(ctx)
if err != nil {
xl.Error("open tunnel error: %v", err)
xl.Errorf("open tunnel error: %v", err)
// no fallback, just return
if sv.cfg.FallbackTo == "" {
return
}
xl.Debug("try to transfer connection to visitor: %s", sv.cfg.FallbackTo)
xl.Debugf("try to transfer connection to visitor: %s", sv.cfg.FallbackTo)
if err := sv.helper.TransferConn(sv.cfg.FallbackTo, userConn); err != nil {
xl.Error("transfer connection to visitor %s error: %v", sv.cfg.FallbackTo, err)
xl.Errorf("transfer connection to visitor %s error: %v", sv.cfg.FallbackTo, err)
return
}
isConnTrasfered = true
@@ -195,7 +195,7 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
if sv.cfg.Transport.UseEncryption {
muxConnRWCloser, err = libio.WithEncryption(muxConnRWCloser, []byte(sv.cfg.SecretKey))
if err != nil {
xl.Error("create encryption stream error: %v", err)
xl.Errorf("create encryption stream error: %v", err)
return
}
}
@@ -206,9 +206,9 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
}
_, _, errs := libio.Join(userConn, muxConnRWCloser)
xl.Debug("join connections closed")
xl.Debugf("join connections closed")
if len(errs) > 0 {
xl.Trace("join connections errors: %v", errs)
xl.Tracef("join connections errors: %v", errs)
}
}
@@ -239,7 +239,7 @@ func (sv *XTCPVisitor) openTunnel(ctx context.Context) (conn net.Conn, err error
if err != nil {
if err != ErrNoTunnelSession {
xl.Warn("get tunnel connection error: %v", err)
xl.Warnf("get tunnel connection error: %v", err)
}
continue
}
@@ -268,19 +268,19 @@ func (sv *XTCPVisitor) getTunnelConn() (net.Conn, error) {
// 4. Create a tunnel session using an underlying UDP connection.
func (sv *XTCPVisitor) makeNatHole() {
xl := xlog.FromContextSafe(sv.ctx)
xl.Trace("makeNatHole start")
xl.Tracef("makeNatHole start")
if err := nathole.PreCheck(sv.ctx, sv.helper.MsgTransporter(), sv.cfg.ServerName, 5*time.Second); err != nil {
xl.Warn("nathole precheck error: %v", err)
xl.Warnf("nathole precheck error: %v", err)
return
}
xl.Trace("nathole prepare start")
xl.Tracef("nathole prepare start")
prepareResult, err := nathole.Prepare([]string{sv.clientCfg.NatHoleSTUNServer})
if err != nil {
xl.Warn("nathole prepare error: %v", err)
xl.Warnf("nathole prepare error: %v", err)
return
}
xl.Info("nathole prepare success, nat type: %s, behavior: %s, addresses: %v, assistedAddresses: %v",
xl.Infof("nathole prepare success, nat type: %s, behavior: %s, addresses: %v, assistedAddresses: %v",
prepareResult.NatType, prepareResult.Behavior, prepareResult.Addrs, prepareResult.AssistedAddrs)
listenConn := prepareResult.ListenConn
@@ -298,30 +298,30 @@ func (sv *XTCPVisitor) makeNatHole() {
AssistedAddrs: prepareResult.AssistedAddrs,
}
xl.Trace("nathole exchange info start")
xl.Tracef("nathole exchange info start")
natHoleRespMsg, err := nathole.ExchangeInfo(sv.ctx, sv.helper.MsgTransporter(), transactionID, natHoleVisitorMsg, 5*time.Second)
if err != nil {
listenConn.Close()
xl.Warn("nathole exchange info error: %v", err)
xl.Warnf("nathole exchange info error: %v", err)
return
}
xl.Info("get natHoleRespMsg, sid [%s], protocol [%s], candidate address %v, assisted address %v, detectBehavior: %+v",
xl.Infof("get natHoleRespMsg, sid [%s], protocol [%s], candidate address %v, assisted address %v, detectBehavior: %+v",
natHoleRespMsg.Sid, natHoleRespMsg.Protocol, natHoleRespMsg.CandidateAddrs,
natHoleRespMsg.AssistedAddrs, natHoleRespMsg.DetectBehavior)
newListenConn, raddr, err := nathole.MakeHole(sv.ctx, listenConn, natHoleRespMsg, []byte(sv.cfg.SecretKey))
if err != nil {
listenConn.Close()
xl.Warn("make hole error: %v", err)
xl.Warnf("make hole error: %v", err)
return
}
listenConn = newListenConn
xl.Info("establishing nat hole connection successful, sid [%s], remoteAddr [%s]", natHoleRespMsg.Sid, raddr)
xl.Infof("establishing nat hole connection successful, sid [%s], remoteAddr [%s]", natHoleRespMsg.Sid, raddr)
if err := sv.session.Init(listenConn, raddr); err != nil {
listenConn.Close()
xl.Warn("init tunnel session error: %v", err)
xl.Warnf("init tunnel session error: %v", err)
return
}
}