Performance improvement - Introduce queue for raw statsd payloads

bridge_test.go

@@ -20,6 +20,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/statsd_exporter/pkg/parser"
+	"github.com/stretchr/testify/require"
+
 	"github.com/go-kit/log"
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
@@ -531,67 +534,77 @@ func TestHandlePacket(t *testing.T) {
 		},
 	}
 
-	parser := line.NewParser()
-	parser.EnableDogstatsdParsing()
-	parser.EnableInfluxdbParsing()
-	parser.EnableLibratoParsing()
-	parser.EnableSignalFXParsing()
+	testParser := line.NewParser()
+	testParser.EnableDogstatsdParsing()
+	testParser.EnableInfluxdbParsing()
+	testParser.EnableLibratoParsing()
+	testParser.EnableSignalFXParsing()
 
 	for k, l := range []statsDPacketHandler{&listener.StatsDUDPListener{
-		Conn:            nil,
-		EventHandler:    nil,
-		Logger:          log.NewNopLogger(),
-		LineParser:      parser,
-		UDPPackets:      udpPackets,
-		LinesReceived:   linesReceived,
-		EventsFlushed:   eventsFlushed,
-		SampleErrors:    *sampleErrors,
-		SamplesReceived: samplesReceived,
-		TagErrors:       tagErrors,
-		TagsReceived:    tagsReceived,
+		Conn:       nil,
+		Logger:     log.NewNopLogger(),
+		UDPPackets: udpPackets,
 	}, &mockStatsDTCPListener{listener.StatsDTCPListener{
-		Conn:            nil,
-		EventHandler:    nil,
-		Logger:          log.NewNopLogger(),
-		LineParser:      parser,
-		LinesReceived:   linesReceived,
-		EventsFlushed:   eventsFlushed,
-		SampleErrors:    *sampleErrors,
-		SamplesReceived: samplesReceived,
-		TagErrors:       tagErrors,
-		TagsReceived:    tagsReceived,
-		TCPConnections:  tcpConnections,
-		TCPErrors:       tcpErrors,
-		TCPLineTooLong:  tcpLineTooLong,
+		Conn:           nil,
+		Logger:         log.NewNopLogger(),
+		TCPConnections: tcpConnections,
+		TCPErrors:      tcpErrors,
+		TCPLineTooLong: tcpLineTooLong,
 	}, log.NewNopLogger()}} {
+		packets := make(chan string, 32)
 		events := make(chan event.Events, 32)
-		l.SetEventHandler(&event.UnbufferedEventHandler{C: events})
+		workers := make([]*parser.Worker, 1)
+		for i := range workers {
+			workers[i] = parser.NewWorker(
+				log.NewNopLogger(),
+				&event.UnbufferedEventHandler{C: events},
+				testParser,
+				nil,
+				linesReceived,
+				*sampleErrors,
+				samplesReceived,
+				tagErrors,
+				tagsReceived,
+			)
+		}
+		go workers[0].Consume(packets)
+		l.SetPacketBuffer(packets)
 		for i, scenario := range scenarios {
-			l.HandlePacket([]byte(scenario.in))
-
-			le := len(events)
-			// Flatten actual events.
-			actual := event.Events{}
-			for j := 0; j < le; j++ {
-				actual = append(actual, <-events...)
-			}
-
-			if len(actual) != len(scenario.out) {
-				t.Fatalf("%d.%d. Expected %d events, got %d in scenario '%s'", k, i, len(scenario.out), len(actual), scenario.name)
-			}
-
-			for j, expected := range scenario.out {
-				if !reflect.DeepEqual(&expected, &actual[j]) {
-					t.Fatalf("%d.%d.%d. Expected %#v, got %#v in scenario '%s'", k, i, j, expected, actual[j], scenario.name)
+			t.Run(scenario.name, func(t *testing.T) {
+				l.HandlePacket([]byte(scenario.in))
+
+				le := 0
+				if len(scenario.out) != 0 {
+					// wait until workers produce events
+					require.Eventually(t, func() bool {
+						le = len(events)
+						return le > 0
+					}, time.Second, time.Millisecond*10)
+				}
+
+				// Flatten actual events.
+				actual := event.Events{}
+				for j := 0; j < le; j++ {
+					actual = append(actual, <-events...)
+				}
+
+				if len(actual) != len(scenario.out) {
+					t.Fatalf("%d.%d. Expected %d events, got %d in scenario '%s'", k, i, len(scenario.out), len(actual), scenario.name)
+				}
+
+				for j, expected := range scenario.out {
+					if !reflect.DeepEqual(&expected, &actual[j]) {
+						t.Fatalf("%d.%d.%d. Expected %#v, got %#v in scenario '%s'", k, i, j, expected, actual[j], scenario.name)
+					}
 				}
-			}
+			})
 		}
 	}
 }
 
 type statsDPacketHandler interface {
 	HandlePacket(packet []byte)
-	SetEventHandler(eh event.EventHandler)
+	SetPacketBuffer(pb chan string)
 }
 
 type mockStatsDTCPListener struct {

exporter_benchmark_test.go

@@ -17,6 +17,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/prometheus/statsd_exporter/pkg/parser"
+
 	"github.com/go-kit/log"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -50,11 +52,11 @@ func benchmarkUDPListener(times int, b *testing.B) {
 		}
 	}
 
