[FAB-5903]Add statsd reporter and test

This patch add a statsd reporter which support tagging.
The current implementation add tag pairs to metrics
name. The test use a mock statsd udp server to receive
the metrics.

Change-Id: I80b4c1e8bda65c65c1b36989e5366e19bb745e8f
Signed-off-by: grapebaba <281165273@qq.com>

Author: GrapeBaBa

common/metrics/tally_provider.go

@@ -12,7 +12,11 @@ import (
 	"sync"
 	"time"
 
+	"sort"
+
+	"github.com/cactus/go-statsd-client/statsd"
 	"github.com/uber-go/tally"
+	statsdreporter "github.com/uber-go/tally/statsd"
 )
 
 var scopeRegistryKey = tally.KeyForPrefixedStringMap
@@ -181,6 +185,65 @@ func (s *scope) SubScope(prefix string) Scope {
 	return subScope
 }
 
+type statsdReporter struct {
+	reporter tally.StatsReporter
+}
+
+func newStatsdReporter(statsd statsd.Statter, opts statsdreporter.Options) tally.StatsReporter {
+	reporter := statsdreporter.NewReporter(statsd, opts)
+	return &statsdReporter{reporter: reporter}
+}
+
+func (r *statsdReporter) ReportCounter(name string, tags map[string]string, value int64) {
+	r.reporter.ReportCounter(tagsToName(name, tags), tags, value)
+}
+
+func (r *statsdReporter) ReportGauge(name string, tags map[string]string, value float64) {
+	r.reporter.ReportGauge(tagsToName(name, tags), tags, value)
+}
+
+func (r *statsdReporter) ReportTimer(name string, tags map[string]string, interval time.Duration) {
+	r.reporter.ReportTimer(tagsToName(name, tags), tags, interval)
+}
+
+func (r *statsdReporter) ReportHistogramValueSamples(
+	name string,
+	tags map[string]string,
+	buckets tally.Buckets,
+	bucketLowerBound,
+	bucketUpperBound float64,
+	samples int64,
+) {
+	r.reporter.ReportHistogramValueSamples(tagsToName(name, tags), tags, buckets, bucketLowerBound, bucketUpperBound, samples)
+}
+
+func (r *statsdReporter) ReportHistogramDurationSamples(
+	name string,
+	tags map[string]string,
+	buckets tally.Buckets,
+	bucketLowerBound,
+	bucketUpperBound time.Duration,
+	samples int64,
+) {
+	r.reporter.ReportHistogramDurationSamples(tagsToName(name, tags), tags, buckets, bucketLowerBound, bucketUpperBound, samples)
+}
+
+func (r *statsdReporter) Capabilities() tally.Capabilities {
+	return r
+}
+
+func (r *statsdReporter) Reporting() bool {
+	return true
+}
+
+func (r *statsdReporter) Tagging() bool {
+	return true
+}
+
+func (r *statsdReporter) Flush() {
+	// no-op
+}
+
 func (s *scope) fullyQualifiedName(name string) string {
 	if len(s.prefix) == 0 {
 		return name
@@ -217,3 +280,17 @@ func copyStringMap(stringMap map[string]string) map[string]string {
 	}
 	return result
 }
+
+func tagsToName(name string, tags map[string]string) string {
+	var keys []string
+	for k := range tags {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	for _, k := range keys {
+		name = name + tally.DefaultSeparator + k + "-" + tags[k]
+	}
+
+	return name
+}

common/metrics/tally_provider_test.go

@@ -12,10 +12,18 @@ import (
 	"testing"
 	"time"
 
+	"io"
+	"net"
+	"strings"
+
+	"github.com/cactus/go-statsd-client/statsd"
 	"github.com/stretchr/testify/assert"
 	"github.com/uber-go/tally"
+	statsdreporter "github.com/uber-go/tally/statsd"
 )
 
+const statsdAddr string = "127.0.0.1:8125"
+
 type testIntValue struct {
 	val      int64
 	tags     map[string]string
@@ -340,3 +348,52 @@ func TestSubScopeTagged(t *testing.T) {
 		"env": "test",
 	}, r.counters[namespace+".sub.haha"].tags)
 }
+
+func TestMetricsByStatsdReporter(t *testing.T) {
+	t.Parallel()
+	udpAddr, err := net.ResolveUDPAddr("udp", statsdAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	server, err := net.ListenUDP("udp", udpAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer server.Close()
+
+	r := newTestStatsdReporter()
+	opts := tally.ScopeOptions{
+		Prefix:    namespace,
+		Separator: tally.DefaultSeparator,
+		Reporter:  r}
+
+	s, c := newRootScope(opts, 1*time.Second)
+	defer c.Close()
+	subs := s.SubScope("peer").Tagged(map[string]string{"component": "committer", "env": "test"})
+	subs.Counter("success_total").Inc(1)
+	subs.Gauge("channel_total").Update(4)
+
+	buffer := make([]byte, 4096)
+	n, _ := io.ReadAtLeast(server, buffer, 1)
+	result := string(buffer[:n])
+
+	expected := []string{
+		`hyperledger.fabric.peer.success_total.component-committer.env-test:1|c`,
+		`hyperledger.fabric.peer.channel_total.component-committer.env-test:4|g`,
+	}
+
+	for i, res := range strings.Split(result, "\n") {
+		if res != expected[i] {
+			t.Errorf("Got `%s`, expected `%s`", res, expected[i])
+		}
+	}
+}
+
+func newTestStatsdReporter() tally.StatsReporter {
+	statter, _ := statsd.NewBufferedClient(statsdAddr,
+		"", 100*time.Millisecond, 512)
+
+	opts := statsdreporter.Options{}
+	return newStatsdReporter(statter, opts)
+}