From 4331fb1c793a34c2a1b5f579066801a4055b40e4 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Tue, 1 Sep 2015 14:01:30 +0200 Subject: [PATCH 1/5] Drop all the dependencies we had on `procspy` by moving some code to `probe`. --- probe/docker/tagger.go | 9 +- probe/docker/tagger_test.go | 4 +- probe/endpoint/reporter.go | 6 +- probe/endpoint/reporter_test.go | 14 +- probe/main.go | 6 +- probe/proc/fixture.go | 26 +++ probe/proc/proc.go | 125 ++++++++++++++ probe/proc/procnet.go | 164 +++++++++++++++++++ probe/proc/spy.go | 40 +++++ probe/proc/spy_linux.go | 52 ++++++ probe/{process => proc}/tree.go | 2 +- probe/{process => proc}/tree_test.go | 8 +- probe/{process => proc}/walker.go | 2 +- probe/{process => proc}/walker_darwin.go | 2 +- probe/{process => proc}/walker_linux.go | 2 +- probe/{process => proc}/walker_linux_test.go | 22 +-- probe/{process => proc}/walker_test.go | 33 ++-- probe/process/reporter.go | 7 +- probe/process/reporter_test.go | 7 +- 19 files changed, 476 insertions(+), 55 deletions(-) create mode 100644 probe/proc/fixture.go create mode 100644 probe/proc/proc.go create mode 100644 probe/proc/procnet.go create mode 100644 probe/proc/spy.go create mode 100644 probe/proc/spy_linux.go rename probe/{process => proc}/tree.go (97%) rename probe/{process => proc}/tree_test.go (80%) rename probe/{process => proc}/walker.go (98%) rename probe/{process => proc}/walker_darwin.go (99%) rename probe/{process => proc}/walker_linux.go (99%) rename probe/{process => proc}/walker_linux_test.go (80%) rename probe/{process => proc}/walker_test.go (64%) diff --git a/probe/docker/tagger.go b/probe/docker/tagger.go index 1ccecadb8..2c1923485 100644 --- a/probe/docker/tagger.go +++ b/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.Walker } // NewTagger returns a usable Tagger. -func NewTagger(registry Registry, procWalker process.Walker) *Tagger { +func NewTagger(registry Registry, procWalker proc.Walker) *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 { diff --git a/probe/docker/tagger_test.go b/probe/docker/tagger_test.go index 261fd4215..ce2d48871 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/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.Walker) (proc.Tree, error) { return &mockProcessTree{map[int]int{2: 1}}, nil } diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index 6df5eb229..1841b2db2 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/procspy" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/probe/process" "github.com/weaveworks/scope/report" ) @@ -42,7 +42,7 @@ 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 @@ -94,7 +94,7 @@ func (r *Reporter) Report() (report.Report, error) { }(time.Now()) rpt := report.MakeReport() - conns, err := procspy.Connections(r.includeProcesses) + conns, err := proc.Connections(r.includeProcesses) if err != nil { return rpt, err } diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index c7ec41995..f6bc0a461 100644 --- a/probe/endpoint/reporter_test.go +++ b/probe/endpoint/reporter_test.go @@ -5,7 +5,7 @@ import ( "strconv" "testing" - "github.com/weaveworks/procspy" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/probe/docker" "github.com/weaveworks/scope/probe/endpoint" "github.com/weaveworks/scope/report" @@ -20,7 +20,7 @@ var ( fixProcessPID = uint(4242) fixProcessName = "nginx" - fixConnections = []procspy.Connection{ + fixConnections = []proc.Connection{ { Transport: "tcp", LocalAddress: fixLocalAddress, @@ -37,14 +37,14 @@ var ( }, } - fixConnectionsWithProcesses = []procspy.Connection{ + fixConnectionsWithProcesses = []proc.Connection{ { Transport: "tcp", LocalAddress: fixLocalAddress, LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, RemotePort: fixRemotePort, - Proc: procspy.Proc{ + Proc: proc.Proc{ PID: fixProcessPID, Name: fixProcessName, }, @@ -55,7 +55,7 @@ var ( LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, RemotePort: fixRemotePort, - Proc: procspy.Proc{ + Proc: proc.Proc{ PID: fixProcessPID, Name: fixProcessName, }, @@ -64,7 +64,7 @@ var ( ) func TestSpyNoProcesses(t *testing.T) { - procspy.SetFixtures(fixConnections) + proc.SetFixtures(fixConnections) const ( nodeID = "heinz-tomato-ketchup" // TODO rename to hostID @@ -100,7 +100,7 @@ func TestSpyNoProcesses(t *testing.T) { } func TestSpyWithProcesses(t *testing.T) { - procspy.SetFixtures(fixConnectionsWithProcesses) + proc.SetFixtures(fixConnectionsWithProcesses) const ( nodeID = "nikon" // TODO rename to hostID diff --git a/probe/main.go b/probe/main.go index eb4495ff9..f0878c04a 100644 --- a/probe/main.go +++ b/probe/main.go @@ -14,7 +14,7 @@ import ( "syscall" "time" - "github.com/weaveworks/procspy" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/probe/docker" "github.com/weaveworks/scope/probe/endpoint" "github.com/weaveworks/scope/probe/host" @@ -67,7 +67,7 @@ func main() { } log.Printf("publishing to: %s", strings.Join(targets, ", ")) - procspy.SetProcRoot(*procRoot) + proc.SetProcRoot(*procRoot) if *httpListen != "" { log.Printf("profiling data being exported to %s", *httpListen) @@ -112,7 +112,7 @@ func main() { var ( endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack) - processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) + processCache = proc.NewCachingWalker(proc.NewWalker(*procRoot)) tickers = []Ticker{processCache} reporters = []Reporter{ endpointReporter, diff --git a/probe/proc/fixture.go b/probe/proc/fixture.go new file mode 100644 index 000000000..1d69a84c9 --- /dev/null +++ b/probe/proc/fixture.go @@ -0,0 +1,26 @@ +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 + } +} diff --git a/probe/proc/proc.go b/probe/proc/proc.go new file mode 100644 index 000000000..fb893e414 --- /dev/null +++ b/probe/proc/proc.go @@ -0,0 +1,125 @@ +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 +} diff --git a/probe/proc/procnet.go b/probe/proc/procnet.go new file mode 100644 index 000000000..4f18229f4 --- /dev/null +++ b/probe/proc/procnet.go @@ -0,0 +1,164 @@ +package proc + +import ( + "bytes" + "net" +) + +// ProcNet is an iterator to parse /proc/net/tcp{,6} files. +type ProcNet struct { + b []byte + c Connection + wantedState uint + bytesLocal, bytesRemote [16]byte +} + +// NewProcNet gives a new ProcNet parser. +func NewProcNet(b []byte, wantedState uint) *ProcNet { + return &ProcNet{ + 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 *ProcNet) 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/spy.go b/probe/proc/spy.go new file mode 100644 index 000000000..d1b93c201 --- /dev/null +++ b/probe/proc/spy.go @@ -0,0 +1,40 @@ +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) +} diff --git a/probe/proc/spy_linux.go b/probe/proc/spy_linux.go new file mode 100644 index 000000000..fc8141c41 --- /dev/null +++ b/probe/proc/spy_linux.go @@ -0,0 +1,52 @@ +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 +} diff --git a/probe/process/tree.go b/probe/proc/tree.go similarity index 97% rename from probe/process/tree.go rename to probe/proc/tree.go index 2d03a9c0f..12506d57b 100644 --- a/probe/process/tree.go +++ b/probe/proc/tree.go @@ -1,4 +1,4 @@ -package process +package proc import ( "fmt" diff --git a/probe/process/tree_test.go b/probe/proc/tree_test.go similarity index 80% rename from probe/process/tree_test.go rename to probe/proc/tree_test.go index 6a93cc226..cb3c79227 100644 --- a/probe/process/tree_test.go +++ b/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{ + processes: []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) } diff --git a/probe/process/walker.go b/probe/proc/walker.go similarity index 98% rename from probe/process/walker.go rename to probe/proc/walker.go index 4ba20027e..a7c5b1541 100644 --- a/probe/process/walker.go +++ b/probe/proc/walker.go @@ -1,4 +1,4 @@ -package process +package proc import "sync" diff --git a/probe/process/walker_darwin.go b/probe/proc/walker_darwin.go similarity index 99% rename from probe/process/walker_darwin.go rename to probe/proc/walker_darwin.go index cd25b3df3..c997d9385 100644 --- a/probe/process/walker_darwin.go +++ b/probe/proc/walker_darwin.go @@ -1,4 +1,4 @@ -package process +package proc import ( "fmt" diff --git a/probe/process/walker_linux.go b/probe/proc/walker_linux.go similarity index 99% rename from probe/process/walker_linux.go rename to probe/proc/walker_linux.go index f84ded6a6..51cee7d07 100644 --- a/probe/process/walker_linux.go +++ b/probe/proc/walker_linux.go @@ -1,4 +1,4 @@ -package process +package proc import ( "bytes" diff --git a/probe/process/walker_linux_test.go b/probe/proc/walker_linux_test.go similarity index 80% rename from probe/process/walker_linux_test.go rename to probe/proc/walker_linux_test.go index 6b449edd9..819b472f8 100644 --- a/probe/process/walker_linux_test.go +++ b/probe/proc/walker_linux_test.go @@ -1,4 +1,4 @@ -package process_test +package proc_test import ( "fmt" @@ -9,7 +9,7 @@ import ( "testing" "time" - "github.com/weaveworks/scope/probe/process" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/test" ) @@ -25,10 +25,10 @@ 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 + oldReadDir, oldReadFile := proc.ReadDir, proc.ReadFile defer func() { - process.ReadDir = oldReadDir - process.ReadFile = oldReadFile + proc.ReadDir = oldReadDir + proc.ReadFile = oldReadFile }() processes := map[string]mockProcess{ @@ -39,7 +39,7 @@ func TestWalker(t *testing.T) { "1": {name: "1", comm: "init\n"}, } - process.ReadDir = func(path string) ([]os.FileInfo, error) { + proc.ReadDir = func(path string) ([]os.FileInfo, error) { result := []os.FileInfo{} for _, p := range processes { result = append(result, p) @@ -47,7 +47,7 @@ func TestWalker(t *testing.T) { return result, nil } - process.ReadFile = func(path string) ([]byte, error) { + proc.ReadFile = func(path string) ([]byte, error) { splits := strings.Split(path, "/") pid := splits[len(splits)-2] @@ -71,16 +71,16 @@ func TestWalker(t *testing.T) { return nil, fmt.Errorf("not found") } - want := map[int]process.Process{ + 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]process.Process{} - walker := process.NewWalker("unused") - err := walker.Walk(func(p process.Process) { + have := map[int]proc.Process{} + walker := proc.NewWalker("unused") + err := walker.Walk(func(p proc.Process) { have[p.PID] = p }) diff --git a/probe/process/walker_test.go b/probe/proc/walker_test.go similarity index 64% rename from probe/process/walker_test.go rename to probe/proc/walker_test.go index 6046006bb..2cac7548c 100644 --- a/probe/process/walker_test.go +++ b/probe/proc/walker_test.go @@ -1,25 +1,36 @@ -package process_test +package proc_test import ( "reflect" "testing" - "github.com/weaveworks/scope/probe/process" + "github.com/weaveworks/scope/probe/proc" "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(process.Process) {} + procFunc = func(proc.Process) {} ) - if err := process.NewWalker(procRoot).Walk(procFunc); err != nil { + if err := proc.NewWalker(procRoot).Walk(procFunc); err != nil { t.Fatal(err) } } func TestCache(t *testing.T) { - processes := []process.Process{ + processes := []proc.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, @@ -28,7 +39,7 @@ func TestCache(t *testing.T) { walker := &mockWalker{ processes: processes, } - cachingWalker := process.NewCachingWalker(walker) + cachingWalker := proc.NewCachingWalker(walker) err := cachingWalker.Tick() if err != nil { t.Fatal(err) @@ -39,7 +50,7 @@ func TestCache(t *testing.T) { t.Errorf("%v (%v)", test.Diff(processes, have), err) } - walker.processes = []process.Process{} + walker.processes = []proc.Process{} have, err = all(cachingWalker) if err != nil || !reflect.DeepEqual(processes, have) { t.Errorf("%v (%v)", test.Diff(processes, have), err) @@ -51,15 +62,15 @@ func TestCache(t *testing.T) { } have, err = all(cachingWalker) - want := []process.Process{} + want := []proc.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) { +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 diff --git a/probe/process/reporter.go b/probe/process/reporter.go index 3d46de07d..64eecef1c 100644 --- a/probe/process/reporter.go +++ b/probe/process/reporter.go @@ -4,6 +4,7 @@ import ( "strconv" "github.com/weaveworks/scope/report" + "github.com/weaveworks/scope/probe/proc" ) // We use these keys in node metadata @@ -18,11 +19,11 @@ const ( // Reporter generates Reports containing the Process topology. type Reporter struct { scope string - walker Walker + walker proc.Walker } // NewReporter makes a new Reporter. -func NewReporter(walker Walker, scope string) *Reporter { +func NewReporter(walker proc.Walker, 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.Walk(func(p proc.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 d13b2611f..1eab60451 100644 --- a/probe/process/reporter_test.go +++ b/probe/process/reporter_test.go @@ -4,16 +4,17 @@ 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 + processes []proc.Process } -func (m *mockWalker) Walk(f func(process.Process)) error { +func (m *mockWalker) Walk(f func(proc.Process)) error { for _, p := range m.processes { f(p) } @@ -22,7 +23,7 @@ func (m *mockWalker) Walk(f func(process.Process)) error { func TestReporter(t *testing.T) { walker := &mockWalker{ - processes: []process.Process{ + processes: []proc.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, From 14dc6a53916dfe5823248d61803953f6e82c0f6e Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Tue, 8 Sep 2015 13:31:30 +0200 Subject: [PATCH 2/5] Cache processes and connections when reading from the /proc Code cleanups --- probe/docker/tagger.go | 4 +- probe/docker/tagger_test.go | 2 +- probe/endpoint/reporter.go | 18 +-- probe/endpoint/reporter_test.go | 22 ++- probe/main.go | 15 +- probe/proc/fixture.go | 26 ---- probe/proc/mocks.go | 59 ++++++++ probe/proc/net.go | 42 ++++++ probe/proc/net_darwin.go | 81 +++++++++++ probe/proc/{procnet.go => net_linux.go} | 22 +-- probe/proc/proc.go | 125 ----------------- probe/proc/reader.go | 117 ++++++++++++++++ .../{walker_darwin.go => reader_darwin.go} | 69 ++++++++-- probe/proc/reader_linux.go | 130 ++++++++++++++++++ probe/proc/reader_linux_test.go | 73 ++++++++++ probe/proc/{walker_test.go => reader_test.go} | 56 +++----- probe/proc/spy.go | 40 ------ probe/proc/spy_linux.go | 52 ------- probe/proc/tree.go | 4 +- probe/proc/tree_test.go | 4 +- probe/proc/walker.go | 56 -------- probe/proc/walker_linux.go | 79 ----------- probe/proc/walker_linux_test.go | 90 ------------ probe/process/reporter.go | 6 +- probe/process/reporter_test.go | 17 +-- 25 files changed, 632 insertions(+), 577 deletions(-) delete mode 100644 probe/proc/fixture.go create mode 100644 probe/proc/mocks.go create mode 100644 probe/proc/net.go create mode 100644 probe/proc/net_darwin.go rename probe/proc/{procnet.go => net_linux.go} (88%) delete mode 100644 probe/proc/proc.go create mode 100644 probe/proc/reader.go rename probe/proc/{walker_darwin.go => reader_darwin.go} (53%) create mode 100644 probe/proc/reader_linux.go create mode 100644 probe/proc/reader_linux_test.go rename probe/proc/{walker_test.go => reader_test.go} (50%) delete mode 100644 probe/proc/spy.go delete mode 100644 probe/proc/spy_linux.go delete mode 100644 probe/proc/walker.go delete mode 100644 probe/proc/walker_linux.go delete mode 100644 probe/proc/walker_linux_test.go diff --git a/probe/docker/tagger.go b/probe/docker/tagger.go index 2c1923485..ec2b7d120 100644 --- a/probe/docker/tagger.go +++ b/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, diff --git a/probe/docker/tagger_test.go b/probe/docker/tagger_test.go index ce2d48871..7a6d3f4b1 100644 --- a/probe/docker/tagger_test.go +++ b/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 } diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index 1841b2db2..ba88b48cd 100644 --- a/probe/endpoint/reporter.go +++ b/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 { diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index f6bc0a461..8ebc1cceb 100644 --- a/probe/endpoint/reporter_test.go +++ b/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) diff --git a/probe/main.go b/probe/main.go index f0878c04a..5b4cff731 100644 --- a/probe/main.go +++ b/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)) } diff --git a/probe/proc/fixture.go b/probe/proc/fixture.go deleted file mode 100644 index 1d69a84c9..000000000 --- a/probe/proc/fixture.go +++ /dev/null @@ -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 - } -} diff --git a/probe/proc/mocks.go b/probe/proc/mocks.go new file mode 100644 index 000000000..057ce2e2c --- /dev/null +++ b/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 +} diff --git a/probe/proc/net.go b/probe/proc/net.go new file mode 100644 index 000000000..1d2f933f6 --- /dev/null +++ b/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) +} diff --git a/probe/proc/net_darwin.go b/probe/proc/net_darwin.go new file mode 100644 index 000000000..013db6c64 --- /dev/null +++ b/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 . + 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/procnet.go b/probe/proc/net_linux.go similarity index 88% rename from probe/proc/procnet.go rename to probe/proc/net_linux.go index 4f18229f4..8a0871514 100644 --- a/probe/proc/procnet.go +++ b/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 diff --git a/probe/proc/proc.go b/probe/proc/proc.go deleted file mode 100644 index fb893e414..000000000 --- a/probe/proc/proc.go +++ /dev/null @@ -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 -} diff --git a/probe/proc/reader.go b/probe/proc/reader.go new file mode 100644 index 000000000..fa77c0805 --- /dev/null +++ b/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 +} diff --git a/probe/proc/walker_darwin.go b/probe/proc/reader_darwin.go similarity index 53% rename from probe/proc/walker_darwin.go rename to probe/proc/reader_darwin.go index c997d9385..67fb70280 100644 --- a/probe/proc/walker_darwin.go +++ b/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 diff --git a/probe/proc/reader_linux.go b/probe/proc/reader_linux.go new file mode 100644 index 000000000..ba41feced --- /dev/null +++ b/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 +} diff --git a/probe/proc/reader_linux_test.go b/probe/proc/reader_linux_test.go new file mode 100644 index 000000000..1974276b0 --- /dev/null +++ b/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) + } +} diff --git a/probe/proc/walker_test.go b/probe/proc/reader_test.go similarity index 50% rename from probe/proc/walker_test.go rename to probe/proc/reader_test.go index 2cac7548c..8273268cd 100644 --- a/probe/proc/walker_test.go +++ b/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 -} diff --git a/probe/proc/spy.go b/probe/proc/spy.go deleted file mode 100644 index d1b93c201..000000000 --- a/probe/proc/spy.go +++ /dev/null @@ -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) -} diff --git a/probe/proc/spy_linux.go b/probe/proc/spy_linux.go deleted file mode 100644 index fc8141c41..000000000 --- a/probe/proc/spy_linux.go +++ /dev/null @@ -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 -} diff --git a/probe/proc/tree.go b/probe/proc/tree.go index 12506d57b..9a4206786 100644 --- a/probe/proc/tree.go +++ b/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 }) diff --git a/probe/proc/tree_test.go b/probe/proc/tree_test.go index cb3c79227..92ec42c30 100644 --- a/probe/proc/tree_test.go +++ b/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}, diff --git a/probe/proc/walker.go b/probe/proc/walker.go deleted file mode 100644 index a7c5b1541..000000000 --- a/probe/proc/walker.go +++ /dev/null @@ -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 -} diff --git a/probe/proc/walker_linux.go b/probe/proc/walker_linux.go deleted file mode 100644 index 51cee7d07..000000000 --- a/probe/proc/walker_linux.go +++ /dev/null @@ -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 -} diff --git a/probe/proc/walker_linux_test.go b/probe/proc/walker_linux_test.go deleted file mode 100644 index 819b472f8..000000000 --- a/probe/proc/walker_linux_test.go +++ /dev/null @@ -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) - } -} diff --git a/probe/process/reporter.go b/probe/process/reporter.go index 64eecef1c..196d2d6dc 100644 --- a/probe/process/reporter.go +++ b/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() diff --git a/probe/process/reporter_test.go b/probe/process/reporter_test.go index 1eab60451..a0c84f4fa 100644 --- a/probe/process/reporter_test.go +++ b/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{ From 2f9f54688fd9e8fb8880e309f1b03e1543081f45 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Thu, 10 Sep 2015 14:05:24 +0200 Subject: [PATCH 3/5] Keep a cache of open files, reducing the number of open/close cycles in the /proc dir --- probe/docker/tagger.go | 4 +- probe/docker/tagger_test.go | 2 +- probe/endpoint/reporter.go | 4 +- probe/endpoint/reporter_test.go | 6 +- probe/main.go | 5 +- probe/proc/mocks.go | 53 ++++----------- probe/proc/mocks_test.go | 27 ++++++++ probe/proc/net_linux.go | 10 ++- probe/proc/reader.go | 84 +++++++++++++++++------ probe/proc/reader_darwin.go | 17 +++-- probe/proc/reader_linux.go | 116 ++++++++++++++++++++++++++------ probe/proc/reader_linux_test.go | 58 ++++++++++------ probe/proc/reader_test.go | 12 +++- probe/proc/tree.go | 2 +- probe/proc/tree_test.go | 2 +- probe/process/reporter.go | 6 +- probe/process/reporter_test.go | 2 +- 17 files changed, 275 insertions(+), 135 deletions(-) create mode 100644 probe/proc/mocks_test.go diff --git a/probe/docker/tagger.go b/probe/docker/tagger.go index ec2b7d120..0c8d7cd57 100644 --- a/probe/docker/tagger.go +++ b/probe/docker/tagger.go @@ -24,11 +24,11 @@ var ( // nodes that have a PID. type Tagger struct { registry Registry - procWalker proc.ProcReader + procWalker proc.Reader } // NewTagger returns a usable Tagger. -func NewTagger(registry Registry, procWalker proc.ProcReader) *Tagger { +func NewTagger(registry Registry, procWalker proc.Reader) *Tagger { return &Tagger{ registry: registry, procWalker: procWalker, diff --git a/probe/docker/tagger_test.go b/probe/docker/tagger_test.go index 7a6d3f4b1..6fc8b697f 100644 --- a/probe/docker/tagger_test.go +++ b/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.ProcReader) (proc.Tree, error) { + docker.NewProcessTreeStub = func(_ proc.Reader) (proc.Tree, error) { return &mockProcessTree{map[int]int{2: 1}}, nil } diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index ba88b48cd..2689700b1 100644 --- a/probe/endpoint/reporter.go +++ b/probe/endpoint/reporter.go @@ -28,7 +28,7 @@ type Reporter struct { conntracker *Conntracker natmapper *natmapper revResolver *ReverseResolver - procReader proc.ProcReader + procReader proc.Reader } // SpyDuration is an exported prometheus metric @@ -48,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, procReader proc.ProcReader, useConntrack bool) *Reporter { +func NewReporter(hostID, hostName string, includeProcesses bool, procReader proc.Reader, useConntrack bool) *Reporter { var ( conntrackModulePresent = ConntrackModulePresent() conntracker *Conntracker diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index 8ebc1cceb..af746048a 100644 --- a/probe/endpoint/reporter_test.go +++ b/probe/endpoint/reporter_test.go @@ -17,7 +17,7 @@ var ( fixRemoteAddress = net.ParseIP("192.168.1.2") fixRemotePort = uint16(12345) fixRemotePortB = uint16(12346) - fixProcessPID = int(4242) + fixProcessPID = int(4242) fixProcessName = "nginx" fixConnections = []proc.Connection{ @@ -69,7 +69,7 @@ func TestSpyNoProcesses(t *testing.T) { nodeName = "frenchs-since-1904" // TODO rename to hostNmae ) - procReader := proc.MockedProcReader{Conns: fixConnections} + procReader := proc.MockedReader{Conns: fixConnections} reporter := endpoint.NewReporter(nodeID, nodeName, false, &procReader, false) r, _ := reporter.Report() //buf, _ := json.MarshalIndent(r, "", " ") @@ -104,7 +104,7 @@ func TestSpyWithProcesses(t *testing.T) { nodeName = "fishermans-friend" // TODO rename to hostNmae ) - procReader := proc.MockedProcReader{Conns: fixConnectionsWithProcesses} + 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) diff --git a/probe/main.go b/probe/main.go index 5b4cff731..33afccf39 100644 --- a/probe/main.go +++ b/probe/main.go @@ -109,8 +109,8 @@ func main() { } var ( - procDir = proc.OSProcDir{Dir: *procRoot} - procReader = proc.NewCachingProcReader(proc.NewProcReader(procDir), *spyProcs) + 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{ @@ -120,6 +120,7 @@ func main() { } taggers = []Tagger{newTopologyTagger(), host.NewTagger(hostID)} ) + defer procReader.Close() defer endpointReporter.Stop() if *dockerEnabled { diff --git a/probe/proc/mocks.go b/probe/proc/mocks.go index 057ce2e2c..f31d2b4b2 100644 --- a/probe/proc/mocks.go +++ b/probe/proc/mocks.go @@ -1,59 +1,30 @@ package proc -import ( - "bytes" - "os" - "time" -) +// note: we must keep this in the "proc" package so it can be used from other packages.. -// 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 { +// MockedReader is a mocked "/proc" reader +type MockedReader struct { Procs []Process Conns []Connection } -func (mw MockedProcReader) Processes(f func(Process)) error { +// 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 } -func (mw *MockedProcReader) Connections(_ bool, f func(Connection)) error { +// 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 new file mode 100644 index 000000000..3b3905a44 --- /dev/null +++ b/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) } diff --git a/probe/proc/net_linux.go b/probe/proc/net_linux.go index 8a0871514..e9b7de05b 100644 --- a/probe/proc/net_linux.go +++ b/probe/proc/net_linux.go @@ -5,17 +5,15 @@ import ( "net" ) -// ProcNet is an iterator to parse /proc/net/tcp{,6} files. -type ProcNet struct { +type netReader struct { b []byte c Connection wantedState uint bytesLocal, bytesRemote [16]byte } -// NewProcNet gives a new ProcNet parser. -func NewProcNet(b []byte, wantedState uint) *ProcNet { - return &ProcNet{ +func newNetReader(b []byte, wantedState uint) *netReader { + return &netReader{ b: b, c: Connection{}, wantedState: wantedState, @@ -24,7 +22,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 { +func (p *netReader) Next() *Connection { again: if len(p.b) == 0 { return nil diff --git a/probe/proc/reader.go b/probe/proc/reader.go index fa77c0805..6ef7c4a4b 100644 --- a/probe/proc/reader.go +++ b/probe/proc/reader.go @@ -2,37 +2,70 @@ package proc import ( "bytes" - "io/ioutil" "os" + "path" "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 +// File is a file in the "/proc" directory +type File interface { + ReadInto(buf *bytes.Buffer) error + Close() error } -type OSProcDir struct{ Dir string } +// OSFile is a native file +type OSFile struct{ *os.File } -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 { +// 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(f) - f.Close() + _, 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 = OSProcDir{Dir: "/proc"} +var DefaultProcDir = OSDir{Dir: "/proc"} // Process represents a single process. type Process struct { @@ -43,13 +76,15 @@ type Process struct { Inodes []uint64 } -// ProcReader is something that reads the /proc directory and +// Reader is something that reads the /proc directory and // returns some info like processes and connections -type ProcReader interface { +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 @@ -57,13 +92,13 @@ type ProcReader interface { type CachingProcReader struct { procsCache []Process connsCache []Connection - source ProcReader + source Reader includeProcs bool sync.RWMutex } // NewCachingProcReader returns a new CachingProcReader -func NewCachingProcReader(source ProcReader, includeProcs bool) *CachingProcReader { +func NewCachingProcReader(source Reader, includeProcs bool) *CachingProcReader { return &CachingProcReader{source: source, includeProcs: includeProcs} } @@ -91,6 +126,11 @@ func (c *CachingProcReader) Connections(_ bool, f func(Connection)) error { } +// Close closes the "/proc" reader +func (c *CachingProcReader) Close() error { + return c.source.Close() +} + // Update updates the cached copy of the processes and connections lists func (c *CachingProcReader) Tick() error { newProcsCache := []Process{} diff --git a/probe/proc/reader_darwin.go b/probe/proc/reader_darwin.go index 67fb70280..3e48f2730 100644 --- a/probe/proc/reader_darwin.go +++ b/probe/proc/reader_darwin.go @@ -8,11 +8,11 @@ import ( "strings" ) -type procReader struct{} +type reader struct{} -// NewProcReader returns a Darwin (lsof-based) '/proc' reader -func NewProcReader(proc ProcDir) *procReader { - return &procReader{} +// NewReader returns a Darwin (lsof-based) '/proc' reader +func NewReader(proc Dir) Reader { + return &reader{} } const ( @@ -22,7 +22,7 @@ const ( lsofBinary = "lsof" ) -func (procReader) Processes(f func(Process)) error { +func (reader) Processes(f func(Process)) error { output, err := exec.Command( lsofBinary, "-i", // only Internet files @@ -45,7 +45,7 @@ func (procReader) Processes(f func(Process)) error { return nil } -func (w *walker) Connections(withProcs bool, f func(Connection)) error { +func (w *reader) Connections(withProcs bool, f func(Connection)) error { out, err := exec.Command( netstatBinary, "-n", // no number resolving @@ -93,6 +93,11 @@ func (w *walker) Connections(withProcs bool, f func(Connection)) error { return nil } +// Close closes the Darwin "/proc" reader +func (w *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/proc/reader_linux.go b/probe/proc/reader_linux.go index ba41feced..8a27ec946 100644 --- a/probe/proc/reader_linux.go +++ b/probe/proc/reader_linux.go @@ -7,39 +7,106 @@ import ( "strings" "sync" "syscall" + "time" + + "github.com/bluele/gcache" ) -type procReader struct { - proc ProcDir +const ( + filesCacheLen = 512 + filesCacheExpiration = 60 * time.Second +) + +var tcpFiles = []string{ + "net/tcp", + "net/tcp6", } -// NewProcReader creates a new /proc reader. -func NewProcReader(proc ProcDir) *procReader { - return &procReader{proc} +// A cache for files handles +// Note: not intended to be used from multiple goroutines +type filesCache struct { + handles gcache.Cache +} + +func newFilesCache(proc Dir) *filesCache { + loadFunc := func(fileName interface{}) (interface{}, error) { + return proc.Open(fileName.(string)) + } + evictionFunc := func(key, value interface{}) { + value.(File).Close() + } + return &filesCache{ + handles: gcache.New(filesCacheLen).LoaderFunc(loadFunc).EvictedFunc(evictionFunc).Expiration(filesCacheExpiration).ARC().Build(), + } +} + +// Read a "/proc" file, identified as a subdir (eg "1134/comm"), into a buffer +// Note: this is not goroutine-safe: two goroutines getting and reading from +// the same handle can obtain some unexpected contents... +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 + } + return h.(File).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 (w *reader) Close() error { + return w.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 (w *procReader) Processes(f func(Process)) error { - dirEntries, err := w.proc.ReadDir(w.proc.Root()) +func (w *reader) Processes(f func(Process)) error { + dirEntries, err := w.proc.ReadDirNames(w.proc.Root()) if err != nil { return err } - for _, dirEntry := range dirEntries { - filename := dirEntry.Name() + buf := bytes.NewBuffer(make([]byte, 0, 5000)) + readIntoBuffer := func(filename string) error { + buf.Reset() + res := w.handles.ReadInto(filename, buf) + return res + } + + for _, filename := range dirEntries { pid, err := strconv.Atoi(filename) if err != nil { continue } - stat, err := w.proc.ReadFile(path.Join(w.proc.Root(), filename, "stat")) - if err != nil { + if readIntoBuffer(path.Join(filename, "stat")) != nil { continue } - splits := strings.Fields(string(stat)) + splits := strings.Fields(buf.String()) ppid, err := strconv.Atoi(splits[3]) if err != nil { return err @@ -51,18 +118,18 @@ func (w *procReader) Processes(f func(Process)) error { } cmdline := "" - if cmdlineBuf, err := w.proc.ReadFile(path.Join(w.proc.Root(), filename, "cmdline")); err == nil { - cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1) + if readIntoBuffer(path.Join(filename, "cmdline")) == nil { + cmdlineBuf := bytes.Replace(buf.Bytes(), []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)) + if readIntoBuffer(path.Join(filename, "comm")) == nil { + comm = strings.TrimSpace(buf.String()) } fdBase := path.Join(w.proc.Root(), strconv.Itoa(pid), "fd") - fdNames, err := w.proc.ReadDir(fdBase) + fdNames, err := w.proc.ReadDirNames(fdBase) if err != nil { return err } @@ -71,7 +138,7 @@ func (w *procReader) Processes(f func(Process)) error { for _, fdName := range fdNames { var fdStat syscall.Stat_t // Direct use of syscall.Stat() to save garbage. - fdPath := path.Join(fdBase, fdName.Name()) + 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) @@ -97,7 +164,8 @@ var bufPool = sync.Pool{ }, } -func (w *procReader) Connections(withProcs bool, f func(Connection)) error { +// Connections walks through all the connections in the "/proc" +func (w *reader) Connections(withProcs bool, f func(Connection)) error { // create a map of inode->Process procs := make(map[uint64]Process) if withProcs { @@ -112,10 +180,14 @@ func (w *procReader) Connections(withProcs bool, f func(Connection)) error { 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) + for _, tcpFile := range tcpFiles { + err := w.handles.ReadInto(tcpFile, buf) + if err != nil { + return err + } + } - pn := NewProcNet(buf.Bytes(), tcpEstablished) + pn := newNetReader(buf.Bytes(), tcpEstablished) for { conn := pn.Next() if conn == nil { diff --git a/probe/proc/reader_linux_test.go b/probe/proc/reader_linux_test.go index 1974276b0..29402eb73 100644 --- a/probe/proc/reader_linux_test.go +++ b/probe/proc/reader_linux_test.go @@ -1,24 +1,37 @@ 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]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"}, + 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{ @@ -29,17 +42,16 @@ func TestProcReaderProcesses(t *testing.T) { } // 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) + procDir := mockedDir{ + ReadDirNamesFunc: func(path string) ([]string, error) { + result := []string{} + for k := range processes { + result = append(result, k) } return result, nil }, - - ReadFileFunc: func(path string) ([]byte, error) { - splits := strings.Split(path, "/") + OpenFunc: func(filename string) (proc.File, error) { + splits := strings.Split(filename, "/") pid := splits[len(splits)-2] process, ok := processes[pid] if !ok { @@ -47,22 +59,30 @@ func TestProcReaderProcesses(t *testing.T) { } file := splits[len(splits)-1] + var content []byte switch file { case "comm": - return []byte(process.Comm), nil + content = []byte(process.Comm) 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 + 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": - return []byte(process.Cmdline), nil + content = []byte(process.Cmdline) + default: + return nil, fmt.Errorf("not found") } - return nil, fmt.Errorf("not found") + return mockedFile{ + ReadIntoFunc: func(buf *bytes.Buffer) error { + _, err := buf.Write(content) + return err + }, + }, nil }, } - procReader := proc.NewProcReader(procDir) + procReader := proc.NewReader(procDir) have := map[int]proc.Process{} err := procReader.Processes(func(p proc.Process) { have[p.PID] = p diff --git a/probe/proc/reader_test.go b/probe/proc/reader_test.go index 8273268cd..9d78997d3 100644 --- a/probe/proc/reader_test.go +++ b/probe/proc/reader_test.go @@ -10,13 +10,19 @@ import ( func TestProcReaderBasic(t *testing.T) { procFunc := func(proc.Process) {} - if err := proc.NewProcReader(proc.EmptyProcDir).Processes(procFunc); err != nil { + 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.ProcReader) ([]proc.Process, error) { + all := func(w proc.Reader) ([]proc.Process, error) { all := []proc.Process{} err := w.Processes(func(p proc.Process) { all = append(all, p) @@ -30,7 +36,7 @@ func TestCachingProcReader(t *testing.T) { {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, {PID: 4, PPID: 2, Comm: "ping", Cmdline: "ping foo.bar.local"}, } - procReader := &proc.MockedProcReader{ + procReader := &proc.MockedReader{ Procs: processes, } cachingProcReader := proc.NewCachingProcReader(procReader, true) diff --git a/probe/proc/tree.go b/probe/proc/tree.go index 9a4206786..f9d48b959 100644 --- a/probe/proc/tree.go +++ b/probe/proc/tree.go @@ -14,7 +14,7 @@ type tree struct { } // NewTree returns a new Tree that can be polled. -func NewTree(walker ProcReader) (Tree, error) { +func NewTree(walker Reader) (Tree, error) { pt := tree{processes: map[int]Process{}} err := walker.Processes(func(p Process) { pt.processes[p.PID] = p diff --git a/probe/proc/tree_test.go b/probe/proc/tree_test.go index 92ec42c30..5377a9e92 100644 --- a/probe/proc/tree_test.go +++ b/probe/proc/tree_test.go @@ -8,7 +8,7 @@ import ( ) func TestTree(t *testing.T) { - walker := &proc.MockedProcReader{ + walker := &proc.MockedReader{ Procs: []proc.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, diff --git a/probe/process/reporter.go b/probe/process/reporter.go index 196d2d6dc..97fb8d67c 100644 --- a/probe/process/reporter.go +++ b/probe/process/reporter.go @@ -3,8 +3,8 @@ package process import ( "strconv" - "github.com/weaveworks/scope/report" "github.com/weaveworks/scope/probe/proc" + "github.com/weaveworks/scope/report" ) // We use these keys in node metadata @@ -19,11 +19,11 @@ const ( // Reporter generates Reports containing the Process topology. type Reporter struct { scope string - walker proc.ProcReader + walker proc.Reader } // NewReporter makes a new Reporter. -func NewReporter(walker proc.ProcReader, scope string) *Reporter { +func NewReporter(walker proc.Reader, scope string) *Reporter { return &Reporter{ scope: scope, walker: walker, diff --git a/probe/process/reporter_test.go b/probe/process/reporter_test.go index a0c84f4fa..0aac3e500 100644 --- a/probe/process/reporter_test.go +++ b/probe/process/reporter_test.go @@ -11,7 +11,7 @@ import ( ) func TestReporter(t *testing.T) { - procReader := &proc.MockedProcReader{ + procReader := &proc.MockedReader{ Procs: []proc.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, From 026d8e906516f9d6f596f0490b5b45b76a3893f8 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Thu, 10 Sep 2015 21:42:50 +0200 Subject: [PATCH 4/5] Minor: refactored readerLinux/readIntoBuffer() for removing a parameter --- probe/proc/reader_linux.go | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/probe/proc/reader_linux.go b/probe/proc/reader_linux.go index 8a27ec946..52dcba7bb 100644 --- a/probe/proc/reader_linux.go +++ b/probe/proc/reader_linux.go @@ -91,19 +91,20 @@ func (w *reader) Processes(f func(Process)) error { } buf := bytes.NewBuffer(make([]byte, 0, 5000)) - readIntoBuffer := func(filename string) error { - buf.Reset() - res := w.handles.ReadInto(filename, buf) - return res - } - for _, filename := range dirEntries { - pid, err := strconv.Atoi(filename) + for _, subdir := range dirEntries { + readIntoBuffer := func(filename string) error { + buf.Reset() + res := w.handles.ReadInto(path.Join(subdir, filename), buf) + return res + } + + pid, err := strconv.Atoi(subdir) if err != nil { continue } - if readIntoBuffer(path.Join(filename, "stat")) != nil { + if readIntoBuffer("stat") != nil { continue } splits := strings.Fields(buf.String()) @@ -118,13 +119,13 @@ func (w *reader) Processes(f func(Process)) error { } cmdline := "" - if readIntoBuffer(path.Join(filename, "cmdline")) == nil { + if readIntoBuffer("cmdline") == nil { cmdlineBuf := bytes.Replace(buf.Bytes(), []byte{'\000'}, []byte{' '}, -1) cmdline = string(cmdlineBuf) } comm := "(unknown)" - if readIntoBuffer(path.Join(filename, "comm")) == nil { + if readIntoBuffer("comm") == nil { comm = strings.TrimSpace(buf.String()) } From a7dc0ac206921f966a7eef2fb51d3d5cd5962d5f Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Mon, 14 Sep 2015 10:31:29 +0200 Subject: [PATCH 5/5] Use a mutex in the /proc files cache Some minor renames and cleanups. --- probe/proc/net.go | 1 + probe/proc/reader.go | 10 ++++++- probe/proc/reader_darwin.go | 4 +-- probe/proc/reader_linux.go | 59 ++++++++++++++++++------------------- 4 files changed, 41 insertions(+), 33 deletions(-) diff --git a/probe/proc/net.go b/probe/proc/net.go index 1d2f933f6..7e9ffbf4d 100644 --- a/probe/proc/net.go +++ b/probe/proc/net.go @@ -30,6 +30,7 @@ func (c Connection) Copy() Connection { c.LocalAddress = dupIP(c.LocalAddress) c.RemoteAddress = dupIP(c.RemoteAddress) + c.Process = c.Process.Copy() return c } diff --git a/probe/proc/reader.go b/probe/proc/reader.go index 6ef7c4a4b..5e43fd967 100644 --- a/probe/proc/reader.go +++ b/probe/proc/reader.go @@ -76,6 +76,14 @@ type Process struct { 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 { @@ -131,7 +139,7 @@ func (c *CachingProcReader) Close() error { return c.source.Close() } -// Update updates the cached copy of the processes and connections lists +// Tick updates the cached copy of the processes and connections lists func (c *CachingProcReader) Tick() error { newProcsCache := []Process{} newConnsCache := []Connection{} diff --git a/probe/proc/reader_darwin.go b/probe/proc/reader_darwin.go index 3e48f2730..c0546ebf7 100644 --- a/probe/proc/reader_darwin.go +++ b/probe/proc/reader_darwin.go @@ -45,7 +45,7 @@ func (reader) Processes(f func(Process)) error { return nil } -func (w *reader) Connections(withProcs bool, f func(Connection)) error { +func (r *reader) Connections(withProcs bool, f func(Connection)) error { out, err := exec.Command( netstatBinary, "-n", // no number resolving @@ -94,7 +94,7 @@ func (w *reader) Connections(withProcs bool, f func(Connection)) error { } // Close closes the Darwin "/proc" reader -func (w *reader) Close() error { +func (reader) Close() error { return nil } diff --git a/probe/proc/reader_linux.go b/probe/proc/reader_linux.go index 52dcba7bb..dcd322f60 100644 --- a/probe/proc/reader_linux.go +++ b/probe/proc/reader_linux.go @@ -23,26 +23,32 @@ var tcpFiles = []string{ } // A cache for files handles -// Note: not intended to be used from multiple goroutines type filesCache struct { handles gcache.Cache } +type filesCacheEntry struct { + sync.RWMutex + File +} + func newFilesCache(proc Dir) *filesCache { loadFunc := func(fileName interface{}) (interface{}, error) { - return proc.Open(fileName.(string)) + f, err := proc.Open(fileName.(string)) + if err != nil { + return nil, err + } + return filesCacheEntry{File: f}, nil } evictionFunc := func(key, value interface{}) { - value.(File).Close() + value.(filesCacheEntry).Close() } return &filesCache{ handles: gcache.New(filesCacheLen).LoaderFunc(loadFunc).EvictedFunc(evictionFunc).Expiration(filesCacheExpiration).ARC().Build(), } } -// Read a "/proc" file, identified as a subdir (eg "1134/comm"), into a buffer -// Note: this is not goroutine-safe: two goroutines getting and reading from -// the same handle can obtain some unexpected contents... +// 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... @@ -50,7 +56,10 @@ func (fc *filesCache) ReadInto(filename string, buf *bytes.Buffer) error { if err != nil { return err } - return h.(File).ReadInto(buf) + handle := h.(filesCacheEntry) + handle.Lock() + defer handle.Unlock() + return handle.ReadInto(buf) } // Close closes all the handles in the cache @@ -76,27 +85,27 @@ func NewReader(proc Dir) Reader { } // Close closes the Linux "/proc" reader -func (w *reader) Close() error { - return w.handles.Close() +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 (w *reader) Processes(f func(Process)) error { - dirEntries, err := w.proc.ReadDirNames(w.proc.Root()) +func (r *reader) Processes(f func(Process)) error { + dirEntries, err := r.proc.ReadDirNames(r.proc.Root()) if err != nil { return err } - buf := bytes.NewBuffer(make([]byte, 0, 5000)) + var fdStat syscall.Stat_t + buf := bytes.Buffer{} for _, subdir := range dirEntries { readIntoBuffer := func(filename string) error { buf.Reset() - res := w.handles.ReadInto(path.Join(subdir, filename), buf) - return res + return r.handles.ReadInto(path.Join(subdir, filename), &buf) } pid, err := strconv.Atoi(subdir) @@ -129,15 +138,14 @@ func (w *reader) Processes(f func(Process)) error { comm = strings.TrimSpace(buf.String()) } - fdBase := path.Join(w.proc.Root(), strconv.Itoa(pid), "fd") - fdNames, err := w.proc.ReadDirNames(fdBase) + 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 { - var fdStat syscall.Stat_t // Direct use of syscall.Stat() to save garbage. fdPath := path.Join(fdBase, fdName) err = syscall.Stat(fdPath, &fdStat) @@ -159,30 +167,21 @@ func (w *reader) Processes(f func(Process)) error { return nil } -var bufPool = sync.Pool{ - New: func() interface{} { - return bytes.NewBuffer(make([]byte, 0, 5000)) - }, -} - // Connections walks through all the connections in the "/proc" -func (w *reader) Connections(withProcs bool, f func(Connection)) error { +func (r *reader) 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) { + r.Processes(func(p Process) { for _, inode := range p.Inodes { procs[inode] = p } }) } - buf := bufPool.Get().(*bytes.Buffer) - buf.Reset() - defer bufPool.Put(buf) - + buf := bytes.Buffer{} for _, tcpFile := range tcpFiles { - err := w.handles.ReadInto(tcpFile, buf) + err := r.handles.ReadInto(tcpFile, &buf) if err != nil { return err }