Always include endpoint nodes brought in by procspy, even if they get merged with conntracked endpoints.

probe/endpoint/reporter.go

@@ -17,6 +17,7 @@ const (
 	Addr        = "addr" // typically IPv4
 	Port        = "port"
 	Conntracked = "conntracked"
+	Procspied   = "procspied"
 )
 
 // Reporter generates Reports containing the Endpoint topology.
@@ -95,26 +96,31 @@ func (r *Reporter) Report() (report.Report, error) {
 
 	hostNodeID := report.MakeHostNodeID(r.hostID)
 	rpt := report.MakeReport()
-	conns, err := procspy.Connections(r.includeProcesses)
-	if err != nil {
-		return rpt, err
-	}
 
-	for conn := conns.Next(); conn != nil; conn = conns.Next() {
-		var (
-			localPort  = conn.LocalPort
-			remotePort = conn.RemotePort
-			localAddr  = conn.LocalAddress.String()
-			remoteAddr = conn.RemoteAddress.String()
-		)
-		extraNodeInfo := report.MakeNode()
-		if conn.Proc.PID > 0 {
-			extraNodeInfo = extraNodeInfo.WithMetadata(report.Metadata{
-				process.PID:       strconv.FormatUint(uint64(conn.Proc.PID), 10),
-				report.HostNodeID: hostNodeID,
-			})
+	{
+		conns, err := procspy.Connections(r.includeProcesses)
+		if err != nil {
+			return rpt, err
+		}
+		commonNodeInfo := report.MakeNode().WithMetadata(report.Metadata{
+			Procspied: "true",
+		})
+		for conn := conns.Next(); conn != nil; conn = conns.Next() {
+			var (
+				localPort  = conn.LocalPort
+				remotePort = conn.RemotePort
+				localAddr  = conn.LocalAddress.String()
+				remoteAddr = conn.RemoteAddress.String()
+			)
+			extraNodeInfo := commonNodeInfo.Copy()
+			if conn.Proc.PID > 0 {
+				extraNodeInfo = extraNodeInfo.WithMetadata(report.Metadata{
+					process.PID:       strconv.FormatUint(uint64(conn.Proc.PID), 10),
+					report.HostNodeID: hostNodeID,
+				})
+			}
+			r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &extraNodeInfo, &commonNodeInfo)
 		}
-		r.addConnection(&rpt, localAddr, remoteAddr, localPort, remotePort, &extraNodeInfo, nil)
 	}
 
 	if r.conntracker != nil {
@@ -136,7 +142,7 @@ func (r *Reporter) Report() (report.Report, error) {
 		r.natmapper.applyNAT(rpt, r.hostID)
 	}
 
-	return rpt, err
+	return rpt, nil
 }
 
 func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr string, localPort, remotePort uint16, extraLocalNode, extraRemoteNode *report.Node) {

render/mapping.go

@@ -48,10 +48,14 @@ func MapEndpointIdentity(m RenderableNode, local report.Networks) RenderableNode
 	}
 
 	// We need to filter out short lived connections from this view,
-	// if they don't also have a pid; see #447
+	// if they don't also have a pid; see #447.  Note if they
+	// we're introduced by proc spy then they are guaranteed to
+	// have a pid on the other end of the adjacency, so we include them
+	// no matter what.
 	pid, pidOK := m.Metadata[process.PID]
 	_, conntracked := m.Metadata[endpoint.Conntracked]
-	if !pidOK && conntracked {
+	_, procspied := m.Metadata[endpoint.Procspied]
+	if !procspied && !pidOK && conntracked {
 		return RenderableNodes{}
 	}
 
@@ -64,8 +68,8 @@ func MapEndpointIdentity(m RenderableNode, local report.Networks) RenderableNode
 		}
 
 		// We are a 'client' pseudo node if the port is in the ephemeral port range.
-		// Linux uses 32768 to 61000.
-		if p, err := strconv.Atoi(port); err == nil && len(m.Adjacency) > 0 && p >= 32768 && p < 61000 {
+		// Linux uses 32768 to 61000, IANA suggests 49152 to 65535.
+		if p, err := strconv.Atoi(port); err == nil && len(m.Adjacency) > 0 && p >= 32768 && p < 65535 {
 			// We only exist if there is something in our adjacency
 			// Generate a single pseudo node for every (client ip, server ip, server port)
 			dstNodeID := m.Adjacency[0]

report/topology.go

@@ -102,7 +102,7 @@ func MakeNodeWith(m map[string]string) Node {
 // WithMetadata returns a fresh copy of n, with Metadata set to m
 func (n Node) WithMetadata(m map[string]string) Node {
 	result := n.Copy()
-	result.Metadata = m
+	result.Metadata = result.Metadata.Merge(m)
 	return result
 }