Adding the option to disable the DNS processor failure or success cache

libbeat/processors/dns/cache.go

@@ -18,6 +18,7 @@
 package dns
 
 import (
+	"math"
 	"sync"
 	"time"
 
@@ -34,13 +35,17 @@ func (r successRecord) IsExpired(now time.Time) bool {
 }
 
 type successCache struct {
+	disabled bool
 	sync.RWMutex
 	data          map[string]successRecord
 	maxSize       int
 	minSuccessTTL time.Duration
 }
 
 func (c *successCache) set(now time.Time, key string, result *result) {
+	if c.disabled {
+		return
+	}
 	c.Lock()
 	defer c.Unlock()
 
@@ -65,12 +70,15 @@ func (c *successCache) evict() {
 }
 
 func (c *successCache) get(now time.Time, key string) *result {
+	if c.disabled {
+		return nil
+	}
 	c.RLock()
 	defer c.RUnlock()
 
 	r, found := c.data[key]
 	if found && !r.IsExpired(now) {
-		return &result{r.data, uint32(r.expires.Sub(now) / time.Second)}
+		return &result{r.data, safeUint32(r.expires.Sub(now).Seconds())}
 	}
 	return nil
 }
@@ -85,13 +93,17 @@ func (r failureRecord) IsExpired(now time.Time) bool {
 }
 
 type failureCache struct {
+	disabled bool
 	sync.RWMutex
 	data       map[string]failureRecord
 	maxSize    int
 	failureTTL time.Duration
 }
 
 func (c *failureCache) set(now time.Time, key string, err error) {
+	if c.disabled {
+		return
+	}
 	c.Lock()
 	defer c.Unlock()
 	if len(c.data) >= c.maxSize {
@@ -115,6 +127,9 @@ func (c *failureCache) evict() {
 }
 
 func (c *failureCache) get(now time.Time, key string) error {
+	if c.disabled {
+		return nil
+	}
 	c.RLock()
 	defer c.RUnlock()
 
@@ -155,11 +170,13 @@ func newLookupCache(reg *monitoring.Registry, conf cacheConfig, resolver resolve
 
 	c := &lookupCache{
 		success: &successCache{
+			disabled:      conf.FailureCache.Disabled,
 			data:          make(map[string]successRecord, conf.SuccessCache.InitialCapacity),
 			maxSize:       conf.SuccessCache.MaxCapacity,
 			minSuccessTTL: conf.SuccessCache.MinTTL,
 		},
 		failure: &failureCache{
+			disabled:   conf.FailureCache.Disabled,
 			data:       make(map[string]failureRecord, conf.FailureCache.InitialCapacity),
 			maxSize:    conf.FailureCache.MaxCapacity,
 			failureTTL: conf.FailureCache.TTL,
@@ -200,7 +217,7 @@ func (c lookupCache) Lookup(q string, qt queryType) (*result, error) {
 	}
 
 	// We set the result TTL to the minimum TTL in case it is less than that.
-	r.TTL = max(r.TTL, uint32(c.success.minSuccessTTL/time.Second))
+	r.TTL = max(r.TTL, safeUint32(c.success.minSuccessTTL.Seconds()))
 
 	c.success.set(now, q, r)
 	return r, nil
@@ -212,3 +229,8 @@ func max(a, b uint32) uint32 {
 	}
 	return b
 }
+
+// bounds check / satisfy G115
+func safeUint32(float float64) uint32 {
+	return uint32(math.Min(math.Abs(float), float64(math.MaxUint32)))
+}

libbeat/processors/dns/cache_test.go

@@ -101,7 +101,7 @@ func TestCache(t *testing.T) {
 		assert.EqualValues(t, 3, c.stats.Miss.Get()) // Cache miss.
 	}
 
-	minTTL := defaultConfig().cacheConfig.SuccessCache.MinTTL
+	minTTL := defaultConfig().SuccessCache.MinTTL
 	// Initial success returned TTL=0 with MinTTL.
 	r, err = c.Lookup(gatewayIP+"2", typePTR)
 	if assert.NoError(t, err) {
@@ -126,3 +126,30 @@ func TestCache(t *testing.T) {
 		assert.EqualValues(t, 4, c.stats.Miss.Get())
 	}
 }
+
+func TestDisabledCache(t *testing.T) {
+	config := defaultConfig().cacheConfig
+	config.SuccessCache.Disabled = true
+	config.FailureCache.Disabled = true
+
+	c, err := newLookupCache(
+		monitoring.NewRegistry(),
+		config,
+		&stubResolver{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Initial success query.
+	var r *result
+	r, err = c.Lookup(gatewayIP, typePTR)
+	if assert.NoError(t, err) {
+		assert.EqualValues(t, []string{gatewayName}, r.Data)
+	}
+	r, err = c.Lookup(gatewayIP, typePTR)
+	if assert.NoError(t, err) {
+		assert.EqualValues(t, []string{gatewayName}, r.Data)
+		assert.EqualValues(t, 0, c.stats.Hit.Get())
+		assert.EqualValues(t, 2, c.stats.Miss.Get())
+	}
+}

libbeat/processors/dns/config.go

@@ -120,6 +120,9 @@ type cacheConfig struct {
 
 // cacheSettings define the caching behavior for an individual cache.
 type cacheSettings struct {
+	// Disable the use of the cache.
+	Disabled bool `config:"disabled" default:"false"`
+
 	// TTL value for items in cache. Not used for success because we use TTL
 	// from the DNS record.
 	TTL time.Duration `config:"ttl"`
@@ -189,11 +192,13 @@ func defaultConfig() config {
 	return config{
 		cacheConfig: cacheConfig{
 			SuccessCache: cacheSettings{
+				Disabled:        false,
 				MinTTL:          time.Minute,
 				InitialCapacity: 1000,
 				MaxCapacity:     10000,
 			},
 			FailureCache: cacheSettings{
+				Disabled:        false,
 				MinTTL:          time.Minute,
 				TTL:             time.Minute,
 				InitialCapacity: 1000,

libbeat/processors/dns/dns_test.go

@@ -39,7 +39,7 @@ func TestDNSProcessorRun(t *testing.T) {
 		resolver: &stubResolver{},
 		log:      logptest.NewTestingLogger(t, logName),
 	}
-	p.config.reverseFlat = map[string]string{
+	p.reverseFlat = map[string]string{
 		"source.ip": "source.domain",
 	}
 	t.Log(p.String())
@@ -60,7 +60,7 @@ func TestDNSProcessorRun(t *testing.T) {
 
 	const forwardDomain = "www." + gatewayName
 	t.Run("append", func(t *testing.T) {
-		p.config.Action = actionAppend
+		p.Action = actionAppend
 
 		event, err := p.Run(&beat.Event{
 			Fields: mapstr.M{
@@ -79,7 +79,7 @@ func TestDNSProcessorRun(t *testing.T) {
 	})
 
 	t.Run("replace", func(t *testing.T) {
-		p.config.Action = actionReplace
+		p.Action = actionReplace
 
 		event, err := p.Run(&beat.Event{
 			Fields: mapstr.M{
@@ -132,8 +132,8 @@ func TestDNSProcessorTagOnFailure(t *testing.T) {
 		resolver: &stubResolver{},
 		log:      logptest.NewTestingLogger(t, logName),
 	}
-	p.config.TagOnFailure = []string{"_lookup_failed"}
-	p.config.reverseFlat = map[string]string{
+	p.TagOnFailure = []string{"_lookup_failed"}
+	p.reverseFlat = map[string]string{
 		"source.ip":      "source.domain",
 		"destination.ip": "destination.domain",
 	}
@@ -151,10 +151,40 @@ func TestDNSProcessorTagOnFailure(t *testing.T) {
 
 	v, _ := event.GetValue("tags")
 	if assert.Len(t, v, 1) {
-		assert.ElementsMatch(t, v, p.config.TagOnFailure)
+		assert.ElementsMatch(t, v, p.TagOnFailure)
 	}
 }
 
+func TestDNSProcessorDisabledCache(t *testing.T) {
+	c := defaultConfig()
+	c.Type = typePTR
+	c.SuccessCache.Disabled = true
+	c.FailureCache.Disabled = true
+	p := &processor{
+		config:   c,
+		resolver: &stubResolver{},
+		log:      logptest.NewTestingLogger(t, logName),
+	}
+	p.reverseFlat = map[string]string{
+		"source.ip": "source.domain",
+	}
+	t.Log(p.String())
+
+	t.Run("default", func(t *testing.T) {
+		event, err := p.Run(&beat.Event{
+			Fields: mapstr.M{
+				"source.ip": gatewayIP,
+			},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		v, _ := event.GetValue("source.domain")
+		assert.Equal(t, gatewayName, v)
+	})
+}
+
 func TestDNSProcessorRunInParallel(t *testing.T) {
 	// This is a simple smoke test to make sure that there are no concurrency
 	// issues. It is most effective when run with the race detector.
@@ -166,7 +196,7 @@ func TestDNSProcessorRunInParallel(t *testing.T) {
 		t.Fatal(err)
 	}
 	p := &processor{config: conf, resolver: cache, log: logptest.NewTestingLogger(t, logName)}
-	p.config.reverseFlat = map[string]string{"source.ip": "source.domain"}
+	p.reverseFlat = map[string]string{"source.ip": "source.domain"}
 
 	const numGoroutines = 10
 	const numEvents = 500