Merge pull request #2920 from weaveworks/endpoints-renderers

Optimisation: replace three map-reduces with Renderers

app/benchmark_internal_test.go

@@ -32,19 +32,43 @@ func loadReport() (report.Report, error) {
 }
 
 func BenchmarkTopologyList(b *testing.B) {
+	benchmarkRender(b, func(report report.Report) {
+		request := &http.Request{
+			Form: url.Values{},
+		}
+		topologyRegistry.renderTopologies(report, request)
+	})
+}
+
+func benchmarkRender(b *testing.B, f func(report.Report)) {
 	report, err := loadReport()
 	if err != nil {
 		b.Fatal(err)
 	}
 	b.ReportAllocs()
 	b.ResetTimer()
-	request := &http.Request{
-		Form: url.Values{},
-	}
 	for i := 0; i < b.N; i++ {
 		b.StopTimer()
 		render.ResetCache()
 		b.StartTimer()
-		topologyRegistry.renderTopologies(report, request)
+		f(report)
 	}
 }
+
+func BenchmarkTopologyHosts(b *testing.B) {
+	benchmarkOneTopology(b, "hosts")
+}
+
+func BenchmarkTopologyContainers(b *testing.B) {
+	benchmarkOneTopology(b, "containers")
+}
+
+func benchmarkOneTopology(b *testing.B, topologyID string) {
+	benchmarkRender(b, func(report report.Report) {
+		renderer, decorator, err := topologyRegistry.RendererForTopology(topologyID, url.Values{}, report)
+		if err != nil {
+			b.Fatal(err)
+		}
+		renderer.Render(report, decorator)
+	})
+}

render/container.go

@@ -39,85 +39,74 @@ var ContainerRenderer = Memoise(MakeFilter(
 	),
 ))
 
-var mapEndpoint2IP = Memoise(MakeMap(endpoint2IP, SelectEndpoint))
-
 const originalNodeID = "original_node_id"
 
 // ConnectionJoin joins the given renderer with connections from the
 // endpoints topology, using the toIPs function to extract IPs from
 // the nodes.
 func ConnectionJoin(toIPs func(report.Node) []string, r Renderer) Renderer {
-	nodeToIP := func(n report.Node, _ report.Networks) report.Nodes {
-		result := report.Nodes{}
-		for _, ip := range toIPs(n) {
-			result[ip] = report.MakeNode(ip).
-				WithTopology(IP).
-				WithLatests(map[string]string{
-					originalNodeID: n.ID,
-				}).
-				WithCounters(map[string]int{IP: 1})
-		}
-		return result
-	}
-
-	return MakeReduce(
-		MakeMap(
-			ipToNode,
-			MakeReduce(
-				MakeMap(nodeToIP, r),
-				mapEndpoint2IP,
-			),
-		),
-		r,
-	)
+	return connectionJoin{toIPs: toIPs, r: r}
 }
 
