temporalio · cv65kr · Sep 15, 2022
@@ -15,6 +15,10 @@ const (
 
 	driverPrometheus string = "prometheus"
 	driverStatsd     string = "statsd"
+
+	// statsd format
+	statsdFormatBasic   string = "basic"
+	statsdFormatDataDog string = "datadog"
 )
 
 // ref:https://github.dev/temporalio/temporal/common/metrics/config.go:79
@@ -37,6 +41,7 @@ type Statsd struct {
 	// TagSeparator allows tags to be appended with a separator. If not specified tag keys and values
 	// are embedded to the stat name directly.
 	TagSeparator string `mapstructure:"tag_separator"`
+	FormatType   string `mapstructure:"format_type"`
 }
 
 type StatsdReporterConfig struct {
@@ -112,6 +117,10 @@ func (c *Config) InitDefault() error {
 				if c.Metrics.Statsd.HostPort == "" {
 					c.Metrics.Statsd.HostPort = "127.0.0.1:8125"
 				}
+
+				if c.Metrics.Statsd.FormatType == "" {
+					c.Metrics.Statsd.FormatType = statsdFormatBasic
+				}
 			}
 		}
 	}

@@ -3,6 +3,7 @@ module github.com/temporalio/roadrunner-temporal
 go 1.19
 
 require (
+	github.com/DataDog/datadog-go v4.8.3+incompatible
 	github.com/cactus/go-statsd-client/statsd v0.0.0-20200423205355-cb0885a1018c
 	github.com/goccy/go-json v0.9.11
 	github.com/golang/protobuf v1.5.2
@@ -20,6 +21,8 @@ require (
 	google.golang.org/protobuf v1.28.1
 )
 
+require github.com/Microsoft/go-winio v0.5.2 // indirect
+
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect

@@ -33,6 +33,10 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/DataDog/datadog-go v4.8.3+incompatible h1:fNGaYSuObuQb5nzeTQqowRAd9bpDIRRV4/gUtIBjh8Q=
+github.com/DataDog/datadog-go v4.8.3+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
+github.com/Microsoft/go-winio v0.5.2 h1:a9IhgEQBCUEk6QCdml9CiJGhAws+YwffDHEMp1VMrpA=
+github.com/Microsoft/go-winio v0.5.2/go.mod h1:WpS1mjBmmwHBEWmogvA2mj8546UReBk4v8QkMxJ6pZY=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -252,6 +256,7 @@ github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMT
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
+github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -427,6 +432,7 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

@@ -1,6 +1,7 @@
 package roadrunner_temporal //nolint:revive,stylecheck
 
 import (
+	"errors"
 	"io"
 	"time"
 
@@ -82,7 +83,7 @@ func newPrometheusScope(c prometheus.Configuration, prefix string, log *zap.Logg
 }
 
 // ref: https://github.dev/temporalio/temporal/common/metrics/config.go:391
-func newStatsdScope(statsdConfig *Statsd) (tally.Scope, io.Closer, error) {
+func newStatsdScope(statsdConfig *Statsd, log *zap.Logger) (tally.Scope, io.Closer, error) {
 	st, err := statsd.NewClientWithConfig(&statsd.ClientConfig{
 		Address:       statsdConfig.HostPort,
 		Prefix:        statsdConfig.Prefix,
@@ -101,7 +102,15 @@ func newStatsdScope(statsdConfig *Statsd) (tally.Scope, io.Closer, error) {
 		TagSeparator: statsdConfig.TagSeparator,
 	}
 
-	reporter := statsdreporter.NewReporter(st, opts)
+	var reporter tally.StatsReporter
+	if statsdConfig.FormatType == statsdFormatBasic {
+		reporter = statsdreporter.NewReporter(st, opts)
+	} else if statsdConfig.FormatType == statsdFormatDataDog {
+		reporter = NewDataDogMetricsReporter(statsdConfig, log)
+	} else {
+		return nil, nil, errors.New("statsd format type is not supported")
+	}
+
 	scopeOpts := tally.ScopeOptions{
 		Tags:     statsdConfig.Tags,
 		Prefix:   statsdConfig.Prefix,

@@ -0,0 +1,91 @@
+// Based on https://github.com/temporalio/temporal/pull/998#issuecomment-857884983
+package roadrunner_temporal
+
+import (
+	"fmt"
+	"go.uber.org/zap"
+	"sort"
+	"time"
+
+	"github.com/DataDog/datadog-go/statsd"
+	"github.com/uber-go/tally/v4"
+)
+
+type dogstatsdReporter struct {
+	dogstatsd *statsd.Client
+	log       *zap.Logger
+}
+
+func NewDataDogMetricsReporter(statsdConfig *Statsd, log *zap.Logger) tally.StatsReporter {
+	client, err := statsd.New(
+		statsdConfig.HostPort,
+		statsd.WithBufferFlushInterval(statsdConfig.FlushInterval),
+		statsd.WithMaxBytesPerPayload(statsdConfig.FlushBytes),
+	)
+	if err != nil {
+		log.Fatal("error creating dogstatsd client", zap.Error(err))
+	}
+	return &dogstatsdReporter{
+		dogstatsd: client,
+		log:       log,
+	}
+}
+
+func (r dogstatsdReporter) Capabilities() tally.Capabilities {
+	return r
+}
+
+func (r dogstatsdReporter) Reporting() bool {
+	return true
+}
+
+func (r dogstatsdReporter) Tagging() bool {
+	return true
+}
+
+func (r dogstatsdReporter) Flush() {
+	if err := r.dogstatsd.Flush(); err != nil {
+		r.log.Error("error while flushing", zap.Error(err))
+	}
+}
+
+func (r dogstatsdReporter) ReportCounter(name string, tags map[string]string, value int64) {
+	if err := r.dogstatsd.Count(name, value, r.marshalTags(tags), 1); err != nil {
+		r.log.Error("failed reporting counter", zap.Error(err))
+	}
+}
+
+func (r dogstatsdReporter) ReportGauge(name string, tags map[string]string, value float64) {
+	if err := r.dogstatsd.Gauge(name, value, r.marshalTags(tags), 1); err != nil {
+		r.log.Error("failed reporting gauge", zap.Error(err))
+	}
+}
+
+func (r dogstatsdReporter) ReportTimer(name string, tags map[string]string, interval time.Duration) {
+	if err := r.dogstatsd.Timing(name, interval, r.marshalTags(tags), 1); err != nil {
+		r.log.Error("failed reporting timer", zap.Error(err))
+	}
+}
+
+func (r dogstatsdReporter) ReportHistogramValueSamples(name string, tags map[string]string, buckets tally.Buckets, bucketLowerBound, bucketUpperBound float64, samples int64) {
+	r.log.Warn("unexpected call to ReportHistogramValueSamples")
+}
+
+func (r dogstatsdReporter) ReportHistogramDurationSamples(name string, tags map[string]string, buckets tally.Buckets, bucketLowerBound, bucketUpperBound time.Duration, samples int64) {
+	r.log.Warn("unexpected call to ReportHistogramDurationSamples")
+}
+
+func (r dogstatsdReporter) marshalTags(tags map[string]string) []string {
+	var keys []string
+	for k := range tags {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	var dogTags []string
+	for _, tk := range keys {
+		dogTags = append(dogTags, fmt.Sprintf("%s:%s", tk, tags[tk]))
+	}
+
+	return dogTags
+}
@@ -202,7 +202,7 @@ func (p *Plugin) Serve() chan error {
 			opts.MetricsHandler = tally.NewMetricsHandler(ms)
 			p.tallyCloser = cl
 		case driverStatsd:
-			ms, cl, errSt := newStatsdScope(p.config.Metrics.Statsd)
+			ms, cl, errSt := newStatsdScope(p.config.Metrics.Statsd, p.log)
 			if errSt != nil {
 				errCh <- errSt
 				return errCh