Merge pull request #212 from weaveworks/process-topology

Process topology

probe/main.go

@@ -29,7 +29,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)")
-		dockerTagger       = flag.Bool("docker", true, "collect Docker-related attributes for processes")
+		dockerEnabled      = flag.Bool("docker", true, "collect Docker-related attributes for processes")
 		dockerInterval     = flag.Duration("docker.interval", 10*time.Second, "how often to update Docker attributes")
 		procRoot           = flag.String("proc.root", "/proc", "location of the proc filesystem")
 	)
@@ -65,13 +65,15 @@ func main() {
 	defer publisher.Close()
 
 	taggers := []tag.Tagger{tag.NewTopologyTagger()}
-	if *dockerTagger && runtime.GOOS == "linux" {
-		t, err := tag.NewDockerTagger(*procRoot, *dockerInterval)
+	var dockerTagger *tag.DockerTagger
+	if *dockerEnabled && runtime.GOOS == "linux" {
+		var err error
+		dockerTagger, err = tag.NewDockerTagger(*procRoot, *dockerInterval)
 		if err != nil {
 			log.Fatalf("failed to start docker tagger: %v", err)
 		}
-		defer t.Stop()
-		taggers = append(taggers, t)
+		defer dockerTagger.Stop()
+		taggers = append(taggers, dockerTagger)
 	}
 
 	log.Printf("listening on %s", *listen)
@@ -97,8 +99,12 @@ func main() {
 
 			case <-spyTick:
 				r.Merge(spy(hostID, hostName, *spyProcs))
+				if pidTree, err := tag.NewPIDTree(*procRoot); err == nil {
+					r.Process.Merge(pidTree.ProcessTopology(hostID))
+				} else {
+					log.Print(err)
+				}
 				r = tag.Apply(r, taggers)
-				// log.Printf("merged report:\n%#v\n", r)
 
 			case <-quit:
 				return

probe/tag/docker_tagger.go

@@ -18,7 +18,7 @@ const (
 
 var (
 	newDockerClientStub = newDockerClient
-	newPIDTreeStub      = newPIDTree
+	newPIDTreeStub      = NewPIDTree
 )
 
 // DockerTagger is a tagger that tags Docker container information to process
@@ -33,7 +33,7 @@ type DockerTagger struct {
 	images          map[string]*docker.APIImages
 
 	procRoot string
-	pidTree  *pidTree
+	pidTree  *PIDTree
 }
 
 // NewDockerTagger returns a usable DockerTagger. Don't forget to Stop it.

probe/tag/docker_tagger_test.go

@@ -40,12 +40,12 @@ func TestDockerTagger(t *testing.T) {
 	oldPIDTree, oldDockerClient := newPIDTreeStub, newDockerClientStub
 	defer func() { newPIDTreeStub, newDockerClientStub = oldPIDTree, oldDockerClient }()
 
-	newPIDTreeStub = func(procRoot string) (*pidTree, error) {
-		pid1 := &process{pid: 1}
-		pid2 := &process{pid: 2, ppid: 1, parent: pid1}
-		pid1.children = []*process{pid2}
-		return &pidTree{
-			processes: map[int]*process{
+	newPIDTreeStub = func(procRoot string) (*PIDTree, error) {
+		pid1 := &Process{PID: 1}
+		pid2 := &Process{PID: 2, PPID: 1, parent: pid1}
+		pid1.children = []*Process{pid2}
+		return &PIDTree{
+			processes: map[int]*Process{
 				1: pid1, 2: pid2,
 			},
 		}, nil

probe/tag/pidtree.go

@@ -6,16 +6,21 @@ import (
 	"path"
 	"strconv"
 	"strings"
+
+	"github.com/weaveworks/scope/report"
 )
 
-type pidTree struct {
-	processes map[int]*process
+// PIDTree represents all processes on the machine.
+type PIDTree struct {
+	processes map[int]*Process
 }
 
-type process struct {
-	pid, ppid int
-	parent    *process
-	children  []*process
+// Process represents a single process.
+type Process struct {
+	PID, PPID int
+	Comm      string
+	parent    *Process
+	children  []*Process
 }
 
 // Hooks for mocking
@@ -24,35 +29,45 @@ var (
 	readFile = ioutil.ReadFile
 )
 
-func newPIDTree(procRoot string) (*pidTree, error) {
+// 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{}}
+	pt := PIDTree{processes: map[int]*Process{}}
 	for _, dirEntry := range dirEntries {
-		pid, err := strconv.Atoi(dirEntry.Name())
+		filename := dirEntry.Name()
+		pid, err := strconv.Atoi(filename)
 		if err != nil {
 			continue
 		}
 
-		stat, err := readFile(path.Join(procRoot, dirEntry.Name(), "stat"))
+		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
 		}
 
-		pt.processes[pid] = &process{pid: pid, ppid: ppid}
+		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]
+		parent, ok := pt.processes[child.PPID]
 		if !ok {
 			// This can happen as listing proc is not a consistent snapshot
 			continue
@@ -64,17 +79,17 @@ func newPIDTree(procRoot string) (*pidTree, error) {
 	return &pt, nil
 }
 
-func (pt *pidTree) getParent(pid int) (int, error) {
+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
+	return proc.PPID, nil
 }
 
 // allChildren returns a flattened list of child pids including the given pid
-func (pt *pidTree) allChildren(pid int) ([]int, error) {
+func (pt *PIDTree) allChildren(pid int) ([]int, error) {
 	proc, ok := pt.processes[pid]
 	if !ok {
 		return []int{}, fmt.Errorf("PID %d not found", pid)
@@ -82,9 +97,9 @@ func (pt *pidTree) allChildren(pid int) ([]int, error) {
 
 	var result []int
 
-	var f func(*process)
-	f = func(p *process) {
-		result = append(result, p.pid)
+	var f func(*Process)
+	f = func(p *Process) {
+		result = append(result, p.PID)
 		for _, child := range p.children {
 			f(child)
 		}
@@ -93,3 +108,20 @@ func (pt *pidTree) allChildren(pid int) ([]int, error) {
 	f(proc)
 	return result, 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
+}

report/detailed_node.go

@@ -52,6 +52,9 @@ func OriginTable(r Report, originID string) (Table, bool) {
 	if nmd, ok := r.Address.NodeMetadatas[originID]; ok {
 		return addressOriginTable(nmd)
 	}
+	if nmd, ok := r.Process.NodeMetadatas[originID]; ok {
+		return processOriginTable(nmd)
+	}
 	if nmd, ok := r.Host.NodeMetadatas[originID]; ok {
 		return hostOriginTable(nmd)
 	}
@@ -97,6 +100,24 @@ func addressOriginTable(nmd NodeMetadata) (Table, bool) {
 	}, len(rows) > 0
 }
 
+func processOriginTable(nmd 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, ""})
+	}
+	return Table{
+		Title:   "Origin Process",
+		Numeric: false,
+		Rows:    rows,
+	}, len(rows) > 0
+}
+
 func hostOriginTable(nmd NodeMetadata) (Table, bool) {
 	rows := []Row{}
 	if val, ok := nmd["host_name"]; ok {

report/detailed_node_test.go

@@ -21,13 +21,6 @@ func TestOriginTable(t *testing.T) {
 			Numeric: false,
 			Rows:    []report.Row{{"Host name", clientHostName, ""}},
 		},
-		//report.MakeProcessNodeID(clientHostID, "4242"): {
-		//	Title:   "Origin Process",
-		//	Numeric: false,
-		//	Rows: []report.Row{
-		//		{"Host name", "client.host.com", ""},
-		//	},
-		//},
 		clientAddressNodeID: {
 			Title:   "Origin Address",
 			Numeric: false,
@@ -35,18 +28,14 @@ func TestOriginTable(t *testing.T) {
 				{"Host name", clientHostName, ""},
 			},
 		},
-		//report.MakeProcessNodeID(clientHostID, "4242"): {
-		//	Title:   "Origin Process",
-		//	Numeric: false,
-		//	Rows: []report.Row{
-		//		{"Process name", "curl", ""},
-		//		{"PID", "4242", ""},
-		//		{"Docker container ID", "a1b2c3d4e5", ""},
-		//		{"Docker container name", "fixture-container", ""},
-		//		{"Docker image ID", "0000000000", ""},
-		//		{"Docker image name", "fixture/container:latest", ""},
-		//	},
-		//},
+		report.MakeProcessNodeID(clientHostID, "4242"): {
+			Title:   "Origin Process",
+			Numeric: false,
+			Rows: []report.Row{
+				{"Name (comm)", "curl", ""},
+				{"PID", "4242", ""},
+			},
+		},
 		serverHostNodeID: {
 			Title:   "Origin Host",
 			Numeric: false,

report/merge.go

@@ -8,6 +8,7 @@ package report
 func (r *Report) Merge(other Report) {
 	r.Endpoint.Merge(other.Endpoint)
 	r.Address.Merge(other.Address)
+	r.Process.Merge(other.Process)
 	r.Host.Merge(other.Host)
 }
 

report/report.go

@@ -19,6 +19,9 @@ type Report struct {
 	// endpoints (e.g. ICMP). Edges are present.
 	Address Topology
 
+	// Process nodes are processes on each host. Edges are not present.
+	Process Topology
+
 	// Host nodes are physical hosts that run probes. Metadata includes things
 	// like operating system, load, etc. The information is scraped by the
 	// probes with each published report. Edges are not present.
@@ -69,6 +72,7 @@ func MakeReport() Report {
 	return Report{
 		Endpoint: NewTopology(),
 		Address:  NewTopology(),
+		Process:  NewTopology(),
 		Host:     NewTopology(),
 	}
 }
@@ -79,6 +83,7 @@ func (r Report) Squash() Report {
 	localNetworks := r.LocalNetworks()
 	r.Endpoint = r.Endpoint.Squash(EndpointIDAddresser, localNetworks)
 	r.Address = r.Address.Squash(AddressIDAddresser, localNetworks)
+	r.Process = r.Process.Squash(PanicIDAddresser, localNetworks)
 	r.Host = r.Host.Squash(PanicIDAddresser, localNetworks)
 	return r
 }

report/report_fixture_test.go

@@ -101,27 +101,27 @@ var reportFixture = report.Report{
 			},
 		},
 	},
-	//Process: report.Topology{
-	//	Adjacency: report.Adjacency{},
-	//	NodeMetadatas: report.NodeMetadatas{
-	//		report.MakeProcessNodeID(clientHostID, "4242"): report.NodeMetadata{
-	//			"host_name":             "client.host.com",
-	//			"pid":                   "4242",
-	//			"process_name":          "curl",
-	//			"docker_container_id":   "a1b2c3d4e5",
-	//			"docker_container_name": "fixture-container",
-	//			"docker_image_id":       "0000000000",
-	//			"docker_image_name":     "fixture/container:latest",
-	//		},
-	//		report.MakeProcessNodeID(serverHostID, "215"): report.NodeMetadata{
-	//			"pid":          "215",
-	//			"process_name": "apache",
-	//		},
-	//
-	//		"no-container": report.NodeMetadata{},
-	//	},
-	//	EdgeMetadatas: report.EdgeMetadatas{},
-	//},
+	Process: report.Topology{
+		Adjacency: report.Adjacency{},
+		NodeMetadatas: report.NodeMetadatas{
+			report.MakeProcessNodeID(clientHostID, "4242"): report.NodeMetadata{
+				"host_name":             "client.host.com",
+				"pid":                   "4242",
+				"comm":                  "curl",
+				"docker_container_id":   "a1b2c3d4e5",
+				"docker_container_name": "fixture-container",
+				"docker_image_id":       "0000000000",
+				"docker_image_name":     "fixture/container:latest",
+			},
+			report.MakeProcessNodeID(serverHostID, "215"): report.NodeMetadata{
+				"pid":          "215",
+				"process_name": "apache",
+			},
+
+			"no-container": report.NodeMetadata{},
+		},
+		EdgeMetadatas: report.EdgeMetadatas{},
+	},
 	Host: report.Topology{
 		Adjacency: report.Adjacency{},
 		NodeMetadatas: report.NodeMetadatas{