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{