diff --git a/tunnel/udp_vconn/conn.go b/tunnel/udp_vconn/conn.go index 57de3b5a9..c28385051 100644 --- a/tunnel/udp_vconn/conn.go +++ b/tunnel/udp_vconn/conn.go @@ -22,6 +22,7 @@ import ( "github.com/netfoundry/ziti-foundation/util/mempool" "io" "net" + "sync/atomic" "time" ) @@ -31,7 +32,7 @@ type udpConn struct { srcAddr net.Addr manager *manager writeConn UDPWriterTo - lastUse time.Time + lastUse atomic.Value closed bool } @@ -40,6 +41,7 @@ func (conn *udpConn) Service() string { } func (conn *udpConn) Accept(buffer mempool.PooledBuffer) { + pfxlog.Logger().WithField("udpConnId", conn.srcAddr.String()).Debugf("udp->ziti: queuing") conn.readC <- buffer } @@ -51,6 +53,15 @@ func (conn *udpConn) String() string { return conn.service } +func (conn *udpConn) markUsed() { + conn.lastUse.Store(time.Now()) +} + +func (conn *udpConn) GetLastUsed() time.Time { + val := conn.lastUse.Load() + return val.(time.Time) +} + func (conn *udpConn) WriteTo(w io.Writer) (n int64, err error) { var bytesWritten int64 for { @@ -60,9 +71,11 @@ func (conn *udpConn) WriteTo(w io.Writer) (n int64, err error) { return bytesWritten, io.EOF } - n, err := w.Write(buf.GetPayload()) + payload := buf.GetPayload() + pfxlog.Logger().WithField("udpConnId", conn.srcAddr.String()).Debugf("udp->ziti: %v bytes", len(payload)) + n, err := w.Write(payload) buf.Release() - conn.lastUse = time.Now() + conn.markUsed() bytesWritten += int64(n) if err != nil { return bytesWritten, err @@ -75,10 +88,10 @@ func (conn *udpConn) Read(b []byte) (n int, err error) { } func (conn *udpConn) Write(b []byte) (int, error) { - pfxlog.Logger().Infof("Received %v bytes from ziti to return to udp", len(b)) + pfxlog.Logger().WithField("udpConnId", conn.srcAddr.String()).Debugf("ziti->udp: %v bytes", len(b)) // TODO: UDP chunking, MTU chunking? n, err := conn.writeConn.WriteTo(b, conn.srcAddr) - conn.lastUse = time.Now() + conn.markUsed() return n, err } diff --git a/tunnel/udp_vconn/manager.go b/tunnel/udp_vconn/manager.go index 2da49ef75..15c401504 100644 --- a/tunnel/udp_vconn/manager.go +++ b/tunnel/udp_vconn/manager.go @@ -71,7 +71,7 @@ func (manager *manager) run() { } func (manager *manager) GetWriteQueue(srcAddr net.Addr) WriteQueue { - pfxlog.Logger().Infof("Looking up address %v", srcAddr.String()) + pfxlog.Logger().Debugf("Looking up address %v", srcAddr.String()) result := manager.connMap[srcAddr.String()] if result == nil { return nil @@ -92,10 +92,10 @@ func (manager *manager) CreateWriteQueue(srcAddr net.Addr, service string, write srcAddr: srcAddr, manager: manager, writeConn: writeConn, - lastUse: time.Now(), } + conn.markUsed() manager.connMap[srcAddr.String()] = conn - pfxlog.Logger().Infof("created new virtual UDP connection for %v", srcAddr.String()) + pfxlog.Logger().WithField("udpConnId", srcAddr.String()).Debug("created new virtual UDP connection") go tunnel.Run(manager.context, service, conn) return conn, nil } @@ -108,10 +108,8 @@ func (manager *manager) dropLRU() { for _, value := range manager.connMap { if oldest == nil { oldest = value - } else { - if oldest.lastUse.After(value.lastUse) { - oldest = value - } + } else if oldest.GetLastUsed().After(value.GetLastUsed()) { + oldest = value } } manager.close(oldest) @@ -121,8 +119,8 @@ func (manager *manager) dropExpired() { log := pfxlog.Logger() now := time.Now() for key, value := range manager.connMap { - if manager.expirationPolicy.IsExpired(now, value.lastUse) { - log.Infof("connection %v expired. removing from UDP vconn manager", key) + if manager.expirationPolicy.IsExpired(now, value.GetLastUsed()) { + log.WithField("udpConnId", key).Debug("connection expired. removing from UDP vconn manager") manager.close(value) } }