Use an exec'd conntrack in 'events' mode instead of repeatedly execing it for NAT mappings. Also use conntrack to populate the endpoint table.

probe/endpoint/conntrack.go

@@ -0,0 +1,225 @@
+package endpoint
+
+import (
+	"bufio"
+	"encoding/xml"
+	"fmt"
+	"log"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/weaveworks/scope/test/exec"
+)
+
+// Constants exported for testing
+const (
+	modules          = "/proc/modules"
+	conntrackModule  = "nf_conntrack"
+	XMLHeader        = "<?xml version=\"1.0\" encoding=\"utf-8\"?>\n"
+	ConntrackOpenTag = "<conntrack>\n"
+	TimeWait         = "TIME_WAIT"
+	TCP              = "tcp"
+	New              = "new"
+	Update           = "update"
+	Destroy          = "destroy"
+)
+
+// Layer3 - these structs are for the parsed conntrack output
+type Layer3 struct {
+	XMLName xml.Name `xml:"layer3"`
+	SrcIP   string   `xml:"src"`
+	DstIP   string   `xml:"dst"`
+}
+
+// Layer4 - these structs are for the parsed conntrack output
+type Layer4 struct {
+	XMLName xml.Name `xml:"layer4"`
+	SrcPort int      `xml:"sport"`
+	DstPort int      `xml:"dport"`
+	Proto   string   `xml:"protoname,attr"`
+}
+
+// Meta - these structs are for the parsed conntrack output
+type Meta struct {
+	XMLName   xml.Name `xml:"meta"`
+	Direction string   `xml:"direction,attr"`
+	Layer3    Layer3   `xml:"layer3"`
+	Layer4    Layer4   `xml:"layer4"`
+	ID        int64    `xml:"id"`
+	State     string   `xml:"state"`
+}
+
+// Flow - these structs are for the parsed conntrack output
+type Flow struct {
+	XMLName xml.Name `xml:"flow"`
+	Metas   []Meta   `xml:"meta"`
+	Type    string   `xml:"type,attr"`
+
+	Original, Reply, Independent *Meta `xml:"-"`
+}
+
+// Conntracker uses the conntrack command to track network connections
+type Conntracker struct {
+	sync.Mutex
+	cmd           exec.Cmd
+	activeFlows   map[int64]Flow // active flows in state != TIME_WAIT
+	bufferedFlows []Flow         // flows coming out of activeFlows spend 1 walk cycle here
+}
+
+// NewConntracker creates and starts a new Conntracter
+func NewConntracker(args ...string) (*Conntracker, error) {
+	if !ConntrackModulePresent() {
+		return nil, fmt.Errorf("No conntrack module")
+	}
+	result := &Conntracker{
+		activeFlows: map[int64]Flow{},
+	}
+	go result.run(args...)
+	return result, nil
+}
+
+// ConntrackModulePresent returns true if the kernel has the conntrack module
+// present.  It is made public for mocking.
+var ConntrackModulePresent = func() bool {
+	f, err := os.Open(modules)
+	if err != nil {
+		return false
+	}
+	defer f.Close()
+
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if strings.HasPrefix(line, conntrackModule) {
+			return true
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		log.Printf("conntrack error: %v", err)
+	}
+
+	log.Printf("conntrack: failed to find module %s", conntrackModule)
+	return false
+}
+
+// NB this is not re-entrant!
+func (c *Conntracker) run(args ...string) {
+	args = append([]string{"-E", "-o", "xml"}, args...)
+	cmd := exec.Command("conntrack", args...)
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		log.Printf("conntrack error: %v", err)
+		return
+	}
+	if err := cmd.Start(); err != nil {
+		log.Printf("conntrack error: %v", err)
+		return
+	}
+
+	c.Lock()
+	c.cmd = cmd
+	c.Unlock()
+	defer func() {
+		if err := cmd.Wait(); err != nil {
+			log.Printf("conntrack error: %v", err)
+		}
+	}()
+
+	// Swallow the first two lines
+	reader := bufio.NewReader(stdout)
+	if line, err := reader.ReadString('\n'); err != nil {
+		log.Printf("conntrack error: %v", err)
+		return
+	} else if line != XMLHeader {
+		log.Printf("conntrack invalid output: '%s'", line)
+		return
+	}
+	if line, err := reader.ReadString('\n'); err != nil {
+		log.Printf("conntrack error: %v", err)
+		return
+	} else if line != ConntrackOpenTag {
+		log.Printf("conntrack invalid output: '%s'", line)
+		return
+	}
+
+	// Now loop on the output stream
+	decoder := xml.NewDecoder(reader)
+	for {
+		var f Flow
+		if err := decoder.Decode(&f); err != nil {
+			log.Printf("conntrack error: %v", err)
+		}
+		c.handleFlow(f)
+	}
+}
+
+// Stop stop stop
+func (c *Conntracker) Stop() {
+	c.Lock()
+	defer c.Unlock()
+	if c.cmd == nil {
+		return
+	}
+
+	if p := c.cmd.Process(); p != nil {
+		p.Kill()
+	}
+}
+
+func (c *Conntracker) handleFlow(f Flow) {
+	// A flow consists of 3 'metas' - the 'original' 4 tuple (as seen by this
+	// host) and the 'reply' 4 tuple, which is what it has been rewritten to.
+	// This code finds those metas, which are identified by a Direction
+	// attribute.
+	for i := range f.Metas {
+		meta := &f.Metas[i]
+		switch meta.Direction {
+		case "original":
+			f.Original = meta
+		case "reply":
+			f.Reply = meta
+		case "independent":
+			f.Independent = meta
+		}
+	}
+
+	// For not, I'm only interested in tcp connections - there is too much udp
+	// traffic going on (every container talking to weave dns, for example) to
+	// render nicely. TODO: revisit this.
+	if f.Original.Layer4.Proto != TCP {
+		return
+	}
+
+	c.Lock()
+	defer c.Unlock()
+
+	switch f.Type {
+	case New, Update:
+		if f.Independent.State != TimeWait {
+			c.activeFlows[f.Independent.ID] = f
+		} else if _, ok := c.activeFlows[f.Independent.ID]; ok {
+			delete(c.activeFlows, f.Independent.ID)
+			c.bufferedFlows = append(c.bufferedFlows, f)
+		}
+	case Destroy:
+		if _, ok := c.activeFlows[f.Independent.ID]; ok {
+			delete(c.activeFlows, f.Independent.ID)
+			c.bufferedFlows = append(c.bufferedFlows, f)
+		}
+	}
+}
+
+// WalkFlows calls f with all active flows and flows that have come and gone
+// since the last call to WalkFlows
+func (c *Conntracker) WalkFlows(f func(Flow)) {
+	c.Lock()
+	defer c.Unlock()
+	for _, flow := range c.activeFlows {
+		f(flow)
+	}
+	for _, flow := range c.bufferedFlows {
+		f(flow)
+	}
+	c.bufferedFlows = c.bufferedFlows[:0]
+}

