Refactor process mappers into taggers.

Makefile

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

app/api_topology.go

@@ -30,17 +30,6 @@ type APIEdge struct {
 	Metadata report.AggregateMetadata `json:"metadata"`
 }
 
-// topologySelecter selects a single topology from a report.
-type topologySelecter func(r report.Report) report.Topology
-
-func selectProcess(r report.Report) report.Topology {
-	return r.Process
-}
-
-func selectNetwork(r report.Report) report.Topology {
-	return r.Network
-}
-
 // Full topology.
 func handleTopology(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
 	respondWith(w, http.StatusOK, APITopology{

app/origin_node.go

@@ -34,8 +34,8 @@ func originNodeForProcess(node report.NodeMetadata) OriginNode {
 		{Key: "Process name", ValueMajor: node["name"], ValueMinor: ""},
 	}
 	for _, tuple := range []struct{ key, human string }{
-		{"docker_id", "Container ID"},
-		{"docker_name", "Container name"},
+		{"docker_container_id", "Container ID"},
+		{"docker_container_name", "Container name"},
 		{"docker_image_id", "Container image ID"},
 		{"docker_image_name", "Container image name"},
 		{"cgroup", "cgroup"},

app/router.go

@@ -47,16 +47,16 @@ func apiHandler(w http.ResponseWriter, r *http.Request) {
 
 type topologyView struct {
 	human           string
-	selector        topologySelecter
+	selector        report.TopologySelector
 	mapper          report.MapFunc
 	pseudo          report.PseudoFunc
 	groupedTopology string
 }
 
 var topologyRegistry = map[string]topologyView{
-	"applications":         {"Applications", selectProcess, report.ProcessPID, report.GenericPseudoNode, "applications-grouped"},
-	"applications-grouped": {"Applications", selectProcess, report.ProcessName, report.GenericGroupedPseudoNode, ""},
-	"containers":           {"Containers", selectProcess, report.ProcessContainer, report.InternetOnlyPseudoNode, "containers-grouped"},
-	"containers-grouped":   {"Containers", selectProcess, report.ProcessContainerImage, report.InternetOnlyPseudoNode, ""},
-	"hosts":                {"Hosts", selectNetwork, report.NetworkHostname, report.GenericPseudoNode, ""},
+	"applications":         {"Applications", report.SelectProcess, report.ProcessPID, report.GenericPseudoNode, "applications-grouped"},
+	"applications-grouped": {"Applications", report.SelectProcess, report.ProcessName, report.GenericGroupedPseudoNode, ""},
+	"containers":           {"Containers", report.SelectProcess, report.ProcessContainer, report.InternetOnlyPseudoNode, "containers-grouped"},
+	"containers-grouped":   {"Containers", report.SelectProcess, report.ProcessContainerImage, report.InternetOnlyPseudoNode, ""},
+	"hosts":                {"Hosts", report.SelectNetwork, report.NetworkHostname, report.GenericPseudoNode, ""},
 }

probe/main.go

@@ -5,7 +5,6 @@ import (
 	"log"
 	"net"
 	"net/http"
-	_ "net/http/pprof"
 	"os"
 	"os/signal"
 	"runtime"
@@ -14,12 +13,12 @@ import (
 	"time"
 
 	"github.com/weaveworks/procspy"
+	"github.com/weaveworks/scope/probe/tag"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/xfer"
 )
 
-// Set during buildtime.
-var version = "unknown"
+var version = "dev" // set at build time
 
 func main() {
 	var (
@@ -29,9 +28,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)")
-		cgroupsRoot        = flag.String("cgroups.root", "", "if provided, enrich -processes with cgroup names from this root (e.g. /mnt/cgroups)")
-		cgroupsInterval    = flag.Duration("cgroups.interval", 10*time.Second, "how often to update cgroup names")
-		dockerMapper       = flag.Bool("docker", true, "collect Docker-related attributes for processes")
+		dockerTagger       = 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")
 	)
@@ -42,7 +39,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	log.Printf("probe starting, version %s", version)
+	log.Printf("probe version %s", version)
 
 	procspy.SetProcRoot(*procRoot)
 
@@ -66,32 +63,14 @@ func main() {
 	}
 	defer publisher.Close()
 
-	pms := []processMapper{identityMapper{}}
-
-	if *cgroupsRoot != "" {
-		if fi, err := os.Stat(*cgroupsRoot); err == nil && fi.IsDir() {
-			log.Printf("enriching -processes with cgroup names from %s", *cgroupsRoot)
-			cgroupMapper := newCgroupMapper(*cgroupsRoot, *cgroupsInterval)
-			defer cgroupMapper.Stop()
-			pms = append(pms, cgroupMapper)
-		} else {
-			log.Printf("-cgroups.root=%s: %v", *cgroupsRoot, err)
-		}
-	}
-
-	if *dockerMapper && runtime.GOOS == "Linux" {
-		docker, err := newDockerMapper(*procRoot, *dockerInterval)
+	taggers := []tag.Tagger{tag.NewTopologyTagger()}
+	if *dockerTagger {
+		t, err := tag.NewDockerTagger(*procRoot, *dockerInterval)
 		if err != nil {
-			log.Fatal(err)
+			log.Fatalf("failed to start docker tagger: %v", err)
 		}
-		defer docker.Stop()
-
-		pms = append(pms,
-			docker.idMapper(),
-			docker.nameMapper(),
-			docker.imageIDMapper(),
-			docker.imageNameMapper(),
-		)
+		defer t.Stop()
+		taggers = append(taggers, t)
 	}
 
 	log.Printf("listening on %s", *listen)
@@ -116,7 +95,8 @@ func main() {
 				r = report.MakeReport()
 
 			case <-spyTick:
-				r.Merge(spy(hostname, hostname, *spyProcs, pms))
+				r.Merge(spy(hostname, hostname, *spyProcs))
+				r = tag.Apply(r, taggers)
 				// log.Printf("merged report:\n%#v\n", r)
 
 			case <-quit: