metrics: rework prometheus integration

The peers_total metric used to live deep down in libp2p,
it is now nicely separated, and there's space for even
more metrics. Tests are a good thing too.

The Prometheus library is now gx'ed too.

License: MIT
Signed-off-by: Lars Gierth <[email protected]>

cmd/ipfs/daemon.go

@@ -23,9 +23,10 @@ import (
 	"github.com/ipfs/go-ipfs/core/corerouting"
 	nodeMount "github.com/ipfs/go-ipfs/fuse/node"
 	fsrepo "github.com/ipfs/go-ipfs/repo/fsrepo"
-	util "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
+	prometheus "gx/ipfs/QmSH7Koc3CCaB8weo9aDE7YWQuGwAi5G7bqGvxatfux1Y5/prometheus"
 	conn "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/net/conn"
 	peer "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/peer"
+	util "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
 )
 
 const (
@@ -302,6 +303,10 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
 		return
 	}
 
+	// initialize metrics collector
+	prometheus.MustRegisterOrGet(&corehttp.IpfsNodeCollector{Node: node})
+	prometheus.EnableCollectChecks(true)
+
 	fmt.Printf("Daemon is ready\n")
 	// collect long-running errors and block for shutdown
 	// TODO(cryptix): our fuse currently doesnt follow this pattern for graceful shutdown
@@ -364,15 +369,15 @@ func serveHTTPApi(req cmds.Request) (error, <-chan error) {
 		},
 	})
 	var opts = []corehttp.ServeOption{
-		corehttp.PrometheusCollectorOption("api"),
+		corehttp.MetricsCollectionOption("api"),
 		corehttp.CommandsOption(*req.InvocContext()),
 		corehttp.WebUIOption,
 		apiGw.ServeOption(),
 		corehttp.VersionOption(),
 		defaultMux("/debug/vars"),
 		defaultMux("/debug/pprof/"),
+		corehttp.MetricsScrapingOption("/debug/metrics/prometheus"),
 		corehttp.LogOption(),
-		corehttp.PrometheusOption("/debug/metrics/prometheus"),
 	}
 
 	if len(cfg.Gateway.RootRedirect) > 0 {
@@ -443,7 +448,7 @@ func serveHTTPGateway(req cmds.Request) (error, <-chan error) {
 	}
 
 	var opts = []corehttp.ServeOption{
-		corehttp.PrometheusCollectorOption("gateway"),
+		corehttp.MetricsCollectionOption("gateway"),
 		corehttp.CommandsROOption(*req.InvocContext()),
 		corehttp.VersionOption(),
 		corehttp.IPNSHostnameOption(),

core/corehttp/metrics.go

@@ -0,0 +1,53 @@
+package corehttp
+
+import (
+	"net"
+	"net/http"
+
+	prometheus "gx/ipfs/QmSH7Koc3CCaB8weo9aDE7YWQuGwAi5G7bqGvxatfux1Y5/prometheus"
+
+	core "github.com/ipfs/go-ipfs/core"
+)
+
+// This adds the scraping endpoint which Prometheus uses to fetch metrics.
+func MetricsScrapingOption(path string) ServeOption {
+	return func(n *core.IpfsNode, _ net.Listener, mux *http.ServeMux) (*http.ServeMux, error) {
+		mux.Handle(path, prometheus.UninstrumentedHandler())
+		return mux, nil
+	}
+}
+
+// This adds collection of net/http-related metrics
+func MetricsCollectionOption(handlerName string) ServeOption {
+	return func(_ *core.IpfsNode, _ net.Listener, mux *http.ServeMux) (*http.ServeMux, error) {
+		childMux := http.NewServeMux()
+		mux.HandleFunc("/", prometheus.InstrumentHandler(handlerName, childMux))
+		return childMux, nil
+	}
+}
+
+var (
+	peersTotalMetric = prometheus.NewDesc(
+		prometheus.BuildFQName("ipfs", "p2p", "peers_total"),
+		"Number of connected peers", nil, nil)
+)
+
+type IpfsNodeCollector struct {
+	Node *core.IpfsNode
+}
+
+func (_ IpfsNodeCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- peersTotalMetric
+}
+
+func (c IpfsNodeCollector) Collect(ch chan<- prometheus.Metric) {
+	ch <- prometheus.MustNewConstMetric(
+		peersTotalMetric,
+		prometheus.GaugeValue,
+		c.PeersTotalValue(),
+	)
+}
+
+func (c IpfsNodeCollector) PeersTotalValue() float64 {
+	return float64(len(c.Node.PeerHost.Network().Conns()))
+}

core/corehttp/metrics_test.go

@@ -0,0 +1,46 @@
+package corehttp
+
+import (
+	"testing"
+	"time"
+
+	core "github.com/ipfs/go-ipfs/core"
+	context "golang.org/x/net/context"
+	bhost "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/host/basic"
+	inet "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/net"
+	testutil "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/test/util"
+)
+
+// This test is based on go-libp2p/p2p/net/swarm.TestConnectednessCorrect
+// It builds 4 nodes and connects them, one being the sole center.
+// Then it checks that the center reports the correct number of peers.
+func TestPeersTotal(t *testing.T) {
+	ctx := context.Background()
+
+	hosts := make([]*bhost.BasicHost, 4)
+	for i := 0; i < 4; i++ {
+		hosts[i] = testutil.GenHostSwarm(t, ctx)
+	}
+
+	dial := func(a, b inet.Network) {
+		testutil.DivulgeAddresses(b, a)
+		if _, err := a.DialPeer(ctx, b.LocalPeer()); err != nil {
+			t.Fatalf("Failed to dial: %s", err)
+		}
+	}
+
+	dial(hosts[0].Network(), hosts[1].Network())
+	dial(hosts[0].Network(), hosts[2].Network())
+	dial(hosts[0].Network(), hosts[3].Network())
+
+	// there's something wrong with dial, i think. it's not finishing
+	// completely. there must be some async stuff.
+	<-time.After(100 * time.Millisecond)
+
+	node := &core.IpfsNode{PeerHost: hosts[0]}
+	collector := IpfsNodeCollector{Node: node}
+	actual := collector.PeersTotalValue()
+	if actual != 3 {
+		t.Fatalf("expected 3 peers, got %d", int(actual))
+	}
+}

package.json

@@ -31,6 +31,11 @@
       "name": "gogo-protobuf",
       "hash": "QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV",
       "version": "0.0.0"
+    },
+    {
+      "name": "prometheus",
+      "hash": "QmSH7Koc3CCaB8weo9aDE7YWQuGwAi5G7bqGvxatfux1Y5",
+      "version": "0.7.0"
     }
   ],
   "language": "go",