Merge pull request #450 from weaveworks/scope-284

Optimize /proc traversal

probe/docker/tagger.go

@@ -3,6 +3,7 @@ package docker
 import (
 	"strconv"
 
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/report"
 )
@@ -16,18 +17,18 @@ const (
 
 // These vars are exported for testing.
 var (
-	NewProcessTreeStub = process.NewTree
+	NewProcessTreeStub = proc.NewTree
 )
 
 // Tagger is a tagger that tags Docker container information to process
 // nodes that have a PID.
 type Tagger struct {
 	registry   Registry
-	procWalker process.Walker
+	procWalker proc.Reader
 }
 
 // NewTagger returns a usable Tagger.
-func NewTagger(registry Registry, procWalker process.Walker) *Tagger {
+func NewTagger(registry Registry, procWalker proc.Reader) *Tagger {
 	return &Tagger{
 		registry:   registry,
 		procWalker: procWalker,
@@ -44,7 +45,7 @@ func (t *Tagger) Tag(r report.Report) (report.Report, error) {
 	return r, nil
 }
 
-func (t *Tagger) tag(tree process.Tree, topology *report.Topology) {
+func (t *Tagger) tag(tree proc.Tree, topology *report.Topology) {
 	for nodeID, nodeMetadata := range topology.Nodes {
 		pidStr, ok := nodeMetadata.Metadata[process.PID]
 		if !ok {

probe/docker/tagger_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 
 	"github.com/weaveworks/scope/probe/docker"
-	"github.com/weaveworks/scope/probe/process"
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
 )
@@ -27,7 +27,7 @@ func TestTagger(t *testing.T) {
 	oldProcessTree := docker.NewProcessTreeStub
 	defer func() { docker.NewProcessTreeStub = oldProcessTree }()
 
-	docker.NewProcessTreeStub = func(_ process.Walker) (process.Tree, error) {
+	docker.NewProcessTreeStub = func(_ proc.Reader) (proc.Tree, error) {
 		return &mockProcessTree{map[int]int{2: 1}}, nil
 	}
 

probe/endpoint/reporter.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 
-	"github.com/weaveworks/procspy"
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/report"
 )
@@ -28,6 +28,7 @@ type Reporter struct {
 	conntracker      *Conntracker
 	natmapper        *natmapper
 	revResolver      *ReverseResolver
+	procReader       proc.Reader
 }
 
 // SpyDuration is an exported prometheus metric
@@ -42,12 +43,12 @@ var SpyDuration = prometheus.NewSummaryVec(
 	[]string{},
 )
 
-// NewReporter creates a new Reporter that invokes procspy.Connections to
+// NewReporter creates a new Reporter that invokes Connections to
 // generate a report.Report that contains every discovered (spied) connection
 // 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, useConntrack bool) *Reporter {
+func NewReporter(hostID, hostName string, includeProcesses bool, procReader proc.Reader, useConntrack bool) *Reporter {
 	var (
 		conntrackModulePresent = ConntrackModulePresent()
 		conntracker            *Conntracker
@@ -73,6 +74,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
 		conntracker:      conntracker,
 		natmapper:        natmapper,
 		revResolver:      NewReverseResolver(),
+		procReader:       procReader,
 	}
 }
 
@@ -94,12 +96,7 @@ func (r *Reporter) Report() (report.Report, error) {
 	}(time.Now())
 
 	rpt := report.MakeReport()
-	conns, err := procspy.Connections(r.includeProcesses)
-	if err != nil {
-		return rpt, err
-	}
-
-	for conn := conns.Next(); conn != nil; conn = conns.Next() {
+	err := r.procReader.Connections(r.includeProcesses, func(conn proc.Connection) {
 		var (
 			localPort  = conn.LocalPort
 			remotePort = conn.RemotePort
@@ -107,12 +104,15 @@ func (r *Reporter) Report() (report.Report, error) {
 			remoteAddr = conn.RemoteAddress.String()
 		)
 		extraNodeInfo := report.MakeNode()
-		if conn.Proc.PID > 0 {
+		if conn.Process.PID > 0 {
 			extraNodeInfo = extraNodeInfo.WithMetadata(report.Metadata{
-				process.PID: strconv.FormatUint(uint64(conn.Proc.PID), 10),
+				process.PID: strconv.FormatUint(uint64(conn.Process.PID), 10),
 			})
 		}
 		r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &extraNodeInfo, nil)
+	})
+	if err != nil {
+		return rpt, err
 	}
 
 	if r.conntracker != nil {

probe/endpoint/reporter_test.go

@@ -5,9 +5,9 @@ import (
 	"strconv"
 	"testing"
 
-	"github.com/weaveworks/procspy"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/endpoint"
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -17,10 +17,10 @@ var (
 	fixRemoteAddress = net.ParseIP("192.168.1.2")
 	fixRemotePort    = uint16(12345)
 	fixRemotePortB   = uint16(12346)
-	fixProcessPID    = uint(4242)
+	fixProcessPID    = int(4242)
 	fixProcessName   = "nginx"
 
-	fixConnections = []procspy.Connection{
+	fixConnections = []proc.Connection{
 		{
 			Transport:     "tcp",
 			LocalAddress:  fixLocalAddress,
@@ -37,16 +37,16 @@ var (
 		},
 	}
 
-	fixConnectionsWithProcesses = []procspy.Connection{
+	fixConnectionsWithProcesses = []proc.Connection{
 		{
 			Transport:     "tcp",
 			LocalAddress:  fixLocalAddress,
 			LocalPort:     fixLocalPort,
 			RemoteAddress: fixRemoteAddress,
 			RemotePort:    fixRemotePort,
-			Proc: procspy.Proc{
+			Process: proc.Process{
 				PID:  fixProcessPID,
-				Name: fixProcessName,
+				Comm: fixProcessName,
 			},
 		},
 		{
@@ -55,23 +55,22 @@ var (
 			LocalPort:     fixLocalPort,
 			RemoteAddress: fixRemoteAddress,
 			RemotePort:    fixRemotePort,
-			Proc: procspy.Proc{
+			Process: proc.Process{
 				PID:  fixProcessPID,
-				Name: fixProcessName,
+				Comm: fixProcessName,
 			},
 		},
 	}
 )
 
 func TestSpyNoProcesses(t *testing.T) {
-	procspy.SetFixtures(fixConnections)
-
 	const (
 		nodeID   = "heinz-tomato-ketchup" // TODO rename to hostID
 		nodeName = "frenchs-since-1904"   // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, false, false)
+	procReader := proc.MockedReader{Conns: fixConnections}
+	reporter := endpoint.NewReporter(nodeID, nodeName, false, &procReader, false)
 	r, _ := reporter.Report()
 	//buf, _ := json.MarshalIndent(r, "", "    ")
 	//t.Logf("\n%s\n", buf)
@@ -100,14 +99,13 @@ func TestSpyNoProcesses(t *testing.T) {
 }
 
 func TestSpyWithProcesses(t *testing.T) {
-	procspy.SetFixtures(fixConnectionsWithProcesses)
-
 	const (
 		nodeID   = "nikon"             // TODO rename to hostID
 		nodeName = "fishermans-friend" // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, true, false)
+	procReader := proc.MockedReader{Conns: fixConnectionsWithProcesses}
+	reporter := endpoint.NewReporter(nodeID, nodeName, true, &procReader, false)
 	r, _ := reporter.Report()
 	// buf, _ := json.MarshalIndent(r, "", "    ") ; t.Logf("\n%s\n", buf)
 

probe/main.go

@@ -14,11 +14,11 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/weaveworks/procspy"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/endpoint"
 	"github.com/weaveworks/scope/probe/host"
 	"github.com/weaveworks/scope/probe/overlay"
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/probe/sniff"
 	"github.com/weaveworks/scope/report"
@@ -67,8 +67,6 @@ func main() {
 	}
 	log.Printf("publishing to: %s", strings.Join(targets, ", "))
 
-	procspy.SetProcRoot(*procRoot)
-
 	if *httpListen != "" {
 		log.Printf("profiling data being exported to %s", *httpListen)
 		log.Printf("go tool pprof http://%s/debug/pprof/{profile,heap,block}", *httpListen)
@@ -111,16 +109,18 @@ func main() {
 	}
 
 	var (
-		endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack)
-		processCache     = process.NewCachingWalker(process.NewWalker(*procRoot))
-		tickers          = []Ticker{processCache}
+		procDir          = proc.OSDir{Dir: *procRoot}
+		procReader       = proc.NewCachingProcReader(proc.NewReader(procDir), *spyProcs)
+		tickers          = []Ticker{procReader}
+		endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, procReader, *useConntrack)
 		reporters        = []Reporter{
 			endpointReporter,
 			host.NewReporter(hostID, hostName, localNets),
-			process.NewReporter(processCache, hostID),
+			process.NewReporter(procReader, hostID),
 		}
 		taggers = []Tagger{newTopologyTagger(), host.NewTagger(hostID)}
 	)
+	defer procReader.Close()
 	defer endpointReporter.Stop()
 
 	if *dockerEnabled {
@@ -134,7 +134,7 @@ func main() {
 		}
 		defer dockerRegistry.Stop()
 
-		taggers = append(taggers, docker.NewTagger(dockerRegistry, processCache))
+		taggers = append(taggers, docker.NewTagger(dockerRegistry, procReader))
 		reporters = append(reporters, docker.NewReporter(dockerRegistry, hostID))
 	}
 

probe/proc/mocks.go

@@ -0,0 +1,30 @@
+package proc
+
+// note: we must keep this in the "proc" package so it can be used from other packages..
+
+// MockedReader is a mocked "/proc" reader
+type MockedReader struct {
+	Procs []Process
+	Conns []Connection
+}
+
+// Processes walks through the processes provided in the mocked "/proc" reader
+func (mw MockedReader) Processes(f func(Process)) error {
+	for _, p := range mw.Procs {
+		f(p)
+	}
+	return nil
+}
+
+// Connections walks through the connections provided in the mocked "/proc" reader
+func (mw *MockedReader) Connections(_ bool, f func(Connection)) error {
+	for _, c := range mw.Conns {
+		f(c)
+	}
+	return nil
+}
+
+// Close (mocked version)
+func (mw *MockedReader) Close() error {
+	return nil
+}

probe/proc/mocks_test.go

@@ -0,0 +1,27 @@
+package proc_test
+
+import (
+	"bytes"
+
+	"github.com/weaveworks/scope/probe/proc"
+)
+
+// mockedFile is a mocked file in the "/proc" directory
+type mockedFile struct {
+	Path         string
+	ReadIntoFunc func(buf *bytes.Buffer) error
+}
+
+func (mpf mockedFile) ReadInto(buf *bytes.Buffer) error { return mpf.ReadIntoFunc(buf) }
+func (mpf mockedFile) Close() error                     { return nil }
+
+// mockedDir is a mocked "/proc" directory
+type mockedDir struct {
+	Dir              string
+	OpenFunc         func(string) (proc.File, error)
+	ReadDirNamesFunc func(string) ([]string, error)
+}
+
+func (p mockedDir) Root() string                            { return p.Dir }
+func (p mockedDir) Open(s string) (proc.File, error)        { return p.OpenFunc(s) }
+func (p mockedDir) ReadDirNames(s string) ([]string, error) { return p.ReadDirNamesFunc(s) }

probe/proc/net.go

@@ -0,0 +1,43 @@
+package proc
+
+import (
+	"fmt"
+	"net"
+)
+
+const (
+	tcpEstablished = 1 // according to /include/net/tcp_states.h
+)
+
+// Connection is a (TCP) connection. The 'Process' struct might not be filled in.
+type Connection struct {
+	Transport     string
+	LocalAddress  net.IP
+	LocalPort     uint16
+	RemoteAddress net.IP
+	RemotePort    uint16
+	inode         uint64
+	Process
+}
+
+// Copy returns a copy of a connection
+func (c Connection) Copy() Connection {
+	dupIP := func(ip net.IP) net.IP {
+		dup := make(net.IP, len(ip))
+		copy(dup, ip)
+		return dup
+	}
+
+	c.LocalAddress = dupIP(c.LocalAddress)
+	c.RemoteAddress = dupIP(c.RemoteAddress)
+	c.Process = c.Process.Copy()
+	return c
+}
+
+// String returns the string repr
+func (c Connection) String() string {
+	return fmt.Sprintf("%s:%d - %s:%d %s#%d",
+		c.LocalAddress, c.LocalPort,
+		c.RemoteAddress, c.RemotePort,
+		c.Transport, c.inode)
+}

probe/proc/net_darwin.go

@@ -0,0 +1,81 @@
+package proc
+
+import (
+	"net"
+	"strconv"
+	"strings"
+)
+
+// parseDarwinNetstat parses netstat output. (Linux has ip:port, darwin
+// ip.port. The 'Proto' column value also differs.)
+func parseDarwinNetstat(out string) []Connection {
+	//
+	//  Active Internet connections
+	//  Proto Recv-Q Send-Q  Local Address          Foreign Address        (state)
+	//  tcp4       0      0  10.0.1.6.58287         1.2.3.4.443      		ESTABLISHED
+	//
+	res := []Connection{}
+	for i, line := range strings.Split(out, "\n") {
+		if i == 0 || i == 1 {
+			// Skip header
+			continue
+		}
+
+		// Fields are:
+		fields := strings.Fields(line)
+		if len(fields) != 6 {
+			continue
+		}
+
+		if fields[5] != "ESTABLISHED" {
+			continue
+		}
+
+		t := Connection{
+			Transport: "tcp",
+		}
+
+		// Format is <ip>.<port>
+		locals := strings.Split(fields[3], ".")
+		if len(locals) < 2 {
+			continue
+		}
+
+		var (
+			localAddress = strings.Join(locals[:len(locals)-1], ".")
+			localPort    = locals[len(locals)-1]
+		)
+
+		t.LocalAddress = net.ParseIP(localAddress)
+
+		p, err := strconv.Atoi(localPort)
+		if err != nil {
+			return nil
+		}
+
+		t.LocalPort = uint16(p)
+
+		remotes := strings.Split(fields[4], ".")
+		if len(remotes) < 2 {
+			continue
+		}
+
+		var (
+			remoteAddress = strings.Join(remotes[:len(remotes)-1], ".")
+			remotePort    = remotes[len(remotes)-1]
+		)
+
+		t.RemoteAddress = net.ParseIP(remoteAddress)
+
+		p, err = strconv.Atoi(remotePort)
+		if err != nil {
+			return nil
+		}
+
+		t.RemotePort = uint16(p)
+
+		res = append(res, t)
+	}
+
+	return res
+}

probe/proc/net_linux.go

@@ -0,0 +1,162 @@
+package proc
+
+import (
+	"bytes"
+	"net"
+)
+
+type netReader struct {
+	b                       []byte
+	c                       Connection
+	wantedState             uint
+	bytesLocal, bytesRemote [16]byte
+}
+
+func newNetReader(b []byte, wantedState uint) *netReader {
+	return &netReader{
+		b:           b,
+		c:           Connection{},
+		wantedState: wantedState,
+	}
+}
+
+// Next returns the next connection. All buffers are re-used, so if you want
+// to keep the IPs you have to copy them.
+func (p *netReader) Next() *Connection {
+again:
+	if len(p.b) == 0 {
+		return nil
+	}
+	b := p.b
+
+	if p.b[2] == 's' {
+		// Skip header
+		p.b = nextLine(b)
+		goto again
+	}
+
+	var (
+		local, remote, state, inode []byte
+	)
+	_, b = nextField(b) // 'sl' column
+	local, b = nextField(b)
+	remote, b = nextField(b)
+	state, b = nextField(b)
+	if parseHex(state) != p.wantedState {
+		p.b = nextLine(b)
+		goto again
+	}
+	_, b = nextField(b) // 'tx_queue' column
+	_, b = nextField(b) // 'rx_queue' column
+	_, b = nextField(b) // 'tr' column
+	_, b = nextField(b) // 'uid' column
+	_, b = nextField(b) // 'timeout' column
+	inode, b = nextField(b)
+
+	p.c.LocalAddress, p.c.LocalPort = scanAddressNA(local, &p.bytesLocal)
+	p.c.RemoteAddress, p.c.RemotePort = scanAddressNA(remote, &p.bytesRemote)
+	p.c.inode = parseDec(inode)
+	p.b = nextLine(b)
+	return &p.c
+}
+
+// scanAddressNA parses 'A12CF62E:00AA' to the address/port. Handles IPv4 and
+// IPv6 addresses. The address is a big endian 32 bit ints, hex encoded. We
+// just decode the hex and flip the bytes in every group of 4.
+func scanAddressNA(in []byte, buf *[16]byte) (net.IP, uint16) {
+	col := bytes.IndexByte(in, ':')
+	if col == -1 {
+		return nil, 0
+	}
+
+	// Network address is big endian. Can be either ipv4 or ipv6.
+	address := hexDecode32bigNA(in[:col], buf)
+	return net.IP(address), uint16(parseHex(in[col+1:]))
+}
+
+// hexDecode32big decodes sequences of 32bit big endian bytes.
+func hexDecode32bigNA(src []byte, buf *[16]byte) []byte {
+	blocks := len(src) / 8
+	for block := 0; block < blocks; block++ {
+		for i := 0; i < 4; i++ {
+			a := fromHexChar(src[block*8+i*2])
+			b := fromHexChar(src[block*8+i*2+1])
+			buf[block*4+3-i] = (a << 4) | b
+		}
+	}
+	return buf[:blocks*4]
+}
+
+func nextField(s []byte) ([]byte, []byte) {
+	// Skip whitespace.
+	for i, b := range s {
+		if b != ' ' {
+			s = s[i:]
+			break
+		}
+	}
+
+	// Up until the next whitespace field.
+	for i, b := range s {
+		if b == ' ' {
+			return s[:i], s[i:]
+		}
+	}
+
+	return nil, nil
+}
+
+func nextLine(s []byte) []byte {
+	i := bytes.IndexByte(s, '\n')
+	if i == -1 {
+		return nil
+	}
+	return s[i+1:]
+}
+
+// Simplified copy of strconv.ParseUint(16).
+func parseHex(s []byte) uint {
+	n := uint(0)
+	for i := 0; i < len(s); i++ {
+		n *= 16
+		n += uint(fromHexChar(s[i]))
+	}
+	return n
+}
+
+// Simplified copy of strconv.ParseUint(10).
+func parseDec(s []byte) uint64 {
+	n := uint64(0)
+	for _, c := range s {
+		n *= 10
+		n += uint64(c - '0')
+	}
+	return n
+}
+
+// hexDecode32big decodes sequences of 32bit big endian bytes.
+func hexDecode32big(src []byte) []byte {
+	dst := make([]byte, len(src)/2)
+	blocks := len(src) / 8
+	for block := 0; block < blocks; block++ {
+		for i := 0; i < 4; i++ {
+			a := fromHexChar(src[block*8+i*2])
+			b := fromHexChar(src[block*8+i*2+1])
+			dst[block*4+3-i] = (a << 4) | b
+		}
+	}
+	return dst
+}
+
+// fromHexChar converts a hex character into its value.
+func fromHexChar(c byte) uint8 {
+	switch {
+	case '0' <= c && c <= '9':
+		return c - '0'
+	case 'a' <= c && c <= 'f':
+		return c - 'a' + 10
+	case 'A' <= c && c <= 'F':
+		return c - 'A' + 10
+	}
+	return 0
+}

probe/proc/reader.go

@@ -0,0 +1,165 @@
+package proc
+
+import (
+	"bytes"
+	"os"
+	"path"
+	"sync"
+)
+
+// File is a file in the "/proc" directory
+type File interface {
+	ReadInto(buf *bytes.Buffer) error
+	Close() error
+}
+
+// OSFile is a native file
+type OSFile struct{ *os.File }
+
+// ReadInto reads the whole file into a buffer
+func (of *OSFile) ReadInto(buf *bytes.Buffer) error {
+	if _, err := of.File.Seek(0, 0); err != nil {
+		return err
+	}
+	_, err := buf.ReadFrom(of.File)
+	return err
+}
+
+// Dir is the '/proc' directory and the associated ops for
+// reading subdirs or files.
+type Dir interface {
+	Root() string                          // the "/proc" directory
+	Open(s string) (File, error)           // open a file in the "/proc" dir
+	ReadDirNames(string) ([]string, error) // list a subdirectory in the "/proc"
+}
+
+// OSDir is a OS "/proc" firectory
+type OSDir struct{ Dir string }
+
+// Root returns the "/proc" top dir
+func (dp OSDir) Root() string {
+	return dp.Dir
+}
+
+// Open returns a ProcFile
+func (dp OSDir) Open(s string) (File, error) {
+	h, err := os.Open(path.Join(dp.Root(), s))
+	if err != nil {
+		return nil, err
+	}
+	return &OSFile{h}, nil
+}
+
+// ReadDirNames reads all the directory entries
+func (dp OSDir) ReadDirNames(s string) ([]string, error) {
+	f, err := os.Open(s)
+	if err != nil {
+		return nil, err
+	}
+	list, err := f.Readdirnames(-1)
+	f.Close()
+	if err != nil {
+		return nil, err
+	}
+	return list, nil
+}
+
+// DefaultProcDir is the default '/proc' directory
+var DefaultProcDir = OSDir{Dir: "/proc"}
+
+// Process represents a single process.
+type Process struct {
+	PID, PPID int
+	Comm      string
+	Cmdline   string
+	Threads   int
+	Inodes    []uint64
+}
+
+// Copy returns a copy of a process
+func (p Process) Copy() Process {
+	dup := make([]uint64, len(p.Inodes))
+	copy(dup, p.Inodes)
+	p.Inodes = dup
+	return p
+}
+
+// Reader is something that reads the /proc directory and
+// returns some info like processes and connections
+type Reader interface {
+	// Processes walks through the processes
+	Processes(func(Process)) error
+	// Connections walks through the connections
+	Connections(bool, func(Connection)) error
+	// Close closes the "/proc" reader
+	Close() error
+}
+
+// CachingProcReader is a '/proc' reader than caches a copy of the output from another
+// '/proc' reader, and then allows other concurrent readers to Walk that copy.
+type CachingProcReader struct {
+	procsCache   []Process
+	connsCache   []Connection
+	source       Reader
+	includeProcs bool
+	sync.RWMutex
+}
+
+// NewCachingProcReader returns a new CachingProcReader
+func NewCachingProcReader(source Reader, includeProcs bool) *CachingProcReader {
+	return &CachingProcReader{source: source, includeProcs: includeProcs}
+}
+
+// Processes walks a cached copy of process list
+func (c *CachingProcReader) Processes(f func(Process)) error {
+	c.RLock()
+	defer c.RUnlock()
+
+	for _, p := range c.procsCache {
+		f(p)
+	}
+	return nil
+}
+
+// Connections walks a cached copy of the connections list
+// Note: specifying 'includeProcs' has no effect here, as the cached copy
+func (c *CachingProcReader) Connections(_ bool, f func(Connection)) error {
+	c.RLock()
+	defer c.RUnlock()
+
+	for _, c := range c.connsCache {
+		f(c)
+	}
+	return nil
+
+}
+
+// Close closes the "/proc" reader
+func (c *CachingProcReader) Close() error {
+	return c.source.Close()
+}
+
+// Tick updates the cached copy of the processes and connections lists
+func (c *CachingProcReader) Tick() error {
+	newProcsCache := []Process{}
+	newConnsCache := []Connection{}
+
+	if err := c.source.Processes(func(p Process) {
+		newProcsCache = append(newProcsCache, p)
+	}); err != nil {
+		return err
+	}
+
+	if err := c.source.Connections(c.includeProcs, func(conn Connection) {
+		newConnsCache = append(newConnsCache, conn.Copy())
+	}); err != nil {
+		return err
+	}
+
+	c.Lock()
+	defer c.Unlock()
+	c.procsCache = newProcsCache
+	c.connsCache = newConnsCache
+
+	return nil
+}

probe/proc/reader_darwin.go

@@ -1,27 +1,28 @@
-package process
+package proc
 
 import (
 	"fmt"
+	"net"
 	"os/exec"
 	"strconv"
 	"strings"
 )
 
-// NewWalker returns a Darwin (lsof-based) walker.
-func NewWalker(_ string) Walker {
-	return &walker{}
+type reader struct{}
+
+// NewReader returns a Darwin (lsof-based) '/proc' reader
+func NewReader(proc Dir) Reader {
+	return &reader{}
 }
 
-type walker struct{}
-
 const (
-	lsofBinary = "lsof"
-	lsofFields = "cn" // parseLSOF() depends on the order
+	lsofBinary    = "lsof"
+	lsofFields    = "cn" // parseLSOF() depends on the order
+	netstatBinary = "netstat"
+	lsofBinary    = "lsof"
 )
 
-// These functions copied from procspy.
-
-func (walker) Walk(f func(Process)) error {
+func (reader) Processes(f func(Process)) error {
 	output, err := exec.Command(
 		lsofBinary,
 		"-i",       // only Internet files
@@ -44,6 +45,59 @@ func (walker) Walk(f func(Process)) error {
 	return nil
 }
 
+func (r *reader) Connections(withProcs bool, f func(Connection)) error {
+	out, err := exec.Command(
+		netstatBinary,
+		"-n", // no number resolving
+		"-W", // Wide output
+		// "-l", // full IPv6 addresses // What does this do?
+		"-p", "tcp", // only TCP
+	).CombinedOutput()
+	if err != nil {
+		return err
+	}
+	connections := parseDarwinNetstat(string(out))
+
+	if withProcs {
+		out, err := exec.Command(
+			lsofBinary,
+			"-i",       // only Internet files
+			"-n", "-P", // no number resolving
+			"-w",             // no warnings
+			"-F", lsofFields, // \n based output of only the fields we want.
+		).CombinedOutput()
+		if err != nil {
+			return err
+		}
+
+		procs, err := parseLSOF(string(out))
+		if err != nil {
+			return err
+		}
+		for local, proc := range procs {
+			for i, c := range connections {
+				localAddr := net.JoinHostPort(
+					c.LocalAddress.String(),
+					strconv.Itoa(int(c.LocalPort)),
+				)
+				if localAddr == local {
+					connections[i].Proc = proc
+				}
+			}
+		}
+	}
+
+	for _, c := range connections {
+		f(c)
+	}
+	return nil
+}
+
+// Close closes the Darwin "/proc" reader
+func (reader) Close() error {
+	return nil
+}
+
 func parseLSOF(output string) (map[string]Process, error) {
 	var (
 		processes = map[string]Process{} // Local addr -> Proc

probe/proc/reader_linux.go

@@ -0,0 +1,202 @@
+package proc
+
+import (
+	"bytes"
+	"path"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/bluele/gcache"
+)
+
+const (
+	filesCacheLen        = 512
+	filesCacheExpiration = 60 * time.Second
+)
+
+var tcpFiles = []string{
+	"net/tcp",
+	"net/tcp6",
+}
+
+// A cache for files handles
+type filesCache struct {
+	handles gcache.Cache
+}
+
+type filesCacheEntry struct {
+	sync.RWMutex
+	File
+}
+
+func newFilesCache(proc Dir) *filesCache {
+	loadFunc := func(fileName interface{}) (interface{}, error) {
+		f, err := proc.Open(fileName.(string))
+		if err != nil {
+			return nil, err
+		}
+		return filesCacheEntry{File: f}, nil
+	}
+	evictionFunc := func(key, value interface{}) {
+		value.(filesCacheEntry).Close()
+	}
+	return &filesCache{
+		handles: gcache.New(filesCacheLen).LoaderFunc(loadFunc).EvictedFunc(evictionFunc).Expiration(filesCacheExpiration).ARC().Build(),
+	}
+}
+
+// Read a "/proc" file, identified as a file in a subdir (eg "1134/comm"), into a buffer
+func (fc *filesCache) ReadInto(filename string, buf *bytes.Buffer) error {
+	// we could use a lock here, but this is only used from Processes()/Connections(),
+	// and they are always invoked sequentially...
+	h, err := fc.handles.Get(filename)
+	if err != nil {
+		return err
+	}
+	handle := h.(filesCacheEntry)
+	handle.Lock()
+	defer handle.Unlock()
+	return handle.ReadInto(buf)
+}
+
+// Close closes all the handles in the cache
+func (fc *filesCache) Close() error {
+	for _, key := range fc.handles.Keys() {
+		fc.handles.Remove(key)
+	}
+	return nil
+}
+
+// the Linux "/proc" reader
+type reader struct {
+	proc    Dir
+	handles *filesCache
+}
+
+// NewReader creates a new /proc reader.
+func NewReader(proc Dir) Reader {
+	return &reader{
+		proc:    proc,
+		handles: newFilesCache(proc),
+	}
+}
+
+// Close closes the Linux "/proc" reader
+func (r *reader) Close() error {
+	return r.handles.Close()
+}
+
+// Processes walks the /proc directory and marshalls the files into
+// instances of Process, which it then passes one-by-one to the
+// supplied function. Processes() is only made public so that is
+// can be tested.
+func (r *reader) Processes(f func(Process)) error {
+	dirEntries, err := r.proc.ReadDirNames(r.proc.Root())
+	if err != nil {
+		return err
+	}
+
+	var fdStat syscall.Stat_t
+	buf := bytes.Buffer{}
+
+	for _, subdir := range dirEntries {
+		readIntoBuffer := func(filename string) error {
+			buf.Reset()
+			return r.handles.ReadInto(path.Join(subdir, filename), &buf)
+		}
+
+		pid, err := strconv.Atoi(subdir)
+		if err != nil {
+			continue
+		}
+
+		if readIntoBuffer("stat") != nil {
+			continue
+		}
+		splits := strings.Fields(buf.String())
+		ppid, err := strconv.Atoi(splits[3])
+		if err != nil {
+			return err
+		}
+
+		threads, err := strconv.Atoi(splits[19])
+		if err != nil {
+			return err
+		}
+
+		cmdline := ""
+		if readIntoBuffer("cmdline") == nil {
+			cmdlineBuf := bytes.Replace(buf.Bytes(), []byte{'\000'}, []byte{' '}, -1)
+			cmdline = string(cmdlineBuf)
+		}
+
+		comm := "(unknown)"
+		if readIntoBuffer("comm") == nil {
+			comm = strings.TrimSpace(buf.String())
+		}
+
+		fdBase := path.Join(r.proc.Root(), strconv.Itoa(pid), "fd")
+		fdNames, err := r.proc.ReadDirNames(fdBase)
+		if err != nil {
+			return err
+		}
+
+		inodes := []uint64{}
+		for _, fdName := range fdNames {
+			// Direct use of syscall.Stat() to save garbage.
+			fdPath := path.Join(fdBase, fdName)
+			err = syscall.Stat(fdPath, &fdStat)
+			if err == nil && (fdStat.Mode&syscall.S_IFMT == syscall.S_IFSOCK) { // We want sockets only.
+				inodes = append(inodes, fdStat.Ino)
+			}
+		}
+
+		f(Process{
+			PID:     pid,
+			PPID:    ppid,
+			Comm:    comm,
+			Cmdline: cmdline,
+			Threads: threads,
+			Inodes:  inodes,
+		})
+	}
+
+	return nil
+}
+
+// Connections walks through all the connections in the "/proc"
+func (r *reader) Connections(withProcs bool, f func(Connection)) error {
+	// create a map of inode->Process
+	procs := make(map[uint64]Process)
+	if withProcs {
+		r.Processes(func(p Process) {
+			for _, inode := range p.Inodes {
+				procs[inode] = p
+			}
+		})
+	}
+
+	buf := bytes.Buffer{}
+	for _, tcpFile := range tcpFiles {
+		err := r.handles.ReadInto(tcpFile, &buf)
+		if err != nil {
+			return err
+		}
+	}
+
+	pn := newNetReader(buf.Bytes(), tcpEstablished)
+	for {
+		conn := pn.Next()
+		if conn == nil {
+			break // Done!
+		}
+		if proc, ok := procs[conn.inode]; ok {
+			conn.Process = proc
+		}
+		f(*conn)
+	}
+	return nil
+}

probe/proc/reader_linux_test.go

@@ -0,0 +1,93 @@
+package proc_test
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"reflect"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/weaveworks/scope/probe/proc"
+	"github.com/weaveworks/scope/test"
+)
+
+type mockedProcess struct {
+	ID, Comm, Cmdline string
+}
+
+func (p mockedProcess) Name() string       { return p.ID }
+func (p mockedProcess) Size() int64        { return 0 }
+func (p mockedProcess) Mode() os.FileMode  { return 0 }
+func (p mockedProcess) ModTime() time.Time { return time.Now() }
+func (p mockedProcess) IsDir() bool        { return true }
+func (p mockedProcess) Sys() interface{}   { return nil }
+
+func TestProcReaderProcesses(t *testing.T) {
+	processes := map[string]mockedProcess{
+		"3":       {ID: "3", Comm: "curl\n", Cmdline: "curl\000google.com"},
+		"2":       {ID: "2", Comm: "bash\n"},
+		"4":       {ID: "4", Comm: "apache\n"},
+		"notapid": {ID: "notapid"},
+		"1":       {ID: "1", Comm: "init\n"},
+	}
+
+	want := map[int]proc.Process{
+		3: {PID: 3, PPID: 2, Comm: "curl", Cmdline: "curl google.com", Threads: 1, Inodes: []uint64{}},
+		2: {PID: 2, PPID: 1, Comm: "bash", Cmdline: "", Threads: 1, Inodes: []uint64{}},
+		4: {PID: 4, PPID: 3, Comm: "apache", Cmdline: "", Threads: 1, Inodes: []uint64{}},
+		1: {PID: 1, PPID: 0, Comm: "init", Cmdline: "", Threads: 1, Inodes: []uint64{}},
+	}
+
+	// use a mocked /proc that reads from our mocked processes
+	procDir := mockedDir{
+		ReadDirNamesFunc: func(path string) ([]string, error) {
+			result := []string{}
+			for k := range processes {
+				result = append(result, k)
+			}
+			return result, nil
+		},
+		OpenFunc: func(filename string) (proc.File, error) {
+			splits := strings.Split(filename, "/")
+			pid := splits[len(splits)-2]
+			process, ok := processes[pid]
+			if !ok {
+				return nil, fmt.Errorf("not found")
+			}
+
+			file := splits[len(splits)-1]
+			var content []byte
+			switch file {
+			case "comm":
+				content = []byte(process.Comm)
+			case "stat":
+				pid, _ := strconv.Atoi(splits[len(splits)-2])
+				parent := pid - 1
+				content = []byte(fmt.Sprintf("%d na R %d 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1", pid, parent))
+			case "cmdline":
+				content = []byte(process.Cmdline)
+			default:
+				return nil, fmt.Errorf("not found")
+			}
+
+			return mockedFile{
+				ReadIntoFunc: func(buf *bytes.Buffer) error {
+					_, err := buf.Write(content)
+					return err
+				},
+			}, nil
+		},
+	}
+
+	procReader := proc.NewReader(procDir)
+	have := map[int]proc.Process{}
+	err := procReader.Processes(func(p proc.Process) {
+		have[p.PID] = p
+	})
+	if err != nil || !reflect.DeepEqual(want, have) {
+		t.Errorf("%v (%v)", test.Diff(want, have), err)
+	}
+}

probe/proc/reader_test.go

@@ -0,0 +1,69 @@
+package proc_test
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/weaveworks/scope/probe/proc"
+	"github.com/weaveworks/scope/test"
+)
+
+func TestProcReaderBasic(t *testing.T) {
+	procFunc := func(proc.Process) {}
+	nullProcDir := mockedDir{
+		Dir:              "",
+		OpenFunc:         func(string) (proc.File, error) { return &proc.OSFile{File: nil}, nil },
+		ReadDirNamesFunc: func(string) ([]string, error) { return []string{}, nil },
+	}
+
+	if err := proc.NewReader(nullProcDir).Processes(procFunc); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCachingProcReader(t *testing.T) {
+	all := func(w proc.Reader) ([]proc.Process, error) {
+		all := []proc.Process{}
+		err := w.Processes(func(p proc.Process) {
+			all = append(all, p)
+		})
+		return all, err
+	}
+
+	processes := []proc.Process{
+		{PID: 1, PPID: 0, Comm: "init"},
+		{PID: 2, PPID: 1, Comm: "bash"},
+		{PID: 3, PPID: 1, Comm: "apache", Threads: 2},
+		{PID: 4, PPID: 2, Comm: "ping", Cmdline: "ping foo.bar.local"},
+	}
+	procReader := &proc.MockedReader{
+		Procs: processes,
+	}
+	cachingProcReader := proc.NewCachingProcReader(procReader, true)
+	err := cachingProcReader.Tick()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	have, err := all(cachingProcReader)
+	if err != nil || !reflect.DeepEqual(processes, have) {
+		t.Errorf("%v (%v)", test.Diff(processes, have), err)
+	}
+
+	procReader.Procs = []proc.Process{}
+	have, err = all(cachingProcReader)
+	if err != nil || !reflect.DeepEqual(processes, have) {
+		t.Errorf("%v (%v)", test.Diff(processes, have), err)
+	}
+
+	err = cachingProcReader.Tick()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	have, err = all(cachingProcReader)
+	want := []proc.Process{}
+	if err != nil || !reflect.DeepEqual(want, have) {
+		t.Errorf("%v (%v)", test.Diff(want, have), err)
+	}
+}

probe/proc/tree.go

@@ -1,4 +1,4 @@
-package process
+package proc
 
 import (
 	"fmt"
@@ -14,9 +14,9 @@ type tree struct {
 }
 
 // NewTree returns a new Tree that can be polled.
-func NewTree(walker Walker) (Tree, error) {
+func NewTree(walker Reader) (Tree, error) {
 	pt := tree{processes: map[int]Process{}}
-	err := walker.Walk(func(p Process) {
+	err := walker.Processes(func(p Process) {
 		pt.processes[p.PID] = p
 	})
 

probe/proc/tree_test.go

@@ -1,15 +1,15 @@
-package process_test
+package proc_test
 
 import (
 	"reflect"
 	"testing"
 
-	"github.com/weaveworks/scope/probe/process"
+	"github.com/weaveworks/scope/probe/proc"
 )
 
 func TestTree(t *testing.T) {
-	walker := &mockWalker{
-		processes: []process.Process{
+	walker := &proc.MockedReader{
+		Procs: []proc.Process{
 			{PID: 1, PPID: 0, Comm: "init"},
 			{PID: 2, PPID: 1, Comm: "bash"},
 			{PID: 3, PPID: 1, Comm: "apache", Threads: 2},
@@ -17,7 +17,7 @@ func TestTree(t *testing.T) {
 		},
 	}
 
-	tree, err := process.NewTree(walker)
+	tree, err := proc.NewTree(walker)
 	if err != nil {
 		t.Fatalf("newProcessTree error: %v", err)
 	}

probe/process/reporter.go

@@ -3,6 +3,7 @@ package process
 import (
 	"strconv"
 
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -18,11 +19,11 @@ const (
 // Reporter generates Reports containing the Process topology.
 type Reporter struct {
 	scope  string
-	walker Walker
+	walker proc.Reader
 }
 
 // NewReporter makes a new Reporter.
-func NewReporter(walker Walker, scope string) *Reporter {
+func NewReporter(walker proc.Reader, scope string) *Reporter {
 	return &Reporter{
 		scope:  scope,
 		walker: walker,
@@ -42,7 +43,7 @@ func (r *Reporter) Report() (report.Report, error) {
 
 func (r *Reporter) processTopology() (report.Topology, error) {
 	t := report.MakeTopology()
-	err := r.walker.Walk(func(p Process) {
+	err := r.walker.Processes(func(p proc.Process) {
 		pidstr := strconv.Itoa(p.PID)
 		nodeID := report.MakeProcessNodeID(r.scope, pidstr)
 		t.Nodes[nodeID] = report.MakeNode()

probe/process/reporter_test.go

@@ -4,25 +4,15 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
 )
 
-type mockWalker struct {
-	processes []process.Process
-}
-
-func (m *mockWalker) Walk(f func(process.Process)) error {
-	for _, p := range m.processes {
-		f(p)
-	}
-	return nil
-}
-
 func TestReporter(t *testing.T) {
-	walker := &mockWalker{
-		processes: []process.Process{
+	procReader := &proc.MockedReader{
+		Procs: []proc.Process{
 			{PID: 1, PPID: 0, Comm: "init"},
 			{PID: 2, PPID: 1, Comm: "bash"},
 			{PID: 3, PPID: 1, Comm: "apache", Threads: 2},
@@ -31,7 +21,7 @@ func TestReporter(t *testing.T) {
 		},
 	}
 
-	reporter := process.NewReporter(walker, "")
+	reporter := process.NewReporter(procReader, "")
 	want := report.MakeReport()
 	want.Process = report.Topology{
 		Nodes: report.Nodes{

probe/process/walker.go

@@ -1,56 +0,0 @@
-package process
-
-import "sync"
-
-// Process represents a single process.
-type Process struct {
-	PID, PPID int
-	Comm      string
-	Cmdline   string
-	Threads   int
-}
-
-// Walker is something that walks the /proc directory
-type Walker interface {
-	Walk(func(Process)) error
-}
-
-// CachingWalker is a walker than caches a copy of the output from another
-// Walker, and then allows other concurrent readers to Walk that copy.
-type CachingWalker struct {
-	cache     []Process
-	cacheLock sync.RWMutex
-	source    Walker
-}
-
-// NewCachingWalker returns a new CachingWalker
-func NewCachingWalker(source Walker) *CachingWalker {
-	return &CachingWalker{source: source}
-}
-
-// Walk walks a cached copy of process list
-func (c *CachingWalker) Walk(f func(Process)) error {
-	c.cacheLock.RLock()
-	defer c.cacheLock.RUnlock()
-
-	for _, p := range c.cache {
-		f(p)
-	}
-	return nil
-}
-
-// Tick updates cached copy of process list
-func (c *CachingWalker) Tick() error {
-	newCache := []Process{}
-	err := c.source.Walk(func(p Process) {
-		newCache = append(newCache, p)
-	})
-	if err != nil {
-		return err
-	}
-
-	c.cacheLock.Lock()
-	defer c.cacheLock.Unlock()
-	c.cache = newCache
-	return nil
-}

probe/process/walker_linux.go

@@ -1,79 +0,0 @@
-package process
-
-import (
-	"bytes"
-	"io/ioutil"
-	"path"
-	"strconv"
-	"strings"
-)
-
-// Hooks exposed for mocking
-var (
-	ReadDir  = ioutil.ReadDir
-	ReadFile = ioutil.ReadFile
-)
-
-type walker struct {
-	procRoot string
-}
-
-// NewWalker creates a new process Walker.
-func NewWalker(procRoot string) Walker {
-	return &walker{procRoot: procRoot}
-}
-
-// Walk walks the supplied directory (expecting it to look like /proc)
-// and marshalls the files into instances of Process, which it then
-// passes one-by-one to the supplied function. Walk is only made public
-// so that is can be tested.
-func (w *walker) Walk(f func(Process)) error {
-	dirEntries, err := ReadDir(w.procRoot)
-	if err != nil {
-		return err
-	}
-
-	for _, dirEntry := range dirEntries {
-		filename := dirEntry.Name()
-		pid, err := strconv.Atoi(filename)
-		if err != nil {
-			continue
-		}
-
-		stat, err := ReadFile(path.Join(w.procRoot, filename, "stat"))
-		if err != nil {
-			continue
-		}
-		splits := strings.Fields(string(stat))
-		ppid, err := strconv.Atoi(splits[3])
-		if err != nil {
-			return err
-		}
-
-		threads, err := strconv.Atoi(splits[19])
-		if err != nil {
-			return err
-		}
-
-		cmdline := ""
-		if cmdlineBuf, err := ReadFile(path.Join(w.procRoot, filename, "cmdline")); err == nil {
-			cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1)
-			cmdline = string(cmdlineBuf)
-		}
-
-		comm := "(unknown)"
-		if commBuf, err := ReadFile(path.Join(w.procRoot, filename, "comm")); err == nil {
-			comm = strings.TrimSpace(string(commBuf))
-		}
-
-		f(Process{
-			PID:     pid,
-			PPID:    ppid,
-			Comm:    comm,
-			Cmdline: cmdline,
-			Threads: threads,
-		})
-	}
-
-	return nil
-}

probe/process/walker_linux_test.go

@@ -1,90 +0,0 @@
-package process_test
-
-import (
-	"fmt"
-	"os"
-	"reflect"
-	"strconv"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/weaveworks/scope/probe/process"
-	"github.com/weaveworks/scope/test"
-)
-
-type mockProcess struct {
-	name, comm, cmdline string
-}
-
-func (p mockProcess) Name() string       { return p.name }
-func (p mockProcess) Size() int64        { return 0 }
-func (p mockProcess) Mode() os.FileMode  { return 0 }
-func (p mockProcess) ModTime() time.Time { return time.Now() }
-func (p mockProcess) IsDir() bool        { return true }
-func (p mockProcess) Sys() interface{}   { return nil }
-
-func TestWalker(t *testing.T) {
-	oldReadDir, oldReadFile := process.ReadDir, process.ReadFile
-	defer func() {
-		process.ReadDir = oldReadDir
-		process.ReadFile = oldReadFile
-	}()
-
-	processes := map[string]mockProcess{
-		"3":       {name: "3", comm: "curl\n", cmdline: "curl\000google.com"},
-		"2":       {name: "2", comm: "bash\n"},
-		"4":       {name: "4", comm: "apache\n"},
-		"notapid": {name: "notapid"},
-		"1":       {name: "1", comm: "init\n"},
-	}
-
-	process.ReadDir = func(path string) ([]os.FileInfo, error) {
-		result := []os.FileInfo{}
-		for _, p := range processes {
-			result = append(result, p)
-		}
-		return result, nil
-	}
-
-	process.ReadFile = func(path string) ([]byte, error) {
-		splits := strings.Split(path, "/")
-
-		pid := splits[len(splits)-2]
-		process, ok := processes[pid]
-		if !ok {
-			return nil, fmt.Errorf("not found")
-		}
-
-		file := splits[len(splits)-1]
-		switch file {
-		case "comm":
-			return []byte(process.comm), nil
-		case "stat":
-			pid, _ := strconv.Atoi(splits[len(splits)-2])
-			parent := pid - 1
-			return []byte(fmt.Sprintf("%d na R %d 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1", pid, parent)), nil
-		case "cmdline":
-			return []byte(process.cmdline), nil
-		}
-
-		return nil, fmt.Errorf("not found")
-	}
-
-	want := map[int]process.Process{
-		3: {PID: 3, PPID: 2, Comm: "curl", Cmdline: "curl google.com", Threads: 1},
-		2: {PID: 2, PPID: 1, Comm: "bash", Cmdline: "", Threads: 1},
-		4: {PID: 4, PPID: 3, Comm: "apache", Cmdline: "", Threads: 1},
-		1: {PID: 1, PPID: 0, Comm: "init", Cmdline: "", Threads: 1},
-	}
-
-	have := map[int]process.Process{}
-	walker := process.NewWalker("unused")
-	err := walker.Walk(func(p process.Process) {
-		have[p.PID] = p
-	})
-
-	if err != nil || !reflect.DeepEqual(want, have) {
-		t.Errorf("%v (%v)", test.Diff(want, have), err)
-	}
-}

probe/process/walker_test.go

@@ -1,66 +0,0 @@
-package process_test
-
-import (
-	"reflect"
-	"testing"
-
-	"github.com/weaveworks/scope/probe/process"
-	"github.com/weaveworks/scope/test"
-)
-
-func TestBasicWalk(t *testing.T) {
-	var (
-		procRoot = "/proc"
-		procFunc = func(process.Process) {}
-	)
-	if err := process.NewWalker(procRoot).Walk(procFunc); err != nil {
-		t.Fatal(err)
-	}
-}
-
-func TestCache(t *testing.T) {
-	processes := []process.Process{
-		{PID: 1, PPID: 0, Comm: "init"},
-		{PID: 2, PPID: 1, Comm: "bash"},
-		{PID: 3, PPID: 1, Comm: "apache", Threads: 2},
-		{PID: 4, PPID: 2, Comm: "ping", Cmdline: "ping foo.bar.local"},
-	}
-	walker := &mockWalker{
-		processes: processes,
-	}
-	cachingWalker := process.NewCachingWalker(walker)
-	err := cachingWalker.Tick()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	have, err := all(cachingWalker)
-	if err != nil || !reflect.DeepEqual(processes, have) {
-		t.Errorf("%v (%v)", test.Diff(processes, have), err)
-	}
-
-	walker.processes = []process.Process{}
-	have, err = all(cachingWalker)
-	if err != nil || !reflect.DeepEqual(processes, have) {
-		t.Errorf("%v (%v)", test.Diff(processes, have), err)
-	}
-
-	err = cachingWalker.Tick()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	have, err = all(cachingWalker)
-	want := []process.Process{}
-	if err != nil || !reflect.DeepEqual(want, have) {
-		t.Errorf("%v (%v)", test.Diff(want, have), err)
-	}
-}
-
-func all(w process.Walker) ([]process.Process, error) {
-	all := []process.Process{}
-	err := w.Walk(func(p process.Process) {
-		all = append(all, p)
-	})
-	return all, err
-}