Skip to content

Commit

Permalink
New asynchronous, caching DNS resolver for reverse resolutions
Browse files Browse the repository at this point in the history
Add nodes for the remote side of connections iff we have a DNS reverse resolution for the IP.
  • Loading branch information
inercia committed Sep 1, 2015
1 parent e867d16 commit 160e0e7
Show file tree
Hide file tree
Showing 4 changed files with 89 additions and 6 deletions.
14 changes: 12 additions & 2 deletions probe/endpoint/reporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import (

// Node metadata keys.
const (
Name = "name"
Addr = "addr" // typically IPv4
Port = "port"
)
Expand All @@ -26,6 +27,7 @@ type Reporter struct {
includeNAT bool
conntracker *Conntracker
natmapper *natmapper
revResolver *ReverseResolver
}

// SpyDuration is an exported prometheus metric
Expand All @@ -45,7 +47,7 @@ var SpyDuration = prometheus.NewSummaryVec(
// on the host machine, at the granularity of host and port. That information
// is stored in the Endpoint topology. It optionally enriches that topology
// with process (PID) information.
func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool) *Reporter {
func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, revRes *ReverseResolver) *Reporter {
var (
conntrackModulePresent = ConntrackModulePresent()
conntracker *Conntracker
Expand All @@ -70,6 +72,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
includeProcesses: includeProcesses,
conntracker: conntracker,
natmapper: natmapper,
revResolver: revRes,
}
}

Expand Down Expand Up @@ -132,6 +135,11 @@ func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr strin

// Update address topology
{
revRemoteAddr := ""
if r, err := r.revResolver.Get(remoteAddr); err != nil {
revRemoteAddr = r
}

var (
localAddressNodeID = report.MakeAddressNodeID(r.hostID, localAddr)
remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, remoteAddr)
Expand All @@ -143,7 +151,8 @@ func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr strin
report.HostNodeID: hostNodeID,
})
remoteNode = report.MakeNodeMetadataWith(map[string]string{
Addr: remoteAddr,
"name": revRemoteAddr,
Addr: remoteAddr,
})
)

Expand All @@ -157,6 +166,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)
}

Expand Down
8 changes: 5 additions & 3 deletions probe/endpoint/reporter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ var (
LocalAddress: fixLocalAddress,
LocalPort: fixLocalPort,
RemoteAddress: fixRemoteAddress,
RemotePort: fixRemotePortB,
RemotePort: fixRemotePort,
Proc: procspy.Proc{
PID: fixProcessPID,
Name: fixProcessName,
Expand All @@ -71,7 +71,8 @@ func TestSpyNoProcesses(t *testing.T) {
nodeName = "frenchs-since-1904" // TODO rename to hostNmae
)

reporter := endpoint.NewReporter(nodeID, nodeName, false, false)
revResolver := endpoint.NewReverseResolver(endpoint.RAddrCacheLen)
reporter := endpoint.NewReporter(nodeID, nodeName, false, false, revResolver)
r, _ := reporter.Report()
//buf, _ := json.MarshalIndent(r, "", " ")
//t.Logf("\n%s\n", buf)
Expand Down Expand Up @@ -107,7 +108,8 @@ func TestSpyWithProcesses(t *testing.T) {
nodeName = "fishermans-friend" // TODO rename to hostNmae
)

reporter := endpoint.NewReporter(nodeID, nodeName, true, false)
revResolver := endpoint.NewReverseResolver(endpoint.RAddrCacheLen)
reporter := endpoint.NewReporter(nodeID, nodeName, true, false, revResolver)
r, _ := reporter.Report()
// buf, _ := json.MarshalIndent(r, "", " ") ; t.Logf("\n%s\n", buf)

Expand Down
70 changes: 70 additions & 0 deletions probe/endpoint/resolver.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
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
// 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)
}
3 changes: 2 additions & 1 deletion probe/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,8 @@ func main() {
}

var (
endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack)
revResolver = endpoint.NewReverseResolver(endpoint.RAddrCacheLen)
endpointReporter = endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack, revResolver)
processCache = process.NewCachingWalker(process.NewWalker(*procRoot))
reporters = []Reporter{
endpointReporter,
Expand Down

0 comments on commit 160e0e7

Please sign in to comment.