Merge pull request openziti/fabric#182 from openziti/pull-fixes

Pull fixes from 0.15 release branch

router/forwarder/forwarder.go

@@ -83,12 +83,12 @@ func (forwarder *Forwarder) UnregisterDestinations(sessionId string) {
 				forwarder.destinations.removeDestination(address)
 				go destination.(XgressDestination).Close() // create close queue?
 			} else {
-				pfxlog.Logger().Debugf("no destinations found for [@/%v] for [s/%v]", address, sessionId)
+				pfxlog.Logger().Warnf("no destinations found for [@/%v] for [s/%v]", address, sessionId)
 			}
 		}
 		forwarder.destinations.unlinkSession(sessionId)
 	} else {
-		pfxlog.Logger().Debugf("found no addresses to unregister for [s/%v]", sessionId)
+		pfxlog.Logger().Warnf("found no addresses to unregister for [s/%v]", sessionId)
 	}
 }
 

router/xgress/link_receive_buffer.go

@@ -58,12 +58,10 @@ func (buffer *LinkReceiveBuffer) ReceiveUnordered(payload *Payload, maxSize uint
 		payloadSize := len(payload.Data)
 		size := atomic.AddUint32(&buffer.size, uint32(payloadSize))
 		pfxlog.Logger().Tracef("Payload %v of size %v added to transmit buffer. New size: %v", payload.Sequence, payloadSize, size)
-		localRecvBufferSizeBytesHistogram.Update(int64(size))
 		if payload.Sequence > buffer.maxSequence {
 			buffer.maxSequence = payload.Sequence
 		}
 	}
-	localRecvBufferSizeMsgsHistogram.Update(int64(buffer.tree.Size()))
 	return true
 }
 
@@ -80,7 +78,6 @@ func (buffer *LinkReceiveBuffer) PeekHead() *Payload {
 func (buffer *LinkReceiveBuffer) Remove(payload *Payload) {
 	buffer.tree.Remove(payload.Sequence)
 	buffer.sequence = payload.Sequence
-	localRecvBufferSizeMsgsHistogram.Update(int64(buffer.tree.Size()))
 }
 
 func (buffer *LinkReceiveBuffer) getLastBufferSizeSent() uint32 {

router/xgress/link_send_buffer.go

@@ -183,9 +183,6 @@ func (buffer *LinkSendBuffer) run() {
 	defer retransmitTicker.Stop()
 
 	for {
-		txBufferSizeHistogram.Update(int64(buffer.linkSendBufferSize))
-		txWindowSize.Update(int64(buffer.windowsSize))
-
 		// don't block when we're closing, since the only thing that should still be coming in is end-of-session
 		if buffer.isBlocked() && !buffer.closeWhenEmpty.Get() {
 			buffered = nil
@@ -286,11 +283,9 @@ func (buffer *LinkSendBuffer) receiveAcknowledgement(ack *Acknowledgement) {
 	}
 
 	buffer.linkRecvBufferSize = ack.RecvBufferSize
-	remoteRecvBufferSizeHistogram.Update(int64(buffer.linkRecvBufferSize))
 	if ack.RTT > 0 {
 		rtt := uint16(info.NowInMilliseconds()) - ack.RTT
 		buffer.retxThreshold = uint32(float64(rtt)*buffer.retxScale) + buffer.x.Options.RetxAddMs
-		rttHistogram.Update(int64(rtt))
 	}
 }
 

router/xgress/metrics.go

@@ -12,14 +12,8 @@ var retransmissions metrics.Meter
 var retransmissionFailures metrics.Meter
 
 var ackFailures metrics.Meter
-var rttHistogram metrics.Histogram
-var txBufferSizeHistogram metrics.Histogram
-var localRecvBufferSizeBytesHistogram metrics.Histogram
-var localRecvBufferSizeMsgsHistogram metrics.Histogram
-var remoteRecvBufferSizeHistogram metrics.Histogram
 var payloadWriteTimer metrics.Timer
 var duplicateAcksMeter metrics.Meter
-var txWindowSize metrics.Histogram
 
 var buffersBlockedByLocalWindow int64
 var buffersBlockedByRemoteWindow int64
@@ -33,14 +27,8 @@ func InitMetrics(registry metrics.UsageRegistry) {
 	ackRxMeter = registry.Meter("xgress.rx.acks")
 	ackTxMeter = registry.Meter("xgress.tx.acks")
 	ackFailures = registry.Meter("xgress.ack_failures")
-	rttHistogram = registry.Histogram("xgress.rtt")
-	txBufferSizeHistogram = registry.Histogram("xgress.tx_buffer_size")
-	localRecvBufferSizeBytesHistogram = registry.Histogram("xgress.local.rx_buffer_bytes_size")
-	localRecvBufferSizeMsgsHistogram = registry.Histogram("xgress.local.rx_buffer_msgs_size")
-	remoteRecvBufferSizeHistogram = registry.Histogram("xgress.remote.rx_buffer_size")
 	payloadWriteTimer = registry.Timer("xgress.tx_write_time")
 	duplicateAcksMeter = registry.Meter("xgress.ack_duplicates")
-	txWindowSize = registry.Histogram("xgress.tx_window_size")
 
 	registry.FuncGauge("xgress.blocked_by_local_window", func() int64 {
 		return atomic.LoadInt64(&buffersBlockedByLocalWindow)

router/xgress/xgress.go

@@ -409,7 +409,6 @@ func (self *Xgress) tx() {
 		payloadSize := len(payload.Data)
 		size := atomic.AddUint32(&self.linkRxBuffer.size, ^uint32(payloadSize-1)) // subtraction for uint32
 		payloadLogger.Debugf("Payload %v of size %v removed from rx buffer. New size: %v", payload.Sequence, payloadSize, size)
-		localRecvBufferSizeBytesHistogram.Update(int64(size))
 
 		lastBufferSizeSent := self.linkRxBuffer.getLastBufferSizeSent()
 		if lastBufferSizeSent > 10000 && (lastBufferSizeSent>>1) > size {
@@ -526,9 +525,7 @@ func (self *Xgress) nextReceiveSequence() int32 {
 func (self *Xgress) closeTimeoutHandler(duration time.Duration) {
 	self.payloadBuffer.CloseWhenEmpty() // If we clear the send buffer, close sooner
 	time.Sleep(duration)
-	if !self.closed.Get() {
-		self.Close()
-	}
+	self.Close()
 }
 
 func (self *Xgress) PayloadReceived(payload *Payload) {