feat(GEN-95|HELIX-1134): add datadog and mixpanel support (#6)

* feat(GEN-95): add datadog support feat(HELIX-1134): add mixpanel support * feat(HELIX-1134): Counter generics * feat(HELIX-1134): remove threshold from count metric using mock
InjectiveLabs · Sep 30, 2024 · 815d3fe · 815d3fe
1 parent c71af53
commit 815d3fe
Show file tree

Hide file tree

Showing 5 changed files with 255 additions and 29 deletions.
diff --git a/client.go b/client.go
@@ -1,16 +1,20 @@
 package metrics
 
 import (
-	"fmt"
+	"runtime"
 	"sync"
 	"time"
 
 	dogstatsd "github.com/DataDog/datadog-go/v5/statsd"
-	statsd "github.com/alexcesaro/statsd"
+	"github.com/alexcesaro/statsd"
+	"github.com/mixpanel/mixpanel-go"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
 	ddotel "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/opentelemetry"
+	"gopkg.in/DataDog/dd-trace-go.v1/profiler"
+
+	log "github.com/InjectiveLabs/suplog"
 )
 
 const (
@@ -25,8 +29,9 @@ var (
 	clientMux = new(sync.RWMutex)
 	config    *StatterConfig
 
-	traceProvider *ddotel.TracerProvider
-	tracer        trace.Tracer
+	traceProvider  *ddotel.TracerProvider
+	tracer         trace.Tracer
+	mixPanelClient *mixpanel.ApiClient
 )
 
 type StatterConfig struct {
@@ -35,10 +40,15 @@ type StatterConfig struct {
 	Agent                string        // telegraf/datadog
 	EnvName              string        // dev/test/staging/prod
 	HostName             string        // hostname
+	Version              string        // version
 	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 DataDog tracing should be enabled (via OpenTelemetry)
+	ProfilingEnabled     bool          // whether Datadog profiling should be enabled
+	MixPanelEnabled      bool          // whether MixPanel should be enabled
+	MixPanelProjectToken string        // MixPanel project token
 }
 
 func (m *StatterConfig) BaseTags() []string {
@@ -82,35 +92,15 @@ func Close() {
 	if client == nil {
 		return
 	}
-	if traceProvider != nil {
-		traceProvider.Shutdown()
-	}
 	client.Close()
 }
 
-func Disable() {
-	config = checkConfig(nil)
-	clientMux.Lock()
-	client = newMockStatter(true)
-	clientMux.Unlock()
-	tracer = nil
-}
-
-func InitWithConfig(cfg *StatterConfig) error {
-	return Init(cfg.Addr, cfg.Prefix, cfg)
-}
-
 func Init(addr string, prefix string, cfg *StatterConfig) error {
-	if cfg.Disabled {
-		Disable()
-		return nil
-	}
-
 	config = checkConfig(cfg)
 	if config.MockingEnabled {
 		// init a mock statter instead of real statsd client
 		clientMux.Lock()
-		client = newMockStatter(false)
+		client = newMockStatter(cfg)
 		clientMux.Unlock()
 		return nil
 	}
@@ -150,12 +140,51 @@ func Init(addr string, prefix string, cfg *StatterConfig) error {
 	clientMux.Unlock()
 
 	// OpenTelemetry tracing via DataDog provider
-	if cfg.TracingEnabled {
+	if cfg.Agent == DatadogAgent && cfg.TracingEnabled {
 		traceProvider = ddotel.NewTracerProvider()
 		otel.SetTracerProvider(traceProvider)
 		tracer = otel.Tracer("")
 	}
 
+	if cfg.Agent == DatadogAgent && cfg.ProfilingEnabled {
+		err = setupProfiler(cfg)
+		if err != nil {
+			return err
+		}
+	}
+
+	if cfg.MixPanelEnabled {
+		StartMixPanel(cfg.MixPanelProjectToken)
+	}
+
+	return nil
+}
+
+func StartMixPanel(projectToken string) {
+	clientMux.Lock()
+	defer clientMux.Unlock()
+	mixPanelClient = mixpanel.NewApiClient(projectToken)
+}
+
+func setupProfiler(cfg *StatterConfig) error {
+	runtime.SetMutexProfileFraction(5)
+	runtime.SetBlockProfileRate(5)
+
+	err := profiler.Start(
+		profiler.WithService(cfg.Prefix),
+		profiler.WithEnv(cfg.EnvName),
+		profiler.WithHostname(cfg.HostName),
+		profiler.WithVersion(cfg.Version),
+		profiler.WithProfileTypes(
+			profiler.CPUProfile,
+			profiler.HeapProfile,
+			profiler.BlockProfile,
+			profiler.MutexProfile,
+		),
+	)
+	if err != nil {
+		return errors.Wrap(err, "profiler start failed")
+	}
 	return nil
 }
 
@@ -173,44 +202,86 @@ func checkConfig(cfg *StatterConfig) *StatterConfig {
 }
 
 func errHandler(err error) {
-	fmt.Printf("statsd error, err: %v\n", err)
+	log.WithError(err).Errorln("statsd error")
 }
 
-func newMockStatter(noop bool) Statter {
-	return &mockStatter{}
+func newMockStatter(cfg *StatterConfig) Statter {
+	return &mockStatter{
+		l: log.WithFields(log.Fields{
+			"module": "mock_statter",
+		}),
+		threshold: cfg.MockingThreshold,
+	}
 }
 
 type mockStatter struct {
+	l         log.Logger
+	threshold time.Duration
 }
 
 func (s *mockStatter) Count(name string, value int64, tags []string, rate float64) error {
+	s.l.WithFields(s.withTagFields(tags)).Debugf("Count %s: %v", name, value)
 	return nil
 }
 
 func (s *mockStatter) Incr(name string, tags []string, rate float64) error {
+	if s.threshold > 0 {
+		return nil
+	}
+	s.l.WithFields(s.withTagFields(tags)).Debugf("Incr %s", name)
 	return nil
 }
 
 func (s *mockStatter) Decr(name string, tags []string, rate float64) error {
+	if s.threshold > 0 {
+		return nil
+	}
+	s.l.WithFields(s.withTagFields(tags)).Debugf("Decr %s", name)
 	return nil
 }
 
 func (s *mockStatter) Gauge(name string, value float64, tags []string, rate float64) error {
+	if s.threshold > 0 {
+		return nil
+	}
+	s.l.WithFields(s.withTagFields(tags)).Debugf("Gauge %s: %v", name, value)
 	return nil
 }
 
 func (s *mockStatter) Timing(name string, value time.Duration, tags []string, rate float64) error {
+	if value > s.threshold {
+		s.l.WithFields(s.withTagFields(tags)).Debugf("Timing %s: %v", name, value)
+	}
 	return nil
 }
 
 func (s *mockStatter) Histogram(name string, value float64, tags []string, rate float64) error {
+	if value > float64(s.threshold.Milliseconds()) {
+		s.l.WithFields(s.withTagFields(tags)).Debugf("Histogram %s: %v", name, value)
+	}
 	return nil
 }
 
 func (s *mockStatter) Unique(bucket string, value string) error {
+	if s.threshold > 0 {
+		return nil
+	}
+	s.l.Debugf("Unique %s: %v", bucket, value)
 	return nil
 }
 
 func (s *mockStatter) Close() error {
+	s.l.Debugf("closed at %s", time.Now())
 	return nil
 }
+
+func (s *mockStatter) withTagFields(tags []string) log.Fields {
+	fields := make(log.Fields)
+	for i := 0; i < len(tags); i += 2 {
+		if i+1 >= len(tags) { // protect against odd number of tags
+			break
+		}
+		fields[tags[i]] = tags[i+1]
+	}
+	return fields
+}
diff --git a/go.mod b/go.mod
@@ -4,8 +4,10 @@ go 1.22
 
 require (
 	github.com/DataDog/datadog-go/v5 v5.3.0
+	github.com/InjectiveLabs/suplog v1.3.3
 	github.com/alexcesaro/statsd v2.0.0+incompatible
 	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/stretchr/testify v1.9.0
 	go.opentelemetry.io/otel v1.24.0
@@ -29,11 +31,16 @@ require (
 	github.com/DataDog/datadog-agent/pkg/remoteconfig/state v0.48.1 // indirect
 	github.com/DataDog/go-libddwaf/v2 v2.3.2 // indirect
 	github.com/DataDog/go-tuf v1.0.2-0.5.2 // indirect
+	github.com/DataDog/gostackparse v0.7.0 // indirect
 	github.com/DataDog/sketches-go v1.4.2 // indirect
 	github.com/DataDog/zstd v1.5.5 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
+	github.com/aws/aws-sdk-go v1.44.327 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/bitly/go-simplejson v0.5.1 // indirect
 	github.com/btcsuite/btcd/btcec/v2 v2.3.2 // indirect
+	github.com/bugsnag/bugsnag-go v1.5.3 // indirect
+	github.com/bugsnag/panicwrap v1.3.4 // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cockroachdb/errors v1.11.1 // indirect
@@ -61,12 +68,14 @@ require (
 	github.com/go-logfmt/logfmt v0.6.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gofrs/uuid v4.4.0+incompatible // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/glog v1.2.0 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/pprof v0.0.0-20230817174616-7a8ec2ada47b // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
@@ -75,7 +84,9 @@ require (
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/iancoleman/strcase v0.3.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/jmhodges/levigo v1.0.0 // indirect
+	github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect
 	github.com/klauspost/compress v1.17.7 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
@@ -84,6 +95,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/oasisprotocol/curve25519-voi v0.0.0-20230904125328-1f23a7beb09a // indirect
+	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/outcaste-io/ristretto v0.2.3 // indirect
 	github.com/petermattis/goid v0.0.0-20231207134359-e60b3f734c67 // indirect
 	github.com/philhofer/fwd v1.1.2 // indirect
@@ -92,10 +104,13 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.52.2 // indirect
 	github.com/prometheus/procfs v0.13.0 // indirect
+	github.com/richardartoul/molecule v1.0.1-0.20221107223329-32cfee06a052 // indirect
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
 	github.com/rs/zerolog v1.32.0 // indirect
 	github.com/sasha-s/go-deadlock v0.3.1 // indirect
 	github.com/secure-systems-lab/go-securesystemslib v0.7.0 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	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