diff --git a/internal/proxy/bondclient/bondclient.go b/internal/proxy/bondclient/bondclient.go index 5ed32c4..8be3fd8 100644 --- a/internal/proxy/bondclient/bondclient.go +++ b/internal/proxy/bondclient/bondclient.go @@ -8,7 +8,6 @@ import ( "context" "fmt" "io" - "log" "net" "strconv" "strings" @@ -64,11 +63,11 @@ func (h *Handler) Handle(ctx context.Context, tcpConn net.Conn, connID uint64, c } stream, err := ps.Sess.OpenStream() if err != nil { - log.Printf("[bond %d] session %d open stream error: %s", connID, ps.ID, err) + h.Deps.log().Errorf("[bond %d] session %d open stream error: %s", connID, ps.ID, err) continue } if err := bondframe.WriteHello(stream, connID, uint16(i), uint16(len(candidates))); err != nil { - log.Printf("[bond %d] session %d hello error: %s", connID, ps.ID, err) + h.Deps.log().Errorf("[bond %d] session %d hello error: %s", connID, ps.ID, err) _ = stream.Close() continue } @@ -79,7 +78,7 @@ func (h *Handler) Handle(ctx context.Context, tcpConn net.Conn, connID uint64, c } if len(lanes) == 0 { - log.Printf("[bond %d] no usable lanes, rejecting TCP from %s", connID, tcpConn.RemoteAddr()) + h.Deps.log().Errorf("[bond %d] no usable lanes, rejecting TCP from %s", connID, tcpConn.RemoteAddr()) return } context.AfterFunc(ctx, func() { @@ -159,7 +158,7 @@ func (h *Handler) copyTCPToBond(ctx context.Context, connID uint64, tcpConn net. if n > 0 { l, writeErr := writeBondFrameToNextLane(ctx, lanes, bondframe.FrameData, seq, buf[:n], &laneIdx) if writeErr != nil { - log.Printf("[bond %d] write data error: %v", connID, writeErr) + h.Deps.log().Errorf("[bond %d] write data error: %v", connID, writeErr) return } l.ps.ToSession.Add(uint64(n)) @@ -177,7 +176,7 @@ func (h *Handler) copyTCPToBond(ctx context.Context, connID uint64, tcpConn net. writeErr := bondframe.WriteFrame(l.stream, bondframe.FrameFIN, seq, nil) l.mu.Unlock() if writeErr != nil && ctx.Err() == nil { - log.Printf("[bond %d] session %d write FIN error: %v", connID, l.ps.ID, writeErr) + h.Deps.log().Errorf("[bond %d] session %d write FIN error: %v", connID, l.ps.ID, writeErr) } } h.Deps.log().Debugf("[bond %d] upload finished chunks=%d", connID, seq) @@ -238,7 +237,7 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net. switch f.Type { case bondframe.FrameData: if len(pending) >= 1024 { - log.Printf("[bond %d] pending map overflow (>1024), closing", connID) + h.Deps.log().Errorf("[bond %d] pending map overflow (>1024), closing", connID) return } pending[f.Seq] = f.Data @@ -248,7 +247,7 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net. finSeq = &v } default: - log.Printf("[bond %d] unknown frame type %d", connID, f.Type) + h.Deps.log().Errorf("[bond %d] unknown frame type %d", connID, f.Type) return } @@ -260,7 +259,7 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net. delete(pending, expect) if len(data) > 0 { if _, err := tcpConn.Write(data); err != nil { - log.Printf("[bond %d] local TCP write error: %v", connID, err) + h.Deps.log().Errorf("[bond %d] local TCP write error: %v", connID, err) return } } diff --git a/internal/proxy/bondserver/bondserver.go b/internal/proxy/bondserver/bondserver.go index f41edd0..35528bb 100644 --- a/internal/proxy/bondserver/bondserver.go +++ b/internal/proxy/bondserver/bondserver.go @@ -7,7 +7,6 @@ package bondserver import ( "context" "io" - "log" "net" "sync" "time" @@ -64,13 +63,13 @@ func (r *Registry) HandleStreamAfterMagic(ctx context.Context, stream *smux.Stre func (r *Registry) handleStream(ctx context.Context, stream *smux.Stream, connectAddr string, readHello func(io.Reader) (bondframe.Hello, error)) { defer func() { if err := stream.Close(); err != nil && err != smux.ErrGoAway { - log.Printf("failed to close bond smux stream: %v", err) + r.deps.log().Errorf("bondserver: close smux stream: %v", err) } }() hello, err := readHello(stream) if err != nil { - log.Printf("bond hello error: %v", err) + r.deps.log().Errorf("bondserver: bond hello: %v", err) return } @@ -233,22 +232,22 @@ func (c *serverConn) run() { backendConn, err := net.DialTimeout("tcp", c.connectAddr, 10*time.Second) if err != nil { - log.Printf("[bond %d] backend dial error: %s", c.id, err) + c.deps.log().Errorf("[bond %d] backend dial: %s", c.id, err) return } defer func() { if err := backendConn.Close(); err != nil { - log.Printf("[bond %d] failed to close backend connection: %v", c.id, err) + c.deps.log().Errorf("[bond %d] close backend connection: %v", c.id, err) } }() context.AfterFunc(c.ctx, func() { now := time.Now() if err := backendConn.SetDeadline(now); err != nil { - log.Printf("[bond %d] backend deadline error: %v", c.id, err) + c.deps.log().Errorf("[bond %d] backend deadline: %v", c.id, err) } for _, lane := range c.snapshotLanes() { if err := lane.stream.SetDeadline(now); err != nil { - log.Printf("[bond %d] lane %d deadline error: %v", c.id, lane.index, err) + c.deps.log().Errorf("[bond %d] lane %d deadline: %v", c.id, lane.index, err) } } }) @@ -287,7 +286,7 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) { // that emits seq with permanent gaps cannot exhaust memory. // Matches bondclient.copyBondToTCP. if len(pending) >= 1024 { - log.Printf("[bond %d] pending map overflow (>1024), closing", c.id) + c.deps.log().Errorf("[bond %d] pending map overflow (>1024), closing", c.id) return } pending[f.Seq] = f.Data @@ -297,7 +296,7 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) { finSeq = &v } default: - log.Printf("[bond %d] unknown frame type %d", c.id, f.Type) + c.deps.log().Errorf("[bond %d] unknown frame type %d", c.id, f.Type) return } @@ -309,7 +308,7 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) { delete(pending, expect) if len(data) > 0 { if _, err := backendConn.Write(data); err != nil { - log.Printf("[bond %d] backend write error: %v", c.id, err) + c.deps.log().Errorf("[bond %d] backend write error: %v", c.id, err) return } } @@ -329,7 +328,7 @@ func (c *serverConn) copyBackendToBond(backendConn net.Conn) { data := make([]byte, n) copy(data, buf[:n]) if writeErr := c.writeToNextLane(bondframe.FrameData, seq, data, &laneIdx); writeErr != nil { - log.Printf("[bond %d] lane write data error: %v", c.id, writeErr) + c.deps.log().Errorf("[bond %d] lane write data error: %v", c.id, writeErr) return } seq++ @@ -341,7 +340,7 @@ func (c *serverConn) copyBackendToBond(backendConn net.Conn) { writeErr := bondframe.WriteFrame(lane.stream, bondframe.FrameFIN, seq, nil) lane.mu.Unlock() if writeErr != nil && c.ctx.Err() == nil { - log.Printf("[bond %d] lane %d write FIN error: %v", c.id, lane.index, writeErr) + c.deps.log().Errorf("[bond %d] lane %d write FIN error: %v", c.id, lane.index, writeErr) } } c.deps.log().Debugf("[bond %d] download from backend finished chunks=%d", c.id, seq) @@ -368,7 +367,7 @@ func (c *serverConn) writeToNextLane(typ byte, seq uint64, data []byte, laneIdx return nil } left := c.removeLane(lane) - log.Printf("[bond %d] lane %d write error: %v (lanes=%d)", c.id, lane.index, err, left) + c.deps.log().Errorf("[bond %d] lane %d write error: %v (lanes=%d)", c.id, lane.index, err, left) if left == 0 { return err } diff --git a/internal/proxy/tcpfwd/tcpfwd.go b/internal/proxy/tcpfwd/tcpfwd.go index ada0361..67826a4 100644 --- a/internal/proxy/tcpfwd/tcpfwd.go +++ b/internal/proxy/tcpfwd/tcpfwd.go @@ -70,7 +70,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis }) } - log.Printf("VLESS mode: waiting for sessions to connect (total: %d)...", numSessions) + deps.log().Errorf("VLESS mode: waiting for sessions to connect (total: %d)...", numSessions) for { select { case <-ctx.Done(): @@ -90,15 +90,15 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis wrappedListener, err := ish.WrapListener(listener) if err != nil { - log.Printf("Warning: failed to wrap listener: %v", err) + deps.log().Errorf("Warning: failed to wrap listener: %v", err) wrappedListener = listener } context.AfterFunc(ctx, func() { _ = wrappedListener.Close() }) if useBond { - log.Printf("VLESS bond mode: listening on %s (striping each TCP connection across active sessions)", listenAddr) + deps.log().Errorf("VLESS bond mode: listening on %s (striping each TCP connection across active sessions)", listenAddr) } else { - log.Printf("VLESS mode: listening on %s (round-robin across %d sessions)", listenAddr, numSessions) + deps.log().Errorf("VLESS mode: listening on %s (round-robin across %d sessions)", listenAddr, numSessions) } var wgConn sync.WaitGroup @@ -112,20 +112,20 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis return default: } - log.Printf("TCP accept error: %s", err) + deps.log().Errorf("TCP accept error: %s", err) continue } if useBond { if deps.BondHandler == nil { - log.Printf("bond requested but no BondHandler set, rejecting") + deps.log().Errorf("bond requested but no BondHandler set, rejecting") _ = tcpConn.Close() continue } connID := (uint64(time.Now().UnixNano()) << 16) ^ pool.NextConnID() lanes := pool.Snapshot() if len(lanes) == 0 { - log.Printf("No active sessions, rejecting connection") + deps.log().Errorf("No active sessions, rejecting connection") _ = tcpConn.Close() continue } @@ -139,7 +139,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis ps := pool.Pick() if ps == nil || ps.Sess.IsClosed() { - log.Printf("No active sessions, rejecting connection") + deps.log().Errorf("No active sessions, rejecting connection") _ = tcpConn.Close() continue } @@ -163,7 +163,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis stream, err := sessRef.Sess.OpenStream() if err != nil { - log.Printf("[session %d] smux open stream error for TCP #%d: %s", sessRef.ID, cid, err) + deps.log().Errorf("[session %d] smux open stream error for TCP #%d: %s", sessRef.ID, cid, err) return } defer func() { _ = stream.Close() }() @@ -188,7 +188,7 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net. smuxSess, cleanup, err := createSmuxSession(ctx, deps, params, peer, id) if err != nil { - log.Printf("[session %d] setup error: %s, retrying...", id, err) + deps.log().Errorf("[session %d] setup error: %s, retrying...", id, err) select { case <-ctx.Done(): return @@ -198,7 +198,7 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net. } ps := pool.Add(id, smuxSess) - log.Printf("[session %d] connected (active: %d)", id, pool.Count()) + deps.log().Errorf("[session %d] connected (active: %d)", id, pool.Count()) for !smuxSess.IsClosed() { select { @@ -212,7 +212,7 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net. pool.Remove(ps) cleanup() - log.Printf("[session %d] disconnected (active: %d), reconnecting...", id, pool.Count()) + deps.log().Errorf("[session %d] disconnected (active: %d), reconnecting...", id, pool.Count()) select { case <-ctx.Done(): @@ -285,10 +285,10 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) { ctx2, cancel := context.WithCancel(ctx) context.AfterFunc(ctx2, func() { if err := c1.SetDeadline(time.Now()); err != nil { - log.Printf("pipe: failed to set deadline c1: %v", err) + deps.log().Errorf("pipe: failed to set deadline c1: %v", err) } if err := c2.SetDeadline(time.Now()); err != nil { - log.Printf("pipe: failed to set deadline c2: %v", err) + deps.log().Errorf("pipe: failed to set deadline c2: %v", err) } }) @@ -300,7 +300,7 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) { n, err := io.Copy(c1, c2) c1FromC2 = n if err != nil && deps.log().DebugEnabled() { - log.Printf("pipe: c1<-c2 copy error: %v", err) + deps.log().Errorf("pipe: c1<-c2 copy error: %v", err) } }) wg.Go(func() { @@ -308,15 +308,15 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) { n, err := io.Copy(c2, c1) c2FromC1 = n if err != nil && deps.log().DebugEnabled() { - log.Printf("pipe: c2<-c1 copy error: %v", err) + deps.log().Errorf("pipe: c2<-c1 copy error: %v", err) } }) wg.Wait() if err := c1.SetDeadline(time.Time{}); err != nil && deps.log().DebugEnabled() { - log.Printf("pipe: failed to reset deadline c1: %v", err) + deps.log().Errorf("pipe: failed to reset deadline c1: %v", err) } if err := c2.SetDeadline(time.Time{}); err != nil && deps.log().DebugEnabled() { - log.Printf("pipe: failed to reset deadline c2: %v", err) + deps.log().Errorf("pipe: failed to reset deadline c2: %v", err) } return c1FromC2, c2FromC1 } diff --git a/internal/proxy/tcpfwdserver/tcpfwdserver.go b/internal/proxy/tcpfwdserver/tcpfwdserver.go index c347a32..6e3591e 100644 --- a/internal/proxy/tcpfwdserver/tcpfwdserver.go +++ b/internal/proxy/tcpfwdserver/tcpfwdserver.go @@ -7,7 +7,6 @@ package tcpfwdserver import ( "context" "io" - "log" "net" "sync" "time" @@ -37,24 +36,24 @@ func Handle(ctx context.Context, logger logx.Logger, registry *bondserver.Regist kcpSess, err := kcptun.NewKCPOverDTLS(&stats.CountingConn{Conn: dtlsConn, Stats: st}, true) if err != nil { - log.Printf("KCP session error: %s", err) + logger.Errorf("tcpfwdserver: KCP session: %s", err) return } defer func() { if closeErr := kcpSess.Close(); closeErr != nil { - log.Printf("failed to close KCP session: %v", closeErr) + logger.Errorf("tcpfwdserver: close KCP session: %v", closeErr) } }() logger.Debugf("KCP session established (server)") smuxSess, err := smux.Server(kcpSess, kcptun.DefaultSmuxConfig()) if err != nil { - log.Printf("smux server error: %s", err) + logger.Errorf("tcpfwdserver: smux server: %s", err) return } defer func() { if err := smuxSess.Close(); err != nil { - log.Printf("failed to close smux session: %v", err) + logger.Errorf("tcpfwdserver: close smux session: %v", err) } }() logger.Debugf("smux session established (server)") @@ -66,7 +65,7 @@ func Handle(ctx context.Context, logger logx.Logger, registry *bondserver.Regist select { case <-ctx.Done(): default: - log.Printf("smux accept error: %s", err) + logger.Errorf("tcpfwdserver: smux accept: %s", err) } break } @@ -83,7 +82,7 @@ func handleStream(ctx context.Context, logger logx.Logger, registry *bondserver. var prefix [4]byte if _, err := io.ReadFull(s, prefix[:]); err != nil { if err != io.EOF && err != io.ErrUnexpectedEOF { - log.Printf("smux stream prefix read error: %v", err) + logger.Errorf("tcpfwdserver: smux stream prefix read: %v", err) } _ = s.Close() return @@ -96,18 +95,18 @@ func handleStream(ctx context.Context, logger logx.Logger, registry *bondserver. defer func() { if err := s.Close(); err != nil && err != smux.ErrGoAway { - log.Printf("failed to close smux stream: %v", err) + logger.Errorf("tcpfwdserver: close smux stream: %v", err) } }() backendConn, err := net.DialTimeout("tcp", connectAddr, 10*time.Second) if err != nil { - log.Printf("backend dial error: %s", err) + logger.Errorf("tcpfwdserver: backend dial: %s", err) return } defer func() { if err := backendConn.Close(); err != nil { - log.Printf("failed to close backend connection: %v", err) + logger.Errorf("tcpfwdserver: close backend connection: %v", err) } }() diff --git a/internal/proxy/udprelay/udprelay.go b/internal/proxy/udprelay/udprelay.go index b4a0e95..d25788b 100644 --- a/internal/proxy/udprelay/udprelay.go +++ b/internal/proxy/udprelay/udprelay.go @@ -91,7 +91,7 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log } context.AfterFunc(ctx, func() { if closeErr := listenConn.Close(); closeErr != nil { - log.Printf("Failed to close local connection: %s", closeErr) + logger.Errorf("udprelay: close local connection: %s", closeErr) } }) @@ -240,7 +240,7 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr if err := <-c; err != nil { if errors.Is(err, vkauth.ErrFatalCaptchaNoStreams) { - log.Printf("[STREAM %d] Fatal manual captcha error. Shutting down application.", streamID) + deps.log().Errorf("[STREAM %d] Fatal manual captcha error. Shutting down application.", streamID) if deps.AppCancel != nil { deps.AppCancel() } @@ -248,7 +248,7 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr } if errors.Is(err, vkauth.ErrCaptchaWaitRequired) { if !errors.Is(err, vkauth.ErrLockoutActive) { - log.Printf("[STREAM %d] Backing off for 60 seconds to avoid IP ban...", streamID) + deps.log().Errorf("[STREAM %d] Backing off for 60 seconds to avoid IP ban...", streamID) select { case <-ctx.Done(): return @@ -267,7 +267,7 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr } } } else { - log.Printf("[STREAM %d] %s", streamID, err) + deps.log().Errorf("[STREAM %d] %s", streamID, err) time.Sleep(2 * time.Second) } } @@ -301,11 +301,11 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net. var dtlsConn net.Conn = dtlsRaw defer func() { if closeErr := dtlsConn.Close(); closeErr != nil { - log.Printf("[STREAM %d] failed to close DTLS connection: %s", streamID, closeErr) + deps.log().Errorf("[STREAM %d] failed to close DTLS connection: %s", streamID, closeErr) } - log.Printf("[STREAM %d] Closed DTLS connection\n", streamID) + deps.log().Errorf("[STREAM %d] Closed DTLS connection\n", streamID) }() - log.Printf("[STREAM %d] Established DTLS connection!\n", streamID) + deps.log().Errorf("[STREAM %d] Established DTLS connection!\n", streamID) if okchan != nil { go func() { @@ -319,7 +319,7 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net. wg := sync.WaitGroup{} context.AfterFunc(dtlsctx, func() { if err := dtlsConn.SetDeadline(time.Now()); err != nil { - log.Printf("[STREAM %d] Warning: SetDeadline failed: %v", streamID, err) + deps.log().Errorf("[STREAM %d] Warning: SetDeadline failed: %v", streamID, err) } }) @@ -348,7 +348,7 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net. if peerAddr := deps.ActiveLocalPeer.Load(); peerAddr != nil { if addr, ok := peerAddr.(net.Addr); ok { if _, err := listenConn.WriteTo(buf[:n], addr); err != nil { - log.Printf("[STREAM %d] failed to forward packet to local peer: %v", streamID, err) + deps.log().Errorf("[STREAM %d] failed to forward packet to local peer: %v", streamID, err) } } } @@ -357,7 +357,7 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net. wg.Wait() if err := dtlsConn.SetDeadline(time.Time{}); err != nil { - log.Printf("[STREAM %d] Failed to clear DTLS deadline: %s", streamID, err) + deps.log().Errorf("[STREAM %d] Failed to clear DTLS deadline: %s", streamID, err) } return nil } @@ -388,7 +388,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, }() if deps.log().DebugEnabled() { - log.Printf("[STREAM %d] relayed-address=%s", streamID, relayConn.LocalAddr().String()) + deps.log().Errorf("[STREAM %d] relayed-address=%s", streamID, relayConn.LocalAddr().String()) } wg := sync.WaitGroup{} @@ -398,13 +398,13 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, context.AfterFunc(turnctx, func() { if err := relayConn.SetDeadline(time.Now()); err != nil { - log.Printf("Failed to set relay deadline: %s", err) + deps.log().Errorf("Failed to set relay deadline: %s", err) } }) var internalPipeAddr atomic.Value wc, wcErr := common.NewClientWrap(params.WrapKey) if wcErr != nil { - log.Printf("[STREAM %d] WRAP init failed: %v", streamID, wcErr) + deps.log().Errorf("[STREAM %d] WRAP init failed: %v", streamID, wcErr) turncancel() return } @@ -434,7 +434,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, if wc != nil { written, wrapErr := wc.WrapInto(wireBuf, out) if wrapErr != nil { - log.Printf("[STREAM %d] WRAP failed: %v", streamID, wrapErr) + deps.log().Errorf("[STREAM %d] WRAP failed: %v", streamID, wrapErr) return } out = wireBuf[:written] @@ -471,7 +471,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, if wc != nil { m, wrapErr := wc.Unwrap(payload, plain) if wrapErr != nil { - log.Printf("[STREAM %d] UNWRAP failed: %v (n=%d)", streamID, wrapErr, n) + deps.log().Errorf("[STREAM %d] UNWRAP failed: %v (n=%d)", streamID, wrapErr, n) continue } payload = plain[:m] @@ -486,6 +486,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, wg.Wait() if err := relayConn.SetDeadline(time.Time{}); err != nil { - log.Printf("Failed to clear relay deadline: %s", err) + deps.log().Errorf("Failed to clear relay deadline: %s", err) } } diff --git a/internal/proxy/udpserver/udpserver.go b/internal/proxy/udpserver/udpserver.go index e458a7e..c0ab32a 100644 --- a/internal/proxy/udpserver/udpserver.go +++ b/internal/proxy/udpserver/udpserver.go @@ -5,7 +5,6 @@ package udpserver import ( "context" "fmt" - "log" "net" "sync" "time" @@ -24,12 +23,12 @@ const ( func Handle(ctx context.Context, logger logx.Logger, conn net.Conn, connectAddr string) { serverConn, err := net.Dial("udp", connectAddr) if err != nil { - log.Println(err) + logger.Errorf("udpserver: dial backend: %v", err) return } defer func() { if err = serverConn.Close(); err != nil { - log.Printf("failed to close outgoing connection: %s", err) + logger.Errorf("udpserver: close outgoing connection: %s", err) } }() @@ -46,28 +45,28 @@ func Handle(ctx context.Context, logger logx.Logger, conn net.Conn, connectAddr context.AfterFunc(ctx2, func() { if err := conn.SetDeadline(time.Now()); err != nil { - log.Printf("failed to set incoming deadline: %s", err) + logger.Errorf("udpserver: set incoming deadline: %s", err) } if err := serverConn.SetDeadline(time.Now()); err != nil { - log.Printf("failed to set outgoing deadline: %s", err) + logger.Errorf("udpserver: set outgoing deadline: %s", err) } }) var wg sync.WaitGroup wg.Go(func() { defer cancel() - copyOne(ctx2, conn, serverConn, st.AddTx) + copyOne(ctx2, logger, conn, serverConn, st.AddTx) }) wg.Go(func() { defer cancel() - copyOne(ctx2, serverConn, conn, st.AddRx) + copyOne(ctx2, logger, serverConn, conn, st.AddRx) }) wg.Wait() } // copyOne reads from src and writes to dst until ctx fires or either side errors. // Each read/write resets an idle timeout so a stuck side closes instead of hanging. -func copyOne(ctx context.Context, src, dst net.Conn, count func(int)) { +func copyOne(ctx context.Context, logger logx.Logger, src, dst net.Conn, count func(int)) { buf := make([]byte, udpRelayBufSize) for { select { @@ -76,22 +75,22 @@ func copyOne(ctx context.Context, src, dst net.Conn, count func(int)) { default: } if err := src.SetReadDeadline(time.Now().Add(udpIdleTimeout)); err != nil { - log.Printf("Failed: %s", err) + logger.Errorf("udpserver: set read deadline: %s", err) return } n, err := src.Read(buf) if err != nil { - log.Printf("Failed: %s", err) + logger.Debugf("udpserver: read: %s", err) return } if werr := dst.SetWriteDeadline(time.Now().Add(udpIdleTimeout)); werr != nil { - log.Printf("Failed: %s", werr) + logger.Errorf("udpserver: set write deadline: %s", werr) return } written, werr := dst.Write(buf[:n]) count(written) if werr != nil { - log.Printf("Failed: %s", werr) + logger.Debugf("udpserver: write: %s", werr) return } }