New antispam

Insanedocfile

@@ -3,21 +3,17 @@ extractors:
   fn-list: '"fn-list" #4 /Plugin\)\s(.+)\s{/'
   match-modes: '"match-modes" /MatchMode(.*),/ /\"(.*)\"/'
   do-if-node: '"do-if-node" /Node(\w+)\s/'
-  do-if-field-op: '"do-if-field-op" /field(\w+)OpTag\s/'
-  do-if-logical-op: '"do-if-logical-op" /logical(\w+)Tag\s/'
 decorators:
   config-params: '_ _ /*`%s`* / /*`default=%s`* / /*`%s`* / /*`options=%s`* /'
   fn-list: '_ _ /`%s`/'
   match-modes: '_ /%s/ /`match_mode: %s`/'
   do-if-node: '_ /%s/'
-  do-if-field-op: '_ /%s/'
-  do-if-logical-op: '_ /%s/'
 templates:
   - template: docs/*.idoc.md
     files: ["../pipeline/*.go"]
   - template: pipeline/*.idoc.md
     files: ["*.go"]
-  - template: pipeline/doif/*.idoc.md
+  - template: pipeline/do_if/*.idoc.md
     files: ["*.go"]
   - template: plugin/*/*/README.idoc.md
     files: ["*.go"]

cfg/config.go

@@ -731,3 +731,20 @@ func mergeYAMLs(a, b map[interface{}]interface{}) map[interface{}]interface{} {
 	}
 	return merged
 }
+
+func AnyToInt(v any) (int, error) {
+	switch vNum := v.(type) {
+	case int:
+		return vNum, nil
+	case float64:
+		return int(vNum), nil
+	case json.Number:
+		vInt64, err := vNum.Int64()
+		if err != nil {
+			return 0, err
+		}
+		return int(vInt64), nil
+	default:
+		return 0, fmt.Errorf("not convertable to int: value=%v type=%T", v, v)
+	}
+}

cfg/matchrule/matchrule.go

@@ -27,6 +27,19 @@ func (m *Mode) UnmarshalJSON(i []byte) error {
 	return nil
 }
 
