Repository: atlassian/gostatsd
Pull request #590 (2023-01-31 05:07:56)

Allow parallel metrics merging with HTTP forwarder mode
In the forwarder mode, metrics merging is a blocking operation which may introduce significant back pressure to the UDP consumption if there is large load. This patch adds a configuration `concurrent-merge` to allow for parallel processing of the merging operation. The default value is set to 1 to maintain backward-compatibility since it aggravates the out-of-orderness of the forwarding metrics.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 37cca8b..eab6331 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,8 @@
+35.1.3
+------
+- HTTP forwarder updated with a new option `concurrent-merge` to support parallel processing, the
+  default value is backward-compatible.
+
 35.1.2
 ------
 - Bump all the libraries.  No user visible impact expected.
diff --git a/README.md b/README.md
index f4fcb4f..33409aa 100644
--- a/README.md
+++ b/README.md
@@ -126,6 +126,7 @@ following configuration options:
 - `api-endpoint`: configures the endpoint to submit raw metrics to.  This setting should be just a base URL, for example
   `https://statsd-aggregator.private`, with no path.  Required, no default
 - `max-requests`: maximum number of requests in flight.  Defaults to `1000` (which is probably too high)
+- `concurrent-merge`: maximum number of concurrent goroutines allowed to merge metrics before forwarding.  Defaults to `1` for backward-compatibility
 - `max-request-elapsed-time`: duration for the maximum amount of time to try submitting data before giving up.  This
   includes retries.  Defaults to `30s` (which is probably too high). Setting this value to `-1` will disable retries.
 - `consolidator-slots`: number of slots in the metric consolidator.  Memory usage is a function of this.  Lower values
diff --git a/pkg/statsd/handler_http_forwarder_v2.go b/pkg/statsd/handler_http_forwarder_v2.go
index ab66298..6a16838 100644
--- a/pkg/statsd/handler_http_forwarder_v2.go
+++ b/pkg/statsd/handler_http_forwarder_v2.go
@@ -33,6 +33,7 @@ const (
 	defaultApiEndpoint               = ""
 	defaultMaxRequestElapsedTime     = 30 * time.Second
 	defaultMaxRequests               = 1000
+	defaultConcurrentMerge           = 1
 	defaultTransport                 = "default"
 )
 
