fix after review

Author: DmitryRomanov

e2e/start_work_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/ozontech/file.d/fd"
 	_ "github.com/ozontech/file.d/plugin/action/add_file_name"
 	_ "github.com/ozontech/file.d/plugin/action/add_host"
+	_ "github.com/ozontech/file.d/plugin/action/cardinality"
 	_ "github.com/ozontech/file.d/plugin/action/convert_date"
 	_ "github.com/ozontech/file.d/plugin/action/convert_log_level"
 	_ "github.com/ozontech/file.d/plugin/action/convert_utf8_bytes"

plugin/action/cardinality/README.md

@@ -50,6 +50,12 @@ Leave empty for default metric naming.
 
 <br>
 
+**`metric_label_value`** *`string`* *`default=unknown`* 
+
+Value assigned to the metric label when cardinality limit is exceeded.
+
+<br>
+
 **`limit`** *`int`* *`default=10000`* 
 
 Maximum allowed number of unique values for monitored fields.

plugin/action/cardinality/cache.go

@@ -14,12 +14,12 @@ type Cache struct {
 	ttl  int64
 }
 
-func NewCache(ttl time.Duration) (*Cache, error) {
+func NewCache(ttl time.Duration) *Cache {
 	return &Cache{
 		tree: radix.New(),
 		ttl:  ttl.Nanoseconds(),
 		mu:   &sync.RWMutex{},
-	}, nil
+	}
 }
 
 func (c *Cache) Set(key string) bool {
@@ -60,8 +60,8 @@ func (c *Cache) delete(key string) {
 
 func (c *Cache) CountPrefix(prefix string) (count int) {
 	var keysToDelete []string
-	c.mu.RLock()
 	now := xtime.GetInaccurateUnixNano()
+	c.mu.RLock()
 	c.tree.WalkPrefix(prefix, func(s string, v any) bool {
 		timeValue := v.(int64)
 		if c.isExpire(now, timeValue) {

plugin/action/cardinality/cache_test.go

@@ -9,13 +9,12 @@ import (
 )
 
 func TestNewCache(t *testing.T) {
-	cache, err := NewCache(time.Minute)
-	assert.NoError(t, err)
+	cache := NewCache(time.Minute)
 	assert.NotNil(t, cache)
 }
 
 func TestSetAndExists(t *testing.T) {
-	cache, _ := NewCache(time.Minute)
+	cache := NewCache(time.Minute)
 
 	t.Run("basic set and get", func(t *testing.T) {
 		key := "test-key"
@@ -32,7 +31,7 @@ func TestSetAndExists(t *testing.T) {
 }
 
 func TestDelete(t *testing.T) {
-	cache, _ := NewCache(time.Minute)
+	cache := NewCache(time.Minute)
 
 	key := "to-delete"
 	cache.Set(key)
@@ -50,7 +49,7 @@ func TestDelete(t *testing.T) {
 }
 
 func TestCountPrefix(t *testing.T) {
-	cache, _ := NewCache(time.Minute)
+	cache := NewCache(time.Minute)
 
 	keys := []string{
 		"key1_subkey1",
@@ -86,7 +85,7 @@ func TestCountPrefix(t *testing.T) {
 }
 
 func TestConcurrentOperations(t *testing.T) {
-	cache, _ := NewCache(time.Minute)
+	cache := NewCache(time.Minute)
 
 	var wg sync.WaitGroup
 	keys := []string{"key1", "key2", "key3"}
@@ -154,7 +153,7 @@ func TestConcurrentOperations(t *testing.T) {
 }
 
 func TestTTL(t *testing.T) {
-	cache, _ := NewCache(100 * time.Millisecond)
+	cache := NewCache(100 * time.Millisecond)
 
 	key := "ttl-key"
 	cache.Set(key)

plugin/action/cardinality/cardinality.go

@@ -41,7 +41,7 @@ type Plugin struct {
 	fields []string
 	logger *zap.Logger
 
-	cardinalityDiscardCounter *prometheus.CounterVec
+	cardinalityApplyCounter *prometheus.CounterVec
 }
 
 const (
@@ -82,6 +82,11 @@ type Config struct {
 	// > Leave empty for default metric naming.
 	MetricPrefix string `json:"metric_prefix" default:""` // *
 
+	// > @3@4@5@6
+	// >
+	// > Value assigned to the metric label when cardinality limit is exceeded.
+	MetricLabelValue string `json:"metric_label_value" default:"unknown"` // *
+
 	// > @3@4@5@6
 	// >
 	// > Maximum allowed number of unique values for monitored fields.
@@ -134,13 +139,13 @@ func (p *Plugin) makeMetric(ctl *metric.Ctl, name, help string, labels ...string
 func (p *Plugin) registerMetrics(ctl *metric.Ctl, prefix string) {
 	var metricName string
 	if prefix == "" {
-		metricName = "cardinality_discard_total"
+		metricName = "cardinality_applied_total"
 	} else {
-		metricName = fmt.Sprintf(`cardinality_discard_%s_total`, prefix)
+		metricName = fmt.Sprintf(`cardinality_applied_%s_total`, prefix)
 	}
-	p.cardinalityDiscardCounter = p.makeMetric(ctl,
+	p.cardinalityApplyCounter = p.makeMetric(ctl,
 		metricName,
-		"Total number of events discarded due to cardinality limits",
+		"Total number of events applied due to cardinality limits",
 		p.keys...,
 	)
 }
@@ -149,11 +154,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginP
 	p.config = config.(*Config)
 	p.logger = params.Logger.Desugar()
 
-	var err error
-	p.cache, err = NewCache(p.config.TTL_)
-	if err != nil {
-		panic(err)
-	}
+	p.cache = NewCache(p.config.TTL_)
 
 	p.keys = make([]string, 0, len(p.config.KeyFields))
 	for _, fs := range p.config.KeyFields {
@@ -198,16 +199,16 @@ func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
 	value := mapToStringSorted(cacheKey, cacheValue)
 	keysCount := p.cache.CountPrefix(key)
 
-	if p.config.Limit > 0 && keysCount >= p.config.Limit {
+	if p.config.Limit >= 0 && keysCount >= p.config.Limit {
 		labelsValues := make([]string, 0, len(p.keys))
 		for _, key := range p.keys {
 			if val, exists := cacheKey[key]; exists {
 				labelsValues = append(labelsValues, val)
 			} else {
-				labelsValues = append(labelsValues, "unknown")
+				labelsValues = append(labelsValues, p.config.MetricLabelValue)
 			}
 		}
-		p.cardinalityDiscardCounter.WithLabelValues(labelsValues...).Inc()
+		p.cardinalityApplyCounter.WithLabelValues(labelsValues...).Inc()
 		switch p.config.Action {
 		case actionDiscard:
 			return pipeline.ActionDiscard

plugin/action/cardinality/cardinality_test.go

@@ -183,7 +183,7 @@ func TestCardinalityLimitDiscardIfNoSetKeyFields(t *testing.T) {
 }
 
 func TestSetAndCountPrefix(t *testing.T) {
-	cache, _ := NewCache(time.Minute)
+	cache := NewCache(time.Minute)
 
 	cacheKey := map[string]string{
 		"host": "localhost",