netstack: add counters for tcp CurrEstab and EstabResets

Signed-off-by: Jianfeng Tan <henry.tjf@antfin.com>

pkg/sentry/socket/netstack/netstack.go

@@ -148,6 +148,8 @@ var Metrics = tcpip.Stats{
 	TCP: tcpip.TCPStats{
 		ActiveConnectionOpenings:           mustCreateMetric("/netstack/tcp/active_connection_openings", "Number of connections opened successfully via Connect."),
 		PassiveConnectionOpenings:          mustCreateMetric("/netstack/tcp/passive_connection_openings", "Number of connections opened successfully via Listen."),
+		CurrentEstablished:                 mustCreateMetric("/netstack/tcp/current_established", "Number of connections in either ESTABLISHED or CLOSE-WAIT state now."),
+		EstablishedResets:                  mustCreateMetric("/netstack/tcp/established_resets", "Number of times TCP connections have made a direct transition to the CLOSED state from either the ESTABLISHED state or the CLOSE-WAIT state"),
 		ListenOverflowSynDrop:              mustCreateMetric("/netstack/tcp/listen_overflow_syn_drop", "Number of times the listen queue overflowed and a SYN was dropped."),
 		ListenOverflowAckDrop:              mustCreateMetric("/netstack/tcp/listen_overflow_ack_drop", "Number of times the listen queue overflowed and the final ACK in the handshake was dropped."),
 		ListenOverflowSynCookieSent:        mustCreateMetric("/netstack/tcp/listen_overflow_syn_cookie_sent", "Number of times a SYN cookie was sent."),

pkg/tcpip/tcpip.go

@@ -673,6 +673,11 @@ func (s *StatCounter) Increment() {
 	s.IncrementBy(1)
 }
 
+// Decrement minuses one to the counter.
+func (s *StatCounter) Decrement() {
+	s.IncrementBy(^uint64(0))
+}
+
 // Value returns the current value of the counter.
 func (s *StatCounter) Value() uint64 {
 	return atomic.LoadUint64(&s.count)
@@ -881,6 +886,15 @@ type TCPStats struct {
 	// successfully via Listen.
 	PassiveConnectionOpenings *StatCounter
 
+	// CurrentEstablished is the number of TCP connections for which the
+	// current state is either ESTABLISHED or CLOSE-WAIT.
+	CurrentEstablished *StatCounter
+
+	// EstablishedResets is the number of times TCP connections have made
+	// a direct transition to the CLOSED state from either the
+	// ESTABLISHED state or the CLOSE-WAIT state.
+	EstablishedResets *StatCounter
+
 	// ListenOverflowSynDrop is the number of times the listen queue overflowed
 	// and a SYN was dropped.
 	ListenOverflowSynDrop *StatCounter

pkg/tcpip/transport/tcp/accept.go

@@ -297,7 +297,10 @@ func (l *listenContext) createEndpointAndPerformHandshake(s *segment, opts *head
 		return nil, err
 	}
 	ep.mu.Lock()
-	ep.state = StateEstablished
+	if ep.state != StateEstablished {
+		ep.stack.Stats().TCP.CurrentEstablished.Increment()
+		ep.state = StateEstablished
+	}
 	ep.mu.Unlock()
 
 	// Update the receive window scaling. We can't do it before the
@@ -519,6 +522,7 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) {
 		n.tsOffset = 0
 
 		// Switch state to connected.
+		n.stack.Stats().TCP.CurrentEstablished.Increment()
 		n.state = StateEstablished
 
 		// Do the delivery in a separate goroutine so

pkg/tcpip/transport/tcp/connect.go

@@ -754,6 +754,10 @@ func (e *endpoint) handleClose() *tcpip.Error {
 func (e *endpoint) resetConnectionLocked(err *tcpip.Error) {
 	// Only send a reset if the connection is being aborted for a reason
 	// other than receiving a reset.
+	if e.state == StateEstablished || e.state == StateCloseWait {
+		e.stack.Stats().TCP.EstablishedResets.Increment()
+		e.stack.Stats().TCP.CurrentEstablished.Decrement()
+	}
 	e.state = StateError
 	e.HardError = err
 	if err != tcpip.ErrConnectionReset {
@@ -924,6 +928,10 @@ func (e *endpoint) protocolMainLoop(handshake bool) *tcpip.Error {
 			e.lastErrorMu.Unlock()
 
 			e.mu.Lock()
+			if e.state == StateEstablished || e.state == StateCloseWait {
+				e.stack.Stats().TCP.EstablishedResets.Increment()
+				e.stack.Stats().TCP.CurrentEstablished.Decrement()
+			}
 			e.state = StateError
 			e.HardError = err
 
@@ -954,7 +962,10 @@ func (e *endpoint) protocolMainLoop(handshake bool) *tcpip.Error {
 
 	// Tell waiters that the endpoint is connected and writable.
 	e.mu.Lock()
-	e.state = StateEstablished
+	if e.state != StateEstablished {
+		e.stack.Stats().TCP.CurrentEstablished.Increment()
+		e.state = StateEstablished
+	}
 	drained := e.drainDone != nil
 	e.mu.Unlock()
 	if drained {
@@ -1115,6 +1126,10 @@ func (e *endpoint) protocolMainLoop(handshake bool) *tcpip.Error {
 	// Mark endpoint as closed.
 	e.mu.Lock()
 	if e.state != StateError {
+		if e.state == StateEstablished || e.state == StateCloseWait {
+			e.stack.Stats().TCP.EstablishedResets.Increment()
+			e.stack.Stats().TCP.CurrentEstablished.Decrement()
+		}
 		e.state = StateClose
 	}
 	// Lock released below.

pkg/tcpip/transport/tcp/endpoint.go

@@ -1729,6 +1729,7 @@ func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool, run bool) *tc
 		e.segmentQueue.mu.Unlock()
 		e.snd.updateMaxPayloadSize(int(e.route.MTU()), 0)
 		e.state = StateEstablished
+		e.stack.Stats().TCP.CurrentEstablished.Increment()
 	}
 
 	if run {

pkg/tcpip/transport/tcp/snd.go

@@ -674,6 +674,7 @@ func (s *sender) maybeSendSegment(seg *segment, limit int, end seqnum.Value) (se
 		default:
 			s.ep.state = StateFinWait1
 		}
+		s.ep.stack.Stats().TCP.CurrentEstablished.Decrement()
 		s.ep.mu.Unlock()
 	} else {
 		// We're sending a non-FIN segment.