From 7d6b5a540ec66473b95197b0c5dfba3bd90d8492 Mon Sep 17 00:00:00 2001 From: Tom Wilkie Date: Tue, 15 Sep 2015 03:14:39 +0000 Subject: [PATCH] Revert "Merge pull request #450 from weaveworks/scope-284" This reverts commit 46345e3bdb62de467f787dcb569ab366292502e2, reversing changes made to bf3e9a1601d56534360c2751c1a29e6c568c2025. Conflicts: probe/process/walker_darwin.go --- probe/docker/tagger.go | 9 +- probe/docker/tagger_test.go | 4 +- probe/endpoint/reporter.go | 22 +- probe/endpoint/reporter_test.go | 26 +-- probe/main.go | 16 +- probe/proc/mocks.go | 30 --- probe/proc/mocks_test.go | 27 --- probe/proc/net.go | 43 ---- probe/proc/net_darwin.go | 81 ------- probe/proc/net_linux.go | 162 -------------- probe/proc/reader.go | 165 -------------- probe/proc/reader_linux.go | 202 ------------------ probe/proc/reader_linux_test.go | 93 -------- probe/proc/reader_test.go | 69 ------ probe/process/reporter.go | 7 +- probe/process/reporter_test.go | 18 +- probe/{proc => process}/tree.go | 6 +- probe/{proc => process}/tree_test.go | 10 +- probe/process/walker.go | 56 +++++ .../walker_darwin.go} | 70 +----- probe/process/walker_linux.go | 79 +++++++ probe/process/walker_linux_test.go | 90 ++++++++ probe/process/walker_test.go | 66 ++++++ 23 files changed, 364 insertions(+), 987 deletions(-) delete mode 100644 probe/proc/mocks.go delete mode 100644 probe/proc/mocks_test.go delete mode 100644 probe/proc/net.go delete mode 100644 probe/proc/net_darwin.go delete mode 100644 probe/proc/net_linux.go delete mode 100644 probe/proc/reader.go delete mode 100644 probe/proc/reader_linux.go delete mode 100644 probe/proc/reader_linux_test.go delete mode 100644 probe/proc/reader_test.go rename probe/{proc => process}/tree.go (84%) rename probe/{proc => process}/tree_test.go (77%) create mode 100644 probe/process/walker.go rename probe/{proc/reader_darwin.go => process/walker_darwin.go} (55%) create mode 100644 probe/process/walker_linux.go create mode 100644 probe/process/walker_linux_test.go create mode 100644 probe/process/walker_test.go diff --git a/probe/docker/tagger.go b/probe/docker/tagger.go index 0c8d7cd57..1ccecadb8 100644 --- a/probe/docker/tagger.go +++ b/probe/docker/tagger.go @@ -3,7 +3,6 @@ package docker import ( "strconv" - "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/probe/process" "github.com/weaveworks/scope/report" ) @@ -17,18 +16,18 @@ const ( // These vars are exported for testing. var ( - NewProcessTreeStub = proc.NewTree + NewProcessTreeStub = process.NewTree ) // Tagger is a tagger that tags Docker container information to process // nodes that have a PID. type Tagger struct { registry Registry - procWalker proc.Reader + procWalker process.Walker } // NewTagger returns a usable Tagger. -func NewTagger(registry Registry, procWalker proc.Reader) *Tagger { +func NewTagger(registry Registry, procWalker process.Walker) *Tagger { return &Tagger{ registry: registry, procWalker: procWalker, @@ -45,7 +44,7 @@ func (t *Tagger) Tag(r report.Report) (report.Report, error) { return r, nil } -func (t *Tagger) tag(tree proc.Tree, topology *report.Topology) { +func (t *Tagger) tag(tree process.Tree, topology *report.Topology) { for nodeID, nodeMetadata := range topology.Nodes { pidStr, ok := nodeMetadata.Metadata[process.PID] if !ok { diff --git a/probe/docker/tagger_test.go b/probe/docker/tagger_test.go index 6fc8b697f..261fd4215 100644 --- a/probe/docker/tagger_test.go +++ b/probe/docker/tagger_test.go @@ -6,7 +6,7 @@ import ( "testing" "github.com/weaveworks/scope/probe/docker" - "github.com/weaveworks/scope/probe/proc" + "github.com/weaveworks/scope/probe/process" "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(_ proc.Reader) (proc.Tree, error) { + docker.NewProcessTreeStub = func(_ process.Walker) (process.Tree, error) { return &mockProcessTree{map[int]int{2: 1}}, nil } diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index 2689700b1..6df5eb229 100644 --- a/probe/endpoint/reporter.go +++ b/probe/endpoint/reporter.go @@ -7,7 +7,7 @@ import ( "github.com/prometheus/client_golang/prometheus" - "github.com/weaveworks/scope/probe/proc" + "github.com/weaveworks/procspy" "github.com/weaveworks/scope/probe/process" "github.com/weaveworks/scope/report" ) @@ -28,7 +28,6 @@ type Reporter struct { conntracker *Conntracker natmapper *natmapper revResolver *ReverseResolver - procReader proc.Reader } // SpyDuration is an exported prometheus metric @@ -43,12 +42,12 @@ var SpyDuration = prometheus.NewSummaryVec( []string{}, ) -// NewReporter creates a new Reporter that invokes Connections to +// NewReporter creates a new Reporter that invokes procspy.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, procReader proc.Reader, useConntrack bool) *Reporter { +func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool) *Reporter { var ( conntrackModulePresent = ConntrackModulePresent() conntracker *Conntracker @@ -74,7 +73,6 @@ func NewReporter(hostID, hostName string, includeProcesses bool, procReader proc conntracker: conntracker, natmapper: natmapper, revResolver: NewReverseResolver(), - procReader: procReader, } } @@ -96,7 +94,12 @@ func (r *Reporter) Report() (report.Report, error) { }(time.Now()) rpt := report.MakeReport() - err := r.procReader.Connections(r.includeProcesses, func(conn proc.Connection) { + conns, err := procspy.Connections(r.includeProcesses) + if err != nil { + return rpt, err + } + + for conn := conns.Next(); conn != nil; conn = conns.Next() { var ( localPort = conn.LocalPort remotePort = conn.RemotePort @@ -104,15 +107,12 @@ func (r *Reporter) Report() (report.Report, error) { remoteAddr = conn.RemoteAddress.String() ) extraNodeInfo := report.MakeNode() - if conn.Process.PID > 0 { + if conn.Proc.PID > 0 { extraNodeInfo = extraNodeInfo.WithMetadata(report.Metadata{ - process.PID: strconv.FormatUint(uint64(conn.Process.PID), 10), + process.PID: strconv.FormatUint(uint64(conn.Proc.PID), 10), }) } r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &extraNodeInfo, nil) - }) - if err != nil { - return rpt, err } if r.conntracker != nil { diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index af746048a..c7ec41995 100644 --- a/probe/endpoint/reporter_test.go +++ b/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 = int(4242) + fixProcessPID = uint(4242) fixProcessName = "nginx" - fixConnections = []proc.Connection{ + fixConnections = []procspy.Connection{ { Transport: "tcp", LocalAddress: fixLocalAddress, @@ -37,16 +37,16 @@ var ( }, } - fixConnectionsWithProcesses = []proc.Connection{ + fixConnectionsWithProcesses = []procspy.Connection{ { Transport: "tcp", LocalAddress: fixLocalAddress, LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, RemotePort: fixRemotePort, - Process: proc.Process{ + Proc: procspy.Proc{ PID: fixProcessPID, - Comm: fixProcessName, + Name: fixProcessName, }, }, { @@ -55,22 +55,23 @@ var ( LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, RemotePort: fixRemotePort, - Process: proc.Process{ + Proc: procspy.Proc{ PID: fixProcessPID, - Comm: fixProcessName, + Name: 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 ) - procReader := proc.MockedReader{Conns: fixConnections} - reporter := endpoint.NewReporter(nodeID, nodeName, false, &procReader, false) + reporter := endpoint.NewReporter(nodeID, nodeName, false, false) r, _ := reporter.Report() //buf, _ := json.MarshalIndent(r, "", " ") //t.Logf("\n%s\n", buf) @@ -99,13 +100,14 @@ 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 ) - procReader := proc.MockedReader{Conns: fixConnectionsWithProcesses} - reporter := endpoint.NewReporter(nodeID, nodeName, true, &procReader, false) + reporter := endpoint.NewReporter(nodeID, nodeName, true, false) r, _ := reporter.Report() // buf, _ := json.MarshalIndent(r, "", " ") ; t.Logf("\n%s\n", buf) diff --git a/probe/main.go b/probe/main.go index 33afccf39..eb4495ff9 100644 --- a/probe/main.go +++ b/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,6 +67,8 @@ 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) @@ -109,18 +111,16 @@ func main() { } var ( - procDir = proc.OSDir{Dir: *procRoot} - procReader = proc.NewCachingProcReader(proc.NewReader(procDir), *spyProcs) - tickers = []Ticker{procReader} - endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, procReader, *useConntrack) + endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack) + processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) + tickers = []Ticker{processCache} reporters = []Reporter{ endpointReporter, host.NewReporter(hostID, hostName, localNets), - process.NewReporter(procReader, hostID), + process.NewReporter(processCache, 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, procReader)) + taggers = append(taggers, docker.NewTagger(dockerRegistry, processCache)) reporters = append(reporters, docker.NewReporter(dockerRegistry, hostID)) } diff --git a/probe/proc/mocks.go b/probe/proc/mocks.go deleted file mode 100644 index f31d2b4b2..000000000 --- a/probe/proc/mocks.go +++ /dev/null @@ -1,30 +0,0 @@ -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 -} diff --git a/probe/proc/mocks_test.go b/probe/proc/mocks_test.go deleted file mode 100644 index 3b3905a44..000000000 --- a/probe/proc/mocks_test.go +++ /dev/null @@ -1,27 +0,0 @@ -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) } diff --git a/probe/proc/net.go b/probe/proc/net.go deleted file mode 100644 index 7e9ffbf4d..000000000 --- a/probe/proc/net.go +++ /dev/null @@ -1,43 +0,0 @@ -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) -} diff --git a/probe/proc/net_darwin.go b/probe/proc/net_darwin.go deleted file mode 100644 index 013db6c64..000000000 --- a/probe/proc/net_darwin.go +++ /dev/null @@ -1,81 +0,0 @@ -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 . - 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 -} diff --git a/probe/proc/net_linux.go b/probe/proc/net_linux.go deleted file mode 100644 index e9b7de05b..000000000 --- a/probe/proc/net_linux.go +++ /dev/null @@ -1,162 +0,0 @@ -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 -} diff --git a/probe/proc/reader.go b/probe/proc/reader.go deleted file mode 100644 index 5e43fd967..000000000 --- a/probe/proc/reader.go +++ /dev/null @@ -1,165 +0,0 @@ -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 -} diff --git a/probe/proc/reader_linux.go b/probe/proc/reader_linux.go deleted file mode 100644 index dcd322f60..000000000 --- a/probe/proc/reader_linux.go +++ /dev/null @@ -1,202 +0,0 @@ -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 -} diff --git a/probe/proc/reader_linux_test.go b/probe/proc/reader_linux_test.go deleted file mode 100644 index 29402eb73..000000000 --- a/probe/proc/reader_linux_test.go +++ /dev/null @@ -1,93 +0,0 @@ -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) - } -} diff --git a/probe/proc/reader_test.go b/probe/proc/reader_test.go deleted file mode 100644 index 9d78997d3..000000000 --- a/probe/proc/reader_test.go +++ /dev/null @@ -1,69 +0,0 @@ -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) - } -} diff --git a/probe/process/reporter.go b/probe/process/reporter.go index 97fb8d67c..3d46de07d 100644 --- a/probe/process/reporter.go +++ b/probe/process/reporter.go @@ -3,7 +3,6 @@ package process import ( "strconv" - "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/report" ) @@ -19,11 +18,11 @@ const ( // Reporter generates Reports containing the Process topology. type Reporter struct { scope string - walker proc.Reader + walker Walker } // NewReporter makes a new Reporter. -func NewReporter(walker proc.Reader, scope string) *Reporter { +func NewReporter(walker Walker, scope string) *Reporter { return &Reporter{ scope: scope, walker: walker, @@ -43,7 +42,7 @@ func (r *Reporter) Report() (report.Report, error) { func (r *Reporter) processTopology() (report.Topology, error) { t := report.MakeTopology() - err := r.walker.Processes(func(p proc.Process) { + err := r.walker.Walk(func(p Process) { pidstr := strconv.Itoa(p.PID) nodeID := report.MakeProcessNodeID(r.scope, pidstr) t.Nodes[nodeID] = report.MakeNode() diff --git a/probe/process/reporter_test.go b/probe/process/reporter_test.go index 0aac3e500..d13b2611f 100644 --- a/probe/process/reporter_test.go +++ b/probe/process/reporter_test.go @@ -4,15 +4,25 @@ 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) { - procReader := &proc.MockedReader{ - Procs: []proc.Process{ + walker := &mockWalker{ + processes: []process.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, @@ -21,7 +31,7 @@ func TestReporter(t *testing.T) { }, } - reporter := process.NewReporter(procReader, "") + reporter := process.NewReporter(walker, "") want := report.MakeReport() want.Process = report.Topology{ Nodes: report.Nodes{ diff --git a/probe/proc/tree.go b/probe/process/tree.go similarity index 84% rename from probe/proc/tree.go rename to probe/process/tree.go index f9d48b959..2d03a9c0f 100644 --- a/probe/proc/tree.go +++ b/probe/process/tree.go @@ -1,4 +1,4 @@ -package proc +package process import ( "fmt" @@ -14,9 +14,9 @@ type tree struct { } // NewTree returns a new Tree that can be polled. -func NewTree(walker Reader) (Tree, error) { +func NewTree(walker Walker) (Tree, error) { pt := tree{processes: map[int]Process{}} - err := walker.Processes(func(p Process) { + err := walker.Walk(func(p Process) { pt.processes[p.PID] = p }) diff --git a/probe/proc/tree_test.go b/probe/process/tree_test.go similarity index 77% rename from probe/proc/tree_test.go rename to probe/process/tree_test.go index 5377a9e92..6a93cc226 100644 --- a/probe/proc/tree_test.go +++ b/probe/process/tree_test.go @@ -1,15 +1,15 @@ -package proc_test +package process_test import ( "reflect" "testing" - "github.com/weaveworks/scope/probe/proc" + "github.com/weaveworks/scope/probe/process" ) func TestTree(t *testing.T) { - walker := &proc.MockedReader{ - Procs: []proc.Process{ + walker := &mockWalker{ + processes: []process.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 := proc.NewTree(walker) + tree, err := process.NewTree(walker) if err != nil { t.Fatalf("newProcessTree error: %v", err) } diff --git a/probe/process/walker.go b/probe/process/walker.go new file mode 100644 index 000000000..4ba20027e --- /dev/null +++ b/probe/process/walker.go @@ -0,0 +1,56 @@ +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 +} diff --git a/probe/proc/reader_darwin.go b/probe/process/walker_darwin.go similarity index 55% rename from probe/proc/reader_darwin.go rename to probe/process/walker_darwin.go index 456e60fb7..73765ed95 100644 --- a/probe/proc/reader_darwin.go +++ b/probe/process/walker_darwin.go @@ -1,27 +1,28 @@ -package proc +package process import ( "fmt" - "net" "os/exec" "strconv" "strings" ) -type reader struct{} - -// NewReader returns a Darwin (lsof-based) '/proc' reader -func NewReader(proc Dir) Reader { - return &reader{} +// NewWalker returns a Darwin (lsof-based) walker. +func NewWalker(_ string) Walker { + return &walker{} } +type walker struct{} + const ( lsofBinary = "lsof" lsofFields = "cn" // parseLSOF() depends on the order netstatBinary = "netstat" ) -func (reader) Processes(f func(Process)) error { +// These functions copied from procspy. + +func (walker) Walk(f func(Process)) error { output, err := exec.Command( lsofBinary, "-i", // only Internet files @@ -44,59 +45,6 @@ func (reader) Processes(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].Process = 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 diff --git a/probe/process/walker_linux.go b/probe/process/walker_linux.go new file mode 100644 index 000000000..f84ded6a6 --- /dev/null +++ b/probe/process/walker_linux.go @@ -0,0 +1,79 @@ +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 +} diff --git a/probe/process/walker_linux_test.go b/probe/process/walker_linux_test.go new file mode 100644 index 000000000..6b449edd9 --- /dev/null +++ b/probe/process/walker_linux_test.go @@ -0,0 +1,90 @@ +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) + } +} diff --git a/probe/process/walker_test.go b/probe/process/walker_test.go new file mode 100644 index 000000000..6046006bb --- /dev/null +++ b/probe/process/walker_test.go @@ -0,0 +1,66 @@ +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 +}