[prometheus] init pushgateway support

* add prometheus.gateway flag support
* instance: peerid
* job: network_type/shard

Signed-off-by: Leo Chen <leo@harmony.one>

api/service/prometheus/service.go

@@ -34,7 +34,7 @@ var (
 
 // NewService sets up a new instance for a given address host:port.
 // An empty host will match with any IP so an address like ":19000" is perfectly acceptable.
-func NewService(config nodeconfig.PrometheusServerConfig, additionalHandlers ...Handler) {
+func NewService(config nodeconfig.PrometheusServerConfig, pubkey string, additionalHandlers ...Handler) {
 	if !config.HTTPEnabled {
 		utils.Logger().Info().Msg("Prometheus http server disabled...")
 		return
@@ -55,6 +55,20 @@ func NewService(config nodeconfig.PrometheusServerConfig, additionalHandlers ...
 	endpoint := fmt.Sprintf("%s:%d", config.HTTPIp, config.HTTPPort)
 	svc.server = &http.Server{Addr: endpoint, Handler: mux}
 
+	// start pusher to push metrics to prometheus pushgateway
+	// every minute
+	go func(pubkey string) {
+		ticker := time.NewTicker(time.Minute)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				if err := utils.PromPusher(pubkey).Add(); err != nil {
+					utils.Logger().Warn().Err(err).Msg("Pushgateway Error")
+				}
+			}
+		}
+	}(pubkey)
 	svc.Start()
 }
 

cmd/harmony/config.go

@@ -146,6 +146,7 @@ type prometheusConfig struct {
 	Enabled bool
 	IP      string
 	Port    int
+	Gateway string
 }
 
 // TODO: use specific type wise validation instead of general string types assertion.

cmd/harmony/default.go

@@ -69,6 +69,7 @@ var defaultConfig = harmonyConfig{
 		Enabled: true,
 		IP:      "0.0.0.0",
 		Port:    nodeconfig.DefaultPrometheusPort,
+		Gateway: "https://gateway.harmony.one",
 	},
 }
 
@@ -102,6 +103,7 @@ var defaultPrometheusConfig = prometheusConfig{
 	Enabled: true,
 	IP:      "0.0.0.0",
 	Port:    9900,
+	Gateway: "https://gateway.harmony.one",
 }
 
 const (

cmd/harmony/flags.go

@@ -1204,6 +1204,11 @@ var (
 		Usage:    "prometheus port to listen for HTTP requests",
 		DefValue: defaultConfig.Prometheus.Port,
 	}
+	prometheusGatewayFlag = cli.StringFlag{
+		Name:     "prometheus.pushgateway",
+		Usage:    "prometheus pushgateway URL",
+		DefValue: defaultConfig.Prometheus.Gateway,
+	}
 )
 
 func applyPrometheusFlags(cmd *cobra.Command, config *harmonyConfig) {
@@ -1224,4 +1229,7 @@ func applyPrometheusFlags(cmd *cobra.Command, config *harmonyConfig) {
 		config.Prometheus.Enabled = cli.GetBoolFlagValue(cmd, prometheusEnabledFlag)
 	}
 
+	if cli.IsFlagChanged(cmd, prometheusGatewayFlag) {
+		config.Prometheus.Gateway = cli.GetStringFlagValue(cmd, prometheusGatewayFlag)
+	}
 }

cmd/harmony/flags_test.go

@@ -109,6 +109,7 @@ func TestHarmonyFlags(t *testing.T) {
 					Enabled: true,
 					IP:      "0.0.0.0",
 					Port:    9900,
+					Gateway: "https://gateway.harmony.one",
 				},
 			},
 		},

cmd/harmony/main.go

@@ -395,7 +395,7 @@ func setupNodeAndRun(hc harmonyConfig) {
 			Msg("Start Rosetta failed")
 	}
 
