Merge pull request #2559 from weaveworks/simplify-connection-join

simplify connection join

render/container.go

@@ -5,7 +5,6 @@ import (
 	"strings"
 
 	"github.com/weaveworks/scope/probe/docker"
-	"github.com/weaveworks/scope/probe/endpoint"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -35,23 +34,29 @@ var ContainerRenderer = MakeFilter(
 			ProcessRenderer,
 		),
 
-		// This mapper brings in short lived connections by joining with container IPs.
-		// We need to be careful to ensure we only include each edge once.  Edges brought in
-		// by the above renders will have a pid, so its enough to filter out any nodes with
-		// pids.
-		ShortLivedConnectionJoin(SelectContainer, MapContainer2IP),
+		// This mapper brings in connections by joining with container
+		// IPs.
+		ConnectionJoin(SelectContainer, MapContainer2IP),
 
 		SelectContainer,
 	),
 )
 
+var mapEndpoint2IP = MakeMap(
+	endpoint2IP,
+	// We drop endpoint nodes which were procspied or eBBF-tracked, as
+	// they will be joined to containers through the process topology,
+	// and we don't want to double count edges.
+	MakeFilter(Complement(procspiedOrEBPF), SelectEndpoint),
+)
+
 const originalNodeID = "original_node_id"
 const originalNodeTopology = "original_node_topology"
 
-// ShortLivedConnectionJoin joins the given renderer with short lived connections
-// from the endpoints topology, using the toIPs function to extract IPs from
+// ConnectionJoin joins the given renderer with connections from the
+// endpoints topology, using the toIPs function to extract IPs from
 // the nodes.
-func ShortLivedConnectionJoin(r Renderer, toIPs func(report.Node) []string) Renderer {
+func ConnectionJoin(r Renderer, toIPs func(report.Node) []string) Renderer {
 	nodeToIP := func(n report.Node, _ report.Networks) report.Nodes {
 		result := report.Nodes{}
 		for _, ip := range toIPs(n) {
@@ -66,82 +71,69 @@ func ShortLivedConnectionJoin(r Renderer, toIPs func(report.Node) []string) Rend
 		return result
 	}
 
-	ipToNode := func(n report.Node, _ report.Networks) report.Nodes {
-		// 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{}
-		}
-
-		// Propagate the internet and service pseudo nodes
-		if strings.HasSuffix(n.ID, TheInternetID) || strings.HasPrefix(n.ID, ServiceNodeIDPrefix) {
-			return report.Nodes{n.ID: n}
-		}
-
-		// 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{}
-		}
-		topology, ok := n.Latest.Lookup(originalNodeTopology)
-		if !ok {
-			return report.Nodes{}
-		}
-
-		return report.Nodes{
-			id: NewDerivedNode(id, n).
-				WithTopology(topology),
-		}
-	}
-
-	// MapEndpoint2IP maps endpoint nodes to their IP address, for joining
-	// with container nodes.  We drop endpoint nodes with pids, as they
-	// will be joined to containers through the process topology, and we
-	// don't want to double count edges.
-	endpoint2IP := func(m report.Node, local report.Networks) report.Nodes {
-		// Don't include procspied connections, to prevent double counting
-		_, ok := m.Latest.Lookup(endpoint.Procspied)
-		if ok {
-			return report.Nodes{}
-		}
-		scope, addr, port, ok := report.ParseEndpointNodeID(m.ID)
-		if !ok {
-			return report.Nodes{}
-		}
-
-		if externalNode, ok := NewDerivedExternalNode(m, addr, local); ok {
-			return report.Nodes{externalNode.ID: externalNode}
-		}
-
-		// 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),
-		}
-	}
-
 	return FilterUnconnected(MakeMap(
 		ipToNode,
 		MakeReduce(
-			MakeMap(
-				nodeToIP,
-				r,
-			),
-			MakeMap(
-				endpoint2IP,
-				SelectEndpoint,
-			),
+			MakeMap(nodeToIP, r),
+			mapEndpoint2IP,
 		),
 	))
 }
 
+func ipToNode(n report.Node, _ report.Networks) report.Nodes {
+	// 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{}
+	}
+
+	// Propagate the internet and service pseudo nodes
+	if strings.HasSuffix(n.ID, TheInternetID) || strings.HasPrefix(n.ID, ServiceNodeIDPrefix) {
+		return report.Nodes{n.ID: n}
+	}
+
+	// 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{}
+	}
+	topology, ok := n.Latest.Lookup(originalNodeTopology)
+	if !ok {
+		return report.Nodes{}
+	}
+
+	return report.Nodes{
+		id: NewDerivedNode(id, n).
+			WithTopology(topology),
+	}
+}
+
+// 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{}
+	}
+
+	if externalNode, ok := NewDerivedExternalNode(m, addr, local); ok {
+		return report.Nodes{externalNode.ID: externalNode}
+	}
+
+	// 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),
+	}
+}
+
 // FilterEmpty is a Renderer which filters out nodes which have no children
 // from the specified topology.
 func FilterEmpty(topology string, r Renderer) Renderer {

render/filters.go

@@ -237,30 +237,18 @@ func IsRunning(n report.Node) bool {
 // IsStopped checks if the node is *not* a running docker container
 var IsStopped = Complement(IsRunning)
 
-func nonProcspiedFilter(node report.Node) bool {
-	_, ok := node.Latest.Lookup(endpoint.Procspied)
-	return ok
-}
-
-func nonEBPFFilter(node report.Node) bool {
+func procspiedOrEBPF(node report.Node) bool {
+	if _, ok := node.Latest.Lookup(endpoint.Procspied); ok {
+		return true
+	}
 	_, ok := node.Latest.Lookup(endpoint.EBPF)
 	return ok
 }
 
-// FilterNonProcspied removes endpoints which were not found in procspy.
-func FilterNonProcspied(r Renderer) Renderer {
-	return MakeFilter(nonProcspiedFilter, r)
-}
-
-// FilterNonEBPF removes endpoints which were not found via eBPF.
-func FilterNonEBPF(r Renderer) Renderer {
-	return MakeFilter(nonEBPFFilter, r)
-}
-
-// FilterNonProcspiedNorEBPF removes endpoints which were not found in procspy
-// nor via eBPF.
-func FilterNonProcspiedNorEBPF(r Renderer) Renderer {
-	return MakeFilter(AnyFilterFunc(nonProcspiedFilter, nonEBPFFilter), r)
+// FilterProcspiedOrEBPF keeps only endpoints which were found via
+// procspy or eBPF.
+func FilterProcspiedOrEBPF(r Renderer) Renderer {
+	return MakeFilter(procspiedOrEBPF, r)
 }
 
 // IsApplication checks if the node is an "application" node

render/pod.go

@@ -44,7 +44,7 @@ var PodRenderer = ConditionalRenderer(renderKubernetesTopologies,
 						ContainerWithImageNameRenderer,
 					),
 				),
-				ShortLivedConnectionJoin(SelectPod, MapPod2IP),
+				ConnectionJoin(SelectPod, MapPod2IP),
 				SelectPod,
 			),
 		),

render/process.go

@@ -23,7 +23,7 @@ func renderProcesses(rpt report.Report) bool {
 }
 
 // EndpointRenderer is a Renderer which produces a renderable endpoint graph.
-var EndpointRenderer = FilterNonProcspiedNorEBPF(SelectEndpoint)
+var EndpointRenderer = FilterProcspiedOrEBPF(SelectEndpoint)
 
 // ProcessRenderer is a Renderer which produces a renderable process
 // graph by merging the endpoint graph and the process topology.