Ignore conntracked connections on which we never saw an update; don't…

… nat map conntracked connections. (#1466)
weaveworks · May 11, 2016 · 23c5e9f · 23c5e9f
1 parent 45c6c00
commit 23c5e9f
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 14 deletions.
diff --git a/probe/endpoint/conntrack.go b/probe/endpoint/conntrack.go
@@ -314,9 +314,13 @@ func (c *conntrackWalker) handleFlow(f flow, forceAdd bool) {
 			c.bufferedFlows = append(c.bufferedFlows, f)
 		}
 	case f.Type == destroyType:
-		if _, ok := c.activeFlows[f.Independent.ID]; ok {
+		if active, ok := c.activeFlows[f.Independent.ID]; ok {
 			delete(c.activeFlows, f.Independent.ID)
-			c.bufferedFlows = append(c.bufferedFlows, f)
+			// Ignore flows for which we never saw an update; they are likely
+			// incomplete or wrong.  See #1462.
+			if active.Type == updateType {
+				c.bufferedFlows = append(c.bufferedFlows, active)
+			}
 		}
 	}
 }

diff --git a/probe/endpoint/conntrack_internal_test.go b/probe/endpoint/conntrack_internal_test.go
@@ -150,20 +150,22 @@ func TestConntracker(t *testing.T) {
 		}
 	}
 
-	flow1 := makeFlow(newType)
+	flow1 := makeFlow(updateType)
 	addMeta(&flow1, "original", "1.2.3.4", "2.3.4.5", 2, 3)
 	addIndependant(&flow1, 1, "")
 	writeFlow(flow1)
 	test.Poll(t, ts, []flow{flow1}, have)
 
 	// Now check when we remove the flow, we still get it in the next Walk
-	flow1.Type = destroyType
-	writeFlow(flow1)
+	flow2 := makeFlow(destroyType)
+	addMeta(&flow2, "original", "1.2.3.4", "2.3.4.5", 2, 3)
+	addIndependant(&flow2, 1, "")
+	writeFlow(flow2)
 	test.Poll(t, ts, []flow{flow1}, have)
 	test.Poll(t, ts, []flow{}, have)
 
 	// This time we're not going to remove it, but put it in state TIME_WAIT
-	flow1.Type = newType
+	flow1.Type = updateType
 	writeFlow(flow1)
 	test.Poll(t, ts, []flow{flow1}, have)
 

diff --git a/probe/endpoint/nat_internal_test.go b/probe/endpoint/nat_internal_test.go
@@ -34,7 +34,7 @@ func TestNat(t *testing.T) {
 
 	// from the PoV of host1
 	{
-		f := makeFlow("")
+		f := makeFlow(updateType)
 		addIndependant(&f, 1, "")
 		f.Original = addMeta(&f, "original", "2.3.4.5", "1.2.3.4", 222222, 80)
 		f.Reply = addMeta(&f, "reply", "10.0.47.1", "2.3.4.5", 80, 222222)
@@ -45,9 +45,10 @@ func TestNat(t *testing.T) {
 		have := report.MakeReport()
 		originalID := report.MakeEndpointNodeID("host1", "10.0.47.1", "80")
 		have.Endpoint.AddNode(report.MakeNodeWith(originalID, map[string]string{
-			Addr:  "10.0.47.1",
-			Port:  "80",
-			"foo": "bar",
+			Addr:      "10.0.47.1",
+			Port:      "80",
+			"foo":     "bar",
+			Procspied: "true",
 		}))
 
 		want := have.Copy()
@@ -57,6 +58,7 @@ func TestNat(t *testing.T) {
 			Port:      "80",
 			"copy_of": originalID,
 			"foo":     "bar",
+			Procspied: "true",
 		}))
 
 		makeNATMapper(ct).applyNAT(have, "host1")
@@ -67,7 +69,7 @@ func TestNat(t *testing.T) {
 
 	// form the PoV of host2
 	{
-		f := makeFlow("")
+		f := makeFlow(updateType)
 		addIndependant(&f, 2, "")
 		f.Original = addMeta(&f, "original", "10.0.47.2", "1.2.3.4", 22222, 80)
 		f.Reply = addMeta(&f, "reply", "1.2.3.4", "2.3.4.5", 80, 22223)
@@ -78,9 +80,10 @@ func TestNat(t *testing.T) {
 		have := report.MakeReport()
 		originalID := report.MakeEndpointNodeID("host2", "10.0.47.2", "22222")
 		have.Endpoint.AddNode(report.MakeNodeWith(originalID, map[string]string{
-			Addr:  "10.0.47.2",
-			Port:  "22222",
-			"foo": "baz",
+			Addr:      "10.0.47.2",
+			Port:      "22222",
+			"foo":     "baz",
+			Procspied: "true",
 		}))
 
 		want := have.Copy()
@@ -89,6 +92,7 @@ func TestNat(t *testing.T) {
 			Port:      "22223",
 			"copy_of": originalID,
 			"foo":     "baz",
+			Procspied: "true",
 		}))
 
 		makeNATMapper(ct).applyNAT(have, "host1")