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 antispamSampler *antispam.Sampler
 
 	metricHoldDuration := pipeline.DefaultMetricHoldDuration
 	metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength
@@ -126,6 +127,11 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 			logger.Fatalf("extract antispam rules: %s", err)
 		}
 
+		antispamSampler, err = extractAntispamSampler(antispamSettings)
+		if err != nil {
+			logger.Fatalf("extract antispam sampler: %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,
+			Sampler:             antispamSampler,
 		},
 		SourceNameMetaField: sourceNameMetaField,
 		MaintenanceInterval: maintenanceInterval,
@@ -242,6 +249,47 @@ func extractAntispamRules(settings *simplejson.Json, antispamMaintenanceInterval
 	return rules, nil
 }
 
+func extractAntispamSampler(settings *simplejson.Json) (*antispam.Sampler, error) {
+	sampleJSON, ok := settings.CheckGet("sampler")
+	if !ok {
+		return nil, nil
+	}
+
+	intervalStr := sampleJSON.Get("interval").MustString()
+	if intervalStr == "" {
+		return nil, fmt.Errorf("interval is required")
+	}
+	interval, err := time.ParseDuration(intervalStr)
+	if err != nil {
+		return nil, fmt.Errorf("parse interval: %w", err)
+	}
+	if interval <= 0 {
+		return nil, fmt.Errorf("interval must be > 0, got %s", interval)
+	}
+
+	first := sampleJSON.Get("first").MustInt64()
+	if first < 0 {
+		return nil, fmt.Errorf("first must be >= 0, got %d", first)
+	}
+
+	thereafter := sampleJSON.Get("thereafter").MustInt64()
+	if thereafter < 0 {
+		return nil, fmt.Errorf("thereafter must be >= 0, got %d", thereafter)
+	}
+	if first == 0 && thereafter == 0 {
+		return nil, fmt.Errorf("at least one of first/thereafter must be > 0")
+	}
+
+	return &antispam.Sampler{
+		Interval:     interval,
+		First:        first,
+		Thereafter:   thereafter,
+		MarkerField:  sampleJSON.Get("marker_field").MustString(),
+		MetricName:   sampleJSON.Get("metric_name").MustString(),
+		MetricLabels: sampleJSON.Get("metric_labels").MustStringArray(),
+	}, nil
+}
+
 func extractMatchMode(actionJSON *simplejson.Json) pipeline.MatchMode {
 	mm := actionJSON.Get("match_mode").MustString()
 	return pipeline.MatchModeFromString(mm)

pipeline/antispam/README.md

@@ -78,3 +78,46 @@ List of rules to check the log against.
 Flag indicating whether to check source name. If set to `true` source name will be checked against all rules. If set to `false` log as raw bytes content will be checked against all rules.
 
 <br>
+
+## Sampler
+
+By default when a source hits the antispam threshold it is banned and all its subsequent logs are silently dropped until unban. That makes debugging a constantly blocked service impossible. The optional block `sampler` enables a per-source sampler that lets a small share of banned logs through. Each banned source has its own sampler state: in every `interval` window the first `first` logs pass, then every `thereafter`-th log passes, the rest are dropped. Each passed event can be marked with a boolean field so you can tell sampled events apart from default ones.
+
+### Sampler parameters
+
+**`interval`** *`duration`* *`required`*
+
+Sampler window per source.
+
+<br>
+
+**`first`** *`int`*
+
+Number of events from a banned source that always pass at the start of each `interval` window.
+
+<br>
+
+**`thereafter`** *`int`*
+
+After `first` events in the window have passed, every `thereafter`-th event is let through. Set to zero to drop everything past `first`.
+
+<br>
+
+**`marker_field`** *`string`*
+
+Field to add to log if it was let through the banned sampler. E.g. with `marker_field: _antispam_sampled`, if the log was sampled, the output
+event will have field `"_antispam_sampled":true`. Only works if the `sampler` block is set. Useful for marking sampled logs.
+
+<br>
+
+**`metric_name`** *`string`*
+
+Name of the metric registered for events that passed through the banned sampler.
+
+<br>
+
+**`metric_labels`** *`[]string`*
+
+Lists the log fields to add to the metric. Blank list means no labels.
+
+<br>

pipeline/antispam/antispammer.go

@@ -17,6 +17,17 @@ const (
 	thresholdBlocked   = 0
 )
 
+type SpamResult int
+
+const (
+	// event is not spam.
+	Pass SpamResult = iota
+	//event is spam.
+	Drop
+	// event is spam, but passed by sampler
+	Sampled
+)
+
 // Antispammer makes a decision on the need to parse the input log.
 // It can be useful when any application writes logs at speed faster than File.d can read it.
 //
@@ -30,19 +41,23 @@ type Antispammer struct {
 	sourcesThresholds   map[string]int
 	exceptions          Exceptions
 	rules               Rules
+	sampler             *Sampler
 
 	logger *zap.Logger
 
 	// antispammer metrics
 	activeMetric    *metric.Gauge
 	banMetric       *metric.GaugeVec
 	exceptionMetric *metric.CounterVec
+	samplerMetric   *metric.CounterVec
 }
 
 type source struct {
-	counter   *atomic.Int32
-	timestamp *atomic.Int64
-	name      string
+	counter       *atomic.Int32
+	timestamp     *atomic.Int64
+	sampleUntil   *atomic.Int64
+	sampleCounter *atomic.Int64
+	name          string
 }
 
 type Options struct {
@@ -51,11 +66,21 @@ type Options struct {
 	UnbanIterations     int
 	Exceptions          Exceptions
 	Rules               Rules
+	Sampler             *Sampler
 
 	Logger            *zap.Logger
 	MetricsController *metric.Ctl
 }
 
+type Sampler struct {
+	Interval     time.Duration
+	First        int64
+	Thereafter   int64
+	MarkerField  string
+	MetricName   string
+	MetricLabels []string
+}
+
 func NewAntispammer(o *Options) *Antispammer {
 	if o.Threshold > 0 {
 		o.Logger.Info("antispam enabled",
@@ -85,15 +110,31 @@ func NewAntispammer(o *Options) *Antispammer {
 		),
 	}
 
+	if o.Sampler != nil {
+		a.sampler = o.Sampler
+		o.Logger.Info("sampling of banned sources is enabled",
+			zap.Duration("interval", o.Sampler.Interval),
+			zap.Int64("first", o.Sampler.First),
+			zap.Int64("thereafter", o.Sampler.Thereafter),
+		)
+
+		if o.Sampler.MetricName != "" {
+			a.samplerMetric = o.MetricsController.RegisterCounterVec(
+				o.Sampler.MetricName,
+				"How many events from banned sources were let through by the sampler",
+				o.Sampler.MetricLabels...,
+			)
+		}
+	}
 	// 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) SpamResult {
 	if a.rules == nil && a.threshold == -1 {
-		return false
+		return Pass
 	}
 
 	threshold := a.threshold
@@ -108,7 +149,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 Pass
 			}
 		}
 	} else {
@@ -124,9 +165,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 Pass
 			case thresholdBlocked:
-				return true
+				return Drop
 			}
 
 			threshold = rule.Threshold
@@ -136,9 +177,9 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 
 	switch threshold {
 	case thresholdUnlimited:
-		return false
+		return Pass
 	case thresholdBlocked:
-		return true
+		return Drop
 	}
 
 	a.mu.RLock()
@@ -153,9 +194,11 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 			src = newSrc
 		} else {
 			src = source{
-				counter:   &atomic.Int32{},
-				name:      name,
-				timestamp: &atomic.Int64{},
+				counter:       &atomic.Int32{},
+				timestamp:     &atomic.Int64{},
+				sampleUntil:   &atomic.Int64{},
+				sampleCounter: &atomic.Int64{},
+				name:          name,
 			}
 			src.timestamp.Add(timeEventSeconds)
 			a.sources[id] = src
@@ -166,7 +209,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 
 	if isNewSource {
 		src.counter.Swap(0)
-		return false
+		return Pass
 	}
 
 	x := src.counter.Load()
@@ -186,7 +229,13 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 		)
 	}
 