+func (m *Mode) ToString() string {
+	switch *m {
+	case ModeContains:
+		return "contains"
+	case ModePrefix:
+		return "prefix"
+	case ModeSuffix:
+		return "suffix"
+	default:
+		panic("unreachable")
+	}
+}
+
 const (
 	ModePrefix Mode = iota
 	ModeContains
@@ -66,6 +79,14 @@ type Rule struct {
 	prepared     bool
 }
 
+func (r *Rule) GetMinValueSize() int {
+	return r.minValueSize
+}
+
+func (r *Rule) GetMaxValueSize() int {
+	return r.maxValueSize
+}
+
 func (r *Rule) Prepare() {
 	if len(r.Values) == 0 {
 		return
@@ -186,6 +207,17 @@ var (
 	condOrBytes  = []byte(`"or"`)
 )
 
+func (c *Cond) ToString() string {
+	switch *c {
+	case CondAnd:
+		return "and"
+	case CondOr:
+		return "or"
+	default:
+		panic("unreachable")
+	}
+}
+
 type RuleSet struct {
 	// > @3@4@5@6
 	// >

decoder/common.go

@@ -1,27 +1,5 @@
 package decoder
 
-import (
-	"encoding/json"
-	"errors"
-)
-
-func anyToInt(v any) (int, error) {
-	switch vNum := v.(type) {
-	case int:
-		return vNum, nil
-	case float64:
-		return int(vNum), nil
-	case json.Number:
-		vInt64, err := vNum.Int64()
-		if err != nil {
-			return 0, err
-		}
-		return int(vInt64), nil
-	default:
-		return 0, errors.New("value is not convertable to int")
-	}
-}
-
 // atoi is allocation free ASCII number to integer conversion
 func atoi(b []byte) (int, bool) {
 	if len(b) == 0 {

decoder/json.go

@@ -6,6 +6,7 @@ import (
 	"slices"
 	"sync"
 
+	"github.com/ozontech/file.d/cfg"
 	insaneJSON "github.com/ozontech/insane-json"
 	"github.com/tidwall/gjson"
 )
@@ -140,7 +141,7 @@ func extractJsonParams(params map[string]any) (jsonParams, error) {
 			return jsonParams{}, fmt.Errorf("%q must be map", jsonMaxFieldsSizeParam)
 		}
 		for k, v := range maxFieldsSizeMap {
-			vInt, err := anyToInt(v)
+			vInt, err := cfg.AnyToInt(v)
 			if err != nil {
 				return jsonParams{}, fmt.Errorf("each value in %q must be int", jsonMaxFieldsSizeParam)
 			}

fd/util.go

@@ -11,13 +11,14 @@ import (
 	"github.com/ozontech/file.d/logger"
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/pipeline/antispam"
-	"github.com/ozontech/file.d/pipeline/doif"
+	"github.com/ozontech/file.d/pipeline/do_if"
 )
 
 func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 	capacity := pipeline.DefaultCapacity
 	antispamThreshold := pipeline.DefaultAntispamThreshold
 	var antispamExceptions antispam.Exceptions
+	var antispamCfg map[string]any
 	sourceNameMetaField := pipeline.DefaultSourceNameMetaField
 	avgInputEventSize := pipeline.DefaultAvgInputEventSize
 	maxInputEventSize := pipeline.DefaultMaxInputEventSize
@@ -101,6 +102,8 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 		}
 		antispamExceptions.Prepare()
 
+		antispamCfg = settings.Get("antispam").MustMap()
+
 		sourceNameMetaField = settings.Get("source_name_meta_field").MustString()
 		isStrict = settings.Get("is_strict").MustBool()
 
@@ -129,6 +132,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 		CutOffEventByLimitField: cutOffEventByLimitField,
 		AntispamThreshold:       antispamThreshold,
 		AntispamExceptions:      antispamExceptions,
+		Antispam:                antispamCfg,
 		SourceNameMetaField:     sourceNameMetaField,
 		MaintenanceInterval:     maintenanceInterval,
 		EventTimeout:            eventTimeout,
@@ -219,13 +223,13 @@ func extractMetrics(actionJSON *simplejson.Json) (string, []string, bool) {
 	return metricName, metricLabels, skipStatus
 }
 
-func extractDoIfChecker(actionJSON *simplejson.Json) (*doif.Checker, error) {
+func extractDoIfChecker(actionJSON *simplejson.Json) (*do_if.Checker, error) {
 	m := actionJSON.MustMap()
 	if m == nil {
 		return nil, nil
 	}
 
-	return doif.NewFromMap(m)
+	return do_if.NewFromMap(m)
 }
 
 func makeActionJSON(actionJSON *simplejson.Json) []byte {

pipeline/antispam/README.md

@@ -6,7 +6,77 @@ In some systems services might explode with logs due to different circumstances.
 
 The main entity is `Antispammer`. It counts input data from the sources (e.g. if data comes from [file input plugin](/plugin/input/file/README.md), source can be filename) and decides whether to ban it or not. For each source it counts how many logs it has got, in other words the counter for the source is incremented for each incoming log. When the counter is greater or equal to the threshold value the source is banned until its counter is less than the threshold value. The counter value is decremented once in maintenance interval by the threshold value. The maintenance interval for antispam is the same as for the pipeline (see `maintenance_interval` in [pipeline settings](/pipeline/README.md#settings)).
 
-## Exceptions
+## Antispam config
+
+Example:
+
+```
+antispam:
+  threshold: 3000
+  rules:
+    - name: alert_agent
+      if:
+        op: and
+        operands:
+          - op: contains
+            data: meta.service
+            values:
+              - alerts-agent
+          - op: prefix
+            data: event
+            values:
+              - '{"level":"debug"'
+      threshold: -1
+    - name: viewer
+      if:
+        op: and
+        operands:
+          - op: contains
+            data: source_name
+            values:
+              - viewer
+      threshold: 5000
+```
+
+Antispammer iterates over rules, checks event and applies first matched rule.
+If event does not match any rule it will be limited with common threshold.
+
+### Antispam fields
+
+**`threshold`** **`int`** 
+
+Common threshold applied to events that don't match any rule.
+Values:
+- `-1` - no limit;
+- `0` - discard all logs;
+- `> 0` - normal threshold value.
+
+**`rules`**
+
+Antispam rules array
+
+### Rule fields
+
+**`name`** **`string`**
+
+Name of rule. If set to nonempty string, adds label value for the `name` label in the `antispam_exceptions` metric.
+
+**`threshold`** **`int`**
+
+Rule threshold. Has the same value meanings as common threshold.
+
+**`if`**
+
+`do_if`-like condition tree (see [doc](../do_if/README.md)). 
+Difference is we allowed only logical and data operations.
+We use `data` to point data to check instead of `field`.
+Values:
+- `event`
+- `source_name`
+- `meta.name` - get data to check from metadata by key `name`
+
+
+## Exceptions [deprecated: use rules instead]
 
 Antispammer has some exception rules which can be applied by checking source name or log as raw bytes contents. If the log is matched by the rules it is not accounted for in the antispammer. It might be helpful for the logs from critical infrastructure services which must not be banned at all.