Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

reverse resolutions for remote side of connections. #404

Merged
merged 3 commits into from
Sep 7, 2015
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions probe/endpoint/reporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ type Reporter struct {
includeNAT bool
conntracker *Conntracker
natmapper *natmapper
revResolver *ReverseResolver
}

// SpyDuration is an exported prometheus metric
Expand Down Expand Up @@ -70,6 +71,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
includeProcesses: includeProcesses,
conntracker: conntracker,
natmapper: natmapper,
revResolver: NewReverseResolver(),
}
}

Expand All @@ -81,6 +83,7 @@ func (r *Reporter) Stop() {
if r.natmapper != nil {
r.natmapper.Stop()
}
r.revResolver.Stop()
}

// Report implements Reporter.
Expand Down Expand Up @@ -145,6 +148,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 {
// New nodes are merged into the report so we don't need to do any counting here; the merge does it for us.
localNode = localNode.WithEdge(remoteAddressNodeID, report.EdgeMetadata{
Expand Down Expand Up @@ -177,6 +187,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 {
// New nodes are merged into the report so we don't need to do any counting here; the merge does it for us.
localNode = localNode.WithEdge(remoteEndpointNodeID, report.EdgeMetadata{
Expand Down
2 changes: 1 addition & 1 deletion 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 Down
76 changes: 76 additions & 0 deletions probe/endpoint/resolver.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
package endpoint

import (
"net"
"strings"
"time"

"github.com/bluele/gcache"
)

const (
rAddrCacheLen = 500 // Default cache length
rAddrBacklog = 1000
rAddrCacheExpiration = 30 * time.Minute
)

type revResFunc func(addr string) (names []string, err error)

// ReverseResolver is a caching, reverse resolver
type ReverseResolver struct {

This comment was marked as abuse.

addresses chan string
cache gcache.Cache
Throttle <-chan time.Time // Made public for mocking
Resolver revResFunc
}

// NewReverseResolver starts a new reverse resolver that
// performs reverse resolutions and caches the result.
func NewReverseResolver() *ReverseResolver {
r := ReverseResolver{
addresses: make(chan string, rAddrBacklog),
cache: gcache.New(rAddrCacheLen).LRU().Expiration(rAddrCacheExpiration).Build(),
Throttle: time.Tick(time.Second / 10),
Resolver: net.LookupAddr,
}
go r.loop()
return &r
}

// Get the reverse resolution for an IP address if already in the cache,
// a gcache.NotFoundKeyError error 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) loop() {
for request := range r.addresses {
<-r.Throttle // rate limit our DNS resolutions
// and check if the answer is already in the cache
if _, err := r.cache.Get(request); err == nil {
continue
}
names, err := r.Resolver(request)
if err == nil && len(names) > 0 {
name := strings.TrimRight(names[0], ".")
r.cache.Set(request, name)
}
}
}

// Stop the async reverse resolver
func (r *ReverseResolver) Stop() {
close(r.addresses)
}
38 changes: 38 additions & 0 deletions probe/endpoint/resolver_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
package endpoint_test

import (
"errors"
"testing"
"time"

This comment was marked as abuse.

This comment was marked as abuse.


. "github.com/weaveworks/scope/probe/endpoint"
"github.com/weaveworks/scope/test"
)

func TestReverseResolver(t *testing.T) {
tests := map[string]string{
"1.2.3.4": "test.domain.name",
"4.3.2.1": "im.a.little.tea.pot",

This comment was marked as abuse.

}

revRes := NewReverseResolver()
defer revRes.Stop()

// use a mocked resolver function
revRes.Resolver = func(addr string) (names []string, err error) {
if name, ok := tests[addr]; ok {
return []string{name}, nil
}
return []string{}, errors.New("invalid IP")
}

// Up the rate limit so the test runs faster
revRes.Throttle = time.Tick(time.Millisecond)

for ip, hostname := range tests {
test.Poll(t, 100*time.Millisecond, hostname, func() interface{} {
result, _ := revRes.Get(ip)
return result
})
}
}
11 changes: 7 additions & 4 deletions render/detailed_node.go
Original file line number Diff line number Diff line change
Expand Up @@ -205,22 +205,25 @@ func OriginTable(r report.Report, originID string, addHostTags bool, addContaine

func connectionDetailsRows(topology report.Topology, originID string) []Row {
rows := []Row{}
labeler := func(nodeID string) (string, bool) {
labeler := func(nodeID string, meta map[string]string) (string, bool) {
if _, addr, port, ok := report.ParseEndpointNodeID(nodeID); ok {
if name, ok := meta["name"]; ok {
return fmt.Sprintf("%s:%s", name, port), true
}
return fmt.Sprintf("%s:%s", addr, port), true
}
if _, addr, ok := report.ParseAddressNodeID(nodeID); ok {
return addr, true
}
return "", false
}
local, ok := labeler(originID)
local, ok := labeler(originID, topology.Nodes[originID].Metadata)
if !ok {
return rows
}
// Firstly, collection outgoing connections from this node.
for _, serverNodeID := range topology.Nodes[originID].Adjacency {
remote, ok := labeler(serverNodeID)
remote, ok := labeler(serverNodeID, topology.Nodes[serverNodeID].Metadata)
if !ok {
continue
}
Expand All @@ -239,7 +242,7 @@ func connectionDetailsRows(topology report.Topology, originID string) []Row {
if !serverNodeIDs.Contains(originID) {
continue
}
remote, ok := labeler(clientNodeID)
remote, ok := labeler(clientNodeID, clientNode.Metadata)
if !ok {
continue
}
Expand Down
6 changes: 6 additions & 0 deletions report/topology.go
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,12 @@ func (n Node) WithMetadata(m map[string]string) Node {
return result
}

// AddMetadata returns a fresh copy of n, with Metadata set to the merge of n and the metadata provided
func (n Node) AddMetadata(m map[string]string) Node {
additional := MakeNodeWith(m)
return n.Merge(additional)
}

// WithCounters returns a fresh copy of n, with Counters set to c
func (n Node) WithCounters(c map[string]int) Node {
result := n.Copy()
Expand Down