Merge pull request #143 from netfoundry/udp-log-cleanup-race-condition-fix

Make udp logging use debug and make format consistent. Ensure last us…

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
 }
 

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)
 		}
 	}