Merge pull request #1410 from weaveworks/1408-dnat-shortlived-connections

Correctly attribute DNAT-ed short-lived connections

probe/docker/container.go

@@ -298,7 +298,7 @@ func (c *container) NetworkMode() (string, bool) {
 func addScopeToIPs(hostID string, ips []string) []string {
 	ipsWithScopes := []string{}
 	for _, ip := range ips {
-		ipsWithScopes = append(ipsWithScopes, report.MakeScopedAddressNodeID(hostID, ip))
+		ipsWithScopes = append(ipsWithScopes, report.MakeAddressNodeID(hostID, ip))
 	}
 	return ipsWithScopes
 }

probe/docker/container_test.go

@@ -113,7 +113,7 @@ func TestContainer(t *testing.T) {
 		want := report.EmptySets.
 			Add("docker_container_ports", report.MakeStringSet("1.2.3.4:80->80/tcp", "81/tcp")).
 			Add("docker_container_ips", report.MakeStringSet("1.2.3.4")).
-			Add("docker_container_ips_with_scopes", report.MakeStringSet("scope;1.2.3.4"))
+			Add("docker_container_ips_with_scopes", report.MakeStringSet(";1.2.3.4"))
 
 		test.Poll(t, 100*time.Millisecond, want, func() interface{} {
 			return c.NetworkInfo([]net.IP{})

probe/endpoint/nat_internal_test.go

@@ -30,7 +30,7 @@ func TestNat(t *testing.T) {
 	// the setup is this:
 	//
 	// container2 (10.0.47.2:222222), host2 (2.3.4.5:22223) ->
-	//     host1 (1.2.3.4:80), container1 (10.0.47.2:80)
+	//     host1 (1.2.3.4:80), container1 (10.0.47.1:80)
 
 	// from the PoV of host1
 	{

probe/endpoint/reporter.go

@@ -106,7 +106,7 @@ func (r *Reporter) Report() (report.Report, error) {
 	rpt := report.MakeReport()
 	seenTuples := map[string]fourTuple{}
 
-	// Consult the flowWalker for short-live connections
+	// Consult the flowWalker for short-lived connections
 	{
 		extraNodeInfo := map[string]string{
 			Conntracked: "true",
@@ -118,6 +118,18 @@ func (r *Reporter) Report() (report.Report, error) {
 				uint16(f.Original.Layer4.SrcPort),
 				uint16(f.Original.Layer4.DstPort),
 			}
+			// Handle DNAT-ed short-lived connections.
+			// The NAT mapper won't help since it only runs periodically,
+			// missing the short-lived connections.
+			if f.Original.Layer3.DstIP != f.Reply.Layer3.SrcIP {
+				tuple = fourTuple{
+					f.Reply.Layer3.DstIP,
+					f.Reply.Layer3.SrcIP,
+					uint16(f.Reply.Layer4.DstPort),
+					uint16(f.Reply.Layer4.SrcPort),
+				}
+			}
+
 			seenTuples[tuple.key()] = tuple
 			r.addConnection(&rpt, tuple, extraNodeInfo, extraNodeInfo)
 		})

probe/kubernetes/reporter.go

@@ -94,7 +94,7 @@ func (r *Reporter) podEvent(e Event, pod Pod) {
 }
 
 func isPauseContainer(n report.Node, rpt report.Report) bool {
-	containerImageIDs, ok := n.Sets.Lookup(report.ContainerImage)
+	containerImageIDs, ok := n.Parents.Lookup(report.ContainerImage)
 	if !ok {
 		return false
 	}