-	parser := line.NewParser()
-	parser.EnableDogstatsdParsing()
-	parser.EnableInfluxdbParsing()
-	parser.EnableLibratoParsing()
-	parser.EnableSignalFXParsing()
+	testParser := line.NewParser()
+	testParser.EnableDogstatsdParsing()
+	testParser.EnableInfluxdbParsing()
+	testParser.EnableLibratoParsing()
+	testParser.EnableSignalFXParsing()
 
 	// reset benchmark timer to not measure startup costs
 	b.ResetTimer()
@@ -65,25 +67,39 @@ func benchmarkUDPListener(times int, b *testing.B) {
 
 		// there are more events than input lines, need bigger buffer
 		events := make(chan event.Events, len(bytesInput)*times*2)
+		packets := make(chan string, len(input)*times*2)
 
 		l := listener.StatsDUDPListener{
-			EventHandler:    &event.UnbufferedEventHandler{C: events},
-			Logger:          logger,
-			LineParser:      parser,
-			UDPPackets:      udpPackets,
-			LinesReceived:   linesReceived,
-			SamplesReceived: samplesReceived,
-			TagsReceived:    tagsReceived,
+			Logger:       logger,
+			UDPPackets:   udpPackets,
+			PacketBuffer: packets,
+		}
+
+		workers := make([]*parser.Worker, 2)
+		for i := 0; i < len(workers); i++ {
+			workers[i] = parser.NewWorker(
+				logger,
+				&event.UnbufferedEventHandler{C: events},
+				testParser,
+				nil,
+				linesReceived,
+				*sampleErrors,
+				samplesReceived,
+				tagErrors,
+				tagsReceived,
+			)
+			go workers[i].Consume(packets)
 		}
 
 		// resume benchmark timer
 		b.StartTimer()
 
 		for i := 0; i < times; i++ {
-			for _, line := range bytesInput {
-				l.HandlePacket([]byte(line))
+			for _, li := range bytesInput {
+				l.HandlePacket([]byte(li))
 			}
 		}
+		close(packets)
 	}
 }
 

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/prometheus/client_golang v1.12.2
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.37.0
+	github.com/stretchr/testify v1.4.0
 	github.com/stvp/go-udp-testing v0.0.0-20201019212854-469649b16807
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
 	gopkg.in/yaml.v2 v2.4.0
@@ -18,9 +19,11 @@ require (
 	github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
 	golang.org/x/sys v0.0.0-20220708085239-5a0f0661e09d // indirect
 	google.golang.org/protobuf v1.28.0 // indirect