From f4dc3689551e0ae6800f8ba1b6a599a29cf4247e Mon Sep 17 00:00:00 2001
From: Bryan Boreham <bjboreham@gmail.com>
Date: Sun, 5 Aug 2018 13:20:00 +0000
Subject: [PATCH] Don't buffer TIME_WAIT flows on conntrack start-up

When the probe first starts we should only be interested in active
connections, and if the loop re-starts it's probably because too many
connections are opening and closing to keep up with, so it's good to
drop any that are already closed then too.

Refactor the code so `handleFlow` is only called on events, and handle
the initial list of connections directly.
---
 probe/endpoint/conntrack.go | 22 ++++++++++++++--------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/probe/endpoint/conntrack.go b/probe/endpoint/conntrack.go
index c49e409da..6bb480e30 100644
--- a/probe/endpoint/conntrack.go
+++ b/probe/endpoint/conntrack.go
@@ -100,15 +100,23 @@ func (c *conntrackWalker) clearFlows() {
 	c.activeFlows = map[uint32]conntrack.Conn{}
 }
 
+func (c *conntrackWalker) relevant(f conntrack.Conn) bool {
+	return !(c.natOnly && (f.Status&conntrack.IPS_NAT_MASK) == 0)
+}
+
 func (c *conntrackWalker) run() {
 	existingFlows, err := conntrack.ConnectionsSize(c.bufferSize)
 	if err != nil {
 		log.Errorf("conntrack Connections error: %v", err)
 		return
 	}
+	c.Lock()
 	for _, flow := range existingFlows {
-		c.handleFlow(flow, true)
+		if c.relevant(flow) && flow.TCPState != "TIME_WAIT" {
+			c.activeFlows[flow.CtId] = flow
+		}
 	}
+	c.Unlock()
 
 	events, stop, err := conntrack.FollowSize(c.bufferSize, conntrack.NF_NETLINK_CONNTRACK_UPDATE|conntrack.NF_NETLINK_CONNTRACK_DESTROY)
 	if err != nil {
@@ -128,7 +136,9 @@ func (c *conntrackWalker) run() {
 			if !ok {
 				return
 			}
-			c.handleFlow(f, false)
+			if c.relevant(f) {
+				c.handleFlow(f)
+			}
 		}
 	}
 }
@@ -139,18 +149,14 @@ func (c *conntrackWalker) stop() {
 	close(c.quit)
 }
 
-func (c *conntrackWalker) handleFlow(f conntrack.Conn, forceAdd bool) {
+func (c *conntrackWalker) handleFlow(f conntrack.Conn) {
 	c.Lock()
 	defer c.Unlock()
 
-	if c.natOnly && (f.Status&conntrack.IPS_NAT_MASK) == 0 {
-		return
-	}
-
 	// Ignore flows for which we never saw an update; they are likely
 	// incomplete or wrong.  See #1462.
 	switch {
-	case forceAdd || f.MsgType == conntrack.NfctMsgUpdate:
+	case f.MsgType == conntrack.NfctMsgUpdate:
 		if f.TCPState != "TIME_WAIT" {
 			c.activeFlows[f.CtId] = f
 		} else if _, ok := c.activeFlows[f.CtId]; ok {