Trim metrics labels

fd/file.d.go

@@ -63,7 +63,7 @@ func (f *FileD) Start() {
 }
 
 func (f *FileD) initMetrics() {
-	f.metricCtl = metric.NewCtl("file_d", f.registry, 0)
+	f.metricCtl = metric.NewCtl("file_d", f.registry, 0, 0)
 	f.versionMetric = f.metricCtl.RegisterGaugeVec("version", "", "version")
 	f.versionMetric.WithLabelValues(buildinfo.Version).Inc()
 }

fd/util.go

@@ -29,10 +29,12 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 	decoderParams := make(map[string]any)
 	isStrict := pipeline.DefaultIsStrict
 	eventTimeout := pipeline.DefaultEventTimeout
-	metricHoldDuration := pipeline.DefaultMetricHoldDuration
 	metaCacheSize := pipeline.DefaultMetaCacheSize
 	pool := ""
 
+	metricHoldDuration := pipeline.DefaultMetricHoldDuration
+	metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength
+
 	if settings != nil {
 		val := settings.Get("capacity").MustInt()
 		if val != 0 {
@@ -104,7 +106,15 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 		sourceNameMetaField = settings.Get("source_name_meta_field").MustString()
 		isStrict = settings.Get("is_strict").MustBool()
 
-		str = settings.Get("metric_hold_duration").MustString()
+		if str := settings.Get("pool").MustString(); str != "" {
+			pool = str
+		}
+
+		metrics := settings.Get("metrics")
+		str = metrics.Get("hold_duration").MustString()
+		if str == "" {
+			str = settings.Get("metric_hold_duration").MustString()
+		}
 		if str != "" {
 			i, err := time.ParseDuration(str)
 			if err != nil {
@@ -113,8 +123,10 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 			metricHoldDuration = i
 		}
 
-		if str := settings.Get("pool").MustString(); str != "" {
-			pool = str
+		metricMaxLabelValueLength = metrics.Get("max_label_value_length").MustInt()
+		if metricMaxLabelValueLength < 0 {
+			logger.Warn("negative max_label_value_length value, metric label truncation is disabled")
+			metricMaxLabelValueLength = 0
 		}
 	}
 
@@ -134,8 +146,11 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 		EventTimeout:            eventTimeout,
 		StreamField:             streamField,
 		IsStrict:                isStrict,
-		MetricHoldDuration:      metricHoldDuration,
 		Pool:                    pipeline.PoolType(pool),
+		Metric: &pipeline.MetricSettings{
+			HoldDuration:        metricHoldDuration,
+			MaxLabelValueLength: metricMaxLabelValueLength,
+		},
 	}
 }
 

metric/controller.go

@@ -21,16 +21,18 @@ type Ctl struct {
 	subsystem string
 	register  *prometheus.Registry
 
-	holder  *Holder
-	metrics map[string]prometheus.Collector
-	mu      sync.RWMutex
+	holder                    *Holder
+	metrics                   map[string]prometheus.Collector
+	metricMaxLabelValueLength int
+	mu                        sync.RWMutex
 }
 
-func NewCtl(subsystem string, registry *prometheus.Registry, metricHoldDuration time.Duration) *Ctl {
+func NewCtl(subsystem string, registry *prometheus.Registry, metricHoldDuration time.Duration, metricMaxLabelValueLength int) *Ctl {
 	ctl := &Ctl{
-		subsystem: subsystem,
-		register:  registry,
-		metrics:   make(map[string]prometheus.Collector),
+		subsystem:                 subsystem,
+		register:                  registry,
+		metrics:                   make(map[string]prometheus.Collector),
+		metricMaxLabelValueLength: metricMaxLabelValueLength,
 	}
 
 	if metricHoldDuration != 0 {
@@ -75,7 +77,7 @@ func (mc *Ctl) RegisterCounterVec(name, help string, labels ...string) *CounterV
 		Help:      help,
 	}, labels)
 
-	return newCounterVec(mc.registerMetric(name, counterVec).(*prometheus.CounterVec))
+	return newCounterVec(mc.registerMetric(name, counterVec).(*prometheus.CounterVec), mc.metricMaxLabelValueLength)
 }
 
 func (mc *Ctl) RegisterGauge(name, help string) *Gauge {
@@ -97,7 +99,7 @@ func (mc *Ctl) RegisterGaugeVec(name, help string, labels ...string) *GaugeVec {
 		Help:      help,
 	}, labels)
 
-	return newGaugeVec(mc.registerMetric(name, gaugeVec).(*prometheus.GaugeVec))
+	return newGaugeVec(mc.registerMetric(name, gaugeVec).(*prometheus.GaugeVec), mc.metricMaxLabelValueLength)
 }
 
 func (mc *Ctl) RegisterHistogram(name, help string, buckets []float64) *Histogram {
@@ -121,7 +123,7 @@ func (mc *Ctl) RegisterHistogramVec(name, help string, buckets []float64, labels
 		Buckets:   buckets,
 	}, labels)
 