-	if err := currentNode.StartPrometheus(); err != nil {
+	if err := currentNode.StartPrometheus(nodeconfig.GetDefaultConfig().ConsensusPriKey.GetPublicKeys().SerializeToHexStr()); err != nil {
 		utils.Logger().Warn().
 			Err(err).
 			Msg("Start Prometheus failed")

consensus/consensus.go

@@ -222,5 +222,8 @@ func New(
 	// Make Sure Verifier is not null
 	consensus.vc = newViewChange()
 
+	// init prometheus metrics
+	initMetrics()
+
 	return &consensus, nil
 }

consensus/metrics.go

@@ -1,13 +1,13 @@
 package consensus
 
 import (
+	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
 var (
 	// consensusCounterVec is used to keep track of consensus reached
-	consensusCounterVec = promauto.NewCounterVec(
+	consensusCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "consensus",
@@ -19,7 +19,7 @@ var (
 		},
 	)
 	// consensusVCCounterVec is used to keep track of number of view change
-	consensusVCCounterVec = promauto.NewCounterVec(
+	consensusVCCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "consensus",
@@ -31,7 +31,7 @@ var (
 		},
 	)
 	// consensusSyncCounterVec is used to keep track of consensus syncing state
-	consensusSyncCounterVec = promauto.NewCounterVec(
+	consensusSyncCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "consensus",
@@ -43,7 +43,7 @@ var (
 		},
 	)
 	// consensusGaugeVec is used to keep track of gauge number of the consensus
-	consensusGaugeVec = promauto.NewGaugeVec(
+	consensusGaugeVec = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "hmy",
 			Subsystem: "consensus",
@@ -57,7 +57,7 @@ var (
 	// consensusFinalityHistogram is used to keep track of finality
 	// 10 ExponentialBuckets are in the unit of millisecond:
 	// 800, 1000, 1250, 1562, 1953, 2441, 3051, 3814, 4768, 5960, inf
-	consensusFinalityHistogram = promauto.NewHistogram(
+	consensusFinalityHistogram = prometheus.NewHistogram(
 		prometheus.HistogramOpts{
 			Namespace: "hmy",
 			Subsystem: "consensus",
@@ -66,6 +66,9 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(800, 1.25, 10),
 		},
 	)
+
+	// TODO: add last consensus timestamp, add view ID
+	// add last view change timestamp
 )
 
 // UpdateValidatorMetrics will udpate validator metrics
@@ -83,3 +86,13 @@ func (consensus *Consensus) UpdateLeaderMetrics(numCommits float64, blockNum flo
 	consensusCounterVec.With(prometheus.Labels{"consensus": "num_commits"}).Add(numCommits)
 	consensusGaugeVec.With(prometheus.Labels{"consensus": "num_commits"}).Set(numCommits)
 }
+
+func initMetrics() {
+	utils.PromRegistry().MustRegister(
+		consensusCounterVec,
+		consensusVCCounterVec,
+		consensusSyncCounterVec,
+		consensusGaugeVec,
+		consensusFinalityHistogram,
+	)
+}

internal/utils/prometheus.go

@@ -0,0 +1,34 @@
+package utils
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/push"
+	"sync"
+)
+
+var (
+	// Prometheus Pusher
+	onceForPusher sync.Once
+	pusher        *push.Pusher
+	registry      *prometheus.Registry
+)
+
+// Pusher returns the pusher, initialized once only
+func PromPusher(pubkey string) *push.Pusher {
+	onceForPusher.Do(func() {
+		if registry == nil {
+			registry = prometheus.NewRegistry()
+		}
+		pusher = push.New("https://gateway.harmony.one", "hmy").
+			Gatherer(registry).
+			Grouping("instance", pubkey)
+	})
+	return pusher
+}
+
+func PromRegistry() *prometheus.Registry {
+	if registry == nil {
+		registry = prometheus.NewRegistry()
+	}
+	return registry
+}

node/api.go

@@ -83,8 +83,8 @@ func (node *Node) StopRPC() error {
 }
 
 // StartPrometheus start promtheus metrics service
-func (node *Node) StartPrometheus() error {
-	prometheus.NewService(node.NodeConfig.PrometheusServer)
+func (node *Node) StartPrometheus(pubkey string) error {
+	prometheus.NewService(node.NodeConfig.PrometheusServer, pubkey)
 	return nil
 }
 

node/metrics.go

@@ -1,14 +1,14 @@
 package node
 
 import (
+	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
 var (
 	// NodeStringCounterVec is used to add version string or other static string
 	// info into the metrics api
-	NodeStringCounterVec = promauto.NewCounterVec(
+	NodeStringCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "node",
@@ -18,7 +18,7 @@ var (
 		[]string{"key", "value"},
 	)
 	// NodeP2PMessageCounterVec is used to keep track of all p2p messages received
-	NodeP2PMessageCounterVec = promauto.NewCounterVec(
+	NodeP2PMessageCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "p2p",
@@ -30,7 +30,7 @@ var (
 		},
 	)
 	// NodeConsensusMessageCounterVec is used to keep track of consensus p2p messages received
-	NodeConsensusMessageCounterVec = promauto.NewCounterVec(
+	NodeConsensusMessageCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "p2p",
@@ -43,7 +43,7 @@ var (
 	)
 
 	// NodeNodeMessageCounterVec is used to keep track of node p2p messages received
-	NodeNodeMessageCounterVec = promauto.NewCounterVec(
+	NodeNodeMessageCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "hmy",
 			Subsystem: "p2p",
@@ -55,3 +55,12 @@ var (
 		},
 	)
 )
+
+func initMetrics() {
+	utils.PromRegistry().MustRegister(
+		NodeStringCounterVec,
+		NodeP2PMessageCounterVec,
+		NodeConsensusMessageCounterVec,
+		NodeNodeMessageCounterVec,
+	)
+}

node/node.go

@@ -988,6 +988,7 @@ func New(
 	}
 
 	// init metrics
+	initMetrics()
 	NodeStringCounterVec.WithLabelValues("version", nodeconfig.GetVersion()).Inc()
 
 	return &node