github/weave-scope
1
0
Fork 0
mirror of https://github.com/weaveworks/scope.git synced 2026-07-16 03:49:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Naïve process walker for Darwin

Browse Source 
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.
This commit is contained in:
Peter Bourgon
2015-07-16 11:04:32 +02:00
parent c8079daf26
commit b585a362ac
11 changed files with 301 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
probe/main.go
View File

@@ -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 {

2
probe/process/reporter.go
View File

@@ -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{

6
probe/process/reporter_test.go
View File

@@ -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},

6
probe/process/tree.go
View File

@@ -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
})

2
probe/process/tree_test.go
View File

@@ -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},

89
probe/process/walker.go
View File

@@ -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 {

93
probe/process/walker_darwin.go Normal file
View File

@@ -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
}

79
probe/process/walker_linux.go Normal file
View File

@@ -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
}

90
probe/process/walker_linux_test.go Normal file
View File

@@ -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)
}
}

93
probe/process/walker_test.go
View File

@@ -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

4
scope
View File

@@ -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