p2p/protocols: added reporter for cross-session accounting metrics

p2p/protocols/accounting.go

@@ -16,11 +16,16 @@
 
 package protocols
 
-import "github.com/ethereum/go-ethereum/metrics"
+import (
+	"time"
+
+	"github.com/ethereum/go-ethereum/metrics"
+)
 
 //define some metrics
 var (
 	//NOTE: these metrics just define the interfaces and are currently *NOT persisted* over sessions
+	metricsInitialized bool
 	//All metrics are cumulative
 
 	//total amount of units credited
@@ -152,6 +157,9 @@ func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error {
 // * if the price is positive, local node has been credited; thus `err` implicitly signals the REMOTE has been dropped
 // * if the price is negative, local node has been debited, thus `err` implicitly signals LOCAL node "overdraft"
 func (ah *Accounting) doMetrics(price int64, size uint32, err error) {
+	if !metricsInitialized {
+		initMetrics()
+	}
 	if price > 0 {
 		mBalanceCredit.Inc(price)
 		mBytesCredit.Inc(int64(size))
@@ -170,3 +178,8 @@ func (ah *Accounting) doMetrics(price int64, size uint32, err error) {
 		}
 	}
 }
+
+func initMetrics() {
+	NewMetricsStateStore(metrics.DefaultRegistry, 10*time.Second, "metrics.db")
+	metricsInitialized = true
+}

p2p/protocols/reporter.go

@@ -0,0 +1,71 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package protocols
+
+import (
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/swarm/state"
+	"github.com/rcrowley/go-metrics"
+)
+
+type reporter struct {
+	reg        metrics.Registry
+	interval   time.Duration
+	stateStore *state.DBStore
+}
+
+func NewMetricsStateStore(r metrics.Registry, d time.Duration, path string) {
+	stateStore, err := state.NewDBStore(path)
+	if err != nil {
+		return
+	}
+
+	rep := &reporter{
+		reg:        r,
+		interval:   d,
+		stateStore: stateStore,
+	}
+
+	rep.run()
+}
+
+func (r *reporter) run() {
+	intervalTicker := time.NewTicker(r.interval)
+
+	for _ = range intervalTicker.C {
+		if err := r.send(); err != nil {
+			log.Error("unable to send metrics to InfluxDB. err=%v", err)
+		}
+	}
+}
+
+func (r *reporter) send() error {
+
+	var err error
+
+	r.reg.Each(func(name string, i interface{}) {
+		switch metric := i.(type) {
+		case metrics.Counter:
+			ms := metric.Snapshot()
+			err = r.stateStore.Put(name, ms.Count())
+		}
+	})
+
+	return err
+}

p2p/protocols/reporter_test.go

@@ -0,0 +1,74 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package protocols
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/swarm/state"
+	"github.com/rcrowley/go-metrics"
+)
+
+func TestReporter(t *testing.T) {
+	dir := os.TempDir()
+	defer os.RemoveAll(dir)
+
+	stateStore, err := state.NewDBStore(dir + "/test.db")
+	if err != nil {
+		return
+	}
+
+	rep := &reporter{
+		reg:        metrics.NewRegistry(),
+		interval:   time.Millisecond,
+		stateStore: stateStore,
+	}
+	go rep.run()
+	time.Sleep(1 * time.Second)
+	mBalanceCredit.Inc(12)
+	mBytesCredit.Inc(34)
+	mMsgDebit.Inc(9)
+
+	rep = nil
+	stateStore.Close()
+	stateStore, err = state.NewDBStore(dir + "/test.db")
+	if err != nil {
+		return
+	}
+	rep = &reporter{
+		reg:        metrics.NewRegistry(),
+		interval:   time.Millisecond,
+		stateStore: stateStore,
+	}
+	go rep.run()
+	time.Sleep(1 * time.Second)
+	mBalanceCredit.Inc(11)
+	mBytesCredit.Inc(22)
+	mMsgDebit.Inc(7)
+
+	if mBalanceCredit.Count() != 23 {
+		t.Fatalf("Expected counter to be %d, but is %d", 23, mBalanceCredit.Count())
+	}
+	if mBytesCredit.Count() != 56 {
+		t.Fatalf("Expected counter to be %d, but is %d", 23, mBytesCredit.Count())
+	}
+	if mMsgDebit.Count() != 16 {
+		t.Fatalf("Expected counter to be %d, but is %d", 23, mMsgDebit.Count())
+	}
+}