-	return x >= int32(threshold)
+	if x >= int32(threshold) {
+		if a.isSampled(src) {
+			return Sampled
+		}
+		return Drop
+	}
+	return Pass
 }
 
 func (a *Antispammer) Maintenance() {
@@ -213,6 +262,8 @@ 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.sampleCounter.Store(0)
+			source.sampleUntil.Store(0)
 		}
 
 		if x >= threshold {
@@ -252,6 +303,34 @@ func (a *Antispammer) Dump() string {
 	return out
 }
 
+func (a *Antispammer) SamplerMetric() *metric.CounterVec {
+	return a.samplerMetric
+}
+
+func (a *Antispammer) isSampled(src source) bool {
+	cfg := a.sampler
+	if cfg == nil {
+		return false
+	}
+
+	now := time.Now().UnixNano()
+	if end := src.sampleUntil.Load(); now >= end {
+		if src.sampleUntil.CompareAndSwap(end, now+cfg.Interval.Nanoseconds()) {
+			src.sampleCounter.Store(0)
+		}
+	}
+
+	val := src.sampleCounter.Inc()
+	if val <= cfg.First {
+		return true
+	}
+	if cfg.Thereafter == 0 {
+		return false
+	}
+
+	return (val-cfg.First)%cfg.Thereafter == 0
+}
+
 type Exception struct {
 	matchrule.RuleSet
 	CheckSourceName bool `json:"check_source_name"`

pipeline/antispam/antispammer_test.go

@@ -34,7 +34,7 @@ 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)
+		return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) == Drop
 	}
 
 	for i := 1; i < threshold; i++ {
@@ -64,7 +64,7 @@ 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)
+		return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) == Drop
 	}
 
 	for i := 1; i < threshold; i++ {
@@ -212,7 +212,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))
+		got := antispamer.IsSpam(source, source, false, []byte(event), now, meta) == Drop
+		r.Equal(expected, got)
 	}
 
 	checkSpam(true, "test_source_name", `{"level":"info","message":test"}`, nil)