-	return newHistogramVec(mc.registerMetric(name, histogramVec).(*prometheus.HistogramVec))
+	return newHistogramVec(mc.registerMetric(name, histogramVec).(*prometheus.HistogramVec), mc.metricMaxLabelValueLength)
 }
 
 func (mc *Ctl) registerMetric(name string, newMetric prometheus.Collector) prometheus.Collector {

metric/counter.go

@@ -37,10 +37,10 @@ type CounterVec struct {
 	vec   *prometheus.CounterVec
 }
 
-func newCounterVec(cv *prometheus.CounterVec) *CounterVec {
+func newCounterVec(cv *prometheus.CounterVec, maxLabelValueLength int) *CounterVec {
 	return &CounterVec{
 		vec:   cv,
-		store: newHeldMetricsStore[prometheus.Counter](),
+		store: newHeldMetricsStore[prometheus.Counter](maxLabelValueLength),
 	}
 }
 

metric/gauge.go

@@ -52,10 +52,10 @@ type GaugeVec struct {
 	vec   *prometheus.GaugeVec
 }
 
-func newGaugeVec(gv *prometheus.GaugeVec) *GaugeVec {
+func newGaugeVec(gv *prometheus.GaugeVec, maxLabelValueLength int) *GaugeVec {
 	return &GaugeVec{
 		vec:   gv,
-		store: newHeldMetricsStore[prometheus.Gauge](),
+		store: newHeldMetricsStore[prometheus.Gauge](maxLabelValueLength),
 	}
 }
 

metric/histogram.go

@@ -26,10 +26,10 @@ type HistogramVec struct {
 	vec   *prometheus.HistogramVec
 }
 
-func newHistogramVec(hv *prometheus.HistogramVec) *HistogramVec {
+func newHistogramVec(hv *prometheus.HistogramVec, maxLabelValueLength int) *HistogramVec {
 	return &HistogramVec{
 		vec:   hv,
-		store: newHeldMetricsStore[prometheus.Histogram](),
+		store: newHeldMetricsStore[prometheus.Histogram](maxLabelValueLength),
 	}
 }
 

metric/metric.go

@@ -41,18 +41,21 @@ func (h *heldMetric[T]) updateUsage() {
 }
 
 type heldMetricsStore[T prometheus.Metric] struct {
-	mu            sync.RWMutex
-	metricsByHash map[uint64][]*heldMetric[T]
+	mu                        sync.RWMutex
+	metricsByHash             map[uint64][]*heldMetric[T]
+	metricMaxLabelValueLength int
 }
 
-func newHeldMetricsStore[T prometheus.Metric]() *heldMetricsStore[T] {
+func newHeldMetricsStore[T prometheus.Metric](metricMaxLabelValueLength int) *heldMetricsStore[T] {
 	return &heldMetricsStore[T]{
-		mu:            sync.RWMutex{},
-		metricsByHash: make(map[uint64][]*heldMetric[T]),
+		mu:                        sync.RWMutex{},
+		metricsByHash:             make(map[uint64][]*heldMetric[T]),
+		metricMaxLabelValueLength: metricMaxLabelValueLength,
 	}
 }
 
 func (h *heldMetricsStore[T]) GetOrCreate(labels []string, newPromMetric func(...string) T) *heldMetric[T] {
+	h.truncateLabels(labels)
 	hash := computeStringsHash(labels)
 	// fast path - metric exists
 	h.mu.RLock()
@@ -66,6 +69,7 @@ func (h *heldMetricsStore[T]) GetOrCreate(labels []string, newPromMetric func(..
 }
 
 func (h *heldMetricsStore[T]) Delete(labels []string, deleter metricDeleter) bool {
+	h.truncateLabels(labels)
 	hash := computeStringsHash(labels)
 
 	h.mu.Lock()
@@ -159,6 +163,18 @@ func (h *heldMetricsStore[T]) DeleteOldMetrics(holdDuration time.Duration, delet
 	}
 }
 
+func (h *heldMetricsStore[T]) truncateLabels(lvs []string) {
+	if h.metricMaxLabelValueLength == 0 {
+		return
+	}
+
+	for i, label := range lvs {
+		if len(label) > h.metricMaxLabelValueLength {
+			lvs[i] = label[:h.metricMaxLabelValueLength]
+		}
+	}
+}
+
 func findHeldMetricIndex[T prometheus.Metric](hMetrics []*heldMetric[T], labels []string) int {
 	idx := -1
 	for i := range hMetrics {

metric/metric_test.go

@@ -14,7 +14,7 @@ import (
 func TestLabelExpiration(t *testing.T) {
 	r := require.New(t)
 
-	ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute)
+	ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute, 0)
 	c := ctl.RegisterCounterVec("errors", "", "level")
 
 	now := time.Now().UnixNano()
@@ -53,7 +53,7 @@ func TestUnsafeStringInMetric(t *testing.T) {
 	bytes := []byte("hello world")
 	unsafeString := unsafe.String(unsafe.SliceData(bytes), len(bytes))
 
-	store := newHeldMetricsStore[prometheus.Counter]()
+	store := newHeldMetricsStore[prometheus.Counter](0)
 
 	labels := []string{unsafeString}
 	m := store.GetOrCreate([]string{unsafeString}, func(s ...string) prometheus.Counter {
@@ -66,6 +66,21 @@ func TestUnsafeStringInMetric(t *testing.T) {
 	r.Equal([]string{"hello world"}, m.labels)
 }
 
+func TestLabelTruncation(t *testing.T) {
+	r := require.New(t)
+
+	maxLabelValueLength := 10
+	label := "some_long_label_value"
+	ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute, maxLabelValueLength)
+	c := ctl.RegisterCounterVec("errors", "", "level")
+
+	c.WithLabelValues(label).Inc()
+	r.Equal(float64(1), c.WithLabelValues("some_long_").ToFloat64())
+
+	c.DeleteLabelValues(label)
+	r.Equal(float64(0), c.WithLabelValues("some_long_").ToFloat64())
+}
+
 var holderBenchCases = []struct {
 	Labels      []string
 	LabelValues [][]string
@@ -98,7 +113,7 @@ var holderBenchCases = []struct {
 
 func BenchmarkMetricHolder(b *testing.B) {
 	for _, benchCase := range holderBenchCases {
-		ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute)
+		ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute, 0)
 
 		counter := ctl.RegisterCounterVec("test_name", "", benchCase.Labels...)
 		ctl.AddToHolder(counter)
@@ -117,7 +132,7 @@ func BenchmarkMetricHolder(b *testing.B) {
 
 func BenchmarkPromVec(b *testing.B) {
 	for _, benchCase := range holderBenchCases {
-		ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute)
+		ctl := NewCtl("test", prometheus.NewRegistry(), time.Minute, 0)
 		counter := ctl.RegisterCounterVec("test_name", "", benchCase.Labels...)
 		name := strings.Join(benchCase.Labels, "_")
 

pipeline/README.idoc.md

@@ -108,6 +108,8 @@ Whether to fatal on decoding error.
 
 The amount of time the metric can be idle until it is deleted. Used for deleting rarely updated metrics to save metrics storage resources. The value must be passed in format of duration (`<number>(ms|s|m|h)`).
 
+> ⚠ DEPRECATED. Use `hold_duration` in `metrics` section instead.
+
 <br>
 
 **`pool`** *`string`* *`options=std|low_memory`*
@@ -131,6 +133,31 @@ Default pool is `low_memory`.
 
 <br>
 
+## Metrics
+
+Section for metrics in settings. Example:
+
+```yaml
+pipelines:
+  test:
+    settings:
+      metrics:
+        hold_duration: 1h
+        max_label_value_length: 100
+```
+
+**`hold_duration`** *`string`* *`default=30m`*
+
+The amount of time the metric can be idle until it is deleted. Used for deleting rarely updated metrics to save metrics storage resources. The value must be passed in format of duration (`<number>(ms|s|m|h)`).
+
+<br>
+
+**`max_label_value_length`** *`int`* *`default=0`*
+
+Maximum length of custom metric labels in action plugins. If zero, no limit is set.
+
+<br>
+
 ## Datetime parse formats
 
 Most of the plugins which work with parsing datetime call `pipeline.ParseTime` function. It accepts datetime layouts the same way as Go [time.Parse](https://pkg.go.dev/time#Parse) (in format of datetime like `2006-01-02T15:04:05.999999999Z07:00`) except unix timestamp formats, they can only be specified via aliases.

pipeline/README.md

@@ -108,6 +108,8 @@ Whether to fatal on decoding error.
 
 The amount of time the metric can be idle until it is deleted. Used for deleting rarely updated metrics to save metrics storage resources. The value must be passed in format of duration (`<number>(ms|s|m|h)`).
 
+> ⚠ DEPRECATED. Use `hold_duration` in `metrics` section instead.
+
 <br>
 
 **`pool`** *`string`* *`options=std|low_memory`*
@@ -131,6 +133,31 @@ Default pool is `low_memory`.
 
 <br>
 
+## Metrics
+
+Section for metrics in settings. Example:
+
+```yaml
+pipelines:
+  test:
+    settings:
+      metrics:
+        hold_duration: 1h
+        max_label_value_length: 100
+```
+
+**`hold_duration`** *`string`* *`default=30m`*
+
+The amount of time the metric can be idle until it is deleted. Used for deleting rarely updated metrics to save metrics storage resources. The value must be passed in format of duration (`<number>(ms|s|m|h)`).
+
+<br>
+
+**`max_label_value_length`** *`int`* *`default=0`*
+
+Maximum length of custom metric labels in action plugins. If zero, no limit is set.
+
+<br>
+
 ## Datetime parse formats
 
 Most of the plugins which work with parsing datetime call `pipeline.ParseTime` function. It accepts datetime layouts the same way as Go [time.Parse](https://pkg.go.dev/time#Parse) (in format of datetime like `2006-01-02T15:04:05.999999999Z07:00`) except unix timestamp formats, they can only be specified via aliases.

pipeline/antispam/antispammer_test.go

@@ -17,7 +17,7 @@ func newAntispammer(threshold, unbanIterations int, maintenanceInterval time.Dur
 		Threshold:           threshold,
 		UnbanIterations:     unbanIterations,
 		Logger:              logger.Instance.Named("antispam").Desugar(),
-		MetricsController:   metric.NewCtl("test", prometheus.NewRegistry(), time.Minute),
+		MetricsController:   metric.NewCtl("test", prometheus.NewRegistry(), time.Minute, 0),
 	})
 }
 

pipeline/backoff_test.go

@@ -24,7 +24,7 @@ func TestBackoff(t *testing.T) {
 
 	batcherBackoff := NewRetriableBatcher(
 		&BatcherOptions{
-			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute),
+			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute, 0),
 		},
 		func(workerData *WorkerData, batch *Batch) error {
 			eventCount.Inc()
@@ -54,7 +54,7 @@ func TestBackoffWithError(t *testing.T) {
 
 	batcherBackoff := NewRetriableBatcher(
 		&BatcherOptions{
-			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute),
+			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute, 0),
 		},
 		func(workerData *WorkerData, batch *Batch) error {
 			return errors.New("some error")
@@ -84,7 +84,7 @@ func TestBackoffWithErrorWithDeadQueue(t *testing.T) {
 
 	batcherBackoff := NewRetriableBatcher(
 		&BatcherOptions{
-			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute),
+			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute, 0),
 		},
 		func(workerData *WorkerData, batch *Batch) error {
 			return errors.New("some error")