@@ -49,6 +50,7 @@ type HttpForwarderHandlerV2 struct {
 	apiEndpoint           string
 	maxRequestElapsedTime time.Duration
 	metricsSem            chan struct{}
+	metricsMergingSem     chan struct{}
 	client                *http.Client
 	eventWg               sync.WaitGroup
 	compress              bool
@@ -70,6 +72,7 @@ func NewHttpForwarderHandlerV2FromViper(logger logrus.FieldLogger, v *viper.Vipe
 	subViper.SetDefault("max-request-elapsed-time", defaultMaxRequestElapsedTime)
 	subViper.SetDefault("consolidator-slots", v.GetInt(gostatsd.ParamMaxParsers))
 	subViper.SetDefault("flush-interval", defaultConsolidatorFlushInterval)
+	subViper.SetDefault("concurrent-merge", defaultConcurrentMerge)
 
 	return NewHttpForwarderHandlerV2(
 		logger,
@@ -77,6 +80,7 @@ func NewHttpForwarderHandlerV2FromViper(logger logrus.FieldLogger, v *viper.Vipe
 		subViper.GetString("api-endpoint"),
 		subViper.GetInt("consolidator-slots"),
 		subViper.GetInt("max-requests"),
+		subViper.GetInt("concurrent-merge"),
 		subViper.GetBool("compress"),
 		subViper.GetDuration("max-request-elapsed-time"),
 		subViper.GetDuration("flush-interval"),
@@ -93,6 +97,7 @@ func NewHttpForwarderHandlerV2(
 	apiEndpoint string,
 	consolidatorSlots,
 	maxRequests int,
+	concurrentMerge int,
 	compress bool,
 	maxRequestElapsedTime time.Duration,
 	flushInterval time.Duration,
@@ -115,6 +120,9 @@ func NewHttpForwarderHandlerV2(
 	if flushInterval <= 0 {
 		return nil, fmt.Errorf("flush-interval must be positive")
 	}
+	if concurrentMerge <= 0 {
+		return nil, fmt.Errorf("concurrent-merge must be positive")
+	}
 
 	httpClient, err := pool.Get(transport)
 	if err != nil {
@@ -162,6 +170,11 @@ func NewHttpForwarderHandlerV2(
 		metricsSem <- struct{}{}
 	}
 
+	metricsMergingSem := make(chan struct{}, concurrentMerge)
+	for i := 0; i < concurrentMerge; i++ {
+		metricsMergingSem <- struct{}{}
+	}
+
 	ch := make(chan []*gostatsd.MetricMap)
 
 	return &HttpForwarderHandlerV2{
@@ -169,6 +182,7 @@ func NewHttpForwarderHandlerV2(
 		apiEndpoint:           apiEndpoint,
 		maxRequestElapsedTime: maxRequestElapsedTime,
 		metricsSem:            metricsSem,
+		metricsMergingSem:     metricsMergingSem,
 		compress:              compress,
 		consolidator:          gostatsd.NewMetricConsolidator(consolidatorSlots, false, flushInterval, ch),
 		consolidatedMetrics:   ch,
@@ -229,23 +243,32 @@ func (hfh *HttpForwarderHandlerV2) Run(ctx context.Context) {
 	var wg wait.Group
 	wg.Start(func() {
 		for metricMaps := range hfh.consolidatedMetrics {
-			mergedMetricMap := gostatsd.MergeMaps(metricMaps)
-			mms := mergedMetricMap.SplitByTags(hfh.dynHeaderNames)
-			for dynHeaderTags, mm := range mms {
-				if mm.IsEmpty() {
-					continue
+			hfh.acquireMergingSem()
+			metricMaps := metricMaps
+			go func() {
+				mergedMetricMap := gostatsd.MergeMaps(metricMaps)
+				mms := mergedMetricMap.SplitByTags(hfh.dynHeaderNames)
+				hfh.releaseMergingSem()
+
+				for dynHeaderTags, mm := range mms {
+					if mm.IsEmpty() {
+						continue
+					}
+					hfh.acquireSem()
+					postId := atomic.AddUint64(&hfh.postId, 1) - 1
+					go func(postId uint64, metricMap *gostatsd.MetricMap, dynHeaderTags string) {
+						hfh.postMetrics(context.Background(), metricMap, dynHeaderTags, postId)
+						hfh.releaseSem()
+					}(postId, mm, dynHeaderTags)
 				}
-				hfh.acquireSem()
-				postId := atomic.AddUint64(&hfh.postId, 1) - 1
-				go func(postId uint64, metricMap *gostatsd.MetricMap, dynHeaderTags string) {
-					hfh.postMetrics(context.Background(), metricMap, dynHeaderTags, postId)
-					hfh.releaseSem()
-				}(postId, mm, dynHeaderTags)
-			}
+			}()
 		}
 		for i := 0; i < cap(hfh.metricsSem); i++ {
 			hfh.acquireSem()
 		}
+		for i := 0; i < cap(hfh.metricsMergingSem); i++ {
+			hfh.acquireMergingSem()
+		}
 	})
 	wg.StartWithContext(ctx, func(ctx context.Context) {
 		hfh.consolidator.Run(ctx)
@@ -270,6 +293,14 @@ func (hfh *HttpForwarderHandlerV2) releaseSem() {
 	hfh.metricsSem <- struct{}{} // will never block
 }
 
+func (hfh *HttpForwarderHandlerV2) acquireMergingSem() {
+	<-hfh.metricsMergingSem
+}
+
+func (hfh *HttpForwarderHandlerV2) releaseMergingSem() {
+	hfh.metricsMergingSem <- struct{}{}
+}
+
 func translateToProtobufV2(metricMap *gostatsd.MetricMap) *pb.RawMessageV2 {
 	var pbMetricMap pb.RawMessageV2
 


Test patch:
diff --git a/pkg/statsd/handler_http_forwarder_v2_test.go b/pkg/statsd/handler_http_forwarder_v2_test.go
index 1d7ebf4..1f406c4 100644
--- a/pkg/statsd/handler_http_forwarder_v2_test.go
+++ b/pkg/statsd/handler_http_forwarder_v2_test.go
@@ -251,6 +251,7 @@ func TestForwardingData(t *testing.T) {
 		s.URL,
 		1,
 		1,
+		1,
 		false,
 		100*time.Millisecond, // maxRequestElapsedTime
 		100*time.Millisecond, // flushInterval
@@ -312,7 +313,7 @@ func TestHttpForwarderV2New(t *testing.T) {
 			expected:   []string{"service:", "deploy:"},
 		},
 	} {
-		h, err := NewHttpForwarderHandlerV2(logger, "default", "endpoint", 1, 1, false, time.Second, time.Second,
+		h, err := NewHttpForwarderHandlerV2(logger, "default", "endpoint", 1, 1, 1, false, time.Second, time.Second,
 			cusHeaders, testcase.dynHeaders, pool)
 		require.Nil(t, err)
 		require.Equal(t, h.dynHeaderNames, testcase.expected)
diff --git a/pkg/web/http_receiver_v2_test.go b/pkg/web/http_receiver_v2_test.go
index 636dcdb..673571c 100644
--- a/pkg/web/http_receiver_v2_test.go
+++ b/pkg/web/http_receiver_v2_test.go
@@ -54,6 +54,7 @@ func TestForwardingEndToEndV2(t *testing.T) {
 		c.URL,
 		5, // deliberately prime, so the loop below doesn't send the same thing to the same MetricMap every time.
 		10,
+		1,
 		false,
 		10*time.Second,
 		10*time.Millisecond,