Merge pull request #320 from weaveworks/process-names-on-darwin

Naïve process walker for Darwin

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 {

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.NewNodeMetadata(map[string]string{

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

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

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

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 {

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

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

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

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

scope

@@ -88,7 +88,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
 }
@@ -162,7 +162,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 command_exists weave && is_running $WEAVE_CONTAINER_NAME && weave_dns_present; then