Merge pull request #272 from tomwilkie/process-details

Improve process code in probe

Makefile

@@ -23,7 +23,7 @@ $(SCOPE_EXPORT): $(APP_EXE) $(PROBE_EXE) docker/*
 
 $(APP_EXE): app/*.go render/*.go report/*.go xfer/*.go
 
-$(PROBE_EXE): probe/*.go probe/tag/*.go probe/docker/*.go report/*.go xfer/*.go
+$(PROBE_EXE): probe/*.go probe/tag/*.go probe/docker/*.go probe/process/*.go report/*.go xfer/*.go
 
 $(APP_EXE) $(PROBE_EXE):
 	go get -tags netgo ./$(@D)

probe/docker/reporter.go

@@ -27,11 +27,11 @@ func NewReporter(registry Registry, scope string) *Reporter {
 }
 
 // Report generates a Report containing Container and ContainerImage topologies
-func (r *Reporter) Report() report.Report {
+func (r *Reporter) Report() (report.Report, error) {
 	result := report.MakeReport()
 	result.Container.Merge(r.containerTopology())
 	result.ContainerImage.Merge(r.containerImageTopology())
-	return result
+	return result, nil
 }
 
 func (r *Reporter) containerTopology() report.Topology {

probe/docker/reporter_test.go

@@ -72,7 +72,7 @@ func TestReporter(t *testing.T) {
 	}
 
 	reporter := docker.NewReporter(mockRegistryInstance, "")
-	have := reporter.Report()
+	have, _ := reporter.Report()
 	if !reflect.DeepEqual(want, have) {
 		t.Errorf("%s", test.Diff(want, have))
 	}

probe/docker/tagger.go

@@ -3,7 +3,7 @@ package docker
 import (
 	"strconv"
 
-	"github.com/weaveworks/scope/probe/tag"
+	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -15,7 +15,7 @@ const (
 
 // These vars are exported for testing.
 var (
-	NewPIDTreeStub = tag.NewPIDTree
+	NewProcessTreeStub = process.NewTree
 )
 
 // Tagger is a tagger that tags Docker container information to process
@@ -35,15 +35,15 @@ func NewTagger(registry Registry, procRoot string) *Tagger {
 
 // Tag implements Tagger.
 func (t *Tagger) Tag(r report.Report) (report.Report, error) {
-	pidTree, err := NewPIDTreeStub(t.procRoot)
+	tree, err := NewProcessTreeStub(t.procRoot)
 	if err != nil {
 		return report.MakeReport(), err
 	}
-	t.tag(pidTree, &r.Process)
+	t.tag(tree, &r.Process)
 	return r, nil
 }
 
-func (t *Tagger) tag(pidTree tag.PIDTree, topology *report.Topology) {
+func (t *Tagger) tag(tree process.Tree, topology *report.Topology) {
 	for nodeID, nodeMetadata := range topology.NodeMetadatas {
 		pidStr, ok := nodeMetadata["pid"]
 		if !ok {
@@ -67,7 +67,7 @@ func (t *Tagger) tag(pidTree tag.PIDTree, topology *report.Topology) {
 					break
 				}
 
-				candidate, err = pidTree.GetParent(candidate)
+				candidate, err = tree.GetParent(candidate)
 				if err != nil {
 					break
 				}

probe/docker/tagger_test.go

@@ -6,16 +6,16 @@ import (
 	"testing"
 
 	"github.com/weaveworks/scope/probe/docker"
-	"github.com/weaveworks/scope/probe/tag"
+	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
 )
 
-type mockPIDTree struct {
+type mockProcessTree struct {
 	parents map[int]int
 }
 
-func (m *mockPIDTree) GetParent(pid int) (int, error) {
+func (m *mockProcessTree) GetParent(pid int) (int, error) {
 	parent, ok := m.parents[pid]
 	if !ok {
 		return -1, fmt.Errorf("Not found %d", pid)
@@ -23,16 +23,12 @@ func (m *mockPIDTree) GetParent(pid int) (int, error) {
 	return parent, nil
 }
 
-func (m *mockPIDTree) ProcessTopology(hostID string) report.Topology {
-	panic("")
-}
-
 func TestTagger(t *testing.T) {
-	oldPIDTree := docker.NewPIDTreeStub
-	defer func() { docker.NewPIDTreeStub = oldPIDTree }()
+	oldProcessTree := docker.NewProcessTreeStub
+	defer func() { docker.NewProcessTreeStub = oldProcessTree }()
 
-	docker.NewPIDTreeStub = func(procRoot string) (tag.PIDTree, error) {
-		return &mockPIDTree{map[int]int{2: 1}}, nil
+	docker.NewProcessTreeStub = func(procRoot string) (process.Tree, error) {
+		return &mockProcessTree{map[int]int{2: 1}}, nil
 	}
 
 	var (

probe/main.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/weaveworks/procspy"
 	"github.com/weaveworks/scope/probe/docker"
+	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/probe/tag"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/xfer"
@@ -105,6 +106,11 @@ func main() {
 		taggers = append(taggers, weaveTagger)
 	}
 
+	// TODO provide an alternate implementation for Darwin.
+	if runtime.GOOS == linux {
+		reporters = append(reporters, process.NewReporter(*procRoot, hostID))
+	}
+
 	log.Printf("listening on %s", *listen)
 
 	quit := make(chan struct{})
@@ -129,18 +135,12 @@ func main() {
 				// Do this every tick so it gets tagged by the OriginHostTagger
 				r.Host = hostTopology(hostID, hostName)
 
-				// TODO abstract PIDTree to a process provider, and provide an
-				// alternate implementation for Darwin.
-				if runtime.GOOS == linux {
-					if pidTree, err := tag.NewPIDTree(*procRoot); err == nil {
-						r.Process.Merge(pidTree.ProcessTopology(hostID))
-					} else {
-						log.Printf("PIDTree: %v", err)
-					}
-				}
-
 				for _, reporter := range reporters {
-					r.Merge(reporter.Report())
+					newReport, err := reporter.Report()
+					if err != nil {
+						log.Printf("error generating report: %v", err)
+					}
+					r.Merge(newReport)
 				}
 
 				if weaveTagger != nil {

probe/process/reporter.go

@@ -0,0 +1,61 @@
+package process
+
+import (
+	"strconv"
+
+	"github.com/weaveworks/scope/probe/tag"
+	"github.com/weaveworks/scope/report"
+)
+
+// We use these keys in node metadata
+const (
+	PID     = "pid"
+	Comm    = "comm"
+	PPID    = "ppid"
+	Cmdline = "cmdline"
+	Threads = "threads"
+)
+
+// Reporter generate Reports containing the Process topology
+type reporter struct {
+	procRoot string
+	scope    string
+}
+
+// NewReporter makes a new Reporter
+func NewReporter(procRoot, scope string) tag.Reporter {
+	return &reporter{
+		procRoot: procRoot,
+		scope:    scope,
+	}
+}
+
+// Report generates a Report containing the Process topology
+func (r *reporter) Report() (report.Report, error) {
+	result := report.MakeReport()
+	processes, err := r.processTopology()
+	if err != nil {
+		return result, err
+	}
+	result.Process.Merge(processes)
+	return result, nil
+}
+
+func (r *reporter) processTopology() (report.Topology, error) {
+	t := report.NewTopology()
+	err := Walk(r.procRoot, func(p *Process) {
+		pidstr := strconv.Itoa(p.PID)
+		nodeID := report.MakeProcessNodeID(r.scope, pidstr)
+		t.NodeMetadatas[nodeID] = report.NodeMetadata{
+			PID:     pidstr,
+			Comm:    p.Comm,
+			Cmdline: p.Cmdline,
+			Threads: strconv.Itoa(p.Threads),
+		}
+		if p.PPID > 0 {
+			t.NodeMetadatas[nodeID][PPID] = strconv.Itoa(p.PPID)
+		}
+	})
+
+	return t, err
+}

probe/process/reporter_test.go

@@ -0,0 +1,68 @@
+package process_test
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/weaveworks/scope/probe/process"
+	"github.com/weaveworks/scope/report"
+	"github.com/weaveworks/scope/test"
+)
+
+func TestReporter(t *testing.T) {
+	oldWalk := process.Walk
+	defer func() { process.Walk = oldWalk }()
+
+	process.Walk = func(_ string, f func(*process.Process)) error {
+		for _, p := range []*process.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"},
+		} {
+			f(p)
+		}
+		return nil
+	}
+
+	reporter := process.NewReporter("", "")
+	want := report.MakeReport()
+	want.Process = report.Topology{
+		Adjacency:     report.Adjacency{},
+		EdgeMetadatas: report.EdgeMetadatas{},
+		NodeMetadatas: report.NodeMetadatas{
+			report.MakeProcessNodeID("", "1"): report.NodeMetadata{
+				process.PID:     "1",
+				process.Comm:    "init",
+				process.Cmdline: "",
+				process.Threads: "0",
+			},
+			report.MakeProcessNodeID("", "2"): report.NodeMetadata{
+				process.PID:     "2",
+				process.Comm:    "bash",
+				process.PPID:    "1",
+				process.Cmdline: "",
+				process.Threads: "0",
+			},
+			report.MakeProcessNodeID("", "3"): report.NodeMetadata{
+				process.PID:     "3",
+				process.Comm:    "apache",
+				process.PPID:    "1",
+				process.Cmdline: "",
+				process.Threads: "2",
+			},
+			report.MakeProcessNodeID("", "4"): report.NodeMetadata{
+				process.PID:     "4",
+				process.Comm:    "ping",
+				process.PPID:    "2",
+				process.Cmdline: "ping foo.bar.local",
+				process.Threads: "0",
+			},
+		},
+	}
+
+	have, err := reporter.Report()
+	if err != nil || !reflect.DeepEqual(want, have) {
+		t.Errorf("%s (%v)", test.Diff(want, have), err)
+	}
+}

probe/process/tree.go

@@ -0,0 +1,34 @@
+package process
+
+import (
+	"fmt"
+)
+
+// Tree represents all processes on the machine.
+type Tree interface {
+	GetParent(pid int) (int, error)
+}
+
+type tree struct {
+	processes map[int]*Process
+}
+
+// NewTree returns a new Tree that can be polled.
+func NewTree(procRoot string) (Tree, error) {
+	pt := tree{processes: map[int]*Process{}}
+	err := Walk(procRoot, func(p *Process) {
+		pt.processes[p.PID] = p
+	})
+
+	return &pt, err
+}
+
+// GetParent returns the pid of the parent process for a given pid
+func (pt *tree) GetParent(pid int) (int, error) {
+	proc, ok := pt.processes[pid]
+	if !ok {
+		return -1, fmt.Errorf("PID %d not found", pid)
+	}
+
+	return proc.PPID, nil
+}

probe/process/tree_test.go

@@ -0,0 +1,37 @@
+package process_test
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/weaveworks/scope/probe/process"
+)
+
+func TestTree(t *testing.T) {
+	oldWalk := process.Walk
+	defer func() { process.Walk = oldWalk }()
+
+	process.Walk = func(_ string, f func(*process.Process)) error {
+		for _, p := range []*process.Process{
+			{PID: 1, PPID: 0},
+			{PID: 2, PPID: 1},
+			{PID: 3, PPID: 1},
+			{PID: 4, PPID: 2},
+		} {
+			f(p)
+		}
+		return nil
+	}
+
+	tree, err := process.NewTree("foo")
+	if err != nil {
+		t.Fatalf("newProcessTree error: %v", err)
+	}
+
+	for pid, want := range map[int]int{2: 1, 3: 1, 4: 2} {
+		have, err := tree.GetParent(pid)
+		if err != nil || !reflect.DeepEqual(want, have) {
+			t.Errorf("%d: want %#v, have %#v (%v)", pid, want, have, err)
+		}
+	}
+}

probe/process/walker.go

@@ -0,0 +1,78 @@
+package process
+
+import (
+	"bytes"
+	"io/ioutil"
+	"path"
+	"strconv"
+	"strings"
+)
+
+// Process represents a single process.
+type Process struct {
+	PID, PPID int
+	Comm      string
+	Cmdline   string
+	Threads   int
+}
+
+// Hooks exposed for mocking
+var (
+	ReadDir  = ioutil.ReadDir
+	ReadFile = ioutil.ReadFile
+)
+
+// 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.
+var Walk = func(procRoot string, f func(*Process)) error {
+	dirEntries, err := ReadDir(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(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(procRoot, filename, "cmdline")); err == nil {
+			cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1)
+			cmdline = string(cmdlineBuf)
+		}
+
+		comm := "(unknown)"
+		if commBuf, err := ReadFile(path.Join(procRoot, filename, "comm")); err == nil {
+			comm = string(commBuf)
+		}
+
+		f(&Process{
+			PID:     pid,
+			PPID:    ppid,
+			Comm:    comm,
+			Cmdline: cmdline,
+			Threads: threads,
+		})
+	}
+
+	return nil
+}

probe/process/walker_test.go

@@ -0,0 +1,88 @@
+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    string
+	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", cmdline: "curl\000google.com"},
+		"2":       {name: "2"},
+		"4":       {name: "4"},
+		"notapid": {name: "notapid"},
+		"1":       {name: "1"},
+	}
+
+	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 "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: "(unknown)", Cmdline: "curl google.com", Threads: 1},
+		2: {PID: 2, PPID: 1, Comm: "(unknown)", Cmdline: "", Threads: 1},
+		4: {PID: 4, PPID: 3, Comm: "(unknown)", Cmdline: "", Threads: 1},
+		1: {PID: 1, PPID: 0, Comm: "(unknown)", Cmdline: "", Threads: 1},
+	}
+
+	have := map[int]*process.Process{}
+	err := process.Walk("unused", 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/tag/pidtree.go

@@ -1,112 +0,0 @@
-package tag
-
-import (
-	"fmt"
-	"io/ioutil"
-	"path"
-	"strconv"
-	"strings"
-
-	"github.com/weaveworks/scope/report"
-)
-
-// PIDTree represents all processes on the machine.
-type PIDTree interface {
-	GetParent(pid int) (int, error)
-	ProcessTopology(hostID string) report.Topology
-}
-
-type pidTree struct {
-	processes map[int]*process
-}
-
-// Process represents a single process.
-type process struct {
-	pid, ppid int
-	comm      string
-	parent    *process
-	children  []*process
-}
-
-// Hooks for mocking
-var (
-	readDir  = ioutil.ReadDir
-	readFile = ioutil.ReadFile
-)
-
-// NewPIDTree returns a new PIDTree that can be polled.
-func NewPIDTree(procRoot string) (PIDTree, error) {
-	dirEntries, err := readDir(procRoot)
-	if err != nil {
-		return nil, err
-	}
-
-	pt := pidTree{processes: map[int]*process{}}
-	for _, dirEntry := range dirEntries {
-		filename := dirEntry.Name()
-		pid, err := strconv.Atoi(filename)
-		if err != nil {
-			continue
-		}
-
-		stat, err := readFile(path.Join(procRoot, filename, "stat"))
-		if err != nil {
-			continue
-		}
-		splits := strings.Split(string(stat), " ")
-		ppid, err := strconv.Atoi(splits[3])
-		if err != nil {
-			return nil, err
-		}
-
-		comm := "(unknown)"
-		if commBuf, err := readFile(path.Join(procRoot, filename, "comm")); err == nil {
-			comm = string(commBuf)
-		}
-
-		pt.processes[pid] = &process{
-			pid:  pid,
-			ppid: ppid,
-			comm: comm,
-		}
-	}
-
-	for _, child := range pt.processes {
-		parent, ok := pt.processes[child.ppid]
-		if !ok {
-			// This can happen as listing proc is not a consistent snapshot
-			continue
-		}
-		child.parent = parent
-		parent.children = append(parent.children, child)
-	}
-
-	return &pt, nil
-}
-
-// GetParent returns the pid of the parent process for a given pid
-func (pt *pidTree) GetParent(pid int) (int, error) {
-	proc, ok := pt.processes[pid]
-	if !ok {
-		return -1, fmt.Errorf("PID %d not found", pid)
-	}
-
-	return proc.ppid, nil
-}
-
-// ProcessTopology returns a process topology based on the current state of the PIDTree.
-func (pt *pidTree) ProcessTopology(hostID string) report.Topology {
-	t := report.NewTopology()
-	for pid, proc := range pt.processes {
-		pidstr := strconv.Itoa(pid)
-		nodeID := report.MakeProcessNodeID(hostID, pidstr)
-		t.NodeMetadatas[nodeID] = report.NodeMetadata{
-			"pid":  pidstr,
-			"comm": proc.comm,
-		}
-		if proc.ppid > 0 {
-			t.NodeMetadatas[nodeID]["ppid"] = strconv.Itoa(proc.ppid)
-		}
-	}
-	return t
-}

probe/tag/pidtree_test.go

@@ -1,65 +0,0 @@
-package tag
-
-import (
-	"fmt"
-	"os"
-	"reflect"
-	"strconv"
-	"strings"
-	"testing"
-	"time"
-)
-
-type fileinfo struct {
-	name string
-}
-
-func (f fileinfo) Name() string       { return f.name }
-func (f fileinfo) Size() int64        { return 0 }
-func (f fileinfo) Mode() os.FileMode  { return 0 }
-func (f fileinfo) ModTime() time.Time { return time.Now() }
-func (f fileinfo) IsDir() bool        { return true }
-func (f fileinfo) Sys() interface{}   { return nil }
-
-func TestPIDTree(t *testing.T) {
-	oldReadDir, oldReadFile := readDir, readFile
-	defer func() {
-		readDir = oldReadDir
-		readFile = oldReadFile
-	}()
-
-	readDir = func(path string) ([]os.FileInfo, error) {
-		return []os.FileInfo{
-			fileinfo{"3"}, fileinfo{"2"}, fileinfo{"4"},
-			fileinfo{"notapid"}, fileinfo{"1"},
-		}, nil
-	}
-
-	readFile = func(path string) ([]byte, error) {
-		splits := strings.Split(path, "/")
-		if splits[len(splits)-1] != "stat" {
-			return nil, fmt.Errorf("not stat")
-		}
-		pid, err := strconv.Atoi(splits[len(splits)-2])
-		if err != nil {
-			return nil, err
-		}
-		parent := pid - 1
-		return []byte(fmt.Sprintf("%d na R %d", pid, parent)), nil
-	}
-
-	pidtree, err := NewPIDTree("/proc")
-	if err != nil {
-		t.Fatalf("newPIDTree error: %v", err)
-	}
-
-	for pid, want := range map[int]int{
-		2: 1,
-		3: 2,
-	} {
-		have, err := pidtree.GetParent(pid)
-		if err != nil || !reflect.DeepEqual(want, have) {
-			t.Errorf("%d: want %#v, have %#v (%v)", pid, want, have, err)
-		}
-	}
-}

probe/tag/tagger.go

@@ -13,7 +13,7 @@ type Tagger interface {
 
 // Reporter generates Reports.
 type Reporter interface {
-	Report() report.Report
+	Report() (report.Report, error)
 }
 
 // Apply tags the report with all the taggers.

render/detailed_node.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 
 	"github.com/weaveworks/scope/probe/docker"
+	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -139,15 +140,18 @@ func addressOriginTable(nmd report.NodeMetadata) (Table, bool) {
 
 func processOriginTable(nmd report.NodeMetadata) (Table, bool) {
 	rows := []Row{}
-	if val, ok := nmd["comm"]; ok {
-		rows = append(rows, Row{"Name (comm)", val, ""})
-	}
-	if val, ok := nmd["pid"]; ok {
-		rows = append(rows, Row{"PID", val, ""})
-	}
-	if val, ok := nmd["ppid"]; ok {
-		rows = append(rows, Row{"Parent PID", val, ""})
+	for _, tuple := range []struct{ key, human string }{
+		{process.Comm, "Name (comm)"},
+		{process.PID, "PID"},
+		{process.PPID, "Parent PID"},
+		{process.Cmdline, "Command"},
+		{process.Threads, "# Threads"},
+	} {
+		if val, ok := nmd[tuple.key]; ok {
+			rows = append(rows, Row{Key: tuple.human, ValueMajor: val, ValueMinor: ""})
+		}
 	}
+
 	return Table{
 		Title:   "Origin Process",
 		Numeric: false,