Merge pull request #218 from tomwilkie/171-map-map-merge

All merging of RenderableNodes, such that we can merge multiple topologies.

app/api_topologies.go

@@ -42,14 +42,14 @@ func makeTopologyList(rep Reporter) func(w http.ResponseWriter, r *http.Request)
 				Name:       def.human,
 				URL:        url,
 				GroupedURL: groupedURL,
-				Stats:      stats(def.selector(rpt).RenderBy(def.mapper, def.pseudo)),
+				Stats:      stats(render(rpt, def.maps)),
 			})
 		}
 		respondWith(w, http.StatusOK, a)
 	}
 }
 
-func stats(r map[string]report.RenderableNode) topologyStats {
+func stats(r report.RenderableNodes) topologyStats {
 	var (
 		nodes     int
 		realNodes int

app/api_topology.go

@@ -17,7 +17,7 @@ const (
 
 // APITopology is returned by the /api/topology/{name} handler.
 type APITopology struct {
-	Nodes map[string]report.RenderableNode `json:"nodes"`
+	Nodes report.RenderableNodes `json:"nodes"`
 }
 
 // APINode is returned by the /api/topology/{name}/{id} handler.
@@ -30,10 +30,19 @@ type APIEdge struct {
 	Metadata report.AggregateMetadata `json:"metadata"`
 }
 
+func render(rpt report.Report, maps []topologyMapper) report.RenderableNodes {
+	result := report.RenderableNodes{}
+	for _, m := range maps {
+		rns := m.selector(rpt).RenderBy(m.mapper, m.pseudo)
+		result.Merge(rns)
+	}
+	return result
+}
+
 // Full topology.
 func handleTopology(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Request) {
 	respondWith(w, http.StatusOK, APITopology{
-		Nodes: t.selector(rep.Report()).RenderBy(t.mapper, t.pseudo),
+		Nodes: render(rep.Report(), t.maps),
 	})
 }
 
@@ -60,7 +69,7 @@ func handleNode(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Req
 		vars     = mux.Vars(r)
 		nodeID   = vars["id"]
 		rpt      = rep.Report()
-		node, ok = t.selector(rpt).RenderBy(t.mapper, t.pseudo)[nodeID]
+		node, ok = render(rpt, t.maps)[nodeID]
 	)
 	if !ok {
 		http.NotFound(w, r)
@@ -76,8 +85,13 @@ func handleEdge(rep Reporter, t topologyView, w http.ResponseWriter, r *http.Req
 		localID  = vars["local"]
 		remoteID = vars["remote"]
 		rpt      = rep.Report()
-		metadata = t.selector(rpt).EdgeMetadata(t.mapper, localID, remoteID).Transform()
+		metadata = report.AggregateMetadata{}
 	)
+
+	for _, m := range t.maps {
+		metadata.Merge(m.selector(rpt).EdgeMetadata(m.mapper, localID, remoteID).Transform())
+	}
+
 	respondWith(w, http.StatusOK, APIEdge{Metadata: metadata})
 }
 
@@ -110,11 +124,11 @@ func handleWebsocket(
 	}(conn)
 
 	var (
-		previousTopo map[string]report.RenderableNode
+		previousTopo report.RenderableNodes
 		tick         = time.Tick(loop)
 	)
 	for {
-		newTopo := t.selector(rep.Report()).RenderBy(t.mapper, t.pseudo)
+		newTopo := render(rep.Report(), t.maps)
 		diff := report.TopoDiff(previousTopo, newTopo)
 		previousTopo = newTopo
 

app/router.go

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

probe/tag/docker_tagger.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	stop  = "stop"
+	die   = "die"
 	start = "start"
 )
 
@@ -206,7 +206,7 @@ func (t *DockerTagger) updateImages(client dockerClient) error {
 
 func (t *DockerTagger) handleEvent(event *docker.APIEvents, client dockerClient) {
 	switch event.Status {
-	case stop:
+	case die:
 		containerID := event.ID
 		t.Lock()
 		if container, ok := t.containers[containerID]; ok {
@@ -299,9 +299,8 @@ func (t *DockerTagger) Tag(r report.Report) report.Report {
 		}
 
 		md := report.NodeMetadata{
-			ContainerID:   container.ID,
-			ContainerName: strings.TrimPrefix(container.Name, "/"),
-			ImageID:       container.Image,
+			ContainerID: container.ID,
+			ImageID:     container.Image,
 		}
 
 		t.RLock()

probe/tag/docker_tagger_test.go

@@ -67,8 +67,13 @@ func TestDockerTagger(t *testing.T) {
 	}
 
 	var (
-		endpoint1NodeID     = "somehost.com;192.168.1.1;12345"
-		endpoint2NodeID     = "somehost.com;192.168.1.1;67890"
+		endpoint1NodeID      = "somehost.com;192.168.1.1;12345"
+		endpoint2NodeID      = "somehost.com;192.168.1.1;67890"
+		endpointNodeMetadata = report.NodeMetadata{
+			ContainerID: "foo",
+			ImageID:     "baz",
+			ImageName:   "bang",
+		}
 		processNodeMetadata = report.NodeMetadata{
 			ContainerID:   "foo",
 			ContainerName: "bar",
@@ -84,7 +89,7 @@ func TestDockerTagger(t *testing.T) {
 	dockerTagger, _ := NewDockerTagger("/irrelevant", 10*time.Second)
 	runtime.Gosched()
 	for _, endpointNodeID := range []string{endpoint1NodeID, endpoint2NodeID} {
-		want := processNodeMetadata.Copy()
+		want := endpointNodeMetadata.Copy()
 		have := dockerTagger.Tag(r).Endpoint.NodeMetadatas[endpointNodeID].Copy()
 		delete(have, "pid")
 		if !reflect.DeepEqual(want, have) {

report/detailed_node.go

@@ -71,10 +71,6 @@ func endpointOriginTable(nmd NodeMetadata) (Table, bool) {
 		{"host_name", "Host name"},
 		{"pid", "PID"},
 		{"name", "Process name"},
-		{"docker_container_id", "Container ID"},
-		{"docker_container_name", "Container name"},
-		{"docker_image_id", "Container image ID"},
-		{"docker_image_name", "Container image name"},
 	} {
 		if val, ok := nmd[tuple.key]; ok {
 			rows = append(rows, Row{Key: tuple.human, ValueMajor: val, ValueMinor: ""})

report/mapping.go

@@ -46,6 +46,11 @@ func SelectAddress(r Report) Topology {
 	return r.Address
 }
 
+// SelectContainer selects the container topology.
+func SelectContainer(r Report) Topology {
+	return r.Container
+}
+
 // ProcessPID takes a node NodeMetadata from topology, and returns a
 // representation with the ID based on the process PID and the labels based on
 // the process name.
@@ -77,11 +82,28 @@ func ProcessName(_ string, m NodeMetadata) (MappedNode, bool) {
 	}, show
 }
 
-// ProcessContainer maps topology nodes to the containers they run in. We
-// consider container and image IDs to be globally unique, and so don't scope
-// them further by e.g. host. If no container metadata is found, nodes are
+// MapEndpoint2Container maps endpoint topology nodes to the containers they run
+// in. We consider container and image IDs to be globally unique, and so don't
+// scope them further by e.g. host. If no container metadata is found, nodes are
 // grouped into the Uncontained node.
-func ProcessContainer(_ string, m NodeMetadata) (MappedNode, bool) {
+func MapEndpoint2Container(_ string, m NodeMetadata) (MappedNode, bool) {
+	var id, major, minor, rank string
+	if m["docker_container_id"] == "" {
+		id, major, minor, rank = "uncontained", "Uncontained", "", "uncontained"
+	} else {
+		id, major, minor, rank = m["docker_container_id"], "", m["domain"], ""
+	}
+
+	return MappedNode{
+		ID:    id,
+		Major: major,
+		Minor: minor,
+		Rank:  rank,
+	}, true
+}
+
+// MapContainerIdentity maps container topology node to container mapped nodes.
+func MapContainerIdentity(_ string, m NodeMetadata) (MappedNode, bool) {
 	var id, major, minor, rank string
 	if m["docker_container_id"] == "" {
 		id, major, minor, rank = "uncontained", "Uncontained", "", "uncontained"

report/mapping_test.go

@@ -52,7 +52,7 @@ func TestUngroupedMapping(t *testing.T) {
 			wantRank:  "42",
 		},
 		{
-			f:  ProcessContainer,
+			f:  MapEndpoint2Container,
 			id: "foo-id",
 			meta: NodeMetadata{
 				"pid":    "42",
@@ -66,7 +66,7 @@ func TestUngroupedMapping(t *testing.T) {
 			wantRank:  "uncontained",
 		},
 		{
-			f:  ProcessContainer,
+			f:  MapEndpoint2Container,
 			id: "bar-id",
 			meta: NodeMetadata{
 				"pid":                   "42",
@@ -79,9 +79,9 @@ func TestUngroupedMapping(t *testing.T) {
 			},
 			wantOK:    true,
 			wantID:    "d321fe0",
-			wantMajor: "walking_sparrow",
+			wantMajor: "",
 			wantMinor: "hosta",
-			wantRank:  "1101fff",
+			wantRank:  "",
 		},
 	} {
 		identity := fmt.Sprintf("(%d %s %v)", i, c.id, c.meta)

report/merge.go

@@ -78,3 +78,39 @@ func (m *EdgeMetadata) Flatten(other EdgeMetadata) {
 		m.MaxConnCountTCP += other.MaxConnCountTCP
 	}
 }
+
+// Merge merges two sets of RenderableNodes
+func (rns RenderableNodes) Merge(other RenderableNodes) {
+	for key, value := range other {
+		if existing, ok := rns[key]; ok {
+			existing.Merge(value)
+			rns[key] = existing
+		} else {
+			rns[key] = value
+		}
+	}
+}
+
+// Merge merges in another RenderableNode
+func (rn *RenderableNode) Merge(other RenderableNode) {
+	if rn.LabelMajor == "" {
+		rn.LabelMajor = other.LabelMajor
+	}
+
+	if rn.LabelMinor == "" {
+		rn.LabelMinor = other.LabelMinor
+	}
+
+	if rn.Rank == "" {
+		rn.Rank = other.Rank
+	}
+
+	if rn.Pseudo != other.Pseudo {
+		panic(rn.ID)
+	}
+
+	rn.Adjacency = rn.Adjacency.Add(other.Adjacency...)
+	rn.Origins = rn.Origins.Add(other.Origins...)
+
+	rn.Metadata.Merge(other.Metadata)
+}

report/report.go

@@ -47,6 +47,9 @@ type RenderableNode struct {
 	Metadata   AggregateMetadata `json:"metadata"`              // Numeric sums
 }
 
+// RenderableNodes is a set of RenderableNodes
+type RenderableNodes map[string]RenderableNode
+
 // DetailedNode is the data type that's yielded to the JavaScript layer when
 // we want deep information about an individual node.
 type DetailedNode struct {

report/topology.go

@@ -82,15 +82,15 @@ func NewTopology() Topology {
 //
 // RenderBy takes a a MapFunc, which defines how to group and label nodes. Npdes
 // with the same mapped IDs will be merged.
-func (t Topology) RenderBy(mapFunc MapFunc, pseudoFunc PseudoFunc) map[string]RenderableNode {
-	nodes := map[string]RenderableNode{}
+func (t Topology) RenderBy(mapFunc MapFunc, pseudoFunc PseudoFunc) RenderableNodes {
+	nodes := RenderableNodes{}
 
 	// Build a set of RenderableNodes for all non-pseudo probes, and an
 	// addressID to nodeID lookup map. Multiple addressIDs can map to the same
 	// RenderableNodes.
 	address2mapped := map[string]string{}
-	for addressID, metadata := range t.NodeMetadatas {
-		mapped, ok := mapFunc(addressID, metadata)
+	for nodeID, metadata := range t.NodeMetadatas {
+		mapped, ok := mapFunc(nodeID, metadata)
 		if !ok {
 			continue
 		}
@@ -104,11 +104,10 @@ func (t Topology) RenderBy(mapFunc MapFunc, pseudoFunc PseudoFunc) map[string]Re
 			LabelMinor: mapped.Minor,
 			Rank:       mapped.Rank,
 			Pseudo:     false,
-			Adjacency:  IDList{},            // later
-			Origins:    IDList{},            // later
+			Origins:    IDList{nodeID},
 			Metadata:   AggregateMetadata{}, // later
 		}
-		address2mapped[addressID] = mapped.ID
+		address2mapped[nodeID] = mapped.ID
 	}
 
 	// Walk the graph and make connections.
@@ -236,7 +235,7 @@ type Diff struct {
 }
 
 // TopoDiff gives you the diff to get from A to B.
-func TopoDiff(a, b map[string]RenderableNode) Diff {
+func TopoDiff(a, b RenderableNodes) Diff {
 	diff := Diff{}
 
 	notSeen := map[string]struct{}{}

report/topology_test.go

@@ -139,7 +139,7 @@ var (
 )
 
 func TestRenderByEndpointPID(t *testing.T) {
-	want := map[string]RenderableNode{
+	want := RenderableNodes{
 		"pid:client-54001-domain:10001": {
 			ID:         "pid:client-54001-domain:10001",
 			LabelMajor: "curl",
@@ -207,7 +207,7 @@ func TestRenderByEndpointPIDGrouped(t *testing.T) {
 	// For grouped, I've somewhat arbitrarily chosen to squash together all
 	// processes with the same name by removing the PID and domain (host)
 	// dimensions from the ID. That could be changed.
-	want := map[string]RenderableNode{
+	want := RenderableNodes{
 		"curl": {
 			ID:         "curl",
 			LabelMajor: "curl",
@@ -258,7 +258,7 @@ func TestRenderByEndpointPIDGrouped(t *testing.T) {
 }
 
 func TestRenderByNetworkHostname(t *testing.T) {
-	want := map[string]RenderableNode{
+	want := RenderableNodes{
 		"host:client.hostname.com": {
 			ID:         "host:client.hostname.com",
 			LabelMajor: "client",       // before first .
@@ -333,8 +333,8 @@ func TestTopoDiff(t *testing.T) {
 	}
 
 	// Helper to make RenderableNode maps.
-	nodes := func(ns ...RenderableNode) map[string]RenderableNode {
-		r := map[string]RenderableNode{}
+	nodes := func(ns ...RenderableNode) RenderableNodes {
+		r := RenderableNodes{}
 		for _, n := range ns {
 			r[n.ID] = n
 		}