Merge pull request #720 from atlassian/htan/make-async-send-async

send batches concurrently in each flush and compress payload

go.mod

@@ -32,6 +32,8 @@ require (
  stathat.com/c/consistent v1.0.0
 )
 
+require golang.org/x/sync v0.6.0
+
 require (
  github.com/PuerkitoBio/purell v1.1.1 // indirect
  github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect

go.sum

@@ -403,6 +403,8 @@ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
+golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

pkg/backends/otlp/backend.go

@@ -13,6 +13,7 @@ import (
  "github.com/sirupsen/logrus"
  "github.com/spf13/viper"
  "go.uber.org/multierr"
+ "golang.org/x/sync/errgroup"
 
  "github.com/atlassian/gostatsd/pkg/stats"
 
@@ -50,6 +51,7 @@ type Backend struct {
  client *http.Client
  requestsBufferSem chan struct{}
  maxRetries int
+ CompressPayload bool
 
  // metricsPerBatch is the maximum number of metrics to send in a single batch.
  metricsPerBatch int
@@ -84,6 +86,7 @@ func NewClientFromViper(v *viper.Viper, logger logrus.FieldLogger, pool *transpo
  logger: logger,
  requestsBufferSem: make(chan struct{}, cfg.MaxRequests),
  maxRetries: cfg.MaxRetries,
+ CompressPayload: cfg.CompressPayload,
  metricsPerBatch: cfg.MetricsPerBatch,
  }, nil
 }
@@ -139,7 +142,7 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  bd.batchesRetried.SendIfChanged(statser, "backend.retried", nil)
  }()
 
- group := newGroups(bd.metricsPerBatch)
+ currentGroup := newGroups(bd.metricsPerBatch)
 
  mm.Counters.Each(func(name, _ string, cm gostatsd.Counter) {
  resources, attributes := data.SplitMetricTagsByKeysAndConvert(cm.Tags, bd.resourceKeys)
@@ -164,8 +167,8 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  ),
  )
 
- group.insert(bd.is, resources, rate)
- group.insert(bd.is, resources, m)
+ currentGroup.insert(bd.is, resources, rate)
+ currentGroup.insert(bd.is, resources, m)
  })
 
  mm.Gauges.Each(func(name, _ string, gm gostatsd.Gauge) {
@@ -181,7 +184,7 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  ),
  )
 
- group.insert(bd.is, resources, m)
+ currentGroup.insert(bd.is, resources, m)
  })
 
  mm.Sets.Each(func(name, _ string, sm gostatsd.Set) {
@@ -197,7 +200,7 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  ),
  )
 
