p2p/protocols: first version of accounting metrics persistance

p2p/protocols/accounting.go

@@ -20,30 +20,29 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/syndtr/goleveldb/leveldb"
 )
 
 //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
-	mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", nil)
+	mBalanceCredit metrics.Counter
 	//total amount of units debited
-	mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", nil)
+	mBalanceDebit metrics.Counter
 	//total amount of bytes credited
-	mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", nil)
+	mBytesCredit metrics.Counter
 	//total amount of bytes debited
-	mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", nil)
+	mBytesDebit metrics.Counter
 	//total amount of credited messages
-	mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", nil)
+	mMsgCredit metrics.Counter
 	//total amount of debited messages
-	mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", nil)
+	mMsgDebit metrics.Counter
 	//how many times local node had to drop remote peers
-	mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", nil)
+	mPeerDrops metrics.Counter
 	//how many times local node overdrafted and dropped
-	mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", nil)
+	mSelfDrops metrics.Counter
 )
 
 //Prices defines how prices are being passed on to the accounting instance
@@ -110,6 +109,20 @@ func NewAccounting(balance Balance, po Prices) *Accounting {
 	return ah
 }
 
+func SetupAccountingMetrics(reportInterval time.Duration, path string) *leveldb.DB {
+	registry := metrics.NewRegistry()
+	mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", registry)
+	mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", registry)
+	mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", registry)
+	mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", registry)
+	mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", registry)
+	mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", registry)
+	mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", registry)
+	mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", registry)
+
+	return NewMetricsDB(registry, reportInterval, path)
+}
+
 //Implement Hook.Send
 // Send takes a peer, a size and a msg and
 // - calculates the cost for the local node sending a msg of size to peer using the Prices interface
@@ -157,9 +170,6 @@ 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))
@@ -178,8 +188,3 @@ 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

@@ -17,32 +17,74 @@
 package protocols
 
 import (
+	"encoding/binary"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/swarm/state"
-	"github.com/rcrowley/go-metrics"
+	"github.com/ethereum/go-ethereum/metrics"
+
+	"github.com/syndtr/goleveldb/leveldb"
 )
 
 type reporter struct {
 	reg      metrics.Registry
 	interval time.Duration
-	stateStore *state.DBStore
+	db       *leveldb.DB
 }
 
-func NewMetricsStateStore(r metrics.Registry, d time.Duration, path string) {
-	stateStore, err := state.NewDBStore(path)
+func NewMetricsDB(r metrics.Registry, d time.Duration, path string) *leveldb.DB {
+	var val = make([]byte, 8)
+	var err error
+
+	db, err := leveldb.OpenFile(path, nil)
 	if err != nil {
-		return
+		log.Error(err.Error())
+		return nil
 	}
 
-	rep := &reporter{
+	val, err = db.Get([]byte("account.balance.credit"), nil)
+	if err == nil {
+		mBalanceCredit.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.balance.debit"), nil)
+	if err == nil {
+		mBalanceDebit.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.bytes.credit"), nil)
+	if err == nil {
+		mBytesCredit.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.bytes.debit"), nil)
+	if err == nil {
+		mBytesDebit.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.msg.credit"), nil)
+	if err == nil {
+		mMsgCredit.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.msg.debit"), nil)
+	if err == nil {
+		mMsgDebit.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.peerdrops"), nil)
+	if err == nil {
+		mPeerDrops.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+	val, err = db.Get([]byte("account.selfdrops"), nil)
+	if err == nil {
+		mSelfDrops.Inc(int64(binary.BigEndian.Uint64(val)))
+	}
+
+	reg := &reporter{
 		reg:      r,
 		interval: d,
-		stateStore: stateStore,
+		db:       db,
 	}
 
-	rep.run()
+	go reg.run()
+
+	return db
+
 }
 
 func (r *reporter) run() {
@@ -50,20 +92,21 @@ func (r *reporter) run() {
 
 	for _ = range intervalTicker.C {
 		if err := r.send(); err != nil {
-			log.Error("unable to send metrics to InfluxDB. err=%v", err)
+			log.Error("unable to send metrics to LevelDB. err=%v", "err", err)
+			return
 		}
 	}
 }
 
 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())
+			byteVal := make([]byte, 8)
+			binary.BigEndian.PutUint64(byteVal, uint64(ms.Count()))
+			err = r.db.Put([]byte(name), byteVal, nil)
 		}
 	})
 

p2p/protocols/reporter_test.go

@@ -21,54 +21,41 @@ import (
 	"testing"
 	"time"
 
-	"github.com/ethereum/go-ethereum/swarm/state"
-	"github.com/rcrowley/go-metrics"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 func TestReporter(t *testing.T) {
 	dir := os.TempDir()
 	defer os.RemoveAll(dir)
 
-	stateStore, err := state.NewDBStore(dir + "/test.db")
-	if err != nil {
-		return
-	}
+	log.Debug("Setting up metrics first time")
+	reportInterval := 100 * time.Millisecond
+	db := SetupAccountingMetrics(reportInterval, dir+"/test.db")
+	log.Debug("Done.")
 
-	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)
+	//give the reporter time to write to DB
+	time.Sleep(500 * time.Millisecond)
 
-	if mBalanceCredit.Count() != 23 {
-		t.Fatalf("Expected counter to be %d, but is %d", 23, mBalanceCredit.Count())
+	mBalanceCredit = nil
+	mBytesCredit = nil
+	mMsgDebit = nil
+	db.Close()
+
+	log.Debug("Setting up metrics second time")
+	SetupAccountingMetrics(reportInterval, dir+"/test.db")
+	log.Debug("Done.")
+
+	if mBalanceCredit.Count() != 12 {
+		t.Fatalf("Expected counter to be %d, but is %d", 12, mBalanceCredit.Count())
 	}
-	if mBytesCredit.Count() != 56 {
+	if mBytesCredit.Count() != 34 {
 		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())
+	if mMsgDebit.Count() != 9 {
+		t.Fatalf("Expected counter to be %d, but is %d", 9, mMsgDebit.Count())
 	}
 }

swarm/swarm.go

@@ -179,6 +179,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 			return nil, err
 		}
 		self.swap = swap.New(balancesStore)
+		protocols.SetupAccountingMetrics(10 *time.Second, filepath.Join(config.Path, "metrics.db"))
 	}
 
 	var nodeID enode.ID