Cache processes and connections when reading from the /proc

Code cleanups

probe/docker/tagger.go

@@ -24,11 +24,11 @@ var (
 // nodes that have a PID.
 type Tagger struct {
 	registry   Registry
-	procWalker proc.Walker
+	procWalker proc.ProcReader
 }
 
 // NewTagger returns a usable Tagger.
-func NewTagger(registry Registry, procWalker proc.Walker) *Tagger {
+func NewTagger(registry Registry, procWalker proc.ProcReader) *Tagger {
 	return &Tagger{
 		registry:   registry,
 		procWalker: procWalker,

probe/docker/tagger_test.go

@@ -27,7 +27,7 @@ func TestTagger(t *testing.T) {
 	oldProcessTree := docker.NewProcessTreeStub
 	defer func() { docker.NewProcessTreeStub = oldProcessTree }()
 
-	docker.NewProcessTreeStub = func(_ proc.Walker) (proc.Tree, error) {
+	docker.NewProcessTreeStub = func(_ proc.ProcReader) (proc.Tree, error) {
 		return &mockProcessTree{map[int]int{2: 1}}, nil
 	}
 

probe/endpoint/reporter.go

@@ -28,6 +28,7 @@ type Reporter struct {
 	conntracker      *Conntracker
 	natmapper        *natmapper
 	revResolver      *ReverseResolver
+	procReader       proc.ProcReader
 }
 
 // SpyDuration is an exported prometheus metric
@@ -47,7 +48,7 @@ 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, useConntrack bool) *Reporter {
+func NewReporter(hostID, hostName string, includeProcesses bool, procReader proc.ProcReader, 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 := proc.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/scope/probe/proc"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/endpoint"
+	"github.com/weaveworks/scope/probe/proc"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -17,7 +17,7 @@ var (
 	fixRemoteAddress = net.ParseIP("192.168.1.2")
 	fixRemotePort    = uint16(12345)
 	fixRemotePortB   = uint16(12346)
-	fixProcessPID    = uint(4242)
+	fixProcessPID = int(4242)
 	fixProcessName   = "nginx"
 
 	fixConnections = []proc.Connection{
@@ -44,9 +44,9 @@ var (
 			LocalPort:     fixLocalPort,
 			RemoteAddress: fixRemoteAddress,
 			RemotePort:    fixRemotePort,
-			Proc: proc.Proc{
+			Process: proc.Process{
 				PID:  fixProcessPID,
-				Name: fixProcessName,
+				Comm: fixProcessName,
 			},
 		},
 		{
@@ -55,23 +55,22 @@ var (
 			LocalPort:     fixLocalPort,
 			RemoteAddress: fixRemoteAddress,
 			RemotePort:    fixRemotePort,
-			Proc: proc.Proc{
+			Process: proc.Process{
 				PID:  fixProcessPID,
-				Name: fixProcessName,
+				Comm: fixProcessName,
 			},
 		},
 	}
 )
 
 func TestSpyNoProcesses(t *testing.T) {
-	proc.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.MockedProcReader{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) {
-	proc.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.MockedProcReader{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/scope/probe/proc"
 	"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, ", "))
 
-	proc.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,13 +109,14 @@ func main() {
 	}
 
 	var (
-		endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack)
-		processCache     = proc.NewCachingWalker(proc.NewWalker(*procRoot))
-		tickers          = []Ticker{processCache}
+		procDir          = proc.OSProcDir{Dir: *procRoot}
+		procReader       = proc.NewCachingProcReader(proc.NewProcReader(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)}
 	)
@@ -134,7 +133,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/fixture.go

@@ -1,26 +0,0 @@
-package proc
-
-// SetFixtures declares constant Connection and ConnectionProcs which will
-// always be returned by the package-level Connections and Processes
-// functions. It's designed to be used in tests.
-
-type fixedConnIter []Connection
-
-func (f *fixedConnIter) Next() *Connection {
-	if len(*f) == 0 {
-		return nil
-	}
-
-	car := (*f)[0]
-	*f = (*f)[1:]
-
-	return &car
-}
-
-// SetFixtures is used in test scenarios to have known output.
-func SetFixtures(c []Connection) {
-	cbConnections = func(bool) (ConnIter, error) {
-		f := fixedConnIter(c)
-		return &f, nil
-	}
-}

probe/proc/mocks.go

@@ -0,0 +1,59 @@
+package proc
+
+import (
+	"bytes"
+	"os"
+	"time"
+)
+
+// a mocked process
+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 }
+
+// a mocked "/proc" directory
+type MockedProcDir struct {
+	Dir              string
+	ReadDirFunc      func(string) ([]os.FileInfo, error)
+	ReadFileFunc     func(string) ([]byte, error)
+	ReadFileIntoFunc func(string, *bytes.Buffer) error
+}
+
+func (p MockedProcDir) Root() string                                 { return p.Dir }
+func (p MockedProcDir) ReadDir(s string) ([]os.FileInfo, error)      { return p.ReadDirFunc(s) }
+func (p MockedProcDir) ReadFile(s string) ([]byte, error)            { return p.ReadFileFunc(s) }
+func (p MockedProcDir) ReadFileInto(s string, b *bytes.Buffer) error { return p.ReadFileIntoFunc(s, b) }
+
+var EmptyProcDir = MockedProcDir{
+	Dir:              "",
+	ReadDirFunc:      func(string) ([]os.FileInfo, error) { return []os.FileInfo{}, nil },
+	ReadFileFunc:     func(string) ([]byte, error) { return []byte{}, nil },
+	ReadFileIntoFunc: func(string, *bytes.Buffer) error { return nil },
+}
+
+// a mocked /proc reader
+type MockedProcReader struct {
+	Procs []Process
+	Conns []Connection
+}
+
+func (mw MockedProcReader) Processes(f func(Process)) error {
+	for _, p := range mw.Procs {
+		f(p)
+	}
+	return nil
+}
+
+func (mw *MockedProcReader) Connections(_ bool, f func(Connection)) error {
+	for _, c := range mw.Conns {
+		f(c)
+	}
+	return nil
+}

probe/proc/net.go

@@ -0,0 +1,42 @@
+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)
+	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

@@ -25,7 +25,7 @@ func NewProcNet(b []byte, wantedState uint) *ProcNet {
 // 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 *ProcNet) Next() *Connection {
-	again:
+again:
 	if len(p.b) == 0 {
 		return nil
 	}
@@ -73,7 +73,7 @@ func scanAddressNA(in []byte, buf *[16]byte) (net.IP, uint16) {
 
 	// 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:]))
+	return net.IP(address), uint16(parseHex(in[col+1:]))
 }
 
 // hexDecode32big decodes sequences of 32bit big endian bytes.
@@ -81,12 +81,12 @@ 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
+			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]
+	return buf[:blocks*4]
 }
 
 func nextField(s []byte) ([]byte, []byte) {
@@ -113,7 +113,7 @@ func nextLine(s []byte) []byte {
 	if i == -1 {
 		return nil
 	}
-	return s[i + 1:]
+	return s[i+1:]
 }
 
 // Simplified copy of strconv.ParseUint(16).
@@ -138,13 +138,13 @@ func parseDec(s []byte) uint64 {
 
 // hexDecode32big decodes sequences of 32bit big endian bytes.
 func hexDecode32big(src []byte) []byte {
-	dst := make([]byte, len(src) / 2)
+	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
+			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

probe/proc/proc.go

@@ -1,125 +0,0 @@
-package proc
-
-// /proc-based implementation.
-
-import (
-	"bytes"
-	"os"
-	"strconv"
-	"syscall"
-	"path"
-)
-
-var (
-	procRoot = "/proc"
-)
-
-// SetProcRoot sets the location of the proc filesystem.
-func SetProcRoot(root string) {
-	procRoot = root
-}
-
-// walkProcPid walks over all numerical (PID) /proc entries, and sees if their
-// ./fd/* files are symlink to sockets. Returns a map from socket ID (inode)
-// to PID. Will return an error if /proc isn't there.
-func walkProcPid() (map[uint64]Proc, error) {
-	fh, err := os.Open(procRoot)
-	if err != nil {
-		return nil, err
-	}
-
-	dirNames, err := fh.Readdirnames(-1)
-	fh.Close()
-	if err != nil {
-		return nil, err
-	}
-
-	var (
-		res = map[uint64]Proc{}
-		stat syscall.Stat_t
-	)
-	for _, dirName := range dirNames {
-		pid, err := strconv.ParseUint(dirName, 10, 0)
-		if err != nil {
-			// Not a number, so not a PID subdir.
-			continue
-		}
-
-		fdBase := procRoot + "/" + dirName + "/fd/"
-		dfh, err := os.Open(fdBase)
-		if err != nil {
-			// Process is be gone by now, or we don't have access.
-			continue
-		}
-
-		fdNames, err := dfh.Readdirnames(-1)
-		dfh.Close()
-		if err != nil {
-			continue
-		}
-
-		var name string
-
-		for _, fdName := range fdNames {
-			// Direct use of syscall.Stat() to save garbage.
-			err = syscall.Stat(fdBase + fdName, &stat)
-			if err != nil {
-				continue
-			}
-
-			// We want sockets only.
-			if stat.Mode & syscall.S_IFMT != syscall.S_IFSOCK {
-				continue
-			}
-
-			if name == "" {
-				if name = procName(path.Join(procRoot, dirName)); name == "" {
-					// Process might be gone by now
-					break
-				}
-			}
-
-			res[stat.Ino] = Proc{
-				PID:  uint(pid),
-				Name: name,
-			}
-		}
-	}
-
-	return res, nil
-}
-
-// procName does a pid->name lookup.
-func procName(base string) string {
-	fh, err := os.Open(base + "/comm")
-	if err != nil {
-		return ""
-	}
-
-	name := make([]byte, 64)
-	l, err := fh.Read(name)
-	fh.Close()
-	if err != nil {
-		return ""
-	}
-
-	if l < 2 {
-		return ""
-	}
-
-	// drop trailing "\n"
-	return string(name[:l - 1])
-}
-
-// readFile reads an arbitrary file into a buffer. It's a variable so it can
-// be overwritten for benchmarks. That's bad practice and we should change it
-// to be a dependency.
-var readFile = func(filename string, buf *bytes.Buffer) error {
-	f, err := os.Open(filename)
-	if err != nil {
-		return err
-	}
-	_, err = buf.ReadFrom(f)
-	f.Close()
-	return err
-}

probe/proc/reader.go

@@ -0,0 +1,117 @@
+package proc
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"sync"
+)
+
+// ProcDir is the '/proc' directory and the associated ops for
+// reading subdirs or files.
+type ProcDir interface {
+	Root() string                             // proc directory
+	ReadDir(string) ([]os.FileInfo, error)    // read a subdirectory in the "root"
+	ReadFile(string) ([]byte, error)          // read a file in the "root"
+	ReadFileInto(string, *bytes.Buffer) error // read a file in the "root" in a buffer
+}
+
+type OSProcDir struct{ Dir string }
+
+func (dp OSProcDir) Root() string                            { return dp.Dir }
+func (dp OSProcDir) ReadDir(s string) ([]os.FileInfo, error) { return ioutil.ReadDir(s) }
+func (dp OSProcDir) ReadFile(s string) ([]byte, error)       { return ioutil.ReadFile(s) }
+func (dp OSProcDir) ReadFileInto(filename string, buf *bytes.Buffer) error {
+	f, err := os.Open(filename)
+	if err != nil {
+		return err
+	}
+	_, err = buf.ReadFrom(f)
+	f.Close()
+	return err
+}
+
+// DefaultProcDir is the default '/proc' directory
+var DefaultProcDir = OSProcDir{Dir: "/proc"}
+
+// Process represents a single process.
+type Process struct {
+	PID, PPID int
+	Comm      string
+	Cmdline   string
+	Threads   int
+	Inodes    []uint64
+}
+
+// ProcReader is something that reads the /proc directory and
+// returns some info like processes and connections
+type ProcReader interface {
+	// Processes walks through the processes
+	Processes(func(Process)) error
+	// Connections walks through the connections
+	Connections(bool, func(Connection)) 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       ProcReader
+	includeProcs bool
+	sync.RWMutex
+}
+
+// NewCachingProcReader returns a new CachingProcReader
+func NewCachingProcReader(source ProcReader, 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
+
+}
+
+// Update 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

@@ -2,26 +2,27 @@ package proc
 
 import (
 	"fmt"
+	"net"
 	"os/exec"
 	"strconv"
 	"strings"
 )
 
-// NewWalker returns a Darwin (lsof-based) walker.
-func NewWalker(_ string) Walker {
-	return &walker{}
+type procReader struct{}
+
+// NewProcReader returns a Darwin (lsof-based) '/proc' reader
+func NewProcReader(proc ProcDir) *procReader {
+	return &procReader{}
 }
 
-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 (procReader) Processes(f func(Process)) error {
 	output, err := exec.Command(
 		lsofBinary,
 		"-i",       // only Internet files
@@ -44,6 +45,54 @@ func (walker) Walk(f func(Process)) error {
 	return nil
 }
 
+func (w *walker) 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
+}
+
 func parseLSOF(output string) (map[string]Process, error) {
 	var (
 		processes = map[string]Process{} // Local addr -> Proc

probe/proc/reader_linux.go

@@ -0,0 +1,130 @@
+package proc
+
+import (
+	"bytes"
+	"path"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+)
+
+type procReader struct {
+	proc ProcDir
+}
+
+// NewProcReader creates a new /proc reader.
+func NewProcReader(proc ProcDir) *procReader {
+	return &procReader{proc}
+}
+
+// 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 (w *procReader) Processes(f func(Process)) error {
+	dirEntries, err := w.proc.ReadDir(w.proc.Root())
+	if err != nil {
+		return err
+	}
+
+	for _, dirEntry := range dirEntries {
+		filename := dirEntry.Name()
+		pid, err := strconv.Atoi(filename)
+		if err != nil {
+			continue
+		}
+
+		stat, err := w.proc.ReadFile(path.Join(w.proc.Root(), 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 := w.proc.ReadFile(path.Join(w.proc.Root(), filename, "cmdline")); err == nil {
+			cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1)
+			cmdline = string(cmdlineBuf)
+		}
+
+		comm := "(unknown)"
+		if commBuf, err := w.proc.ReadFile(path.Join(w.proc.Root(), filename, "comm")); err == nil {
+			comm = strings.TrimSpace(string(commBuf))
+		}
+
+		fdBase := path.Join(w.proc.Root(), strconv.Itoa(pid), "fd")
+		fdNames, err := w.proc.ReadDir(fdBase)
+		if err != nil {
+			return err
+		}
+
+		inodes := []uint64{}
+		for _, fdName := range fdNames {
+			var fdStat syscall.Stat_t
+			// Direct use of syscall.Stat() to save garbage.
+			fdPath := path.Join(fdBase, fdName.Name())
+			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
+}
+
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		return bytes.NewBuffer(make([]byte, 0, 5000))
+	},
+}
+
+func (w *procReader) Connections(withProcs bool, f func(Connection)) error {
+	// create a map of inode->Process
+	procs := make(map[uint64]Process)
+	if withProcs {
+		w.Processes(func(p Process) {
+			for _, inode := range p.Inodes {
+				procs[inode] = p
+			}
+		})
+	}
+
+	buf := bufPool.Get().(*bytes.Buffer)
+	buf.Reset()
+	defer bufPool.Put(buf)
+
+	w.proc.ReadFileInto(path.Join(w.proc.Root(), "net", "tcp"), buf)
+	w.proc.ReadFileInto(path.Join(w.proc.Root(), "net", "tcp6"), buf)
+
+	pn := NewProcNet(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,73 @@
+package proc_test
+
+import (
+	"fmt"
+	"os"
+	"reflect"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/weaveworks/scope/probe/proc"
+	"github.com/weaveworks/scope/test"
+)
+
+func TestProcReaderProcesses(t *testing.T) {
+	processes := map[string]proc.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 := proc.MockedProcDir{
+		ReadDirFunc: func(path string) ([]os.FileInfo, error) {
+			result := []os.FileInfo{}
+			for _, p := range processes {
+				result = append(result, p)
+			}
+			return result, nil
+		},
+
+		ReadFileFunc: 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")
+		},
+	}
+
+	procReader := proc.NewProcReader(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

@@ -8,70 +8,56 @@ import (
 	"github.com/weaveworks/scope/test"
 )
 
-type mockWalker struct {
-	processes []proc.Process
-}
-
-func (m *mockWalker) Walk(f func(proc.Process)) error {
-	for _, p := range m.processes {
-		f(p)
-	}
-	return nil
-}
-
-func TestBasicWalk(t *testing.T) {
-	var (
-		procRoot = "/proc"
-		procFunc = func(proc.Process) {}
-	)
-	if err := proc.NewWalker(procRoot).Walk(procFunc); err != nil {
+func TestProcReaderBasic(t *testing.T) {
+	procFunc := func(proc.Process) {}
+	if err := proc.NewProcReader(proc.EmptyProcDir).Processes(procFunc); err != nil {
 		t.Fatal(err)
 	}
 }
 
-func TestCache(t *testing.T) {
+func TestCachingProcReader(t *testing.T) {
+	all := func(w proc.ProcReader) ([]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"},
 	}
-	walker := &mockWalker{
-		processes: processes,
+	procReader := &proc.MockedProcReader{
+		Procs: processes,
 	}
-	cachingWalker := proc.NewCachingWalker(walker)
-	err := cachingWalker.Tick()
+	cachingProcReader := proc.NewCachingProcReader(procReader, true)
+	err := cachingProcReader.Tick()
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	have, err := all(cachingWalker)
+	have, err := all(cachingProcReader)
 	if err != nil || !reflect.DeepEqual(processes, have) {
 		t.Errorf("%v (%v)", test.Diff(processes, have), err)
 	}
 
-	walker.processes = []proc.Process{}
-	have, err = all(cachingWalker)
+	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 = cachingWalker.Tick()
+	err = cachingProcReader.Tick()
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	have, err = all(cachingWalker)
+	have, err = all(cachingProcReader)
 	want := []proc.Process{}
 	if err != nil || !reflect.DeepEqual(want, have) {
 		t.Errorf("%v (%v)", test.Diff(want, have), err)
 	}
 }
-
-func all(w proc.Walker) ([]proc.Process, error) {
-	all := []proc.Process{}
-	err := w.Walk(func(p proc.Process) {
-		all = append(all, p)
-	})
-	return all, err
-}

probe/proc/spy.go

@@ -1,40 +0,0 @@
-package proc
-
-import (
-	"net"
-)
-
-const (
-	tcpEstablished = 1 // according to /include/net/tcp_states.h
-)
-
-// Connection is a (TCP) connection. The Proc struct might not be filled in.
-type Connection struct {
-	Transport     string
-	LocalAddress  net.IP
-	LocalPort     uint16
-	RemoteAddress net.IP
-	RemotePort    uint16
-	inode         uint64
-	Proc
-}
-
-// Proc is a single process with PID and process name.
-type Proc struct {
-	PID  uint
-	Name string
-}
-
-// ConnIter is returned by Connections().
-type ConnIter interface {
-	Next() *Connection
-}
-
-// Connections returns all established (TCP) connections.  If processes is
-// false we'll just list all TCP connections, and there is no need to be root.
-// If processes is true it'll additionally try to lookup the process owning the
-// connection, filling in the Proc field. You will need to run this as root to
-// find all processes.
-func Connections(processes bool) (ConnIter, error) {
-	return cbConnections(processes)
-}

probe/proc/spy_linux.go

@@ -1,52 +0,0 @@
-package proc
-
-import (
-	"bytes"
-	"path"
-	"sync"
-)
-
-var bufPool = sync.Pool{
-	New: func() interface{} {
-		return bytes.NewBuffer(make([]byte, 0, 5000))
-	},
-}
-
-type pnConnIter struct {
-	pn    *ProcNet
-	buf   *bytes.Buffer
-	procs map[uint64]Proc
-}
-
-func (c *pnConnIter) Next() *Connection {
-	n := c.pn.Next()
-	if n == nil {
-		// Done!
-		bufPool.Put(c.buf)
-		return nil
-	}
-	if proc, ok := c.procs[n.inode]; ok {
-		n.Proc = proc
-	}
-	return n
-}
-
-// cbConnections sets Connections()
-var cbConnections = func(processes bool) (ConnIter, error) {
-	buf := bufPool.Get().(*bytes.Buffer)
-	buf.Reset()
-	readFile(path.Join(procRoot, "net", "tcp"), buf)
-	readFile(path.Join(procRoot, "net", "tcp6"), buf)
-	var procs map[uint64]Proc
-	if processes {
-		var err error
-		if procs, err = walkProcPid(); err != nil {
-			return nil, err
-		}
-	}
-	return &pnConnIter{
-		pn:    NewProcNet(buf.Bytes(), tcpEstablished),
-		buf:   buf,
-		procs: procs,
-	}, nil
-}

probe/proc/tree.go

@@ -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 ProcReader) (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

@@ -8,8 +8,8 @@ import (
 )
 
 func TestTree(t *testing.T) {
-	walker := &mockWalker{
-		processes: []proc.Process{
+	walker := &proc.MockedProcReader{
+		Procs: []proc.Process{
 			{PID: 1, PPID: 0, Comm: "init"},
 			{PID: 2, PPID: 1, Comm: "bash"},
 			{PID: 3, PPID: 1, Comm: "apache", Threads: 2},

probe/proc/walker.go

@@ -1,56 +0,0 @@
-package proc
-
-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/proc/walker_linux.go

@@ -1,79 +0,0 @@
-package proc
-
-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/proc/walker_linux_test.go

@@ -1,90 +0,0 @@
-package proc_test
-
-import (
-	"fmt"
-	"os"
-	"reflect"
-	"strconv"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/weaveworks/scope/probe/proc"
-	"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 := proc.ReadDir, proc.ReadFile
-	defer func() {
-		proc.ReadDir = oldReadDir
-		proc.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"},
-	}
-
-	proc.ReadDir = func(path string) ([]os.FileInfo, error) {
-		result := []os.FileInfo{}
-		for _, p := range processes {
-			result = append(result, p)
-		}
-		return result, nil
-	}
-
-	proc.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]proc.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]proc.Process{}
-	walker := proc.NewWalker("unused")
-	err := walker.Walk(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/process/reporter.go

@@ -19,11 +19,11 @@ const (
 // Reporter generates Reports containing the Process topology.
 type Reporter struct {
 	scope  string
-	walker proc.Walker
+	walker proc.ProcReader
 }
 
 // NewReporter makes a new Reporter.
-func NewReporter(walker proc.Walker, scope string) *Reporter {
+func NewReporter(walker proc.ProcReader, scope string) *Reporter {
 	return &Reporter{
 		scope:  scope,
 		walker: walker,
@@ -43,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 proc.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

@@ -10,20 +10,9 @@ import (
 	"github.com/weaveworks/scope/test"
 )
 
-type mockWalker struct {
-	processes []proc.Process
-}
-
-func (m *mockWalker) Walk(f func(proc.Process)) error {
-	for _, p := range m.processes {
-		f(p)
-	}
-	return nil
-}
-
 func TestReporter(t *testing.T) {
-	walker := &mockWalker{
-		processes: []proc.Process{
+	procReader := &proc.MockedProcReader{
+		Procs: []proc.Process{
 			{PID: 1, PPID: 0, Comm: "init"},
 			{PID: 2, PPID: 1, Comm: "bash"},
 			{PID: 3, PPID: 1, Comm: "apache", Threads: 2},
@@ -32,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{