- group.insert(bd.is, resources, m)
+ currentGroup.insert(bd.is, resources, m)
  })
 
  mm.Timers.Each(func(name, _ string, t gostatsd.Timer) {
@@ -214,7 +217,7 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  } else {
  btags.Insert("le", strconv.FormatFloat(float64(boundry), 'f', -1, 64))
  }
- group.insert(
+ currentGroup.insert(
  bd.is,
  resources,
  data.NewMetric(fmt.Sprintf("%s.histogram", name)).SetGauge(
@@ -248,7 +251,7 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  if calc.discarded {
  continue
  }
- group.insert(
+ currentGroup.insert(
  bd.is,
  resources,
  data.NewMetric(fmt.Sprintf("%s.%s", name, calc.suffix)).SetGauge(
@@ -262,7 +265,7 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  }
 
  for _, pct := range t.Percentiles {
- group.insert(bd.is, resources, data.NewMetric(fmt.Sprintf("%s.%s", name, pct.Str)).SetGauge(
+ currentGroup.insert(bd.is, resources, data.NewMetric(fmt.Sprintf("%s.%s", name, pct.Str)).SetGauge(
  data.NewGauge(data.NewNumberDataPoint(
  uint64(t.Timestamp),
  data.WithNumberDataPointMap(attributes),
@@ -280,27 +283,32 @@ func (bd *Backend) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap,
  if len(t.Histogram) != 0 {
  opts = append(opts, data.WithHistogramDataPointCumulativeBucketValues(t.Histogram))
  }
- group.insert(bd.is, resources, data.NewMetric(name).SetHistogram(
+ currentGroup.insert(bd.is, resources, data.NewMetric(name).SetHistogram(
  data.NewHistogram(data.NewHistogramDataPoint(uint64(t.Timestamp), opts...)),
  ))
  }
  })
 
- var errs error
- for _, b := range group.batches {
+ eg, ectx := errgroup.WithContext(ctx)
+ for _, b := range currentGroup.batches {
  atomic.AddUint64(&bd.batchesCreated, 1)
- err := bd.postMetrics(ctx, b)
- if err != nil {
- bd.logger.WithError(err).WithFields(logrus.Fields{
- "endpoint": bd.metricsEndpoint,
- }).Error("Issues trying to submit data")
- errs = multierr.Append(errs, err)
- } else {
- atomic.AddUint64(&bd.batchesSent, 1)
- atomic.AddUint64(&bd.seriesSent, uint64(b.lenMetrics()))
- }
+ func(g group) {
+ eg.Go(func() error {
+ err := bd.postMetrics(ectx, b)
+ if err != nil {
+ bd.logger.WithError(err).WithFields(logrus.Fields{
+ "endpoint": bd.metricsEndpoint,
+ }).Error("Issues trying to submit data")
+ atomic.AddUint64(&bd.batchesDropped, 1)
+ } else {
+ atomic.AddUint64(&bd.batchesSent, 1)
+ atomic.AddUint64(&bd.seriesSent, uint64(b.lenMetrics()))
+ }
+ return err
+ })
+ }(b)
  }
- cb(multierr.Errors(errs))
+ cb(multierr.Errors(eg.Wait()))
 }
 
 func (c *Backend) postMetrics(ctx context.Context, batch group) error {
@@ -312,9 +320,8 @@ func (c *Backend) postMetrics(ctx context.Context, batch group) error {
  )
 
  resourceMetrics := batch.values()
- req, err = data.NewMetricsRequest(ctx, c.metricsEndpoint, resourceMetrics)
+ req, err = data.NewMetricsRequest(ctx, c.metricsEndpoint, resourceMetrics, c.CompressPayload)
  if err != nil {
- atomic.AddUint64(&c.batchesDropped, 1)
  return err
  }
 

pkg/backends/otlp/backend_test.go

@@ -362,6 +362,7 @@ func TestBackendSendAsyncMetrics(t *testing.T) {
  v := viper.New()
  v.Set("otlp.metrics_endpoint", fmt.Sprintf("%s/%s", s.URL, "v1/metrics"))
  v.Set("otlp.logs_endpoint", fmt.Sprintf("%s/%s", s.URL, "v1/logs"))
+ v.Set("otlp.compress_payload", false)
  if tc.enableHistograms {
  v.Set("otlp.conversion", ConversionAsHistogram)
  }

pkg/backends/otlp/config.go

@@ -29,6 +29,8 @@ type Config struct {
  MaxRequests int `mapstructure:"max_requests"`
  // MaxRetries (Optional, default: 3) is the maximum number of retries to send a batch
  MaxRetries int `mapstructure:"max_retries"`
+ // CompressPayload (Optional, default: true) is used to enable payload compression
+ CompressPayload bool `mapstructure:"compress_payload"`
  // MetricsPerBatch (Optional, default: 1000) is the maximum number of metrics to send in a single batch.
  MetricsPerBatch int `mapstructure:"metrics_per_batch"`
  // ResourceKeys (Optional) is used to extract values from provided tags
@@ -62,6 +64,7 @@ func newDefaultConfig() *Config {
  Transport: "default",
  MaxRequests: runtime.NumCPU() * 2,
  MaxRetries: 3,
+ CompressPayload: true,
  MetricsPerBatch: defaultMetricsPerBatch,
  Conversion: ConversionAsGauge,
  UserAgent: "gostatsd",

pkg/backends/otlp/config_test.go

@@ -30,6 +30,7 @@ func TestNewConfig(t *testing.T) {
  v.SetDefault("otlp.logs_endpoint", "http://local/v1/logs")
  v.SetDefault("otlp.max_requests", 1)
  v.SetDefault("otlp.max_retries", 3)
+ v.SetDefault("otlp.compress_payload", true)
  v.SetDefault("otlp.metrics_per_batch", 999)
  return v
  }(),
@@ -38,6 +39,7 @@ func TestNewConfig(t *testing.T) {
  LogsEndpoint: "http://local/v1/logs",
  MaxRequests: 1,
  MaxRetries: 3,
+ CompressPayload: true,
  MetricsPerBatch: 999,
  Conversion: "AsGauge",
  Transport: "default",

pkg/backends/otlp/internal/data/metric_request.go

@@ -1,6 +1,8 @@
 package data
 
 import (
+ "bytes"
+ "compress/gzip"
  "context"
  "net/http"
 
@@ -13,7 +15,7 @@ type metricsRequest struct {
  raw *v1export.ExportMetricsServiceRequest
 }
 
-func NewMetricsRequest(ctx context.Context, endpoint string, metrics []ResourceMetrics) (*http.Request, error) {
+func NewMetricsRequest(ctx context.Context, endpoint string, metrics []ResourceMetrics, compressPayload bool) (*http.Request, error) {
  mr := metricsRequest{
  raw: &v1export.ExportMetricsServiceRequest{
  ResourceMetrics: make([]*v1metrics.ResourceMetrics, 0, len(metrics)),
@@ -29,5 +31,19 @@ func NewMetricsRequest(ctx context.Context, endpoint string, metrics []ResourceM
  return nil, err
  }
 
+ if compressPayload {
+ var b bytes.Buffer
+ w := gzip.NewWriter(&b)
+ if _, err = w.Write(buf); err != nil {
+ return nil, err
+ }
+
+ if err = w.Close(); err != nil {
+ return nil, err
+ }
+
+ return createProtobufRequest(ctx, endpoint, b.Bytes(), withHeader("Content-Encoding", "gzip"))
+ }
+
  return createProtobufRequest(ctx, endpoint, buf)
 }

pkg/backends/otlp/internal/data/metric_request_test.go

@@ -15,6 +15,7 @@ func TestMetricsRequest(t *testing.T) {
  context.Background(),
  "not-a-valid-url",
  []ResourceMetrics{NewResourceMetrics(NewResource())},
+ false,
  )
  assert.NoError(t, err, "Must not error creating request")
  assert.NotNil(t, req, "Must have a valid request")

pkg/backends/otlp/internal/data/request.go

@@ -6,7 +6,13 @@ import (
  "net/http"
 )
 
-func createProtobufRequest(ctx context.Context, endpoint string, buf []byte) (*http.Request, error) {
+func withHeader(key, value string) func(*http.Request) {
+ return func(req *http.Request) {
+ req.Header.Set(key, value)
+ }
+}
+
+func createProtobufRequest(ctx context.Context, endpoint string, buf []byte, option ...func(*http.Request)) (*http.Request, error) {
  req, err := http.NewRequestWithContext(
  ctx,
  http.MethodPost,
@@ -18,5 +24,8 @@ func createProtobufRequest(ctx context.Context, endpoint string, buf []byte) (*h
  }
 
  req.Header.Set("Content-Type", RequestContentTypeProtobuf)
+ for _, opt := range option {
+ opt(req)
+ }
  return req, nil
 }