From b585a362ac08ef5555d9529c5e7c4d9112e4d60d Mon Sep 17 00:00:00 2001 From: Peter Bourgon Date: Thu, 16 Jul 2015 11:04:32 +0200 Subject: [PATCH] =?UTF-8?q?Na=C3=AFve=20process=20walker=20for=20Darwin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This fixes the regression where process names weren't appearing for Darwin probes. Makes testing easier. Also, changes the process walker to operate on value types. There's no performance advantage to using reference types for something of this size, and there appeared to be a data race in the Darwin port that caused nodes to gain and lose process names over time. Also, restructures how to enable docker scraping. Default false when run manually, and enabled via --probe.docker true in the scope script. --- probe/main.go | 42 ++++++-------- probe/process/reporter.go | 2 +- probe/process/reporter_test.go | 6 +- probe/process/tree.go | 6 +- probe/process/tree_test.go | 2 +- probe/process/walker.go | 89 ++-------------------------- probe/process/walker_darwin.go | 93 ++++++++++++++++++++++++++++++ probe/process/walker_linux.go | 79 +++++++++++++++++++++++++ probe/process/walker_linux_test.go | 90 +++++++++++++++++++++++++++++ probe/process/walker_test.go | 93 ++---------------------------- scope | 4 +- 11 files changed, 301 insertions(+), 205 deletions(-) create mode 100644 probe/process/walker_darwin.go create mode 100644 probe/process/walker_linux.go create mode 100644 probe/process/walker_linux_test.go diff --git a/probe/main.go b/probe/main.go index 999ea5c7b..19c67359a 100644 --- a/probe/main.go +++ b/probe/main.go @@ -7,7 +7,6 @@ import ( _ "net/http/pprof" "os" "os/signal" - "runtime" "strconv" "syscall" "time" @@ -24,8 +23,6 @@ import ( var version = "dev" // set at build time -const linux = "linux" // runtime.GOOS - func main() { var ( httpListen = flag.String("http.listen", "", "listen address for HTTP profiling and instrumentation server") @@ -34,7 +31,7 @@ func main() { listen = flag.String("listen", ":"+strconv.Itoa(xfer.ProbePort), "listen address") prometheusEndpoint = flag.String("prometheus.endpoint", "/metrics", "Prometheus metrics exposition endpoint (requires -http.listen)") spyProcs = flag.Bool("processes", true, "report processes (needs root)") - dockerEnabled = flag.Bool("docker", true, "collect Docker-related attributes for processes") + dockerEnabled = flag.Bool("docker", false, "collect Docker-related attributes for processes") dockerInterval = flag.Duration("docker.interval", 10*time.Second, "how often to update Docker attributes") dockerBridge = flag.String("docker.bridge", "docker0", "the docker bridge name") weaveRouterAddr = flag.String("weave.router.addr", "", "IP address or FQDN of the Weave router") @@ -82,25 +79,22 @@ func main() { processCache *process.CachingWalker ) - // TODO provide an alternate implementation for Darwin. - if runtime.GOOS == linux { - processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) - reporters = append(reporters, process.NewReporter(processCache, hostID)) + processCache = process.NewCachingWalker(process.NewWalker(*procRoot)) + reporters = append(reporters, process.NewReporter(processCache, hostID)) - if *dockerEnabled { - if err = report.AddLocalBridge(*dockerBridge); err != nil { - log.Fatalf("failed to get docker bridge address: %v", err) - } - - dockerRegistry, err := docker.NewRegistry(*dockerInterval) - if err != nil { - log.Fatalf("failed to start docker registry: %v", err) - } - defer dockerRegistry.Stop() - - taggers = append(taggers, docker.NewTagger(dockerRegistry, processCache)) - reporters = append(reporters, docker.NewReporter(dockerRegistry, hostID)) + if *dockerEnabled { + if err := report.AddLocalBridge(*dockerBridge); err != nil { + log.Fatalf("failed to get docker bridge address: %v", err) } + + dockerRegistry, err := docker.NewRegistry(*dockerInterval) + if err != nil { + log.Fatalf("failed to start docker registry: %v", err) + } + defer dockerRegistry.Stop() + + taggers = append(taggers, docker.NewTagger(dockerRegistry, processCache)) + reporters = append(reporters, docker.NewReporter(dockerRegistry, hostID)) } if *weaveRouterAddr != "" { @@ -131,10 +125,8 @@ func main() { r = report.MakeReport() case <-spyTick: - if processCache != nil { - if err := processCache.Update(); err != nil { - log.Printf("error reading processes: %v", err) - } + if err := processCache.Update(); err != nil { + log.Printf("error reading processes: %v", err) } for _, reporter := range reporters { diff --git a/probe/process/reporter.go b/probe/process/reporter.go index efdf4ba4d..e99477ef0 100644 --- a/probe/process/reporter.go +++ b/probe/process/reporter.go @@ -42,7 +42,7 @@ func (r *Reporter) Report() (report.Report, error) { func (r *Reporter) processTopology() (report.Topology, error) { t := report.NewTopology() - err := r.walker.Walk(func(p *Process) { + err := r.walker.Walk(func(p Process) { pidstr := strconv.Itoa(p.PID) nodeID := report.MakeProcessNodeID(r.scope, pidstr) t.NodeMetadatas[nodeID] = report.NodeMetadata{ diff --git a/probe/process/reporter_test.go b/probe/process/reporter_test.go index cd0ffc501..2662a463f 100644 --- a/probe/process/reporter_test.go +++ b/probe/process/reporter_test.go @@ -10,10 +10,10 @@ import ( ) type mockWalker struct { - processes []*process.Process + processes []process.Process } -func (m *mockWalker) Walk(f func(*process.Process)) error { +func (m *mockWalker) Walk(f func(process.Process)) error { for _, p := range m.processes { f(p) } @@ -22,7 +22,7 @@ func (m *mockWalker) Walk(f func(*process.Process)) error { func TestReporter(t *testing.T) { walker := &mockWalker{ - processes: []*process.Process{ + processes: []process.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/process/tree.go b/probe/process/tree.go index 2fd8c65e2..2d03a9c0f 100644 --- a/probe/process/tree.go +++ b/probe/process/tree.go @@ -10,13 +10,13 @@ type Tree interface { } type tree struct { - processes map[int]*Process + processes map[int]Process } // NewTree returns a new Tree that can be polled. func NewTree(walker Walker) (Tree, error) { - pt := tree{processes: map[int]*Process{}} - err := walker.Walk(func(p *Process) { + pt := tree{processes: map[int]Process{}} + err := walker.Walk(func(p Process) { pt.processes[p.PID] = p }) diff --git a/probe/process/tree_test.go b/probe/process/tree_test.go index 8278f53b2..6a93cc226 100644 --- a/probe/process/tree_test.go +++ b/probe/process/tree_test.go @@ -9,7 +9,7 @@ import ( func TestTree(t *testing.T) { walker := &mockWalker{ - processes: []*process.Process{ + processes: []process.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/process/walker.go b/probe/process/walker.go index ca44934f3..4a88c45cd 100644 --- a/probe/process/walker.go +++ b/probe/process/walker.go @@ -1,13 +1,6 @@ package process -import ( - "bytes" - "io/ioutil" - "path" - "strconv" - "strings" - "sync" -) +import "sync" // Process represents a single process. type Process struct { @@ -19,83 +12,13 @@ type Process struct { // Walker is something that walks the /proc directory type Walker interface { - Walk(func(*Process)) error -} - -// 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 + 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 + cache []Process cacheLock sync.RWMutex source Walker } @@ -106,7 +29,7 @@ func NewCachingWalker(source Walker) *CachingWalker { } // Walk walks a cached copy of process list -func (c *CachingWalker) Walk(f func(*Process)) error { +func (c *CachingWalker) Walk(f func(Process)) error { c.cacheLock.RLock() defer c.cacheLock.RUnlock() @@ -118,8 +41,8 @@ func (c *CachingWalker) Walk(f func(*Process)) error { // Update updates cached copy of process list func (c *CachingWalker) Update() error { - newCache := []*Process{} - err := c.source.Walk(func(p *Process) { + newCache := []Process{} + err := c.source.Walk(func(p Process) { newCache = append(newCache, p) }) if err != nil { diff --git a/probe/process/walker_darwin.go b/probe/process/walker_darwin.go new file mode 100644 index 000000000..cd25b3df3 --- /dev/null +++ b/probe/process/walker_darwin.go @@ -0,0 +1,93 @@ +package process + +import ( + "fmt" + "os/exec" + "strconv" + "strings" +) + +// NewWalker returns a Darwin (lsof-based) walker. +func NewWalker(_ string) Walker { + return &walker{} +} + +type walker struct{} + +const ( + lsofBinary = "lsof" + lsofFields = "cn" // parseLSOF() depends on the order +) + +// These functions copied from procspy. + +func (walker) Walk(f func(Process)) error { + output, 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 + } + + processes, err := parseLSOF(string(output)) + if err != nil { + return err + } + + for _, process := range processes { + f(process) + } + return nil +} + +func parseLSOF(output string) (map[string]Process, error) { + var ( + processes = map[string]Process{} // Local addr -> Proc + process Process + ) + for _, line := range strings.Split(output, "\n") { + if len(line) <= 1 { + continue + } + + var ( + field = line[0] + value = line[1:] + ) + switch field { + case 'p': + pid, err := strconv.Atoi(value) + if err != nil { + return nil, fmt.Errorf("invalid 'p' field in lsof output: %#v", value) + } + process.PID = pid + + case 'c': + process.Comm = value + + case 'n': + // 'n' is the last field, with '-F cn' + // format examples: + // "192.168.2.111:44013->54.229.241.196:80" + // "[2003:45:2b57:8900:1869:2947:f942:aba7]:55711->[2a00:1450:4008:c01::11]:443" + // "*:111" <- a listen + addresses := strings.SplitN(value, "->", 2) + if len(addresses) != 2 { + // That's a listen entry. + continue + } + processes[addresses[0]] = Process{ + PID: process.PID, + Comm: process.Comm, + } + + default: + return nil, fmt.Errorf("unexpected lsof field: %c in %#v", field, value) + } + } + return processes, nil +} diff --git a/probe/process/walker_linux.go b/probe/process/walker_linux.go new file mode 100644 index 000000000..f84ded6a6 --- /dev/null +++ b/probe/process/walker_linux.go @@ -0,0 +1,79 @@ +package process + +import ( + "bytes" + "io/ioutil" + "path" + "strconv" + "strings" +) + +// Hooks exposed for mocking +var ( + ReadDir = ioutil.ReadDir + ReadFile = ioutil.ReadFile +) + +type walker struct { + procRoot string +} + +// NewWalker creates a new process Walker. +func NewWalker(procRoot string) Walker { + return &walker{procRoot: procRoot} +} + +// Walk walks the supplied directory (expecting it to look like /proc) +// and marshalls the files into instances of Process, which it then +// passes one-by-one to the supplied function. Walk is only made public +// so that is can be tested. +func (w *walker) Walk(f func(Process)) error { + dirEntries, err := ReadDir(w.procRoot) + if err != nil { + return err + } + + for _, dirEntry := range dirEntries { + filename := dirEntry.Name() + pid, err := strconv.Atoi(filename) + if err != nil { + continue + } + + stat, err := ReadFile(path.Join(w.procRoot, filename, "stat")) + if err != nil { + continue + } + splits := strings.Fields(string(stat)) + ppid, err := strconv.Atoi(splits[3]) + if err != nil { + return err + } + + threads, err := strconv.Atoi(splits[19]) + if err != nil { + return err + } + + cmdline := "" + if cmdlineBuf, err := ReadFile(path.Join(w.procRoot, filename, "cmdline")); err == nil { + cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1) + cmdline = string(cmdlineBuf) + } + + comm := "(unknown)" + if commBuf, err := ReadFile(path.Join(w.procRoot, filename, "comm")); err == nil { + comm = strings.TrimSpace(string(commBuf)) + } + + f(Process{ + PID: pid, + PPID: ppid, + Comm: comm, + Cmdline: cmdline, + Threads: threads, + }) + } + + return nil +} diff --git a/probe/process/walker_linux_test.go b/probe/process/walker_linux_test.go new file mode 100644 index 000000000..6b449edd9 --- /dev/null +++ b/probe/process/walker_linux_test.go @@ -0,0 +1,90 @@ +package process_test + +import ( + "fmt" + "os" + "reflect" + "strconv" + "strings" + "testing" + "time" + + "github.com/weaveworks/scope/probe/process" + "github.com/weaveworks/scope/test" +) + +type mockProcess struct { + name, comm, cmdline string +} + +func (p mockProcess) Name() string { return p.name } +func (p mockProcess) Size() int64 { return 0 } +func (p mockProcess) Mode() os.FileMode { return 0 } +func (p mockProcess) ModTime() time.Time { return time.Now() } +func (p mockProcess) IsDir() bool { return true } +func (p mockProcess) Sys() interface{} { return nil } + +func TestWalker(t *testing.T) { + oldReadDir, oldReadFile := process.ReadDir, process.ReadFile + defer func() { + process.ReadDir = oldReadDir + process.ReadFile = oldReadFile + }() + + processes := map[string]mockProcess{ + "3": {name: "3", comm: "curl\n", cmdline: "curl\000google.com"}, + "2": {name: "2", comm: "bash\n"}, + "4": {name: "4", comm: "apache\n"}, + "notapid": {name: "notapid"}, + "1": {name: "1", comm: "init\n"}, + } + + process.ReadDir = func(path string) ([]os.FileInfo, error) { + result := []os.FileInfo{} + for _, p := range processes { + result = append(result, p) + } + return result, nil + } + + process.ReadFile = func(path string) ([]byte, error) { + splits := strings.Split(path, "/") + + pid := splits[len(splits)-2] + process, ok := processes[pid] + if !ok { + return nil, fmt.Errorf("not found") + } + + file := splits[len(splits)-1] + switch file { + case "comm": + return []byte(process.comm), nil + case "stat": + pid, _ := strconv.Atoi(splits[len(splits)-2]) + parent := pid - 1 + return []byte(fmt.Sprintf("%d na R %d 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1", pid, parent)), nil + case "cmdline": + return []byte(process.cmdline), nil + } + + return nil, fmt.Errorf("not found") + } + + want := map[int]process.Process{ + 3: {PID: 3, PPID: 2, Comm: "curl", Cmdline: "curl google.com", Threads: 1}, + 2: {PID: 2, PPID: 1, Comm: "bash", Cmdline: "", Threads: 1}, + 4: {PID: 4, PPID: 3, Comm: "apache", Cmdline: "", Threads: 1}, + 1: {PID: 1, PPID: 0, Comm: "init", Cmdline: "", Threads: 1}, + } + + have := map[int]process.Process{} + walker := process.NewWalker("unused") + err := walker.Walk(func(p process.Process) { + have[p.PID] = p + }) + + if err != nil || !reflect.DeepEqual(want, have) { + t.Errorf("%v (%v)", test.Diff(want, have), err) + } +} diff --git a/probe/process/walker_test.go b/probe/process/walker_test.go index 729588c22..8008eda73 100644 --- a/probe/process/walker_test.go +++ b/probe/process/walker_test.go @@ -1,96 +1,15 @@ 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) - } -} - func TestCache(t *testing.T) { - processes := []*process.Process{ + processes := []process.Process{ {PID: 1, PPID: 0, Comm: "init"}, {PID: 2, PPID: 1, Comm: "bash"}, {PID: 3, PPID: 1, Comm: "apache", Threads: 2}, @@ -110,7 +29,7 @@ func TestCache(t *testing.T) { t.Errorf("%v (%v)", test.Diff(processes, have), err) } - walker.processes = []*process.Process{} + walker.processes = []process.Process{} have, err = all(cachingWalker) if err != nil || !reflect.DeepEqual(processes, have) { t.Errorf("%v (%v)", test.Diff(processes, have), err) @@ -122,15 +41,15 @@ func TestCache(t *testing.T) { } have, err = all(cachingWalker) - want := []*process.Process{} + 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) { +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 diff --git a/scope b/scope index 7d61b5073..a3b6e8d6c 100755 --- a/scope +++ b/scope @@ -90,7 +90,7 @@ weave_dns_add() { CONTAINER_FQDN="$2" shift 2 - for ip in $*; do + for ip in $*; do weave dns-add $ip $CONTAINER_ID -h $CONTAINER_FQDN done } @@ -160,7 +160,7 @@ case "$COMMAND" in CONTAINER=$(docker run --privileged -d --name=$SCOPE_CONTAINER_NAME --net=host --pid=host \ -v /var/run/docker.sock:/var/run/docker.sock \ - $WEAVESCOPE_DOCKER_ARGS $SCOPE_IMAGE $WEAVESCOPE_DNS_ARGS "$@") + $WEAVESCOPE_DOCKER_ARGS $SCOPE_IMAGE $WEAVESCOPE_DNS_ARGS --probe.docker true "$@") IP_ADDRS=$(docker run --net=host gliderlabs/alpine /bin/sh -c "$IP_ADDR_CMD") if is_running $DNS_CONTAINER_NAME && is_running $WEAVE_CONTAINER_NAME && command_exists weave; then