Fix tests

pipeline/pipeline_test.go

@@ -29,8 +29,9 @@ func TestInUnparsableMessages(t *testing.T) {
 	name := "invalid_json"
 	message := []byte("{wHo Is Json: YoU MeAn SoN oF JoHn???")
 	pipelineSettings := &pipeline.Settings{
-		Capacity: 5,
-		Decoder:  "json",
+		Capacity:           5,
+		Decoder:            "json",
+		MetricHoldDuration: pipeline.DefaultMetricHoldDuration,
 	}
 	offset := int64(666)
 	sourceID := pipeline.SourceID(3<<16 + int(10))
@@ -91,8 +92,9 @@ func TestInInvalidMessages(t *testing.T) {
 			name:    "empty_message",
 			message: []byte(""),
 			pipelineSettings: &pipeline.Settings{
-				Capacity: 5,
-				Decoder:  "json",
+				Capacity:           5,
+				Decoder:            "json",
+				MetricHoldDuration: pipeline.DefaultMetricHoldDuration,
 			},
 			offset:   int64(666),
 			sourceID: pipeline.SourceID(1<<16 + int(1)),
@@ -101,9 +103,10 @@ func TestInInvalidMessages(t *testing.T) {
 			name:    "too_long_message",
 			message: []byte("{\"value\":\"i'm longer than 1 byte\""),
 			pipelineSettings: &pipeline.Settings{
-				Capacity:     5,
-				Decoder:      "json",
-				MaxEventSize: 1,
+				Capacity:           5,
+				Decoder:            "json",
+				MaxEventSize:       1,
+				MetricHoldDuration: pipeline.DefaultMetricHoldDuration,
 			},
 			offset:   int64(666),
 			sourceID: pipeline.SourceID(2<<16 + int(3)),

pipeline/pipeline_whitebox_test.go

@@ -10,8 +10,9 @@ import (
 
 func TestPipeline_streamEvent(t *testing.T) {
 	settings := &Settings{
-		Capacity: 5,
-		Decoder:  "json",
+		Capacity:           5,
+		Decoder:            "json",
+		MetricHoldDuration: DefaultMetricHoldDuration,
 	}
 	p := New("test", settings, prometheus.NewRegistry())
 

plugin/output/file/helpers_test.go

@@ -54,6 +54,7 @@ func newPipeline(t *testing.T, configOutput *Config) *pipeline.Pipeline {
 		AvgEventSize:        2048,
 		StreamField:         "stream",
 		Decoder:             "json",
+		MetricHoldDuration:  pipeline.DefaultMetricHoldDuration,
 	}
 
 	p := pipeline.New("test_pipeline", settings, prometheus.NewRegistry())

plugin/output/s3/s3_test.go

@@ -393,10 +393,11 @@ func newPipeline(t *testing.T, configOutput *Config, objStoreF objStoreFactory)
 		Capacity:            4096,
 		MaintenanceInterval: time.Second * 10,
 		// MaintenanceInterval: time.Second * 100000,
-		AntispamThreshold: 0,
-		AvgEventSize:      2048,
-		StreamField:       "stream",
-		Decoder:           "json",
+		AntispamThreshold:  0,
+		AvgEventSize:       2048,
+		StreamField:        "stream",
+		Decoder:            "json",
+		MetricHoldDuration: pipeline.DefaultMetricHoldDuration,
 	}
 
 	p := pipeline.New("test_pipeline", settings, prometheus.NewRegistry())

test/test.go

@@ -122,6 +122,7 @@ func NewPipeline(actions []*pipeline.ActionPluginStaticInfo, pipelineOpts ...str
 		AvgEventSize:        2048,
 		StreamField:         "stream",
 		Decoder:             "json",
+		MetricHoldDuration:  pipeline.DefaultMetricHoldDuration,
 	}
 
 	pName := "test_pipeline"