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)
+
+	// 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()
 
-	// 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}
+	// 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
+}
 
-	// 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),
-	}
+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)
+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
+}
 
-	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}
+func (e endpoints2Hosts) Stats(rpt report.Report, _ Decorator) Stats {
+	return Stats{} // nothing to report
 }

render/id.go

@@ -36,27 +36,46 @@ 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
 	// internet pseudoNode
 	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