feat(telemetry): enable statsd and dogstatsd telemetry sinks (#18646)

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: Aleksandr Bezobchuk <alexanderbez@users.noreply.github.com>
Co-authored-by: marbar3778 <marbar3778@yahoo.com>
Co-authored-by: Marko <marko@baricevic.me>

telemetry/metrics.go

@@ -4,9 +4,11 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"time"
 
 	"github.com/hashicorp/go-metrics"
+	"github.com/hashicorp/go-metrics/datadog"
 	metricsprom "github.com/hashicorp/go-metrics/prometheus"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/expfmt"
@@ -21,8 +23,17 @@ const (
 	FormatDefault    = ""
 	FormatPrometheus = "prometheus"
 	FormatText       = "text"
+
+	MetricSinkInMem      = "mem"
+	MetricSinkStatsd     = "statsd"
+	MetricSinkDogsStatsd = "dogstatsd"
 )
 
+// DisplayableSink is an interface that defines a method for displaying metrics.
+type DisplayableSink interface {
+	DisplayMetrics(resp http.ResponseWriter, req *http.Request) (any, error)
+}
+
 // Config defines the configuration options for application telemetry.
 type Config struct {
 	// Prefixed with keys to separate services
@@ -52,6 +63,17 @@ type Config struct {
 	// Example:
 	// [["chain_id", "cosmoshub-1"]]
 	GlobalLabels [][]string `mapstructure:"global-labels"`
+
+	// MetricsSink defines the type of metrics backend to use.
+	MetricsSink string `mapstructure:"type" default:"mem"`
+
+	// StatsdAddr defines the address of a statsd server to send metrics to.
+	// Only utilized if MetricsSink is set to "statsd" or "dogstatsd".
+	StatsdAddr string `mapstructure:"statsd-addr"`
+
+	// DatadogHostname defines the hostname to use when emitting metrics to
+	// Datadog. Only utilized if MetricsSink is set to "dogstatsd".
+	DatadogHostname string `mapstructure:"datadog-hostname"`
 }
 
 // Metrics defines a wrapper around application telemetry functionality. It allows
@@ -60,7 +82,7 @@ type Config struct {
 // by the operator. In addition to the sinks, when a process gets a SIGUSR1, a
 // dump of formatted recent metrics will be sent to STDERR.
 type Metrics struct {
-	memSink           *metrics.InmemSink
+	sink              metrics.MetricSink
 	prometheusEnabled bool
 }
 
@@ -81,7 +103,6 @@ func New(cfg Config) (_ *Metrics, rerr error) {
 		for i, gl := range cfg.GlobalLabels {
 			parsedGlobalLabels[i] = NewLabel(gl[0], gl[1])
 		}
-
 		globalLabels = parsedGlobalLabels
 	}
 
@@ -89,16 +110,32 @@ func New(cfg Config) (_ *Metrics, rerr error) {
 	metricsConf.EnableHostname = cfg.EnableHostname
 	metricsConf.EnableHostnameLabel = cfg.EnableHostnameLabel
 
-	memSink := metrics.NewInmemSink(10*time.Second, time.Minute)
-	inMemSig := metrics.DefaultInmemSignal(memSink)
-	defer func() {
-		if rerr != nil {
-			inMemSig.Stop()
-		}
-	}()
+	var (
+		sink metrics.MetricSink
+		err  error
+	)
+	switch cfg.MetricsSink {
+	case MetricSinkStatsd:
+		sink, err = metrics.NewStatsdSink(cfg.StatsdAddr)
+	case MetricSinkDogsStatsd:
+		sink, err = datadog.NewDogStatsdSink(cfg.StatsdAddr, cfg.DatadogHostname)
+	default:
+		memSink := metrics.NewInmemSink(10*time.Second, time.Minute)
+		sink = memSink
+		inMemSig := metrics.DefaultInmemSignal(memSink)
+		defer func() {
+			if rerr != nil {
+				inMemSig.Stop()
+			}
+		}()
+	}
 
-	m := &Metrics{memSink: memSink}
-	fanout := metrics.FanoutSink{memSink}
+	if err != nil {
+		return nil, err
+	}
+
+	m := &Metrics{sink: sink}
+	fanout := metrics.FanoutSink{sink}
 
 	if cfg.PrometheusRetentionTime > 0 {
 		m.prometheusEnabled = true
@@ -140,6 +177,8 @@ func (m *Metrics) Gather(format string) (GatherResponse, error) {
 	}
 }
 
+// gatherPrometheus collects Prometheus metrics and returns a GatherResponse.
+// If Prometheus metrics are not enabled, it returns an error.
 func (m *Metrics) gatherPrometheus() (GatherResponse, error) {
 	if !m.prometheusEnabled {
 		return GatherResponse{}, fmt.Errorf("prometheus metrics are not enabled")
@@ -154,6 +193,7 @@ func (m *Metrics) gatherPrometheus() (GatherResponse, error) {
 	defer buf.Reset()
 
 	e := expfmt.NewEncoder(buf, expfmt.FmtText)
+
 	for _, mf := range metricsFamilies {
 		if err := e.Encode(mf); err != nil {
 			return GatherResponse{}, fmt.Errorf("failed to encode prometheus metrics: %w", err)
@@ -163,8 +203,14 @@ func (m *Metrics) gatherPrometheus() (GatherResponse, error) {
 	return GatherResponse{ContentType: string(expfmt.FmtText), Metrics: buf.Bytes()}, nil
 }
 
+// gatherGeneric collects generic metrics and returns a GatherResponse.
 func (m *Metrics) gatherGeneric() (GatherResponse, error) {
-	summary, err := m.memSink.DisplayMetrics(nil, nil)
+	gm, ok := m.sink.(DisplayableSink)
+	if !ok {
+		return GatherResponse{}, fmt.Errorf("non in-memory metrics sink does not support generic format")
+	}
+
+	summary, err := gm.DisplayMetrics(nil, nil)
 	if err != nil {
 		return GatherResponse{}, fmt.Errorf("failed to gather in-memory metrics: %w", err)
 	}

telemetry/metrics_test.go

@@ -19,6 +19,7 @@ func TestMetrics_Disabled(t *testing.T) {
 
 func TestMetrics_InMem(t *testing.T) {
 	m, err := New(Config{
+		MetricsSink:    MetricSinkInMem,
 		Enabled:        true,
 		EnableHostname: false,
 		ServiceName:    "test",
@@ -42,6 +43,7 @@ func TestMetrics_InMem(t *testing.T) {
 
 func TestMetrics_Prom(t *testing.T) {
 	m, err := New(Config{
+		MetricsSink:             MetricSinkInMem,
 		Enabled:                 true,
 		EnableHostname:          false,
 		ServiceName:             "test",