[FAB-6006]Add prometheus reporter in metrics module

This patchset add prometheus reporter support in metrics
module, it use prometheus default method and host a http
server for exposing metrics endpoint. You can pull metrics
set from this endpoint.

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

Author: GrapeBaBa

common/metrics/service.go

@@ -16,7 +16,7 @@ import (
 )
 
 const (
-	namespace string = "hyperledger.fabric"
+	namespace string = "hyperledger_fabric"
 )
 
 var rootScope Scope

common/metrics/tally_provider.go

@@ -7,18 +7,25 @@ SPDX-License-Identifier: Apache-2.0
 package metrics
 
 import (
+	"context"
 	"fmt"
 	"io"
+	"net/http"
+	"sort"
 	"sync"
 	"time"
 
-	"sort"
-
 	"github.com/cactus/go-statsd-client/statsd"
+	"github.com/op/go-logging"
+	prom "github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/uber-go/tally"
+	promreporter "github.com/uber-go/tally/prometheus"
 	statsdreporter "github.com/uber-go/tally/statsd"
 )
 
+var logger = logging.MustGetLogger("common/metrics/tally")
+
 var scopeRegistryKey = tally.KeyForPrefixedStringMap
 
 type counter struct {
@@ -187,11 +194,33 @@ func (s *scope) SubScope(prefix string) Scope {
 
 type statsdReporter struct {
 	reporter tally.StatsReporter
+	statter  statsd.Statter
+}
+
+type promReporter struct {
+	reporter promreporter.Reporter
+	server   *http.Server
+	registry *prom.Registry
 }
 
 func newStatsdReporter(statsd statsd.Statter, opts statsdreporter.Options) tally.StatsReporter {
 	reporter := statsdreporter.NewReporter(statsd, opts)
-	return &statsdReporter{reporter: reporter}
+	return &statsdReporter{reporter: reporter, statter: statsd}
+}
+
+func newPrometheusReporter(opts promreporter.Options) promreporter.Reporter {
+	reporter := promreporter.NewReporter(opts)
+	//TODO:Use config instead of hard code
+	server := &http.Server{Addr: ":8080", Handler: nil}
+	//TODO:Return this error to caller
+	go func() {
+		if err := server.ListenAndServe(); err != nil {
+			logger.Errorf("Metrics prometheus reporter start failed %s", err)
+		}
+	}()
+	promReporter := &promReporter{reporter: reporter, server: server, registry: opts.Registerer.(*prom.Registry)}
+	http.Handle("/metrics", promReporter.HTTPHandler())
+	return promReporter
 }
 
 func (r *statsdReporter) ReportCounter(name string, tags map[string]string, value int64) {
@@ -244,6 +273,84 @@ func (r *statsdReporter) Flush() {
 	// no-op
 }
 
+func (r *statsdReporter) Close() error {
+	return r.statter.Close()
+}
+
+func (r *promReporter) RegisterCounter(
+	name string,
+	tagKeys []string,
+	desc string,
+) (*prom.CounterVec, error) {
+	return r.reporter.RegisterCounter(name, tagKeys, desc)
+}
+
+// AllocateCounter implements tally.CachedStatsReporter.
+func (r *promReporter) AllocateCounter(name string, tags map[string]string) tally.CachedCount {
+	return r.reporter.AllocateCounter(name, tags)
+}
+
+func (r *promReporter) RegisterGauge(
+	name string,
+	tagKeys []string,
+	desc string,
+) (*prom.GaugeVec, error) {
+	return r.reporter.RegisterGauge(name, tagKeys, desc)
+}
+
+// AllocateGauge implements tally.CachedStatsReporter.
+func (r *promReporter) AllocateGauge(name string, tags map[string]string) tally.CachedGauge {
+	return r.reporter.AllocateGauge(name, tags)
+}
+
+func (r *promReporter) RegisterTimer(
+	name string,
+	tagKeys []string,
+	desc string,
+	opts *promreporter.RegisterTimerOptions,
+) (promreporter.TimerUnion, error) {
+	return r.reporter.RegisterTimer(name, tagKeys, desc, opts)
+}
+
+// AllocateTimer implements tally.CachedStatsReporter.
+func (r *promReporter) AllocateTimer(name string, tags map[string]string) tally.CachedTimer {
+	return r.reporter.AllocateTimer(name, tags)
+}
+
+func (r *promReporter) AllocateHistogram(
+	name string,
+	tags map[string]string,
+	buckets tally.Buckets,
+) tally.CachedHistogram {
+	return r.reporter.AllocateHistogram(name, tags, buckets)
+}
+
+func (r *promReporter) Capabilities() tally.Capabilities {
+	return r
+}
+
+func (r *promReporter) Reporting() bool {
+	return true
+}
+
+func (r *promReporter) Tagging() bool {
+	return true
+}
+
+// Flush does nothing for prometheus
+func (r *promReporter) Flush() {
+
+}
+
+func (r *promReporter) Close() error {
+	//TODO: Timeout here?
+	return r.server.Shutdown(context.Background())
+}
+
+func (r *promReporter) HTTPHandler() http.Handler {
+	return promhttp.HandlerFor(r.registry, promhttp.HandlerOpts{})
+}
+
 func (s *scope) fullyQualifiedName(name string) string {
 	if len(s.prefix) == 0 {
 		return name

common/metrics/tally_provider_test.go

@@ -7,18 +7,21 @@ SPDX-License-Identifier: Apache-2.0
 package metrics
 
 import (
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
-	"io"
-	"net"
-	"strings"
-
 	"github.com/cactus/go-statsd-client/statsd"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/uber-go/tally"
+	promreporter "github.com/uber-go/tally/prometheus"
 	statsdreporter "github.com/uber-go/tally/statsd"
 )
 
@@ -379,8 +382,8 @@ func TestMetricsByStatsdReporter(t *testing.T) {
 	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`,
+		`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") {
@@ -390,10 +393,57 @@ func TestMetricsByStatsdReporter(t *testing.T) {
 	}
 }
 
+func TestMetricsByPrometheusReporter(t *testing.T) {
+	t.Parallel()
+	r := newTestPrometheusReporter()
+
+	opts := tally.ScopeOptions{
+		Prefix:         namespace,
+		Separator:      promreporter.DefaultSeparator,
+		CachedReporter: r}
+
+	s, c := newRootScope(opts, 1*time.Second)
+	defer c.Close()
+
+	scrape := func() string {
+		resp, _ := http.Get("http://localhost:8080/metrics")
+		buf, _ := ioutil.ReadAll(resp.Body)
+		return string(buf)
+	}
+	subs := s.SubScope("peer").Tagged(map[string]string{"component": "committer", "env": "test"})
+	subs.Counter("success_total").Inc(1)
+	subs.Gauge("channel_total").Update(4)
+
+	time.Sleep(2 * time.Second)
+
+	expected := []string{
+		`# HELP hyperledger_fabric_peer_channel_total hyperledger_fabric_peer_channel_total gauge`,
+		`# TYPE hyperledger_fabric_peer_channel_total gauge`,
+		`hyperledger_fabric_peer_channel_total{component="committer",env="test"} 4`,
+		`# HELP hyperledger_fabric_peer_success_total hyperledger_fabric_peer_success_total counter`,
+		`# TYPE hyperledger_fabric_peer_success_total counter`,
+		`hyperledger_fabric_peer_success_total{component="committer",env="test"} 1`,
+		``,
+	}
+
+	result := strings.Split(scrape(), "\n")
+
+	for i, res := range result {
+		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)
 }
+
+func newTestPrometheusReporter() promreporter.Reporter {
+	opts := promreporter.Options{Registerer: prometheus.NewRegistry()}
+	return newPrometheusReporter(opts)
+}