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