[NDMII-3058] Add synchronous rDNS lookup to rdnsquerier component

comp/netflow/flowaggregator/flowaccumulator.go

@@ -194,7 +194,7 @@ func (f *flowAccumulator) setDstReverseDNSHostname(aggHash uint64, hostname stri
 }
 
 func (f *flowAccumulator) addRDNSEnrichment(aggHash uint64, srcAddr []byte, dstAddr []byte) {
-	err := f.rdnsQuerier.GetHostname(
+	err := f.rdnsQuerier.GetHostnameAsync(
 		srcAddr,
 		// Sync callback, lock is already held
 		func(hostname string) {
@@ -213,7 +213,7 @@ func (f *flowAccumulator) addRDNSEnrichment(aggHash uint64, srcAddr []byte, dstA
 		f.logger.Debugf("Error requesting reverse DNS enrichment for source IP address: %v error: %v", srcAddr, err)
 	}
 
-	err = f.rdnsQuerier.GetHostname(
+	err = f.rdnsQuerier.GetHostnameAsync(
 		dstAddr,
 		// Sync callback, lock is held
 		func(hostname string) {

comp/rdnsquerier/def/component.go

@@ -6,9 +6,22 @@
 // Package rdnsquerier provides the reverse DNS querier component.
 package rdnsquerier
 
+import (
+	"context"
+)
+
 // team: network-device-monitoring
 
+// ReverseDNSResult is the result of a reverse DNS lookup
+type ReverseDNSResult struct {
+	IP       string
+	Hostname string
+	Err      error
+}
+
 // Component is the component type.
 type Component interface {
-	GetHostname([]byte, func(string), func(string, error)) error
+	GetHostnameAsync([]byte, func(string), func(string, error)) error
+	GetHostnameSync(context.Context, string) (string, error)
+	GetHostnames(context.Context, []string) map[string]ReverseDNSResult
 }

comp/rdnsquerier/impl-none/none.go

@@ -7,6 +7,8 @@
 package rdnsquerierimpl
 
 import (
+	"context"
+
 	rdnsquerier "github.com/DataDog/datadog-agent/comp/rdnsquerier/def"
 )
 
@@ -25,7 +27,17 @@ func NewNone() Provides {
 }
 
 // GetHostnameAsync does nothing for the noop rdnsquerier implementation
-func (q *rdnsQuerierImplNone) GetHostname(_ []byte, _ func(string), _ func(string, error)) error {
+func (q *rdnsQuerierImplNone) GetHostnameAsync(_ []byte, _ func(string), _ func(string, error)) error {
+	// noop
+	return nil
+}
+
+func (q *rdnsQuerierImplNone) GetHostnameSync(_ context.Context, _ string) (string, error) {
+	// noop
+	return "", nil
+}
+
+func (q *rdnsQuerierImplNone) GetHostnames(_ context.Context, _ []string) map[string]rdnsquerier.ReverseDNSResult {
 	// noop
 	return nil
 }

comp/rdnsquerier/impl/rdnsquerier.go

@@ -10,13 +10,15 @@ import (
 	"context"
 	"fmt"
 	"net/netip"
+	"sync"
 
 	"github.com/DataDog/datadog-agent/comp/core/config"
 	log "github.com/DataDog/datadog-agent/comp/core/log/def"
 	"github.com/DataDog/datadog-agent/comp/core/telemetry"
 	compdef "github.com/DataDog/datadog-agent/comp/def"
 	rdnsquerier "github.com/DataDog/datadog-agent/comp/rdnsquerier/def"
 	rdnsquerierimplnone "github.com/DataDog/datadog-agent/comp/rdnsquerier/impl-none"
+	"go.uber.org/multierr"
 )
 
 // Requires defines the dependencies for the rdnsquerier component
@@ -157,7 +159,7 @@ func NewComponent(reqs Requires) (Provides, error) {
 // If the hostname for the IP address is immediately available (i.e. cache is enabled and entry is cached) then the updateHostnameSync callback
 // will be invoked synchronously, otherwise a query is sent to a channel to be processed asynchronously.  If the channel is full then an error
 // is returned.  When the request completes the updateHostnameAsync callback will be invoked asynchronously.
-func (q *rdnsQuerierImpl) GetHostname(ipAddr []byte, updateHostnameSync func(string), updateHostnameAsync func(string, error)) error {
+func (q *rdnsQuerierImpl) GetHostnameAsync(ipAddr []byte, updateHostnameSync func(string), updateHostnameAsync func(string, error)) error {
 	q.internalTelemetry.total.Inc()
 
 	netipAddr, ok := netip.AddrFromSlice(ipAddr)
@@ -167,18 +169,107 @@ func (q *rdnsQuerierImpl) GetHostname(ipAddr []byte, updateHostnameSync func(str
 	}
 
 	if !netipAddr.IsPrivate() {
-		q.logger.Tracef("Reverse DNS Enrichment IP address %s is not in the private address space", netipAddr)
+		q.logger.Tracef("Reverse DNS Enrichment IP address %s is not in the private address space", ipAddr)
 		return nil
 	}
 	q.internalTelemetry.private.Inc()
 
 	err := q.cache.getHostname(netipAddr.String(), updateHostnameSync, updateHostnameAsync)
 	if err != nil {
 		q.logger.Debugf("Reverse DNS Enrichment cache.getHostname() for addr %s returned error: %v", netipAddr.String(), err)
-		return err
 	}
 
-	return nil
+	return err
+}
+
+// GetHostnameSync attempts to resolve the hostname for the given IP address synchronously.
+// If the IP address is invalid then an error is returned.
+// If the IP address is not in the private address space then it is ignored - no lookup is performed and nil error is returned.
+// If the IP address is in the private address space then the IP address will be resolved to a hostname.
+// The function accepts a timeout via context and will return an error if the timeout is reached.
+func (q *rdnsQuerierImpl) GetHostnameSync(ctx context.Context, ipAddr string) (string, error) {
+	results := q.GetHostnames(ctx, []string{ipAddr})
+	result, ok := results[ipAddr]
+	if !ok {
+		return "", fmt.Errorf("no result for IP address %s", ipAddr)
+	}
+
+	return result.Hostname, result.Err
+}
+
+// GetHostnames attempts to resolve the hostname for the given IP addresses.
+// If the IP address is invalid then an error is returned.
+// If the IP address is not in the private address space then it is ignored - no lookup is performed and nil error is returned.
+// If the IP address is in the private address space then the IP address will be resolved to a hostname.
+// The function accepts a timeout via context and will return an error if the timeout is reached.
+func (q *rdnsQuerierImpl) GetHostnames(ctx context.Context, ipAddrs []string) map[string]rdnsquerier.ReverseDNSResult {
+	q.internalTelemetry.total.Add(float64(len(ipAddrs)))
+
+	var wg sync.WaitGroup
+	resultsChan := make(chan rdnsquerier.ReverseDNSResult, len(ipAddrs))
+
+	for _, ipAddr := range ipAddrs {
+		wg.Add(1)
+		go func(ipAddr string) {
+			defer wg.Done()
+			netipAddr, err := netip.ParseAddr(ipAddr)
+			if err != nil {
+				q.internalTelemetry.invalidIPAddress.Inc()
+				resultsChan <- rdnsquerier.ReverseDNSResult{IP: ipAddr, Err: fmt.Errorf("invalid IP address %s: %v", ipAddr, err)}
+				return
+			}
+
+			if !netipAddr.IsPrivate() {
+				q.logger.Tracef("Reverse DNS Enrichment IP address %s is not in the private address space", ipAddr)
+				resultsChan <- rdnsquerier.ReverseDNSResult{IP: ipAddr}
+				return
+			}
+			q.internalTelemetry.private.Inc()
+
+			hostnameChan := make(chan string, 1)
+			asyncErrChan := make(chan error, 1)
+
+			err = q.cache.getHostname(
+				netipAddr.String(),
+				func(h string) {
+					hostnameChan <- h
+					asyncErrChan <- nil
+				},
+				func(h string, e error) {
+					hostnameChan <- h
+					asyncErrChan <- e
+				},
+			)
+			if err != nil {
+				q.logger.Debugf("Reverse DNS Enrichment cache.getHostname() for addr %s returned error: %v", netipAddr.String(), err)
+			}
+
+			select {
+			case hostname := <-hostnameChan:
+				asyncErr := <-asyncErrChan // this is okay because we know that as soon as we send hostname, we send asyncErr
+				err = multierr.Append(err, asyncErr)
+				resultsChan <- rdnsquerier.ReverseDNSResult{
+					IP:       ipAddr,
+					Hostname: hostname,
+					Err:      err,
+				}
+			case <-ctx.Done():
+				resultsChan <- rdnsquerier.ReverseDNSResult{IP: ipAddr, Err: fmt.Errorf("timeout reached while resolving hostname for IP address %v", ipAddr)}
+			}
+		}(ipAddr)
+	}
+
+	go func() {
+		wg.Wait()
+		close(resultsChan)
+	}()
+
+	results := make(map[string]rdnsquerier.ReverseDNSResult, len(ipAddrs))
+	for result := range resultsChan {
+		results[result.IP] = result
+	}
+
+	return results
 }
 
 func (q *rdnsQuerierImpl) start(_ context.Context) error {