sample events from banned sources by antispam

fd/util.go

@@ -34,6 +34,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 	antispamMaintenanceInterval := pipeline.DefaultMaintenanceInterval
 	var antispamExceptions antispam.Exceptions
 	var antispamRules antispam.Rules
+	var bannedSourcesSample *antispam.BannedSourcesSampleOptions
 
 	metricHoldDuration := pipeline.DefaultMetricHoldDuration
 	metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength
@@ -126,6 +127,11 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 			logger.Fatalf("extract antispam rules: %s", err)
 		}
 
+		bannedSourcesSample, err = extractBannedSourcesSample(antispamSettings)
+		if err != nil {
+			logger.Fatalf("extract antispam banned sources sample: %s", err)
+		}
+
 		sourceNameMetaField = settings.Get("source_name_meta_field").MustString()
 		isStrict = settings.Get("is_strict").MustBool()
 
@@ -167,6 +173,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 			Rules:               antispamRules,
 			Exceptions:          antispamExceptions,
 			MaintenanceInterval: antispamMaintenanceInterval,
+			BannedSourcesSample: bannedSourcesSample,
 		},
 		SourceNameMetaField: sourceNameMetaField,
 		MaintenanceInterval: maintenanceInterval,
@@ -242,6 +249,53 @@ func extractAntispamRules(settings *simplejson.Json, antispamMaintenanceInterval
 	return rules, nil
 }
 
