New asynchronous, caching DNS resolver for reverse resolutions

Add nodes for the remote side of connections iff we have a DNS reverse resolution for the IP.

probe/endpoint/reporter.go

@@ -14,6 +14,7 @@ import (
 
 // Node metadata keys.
 const (
+	Name = "name"
 	Addr = "addr" // typically IPv4
 	Port = "port"
 )
@@ -26,6 +27,7 @@ type Reporter struct {
 	includeNAT       bool
 	conntracker      *Conntracker
 	natmapper        *natmapper
+	revResolver      *ReverseResolver
 }
 
 // SpyDuration is an exported prometheus metric
@@ -64,12 +66,16 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
 			log.Printf("Failed to start natMapper: %v", err)
 		}
 	}
+
+	revRes := NewReverseResolver(rAddrCacheLen)
+
 	return &Reporter{
 		hostID:           hostID,
 		hostName:         hostName,
 		includeProcesses: includeProcesses,
 		conntracker:      conntracker,
 		natmapper:        natmapper,
+		revResolver:      revRes,
 	}
 }
 
@@ -81,6 +87,7 @@ func (r *Reporter) Stop() {
 	if r.natmapper != nil {
 		r.natmapper.Stop()
 	}
+	r.revResolver.Stop()
 }
 
 // Report implements Reporter.
@@ -147,6 +154,13 @@ func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr strin
 			})
 		)
 
+		// in case we have a reverse resolution for the IP, we can use it for the name...
+		if revRemoteName, err := r.revResolver.Get(remoteAddr); err != nil {
+			remoteNode = remoteNode.AddMetadata(map[string]string{
+				"name": revRemoteName,
+			})
+		}
+
 		if localIsClient {
 			localNode.Adjacency = localNode.Adjacency.Add(remoteAddressNodeID)
 			edgeID = report.MakeEdgeID(localAddressNodeID, remoteAddressNodeID)
@@ -157,6 +171,7 @@ func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr strin
 
 		rpt.Address = rpt.Address.WithNode(localAddressNodeID, localNode)
 		rpt.Address = rpt.Address.WithNode(remoteAddressNodeID, remoteNode)
+
 		countTCPConnection(rpt.Address.EdgeMetadatas, edgeID)
 	}
 
@@ -178,6 +193,13 @@ func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr strin
 			})
 		)
 
+		// in case we have a reverse resolution for the IP, we can use it for the name...
+		if revRemoteName, err := r.revResolver.Get(remoteAddr); err != nil {
+			remoteNode = remoteNode.AddMetadata(map[string]string{
+				"name": revRemoteName,
+			})
+		}
+
 		if localIsClient {
 			localNode.Adjacency = localNode.Adjacency.Add(remoteEndpointNodeID)
 			edgeID = report.MakeEdgeID(localEndpointNodeID, remoteEndpointNodeID)

probe/endpoint/reporter_test.go

@@ -54,7 +54,7 @@ var (
 			LocalAddress:  fixLocalAddress,
 			LocalPort:     fixLocalPort,
 			RemoteAddress: fixRemoteAddress,
-			RemotePort:    fixRemotePortB,
+			RemotePort:    fixRemotePort,
 			Proc: procspy.Proc{
 				PID:  fixProcessPID,
 				Name: fixProcessName,

probe/endpoint/resolver.go

@@ -0,0 +1,69 @@
+package endpoint
+
+import (
+	"net"
+	"time"
+
+	"github.com/bluele/gcache"
+)
+
+const (
+	rAddrCacheLen        = 500 // Default cache length
+	rAddrBacklog         = 1000
+	rAddrCacheExpiration = 30 * time.Minute
+)
+
+// ReverseResolver is a caching, reverse resolver
+type ReverseResolver struct {
+	addresses chan string
+	cache     gcache.Cache
+}
+
+// NewReverseResolver starts a new reverse resolver that
+// performs reverse resolutions and caches the result.
+func NewReverseResolver(cacheLen int) *ReverseResolver {
+	r := ReverseResolver{
+		addresses: make(chan string, rAddrBacklog),
+		cache:     gcache.New(cacheLen).LRU().Expiration(rAddrCacheExpiration).Build(),
+	}
+
+	go r.run()
+	return &r
+}
+
+// Get the reverse resolution for an IP address if already in the cache, gcache.NotFoundKeyError otherwise
+// Note: it returns one of the possible names that can be obtained for that IP
+func (r *ReverseResolver) Get(address string) (string, error) {
+	val, err := r.cache.Get(address)
+	if err == nil {
+		return val.(string), nil
+	}
+
+	if err == gcache.NotFoundKeyError {
+		// we trigger a asynchronous reverse resolution when not cached
+		select {
+		case r.addresses <- address:
+		default:
+		}
+	}
+	return "", err
+}
+
+func (r *ReverseResolver) run() {
+	throttle := time.Tick(time.Second / 10)
+	for address := range r.addresses {
+		<-throttle // rate limit our DNS resolutions
+		names, err := net.LookupAddr(address)
+		if err != nil {
+			continue
+		}
+		if len(names) > 0 {
+			r.cache.Set(address, names[0])
+		}
+	}
+}
+
+// Stop the async reverse resolver
+func (r *ReverseResolver) Stop() {
+	close(r.addresses)
+}

report/topology.go

@@ -184,6 +184,12 @@ func (n NodeMetadata) WithMetadata(m map[string]string) NodeMetadata {
 	return result
 }
 
+// AddMetadata returns a fresh copy of n, with Metadata set to the merge of n and the metadata provided
+func (n NodeMetadata) AddMetadata(m map[string]string) NodeMetadata {
+	additional := MakeNodeMetadataWith(m)
+	return n.Merge(additional)
+}
+
 // WithCounters returns a fresh copy of n, with Counters set to c
 func (n NodeMetadata) WithCounters(c map[string]int) NodeMetadata {
 	result := n.Copy()