latency as transform (dmachard#248)

flowintel · Mar 17, 2023 · 666721d · 666721d
1 parent 225010f
commit 666721d
Show file tree

Hide file tree

Showing 33 changed files with 420 additions and 210 deletions.
diff --git a/README.md b/README.md
@@ -51,6 +51,10 @@ Additionally, DNS-collector also support
 
 **Transformers**:
 
+- [`Latency Computing`](doc/transformers.md#dns-latency)
+    - Compute latency between replies and queries
+    - Detect evicted queries
+
 - [`Traffic filtering`](doc/transformers.md#dns-filtering)
     - Downsampling
     - Dropping per Qname, QueryIP or Rcode
@@ -109,6 +113,7 @@ You will find below some examples of configuration to manage your DNS logs.
     - [x] [Filtering incoming traffic with downsample and whitelist of domains](example-config/use-case-9.yml)
     - [x] [Transform all domains to lowercase](example-config/use-case-10.yml)
     - [x] [Add geographical metadata with GeoIP](example-config/use-case-11.yml)
+    - [x] [Count the number of no responses with outgoing queries](example-config/use-case-18.yml)
 
 - Capture DNS traffic from FRSTRM/dnstap files
     - [x] [Save incoming DNStap streams to file (frstrm)](example-config/use-case-13.yml)

diff --git a/collectors/dns_processor.go b/collectors/dns_processor.go
@@ -2,9 +2,6 @@ package collectors
 
 import (
 	"fmt"
-	"hash/fnv"
-	"strconv"
-	"strings"
 	"time"
 
 	"github.com/dmachard/go-dnscollector/dnsutils"
@@ -20,13 +17,11 @@ func GetFakeDns() ([]byte, error) {
 }
 
 type DnsProcessor struct {
-	done         chan bool
-	recvFrom     chan dnsutils.DnsMessage
-	logger       *logger.Logger
-	config       *dnsutils.Config
-	cacheSupport bool
-	queryTimeout int
-	name         string
+	done     chan bool
+	recvFrom chan dnsutils.DnsMessage
+	logger   *logger.Logger
+	config   *dnsutils.Config
+	name     string
 }
 
 func NewDnsProcessor(config *dnsutils.Config, logger *logger.Logger, name string) DnsProcessor {
@@ -44,10 +39,7 @@ func NewDnsProcessor(config *dnsutils.Config, logger *logger.Logger, name string
 	return d
 }
 
-func (d *DnsProcessor) ReadConfig() {
-	d.cacheSupport = false
-	d.queryTimeout = 5
-}
+func (d *DnsProcessor) ReadConfig() {}
 
 func (c *DnsProcessor) LogInfo(msg string, v ...interface{}) {
 	c.logger.Info("["+c.name+"] dns processor - "+msg, v...)
@@ -77,12 +69,8 @@ func (d *DnsProcessor) Stop() {
 
 func (d *DnsProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 
-	// dns cache to compute latency between response and query
-	cache_ttl := dnsutils.NewDnsCache(time.Duration(d.queryTimeout) * time.Second)
-	d.LogInfo("dns cached enabled: %t", d.cacheSupport)
-
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name)
+	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name, sendTo)
 
 	// read incoming dns message
 	d.LogInfo("running... waiting incoming dns message")
@@ -127,35 +115,14 @@ func (d *DnsProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 			}
 		}
 
-		// compute latency if possible
-		if d.cacheSupport {
-			queryport, _ := strconv.Atoi(dm.NetworkInfo.QueryPort)
-			if len(dm.NetworkInfo.QueryIp) > 0 && queryport > 0 && !dm.DNS.MalformedPacket {
-				// compute the hash of the query
-				hash_data := []string{dm.NetworkInfo.QueryIp, dm.NetworkInfo.QueryPort, strconv.Itoa(dm.DNS.Id)}
-
-				hashfnv := fnv.New64a()
-				hashfnv.Write([]byte(strings.Join(hash_data[:], "+")))
-
-				if dm.DNS.Type == dnsutils.DnsQuery {
-					cache_ttl.Set(hashfnv.Sum64(), dm.DnsTap.Timestamp)
-				} else {
-					value, ok := cache_ttl.Get(hashfnv.Sum64())
-					if ok {
-						dm.DnsTap.Latency = dm.DnsTap.Timestamp - value
-					}
-				}
-			}
-		}
-
-		// convert latency to human
-		dm.DnsTap.LatencySec = fmt.Sprintf("%.6f", dm.DnsTap.Latency)
-
 		// apply all enabled transformers
 		if subprocessors.ProcessMessage(&dm) == transformers.RETURN_DROP {
 			continue
 		}
 
+		// convert latency to human
+		dm.DnsTap.LatencySec = fmt.Sprintf("%.6f", dm.DnsTap.Latency)
+
 		// dispatch dns message to all generators
 		for i := range sendTo {
 			sendTo[i] <- dm

diff --git a/collectors/dnstap_processor.go b/collectors/dnstap_processor.go
@@ -2,10 +2,8 @@ package collectors
 
 import (
 	"fmt"
-	"hash/fnv"
 	"net"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/dmachard/go-dnscollector/dnsutils"
@@ -100,12 +98,8 @@ func (d *DnstapProcessor) Stop() {
 func (d *DnstapProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 	dt := &dnstap.Dnstap{}
 
-	// dns cache to compute latency between response and query
-	cache_ttl := dnsutils.NewDnsCache(time.Duration(d.config.Collectors.Dnstap.QueryTimeout) * time.Second)
-	d.LogInfo("dns cached enabled: %t", d.config.Collectors.Dnstap.CacheSupport)
-
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name)
+	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name, sendTo)
 
 	// read incoming dns message
 	d.LogInfo("running... waiting incoming dns message")
@@ -195,34 +189,14 @@ func (d *DnstapProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 			}
 		}
 
-		// compute latency if possible
-		if d.config.Collectors.Dnstap.CacheSupport {
-			if len(dm.NetworkInfo.QueryIp) > 0 && queryport > 0 && !dm.DNS.MalformedPacket {
-				// compute the hash of the query
-				hash_data := []string{dm.NetworkInfo.QueryIp, dm.NetworkInfo.QueryPort, strconv.Itoa(dm.DNS.Id)}
-
-				hashfnv := fnv.New64a()
-				hashfnv.Write([]byte(strings.Join(hash_data[:], "+")))
-
-				if dm.DNS.Type == dnsutils.DnsQuery {
-					cache_ttl.Set(hashfnv.Sum64(), dm.DnsTap.Timestamp)
-				} else {
-					value, ok := cache_ttl.Get(hashfnv.Sum64())
-					if ok {
-						dm.DnsTap.Latency = dm.DnsTap.Timestamp - value
-					}
-				}
-			}
-		}
-
-		// convert latency to human
-		dm.DnsTap.LatencySec = fmt.Sprintf("%.6f", dm.DnsTap.Latency)
-
 		// apply all enabled transformers
 		if subprocessors.ProcessMessage(&dm) == transformers.RETURN_DROP {
 			continue
 		}
 
+		// convert latency to human
+		dm.DnsTap.LatencySec = fmt.Sprintf("%.6f", dm.DnsTap.Latency)
+
 		// dispatch dns message to all generators
 		for i := range sendTo {
 			sendTo[i] <- dm

diff --git a/collectors/file_ingestor.go b/collectors/file_ingestor.go
@@ -290,7 +290,6 @@ func (c *FileIngestor) ProcessPcap(filePath string) {
 
 	// remove event timer for this file
 	c.RemoveEvent(filePath)
-
 }
 
 func (c *FileIngestor) ProcessDnstap(filePath string) error {

diff --git a/collectors/file_tail.go b/collectors/file_tail.go
@@ -98,7 +98,7 @@ func (c *Tail) Run() {
 	}
 
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(&c.config.IngoingTransformers, c.logger, c.name)
+	subprocessors := transformers.NewTransforms(&c.config.IngoingTransformers, c.logger, c.name, c.Loggers())
 
 	// init dns message
 	dm := dnsutils.DnsMessage{}

diff --git a/collectors/powerdns_processor.go b/collectors/powerdns_processor.go
@@ -76,7 +76,7 @@ func (d *PdnsProcessor) Run(sendTo []chan dnsutils.DnsMessage) {
 	pbdm := &powerdns_protobuf.PBDNSMessage{}
 
 	// prepare enabled transformers
-	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name)
+	subprocessors := transformers.NewTransforms(&d.config.IngoingTransformers, d.logger, d.name, sendTo)
 
 	// read incoming dns message
 	d.LogInfo("running... waiting incoming dns message")

diff --git a/collectors/sniffer_afpacket.go b/collectors/sniffer_afpacket.go
@@ -272,8 +272,6 @@ func (c *AfpacketSniffer) Run() {
 	}
 
 	dnsProcessor := NewDnsProcessor(c.config, c.logger, c.name)
-	dnsProcessor.cacheSupport = c.config.Collectors.AfpacketLiveCapture.CacheSupport
-	dnsProcessor.queryTimeout = c.config.Collectors.AfpacketLiveCapture.QueryTimeout
 	go dnsProcessor.Run(c.Loggers())
 
 	dnsChan := make(chan netlib.DnsPacket)

diff --git a/collectors/sniffer_xdp.go b/collectors/sniffer_xdp.go
@@ -107,7 +107,6 @@ func (c *XdpSniffer) Run() {
 	c.LogInfo("starting collector...")
 
 	dnsProcessor := NewDnsProcessor(c.config, c.logger, c.name)
-	dnsProcessor.cacheSupport = false
 	go dnsProcessor.Run(c.Loggers())
 
 	iface, err := net.InterfaceByName("wlp2s0")

diff --git a/collectors/tzsp.go b/collectors/tzsp.go
@@ -126,8 +126,6 @@ func (c *TzspSniffer) Run() {
 	}
 
 	dnsProcessor := NewDnsProcessor(c.config, c.logger, c.name)
-	dnsProcessor.cacheSupport = c.config.Collectors.Tzsp.CacheSupport
-	dnsProcessor.queryTimeout = c.config.Collectors.Tzsp.QueryTimeout
 
 	go dnsProcessor.Run(c.Loggers())
 

diff --git a/config.yml b/config.yml
@@ -52,8 +52,8 @@ global:
   # - ra: recursion available
   # - ad: authenticated data
   # - edns-csubnet: client subnet
-  # - df: flag when ip defragmentation occured
-  # - tr: flag when tcp reassembled occured
+  # - df: ip defragmentation flag
+  # - tr: tcp reassembled flag
   text-format: "timestamp-rfc3339ns identity operation rcode queryip queryport family protocol length qname qtype latency"
 
 # create your dns collector, please refer bellow to see the list 
@@ -97,11 +97,6 @@ multiplexer:
 #   cert-file: ""
 #   # private key server file
 #   key-file: ""
-#   # The cache is used to compute latency between replies and queries 
-#   # This cache can be enabled if your dns server doesn't add the latency information
-#   cache-support: false
-#   # Ttl in second, max time to keep the query record in memory cache
-#   query-timeout: 5
 #   # Sets the socket receive buffer in bytes SO_RCVBUF, set to zero to use the default system value
 #   sock-rcvbuf: 0
 
@@ -128,10 +123,6 @@ multiplexer:
 #   port: 53
 #   # if "" bind on all interfaces
 #   device: wlp2s0
-#   # The cache is used to compute latency between replies and queries
-#   cache-support: true
-#   # Ttl in second, max time to keep the query record in memory cache
-#   query-timeout: 5
 
 # # live capture with XDP
 # xdp-sniffer:
@@ -193,11 +184,6 @@ multiplexer:
 #  listen-ip: 0.0.0.0
 #  # listen on port
 #  listen-port: 10000
-#   # The cache is used to compute latency between replies and queries 
-#   # This cache can be enabled if your dns server doesn't add the latency information
-#   cache-support: false
-#   # Ttl in second, max time to keep the query record in memory cache
-#   query-timeout: 5
 
 ################################################
 # list of supported loggers
@@ -430,6 +416,15 @@ multiplexer:
 # list of transforms to apply on collectors or loggers
 ################################################
 
+# # Use this transformer to compute latency and detect timeout on queries
+# latency:
+#   # Measure latency between replies and queries
+#   measure-latency: false
+#   # Detect queries timeout
+#   detect-evicted-queries: false
+#   # timeout in second for queries
+#   queries-timeout: 2
+
 # # Use this option to protect user privacy
 # user-privacy:
 #   # IP-Addresses are anonymities by zeroing the host-part of an address.

diff --git a/dnsutils/cache.go b/dnsutils/cache.go
-Original file line number
+Diff line change
@@ Expand Up / @@ -290,7 +290,6 @@ func (c *FileIngestor) ProcessPcap(filePath string) { @@
     	// remove event timer for this file
     	c.RemoveEvent(filePath)
     }
     func (c *FileIngestor) ProcessDnstap(filePath string) error {
@@ Expand Down @@