-func ipToNode(n report.Node, _ report.Networks) report.Nodes {
-	// propagate non-IP nodes
-	if n.Topology != IP {
-		return report.Nodes{n.ID: n}
-	}
-	// If an IP is shared between multiple nodes, we can't reliably
-	// attribute an connection based on its IP
-	if count, _ := n.Counters.Lookup(IP); count > 1 {
-		return report.Nodes{}
-	}
-	// If this node is not of the original type, exclude it.  This
-	// excludes all the nodes we've dragged in from endpoint that we
-	// failed to join to a node.
-	id, ok := n.Latest.Lookup(originalNodeID)
-	if !ok {
-		return report.Nodes{}
-	}
-
-	return report.Nodes{id: NewDerivedNode(id, n)}
+type connectionJoin struct {
+	toIPs func(report.Node) []string
+	r     Renderer
 }
 
-// endpoint2IP maps endpoint nodes to their IP address, for joining
-// with container nodes.
-func endpoint2IP(m report.Node, local report.Networks) report.Nodes {
-	scope, addr, port, ok := report.ParseEndpointNodeID(m.ID)
-	if !ok {
-		return report.Nodes{}
-	}
+func (c connectionJoin) Render(rpt report.Report, dct Decorator) report.Nodes {
+	local := LocalNetworks(rpt)
+	inputNodes := c.r.Render(rpt, dct)
+	endpoints := SelectEndpoint.Render(rpt, dct)
 
-	// Nodes without a hostid may be pseudo nodes
-	if _, ok := m.Latest.Lookup(report.HostNodeID); !ok {
-		if externalNode, ok := NewDerivedExternalNode(m, addr, local); ok {
-			return report.Nodes{externalNode.ID: externalNode}
+	// Collect all the IPs we are trying to map to, and which ID they map from
+	var ipNodes = map[string]string{}
+	for _, n := range inputNodes {
+		for _, ip := range c.toIPs(n) {
+			if _, exists := ipNodes[ip]; exists {
+				// If an IP is shared between multiple nodes, we can't reliably
+				// attribute an connection based on its IP
+				ipNodes[ip] = "" // blank out the mapping so we don't use it
+			} else {
+				ipNodes[ip] = n.ID
+			}
 		}
 	}
+	ret := newJoinResults()
 
-	// We also allow for joining on ip:port pairs.  This is useful for
-	// connections to the host IPs which have been port mapped to a
-	// container can only be unambiguously identified with the port.
-	// So we need to emit two nodes, for two different cases.
-	id := report.MakeScopedEndpointNodeID(scope, addr, "")
-	idWithPort := report.MakeScopedEndpointNodeID(scope, addr, port)
-	return report.Nodes{
-		id:         NewDerivedNode(id, m).WithTopology(IP),
-		idWithPort: NewDerivedNode(idWithPort, m).WithTopology(IP),
+	// Now look at all the endpoints and see which map to IP nodes
+	for _, m := range endpoints {
+		scope, addr, port, ok := report.ParseEndpointNodeID(m.ID)
+		if !ok {
+			continue
+		}
+		// Nodes without a hostid may be pseudo nodes - if so, pass through to result
+		if _, ok := m.Latest.Lookup(report.HostNodeID); !ok {
+			if id, ok := externalNodeID(m, addr, local); ok {
+				ret.addToResults(m, id, newPseudoNode)
+				continue
+			}
+		}
+		id, found := ipNodes[report.MakeScopedEndpointNodeID(scope, addr, "")]
+		// We also allow for joining on ip:port pairs.  This is useful for
+		// connections to the host IPs which have been port mapped to a
+		// container can only be unambiguously identified with the port.
+		if !found {
+			id, found = ipNodes[report.MakeScopedEndpointNodeID(scope, addr, port)]
+		}
+		if found && id != "" { // not one we blanked out earlier
+			ret.addToResults(m, id, func(id string) report.Node {
+				return inputNodes[id]
+			})
+		}
 	}
+	ret.copyUnmatched(inputNodes)
+	ret.fixupAdjacencies(inputNodes)
+	ret.fixupAdjacencies(endpoints)
+	return ret.nodes
+}
+
+func (c connectionJoin) Stats(rpt report.Report, _ Decorator) Stats {
+	return Stats{} // nothing to report
 }
 
 // FilterEmpty is a Renderer which filters out nodes which have no children

render/host.go

@@ -7,12 +7,8 @@ import (
 // HostRenderer is a Renderer which produces a renderable host
 // graph from the host topology.
 //
-// not memoised
-var HostRenderer = MakeReduce(
-	MakeMap(
-		MapEndpoint2Host,
-		EndpointRenderer,
-	),
+var HostRenderer = Memoise(MakeReduce(
+	endpoints2Hosts{},
 	MakeMap(
 		MapX2Host,
 		ColorConnectedProcessRenderer,
@@ -30,7 +26,7 @@ var HostRenderer = MakeReduce(
 		PodRenderer,
 	),
 	SelectHost,
-)
+))
 
 // MapX2Host maps any Nodes to host Nodes.
 //
@@ -64,18 +60,37 @@ func MapX2Host(n report.Node, _ report.Networks) report.Nodes {
 	return result
 }
 
-// MapEndpoint2Host takes nodes from the endpoint topology and produces
+// endpoints2Hosts takes nodes from the endpoint topology and produces
 // host nodes or pseudo nodes.
-func MapEndpoint2Host(n report.Node, local report.Networks) report.Nodes {
-	// Nodes without a hostid are treated as pseudo nodes
-	hostNodeID, timestamp, ok := n.Latest.LookupEntry(report.HostNodeID)
-	if !ok {
-		return MapEndpoint2Pseudo(n, local)
-	}
-
-	id := report.MakeHostNodeID(report.ExtractHostID(n))
-	result := NewDerivedNode(id, n).WithTopology(report.Host)
-	result.Latest = result.Latest.Set(report.HostNodeID, timestamp, hostNodeID)
-	result.Counters = result.Counters.Add(n.Topology, 1)
-	return report.Nodes{id: result}
+type endpoints2Hosts struct {
+}
+
+func (e endpoints2Hosts) Render(rpt report.Report, dct Decorator) report.Nodes {
+	local := LocalNetworks(rpt)
+	endpoints := SelectEndpoint.Render(rpt, dct)
+	ret := newJoinResults()
+
+	for _, n := range endpoints {
+		// Nodes without a hostid are treated as pseudo nodes
+		hostNodeID, timestamp, ok := n.Latest.LookupEntry(report.HostNodeID)
+		if !ok {
+			id, ok := pseudoNodeID(n, local)
+			if !ok {
+				continue
+			}
+			ret.addToResults(n, id, newPseudoNode)
+		} else {
+			id := report.MakeHostNodeID(report.ExtractHostID(n))
+			ret.addToResults(n, id, func(id string) report.Node {
+				return report.MakeNode(id).WithTopology(report.Host).
+					WithLatest(report.HostNodeID, timestamp, hostNodeID)
+			})
+		}
+	}
+	ret.fixupAdjacencies(endpoints)
+	return ret.nodes
+}
+
+func (e endpoints2Hosts) Stats(rpt report.Report, _ Decorator) Stats {
+	return Stats{} // nothing to report
 }

render/id.go

@@ -36,13 +36,32 @@ func NewDerivedPseudoNode(id string, node report.Node) report.Node {
 	return output
 }
 
-// NewDerivedExternalNode figures out if a node should be considered external and creates the corresponding pseudo node
-func NewDerivedExternalNode(n report.Node, addr string, local report.Networks) (report.Node, bool) {
+func newPseudoNode(id string) report.Node {
+	return report.MakeNode(id).WithTopology(Pseudo)
+}
+
+func pseudoNodeID(n report.Node, local report.Networks) (string, bool) {
+	_, addr, _, ok := report.ParseEndpointNodeID(n.ID)
+	if !ok {
+		return "", false
+	}
+
+	if id, ok := externalNodeID(n, addr, local); ok {
+		return id, ok
+	}
+
+	// due to https://github.com/weaveworks/scope/issues/1323 we are dropping
+	// all non-external pseudo nodes for now.
+	return "", false
+}
+
+// figure out if a node should be considered external and returns an ID which can be used to create a pseudo node
+func externalNodeID(n report.Node, addr string, local report.Networks) (string, bool) {
 	// First, check if it's a known service and emit a a specific node if it
 	// is. This needs to be done before checking IPs since known services can
 	// live in the same network, see https://github.com/weaveworks/scope/issues/2163
 	if hostname, found := DNSFirstMatch(n, isKnownService); found {
-		return NewDerivedPseudoNode(ServiceNodeIDPrefix+hostname, n), true
+		return ServiceNodeIDPrefix + hostname, true
 	}
 
 	// If the dstNodeAddr is not in a network local to this report, we emit an
@@ -50,13 +69,13 @@ func NewDerivedExternalNode(n report.Node, addr string, local report.Networks) (
 	if ip := net.ParseIP(addr); ip != nil && !local.Contains(ip) {
 		// emit one internet node for incoming, one for outgoing
 		if len(n.Adjacency) > 0 {
-			return NewDerivedPseudoNode(IncomingInternetID, n), true
+			return IncomingInternetID, true
 		}
-		return NewDerivedPseudoNode(OutgoingInternetID, n), true
+		return OutgoingInternetID, true
 	}
 
 	// The node is not external
-	return report.Node{}, false
+	return "", false
 }
 
 // DNSFirstMatch returns the first DNS name where match() returns

render/process.go

@@ -26,15 +26,7 @@ var EndpointRenderer = SelectEndpoint
 
 // ProcessRenderer is a Renderer which produces a renderable process
 // graph by merging the endpoint graph and the process topology.
-var ProcessRenderer = Memoise(ConditionalRenderer(renderProcesses,
-	MakeReduce(
-		MakeMap(
-			MapEndpoint2Process,
-			EndpointRenderer,
-		),
-		SelectProcess,
-	),
-))
+var ProcessRenderer = Memoise(endpoints2Processes{})
 
 // ColorConnectedProcessRenderer colors connected nodes from
 // ProcessRenderer. Since the process topology views only show
@@ -89,56 +81,59 @@ var ProcessNameRenderer = ConditionalRenderer(renderProcesses,
 	),
 )
 
-// MapEndpoint2Pseudo makes internet of host pesudo nodes from a endpoint node.
-func MapEndpoint2Pseudo(n report.Node, local report.Networks) report.Nodes {
-	_, addr, _, ok := report.ParseEndpointNodeID(n.ID)
-	if !ok {
-		return report.Nodes{}
-	}
-
-	if externalNode, ok := NewDerivedExternalNode(n, addr, local); ok {
-		return report.Nodes{externalNode.ID: externalNode}
-	}
-
-	// due to https://github.com/weaveworks/scope/issues/1323 we are dropping
-	// all non-external pseudo nodes for now.
-	return report.Nodes{}
+// endpoints2Processes joins the endpoint topology to the process
+// topology, matching on hostID and pid.
+type endpoints2Processes struct {
 }
 
-// MapEndpoint2Process maps endpoint Nodes to process
-// Nodes.
-//
-// If this function is given a pseudo node, then it will just return it;
-// Pseudo nodes will never have pids in them, and therefore will never
-// be able to be turned into a Process node.
-//
-// Otherwise, this function will produce a node with the correct ID
-// format for a process, but without any Major or Minor labels.
-// It does not have enough info to do that, and the resulting graph
-// must be merged with a process graph to get that info.
-func MapEndpoint2Process(n report.Node, local report.Networks) report.Nodes {
-	// Nodes without a hostid are treated as pseudo nodes
-	if _, ok := n.Latest.Lookup(report.HostNodeID); !ok {
-		return MapEndpoint2Pseudo(n, local)
-	}
-
-	pid, timestamp, ok := n.Latest.LookupEntry(process.PID)
-	if !ok {
+func (e endpoints2Processes) Render(rpt report.Report, dct Decorator) report.Nodes {
+	if len(rpt.Process.Nodes) == 0 {
 		return report.Nodes{}
 	}
+	local := LocalNetworks(rpt)
+	processes := SelectProcess.Render(rpt, dct)
+	endpoints := SelectEndpoint.Render(rpt, dct)
+	ret := newJoinResults()
 
-	if len(n.Adjacency) > 1 {
-		// We cannot be sure that the pid is associated with all the
-		// connections. It is better to drop such an endpoint than
-		// risk rendering bogus connections.
-		return report.Nodes{}
+	for _, n := range endpoints {
+		// Nodes without a hostid are treated as pseudo nodes
+		if hostNodeID, ok := n.Latest.Lookup(report.HostNodeID); !ok {
+			if id, ok := pseudoNodeID(n, local); ok {
+				ret.addToResults(n, id, newPseudoNode)
+			}
+		} else {
+			pid, timestamp, ok := n.Latest.LookupEntry(process.PID)
+			if !ok {
+				continue
+			}
+
+			if len(n.Adjacency) > 1 {
+				// We cannot be sure that the pid is associated with all the
+				// connections. It is better to drop such an endpoint than
+				// risk rendering bogus connections.
+				continue
+			}
+
+			hostID, _, _ := report.ParseNodeID(hostNodeID)
+			id := report.MakeProcessNodeID(hostID, pid)
+			ret.addToResults(n, id, func(id string) report.Node {
+				if processNode, found := processes[id]; found {
+					return processNode
+				}
+				// we have a pid, but no matching process node; create a new one rather than dropping the data
+				return report.MakeNode(id).WithTopology(report.Process).
+					WithLatest(process.PID, timestamp, pid)
+			})
+		}
 	}
+	ret.copyUnmatched(processes)
+	ret.fixupAdjacencies(processes)
+	ret.fixupAdjacencies(endpoints)
+	return ret.nodes
+}
 
-	id := report.MakeProcessNodeID(report.ExtractHostID(n), pid)
-	node := NewDerivedNode(id, n).WithTopology(report.Process)
-	node.Latest = node.Latest.Set(process.PID, timestamp, pid)
-	node.Counters = node.Counters.Add(n.Topology, 1)
-	return report.Nodes{id: node}
+func (e endpoints2Processes) Stats(rpt report.Report, _ Decorator) Stats {
+	return Stats{} // nothing to report
 }
 
 // MapProcess2Name maps process Nodes to Nodes

render/render.go

@@ -207,3 +207,55 @@ func (c ConstantRenderer) Render(_ report.Report, _ Decorator) report.Nodes {
 func (c ConstantRenderer) Stats(_ report.Report, _ Decorator) Stats {
 	return Stats{}
 }
+
+// joinResults is used by Renderers that join sets of nodes
+type joinResults struct {
+	nodes  report.Nodes
+	mapped map[string]string // input node ID -> output node ID
+}
+
+func newJoinResults() joinResults {
+	return joinResults{nodes: make(report.Nodes), mapped: map[string]string{}}
+}
+
+// Add Node M under id, creating a new result node if not already there
+// and updating the mapping from old ID to new ID
+// Note we do not update any counters for child topologies here, because addToResults
+// is only ever called when m is an endpoint and we never look at endpoint counts
+func (ret *joinResults) addToResults(m report.Node, id string, create func(string) report.Node) {
+	result, exists := ret.nodes[id]
+	if !exists {
+		result = create(id)
+	}
+	result.Children = result.Children.Add(m)
+	result.Children = result.Children.Merge(m.Children)
+	ret.nodes[id] = result
+	ret.mapped[m.ID] = id
+}
+
+// Rewrite Adjacency for new nodes in ret for original nodes in input
+func (ret *joinResults) fixupAdjacencies(input report.Nodes) {
+	for _, n := range input {
+		outID, ok := ret.mapped[n.ID]
+		if !ok {
+			continue
+		}
+		out := ret.nodes[outID]
+		// for each adjacency in the original node, find out what it maps to (if any),
+		// and add that to the new node
+		for _, a := range n.Adjacency {
+			if mappedDest, found := ret.mapped[a]; found {
+				out.Adjacency = out.Adjacency.Add(mappedDest)
+			}
+		}
+		ret.nodes[outID] = out
+	}
+}
+
+func (ret *joinResults) copyUnmatched(input report.Nodes) {
+	for _, n := range input {
+		if _, found := ret.nodes[n.ID]; !found {
+			ret.nodes[n.ID] = n
+		}
+	}
+}