Metrics module

cmd/start.go

@@ -5,6 +5,9 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/keep-network/keep-core/pkg/metrics"
+	"github.com/keep-network/keep-core/pkg/net"
+
 	"github.com/ipfs/go-log"
 	"github.com/keep-network/keep-common/pkg/chain/ethereum/ethutil"
 	"github.com/keep-network/keep-common/pkg/persistence"
@@ -150,6 +153,11 @@ func Start(c *cli.Context) error {
 		return fmt.Errorf("error initializing beacon: [%v]", err)
 	}
 
+	err = initializeMetrics(ctx, config, netProvider, stakeMonitor)
+	if err != nil {
+		return fmt.Errorf("error initializing metrics: [%v]", err)
+	}
+
 	select {
 	case <-ctx.Done():
 		if err != nil {
@@ -195,3 +203,53 @@ func waitForStake(stakeMonitor chain.StakeMonitor, address string, timeout int)
 	}
 	return fmt.Errorf("timed out waiting for %s to have required minimum stake", address)
 }
+
+func initializeMetrics(
+	ctx context.Context,
+	config *config.Config,
+	netProvider net.Provider,
+	stakeMonitor chain.StakeMonitor,
+) error {
+	registry, isConfigured := metrics.Initialize(
+		config.Metrics.Port,
+	)
+	if !isConfigured {
+		logger.Infof("metrics are not configured")
+		return nil
+	}
+
+	logger.Infof(
+		"enabled metrics on port [%v]",
+		config.Metrics.Port,
+	)
+
+	metrics.ObserveConnectedPeersCount(
+		ctx,
+		registry,
+		netProvider,
+		time.Duration(config.Metrics.NetworkMetricsTick)*time.Second,
+	)
+
+	metrics.ObserveConnectedBootstrapCount(
+		ctx,
+		registry,
+		netProvider,
+		config.LibP2P.Peers,
+		time.Duration(config.Metrics.NetworkMetricsTick)*time.Second,
+	)
+
+	metrics.ObserveEthConnectivity(
+		ctx,
+		registry,
+		stakeMonitor,
+		config.Ethereum.Account.Address,
+		time.Duration(config.Metrics.EthereumMetricsTick)*time.Second,
+	)
+
+	metrics.ExposeLibP2PInfo(
+		registry,
+		netProvider,
+	)
+
+	return nil
+}

config/config.go

@@ -19,13 +19,21 @@ type Config struct {
 	Ethereum ethereum.Config
 	LibP2P   libp2p.Config
 	Storage  Storage
+	Metrics  Metrics
 }
 
 // Storage stores meta-info about keeping data on disk
 type Storage struct {
 	DataDir string
 }
 
