cardinality: dig sub-fields

Author: DmitryRomanov

plugin/action/cardinality/README.md

@@ -50,12 +50,6 @@ 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/cardinality.go

@@ -37,13 +37,18 @@ pipelines:
 type Plugin struct {
 	cache  *Cache
 	config *Config
-	keys   []string
-	fields []string
+	keys   []parsedField
+	fields []parsedField
 	logger *zap.Logger
 
 	cardinalityApplyCounter *prometheus.CounterVec
 }
 
+type parsedField struct {
+	name  string
+	value []string
+}
+
 const (
 	actionDiscard      = "discard"
 	actionRemoveFields = "remove_fields"
@@ -82,11 +87,6 @@ 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.
@@ -136,6 +136,14 @@ func (p *Plugin) makeMetric(ctl *metric.Ctl, name, help string, labels ...string
 	return ctl.RegisterCounterVec(name, help, labelNames...)
 }
 
+func keyMetricLabels(fields []parsedField) []string {
+	result := make([]string, 0, len(fields))
+	for i := range fields {
+		result = append(result, fields[i].name)
+	}
+	return result
+}
+
 func (p *Plugin) registerMetrics(ctl *metric.Ctl, prefix string) {
 	var metricName string
 	if prefix == "" {
@@ -146,7 +154,7 @@ func (p *Plugin) registerMetrics(ctl *metric.Ctl, prefix string) {
 	p.cardinalityApplyCounter = p.makeMetric(ctl,
 		metricName,
 		"Total number of events applied due to cardinality limits",
-		p.keys...,
+		keyMetricLabels(p.keys)...,
 	)
 }
 
@@ -156,21 +164,35 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginP
 
 	p.cache = NewCache(p.config.TTL_)
 
-	p.keys = make([]string, 0, len(p.config.KeyFields))
+	p.keys = make([]parsedField, 0, len(p.config.KeyFields))
 	for _, fs := range p.config.KeyFields {
 		if fs != "" {
-			p.keys = append(p.keys, cfg.ParseFieldSelector(string(fs))[0])
+			parsedFields := cfg.ParseFieldSelector(string(fs))
+			p.keys = append(p.keys, struct {
+				name  string
+				value []string
+			}{
+				name:  strings.Join(parsedFields, "_"),
+				value: parsedFields,
+			})
 		}
 	}
 
 	if len(p.config.Fields) == 0 {
 		p.logger.Fatal("you have to set key fields")
 	}
 
-	p.fields = make([]string, 0, len(p.config.Fields))
+	p.fields = make([]parsedField, 0, len(p.config.Fields))
 	for _, fs := range p.config.Fields {
 		if fs != "" {
-			p.fields = append(p.fields, cfg.ParseFieldSelector(string(fs))[0])
+			parsedFields := cfg.ParseFieldSelector(string(fs))
+			p.fields = append(p.fields, struct {
+				name  string
+				value []string
+			}{
+				name:  strings.Join(parsedFields, "_"),
+				value: parsedFields,
+			})
 		}
 	}
 
@@ -185,27 +207,26 @@ func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
 	cacheKey := make(map[string]string, len(p.keys))
 
 	for _, key := range p.keys {
-		value := pipeline.CloneString(event.Root.Dig(key).AsString())
-		cacheKey[key] = value
+		value := pipeline.CloneString(event.Root.Dig(key.value...).AsString())
+		cacheKey[key.name] = value
 	}
 
 	cacheValue := make(map[string]string, len(p.fields))
 	for _, key := range p.fields {
-		value := pipeline.CloneString(event.Root.Dig(key).AsString())
-		cacheValue[key] = value
+		value := pipeline.CloneString(event.Root.Dig(key.value...).AsString())
+		cacheValue[key.name] = value
 	}
 
 	key := mapToKey(cacheKey)
 	value := mapToStringSorted(cacheKey, cacheValue)
 	keysCount := p.cache.CountPrefix(key)
 
 	if p.config.Limit >= 0 && keysCount >= p.config.Limit {
-		labelsValues := make([]string, 0, len(p.keys))
+		var labelsValues []string
+		labelsValues = make([]string, 0, len(p.keys))
 		for _, key := range p.keys {
-			if val, exists := cacheKey[key]; exists {
+			if val, exists := cacheKey[key.name]; exists {
 				labelsValues = append(labelsValues, val)
-			} else {
-				labelsValues = append(labelsValues, p.config.MetricLabelValue)
 			}
 		}
 		p.cardinalityApplyCounter.WithLabelValues(labelsValues...).Inc()
@@ -214,7 +235,7 @@ func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
 			return pipeline.ActionDiscard
 		case actionRemoveFields:
 			for _, key := range p.fields {
-				event.Root.Dig(key).Suicide()
+				event.Root.Dig(key.value...).Suicide()
 			}
 		}
 	} else {

plugin/action/cardinality/cardinality_test.go

@@ -59,8 +59,8 @@ func TestMapToStringSorted(t *testing.T) {
 func TestCardinalityLimitDiscard(t *testing.T) {
 	limit := 10
 	config := &Config{
-		KeyFields: []cfg.FieldSelector{"host"},
-		Fields:    []cfg.FieldSelector{"i"},
+		KeyFields: []cfg.FieldSelector{"info.host", "not_exists_fields"},
+		Fields:    []cfg.FieldSelector{"value.i"},
 		Limit:     limit,
 		Action:    actionDiscard,
 		TTL_:      1 * time.Hour,
@@ -84,7 +84,7 @@ func TestCardinalityLimitDiscard(t *testing.T) {
 	})
 
 	for i := 0; i < genEventsCnt; i++ {
-		json := fmt.Sprintf(`{"host":"localhost","i":"%d"}`, i)
+		json := fmt.Sprintf(`{"info": {"host":"localhost"},"value":{"i":"%d"}}`, i)
 		input.In(10, "test", test.NewOffset(0), []byte(json))
 	}
 	inWg.Wait()
@@ -118,17 +118,23 @@ func TestCardinalityLimitRemoveFields(t *testing.T) {
 	})
 
 	outEventsCnt := 0
+	wrongEventsCnt := 0
 	outWg := sync.WaitGroup{}
 	outWg.Add(genEventsCnt)
 	output.SetOutFn(func(e *pipeline.Event) {
 		defer outWg.Done()
 
 		// check exists field
 		value := pipeline.CloneString(e.Root.Dig(string(config.Fields[0])).AsString())
+		keyValue := pipeline.CloneString(e.Root.Dig(string(config.KeyFields[0])).AsString())
 
 		if value != "" {
 			outEventsCnt++
 		}
+
+		if keyValue == "" {
+			wrongEventsCnt++
+		}
 	})
 
 	for i := 0; i < genEventsCnt; i++ {
@@ -140,6 +146,7 @@ func TestCardinalityLimitRemoveFields(t *testing.T) {
 
 	p.Stop()
 	assert.Equal(t, inEventsCnt, genEventsCnt, "wrong in events count")
+	assert.Equal(t, 0, wrongEventsCnt)
 	assert.Equal(t, limit, outEventsCnt, "wrong out events count")
 }