diff --git a/probe/docker/tagger.go b/probe/docker/tagger.go index 1ccecadb8..0c8d7cd57 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.Reader } // NewTagger returns a usable Tagger. -func NewTagger(registry Registry, procWalker process.Walker) *Tagger { +func NewTagger(registry Registry, procWalker proc.Reader) *Tagger { return &Tagger{ registry: registry, procWalker: procWalker, @@ -44,7 +45,7 @@ func (t *Tagger) Tag(r report.Report) (report.Report, error) { return r, nil } -func (t *Tagger) tag(tree process.Tree, topology *report.Topology) { +func (t *Tagger) tag(tree proc.Tree, topology *report.Topology) { for nodeID, nodeMetadata := range topology.Nodes { pidStr, ok := nodeMetadata.Metadata[process.PID] if !ok { diff --git a/probe/docker/tagger_test.go b/probe/docker/tagger_test.go index 261fd4215..6fc8b697f 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.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 6df5eb229..2689700b1 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" ) @@ -28,6 +28,7 @@ type Reporter struct { conntracker *Conntracker natmapper *natmapper revResolver *ReverseResolver + procReader proc.Reader } // SpyDuration is an exported prometheus metric @@ -42,12 +43,12 @@ var SpyDuration = prometheus.NewSummaryVec( []string{}, ) -// NewReporter creates a new Reporter that invokes procspy.Connections to +// NewReporter creates a new Reporter that invokes Connections to // generate a report.Report that contains every discovered (spied) connection // on the host machine, at the granularity of host and port. That information // is stored in the Endpoint topology. It optionally enriches that topology // with process (PID) information. -func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool) *Reporter { +func NewReporter(hostID, hostName string, includeProcesses bool, procReader proc.Reader, useConntrack bool) *Reporter { var ( conntrackModulePresent = ConntrackModulePresent() conntracker *Conntracker @@ -73,6 +74,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo conntracker: conntracker, natmapper: natmapper, revResolver: NewReverseResolver(), + procReader: procReader, } } @@ -94,12 +96,7 @@ func (r *Reporter) Report() (report.Report, error) { }(time.Now()) rpt := report.MakeReport() - conns, err := procspy.Connections(r.includeProcesses) - if err != nil { - return rpt, err - } - - for conn := conns.Next(); conn != nil; conn = conns.Next() { + err := r.procReader.Connections(r.includeProcesses, func(conn proc.Connection) { var ( localPort = conn.LocalPort remotePort = conn.RemotePort @@ -107,12 +104,15 @@ func (r *Reporter) Report() (report.Report, error) { remoteAddr = conn.RemoteAddress.String() ) extraNodeInfo := report.MakeNode() - if conn.Proc.PID > 0 { + if conn.Process.PID > 0 { extraNodeInfo = extraNodeInfo.WithMetadata(report.Metadata{ - process.PID: strconv.FormatUint(uint64(conn.Proc.PID), 10), + process.PID: strconv.FormatUint(uint64(conn.Process.PID), 10), }) } r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &extraNodeInfo, nil) + }) + if err != nil { + return rpt, err } if r.conntracker != nil { diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index c7ec41995..af746048a 100644 --- a/probe/endpoint/reporter_test.go +++ b/probe/endpoint/reporter_test.go @@ -5,9 +5,9 @@ import ( "strconv" "testing" - "github.com/weaveworks/procspy" "github.com/weaveworks/scope/probe/docker" "github.com/weaveworks/scope/probe/endpoint" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/report" ) @@ -17,10 +17,10 @@ var ( fixRemoteAddress = net.ParseIP("192.168.1.2") fixRemotePort = uint16(12345) fixRemotePortB = uint16(12346) - fixProcessPID = uint(4242) + fixProcessPID = int(4242) fixProcessName = "nginx" - fixConnections = []procspy.Connection{ + fixConnections = []proc.Connection{ { Transport: "tcp", LocalAddress: fixLocalAddress, @@ -37,16 +37,16 @@ var ( }, } - fixConnectionsWithProcesses = []procspy.Connection{ + fixConnectionsWithProcesses = []proc.Connection{ { Transport: "tcp", LocalAddress: fixLocalAddress, LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, RemotePort: fixRemotePort, - Proc: procspy.Proc{ + Process: proc.Process{ PID: fixProcessPID, - Name: fixProcessName, + Comm: fixProcessName, }, }, { @@ -55,23 +55,22 @@ var ( LocalPort: fixLocalPort, RemoteAddress: fixRemoteAddress, RemotePort: fixRemotePort, - Proc: procspy.Proc{ + Process: proc.Process{ PID: fixProcessPID, - Name: fixProcessName, + Comm: fixProcessName, }, }, } ) func TestSpyNoProcesses(t *testing.T) { - procspy.SetFixtures(fixConnections) - const ( nodeID = "heinz-tomato-ketchup" // TODO rename to hostID nodeName = "frenchs-since-1904" // TODO rename to hostNmae ) - reporter := endpoint.NewReporter(nodeID, nodeName, false, false) + procReader := proc.MockedReader{Conns: fixConnections} + reporter := endpoint.NewReporter(nodeID, nodeName, false, &procReader, false) r, _ := reporter.Report() //buf, _ := json.MarshalIndent(r, "", " ") //t.Logf("\n%s\n", buf) @@ -100,14 +99,13 @@ func TestSpyNoProcesses(t *testing.T) { } func TestSpyWithProcesses(t *testing.T) { - procspy.SetFixtures(fixConnectionsWithProcesses) - const ( nodeID = "nikon" // TODO rename to hostID nodeName = "fishermans-friend" // TODO rename to hostNmae ) - reporter := endpoint.NewReporter(nodeID, nodeName, true, false) + procReader := proc.MockedReader{Conns: fixConnectionsWithProcesses} + reporter := endpoint.NewReporter(nodeID, nodeName, true, &procReader, false) r, _ := reporter.Report() // buf, _ := json.MarshalIndent(r, "", " ") ; t.Logf("\n%s\n", buf) diff --git a/probe/main.go b/probe/main.go index eb4495ff9..33afccf39 100644 --- a/probe/main.go +++ b/probe/main.go @@ -14,11 +14,11 @@ import ( "syscall" "time" - "github.com/weaveworks/procspy" "github.com/weaveworks/scope/probe/docker" "github.com/weaveworks/scope/probe/endpoint" "github.com/weaveworks/scope/probe/host" "github.com/weaveworks/scope/probe/overlay" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/probe/process" "github.com/weaveworks/scope/probe/sniff" "github.com/weaveworks/scope/report" @@ -67,8 +67,6 @@ func main() { } log.Printf("publishing to: %s", strings.Join(targets, ", ")) - procspy.SetProcRoot(*procRoot) - if *httpListen != "" { log.Printf("profiling data being exported to %s", *httpListen) log.Printf("go tool pprof http://%s/debug/pprof/{profile,heap,block}", *httpListen) @@ -111,16 +109,18 @@ func main() { } var ( - endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack) - processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) - tickers = []Ticker{processCache} + procDir = proc.OSDir{Dir: *procRoot} + procReader = proc.NewCachingProcReader(proc.NewReader(procDir), *spyProcs) + tickers = []Ticker{procReader} + endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, procReader, *useConntrack) reporters = []Reporter{ endpointReporter, host.NewReporter(hostID, hostName, localNets), - process.NewReporter(processCache, hostID), + process.NewReporter(procReader, hostID), } taggers = []Tagger{newTopologyTagger(), host.NewTagger(hostID)} ) + defer procReader.Close() defer endpointReporter.Stop() if *dockerEnabled { @@ -134,7 +134,7 @@ func main() { } defer dockerRegistry.Stop() - taggers = append(taggers, docker.NewTagger(dockerRegistry, processCache)) + taggers = append(taggers, docker.NewTagger(dockerRegistry, procReader)) reporters = append(reporters, docker.NewReporter(dockerRegistry, hostID)) } diff --git a/probe/proc/mocks.go b/probe/proc/mocks.go new file mode 100644 index 000000000..f31d2b4b2 --- /dev/null +++ b/probe/proc/mocks.go @@ -0,0 +1,30 @@ +package proc + +// note: we must keep this in the "proc" package so it can be used from other packages.. + +// MockedReader is a mocked "/proc" reader +type MockedReader struct { + Procs []Process + Conns []Connection +} + +// Processes walks through the processes provided in the mocked "/proc" reader +func (mw MockedReader) Processes(f func(Process)) error { + for _, p := range mw.Procs { + f(p) + } + return nil +} + +// Connections walks through the connections provided in the mocked "/proc" reader +func (mw *MockedReader) Connections(_ bool, f func(Connection)) error { + for _, c := range mw.Conns { + f(c) + } + return nil +} + +// Close (mocked version) +func (mw *MockedReader) Close() error { + return nil +} 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.go b/probe/proc/net.go new file mode 100644 index 000000000..7e9ffbf4d --- /dev/null +++ b/probe/proc/net.go @@ -0,0 +1,43 @@ +package proc + +import ( + "fmt" + "net" +) + +const ( + tcpEstablished = 1 // according to /include/net/tcp_states.h +) + +// Connection is a (TCP) connection. The 'Process' struct might not be filled in. +type Connection struct { + Transport string + LocalAddress net.IP + LocalPort uint16 + RemoteAddress net.IP + RemotePort uint16 + inode uint64 + Process +} + +// Copy returns a copy of a connection +func (c Connection) Copy() Connection { + dupIP := func(ip net.IP) net.IP { + dup := make(net.IP, len(ip)) + copy(dup, ip) + return dup + } + + c.LocalAddress = dupIP(c.LocalAddress) + c.RemoteAddress = dupIP(c.RemoteAddress) + c.Process = c.Process.Copy() + return c +} + +// String returns the string repr +func (c Connection) String() string { + return fmt.Sprintf("%s:%d - %s:%d %s#%d", + c.LocalAddress, c.LocalPort, + c.RemoteAddress, c.RemotePort, + c.Transport, c.inode) +} 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/net_linux.go b/probe/proc/net_linux.go new file mode 100644 index 000000000..e9b7de05b --- /dev/null +++ b/probe/proc/net_linux.go @@ -0,0 +1,162 @@ +package proc + +import ( + "bytes" + "net" +) + +type netReader struct { + b []byte + c Connection + wantedState uint + bytesLocal, bytesRemote [16]byte +} + +func newNetReader(b []byte, wantedState uint) *netReader { + return &netReader{ + b: b, + c: Connection{}, + wantedState: wantedState, + } +} + +// Next returns the next connection. All buffers are re-used, so if you want +// to keep the IPs you have to copy them. +func (p *netReader) Next() *Connection { +again: + if len(p.b) == 0 { + return nil + } + b := p.b + + if p.b[2] == 's' { + // Skip header + p.b = nextLine(b) + goto again + } + + var ( + local, remote, state, inode []byte + ) + _, b = nextField(b) // 'sl' column + local, b = nextField(b) + remote, b = nextField(b) + state, b = nextField(b) + if parseHex(state) != p.wantedState { + p.b = nextLine(b) + goto again + } + _, b = nextField(b) // 'tx_queue' column + _, b = nextField(b) // 'rx_queue' column + _, b = nextField(b) // 'tr' column + _, b = nextField(b) // 'uid' column + _, b = nextField(b) // 'timeout' column + inode, b = nextField(b) + + p.c.LocalAddress, p.c.LocalPort = scanAddressNA(local, &p.bytesLocal) + p.c.RemoteAddress, p.c.RemotePort = scanAddressNA(remote, &p.bytesRemote) + p.c.inode = parseDec(inode) + p.b = nextLine(b) + return &p.c +} + +// scanAddressNA parses 'A12CF62E:00AA' to the address/port. Handles IPv4 and +// IPv6 addresses. The address is a big endian 32 bit ints, hex encoded. We +// just decode the hex and flip the bytes in every group of 4. +func scanAddressNA(in []byte, buf *[16]byte) (net.IP, uint16) { + col := bytes.IndexByte(in, ':') + if col == -1 { + return nil, 0 + } + + // Network address is big endian. Can be either ipv4 or ipv6. + address := hexDecode32bigNA(in[:col], buf) + return net.IP(address), uint16(parseHex(in[col+1:])) +} + +// hexDecode32big decodes sequences of 32bit big endian bytes. +func hexDecode32bigNA(src []byte, buf *[16]byte) []byte { + blocks := len(src) / 8 + for block := 0; block < blocks; block++ { + for i := 0; i < 4; i++ { + a := fromHexChar(src[block*8+i*2]) + b := fromHexChar(src[block*8+i*2+1]) + buf[block*4+3-i] = (a << 4) | b + } + } + return buf[:blocks*4] +} + +func nextField(s []byte) ([]byte, []byte) { + // Skip whitespace. + for i, b := range s { + if b != ' ' { + s = s[i:] + break + } + } + + // Up until the next whitespace field. + for i, b := range s { + if b == ' ' { + return s[:i], s[i:] + } + } + + return nil, nil +} + +func nextLine(s []byte) []byte { + i := bytes.IndexByte(s, '\n') + if i == -1 { + return nil + } + return s[i+1:] +} + +// Simplified copy of strconv.ParseUint(16). +func parseHex(s []byte) uint { + n := uint(0) + for i := 0; i < len(s); i++ { + n *= 16 + n += uint(fromHexChar(s[i])) + } + return n +} + +// Simplified copy of strconv.ParseUint(10). +func parseDec(s []byte) uint64 { + n := uint64(0) + for _, c := range s { + n *= 10 + n += uint64(c - '0') + } + return n +} + +// hexDecode32big decodes sequences of 32bit big endian bytes. +func hexDecode32big(src []byte) []byte { + dst := make([]byte, len(src)/2) + blocks := len(src) / 8 + for block := 0; block < blocks; block++ { + for i := 0; i < 4; i++ { + a := fromHexChar(src[block*8+i*2]) + b := fromHexChar(src[block*8+i*2+1]) + dst[block*4+3-i] = (a << 4) | b + } + } + return dst +} + +// fromHexChar converts a hex character into its value. +func fromHexChar(c byte) uint8 { + switch { + case '0' <= c && c <= '9': + return c - '0' + case 'a' <= c && c <= 'f': + return c - 'a' + 10 + case 'A' <= c && c <= 'F': + return c - 'A' + 10 + } + return 0 +} diff --git a/probe/proc/reader.go b/probe/proc/reader.go new file mode 100644 index 000000000..5e43fd967 --- /dev/null +++ b/probe/proc/reader.go @@ -0,0 +1,165 @@ +package proc + +import ( + "bytes" + "os" + "path" + "sync" +) + +// File is a file in the "/proc" directory +type File interface { + ReadInto(buf *bytes.Buffer) error + Close() error +} + +// OSFile is a native file +type OSFile struct{ *os.File } + +// ReadInto reads the whole file into a buffer +func (of *OSFile) ReadInto(buf *bytes.Buffer) error { + if _, err := of.File.Seek(0, 0); err != nil { + return err + } + _, err := buf.ReadFrom(of.File) + return err +} + +// Dir is the '/proc' directory and the associated ops for +// reading subdirs or files. +type Dir interface { + Root() string // the "/proc" directory + Open(s string) (File, error) // open a file in the "/proc" dir + ReadDirNames(string) ([]string, error) // list a subdirectory in the "/proc" +} + +// OSDir is a OS "/proc" firectory +type OSDir struct{ Dir string } + +// Root returns the "/proc" top dir +func (dp OSDir) Root() string { + return dp.Dir +} + +// Open returns a ProcFile +func (dp OSDir) Open(s string) (File, error) { + h, err := os.Open(path.Join(dp.Root(), s)) + if err != nil { + return nil, err + } + return &OSFile{h}, nil +} + +// ReadDirNames reads all the directory entries +func (dp OSDir) ReadDirNames(s string) ([]string, error) { + f, err := os.Open(s) + if err != nil { + return nil, err + } + list, err := f.Readdirnames(-1) + f.Close() + if err != nil { + return nil, err + } + return list, nil +} + +// DefaultProcDir is the default '/proc' directory +var DefaultProcDir = OSDir{Dir: "/proc"} + +// Process represents a single process. +type Process struct { + PID, PPID int + Comm string + Cmdline string + Threads int + Inodes []uint64 +} + +// Copy returns a copy of a process +func (p Process) Copy() Process { + dup := make([]uint64, len(p.Inodes)) + copy(dup, p.Inodes) + p.Inodes = dup + return p +} + +// Reader is something that reads the /proc directory and +// returns some info like processes and connections +type Reader interface { + // Processes walks through the processes + Processes(func(Process)) error + // Connections walks through the connections + Connections(bool, func(Connection)) error + // Close closes the "/proc" reader + Close() error +} + +// CachingProcReader is a '/proc' reader than caches a copy of the output from another +// '/proc' reader, and then allows other concurrent readers to Walk that copy. +type CachingProcReader struct { + procsCache []Process + connsCache []Connection + source Reader + includeProcs bool + sync.RWMutex +} + +// NewCachingProcReader returns a new CachingProcReader +func NewCachingProcReader(source Reader, includeProcs bool) *CachingProcReader { + return &CachingProcReader{source: source, includeProcs: includeProcs} +} + +// Processes walks a cached copy of process list +func (c *CachingProcReader) Processes(f func(Process)) error { + c.RLock() + defer c.RUnlock() + + for _, p := range c.procsCache { + f(p) + } + return nil +} + +// Connections walks a cached copy of the connections list +// Note: specifying 'includeProcs' has no effect here, as the cached copy +func (c *CachingProcReader) Connections(_ bool, f func(Connection)) error { + c.RLock() + defer c.RUnlock() + + for _, c := range c.connsCache { + f(c) + } + return nil + +} + +// Close closes the "/proc" reader +func (c *CachingProcReader) Close() error { + return c.source.Close() +} + +// Tick updates the cached copy of the processes and connections lists +func (c *CachingProcReader) Tick() error { + newProcsCache := []Process{} + newConnsCache := []Connection{} + + if err := c.source.Processes(func(p Process) { + newProcsCache = append(newProcsCache, p) + }); err != nil { + return err + } + + if err := c.source.Connections(c.includeProcs, func(conn Connection) { + newConnsCache = append(newConnsCache, conn.Copy()) + }); err != nil { + return err + } + + c.Lock() + defer c.Unlock() + c.procsCache = newProcsCache + c.connsCache = newConnsCache + + return nil +} diff --git a/probe/process/walker_darwin.go b/probe/proc/reader_darwin.go similarity index 51% rename from probe/process/walker_darwin.go rename to probe/proc/reader_darwin.go index cd25b3df3..c0546ebf7 100644 --- a/probe/process/walker_darwin.go +++ b/probe/proc/reader_darwin.go @@ -1,27 +1,28 @@ -package process +package proc import ( "fmt" + "net" "os/exec" "strconv" "strings" ) -// NewWalker returns a Darwin (lsof-based) walker. -func NewWalker(_ string) Walker { - return &walker{} +type reader struct{} + +// NewReader returns a Darwin (lsof-based) '/proc' reader +func NewReader(proc Dir) Reader { + return &reader{} } -type walker struct{} - const ( - lsofBinary = "lsof" - lsofFields = "cn" // parseLSOF() depends on the order + lsofBinary = "lsof" + lsofFields = "cn" // parseLSOF() depends on the order + netstatBinary = "netstat" + lsofBinary = "lsof" ) -// These functions copied from procspy. - -func (walker) Walk(f func(Process)) error { +func (reader) Processes(f func(Process)) error { output, err := exec.Command( lsofBinary, "-i", // only Internet files @@ -44,6 +45,59 @@ func (walker) Walk(f func(Process)) error { return nil } +func (r *reader) Connections(withProcs bool, f func(Connection)) error { + out, err := exec.Command( + netstatBinary, + "-n", // no number resolving + "-W", // Wide output + // "-l", // full IPv6 addresses // What does this do? + "-p", "tcp", // only TCP + ).CombinedOutput() + if err != nil { + return err + } + connections := parseDarwinNetstat(string(out)) + + if withProcs { + out, err := exec.Command( + lsofBinary, + "-i", // only Internet files + "-n", "-P", // no number resolving + "-w", // no warnings + "-F", lsofFields, // \n based output of only the fields we want. + ).CombinedOutput() + if err != nil { + return err + } + + procs, err := parseLSOF(string(out)) + if err != nil { + return err + } + for local, proc := range procs { + for i, c := range connections { + localAddr := net.JoinHostPort( + c.LocalAddress.String(), + strconv.Itoa(int(c.LocalPort)), + ) + if localAddr == local { + connections[i].Proc = proc + } + } + } + } + + for _, c := range connections { + f(c) + } + return nil +} + +// Close closes the Darwin "/proc" reader +func (reader) Close() error { + return nil +} + func parseLSOF(output string) (map[string]Process, error) { var ( processes = map[string]Process{} // Local addr -> Proc diff --git a/probe/proc/reader_linux.go b/probe/proc/reader_linux.go new file mode 100644 index 000000000..dcd322f60 --- /dev/null +++ b/probe/proc/reader_linux.go @@ -0,0 +1,202 @@ +package proc + +import ( + "bytes" + "path" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "github.com/bluele/gcache" +) + +const ( + filesCacheLen = 512 + filesCacheExpiration = 60 * time.Second +) + +var tcpFiles = []string{ + "net/tcp", + "net/tcp6", +} + +// A cache for files handles +type filesCache struct { + handles gcache.Cache +} + +type filesCacheEntry struct { + sync.RWMutex + File +} + +func newFilesCache(proc Dir) *filesCache { + loadFunc := func(fileName interface{}) (interface{}, error) { + f, err := proc.Open(fileName.(string)) + if err != nil { + return nil, err + } + return filesCacheEntry{File: f}, nil + } + evictionFunc := func(key, value interface{}) { + value.(filesCacheEntry).Close() + } + return &filesCache{ + handles: gcache.New(filesCacheLen).LoaderFunc(loadFunc).EvictedFunc(evictionFunc).Expiration(filesCacheExpiration).ARC().Build(), + } +} + +// Read a "/proc" file, identified as a file in a subdir (eg "1134/comm"), into a buffer +func (fc *filesCache) ReadInto(filename string, buf *bytes.Buffer) error { + // we could use a lock here, but this is only used from Processes()/Connections(), + // and they are always invoked sequentially... + h, err := fc.handles.Get(filename) + if err != nil { + return err + } + handle := h.(filesCacheEntry) + handle.Lock() + defer handle.Unlock() + return handle.ReadInto(buf) +} + +// Close closes all the handles in the cache +func (fc *filesCache) Close() error { + for _, key := range fc.handles.Keys() { + fc.handles.Remove(key) + } + return nil +} + +// the Linux "/proc" reader +type reader struct { + proc Dir + handles *filesCache +} + +// NewReader creates a new /proc reader. +func NewReader(proc Dir) Reader { + return &reader{ + proc: proc, + handles: newFilesCache(proc), + } +} + +// Close closes the Linux "/proc" reader +func (r *reader) Close() error { + return r.handles.Close() +} + +// Processes walks the /proc directory and marshalls the files into +// instances of Process, which it then passes one-by-one to the +// supplied function. Processes() is only made public so that is +// can be tested. +func (r *reader) Processes(f func(Process)) error { + dirEntries, err := r.proc.ReadDirNames(r.proc.Root()) + if err != nil { + return err + } + + var fdStat syscall.Stat_t + buf := bytes.Buffer{} + + for _, subdir := range dirEntries { + readIntoBuffer := func(filename string) error { + buf.Reset() + return r.handles.ReadInto(path.Join(subdir, filename), &buf) + } + + pid, err := strconv.Atoi(subdir) + if err != nil { + continue + } + + if readIntoBuffer("stat") != nil { + continue + } + splits := strings.Fields(buf.String()) + ppid, err := strconv.Atoi(splits[3]) + if err != nil { + return err + } + + threads, err := strconv.Atoi(splits[19]) + if err != nil { + return err + } + + cmdline := "" + if readIntoBuffer("cmdline") == nil { + cmdlineBuf := bytes.Replace(buf.Bytes(), []byte{'\000'}, []byte{' '}, -1) + cmdline = string(cmdlineBuf) + } + + comm := "(unknown)" + if readIntoBuffer("comm") == nil { + comm = strings.TrimSpace(buf.String()) + } + + fdBase := path.Join(r.proc.Root(), strconv.Itoa(pid), "fd") + fdNames, err := r.proc.ReadDirNames(fdBase) + if err != nil { + return err + } + + inodes := []uint64{} + for _, fdName := range fdNames { + // Direct use of syscall.Stat() to save garbage. + fdPath := path.Join(fdBase, fdName) + err = syscall.Stat(fdPath, &fdStat) + if err == nil && (fdStat.Mode&syscall.S_IFMT == syscall.S_IFSOCK) { // We want sockets only. + inodes = append(inodes, fdStat.Ino) + } + } + + f(Process{ + PID: pid, + PPID: ppid, + Comm: comm, + Cmdline: cmdline, + Threads: threads, + Inodes: inodes, + }) + } + + return nil +} + +// Connections walks through all the connections in the "/proc" +func (r *reader) Connections(withProcs bool, f func(Connection)) error { + // create a map of inode->Process + procs := make(map[uint64]Process) + if withProcs { + r.Processes(func(p Process) { + for _, inode := range p.Inodes { + procs[inode] = p + } + }) + } + + buf := bytes.Buffer{} + for _, tcpFile := range tcpFiles { + err := r.handles.ReadInto(tcpFile, &buf) + if err != nil { + return err + } + } + + pn := newNetReader(buf.Bytes(), tcpEstablished) + for { + conn := pn.Next() + if conn == nil { + break // Done! + } + if proc, ok := procs[conn.inode]; ok { + conn.Process = proc + } + f(*conn) + } + return nil +} diff --git a/probe/proc/reader_linux_test.go b/probe/proc/reader_linux_test.go new file mode 100644 index 000000000..29402eb73 --- /dev/null +++ b/probe/proc/reader_linux_test.go @@ -0,0 +1,93 @@ +package proc_test + +import ( + "bytes" + "fmt" + "os" + "reflect" + "strconv" + "strings" + "testing" + "time" + + "github.com/weaveworks/scope/probe/proc" + "github.com/weaveworks/scope/test" +) + +type mockedProcess struct { + ID, Comm, Cmdline string +} + +func (p mockedProcess) Name() string { return p.ID } +func (p mockedProcess) Size() int64 { return 0 } +func (p mockedProcess) Mode() os.FileMode { return 0 } +func (p mockedProcess) ModTime() time.Time { return time.Now() } +func (p mockedProcess) IsDir() bool { return true } +func (p mockedProcess) Sys() interface{} { return nil } + +func TestProcReaderProcesses(t *testing.T) { + processes := map[string]mockedProcess{ + "3": {ID: "3", Comm: "curl\n", Cmdline: "curl\000google.com"}, + "2": {ID: "2", Comm: "bash\n"}, + "4": {ID: "4", Comm: "apache\n"}, + "notapid": {ID: "notapid"}, + "1": {ID: "1", Comm: "init\n"}, + } + + want := map[int]proc.Process{ + 3: {PID: 3, PPID: 2, Comm: "curl", Cmdline: "curl google.com", Threads: 1, Inodes: []uint64{}}, + 2: {PID: 2, PPID: 1, Comm: "bash", Cmdline: "", Threads: 1, Inodes: []uint64{}}, + 4: {PID: 4, PPID: 3, Comm: "apache", Cmdline: "", Threads: 1, Inodes: []uint64{}}, + 1: {PID: 1, PPID: 0, Comm: "init", Cmdline: "", Threads: 1, Inodes: []uint64{}}, + } + + // use a mocked /proc that reads from our mocked processes + procDir := mockedDir{ + ReadDirNamesFunc: func(path string) ([]string, error) { + result := []string{} + for k := range processes { + result = append(result, k) + } + return result, nil + }, + OpenFunc: func(filename string) (proc.File, error) { + splits := strings.Split(filename, "/") + pid := splits[len(splits)-2] + process, ok := processes[pid] + if !ok { + return nil, fmt.Errorf("not found") + } + + file := splits[len(splits)-1] + var content []byte + switch file { + case "comm": + content = []byte(process.Comm) + case "stat": + pid, _ := strconv.Atoi(splits[len(splits)-2]) + parent := pid - 1 + content = []byte(fmt.Sprintf("%d na R %d 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1", pid, parent)) + case "cmdline": + content = []byte(process.Cmdline) + default: + return nil, fmt.Errorf("not found") + } + + return mockedFile{ + ReadIntoFunc: func(buf *bytes.Buffer) error { + _, err := buf.Write(content) + return err + }, + }, nil + }, + } + + procReader := proc.NewReader(procDir) + have := map[int]proc.Process{} + err := procReader.Processes(func(p proc.Process) { + have[p.PID] = p + }) + if err != nil || !reflect.DeepEqual(want, have) { + t.Errorf("%v (%v)", test.Diff(want, have), err) + } +} diff --git a/probe/proc/reader_test.go b/probe/proc/reader_test.go new file mode 100644 index 000000000..9d78997d3 --- /dev/null +++ b/probe/proc/reader_test.go @@ -0,0 +1,69 @@ +package proc_test + +import ( + "reflect" + "testing" + + "github.com/weaveworks/scope/probe/proc" + "github.com/weaveworks/scope/test" +) + +func TestProcReaderBasic(t *testing.T) { + procFunc := func(proc.Process) {} + nullProcDir := mockedDir{ + Dir: "", + OpenFunc: func(string) (proc.File, error) { return &proc.OSFile{File: nil}, nil }, + ReadDirNamesFunc: func(string) ([]string, error) { return []string{}, nil }, + } + + if err := proc.NewReader(nullProcDir).Processes(procFunc); err != nil { + t.Fatal(err) + } +} + +func TestCachingProcReader(t *testing.T) { + all := func(w proc.Reader) ([]proc.Process, error) { + all := []proc.Process{} + err := w.Processes(func(p proc.Process) { + all = append(all, p) + }) + return all, err + } + + processes := []proc.Process{ + {PID: 1, PPID: 0, Comm: "init"}, + {PID: 2, PPID: 1, Comm: "bash"}, + {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, + {PID: 4, PPID: 2, Comm: "ping", Cmdline: "ping foo.bar.local"}, + } + procReader := &proc.MockedReader{ + Procs: processes, + } + cachingProcReader := proc.NewCachingProcReader(procReader, true) + err := cachingProcReader.Tick() + if err != nil { + t.Fatal(err) + } + + have, err := all(cachingProcReader) + if err != nil || !reflect.DeepEqual(processes, have) { + t.Errorf("%v (%v)", test.Diff(processes, have), err) + } + + procReader.Procs = []proc.Process{} + have, err = all(cachingProcReader) + if err != nil || !reflect.DeepEqual(processes, have) { + t.Errorf("%v (%v)", test.Diff(processes, have), err) + } + + err = cachingProcReader.Tick() + if err != nil { + t.Fatal(err) + } + + have, err = all(cachingProcReader) + want := []proc.Process{} + if err != nil || !reflect.DeepEqual(want, have) { + t.Errorf("%v (%v)", test.Diff(want, have), err) + } +} diff --git a/probe/process/tree.go b/probe/proc/tree.go similarity index 84% rename from probe/process/tree.go rename to probe/proc/tree.go index 2d03a9c0f..f9d48b959 100644 --- a/probe/process/tree.go +++ b/probe/proc/tree.go @@ -1,4 +1,4 @@ -package process +package proc import ( "fmt" @@ -14,9 +14,9 @@ type tree struct { } // NewTree returns a new Tree that can be polled. -func NewTree(walker Walker) (Tree, error) { +func NewTree(walker Reader) (Tree, error) { pt := tree{processes: map[int]Process{}} - err := walker.Walk(func(p Process) { + err := walker.Processes(func(p Process) { pt.processes[p.PID] = p }) diff --git a/probe/process/tree_test.go b/probe/proc/tree_test.go similarity index 77% rename from probe/process/tree_test.go rename to probe/proc/tree_test.go index 6a93cc226..5377a9e92 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{ + walker := &proc.MockedReader{ + Procs: []proc.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, @@ -17,7 +17,7 @@ func TestTree(t *testing.T) { }, } - tree, err := process.NewTree(walker) + tree, err := proc.NewTree(walker) if err != nil { t.Fatalf("newProcessTree error: %v", err) } diff --git a/probe/process/reporter.go b/probe/process/reporter.go index 3d46de07d..97fb8d67c 100644 --- a/probe/process/reporter.go +++ b/probe/process/reporter.go @@ -3,6 +3,7 @@ package process import ( "strconv" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/report" ) @@ -18,11 +19,11 @@ const ( // Reporter generates Reports containing the Process topology. type Reporter struct { scope string - walker Walker + walker proc.Reader } // NewReporter makes a new Reporter. -func NewReporter(walker Walker, scope string) *Reporter { +func NewReporter(walker proc.Reader, scope string) *Reporter { return &Reporter{ scope: scope, walker: walker, @@ -42,7 +43,7 @@ func (r *Reporter) Report() (report.Report, error) { func (r *Reporter) processTopology() (report.Topology, error) { t := report.MakeTopology() - err := r.walker.Walk(func(p Process) { + err := r.walker.Processes(func(p proc.Process) { pidstr := strconv.Itoa(p.PID) nodeID := report.MakeProcessNodeID(r.scope, pidstr) t.Nodes[nodeID] = report.MakeNode() diff --git a/probe/process/reporter_test.go b/probe/process/reporter_test.go index d13b2611f..0aac3e500 100644 --- a/probe/process/reporter_test.go +++ b/probe/process/reporter_test.go @@ -4,25 +4,15 @@ import ( "reflect" "testing" + "github.com/weaveworks/scope/probe/proc" "github.com/weaveworks/scope/probe/process" "github.com/weaveworks/scope/report" "github.com/weaveworks/scope/test" ) -type mockWalker struct { - processes []process.Process -} - -func (m *mockWalker) Walk(f func(process.Process)) error { - for _, p := range m.processes { - f(p) - } - return nil -} - func TestReporter(t *testing.T) { - walker := &mockWalker{ - processes: []process.Process{ + procReader := &proc.MockedReader{ + Procs: []proc.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, @@ -31,7 +21,7 @@ func TestReporter(t *testing.T) { }, } - reporter := process.NewReporter(walker, "") + reporter := process.NewReporter(procReader, "") want := report.MakeReport() want.Process = report.Topology{ Nodes: report.Nodes{ diff --git a/probe/process/walker.go b/probe/process/walker.go deleted file mode 100644 index 4ba20027e..000000000 --- a/probe/process/walker.go +++ /dev/null @@ -1,56 +0,0 @@ -package process - -import "sync" - -// Process represents a single process. -type Process struct { - PID, PPID int - Comm string - Cmdline string - Threads int -} - -// Walker is something that walks the /proc directory -type Walker interface { - Walk(func(Process)) error -} - -// CachingWalker is a walker than caches a copy of the output from another -// Walker, and then allows other concurrent readers to Walk that copy. -type CachingWalker struct { - cache []Process - cacheLock sync.RWMutex - source Walker -} - -// NewCachingWalker returns a new CachingWalker -func NewCachingWalker(source Walker) *CachingWalker { - return &CachingWalker{source: source} -} - -// Walk walks a cached copy of process list -func (c *CachingWalker) Walk(f func(Process)) error { - c.cacheLock.RLock() - defer c.cacheLock.RUnlock() - - for _, p := range c.cache { - f(p) - } - return nil -} - -// Tick updates cached copy of process list -func (c *CachingWalker) Tick() error { - newCache := []Process{} - err := c.source.Walk(func(p Process) { - newCache = append(newCache, p) - }) - if err != nil { - return err - } - - c.cacheLock.Lock() - defer c.cacheLock.Unlock() - c.cache = newCache - return nil -} diff --git a/probe/process/walker_linux.go b/probe/process/walker_linux.go deleted file mode 100644 index f84ded6a6..000000000 --- a/probe/process/walker_linux.go +++ /dev/null @@ -1,79 +0,0 @@ -package process - -import ( - "bytes" - "io/ioutil" - "path" - "strconv" - "strings" -) - -// Hooks exposed for mocking -var ( - ReadDir = ioutil.ReadDir - ReadFile = ioutil.ReadFile -) - -type walker struct { - procRoot string -} - -// NewWalker creates a new process Walker. -func NewWalker(procRoot string) Walker { - return &walker{procRoot: procRoot} -} - -// Walk walks the supplied directory (expecting it to look like /proc) -// and marshalls the files into instances of Process, which it then -// passes one-by-one to the supplied function. Walk is only made public -// so that is can be tested. -func (w *walker) Walk(f func(Process)) error { - dirEntries, err := ReadDir(w.procRoot) - if err != nil { - return err - } - - for _, dirEntry := range dirEntries { - filename := dirEntry.Name() - pid, err := strconv.Atoi(filename) - if err != nil { - continue - } - - stat, err := ReadFile(path.Join(w.procRoot, filename, "stat")) - if err != nil { - continue - } - splits := strings.Fields(string(stat)) - ppid, err := strconv.Atoi(splits[3]) - if err != nil { - return err - } - - threads, err := strconv.Atoi(splits[19]) - if err != nil { - return err - } - - cmdline := "" - if cmdlineBuf, err := ReadFile(path.Join(w.procRoot, filename, "cmdline")); err == nil { - cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1) - cmdline = string(cmdlineBuf) - } - - comm := "(unknown)" - if commBuf, err := ReadFile(path.Join(w.procRoot, filename, "comm")); err == nil { - comm = strings.TrimSpace(string(commBuf)) - } - - f(Process{ - PID: pid, - PPID: ppid, - Comm: comm, - Cmdline: cmdline, - Threads: threads, - }) - } - - return nil -} diff --git a/probe/process/walker_linux_test.go b/probe/process/walker_linux_test.go deleted file mode 100644 index 6b449edd9..000000000 --- a/probe/process/walker_linux_test.go +++ /dev/null @@ -1,90 +0,0 @@ -package process_test - -import ( - "fmt" - "os" - "reflect" - "strconv" - "strings" - "testing" - "time" - - "github.com/weaveworks/scope/probe/process" - "github.com/weaveworks/scope/test" -) - -type mockProcess struct { - name, comm, cmdline string -} - -func (p mockProcess) Name() string { return p.name } -func (p mockProcess) Size() int64 { return 0 } -func (p mockProcess) Mode() os.FileMode { return 0 } -func (p mockProcess) ModTime() time.Time { return time.Now() } -func (p mockProcess) IsDir() bool { return true } -func (p mockProcess) Sys() interface{} { return nil } - -func TestWalker(t *testing.T) { - oldReadDir, oldReadFile := process.ReadDir, process.ReadFile - defer func() { - process.ReadDir = oldReadDir - process.ReadFile = oldReadFile - }() - - processes := map[string]mockProcess{ - "3": {name: "3", comm: "curl\n", cmdline: "curl\000google.com"}, - "2": {name: "2", comm: "bash\n"}, - "4": {name: "4", comm: "apache\n"}, - "notapid": {name: "notapid"}, - "1": {name: "1", comm: "init\n"}, - } - - process.ReadDir = func(path string) ([]os.FileInfo, error) { - result := []os.FileInfo{} - for _, p := range processes { - result = append(result, p) - } - return result, nil - } - - process.ReadFile = func(path string) ([]byte, error) { - splits := strings.Split(path, "/") - - pid := splits[len(splits)-2] - process, ok := processes[pid] - if !ok { - return nil, fmt.Errorf("not found") - } - - file := splits[len(splits)-1] - switch file { - case "comm": - return []byte(process.comm), nil - case "stat": - pid, _ := strconv.Atoi(splits[len(splits)-2]) - parent := pid - 1 - return []byte(fmt.Sprintf("%d na R %d 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1", pid, parent)), nil - case "cmdline": - return []byte(process.cmdline), nil - } - - return nil, fmt.Errorf("not found") - } - - want := map[int]process.Process{ - 3: {PID: 3, PPID: 2, Comm: "curl", Cmdline: "curl google.com", Threads: 1}, - 2: {PID: 2, PPID: 1, Comm: "bash", Cmdline: "", Threads: 1}, - 4: {PID: 4, PPID: 3, Comm: "apache", Cmdline: "", Threads: 1}, - 1: {PID: 1, PPID: 0, Comm: "init", Cmdline: "", Threads: 1}, - } - - have := map[int]process.Process{} - walker := process.NewWalker("unused") - err := walker.Walk(func(p process.Process) { - have[p.PID] = p - }) - - if err != nil || !reflect.DeepEqual(want, have) { - t.Errorf("%v (%v)", test.Diff(want, have), err) - } -} diff --git a/probe/process/walker_test.go b/probe/process/walker_test.go deleted file mode 100644 index 6046006bb..000000000 --- a/probe/process/walker_test.go +++ /dev/null @@ -1,66 +0,0 @@ -package process_test - -import ( - "reflect" - "testing" - - "github.com/weaveworks/scope/probe/process" - "github.com/weaveworks/scope/test" -) - -func TestBasicWalk(t *testing.T) { - var ( - procRoot = "/proc" - procFunc = func(process.Process) {} - ) - if err := process.NewWalker(procRoot).Walk(procFunc); err != nil { - t.Fatal(err) - } -} - -func TestCache(t *testing.T) { - processes := []process.Process{ - {PID: 1, PPID: 0, Comm: "init"}, - {PID: 2, PPID: 1, Comm: "bash"}, - {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, - {PID: 4, PPID: 2, Comm: "ping", Cmdline: "ping foo.bar.local"}, - } - walker := &mockWalker{ - processes: processes, - } - cachingWalker := process.NewCachingWalker(walker) - err := cachingWalker.Tick() - if err != nil { - t.Fatal(err) - } - - have, err := all(cachingWalker) - if err != nil || !reflect.DeepEqual(processes, have) { - t.Errorf("%v (%v)", test.Diff(processes, have), err) - } - - walker.processes = []process.Process{} - have, err = all(cachingWalker) - if err != nil || !reflect.DeepEqual(processes, have) { - t.Errorf("%v (%v)", test.Diff(processes, have), err) - } - - err = cachingWalker.Tick() - if err != nil { - t.Fatal(err) - } - - have, err = all(cachingWalker) - want := []process.Process{} - if err != nil || !reflect.DeepEqual(want, have) { - t.Errorf("%v (%v)", test.Diff(want, have), err) - } -} - -func all(w process.Walker) ([]process.Process, error) { - all := []process.Process{} - err := w.Walk(func(p process.Process) { - all = append(all, p) - }) - return all, err -}