+// Metrics stores meta-info about metrics.
+type Metrics struct {
+	Port                int
+	NetworkMetricsTick  int
+	EthereumMetricsTick int
+}
+
 var (
 	// KeepOpts contains global application settings
 	KeepOpts Config

configs/config.toml.SAMPLE

@@ -54,3 +54,8 @@
 
 [Storage]
   DataDir = "/my/secure/location"
+
+# [Metrics]
+    # Port = 8080
+    # NetworkMetricsTick = 60
+    # EthereumMetricsTick = 600

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/ipfs/go-datastore v0.1.1
 	github.com/ipfs/go-log v0.0.1
 	github.com/keep-network/go-libp2p-bootstrap v0.0.0-20200423153828-ed815bc50aec
-	github.com/keep-network/keep-common v1.1.0
+	github.com/keep-network/keep-common v1.1.1-0.20200703125023-d9872a19ebd1
 	github.com/libp2p/go-addr-util v0.0.1
 	github.com/libp2p/go-libp2p v0.4.1
 	github.com/libp2p/go-libp2p-connmgr v0.1.0

go.sum

@@ -248,6 +248,20 @@ github.com/keep-network/go-libp2p-bootstrap v0.0.0-20200423153828-ed815bc50aec h
 github.com/keep-network/go-libp2p-bootstrap v0.0.0-20200423153828-ed815bc50aec/go.mod h1:xR8jf3/VJAjh3nWu5tFe8Yxnt2HvWsqZHfGef1P5oDk=
 github.com/keep-network/keep-common v1.1.0 h1:m5ZDfUpH+DVqQz3qIi+E53utWHv7kVSooPD01kVG3n8=
 github.com/keep-network/keep-common v1.1.0/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200610113255-c9d58a9011ff h1:MhKVpUYCChwjKFoDt4G7aBwX4B+d+lxQYt6keam2OKk=
+github.com/keep-network/keep-common v1.1.1-0.20200610113255-c9d58a9011ff/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200610130035-55afd4237caf h1:bNOzvCwlxnWANqoUzN+FWVKfzToA86EZrfLquERwcUY=
+github.com/keep-network/keep-common v1.1.1-0.20200610130035-55afd4237caf/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200612104554-481df8250e84 h1:EgCeKoy6zRQEzZWi/z7Z5IQ1/5imb9Un/ucuJNqjCL0=
+github.com/keep-network/keep-common v1.1.1-0.20200612104554-481df8250e84/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200612111801-12829c0d1e0f h1:m0uA5CD9h/aNKTn4Z5r760D/gggBIqFjPSdcE4KGWKA=
+github.com/keep-network/keep-common v1.1.1-0.20200612111801-12829c0d1e0f/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200612121439-e1944b162625 h1:ZDb+ewPufSqbbO49hTPotyO8RMzVODcOPxGWm8xzHeg=
+github.com/keep-network/keep-common v1.1.1-0.20200612121439-e1944b162625/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200701100953-c1c03e75403b h1:sQfkPbDdCp8zTp486gIG/LrpMJscbJ4CQvsYSuGg81I=
+github.com/keep-network/keep-common v1.1.1-0.20200701100953-c1c03e75403b/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
+github.com/keep-network/keep-common v1.1.1-0.20200703125023-d9872a19ebd1 h1:SCjStilprtxkLaY6+pw/xqvCJFPEm71n3GJyC+cbLU0=
+github.com/keep-network/keep-common v1.1.1-0.20200703125023-d9872a19ebd1/go.mod h1:0uY+hufkP66nFnL+3GXMwWXRiKKsFyXFQTby1xR7AwY=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=

pkg/metrics/metrics.go

@@ -0,0 +1,157 @@
+package metrics
+
+import (
+	"context"
+	"time"
+
+	"github.com/ipfs/go-log"
+	"github.com/keep-network/keep-common/pkg/metrics"
+	"github.com/keep-network/keep-core/pkg/chain"
+	"github.com/keep-network/keep-core/pkg/net"
+)
+
+var logger = log.Logger("keep-metrics")
+
+const (
+	// DefaultNetworkMetricsTick is the default duration of the
+	// observation tick for network metrics.
+	DefaultNetworkMetricsTick = 1 * time.Minute
+	// DefaultEthereumMetricsTick is the default duration of the
+	// observation tick for Ethereum metrics.
+	DefaultEthereumMetricsTick = 10 * time.Minute
+)
+
+// Initialize set up the metrics registry and enables metrics server.
+func Initialize(
+	port int,
+) (*metrics.Registry, bool) {
+	if port == 0 {
+		return nil, false
+	}
+
+	registry := metrics.NewRegistry()
+
+	registry.EnableServer(port)
+
+	return registry, true
+}
+
+// ObserveConnectedPeersCount triggers an observation process of the
+// connected_peers_count metric.
+func ObserveConnectedPeersCount(
+	ctx context.Context,
+	registry *metrics.Registry,
+	netProvider net.Provider,
+	tick time.Duration,
+) {
+	input := func() float64 {
+		connectedPeers := netProvider.ConnectionManager().ConnectedPeers()
+		return float64(len(connectedPeers))
+	}
+
+	observe(
+		ctx,
+		"connected_peers_count",
+		input,
+		registry,
+		validateTick(tick, DefaultNetworkMetricsTick),
+	)
+}
+
+// ObserveConnectedBootstrapCount triggers an observation process of the
+// connected_bootstrap_count metric.
+func ObserveConnectedBootstrapCount(
+	ctx context.Context,
+	registry *metrics.Registry,
+	netProvider net.Provider,
+	bootstraps []string,
+	tick time.Duration,
+) {
+	input := func() float64 {
+		currentCount := 0
+
+		for _, address := range bootstraps {
+			if netProvider.ConnectionManager().IsConnected(address) {
+				currentCount++
+			}
+		}
+
+		return float64(currentCount)
+	}
+
+	observe(
+		ctx,
+		"connected_bootstrap_count",
+		input,
+		registry,
+		validateTick(tick, DefaultNetworkMetricsTick),
+	)
+}
+
+// ObserveEthConnectivity triggers an observation process of the
+// eth_connectivity metric.
+func ObserveEthConnectivity(
+	ctx context.Context,
+	registry *metrics.Registry,
+	stakeMonitor chain.StakeMonitor,
+	address string,
+	tick time.Duration,
+) {
+	input := func() float64 {
+		_, err := stakeMonitor.HasMinimumStake(address)
+
+		if err != nil {
+			return 0
+		}
+
+		return 1
+	}
+
+	observe(
+		ctx,
+		"eth_connectivity",
+		input,
+		registry,
+		validateTick(tick, DefaultEthereumMetricsTick),
+	)
+}
+
+func observe(
+	ctx context.Context,
+	name string,
+	input metrics.ObserverInput,
+	registry *metrics.Registry,
+	tick time.Duration,
+) {
+	observer, err := registry.NewGaugeObserver(name, input)
+	if err != nil {
+		logger.Warningf("could not create gauge observer [%v]", name)
+		return
+	}
+
+	observer.Observe(ctx, tick)
+}
+
+func validateTick(tick time.Duration, defaultTick time.Duration) time.Duration {
+	if tick > 0 {
+		return tick
+	}
+
+	return defaultTick
+}
+
+// ExposeLibP2PInfo provides some basic information about libp2p config.
+func ExposeLibP2PInfo(
+	registry *metrics.Registry,
+	netProvider net.Provider,
+) {
+	name := "libp2p_info"
+
+	id := metrics.NewLabel("id", netProvider.ID().String())
+
+	_, err := registry.NewInfo(name, []metrics.Label{id})
+	if err != nil {
+		logger.Warningf("could not create info metric [%v]", name)
+		return
+	}
+}

pkg/net/libp2p/libp2p.go

@@ -216,6 +216,15 @@ func (cm *connectionManager) AddrStrings() []string {
 	return multiaddrStrings
 }
 
+func (cm *connectionManager) IsConnected(address string) bool {
+	peerInfos, err := extractMultiAddrFromPeers([]string{address})
+	if err != nil {
+		return false
+	}
+
+	return cm.Network().Connectedness(peerInfos[0].ID) == libp2pnet.Connected
+}
+
 func (cm *connectionManager) monitorConnectedPeers(ctx context.Context) {
 	ticker := time.NewTicker(ConnectedPeersCheckTick)
 	defer ticker.Stop()

pkg/net/local/local.go

@@ -135,3 +135,7 @@ func (lcm *localConnectionManager) DisconnectPeer(connectedPeer string) {
 func (lcm *localConnectionManager) AddrStrings() []string {
 	return make([]string, 0)
 }
+
+func (lcm *localConnectionManager) IsConnected(address string) bool {
+	panic("not implemented")
+}

pkg/net/net.go

@@ -77,6 +77,8 @@ type ConnectionManager interface {
 
 	// AddrStrings returns all listen addresses of the provider.
 	AddrStrings() []string
+
+	IsConnected(address string) bool
 }
 
 // TaggedUnmarshaler is an interface that includes the proto.Unmarshaler