+func extractBannedSourcesSample(settings *simplejson.Json) (*antispam.BannedSourcesSampleOptions, error) {
+	sampleJSON, ok := settings.CheckGet("banned_sources_sample")
+	if !ok {
+		return nil, nil
+	}
+
+	options := &antispam.BannedSourcesSampleOptions{}
+	if str := sampleJSON.Get("interval").MustString(); str != "" {
+		interval, err := time.ParseDuration(str)
+		if err != nil {
+			return nil, fmt.Errorf("parse interval: %w", err)
+		}
+		if interval < pipeline.DefaultBannedSourcesSampleInterval {
+			logger.Warnf("interval must be >= default value, using default %s", pipeline.DefaultBannedSourcesSampleInterval)
+			interval = pipeline.DefaultBannedSourcesSampleInterval
+		}
+		options.Interval = interval
+	} else {
+		options.Interval = pipeline.DefaultBannedSourcesSampleInterval
+	}
+
+	first := sampleJSON.Get("first").MustInt()
+	if first < pipeline.DefaultBannedSourcesSampleFirst {
+		logger.Warnf("first must be >= default value, using default %d", pipeline.DefaultBannedSourcesSampleFirst)
+		first = pipeline.DefaultBannedSourcesSampleFirst
+	}
+	options.First = first
+
+	thereafter := sampleJSON.Get("thereafter").MustInt()
+	if thereafter < pipeline.DefaultBannedSourcesSampleThereafter {
+		logger.Warnf("thereafter must be >= default value, using default %d", pipeline.DefaultBannedSourcesSampleThereafter)
+		thereafter = pipeline.DefaultBannedSourcesSampleThereafter
+	}
+	options.Thereafter = thereafter
+
+	if str := sampleJSON.Get("sampled_field").MustString(); str == "" {
+		options.SampledField = pipeline.DefaultBannedSourcesSampleField
+	} else {
+		options.SampledField = str
+	}
+
+	options.SampledMetricName = sampleJSON.Get("sampled_metric_name").MustString()
+	options.SampledMetricLabels = sampleJSON.Get("sampled_metric_labels").MustStringArray()
+
+	return options, nil
+}
+
 func extractMatchMode(actionJSON *simplejson.Json) pipeline.MatchMode {
 	mm := actionJSON.Get("match_mode").MustString()
 	return pipeline.MatchModeFromString(mm)

pipeline/antispam/antispammer.go

@@ -26,23 +26,29 @@ type Antispammer struct {
 	threshold           int
 	maintenanceInterval time.Duration
 	mu                  sync.RWMutex
-	sources             map[string]source
+	sources             map[string]*source
 	sourcesThresholds   map[string]int
 	exceptions          Exceptions
 	rules               Rules
+	bannedSourcesSample *BannedSourcesSampleOptions
 
 	logger *zap.Logger
 
 	// antispammer metrics
 	activeMetric    *metric.Gauge
 	banMetric       *metric.GaugeVec
 	exceptionMetric *metric.CounterVec
+	sampledMetric   *metric.CounterVec
 }
 
 type source struct {
 	counter   *atomic.Int32
 	timestamp *atomic.Int64
 	name      string
+
+	sampleMu      sync.Mutex
+	sampleUntil   time.Time
+	sampleCounter int
 }
 
 type Options struct {
@@ -51,11 +57,21 @@ type Options struct {
 	UnbanIterations     int
 	Exceptions          Exceptions
 	Rules               Rules
+	BannedSourcesSample *BannedSourcesSampleOptions
 
 	Logger            *zap.Logger
 	MetricsController *metric.Ctl
 }
 
+type BannedSourcesSampleOptions struct {
+	Interval            time.Duration
+	First               int
+	Thereafter          int
+	SampledField        string
+	SampledMetricName   string
+	SampledMetricLabels []string
+}
+
 func NewAntispammer(o *Options) *Antispammer {
 	if o.Threshold > 0 {
 		o.Logger.Info("antispam enabled",
@@ -67,7 +83,7 @@ func NewAntispammer(o *Options) *Antispammer {
 		unbanIterations:     o.UnbanIterations,
 		threshold:           o.Threshold,
 		maintenanceInterval: o.MaintenanceInterval,
-		sources:             make(map[string]source),
+		sources:             make(map[string]*source),
 		sourcesThresholds:   make(map[string]int),
 		exceptions:          o.Exceptions,
 		rules:               o.Rules,
@@ -85,15 +101,31 @@ func NewAntispammer(o *Options) *Antispammer {
 		),
 	}
 
+	if o.BannedSourcesSample != nil {
+		a.bannedSourcesSample = o.BannedSourcesSample
+		o.Logger.Info("antispam banned sources sample enabled",
+			zap.Duration("interval", o.BannedSourcesSample.Interval),
+			zap.Int("first", o.BannedSourcesSample.First),
+			zap.Int("thereafter", o.BannedSourcesSample.Thereafter),
+		)
+
+		if o.BannedSourcesSample.SampledMetricName != "" {
+			a.sampledMetric = o.MetricsController.RegisterCounterVec(
+				o.BannedSourcesSample.SampledMetricName,
+				"How many events from banned sources were let through by the sampler",
+				o.BannedSourcesSample.SampledMetricLabels...,
+			)
+		}
+	}
 	// not enabled by default
 	a.activeMetric.Set(0)
 
 	return a
 }
 
-func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []byte, timeEvent time.Time, meta map[string]string) bool {
+func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []byte, timeEvent time.Time, meta map[string]string) (spam, sampled bool) {
 	if a.rules == nil && a.threshold == -1 {
-		return false
+		return false, false
 	}
 
 	threshold := a.threshold
@@ -108,7 +140,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 				if e.Name != "" {
 					a.exceptionMetric.WithLabelValues(e.Name).Inc()
 				}
-				return false
+				return false, false
 			}
 		}
 	} else {
@@ -124,9 +156,9 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 			switch rule.Threshold {
 			case thresholdUnlimited:
 				a.exceptionMetric.WithLabelValues(rule.Name).Inc()
-				return false
+				return false, false
 			case thresholdBlocked:
-				return true
+				return true, false
 			}
 
 			threshold = rule.Threshold
@@ -136,9 +168,9 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 
 	switch threshold {
 	case thresholdUnlimited:
-		return false
+		return false, false
 	case thresholdBlocked:
-		return true
+		return true, false
 	}
 
 	a.mu.RLock()
@@ -152,7 +184,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 		if newSrc, has := a.sources[id]; has {
 			src = newSrc
 		} else {
-			src = source{
+			src = &source{
 				counter:   &atomic.Int32{},
 				name:      name,
 				timestamp: &atomic.Int64{},
@@ -166,7 +198,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 
 	if isNewSource {
 		src.counter.Swap(0)
-		return false
+		return false, false
 	}
 
 	x := src.counter.Load()
@@ -186,7 +218,13 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 		)
 	}
 
-	return x >= int32(threshold)
+	if x >= int32(threshold) {
+		if a.samplerAllow(src) {
+			return true, true
+		}
+		return true, false
+	}
+	return false, false
 }
 
 func (a *Antispammer) Maintenance() {
@@ -213,6 +251,10 @@ func (a *Antispammer) Maintenance() {
 		if isMore && x < threshold {
 			a.banMetric.WithLabelValues(source.name).Dec()
 			a.logger.Info("source has been unbanned", zap.Any("id", sourceID))
+			source.sampleMu.Lock()
+			source.sampleUntil = time.Time{}
+			source.sampleCounter = 0
+			source.sampleMu.Unlock()
 		}
 
 		if x >= threshold {
@@ -252,6 +294,36 @@ func (a *Antispammer) Dump() string {
 	return out
 }
 
+func (a *Antispammer) samplerAllow(src *source) bool {
+	cfg := a.bannedSourcesSample
+	if cfg == nil {
+		return false
+	}
+
+	src.sampleMu.Lock()
+	defer src.sampleMu.Unlock()
+
+	now := time.Now()
+	if now.After(src.sampleUntil) {
+		src.sampleUntil = now.Add(cfg.Interval)
+		src.sampleCounter = 0
+	}
+
+	src.sampleCounter++
+	if src.sampleCounter <= cfg.First {
+		return true
+	}
+	if cfg.Thereafter == 0 {
+		return false
+	}
+
+	return (src.sampleCounter-cfg.First)%cfg.Thereafter == 0
+}
+
+func (a *Antispammer) SampledMetric() *metric.CounterVec {
+	return a.sampledMetric
+}
+
 type Exception struct {
 	matchrule.RuleSet
 	CheckSourceName bool `json:"check_source_name"`

pipeline/antispam/antispammer_test.go

@@ -34,7 +34,8 @@ func TestAntispam(t *testing.T) {
 	startTime := time.Now()
 	checkSpam := func(i int) bool {
 		eventTime := startTime.Add(time.Duration(i) * maintenanceInterval / 2)
-		return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil)
+		spam, _ := antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil)
+		return spam
 	}
 
 	for i := 1; i < threshold; i++ {
@@ -64,7 +65,8 @@ func TestAntispamAfterRestart(t *testing.T) {
 	startTime := time.Now()
 	checkSpam := func(i int) bool {
 		eventTime := startTime.Add(time.Duration(i) * maintenanceInterval)
-		return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil)
+		spam, _ := antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil)
+		return spam
 	}
 
 	for i := 1; i < threshold; i++ {
@@ -212,7 +214,8 @@ func TestAntispamRules(t *testing.T) {
 	}
 
 	checkSpam := func(expected bool, source, event string, meta map[string]string) {
-		r.Equal(expected, antispamer.IsSpam(source, source, false, []byte(event), now, meta))
+		spam, _ := antispamer.IsSpam(source, source, false, []byte(event), now, meta)
+		r.Equal(expected, spam)
 	}
 
 	checkSpam(true, "test_source_name", `{"level":"info","message":test"}`, nil)