Prometheus

Author: aopoltorzhicky

config/config.go

@@ -15,8 +15,8 @@ type Config struct {
 	Database    Database              `yaml:"database" validate:"required"`
 	DataSources map[string]DataSource `yaml:"datasources"`
 	Contracts   map[string]Contract   `yaml:"contracts"`
-	Hasura      Hasura                `yaml:"hasura" validate:"omitempty"`
-	Prometheus  Prometheus            `yaml:"prometheus" validate:"omitempty"`
+	Hasura      *Hasura               `yaml:"hasura" validate:"omitempty"`
+	Prometheus  *Prometheus           `yaml:"prometheus" validate:"omitempty"`
 }
 
 // Substitute -

go.mod

@@ -8,8 +8,8 @@ require (
 	github.com/gorilla/websocket v1.4.2
 	github.com/iancoleman/strcase v0.1.3
 	github.com/json-iterator/go v1.1.12
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.11.0
 	github.com/sergi/go-diff v1.2.0
 	github.com/shopspring/decimal v0.0.0-20200227202807-02e2044944cc
 	github.com/sirupsen/logrus v1.7.0

go.sum

@@ -41,12 +41,15 @@ func checkHealth(ctx context.Context, api *API) {
 }
 
 // Create - creates hasura models
-func Create(ctx context.Context, hasura config.Hasura, cfg config.Database, views []string, models ...interface{}) error {
+func Create(ctx context.Context, hasura *config.Hasura, cfg config.Database, views []string, models ...interface{}) error {
+	if hasura == nil {
+		return nil
+	}
 	api := New(hasura.URL, hasura.Secret)
 
 	checkHealth(ctx, api)
 
-	metadata, err := Generate(hasura, cfg, models...)
+	metadata, err := Generate(*hasura, cfg, models...)
 	if err != nil {
 		return err
 	}

hasura/hasura.go

@@ -0,0 +1,93 @@
+package prometheus
+
+import (
+	"log"
+	"net/http"
+	"sync"
+
+	"github.com/dipdup-net/go-lib/config"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// Service -
+type Service struct {
+	counters map[string]*prometheus.CounterVec
+	server   *http.Server
+	wg       sync.WaitGroup
+}
+
+// NewService -
+func NewService(cfg *config.Prometheus) *Service {
+	var s Service
+	s.counters = make(map[string]*prometheus.CounterVec)
+
+	if cfg != nil && cfg.URL != "" {
+		s.server = &http.Server{Addr: cfg.URL}
+		http.Handle("/metrics", promhttp.Handler())
+	}
+
+	return &s
+}
+
+// Start -
+func (service *Service) Start() {
+	if service.server == nil {
+		return
+	}
+
+	service.wg.Add(1)
+	go func() {
+		defer service.wg.Done()
+
+		if err := service.server.ListenAndServe(); err != http.ErrServerClosed {
+			log.Fatalf("ListenAndServe(): %v", err)
+		}
+	}()
+}
+
+// Close -
+func (service *Service) Close() error {
+	if service.server != nil {
+		if err := service.server.Close(); err != nil {
+			return err
+		}
+	}
+
+	service.wg.Wait()
+
+	return nil
+}
+
+// RegisterCounter -
+func (service *Service) RegisterCounter(name, help string, labels ...string) {
+	vec := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: name,
+		Help: help,
+	}, labels)
+	service.counters[name] = vec
+	prometheus.MustRegister(vec)
+}
+
+// Counter -
+func (service *Service) Counter(name string) *prometheus.CounterVec {
+	counter, ok := service.counters[name]
+	if ok {
+		return counter
+	}
+	return nil
+}
+
+// IncrementCounter -
+func (service *Service) IncrementCounter(name string, labels map[string]string) {
+	counter, ok := service.counters[name]
+	if ok {
+		counter.With(labels).Inc()
+	}
+}
+
+// RegisterGoBuildMetrics -
+func (service *Service) RegisterGoBuildMetrics() {
+	prometheus.MustRegister(collectors.NewBuildInfoCollector())
+}

prometheus/prometheus.go

@@ -58,3 +58,8 @@ func (s *SignalR) Send(msg interface{}) error {
 func (s *SignalR) SetOnReconnect(onReconnect func() error) {
 	s.hub.onReconnect = onReconnect
 }
+
+// IsConnected -
+func (s *SignalR) IsConnected() bool {
+	return s.hub != nil && s.hub.conn != nil
+}

tzkt/events/signalr/signalr.go

@@ -31,7 +31,7 @@ func NewTzKT(url string) *TzKT {
 	return &TzKT{
 		s:             signalr.NewSignalR(url),
 		msgs:          make(chan Message, 1024),
-		stop:          make(chan struct{}),
+		stop:          make(chan struct{}, 1),
 		subscriptions: make([]signalr.Invocation, 0),
 	}
 }
@@ -59,6 +59,11 @@ func (tzkt *TzKT) Close() error {
 	return nil
 }
 
+// IsConnected - reports whether the connection to TzKT events is established
+func (tzkt *TzKT) IsConnected() bool {
+	return tzkt.s != nil && tzkt.s.IsConnected()
+}
+
 // Listen - listen channel with all received messages
 func (tzkt *TzKT) Listen() <-chan Message {
 	return tzkt.msgs