Merge pull request #22 from InjectiveLabs/f/otel-counters

- feat: allow otel counters
- feat: configure metrics

Author: hmoragrega

client.go

@@ -2,7 +2,9 @@ package metrics
 
 import (
 	"context"
+	"os"
 	"runtime"
+	"strings"
 	"sync"
 	"time"
 
@@ -38,23 +40,24 @@ var (
 )
 
 type StatterConfig struct {
-	Addr                 string            // localhost:8125
-	Prefix               string            // metrics prefix
-	Agent                string            // telegraf/datadog
-	EnvName              string            // dev/test/staging/prod
-	HostName             string            // hostname
-	Version              string            // version
-	DefaultTags          []interface{}     // default tags for all metrics
-	StuckFunctionTimeout time.Duration     // stuck time
-	MockingThreshold     time.Duration     // mocking threshold
-	MockingEnabled       bool              // whether to enable mock statter, which only produce logs
-	Disabled             bool              // whether to disable metrics completely
-	TracingEnabled       bool              // whether tracing should be enabled
-	ProfilingEnabled     bool              // whether Datadog profiling should be enabled
-	MixPanelEnabled      bool              // whether MixPanel should be enabled
-	MixPanelProjectToken string            // MixPanel project token
-	OTELInsecure         bool              // disable TLS (use for self-hosted SigNoz without TLS)
-	OTELHeaders          map[string]string // extra headers, e.g. {"signoz-access-token": "<token>"} for SigNoz Cloud
+	Addr                   string            // localhost:8125
+	Prefix                 string            // metrics prefix
+	Agent                  string            // telegraf/datadog
+	EnvName                string            // dev/test/staging/prod
+	HostName               string            // hostname
+	Version                string            // version
+	DefaultTags            []interface{}     // default tags for all metrics
+	StuckFunctionTimeout   time.Duration     // stuck time
+	MockingThreshold       time.Duration     // mocking threshold
+	MockingEnabled         bool              // whether to enable mock statter, which only produce logs
+	Disabled               bool              // whether to disable metrics completely
+	TracingEnabled         bool              // whether tracing should be enabled
+	ProfilingEnabled       bool              // whether Datadog profiling should be enabled
+	MixPanelEnabled        bool              // whether MixPanel should be enabled
+	MixPanelProjectToken   string            // MixPanel project token
+	OTELInsecure           bool              // disable TLS (use for self-hosted SigNoz without TLS)
+	OTELHeaders            map[string]string // extra headers, e.g. {"signoz-access-token": "<token>"} for SigNoz Cloud
+	OTELUseCounterForCount bool              // use monotonic OTel counters for Count/Incr instead of UpDownCounters
 }
 
 func (m *StatterConfig) BaseTags() []string {
@@ -177,6 +180,7 @@ func Init(addr string, prefix string, cfg *StatterConfig) error {
 			cfg.OTELInsecure,
 			cfg.OTELHeaders,
 			config.BaseTags(),
+			cfg.OTELUseCounterForCount,
 		)
 
 	default:
@@ -229,6 +233,72 @@ func Init(addr string, prefix string, cfg *StatterConfig) error {
 	return nil
 }
 
+type ServiceConfig struct {
+	Disabled             bool
+	ServiceName          string
+	AgentID              string
+	AgentAddress         string
+	MetricsPrefix        string
+	EnvName              string
+	MockingEnabled       bool
+	MockingThreshold     time.Duration
+	MixPanelEnabled      bool
+	MixPanelProjectToken string
+	OTelInsecure         bool
+	OTelUseCounters      bool
+	TracingEnabled       bool
+	RetryInitialInterval time.Duration
+}
+
+func (c ServiceConfig) normalizePrefix() string {
+	return strings.TrimRight(c.MetricsPrefix, ".") + "."
+}
+
+func InitService(ctx context.Context, cfg ServiceConfig) (func(timeout time.Duration), error) {
+	closeFn := func(time.Duration) {}
+	if err := cfg.Validate(); err != nil {
+		return closeFn, err
+	}
+	if cfg.Disabled {
+		return closeFn, nil
+	}
+
+	if cfg.RetryInitialInterval <= 0 {
+		cfg.RetryInitialInterval = 10 * time.Second
+	}
+
+	for {
+		hostname, _ := os.Hostname()
+		err := Init(cfg.AgentAddress, cfg.normalizePrefix(), &StatterConfig{
+			Agent:                  cfg.AgentID,
+			EnvName:                cfg.EnvName,
+			HostName:               hostname,
+			MockingEnabled:         cfg.MockingEnabled,
+			MockingThreshold:       cfg.MockingThreshold,
+			MixPanelEnabled:        cfg.MixPanelEnabled,
+			MixPanelProjectToken:   cfg.MixPanelProjectToken,
+			OTELInsecure:           cfg.OTelInsecure,
+			OTELUseCounterForCount: cfg.OTelUseCounters,
+			TracingEnabled:         cfg.TracingEnabled,
+			DefaultTags:            []interface{}{"service.name", cfg.ServiceName},
+		})
+		if err != nil {
+			log.WithError(err).Warningf("metrics init failed, will retry in %s seconds", cfg.RetryInitialInterval)
+			select {
+			case <-ctx.Done():
+				return closeFn, nil
+			case <-time.After(cfg.RetryInitialInterval):
+			}
+			continue
+		}
+		log.Debugf("metrics %s client initialized at %s with prefix %s (mocking %v)",
+			cfg.AgentID, cfg.AgentAddress, cfg.normalizePrefix(), cfg.MockingEnabled)
+		break
+	}
+
+	return CloseWithTimeout, nil
+}
+
 func StartMixPanel(projectToken string) {
 	clientMux.Lock()
 	defer clientMux.Unlock()

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/cosmos/cosmos-sdk v0.50.6
 	github.com/mixpanel/mixpanel-go v1.2.1
 	github.com/pkg/errors v0.9.1
+	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/otel v1.43.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.43.0
@@ -120,7 +121,6 @@ require (
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
 	github.com/spf13/cobra v1.8.0 // indirect
-	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect
 	github.com/tendermint/go-amino v0.16.0 // indirect
 	github.com/tinylib/msgp v1.1.8 // indirect

otel.go

@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"sync"
 	"time"
@@ -20,8 +21,10 @@ type otelStatter struct {
 	meter         otelmetric.Meter
 	meterProvider *sdkmetric.MeterProvider
 	prefix        string
+	useCounters   bool
 
 	mu             sync.RWMutex
+	counters       map[string]otelmetric.Int64Counter
 	updownCounters map[string]otelmetric.Int64UpDownCounter
 	gauges         map[string]otelmetric.Float64Gauge
 	histograms     map[string]otelmetric.Float64Histogram
@@ -41,7 +44,7 @@ func newOTELResource(baseTags []string) *resource.Resource {
 	return res
 }
 
-func newOTELStatter(endpoint, prefix string, insecure bool, headers map[string]string, baseTags []string) (Statter, error) {
+func newOTELStatter(endpoint, prefix string, insecure bool, headers map[string]string, baseTags []string, useCounters bool) (Statter, error) {
 	ctx := context.Background()
 
 	metricOpts := []otlpmetricgrpc.Option{
@@ -70,6 +73,8 @@ func newOTELStatter(endpoint, prefix string, insecure bool, headers map[string]s
 		meter:          mp.Meter(prefix),
 		meterProvider:  mp,
 		prefix:         prefix,
+		useCounters:    useCounters,
+		counters:       make(map[string]otelmetric.Int64Counter),
 		updownCounters: make(map[string]otelmetric.Int64UpDownCounter),
 		gauges:         make(map[string]otelmetric.Float64Gauge),
 		histograms:     make(map[string]otelmetric.Float64Histogram),
@@ -114,6 +119,27 @@ func (s *otelStatter) tagsToAttrs(tags []string) []attribute.KeyValue {
 	return attrs
 }
 
+func (s *otelStatter) getCounter(name string) (otelmetric.Int64Counter, error) {
+	fullName := s.prefix + name
+	s.mu.RLock()
+	c, ok := s.counters[fullName]
+	s.mu.RUnlock()
+	if ok {
+		return c, nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if c, ok = s.counters[fullName]; ok {
+		return c, nil
+	}
+	c, err := s.meter.Int64Counter(fullName)
+	if err != nil {
+		return nil, err
+	}
+	s.counters[fullName] = c
+	return c, nil
+}
+
 func (s *otelStatter) getUpDownCounter(name string) (otelmetric.Int64UpDownCounter, error) {
 	fullName := s.prefix + name
 	s.mu.RLock()
@@ -178,6 +204,17 @@ func (s *otelStatter) getHistogram(name string) (otelmetric.Float64Histogram, er
 }
 
 func (s *otelStatter) Count(name string, value int64, tags []string, rate float64) error {
+	if s.useCounters {
+		if value < 0 {
+			return fmt.Errorf("counter value must be non-negative")
+		}
+		c, err := s.getCounter(name)
+		if err != nil {
+			return err
+		}
+		c.Add(context.Background(), value, otelmetric.WithAttributes(s.tagsToAttrs(tags)...))
+		return nil
+	}
 	c, err := s.getUpDownCounter(name)
 	if err != nil {
 		return err
@@ -187,6 +224,14 @@ func (s *otelStatter) Count(name string, value int64, tags []string, rate float6
 }
 
 func (s *otelStatter) Incr(name string, tags []string, rate float64) error {
+	if s.useCounters {
+		c, err := s.getCounter(name)
+		if err != nil {
+			return err
+		}
+		c.Add(context.Background(), 1, otelmetric.WithAttributes(s.tagsToAttrs(tags)...))
+		return nil
+	}
 	c, err := s.getUpDownCounter(name)
 	if err != nil {
 		return err
@@ -239,3 +284,39 @@ func (s *otelStatter) Close() error {
 func (s *otelStatter) CloseCtx(ctx context.Context) error {
 	return s.meterProvider.Shutdown(ctx)
 }
+
+func OTelSetupHistogram(name string, opts ...otelmetric.Float64HistogramOption) error {
+	clientMux.RLock()
+	c := client
+	clientMux.RUnlock()
+
+	if c == nil {
+		return nil
+	}
+
+	s, ok := c.(*otelStatter)
+	if !ok {
+		return nil
+	}
+
+	fullName := s.prefix + name
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, exists := s.histograms[fullName]; exists {
+		return fmt.Errorf("histogram %s already initialized", fullName)
+	}
+
+	finalOpts := append([]otelmetric.Float64HistogramOption{
+		otelmetric.WithUnit("ms"),
+	}, opts...)
+
+	h, err := s.meter.Float64Histogram(fullName, finalOpts...)
+	if err != nil {
+		return err
+	}
+
+	s.histograms[fullName] = h
+	return nil
+}

otel_test.go

@@ -0,0 +1,102 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	otelmetric "go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+func newTestOTELStatter(useCounters bool) (*otelStatter, *sdkmetric.ManualReader) {
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	return &otelStatter{
+		meter:          mp.Meter("test"),
+		meterProvider:  mp,
+		useCounters:    useCounters,
+		counters:       make(map[string]otelmetric.Int64Counter),
+		updownCounters: make(map[string]otelmetric.Int64UpDownCounter),
+		gauges:         make(map[string]otelmetric.Float64Gauge),
+		histograms:     make(map[string]otelmetric.Float64Histogram),
+	}, reader
+}
+
+func collectOTELMetric(t *testing.T, reader *sdkmetric.ManualReader, name string) metricdata.Metrics {
+	t.Helper()
+
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(t.Context(), &rm))
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name == name {
+				return m
+			}
+		}
+	}
+	require.Failf(t, "metric not found", "metric %q not found", name)
+	return metricdata.Metrics{}
+}
+
+func requireOTELInt64Sum(t *testing.T, m metricdata.Metrics) metricdata.Sum[int64] {
+	t.Helper()
+
+	sum, ok := m.Data.(metricdata.Sum[int64])
+	require.Truef(t, ok, "expected int64 sum, got %T", m.Data)
+	return sum
+}
+
+func TestOTELCountDefaultsToUpDownCounter(t *testing.T) {
+	statter, reader := newTestOTELStatter(false)
+	t.Cleanup(func() {
+		require.NoError(t, statter.Close())
+	})
+
+	require.NoError(t, statter.Count("events.total", 5, []string{"status=ok"}, 1))
+
+	sum := requireOTELInt64Sum(t, collectOTELMetric(t, reader, "events.total"))
+	require.False(t, sum.IsMonotonic)
+	require.Len(t, sum.DataPoints, 1)
+	require.EqualValues(t, 5, sum.DataPoints[0].Value)
+}
+
+func TestOTELCountCanUseCounter(t *testing.T) {
+	statter, reader := newTestOTELStatter(true)
+	t.Cleanup(func() {
+		require.NoError(t, statter.Close())
+	})
+
+	require.NoError(t, statter.Count("events.total", 5, []string{"status=ok"}, 1))
+	require.NoError(t, statter.Incr("events.total", []string{"status=ok"}, 1))
+
+	sum := requireOTELInt64Sum(t, collectOTELMetric(t, reader, "events.total"))
+	require.True(t, sum.IsMonotonic)
+	require.Len(t, sum.DataPoints, 1)
+	require.EqualValues(t, 6, sum.DataPoints[0].Value)
+}
+
+func TestOTELCountRejectsNegativeValueWithCounterOption(t *testing.T) {
+	statter, _ := newTestOTELStatter(true)
+	t.Cleanup(func() {
+		require.NoError(t, statter.Close())
+	})
+
+	err := statter.Count("events.total", -1, []string{"status=ok"}, 1)
+	require.EqualError(t, err, "counter value must be non-negative")
+	require.Empty(t, statter.counters)
+}
+
+func TestOTELDecrUsesUpDownCounterWithCounterOption(t *testing.T) {
+	statter, reader := newTestOTELStatter(true)
+	t.Cleanup(func() {
+		require.NoError(t, statter.Close())
+	})
+
+	require.NoError(t, statter.Decr("active.requests", []string{"status=ok"}, 1))
+
+	sum := requireOTELInt64Sum(t, collectOTELMetric(t, reader, "active.requests"))
+	require.False(t, sum.IsMonotonic)
+	require.Len(t, sum.DataPoints, 1)
+	require.EqualValues(t, -1, sum.DataPoints[0].Value)
+}