fix antispam && test (#593)

pipeline/antispam/antispammer.go

@@ -33,7 +33,7 @@ type Antispammer struct {
 }
 
 type source struct {
-	counter atomic.Int32
+	counter *atomic.Int32
 	name    string
 }
 
@@ -103,7 +103,7 @@ func (a *Antispammer) IsSpam(id uint64, name string, isNewSource bool, event []b
 			src = newSrc
 		} else {
 			src = source{
-				counter: atomic.Int32{},
+				counter: &atomic.Int32{},
 				name:    name,
 			}
 			a.sources[id] = src

pipeline/antispam/antispammer_test.go

@@ -0,0 +1,44 @@
+package antispam
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ozontech/file.d/logger"
+	"github.com/ozontech/file.d/metric"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAntispam(t *testing.T) {
+	r := require.New(t)
+
+	threshold := 5
+	unbanIterations := 2
+	antispamer := NewAntispammer(Options{
+		MaintenanceInterval: time.Second * 1,
+		Threshold:           threshold,
+		UnbanIterations:     unbanIterations,
+		Logger:              logger.Instance.Named("antispam").Desugar(),
+		MetricsController:   metric.NewCtl("test", prometheus.NewRegistry()),
+	})
+
+	checkSpam := func() bool {
+		return antispamer.IsSpam(1, "test", false, []byte(`{}`))
+	}
+
+	for i := 0; i < threshold-1; i++ {
+		result := checkSpam()
+		r.False(result)
+	}
+
+	result := checkSpam()
+	r.True(result)
+
+	for i := 0; i <= unbanIterations; i++ {
+		antispamer.Maintenance()
+	}
+
+	result = checkSpam()
+	r.False(result)
+}