probe/endpoint/conntrack_test.go

@@ -0,0 +1,142 @@
+package endpoint_test
+
+import (
+	"bufio"
+	"encoding/xml"
+	"io"
+	"testing"
+	"time"
+
+	. "github.com/weaveworks/scope/probe/endpoint"
+	"github.com/weaveworks/scope/test"
+	"github.com/weaveworks/scope/test/exec"
+)
+
+func makeFlow(id int64, srcIP, dstIP string, srcPort, dstPort int, ty, state string) Flow {
+	return Flow{
+		XMLName: xml.Name{
+			Local: "flow",
+		},
+		Type: ty,
+		Metas: []Meta{
+			{
+				XMLName: xml.Name{
+					Local: "meta",
+				},
+				Direction: "original",
+				Layer3: Layer3{
+					XMLName: xml.Name{
+						Local: "layer3",
+					},
+					SrcIP: srcIP,
+					DstIP: dstIP,
+				},
+				Layer4: Layer4{
+					XMLName: xml.Name{
+						Local: "layer4",
+					},
+					SrcPort: srcPort,
+					DstPort: dstPort,
+					Proto:   TCP,
+				},
+			},
+			{
+				XMLName: xml.Name{
+					Local: "meta",
+				},
+				Direction: "independent",
+				ID:        id,
+				State:     state,
+				Layer3: Layer3{
+					XMLName: xml.Name{
+						Local: "layer3",
+					},
+				},
+				Layer4: Layer4{
+					XMLName: xml.Name{
+						Local: "layer4",
+					},
+				},
+			},
+		},
+	}
+}
+
+func TestConntracker(t *testing.T) {
+	oldExecCmd, oldConntrackPresent := exec.Command, ConntrackModulePresent
+	defer func() { exec.Command, ConntrackModulePresent = oldExecCmd, oldConntrackPresent }()
+
+	ConntrackModulePresent = func() bool {
+		return true
+	}
+
+	reader, writer := io.Pipe()
+	exec.Command = func(name string, args ...string) exec.Cmd {
+		return exec.NewMockCmd(reader)
+	}
+
+	conntracker, err := NewConntracker()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	bw := bufio.NewWriter(writer)
+	if _, err := bw.WriteString(XMLHeader); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := bw.WriteString(ConntrackOpenTag); err != nil {
+		t.Fatal(err)
+	}
+	if err := bw.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	have := func() interface{} {
+		result := []Flow{}
+		conntracker.WalkFlows(func(f Flow) {
+			f.Original = nil
+			f.Reply = nil
+			f.Independent = nil
+			result = append(result, f)
+		})
+		return result
+	}
+	ts := 100 * time.Millisecond
+
+	// First, assert we have no flows
+	test.Poll(t, ts, []Flow{}, have)
+
+	// Now add some flows
+	xmlEncoder := xml.NewEncoder(bw)
+	writeFlow := func(f Flow) {
+		if err := xmlEncoder.Encode(f); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := bw.WriteString("\n"); err != nil {
+			t.Fatal(err)
+		}
+		if err := bw.Flush(); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	flow1 := makeFlow(1, "1.2.3.4", "2.3.4.5", 2, 3, New, "")
+	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 = Destroy
+	writeFlow(flow1)
+	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 = New
+	writeFlow(flow1)
+	test.Poll(t, ts, []Flow{flow1}, have)
+
+	flow1.Metas[1].State = TimeWait
+	writeFlow(flow1)
+	test.Poll(t, ts, []Flow{flow1}, have)
+	test.Poll(t, ts, []Flow{}, have)
+}

probe/endpoint/nat.go

@@ -1,54 +1,11 @@
 package endpoint
 
 import (
-	"bufio"
-	"encoding/xml"
-	"io"
-	"log"
-	"os"
-	"os/exec"
 	"strconv"
-	"strings"
 
 	"github.com/weaveworks/scope/report"
 )
 
-const (
-	modules         = "/proc/modules"
-	conntrackModule = "nf_conntrack"
-)
-
-// these structs are for the parsed conntrack output
-type layer3 struct {
-	XMLName xml.Name `xml:"layer3"`
-	SrcIP   string   `xml:"src"`
-	DstIP   string   `xml:"dst"`
-}
-
-type layer4 struct {
-	XMLName xml.Name `xml:"layer4"`
-	SrcPort int      `xml:"sport"`
-	DstPort int      `xml:"dport"`
-	Proto   string   `xml:"protoname,attr"`
-}
-
-type meta struct {
-	XMLName   xml.Name `xml:"meta"`
-	Direction string   `xml:"direction,attr"`
-	Layer3    layer3   `xml:"layer3"`
-	Layer4    layer4   `xml:"layer4"`
-}
-
-type flow struct {
-	XMLName xml.Name `xml:"flow"`
-	Metas   []meta   `xml:"meta"`
-}
-
-type conntrack struct {
-	XMLName xml.Name `xml:"conntrack"`
-	Flows   []flow   `xml:"flow"`
-}
-
 // This is our 'abstraction' of the endpoint that have been rewritten by NAT.
 // Original is the private IP that has been rewritten.
 type endpointMapping struct {
@@ -59,111 +16,51 @@ type endpointMapping struct {
 	rewrittenPort int
 }
 
-// natTable returns a list of endpoints that have been remapped by NAT.
-func natTable() ([]endpointMapping, error) {
-	var conntrack conntrack
-	cmd := exec.Command("conntrack", "-L", "--any-nat", "-o", "xml")
-	stdout, err := cmd.StdoutPipe()
+type natmapper struct {
+	*Conntracker
+}
+
+func newNATMapper() (*natmapper, error) {
+	ct, err := NewConntracker("--any-nat")
 	if err != nil {
 		return nil, err
 	}
-	if err := cmd.Start(); err != nil {
-		return nil, err
-	}
-	defer func() {
-		if err := cmd.Wait(); err != nil {
-			log.Printf("conntrack error: %v", err)
+	return &natmapper{ct}, nil
+}
+
+func toMapping(f Flow) *endpointMapping {
+	var mapping endpointMapping
+	if f.Original.Layer3.SrcIP == f.Reply.Layer3.DstIP {
+		mapping = endpointMapping{
+			originalIP:    f.Reply.Layer3.SrcIP,
+			originalPort:  f.Reply.Layer4.SrcPort,
+			rewrittenIP:   f.Original.Layer3.DstIP,
+			rewrittenPort: f.Original.Layer4.DstPort,
 		}
-	}()
-	if err := xml.NewDecoder(stdout).Decode(&conntrack); err != nil {
-		if err == io.EOF {
-			return []endpointMapping{}, nil
+	} else {
+		mapping = endpointMapping{
+			originalIP:    f.Original.Layer3.SrcIP,
+			originalPort:  f.Original.Layer4.SrcPort,
+			rewrittenIP:   f.Reply.Layer3.DstIP,
+			rewrittenPort: f.Reply.Layer4.DstPort,
 		}
-		return nil, err
 	}
 
-	output := []endpointMapping{}
-	for _, flow := range conntrack.Flows {
-		// A flow consists of 3 'metas' - the 'original' 4 tuple (as seen by this
-		// host) and the 'reply' 4 tuple, which is what it has been rewritten to.
-		// This code finds those metas, which are identified by a Direction
-		// attribute.
-		original, reply := meta{}, meta{}
-		for _, meta := range flow.Metas {
-			if meta.Direction == "original" {
-				original = meta
-			} else if meta.Direction == "reply" {
-				reply = meta
-			}
-		}
-
-		if original.Layer4.Proto != "tcp" {
-			continue
-		}
-
-		var conn endpointMapping
-		if original.Layer3.SrcIP == reply.Layer3.DstIP {
-			conn = endpointMapping{
-				originalIP:    reply.Layer3.SrcIP,
-				originalPort:  reply.Layer4.SrcPort,
-				rewrittenIP:   original.Layer3.DstIP,
-				rewrittenPort: original.Layer4.DstPort,
-			}
-		} else {
-			conn = endpointMapping{
-				originalIP:    original.Layer3.SrcIP,
-				originalPort:  original.Layer4.SrcPort,
-				rewrittenIP:   reply.Layer3.DstIP,
-				rewrittenPort: reply.Layer4.DstPort,
-			}
-		}
-
-		output = append(output, conn)
-	}
-
-	return output, nil
+	return &mapping
 }
 
 // applyNAT duplicates NodeMetadatas in the endpoint topology of a
 // report, based on the NAT table as returns by natTable.
-func applyNAT(rpt report.Report, scope string) error {
-	mappings, err := natTable()
-	if err != nil {
-		return err
-	}
-
-	for _, mapping := range mappings {
+func (n *natmapper) applyNAT(rpt report.Report, scope string) {
+	n.WalkFlows(func(f Flow) {
+		mapping := toMapping(f)
 		realEndpointID := report.MakeEndpointNodeID(scope, mapping.originalIP, strconv.Itoa(mapping.originalPort))
 		copyEndpointID := report.MakeEndpointNodeID(scope, mapping.rewrittenIP, strconv.Itoa(mapping.rewrittenPort))
 		nmd, ok := rpt.Endpoint.NodeMetadatas[realEndpointID]
 		if !ok {
-			continue
+			return
 		}
 
 		rpt.Endpoint.NodeMetadatas[copyEndpointID] = nmd.Copy()
-	}
-
-	return nil
-}
-
-func conntrackModulePresent() bool {
-	f, err := os.Open(modules)
-	if err != nil {
-		return false
-	}
-	defer f.Close()
-
-	scanner := bufio.NewScanner(f)
-	for scanner.Scan() {
-		line := scanner.Text()
-		if strings.HasPrefix(line, conntrackModule) {
-			return true
-		}
-	}
-	if err := scanner.Err(); err != nil {
-		log.Printf("conntrack error: %v", err)
-	}
-
-	log.Printf("conntrack: failed to find module %s", conntrackModule)
-	return false
+	})
 }

probe/endpoint/reporter.go

@@ -1,7 +1,7 @@
 package endpoint
 
 import (
-	"fmt"
+	"log"
 	"strconv"
 	"time"
 
@@ -24,6 +24,8 @@ type Reporter struct {
 	hostName         string
 	includeProcesses bool
 	includeNAT       bool
+	conntracker      *Conntracker
+	natmapper        *natmapper
 }
 
 // SpyDuration is an exported prometheus metric
@@ -43,12 +45,41 @@ var SpyDuration = prometheus.NewSummaryVec(
 // on the host machine, at the granularity of host and port. That information
 // is stored in the Endpoint topology. It optionally enriches that topology
 // with process (PID) information.
-func NewReporter(hostID, hostName string, includeProcesses bool) *Reporter {
+func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool) *Reporter {
+	var (
+		conntrackModulePresent = ConntrackModulePresent()
+		conntracker            *Conntracker
+		natmapper              *natmapper
+		err                    error
+	)
+	if conntrackModulePresent && useConntrack {
+		conntracker, err = NewConntracker()
+		if err != nil {
+			log.Printf("Failed to start conntracker: %v", err)
+		}
+	}
+	if conntrackModulePresent {
+		natmapper, err = newNATMapper()
+		if err != nil {
+			log.Printf("Failed to start natMapper: %v", err)
+		}
+	}
 	return &Reporter{
 		hostID:           hostID,
 		hostName:         hostName,
 		includeProcesses: includeProcesses,
-		includeNAT:       conntrackModulePresent(),
+		conntracker:      conntracker,
+		natmapper:        natmapper,
+	}
+}
+
+// Stop stop stop
+func (r *Reporter) Stop() {
+	if r.conntracker != nil {
+		r.conntracker.Stop()
+	}
+	if r.natmapper != nil {
+		r.natmapper.Stop()
 	}
 }
 
@@ -65,50 +96,73 @@ func (r *Reporter) Report() (report.Report, error) {
 	}
 
 	for conn := conns.Next(); conn != nil; conn = conns.Next() {
-		r.addConnection(&rpt, conn)
+		var (
+			localPort  = conn.LocalPort
+			remotePort = conn.RemotePort
+			localAddr  = conn.LocalAddress.String()
+			remoteAddr = conn.RemoteAddress.String()
+		)
+		r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &conn.Proc)
 	}
 
-	if r.includeNAT {
-		err = applyNAT(rpt, r.hostID)
+	if r.conntracker != nil {
+		r.conntracker.WalkFlows(func(f Flow) {
+			var (
+				localPort  = f.Original.Layer4.SrcPort
+				remotePort = f.Original.Layer4.DstPort
+				localAddr  = f.Original.Layer3.SrcIP
+				remoteAddr = f.Original.Layer3.DstIP
+			)
+			r.addConnection(&rpt, localAddr, remoteAddr, uint16(localPort), uint16(remotePort), nil)
+		})
+	}
+
+	if r.natmapper != nil {
+		r.natmapper.applyNAT(rpt, r.hostID)
 	}
 
 	return rpt, err
 }
 
-func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) {
-	var (
-		localIsClient       = int(c.LocalPort) > int(c.RemotePort)
-		localAddressNodeID  = report.MakeAddressNodeID(r.hostID, c.LocalAddress.String())
-		remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, c.RemoteAddress.String())
-		adjacencyID         = ""
-		edgeID              = ""
-	)
+func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr string, localPort, remotePort uint16, proc *procspy.Proc) {
+	localIsClient := int(localPort) > int(remotePort)
 
-	if localIsClient {
-		adjacencyID = report.MakeAdjacencyID(localAddressNodeID)
-		rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(remoteAddressNodeID)
-
-		edgeID = report.MakeEdgeID(localAddressNodeID, remoteAddressNodeID)
-	} else {
-		adjacencyID = report.MakeAdjacencyID(remoteAddressNodeID)
-		rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(localAddressNodeID)
-
-		edgeID = report.MakeEdgeID(remoteAddressNodeID, localAddressNodeID)
-	}
-
-	if _, ok := rpt.Address.NodeMetadatas[localAddressNodeID]; !ok {
-		rpt.Address.NodeMetadatas[localAddressNodeID] = report.MakeNodeMetadataWith(map[string]string{
-			"name": r.hostName,
-			Addr:   c.LocalAddress.String(),
-		})
-	}
-
-	countTCPConnection(rpt.Address.EdgeMetadatas, edgeID)
-
-	if c.Proc.PID > 0 {
+	// Update address topology
+	{
 		var (
-			localEndpointNodeID  = report.MakeEndpointNodeID(r.hostID, c.LocalAddress.String(), strconv.Itoa(int(c.LocalPort)))
-			remoteEndpointNodeID = report.MakeEndpointNodeID(r.hostID, c.RemoteAddress.String(), strconv.Itoa(int(c.RemotePort)))
+			localAddressNodeID  = report.MakeAddressNodeID(r.hostID, localAddr)
+			remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, remoteAddr)
+			adjacencyID         = ""
+			edgeID              = ""
+		)
+
+		if localIsClient {
+			adjacencyID = report.MakeAdjacencyID(localAddressNodeID)
+			rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(remoteAddressNodeID)
+
+			edgeID = report.MakeEdgeID(localAddressNodeID, remoteAddressNodeID)
+		} else {
+			adjacencyID = report.MakeAdjacencyID(remoteAddressNodeID)
+			rpt.Address.Adjacency[adjacencyID] = rpt.Address.Adjacency[adjacencyID].Add(localAddressNodeID)
+
+			edgeID = report.MakeEdgeID(remoteAddressNodeID, localAddressNodeID)
+		}
+
+		countTCPConnection(rpt.Address.EdgeMetadatas, edgeID)
+
+		if _, ok := rpt.Address.NodeMetadatas[localAddressNodeID]; !ok {
+			rpt.Address.NodeMetadatas[localAddressNodeID] = report.MakeNodeMetadataWith(map[string]string{
+				"name": r.hostName,
+				Addr:   localAddr,
+			})
+		}
+	}
+
+	// Update endpoint topology
+	if r.includeProcesses {
+		var (
+			localEndpointNodeID  = report.MakeEndpointNodeID(r.hostID, localAddr, strconv.Itoa(int(localPort)))
+			remoteEndpointNodeID = report.MakeEndpointNodeID(r.hostID, remoteAddr, strconv.Itoa(int(remotePort)))
 			adjacencyID          = ""
 			edgeID               = ""
 		)
@@ -125,18 +179,24 @@ func (r *Reporter) addConnection(rpt *report.Report, c *procspy.Connection) {
 			edgeID = report.MakeEdgeID(remoteEndpointNodeID, localEndpointNodeID)
 		}
 
-		if _, ok := rpt.Endpoint.NodeMetadatas[localEndpointNodeID]; !ok {
-			// First hit establishes NodeMetadata for scoped local address + port
-			md := report.MakeNodeMetadataWith(map[string]string{
-				Addr:        c.LocalAddress.String(),
-				Port:        strconv.Itoa(int(c.LocalPort)),
-				process.PID: fmt.Sprint(c.Proc.PID),
-			})
+		countTCPConnection(rpt.Endpoint.EdgeMetadatas, edgeID)
 
+		md, ok := rpt.Endpoint.NodeMetadatas[localEndpointNodeID]
+		updated := !ok
+		if !ok {
+			md = report.MakeNodeMetadataWith(map[string]string{
+				Addr: localAddr,
+				Port: strconv.Itoa(int(localPort)),
+			})
+		}
+		if proc != nil && proc.PID > 0 {
+			pid := strconv.FormatUint(uint64(proc.PID), 10)
+			updated = updated || md.Metadata[process.PID] != pid
+			md.Metadata[process.PID] = pid
+		}
+		if updated {
 			rpt.Endpoint.NodeMetadatas[localEndpointNodeID] = md
 		}
-
-		countTCPConnection(rpt.Endpoint.EdgeMetadatas, edgeID)
 	}
 }
 

probe/endpoint/reporter_test.go

@@ -19,7 +19,6 @@ var (
 	fixRemotePortB   = uint16(12346)
 	fixProcessPID    = uint(4242)
 	fixProcessName   = "nginx"
-	fixProcessPIDB   = uint(4243)
 
 	fixConnections = []procspy.Connection{
 		{
@@ -55,9 +54,9 @@ var (
 			LocalAddress:  fixLocalAddress,
 			LocalPort:     fixLocalPort,
 			RemoteAddress: fixRemoteAddress,
-			RemotePort:    fixRemotePort,
+			RemotePort:    fixRemotePortB,
 			Proc: procspy.Proc{
-				PID:  fixProcessPIDB,
+				PID:  fixProcessPID,
 				Name: fixProcessName,
 			},
 		},
@@ -72,7 +71,7 @@ func TestSpyNoProcesses(t *testing.T) {
 		nodeName = "frenchs-since-1904"   // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, false)
+	reporter := endpoint.NewReporter(nodeID, nodeName, false, false)
 	r, _ := reporter.Report()
 	//buf, _ := json.MarshalIndent(r, "", "    ")
 	//t.Logf("\n%s\n", buf)
@@ -109,7 +108,7 @@ func TestSpyWithProcesses(t *testing.T) {
 		nodeName = "fishermans-friend" // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, false)
+	reporter := endpoint.NewReporter(nodeID, nodeName, true, false)
 	r, _ := reporter.Report()
 	// buf, _ := json.MarshalIndent(r, "", "    ") ; t.Logf("\n%s\n", buf)
 

probe/main.go

@@ -46,6 +46,7 @@ func main() {
 		captureOn          = flag.Duration("capture.on", 1*time.Second, "packet capture duty cycle 'on'")
 		captureOff         = flag.Duration("capture.off", 5*time.Second, "packet capture duty cycle 'off'")
 		printVersion       = flag.Bool("version", false, "print version number and exit")
+		useConntrack       = flag.Bool("conntrack", true, "also use conntrack to track connections")
 	)
 	flag.Parse()
 
@@ -103,13 +104,16 @@ func main() {
 	}
 
 	var (
-		taggers      = []Tagger{newTopologyTagger(), host.NewTagger(hostID)}
-		reporters    = []Reporter{host.NewReporter(hostID, hostName, localNets), endpoint.NewReporter(hostID, hostName, *spyProcs)}
-		processCache *process.CachingWalker
+		endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack)
+		processCache     = process.NewCachingWalker(process.NewWalker(*procRoot))
+		reporters        = []Reporter{
+			endpointReporter,
+			host.NewReporter(hostID, hostName, localNets),
+			process.NewReporter(processCache, hostID),
+		}
+		taggers = []Tagger{newTopologyTagger(), host.NewTagger(hostID)}
 	)
-
-	processCache = process.NewCachingWalker(process.NewWalker(*procRoot))
-	reporters = append(reporters, process.NewReporter(processCache, hostID))
+	defer endpointReporter.Stop()
 
 	if *dockerEnabled {
 		if err := report.AddLocalBridge(*dockerBridge); err != nil {

probe/overlay/weave.go

@@ -4,17 +4,16 @@ import (
 	"bufio"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log"
 	"net"
 	"net/http"
 	"net/url"
-	"os/exec"
 	"regexp"
 	"strings"
 
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/report"
+	"github.com/weaveworks/scope/test/exec"
 )
 
 const (
@@ -36,16 +35,6 @@ const (
 var weavePsMatch = regexp.MustCompile(`^([0-9a-f]{12}) ((?:[0-9a-f][0-9a-f]\:){5}(?:[0-9a-f][0-9a-f]))(.*)$`)
 var ipMatch = regexp.MustCompile(`([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})(/[0-9]+)`)
 
-// Cmd is a hook for mocking
-type Cmd interface {
-	StdoutPipe() (io.ReadCloser, error)
-	Start() error
-	Wait() error
-}
-
-// ExecCommand is a hook for mocking
-var ExecCommand = func(name string, args ...string) Cmd { return exec.Command(name, args...) }
-
 // Weave represents a single Weave router, presumably on the same host
 // as the probe. It is both a Reporter and a Tagger: it produces an Overlay
 // topology, and (in theory) can tag existing topologies with foreign keys to
@@ -114,7 +103,7 @@ type psEntry struct {
 
 func (w Weave) ps() ([]psEntry, error) {
 	var result []psEntry
-	cmd := ExecCommand("weave", "--local", "ps")
+	cmd := exec.Command("weave", "--local", "ps")
 	out, err := cmd.StdoutPipe()
 	if err != nil {
 		return result, err

probe/overlay/weave_test.go

@@ -1,10 +1,7 @@
 package overlay_test
 
 import (
-	"bytes"
 	"fmt"
-	"io"
-	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"reflect"
@@ -14,37 +11,14 @@ import (
 	"github.com/weaveworks/scope/probe/overlay"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
+	"github.com/weaveworks/scope/test/exec"
 )
 
-type mockCmd struct {
-	*bytes.Buffer
-}
-
-func (c *mockCmd) Start() error {
-	return nil
-}
-
-func (c *mockCmd) Wait() error {
-	return nil
-}
-
-func (c *mockCmd) StdoutPipe() (io.ReadCloser, error) {
-	return struct {
-		io.Reader
-		io.Closer
-	}{
-		c.Buffer,
-		ioutil.NopCloser(nil),
-	}, nil
-}
-
 func TestWeaveTaggerOverlayTopology(t *testing.T) {
-	oldExecCmd := overlay.ExecCommand
-	defer func() { overlay.ExecCommand = oldExecCmd }()
-	overlay.ExecCommand = func(name string, args ...string) overlay.Cmd {
-		return &mockCmd{
-			bytes.NewBufferString(fmt.Sprintf("%s %s %s/24\n", mockContainerID, mockContainerMAC, mockContainerIP)),
-		}
+	oldExecCmd := exec.Command
+	defer func() { exec.Command = oldExecCmd }()
+	exec.Command = func(name string, args ...string) exec.Cmd {
+		return exec.NewMockCmdString(fmt.Sprintf("%s %s %s/24\n", mockContainerID, mockContainerMAC, mockContainerIP))
 	}
 
 	s := httptest.NewServer(http.HandlerFunc(mockWeaveRouter))

test/exec/exec.go

@@ -0,0 +1,68 @@
+package exec
+
+import (
+	"bytes"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+)
+
+// Cmd is a hook for mocking
+type Cmd interface {
+	StdoutPipe() (io.ReadCloser, error)
+	Start() error
+	Wait() error
+	Process() *os.Process
+}
+
+// Command is a hook for mocking
+var Command = func(name string, args ...string) Cmd {
+	return &realCmd{exec.Command(name, args...)}
+}
+
+type realCmd struct {
+	*exec.Cmd
+}
+
+func (c *realCmd) Process() *os.Process {
+	return c.Cmd.Process
+}
+
+type mockCmd struct {
+	io.ReadCloser
+}
+
+// NewMockCmdString creates a new mock Cmd which has s on its stdout pipe
+func NewMockCmdString(s string) Cmd {
+	return &mockCmd{
+		struct {
+			io.Reader
+			io.Closer
+		}{
+			bytes.NewBufferString(s),
+			ioutil.NopCloser(nil),
+		},
+	}
+}
+
+// NewMockCmd creates a new mock Cmd with rc as its stdout pipe
+func NewMockCmd(rc io.ReadCloser) Cmd {
+	return &mockCmd{rc}
+}
+
+func (c *mockCmd) Start() error {
+	return nil
+}
+
+func (c *mockCmd) Wait() error {
+	return nil
+}
+
+func (c *mockCmd) StdoutPipe() (io.ReadCloser, error) {
+	return c.ReadCloser, nil
+}
+
+func (c *mockCmd) Process() *os.Process {
+	return nil
+}

test/poll.go

@@ -2,6 +2,7 @@ package test
 
 import (
 	"reflect"
+	"runtime"
 	"testing"
 	"time"
 )
@@ -20,6 +21,7 @@ func Poll(t *testing.T, d time.Duration, want interface{}, have func() interface
 	}
 	h := have()
 	if !reflect.DeepEqual(want, h) {
-		t.Fatal(Diff(want, h))
+		_, file, line, _ := runtime.Caller(1)
+		t.Fatalf("%s:%d: %s", file, line, Diff(want, h))
 	}
 }