From 18e68e7ab717e6b7e78d2c77edee3b07100bceca Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Thu, 13 Aug 2026 19:48:55 +0300 Subject: [PATCH 1/8] add logic antispam sampler --- fd/util.go | 52 +++++++++++++++ pipeline/antispam/antispammer.go | 92 +++++++++++++++++++++++---- pipeline/antispam/antispammer_test.go | 9 ++- pipeline/pipeline.go | 68 +++++++++++++------- 4 files changed, 184 insertions(+), 37 deletions(-) diff --git a/fd/util.go b/fd/util.go index 9aa5af788..b29eed767 100644 --- a/fd/util.go +++ b/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,51 @@ 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 + } + + 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) diff --git a/pipeline/antispam/antispammer.go b/pipeline/antispam/antispammer.go index 229165650..d432d62e4 100644 --- a/pipeline/antispam/antispammer.go +++ b/pipeline/antispam/antispammer.go @@ -26,10 +26,11 @@ 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 @@ -37,12 +38,17 @@ type Antispammer struct { 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() { @@ -252,6 +290,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"` diff --git a/pipeline/antispam/antispammer_test.go b/pipeline/antispam/antispammer_test.go index c0c5552ce..5bfe6f0da 100644 --- a/pipeline/antispam/antispammer_test.go +++ b/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) diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index e322d35f7..09adba299 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -24,24 +24,28 @@ import ( ) const ( - DefaultAntispamThreshold = -1 - DefaultSourceNameMetaField = "" - DefaultDecoder = "auto" - DefaultIsStrict = false - DefaultStreamField = "stream" - DefaultCapacity = 1024 - DefaultAvgInputEventSize = 4 * 1024 - DefaultMaxInputEventSize = 0 - DefaultCutOffEventByLimit = false - DefaultCutOffEventByLimitField = "" - DefaultJSONNodePoolSize = 16 - DefaultMaintenanceInterval = time.Second * 5 - DefaultEventTimeout = time.Second * 30 - DefaultFieldValue = "not_set" - DefaultStreamName = StreamName("not_set") - DefaultMetricHoldDuration = time.Minute * 30 - DefaultMetaCacheSize = 1024 - DefaultMetricMaxLabelValueLength = 0 + DefaultAntispamThreshold = -1 + DefaultSourceNameMetaField = "" + DefaultDecoder = "auto" + DefaultIsStrict = false + DefaultStreamField = "stream" + DefaultCapacity = 1024 + DefaultAvgInputEventSize = 4 * 1024 + DefaultMaxInputEventSize = 0 + DefaultCutOffEventByLimit = false + DefaultCutOffEventByLimitField = "" + DefaultJSONNodePoolSize = 16 + DefaultMaintenanceInterval = time.Second * 5 + DefaultEventTimeout = time.Second * 30 + DefaultFieldValue = "not_set" + DefaultStreamName = StreamName("not_set") + DefaultMetricHoldDuration = time.Minute * 30 + DefaultMetaCacheSize = 1024 + DefaultMetricMaxLabelValueLength = 0 + DefaultBannedSourcesSampleInterval = time.Second * 3 // idk + DefaultBannedSourcesSampleFirst = 3 // too + DefaultBannedSourcesSampleThereafter = 0 // and too + DefaultBannedSourcesSampleField = "_antispam_sampled" // may be just _sampled ? EventSeqIDError = uint64(0) @@ -172,6 +176,7 @@ type AntispamSettings struct { Rules antispam.Rules Exceptions antispam.Exceptions MaintenanceInterval time.Duration + BannedSourcesSample *antispam.BannedSourcesSampleOptions } type PoolType string @@ -221,6 +226,7 @@ func New(name string, settings *Settings, registry *prometheus.Registry, lg *zap MetricsController: metricCtl, Rules: settings.Antispam.Rules, Exceptions: settings.Antispam.Exceptions, + BannedSourcesSample: settings.Antispam.BannedSourcesSample, }), metricCtl: metricCtl, @@ -400,8 +406,9 @@ func (p *Pipeline) GetOutput() OutputPlugin { // In decodes message and passes it to event stream. func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, bytes []byte, isNewSource bool, meta metadata.MetaData) (seqID uint64) { var ( - ok bool - cutoff bool + ok bool + cutoff bool + antispamSampled bool ) // don't process mud. bytes, cutoff, ok = p.checkInputBytes(bytes, sourceName, meta) @@ -476,10 +483,11 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt p.Error(fmt.Sprintf("cannot parse raw time %s: %v", row.Time, err)) } } - isSpam := p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) - if isSpam { + isSpam, sampled := p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) + if isSpam && !sampled { return EventSeqIDError } + antispamSampled = sampled } p.inputEvents.Inc() @@ -550,6 +558,22 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt if cutoff && p.settings.CutOffEventByLimitField != "" { event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true) } + if antispamSampled { + cfg := p.settings.Antispam.BannedSourcesSample + event.Root.AddFieldNoAlloc(event.Root, cfg.SampledField).MutateToBool(true) + + if sm := p.antispamer.SampledMetric(); sm != nil { + values := make([]string, 0, len(cfg.SampledMetricLabels)) + for _, path := range cfg.SampledMetricLabels { + v := "not_set" + if node := event.Root.Dig(path); node != nil { + v = node.AsString() + } + values = append(values, v) + } + sm.WithLabelValues(values...).Inc() + } + } event.Offset = offsets.current event.SourceID = sourceID From 9fce437ee13205939cf8fc6eb2fa927a8b629795 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Thu, 13 Aug 2026 20:05:24 +0300 Subject: [PATCH 2/8] update --- fd/util.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/fd/util.go b/fd/util.go index b29eed767..513478d0b 100644 --- a/fd/util.go +++ b/fd/util.go @@ -266,6 +266,8 @@ func extractBannedSourcesSample(settings *simplejson.Json) (*antispam.BannedSour interval = pipeline.DefaultBannedSourcesSampleInterval } options.Interval = interval + } else { + options.Interval = pipeline.DefaultBannedSourcesSampleInterval } first := sampleJSON.Get("first").MustInt() From 6dc5c1c68261b34dd514bf10ce2054fda170c5aa Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Fri, 14 Aug 2026 02:11:47 +0300 Subject: [PATCH 3/8] add sample counter reset --- pipeline/antispam/antispammer.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pipeline/antispam/antispammer.go b/pipeline/antispam/antispammer.go index d432d62e4..eafec51fd 100644 --- a/pipeline/antispam/antispammer.go +++ b/pipeline/antispam/antispammer.go @@ -251,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 { From b5430ec084242ad80327ae315cdf2a07ea0a800a Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Thu, 10 Sep 2026 12:48:04 +0300 Subject: [PATCH 4/8] some refactor + update antispam/README --- fd/util.go | 58 +++++++++++------------- pipeline/antispam/README.md | 43 ++++++++++++++++++ pipeline/antispam/antispammer.go | 75 ++++++++++++++++---------------- pipeline/pipeline.go | 44 +++++++++---------- 4 files changed, 128 insertions(+), 92 deletions(-) diff --git a/fd/util.go b/fd/util.go index 513478d0b..d03cb1fca 100644 --- a/fd/util.go +++ b/fd/util.go @@ -255,45 +255,39 @@ func extractBannedSourcesSample(settings *simplejson.Json) (*antispam.BannedSour 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 + intervalStr := sampleJSON.Get("interval").MustString() + if intervalStr == "" { + return nil, fmt.Errorf("banned_sources_sample.interval is required") } - - first := sampleJSON.Get("first").MustInt() - if first < pipeline.DefaultBannedSourcesSampleFirst { - logger.Warnf("first must be >= default value, using default %d", pipeline.DefaultBannedSourcesSampleFirst) - first = pipeline.DefaultBannedSourcesSampleFirst + interval, err := time.ParseDuration(intervalStr) + if err != nil { + return nil, fmt.Errorf("parse banned_sources_sample.interval: %w", err) } - 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 + if interval <= 0 { + return nil, fmt.Errorf("banned_sources_sample.interval must be > 0, got %s", interval) } - options.Thereafter = thereafter - if str := sampleJSON.Get("sampled_field").MustString(); str == "" { - options.SampledField = pipeline.DefaultBannedSourcesSampleField - } else { - options.SampledField = str + first := sampleJSON.Get("first").MustInt64() + if first < 0 { + return nil, fmt.Errorf("banned_sources_sample.first must be >= 0, got %d", first) } - options.SampledMetricName = sampleJSON.Get("sampled_metric_name").MustString() - options.SampledMetricLabels = sampleJSON.Get("sampled_metric_labels").MustStringArray() + thereafter := sampleJSON.Get("thereafter").MustInt64() + if thereafter < 0 { + return nil, fmt.Errorf("banned_sources_sample.thereafter must be >= 0, got %d", thereafter) + } + if first == 0 && thereafter == 0 { + return nil, fmt.Errorf("banned_sources_sample: at least one of first/thereafter must be > 0") + } - return options, nil + return &antispam.BannedSourcesSampleOptions{ + Interval: interval, + First: first, + Thereafter: thereafter, + SampledField: sampleJSON.Get("sampled_field").MustString(), + SampledMetricName: sampleJSON.Get("sampled_metric_name").MustString(), + SampledMetricLabels: sampleJSON.Get("sampled_metric_labels").MustStringArray(), + }, nil } func extractMatchMode(actionJSON *simplejson.Json) pipeline.MatchMode { diff --git a/pipeline/antispam/README.md b/pipeline/antispam/README.md index b31428e4d..3af7611ae 100644 --- a/pipeline/antispam/README.md +++ b/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.
+ +# Banned sources sample + +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 `banned_sources_sample` 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 events can be marked with a boolean field so you can tell sampled events apart from default ones. + +## Sample parameters + +**`interval`** *`duration`* *`required`* + +Sampler window per source. + +
+ +**`first`** *`int`* + +Number of events from a banned source that always pass at the start of each `interval` window. + +
+ +**`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`. + +
+ +**`sampled_field`** *`string`* + +Field to add to log if it was let through the banned sampler. E.g. with `sampled_field:_antispam_sampled`, if the log was sampled, the output +event will have field `"_antispam_sampled":true`. Only works if the "banned_sources_sample" block is set. Useful for marking sampled logs. + +
+ +**`sampled_metric_name`** *`string`* + +Name of the metric registered for events that passed through the banned sampler. + +
+ +*`sampled_metric_labels`* *`[]string`* + +Lists the log fields to add to the metric. Blank list means no labels. + +
diff --git a/pipeline/antispam/antispammer.go b/pipeline/antispam/antispammer.go index eafec51fd..78e8efe86 100644 --- a/pipeline/antispam/antispammer.go +++ b/pipeline/antispam/antispammer.go @@ -26,7 +26,7 @@ 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 @@ -42,13 +42,11 @@ type Antispammer struct { } type source struct { - counter *atomic.Int32 - timestamp *atomic.Int64 - name string - - sampleMu sync.Mutex - sampleUntil time.Time - sampleCounter int + counter *atomic.Int32 + timestamp *atomic.Int64 + sampleUntil *atomic.Int64 + sampleCounter *atomic.Int64 + name string } type Options struct { @@ -65,8 +63,8 @@ type Options struct { type BannedSourcesSampleOptions struct { Interval time.Duration - First int - Thereafter int + First int64 + Thereafter int64 SampledField string SampledMetricName string SampledMetricLabels []string @@ -83,7 +81,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, @@ -103,10 +101,10 @@ func NewAntispammer(o *Options) *Antispammer { if o.BannedSourcesSample != nil { a.bannedSourcesSample = o.BannedSourcesSample - o.Logger.Info("antispam banned sources sample enabled", + o.Logger.Info("sampling of banned sources is enabled", zap.Duration("interval", o.BannedSourcesSample.Interval), - zap.Int("first", o.BannedSourcesSample.First), - zap.Int("thereafter", o.BannedSourcesSample.Thereafter), + zap.Int64("first", o.BannedSourcesSample.First), + zap.Int64("thereafter", o.BannedSourcesSample.Thereafter), ) if o.BannedSourcesSample.SampledMetricName != "" { @@ -184,10 +182,12 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b if newSrc, has := a.sources[id]; has { src = newSrc } else { - src = &source{ - counter: &atomic.Int32{}, - name: name, - timestamp: &atomic.Int64{}, + src = source{ + counter: &atomic.Int32{}, + timestamp: &atomic.Int64{}, + sampleUntil: &atomic.Int64{}, + sampleCounter: &atomic.Int64{}, + name: name, } src.timestamp.Add(timeEventSeconds) a.sources[id] = src @@ -251,10 +251,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.sampleMu.Lock() - source.sampleUntil = time.Time{} - source.sampleCounter = 0 - source.sampleMu.Unlock() + source.sampleCounter.Store(0) + source.sampleUntil.Store(0) } if x >= threshold { @@ -294,34 +292,37 @@ func (a *Antispammer) Dump() string { return out } -func (a *Antispammer) samplerAllow(src *source) bool { +func (a *Antispammer) SampledMetric() *metric.CounterVec { + return a.sampledMetric +} + +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 + now := time.Now().UnixNano() + for { + end := src.sampleUntil.Load() + if now < end { + break + } + if src.sampleUntil.CompareAndSwap(end, now+cfg.Interval.Nanoseconds()) { + src.sampleCounter.Store(0) + break + } } - src.sampleCounter++ - if src.sampleCounter <= cfg.First { + val := src.sampleCounter.Inc() + if val <= 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 + return (val-cfg.First)%cfg.Thereafter == 0 } type Exception struct { diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index 09adba299..1d4675526 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -24,28 +24,24 @@ import ( ) const ( - DefaultAntispamThreshold = -1 - DefaultSourceNameMetaField = "" - DefaultDecoder = "auto" - DefaultIsStrict = false - DefaultStreamField = "stream" - DefaultCapacity = 1024 - DefaultAvgInputEventSize = 4 * 1024 - DefaultMaxInputEventSize = 0 - DefaultCutOffEventByLimit = false - DefaultCutOffEventByLimitField = "" - DefaultJSONNodePoolSize = 16 - DefaultMaintenanceInterval = time.Second * 5 - DefaultEventTimeout = time.Second * 30 - DefaultFieldValue = "not_set" - DefaultStreamName = StreamName("not_set") - DefaultMetricHoldDuration = time.Minute * 30 - DefaultMetaCacheSize = 1024 - DefaultMetricMaxLabelValueLength = 0 - DefaultBannedSourcesSampleInterval = time.Second * 3 // idk - DefaultBannedSourcesSampleFirst = 3 // too - DefaultBannedSourcesSampleThereafter = 0 // and too - DefaultBannedSourcesSampleField = "_antispam_sampled" // may be just _sampled ? + DefaultAntispamThreshold = -1 + DefaultSourceNameMetaField = "" + DefaultDecoder = "auto" + DefaultIsStrict = false + DefaultStreamField = "stream" + DefaultCapacity = 1024 + DefaultAvgInputEventSize = 4 * 1024 + DefaultMaxInputEventSize = 0 + DefaultCutOffEventByLimit = false + DefaultCutOffEventByLimitField = "" + DefaultJSONNodePoolSize = 16 + DefaultMaintenanceInterval = time.Second * 5 + DefaultEventTimeout = time.Second * 30 + DefaultFieldValue = "not_set" + DefaultStreamName = StreamName("not_set") + DefaultMetricHoldDuration = time.Minute * 30 + DefaultMetaCacheSize = 1024 + DefaultMetricMaxLabelValueLength = 0 EventSeqIDError = uint64(0) @@ -560,7 +556,9 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt } if antispamSampled { cfg := p.settings.Antispam.BannedSourcesSample - event.Root.AddFieldNoAlloc(event.Root, cfg.SampledField).MutateToBool(true) + if cfg.SampledField != "" { + event.Root.AddFieldNoAlloc(event.Root, cfg.SampledField).MutateToBool(true) + } if sm := p.antispamer.SampledMetric(); sm != nil { values := make([]string, 0, len(cfg.SampledMetricLabels)) From 2b687329b7ec9ee322de2f8bc9662d4156ecee23 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Thu, 10 Sep 2026 12:53:12 +0300 Subject: [PATCH 5/8] fix antispam/README --- pipeline/antispam/README.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/pipeline/antispam/README.md b/pipeline/antispam/README.md index 3af7611ae..c40f0a37d 100644 --- a/pipeline/antispam/README.md +++ b/pipeline/antispam/README.md @@ -79,11 +79,11 @@ Flag indicating whether to check source name. If set to `true` source name will
-# Banned sources sample +## Banned sources sample -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 `banned_sources_sample` 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 events can be marked with a boolean field so you can tell sampled events apart from default ones. +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 `banned_sources_sample` 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. -## Sample parameters +### Sample parameters **`interval`** *`duration`* *`required`* @@ -105,8 +105,8 @@ After `first` events in the window have passed, every `thereafter`-th event is l **`sampled_field`** *`string`* -Field to add to log if it was let through the banned sampler. E.g. with `sampled_field:_antispam_sampled`, if the log was sampled, the output -event will have field `"_antispam_sampled":true`. Only works if the "banned_sources_sample" block is set. Useful for marking sampled logs. +Field to add to log if it was let through the banned sampler. E.g. with `sampled_field: _antispam_sampled`, if the log was sampled, the output +event will have field `"_antispam_sampled":true`. Only works if the `banned_sources_sample` block is set. Useful for marking sampled logs.
@@ -116,7 +116,7 @@ Name of the metric registered for events that passed through the banned sampler.
-*`sampled_metric_labels`* *`[]string`* +**`sampled_metric_labels`** *`[]string`* Lists the log fields to add to the metric. Blank list means no labels. From fb32a2a22d4047b204287f5662b74442b6ae0495 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Mon, 14 Sep 2026 17:21:47 +0300 Subject: [PATCH 6/8] update sampler - add enum, refactor: isSampled, README.md --- fd/util.go | 38 ++++++------ pipeline/antispam/README.md | 16 ++--- pipeline/antispam/antispammer.go | 88 ++++++++++++++------------- pipeline/antispam/antispammer_test.go | 10 ++- pipeline/pipeline.go | 25 ++++---- 5 files changed, 92 insertions(+), 85 deletions(-) diff --git a/fd/util.go b/fd/util.go index d03cb1fca..2071c736a 100644 --- a/fd/util.go +++ b/fd/util.go @@ -34,7 +34,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { antispamMaintenanceInterval := pipeline.DefaultMaintenanceInterval var antispamExceptions antispam.Exceptions var antispamRules antispam.Rules - var bannedSourcesSample *antispam.BannedSourcesSampleOptions + var antispamSampler *antispam.Sampler metricHoldDuration := pipeline.DefaultMetricHoldDuration metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength @@ -127,9 +127,9 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { logger.Fatalf("extract antispam rules: %s", err) } - bannedSourcesSample, err = extractBannedSourcesSample(antispamSettings) + antispamSampler, err = extractAntispamSampler(antispamSettings) if err != nil { - logger.Fatalf("extract antispam banned sources sample: %s", err) + logger.Fatalf("extract antispam sampler: %s", err) } sourceNameMetaField = settings.Get("source_name_meta_field").MustString() @@ -173,7 +173,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { Rules: antispamRules, Exceptions: antispamExceptions, MaintenanceInterval: antispamMaintenanceInterval, - BannedSourcesSample: bannedSourcesSample, + Sampler: antispamSampler, }, SourceNameMetaField: sourceNameMetaField, MaintenanceInterval: maintenanceInterval, @@ -249,44 +249,44 @@ func extractAntispamRules(settings *simplejson.Json, antispamMaintenanceInterval return rules, nil } -func extractBannedSourcesSample(settings *simplejson.Json) (*antispam.BannedSourcesSampleOptions, error) { - sampleJSON, ok := settings.CheckGet("banned_sources_sample") +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("banned_sources_sample.interval is required") + return nil, fmt.Errorf("interval is required") } interval, err := time.ParseDuration(intervalStr) if err != nil { - return nil, fmt.Errorf("parse banned_sources_sample.interval: %w", err) + return nil, fmt.Errorf("parse interval: %w", err) } if interval <= 0 { - return nil, fmt.Errorf("banned_sources_sample.interval must be > 0, got %s", interval) + return nil, fmt.Errorf("interval must be > 0, got %s", interval) } first := sampleJSON.Get("first").MustInt64() if first < 0 { - return nil, fmt.Errorf("banned_sources_sample.first must be >= 0, got %d", first) + return nil, fmt.Errorf("first must be >= 0, got %d", first) } thereafter := sampleJSON.Get("thereafter").MustInt64() if thereafter < 0 { - return nil, fmt.Errorf("banned_sources_sample.thereafter must be >= 0, got %d", thereafter) + return nil, fmt.Errorf("thereafter must be >= 0, got %d", thereafter) } if first == 0 && thereafter == 0 { - return nil, fmt.Errorf("banned_sources_sample: at least one of first/thereafter must be > 0") + return nil, fmt.Errorf("at least one of first/thereafter must be > 0") } - return &antispam.BannedSourcesSampleOptions{ - Interval: interval, - First: first, - Thereafter: thereafter, - SampledField: sampleJSON.Get("sampled_field").MustString(), - SampledMetricName: sampleJSON.Get("sampled_metric_name").MustString(), - SampledMetricLabels: sampleJSON.Get("sampled_metric_labels").MustStringArray(), + 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 } diff --git a/pipeline/antispam/README.md b/pipeline/antispam/README.md index c40f0a37d..70dacc01b 100644 --- a/pipeline/antispam/README.md +++ b/pipeline/antispam/README.md @@ -79,11 +79,11 @@ Flag indicating whether to check source name. If set to `true` source name will
-## Banned sources sample +## 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 `banned_sources_sample` 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. +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. -### Sample parameters +### Sampler parameters **`interval`** *`duration`* *`required`* @@ -103,20 +103,20 @@ After `first` events in the window have passed, every `thereafter`-th event is l
-**`sampled_field`** *`string`* +**`marker_field`** *`string`* -Field to add to log if it was let through the banned sampler. E.g. with `sampled_field: _antispam_sampled`, if the log was sampled, the output -event will have field `"_antispam_sampled":true`. Only works if the `banned_sources_sample` block is set. Useful for marking sampled logs. +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.
-**`sampled_metric_name`** *`string`* +**`metric_name`** *`string`* Name of the metric registered for events that passed through the banned sampler.
-**`sampled_metric_labels`** *`[]string`* +**`metric_labels`** *`[]string`* Lists the log fields to add to the metric. Blank list means no labels. diff --git a/pipeline/antispam/antispammer.go b/pipeline/antispam/antispammer.go index 78e8efe86..c75b33814 100644 --- a/pipeline/antispam/antispammer.go +++ b/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,7 +41,7 @@ type Antispammer struct { sourcesThresholds map[string]int exceptions Exceptions rules Rules - bannedSourcesSample *BannedSourcesSampleOptions + sampler *Sampler logger *zap.Logger @@ -38,7 +49,7 @@ type Antispammer struct { activeMetric *metric.Gauge banMetric *metric.GaugeVec exceptionMetric *metric.CounterVec - sampledMetric *metric.CounterVec + samplerMetric *metric.CounterVec } type source struct { @@ -55,19 +66,19 @@ type Options struct { UnbanIterations int Exceptions Exceptions Rules Rules - BannedSourcesSample *BannedSourcesSampleOptions + Sampler *Sampler Logger *zap.Logger MetricsController *metric.Ctl } -type BannedSourcesSampleOptions struct { - Interval time.Duration - First int64 - Thereafter int64 - SampledField string - SampledMetricName string - SampledMetricLabels []string +type Sampler struct { + Interval time.Duration + First int64 + Thereafter int64 + MarkerField string + MetricName string + MetricLabels []string } func NewAntispammer(o *Options) *Antispammer { @@ -99,19 +110,19 @@ func NewAntispammer(o *Options) *Antispammer { ), } - if o.BannedSourcesSample != nil { - a.bannedSourcesSample = o.BannedSourcesSample + if o.Sampler != nil { + a.sampler = o.Sampler o.Logger.Info("sampling of banned sources is enabled", - zap.Duration("interval", o.BannedSourcesSample.Interval), - zap.Int64("first", o.BannedSourcesSample.First), - zap.Int64("thereafter", o.BannedSourcesSample.Thereafter), + zap.Duration("interval", o.Sampler.Interval), + zap.Int64("first", o.Sampler.First), + zap.Int64("thereafter", o.Sampler.Thereafter), ) - if o.BannedSourcesSample.SampledMetricName != "" { - a.sampledMetric = o.MetricsController.RegisterCounterVec( - o.BannedSourcesSample.SampledMetricName, + 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.BannedSourcesSample.SampledMetricLabels..., + o.Sampler.MetricLabels..., ) } } @@ -121,9 +132,9 @@ func NewAntispammer(o *Options) *Antispammer { return a } -func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []byte, timeEvent time.Time, meta map[string]string) (spam, sampled 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, false + return Pass } threshold := a.threshold @@ -138,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, false + return Pass } } } else { @@ -154,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, false + return Pass case thresholdBlocked: - return true, false + return Drop } threshold = rule.Threshold @@ -166,9 +177,9 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b switch threshold { case thresholdUnlimited: - return false, false + return Pass case thresholdBlocked: - return true, false + return Drop } a.mu.RLock() @@ -198,7 +209,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b if isNewSource { src.counter.Swap(0) - return false, false + return Pass } x := src.counter.Load() @@ -219,12 +230,12 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b } if x >= int32(threshold) { - if a.samplerAllow(src) { - return true, true + if a.isSampled(src) { + return Sampled } - return true, false + return Drop } - return false, false + return Pass } func (a *Antispammer) Maintenance() { @@ -292,25 +303,20 @@ func (a *Antispammer) Dump() string { return out } -func (a *Antispammer) SampledMetric() *metric.CounterVec { - return a.sampledMetric +func (a *Antispammer) SamplerMetric() *metric.CounterVec { + return a.samplerMetric } -func (a *Antispammer) samplerAllow(src source) bool { - cfg := a.bannedSourcesSample +func (a *Antispammer) isSampled(src source) bool { + cfg := a.sampler if cfg == nil { return false } now := time.Now().UnixNano() - for { - end := src.sampleUntil.Load() - if now < end { - break - } + if end := src.sampleUntil.Load(); now >= end { if src.sampleUntil.CompareAndSwap(end, now+cfg.Interval.Nanoseconds()) { src.sampleCounter.Store(0) - break } } diff --git a/pipeline/antispam/antispammer_test.go b/pipeline/antispam/antispammer_test.go index 5bfe6f0da..5e592136c 100644 --- a/pipeline/antispam/antispammer_test.go +++ b/pipeline/antispam/antispammer_test.go @@ -34,8 +34,7 @@ func TestAntispam(t *testing.T) { startTime := time.Now() checkSpam := func(i int) bool { eventTime := startTime.Add(time.Duration(i) * maintenanceInterval / 2) - spam, _ := antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) - return spam + return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) == Drop } for i := 1; i < threshold; i++ { @@ -65,8 +64,7 @@ func TestAntispamAfterRestart(t *testing.T) { startTime := time.Now() checkSpam := func(i int) bool { eventTime := startTime.Add(time.Duration(i) * maintenanceInterval) - spam, _ := antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) - return spam + return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) == Drop } for i := 1; i < threshold; i++ { @@ -214,8 +212,8 @@ func TestAntispamRules(t *testing.T) { } checkSpam := func(expected bool, source, event string, meta map[string]string) { - spam, _ := antispamer.IsSpam(source, source, false, []byte(event), now, meta) - r.Equal(expected, spam) + got := antispamer.IsSpam(source, source, false, []byte(event), now, meta) + r.Equal(expected, got) } checkSpam(true, "test_source_name", `{"level":"info","message":test"}`, nil) diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index 1d4675526..c1b4078e3 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -172,7 +172,7 @@ type AntispamSettings struct { Rules antispam.Rules Exceptions antispam.Exceptions MaintenanceInterval time.Duration - BannedSourcesSample *antispam.BannedSourcesSampleOptions + Sampler *antispam.Sampler } type PoolType string @@ -222,7 +222,7 @@ func New(name string, settings *Settings, registry *prometheus.Registry, lg *zap MetricsController: metricCtl, Rules: settings.Antispam.Rules, Exceptions: settings.Antispam.Exceptions, - BannedSourcesSample: settings.Antispam.BannedSourcesSample, + Sampler: settings.Antispam.Sampler, }), metricCtl: metricCtl, @@ -479,11 +479,13 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt p.Error(fmt.Sprintf("cannot parse raw time %s: %v", row.Time, err)) } } - isSpam, sampled := p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) - if isSpam && !sampled { + + switch p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) { + case antispam.Drop: return EventSeqIDError + case antispam.Sampled: + antispamSampled = true } - antispamSampled = sampled } p.inputEvents.Inc() @@ -555,14 +557,15 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true) } if antispamSampled { - cfg := p.settings.Antispam.BannedSourcesSample - if cfg.SampledField != "" { - event.Root.AddFieldNoAlloc(event.Root, cfg.SampledField).MutateToBool(true) + // cfg is always non-nil here: antispamSampled is set only when the sampler is configured. + cfg := p.settings.Antispam.Sampler + if cfg.MarkerField != "" { + event.Root.AddFieldNoAlloc(event.Root, cfg.MarkerField).MutateToBool(true) } - if sm := p.antispamer.SampledMetric(); sm != nil { - values := make([]string, 0, len(cfg.SampledMetricLabels)) - for _, path := range cfg.SampledMetricLabels { + if sm := p.antispamer.SamplerMetric(); sm != nil { + values := make([]string, 0, len(cfg.MetricLabels)) + for _, path := range cfg.MetricLabels { v := "not_set" if node := event.Root.Dig(path); node != nil { v = node.AsString() From 519ede75bce993dba183caf9bdac46e2a119fe36 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Mon, 14 Sep 2026 17:30:32 +0300 Subject: [PATCH 7/8] fix test --- pipeline/antispam/antispammer_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pipeline/antispam/antispammer_test.go b/pipeline/antispam/antispammer_test.go index 5e592136c..663a0a8ff 100644 --- a/pipeline/antispam/antispammer_test.go +++ b/pipeline/antispam/antispammer_test.go @@ -212,7 +212,7 @@ func TestAntispamRules(t *testing.T) { } checkSpam := func(expected bool, source, event string, meta map[string]string) { - got := antispamer.IsSpam(source, source, false, []byte(event), now, meta) + got := antispamer.IsSpam(source, source, false, []byte(event), now, meta) == Drop r.Equal(expected, got) } From c6c517ea9c8e9b007efdea6f8737a173b0273ce5 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Tue, 15 Sep 2026 14:55:31 +0300 Subject: [PATCH 8/8] some refactor --- pipeline/antispam/README.md | 4 +- pipeline/antispam/antispammer.go | 48 ++++++++--------- pipeline/antispam/antispammer_test.go | 75 +++++++++++++++++++++++++-- pipeline/pipeline.go | 2 +- 4 files changed, 99 insertions(+), 30 deletions(-) diff --git a/pipeline/antispam/README.md b/pipeline/antispam/README.md index 70dacc01b..f47a44cb3 100644 --- a/pipeline/antispam/README.md +++ b/pipeline/antispam/README.md @@ -81,7 +81,7 @@ Flag indicating whether to check source name. If set to `true` source name will ## 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. +By default when a source hits the antispam threshold it is banned and all its subsequent events 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 events through. Each banned source has its own sampler state: in every `interval` window the first `first` events 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 @@ -106,7 +106,7 @@ After `first` events in the window have passed, every `thereafter`-th event is l **`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. +event will have field `"_antispam_sampled":true`. Only works if the `sampler` block is set. Useful for marking sampled events.
diff --git a/pipeline/antispam/antispammer.go b/pipeline/antispam/antispammer.go index c75b33814..986316f13 100644 --- a/pipeline/antispam/antispammer.go +++ b/pipeline/antispam/antispammer.go @@ -5,10 +5,11 @@ import ( "sync" "time" + "sync/atomic" + "github.com/ozontech/file.d/cfg/matchrule" "github.com/ozontech/file.d/logger" "github.com/ozontech/file.d/metric" - "go.uber.org/atomic" "go.uber.org/zap" ) @@ -20,10 +21,10 @@ const ( type SpamResult int const ( - // event is not spam. - Pass SpamResult = iota - //event is spam. - Drop + // event is not spam + Passed SpamResult = iota + // event is spam + Dropped // event is spam, but passed by sampler Sampled ) @@ -134,7 +135,7 @@ func NewAntispammer(o *Options) *Antispammer { 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 Pass + return Passed } threshold := a.threshold @@ -149,7 +150,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b if e.Name != "" { a.exceptionMetric.WithLabelValues(e.Name).Inc() } - return Pass + return Passed } } } else { @@ -165,9 +166,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 Pass + return Passed case thresholdBlocked: - return Drop + return Dropped } threshold = rule.Threshold @@ -177,9 +178,9 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b switch threshold { case thresholdUnlimited: - return Pass + return Passed case thresholdBlocked: - return Drop + return Dropped } a.mu.RLock() @@ -209,13 +210,13 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b if isNewSource { src.counter.Swap(0) - return Pass + return Passed } x := src.counter.Load() diff := timeEventSeconds - src.timestamp.Swap(timeEventSeconds) if diff < a.maintenanceInterval.Nanoseconds() { - x = src.counter.Inc() + x = src.counter.Add(1) } if x == int32(threshold) { src.counter.Swap(int32(a.unbanIterations * threshold)) @@ -230,12 +231,12 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b } if x >= int32(threshold) { - if a.isSampled(src) { + if a.sampler.isSampled(src) { return Sampled } - return Drop + return Dropped } - return Pass + return Passed } func (a *Antispammer) Maintenance() { @@ -307,28 +308,27 @@ func (a *Antispammer) SamplerMetric() *metric.CounterVec { return a.samplerMetric } -func (a *Antispammer) isSampled(src source) bool { - cfg := a.sampler - if cfg == nil { +func (s *Sampler) isSampled(src source) bool { + if s == nil { return false } now := time.Now().UnixNano() if end := src.sampleUntil.Load(); now >= end { - if src.sampleUntil.CompareAndSwap(end, now+cfg.Interval.Nanoseconds()) { + if src.sampleUntil.CompareAndSwap(end, now+s.Interval.Nanoseconds()) { src.sampleCounter.Store(0) } } - val := src.sampleCounter.Inc() - if val <= cfg.First { + val := src.sampleCounter.Add(1) + if val <= s.First { return true } - if cfg.Thereafter == 0 { + if s.Thereafter == 0 { return false } - return (val-cfg.First)%cfg.Thereafter == 0 + return (val-s.First)%s.Thereafter == 0 } type Exception struct { diff --git a/pipeline/antispam/antispammer_test.go b/pipeline/antispam/antispammer_test.go index 663a0a8ff..4b79ced33 100644 --- a/pipeline/antispam/antispammer_test.go +++ b/pipeline/antispam/antispammer_test.go @@ -1,7 +1,9 @@ package antispam import ( + "sync/atomic" "testing" + "testing/synctest" "time" "github.com/ozontech/file.d/cfg/matchrule" @@ -34,7 +36,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) == Drop + return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) == Dropped } for i := 1; i < threshold; i++ { @@ -64,7 +66,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) == Drop + return antispamer.IsSpam("1", "test", false, []byte(`{}`), eventTime, nil) == Dropped } for i := 1; i < threshold; i++ { @@ -212,7 +214,7 @@ func TestAntispamRules(t *testing.T) { } checkSpam := func(expected bool, source, event string, meta map[string]string) { - got := antispamer.IsSpam(source, source, false, []byte(event), now, meta) == Drop + got := antispamer.IsSpam(source, source, false, []byte(event), now, meta) == Dropped r.Equal(expected, got) } @@ -229,3 +231,70 @@ func TestAntispamRules(t *testing.T) { checkSpam(false, "test", `{"level":"info","message":test"}`, nil) checkSpam(true, "test", `{"level":"info","message":test"}`, nil) } + +func TestIsSampled(t *testing.T) { + type samplerStep struct { + timeSleep time.Duration + want bool + } + + cases := []struct { + name string + sampler *Sampler + steps []samplerStep + }{ + { + name: "first_only", + sampler: &Sampler{Interval: time.Hour, First: 3}, + steps: []samplerStep{ + {want: true}, {want: true}, {want: true}, {want: false}, {want: false}, + }, + }, + { + name: "thereafter_only", + sampler: &Sampler{Interval: time.Hour, Thereafter: 2}, + steps: []samplerStep{ + {want: false}, // 1 % 2 != 0 + {want: true}, // 2 % 2 == 0 + {want: false}, // 3 % 2 != 0 + }, + }, + { + name: "first_and_thereafter", + sampler: &Sampler{Interval: time.Hour, First: 2, Thereafter: 3}, + steps: []samplerStep{ + {want: true}, // 1 + {want: true}, // 2 + {want: false}, // (3-2)% 3 != 0 + {want: false}, // (4-2) % 3 != 0 + {want: true}, // (5-2) % 3 == 0 + {want: false}, // (6-2) % 3 != 0 + }, + }, + { + name: "window_rotation", + sampler: &Sampler{Interval: time.Second, First: 2}, + steps: []samplerStep{ + {want: true}, {want: true}, {want: false}, + {timeSleep: 2 * time.Second, want: true}, {want: true}, {want: false}, + }, + }, + } + + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + src := source{ + sampleUntil: &atomic.Int64{}, + sampleCounter: &atomic.Int64{}, + } + for _, step := range tt.steps { + if step.timeSleep > 0 { + time.Sleep(step.timeSleep) + } + require.Equal(t, step.want, tt.sampler.isSampled(src)) + } + }) + }) + } +} diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index c1b4078e3..6ba6901f2 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -481,7 +481,7 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt } switch p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) { - case antispam.Drop: + case antispam.Dropped: return EventSeqIDError case antispam.Sampled: antispamSampled = true