From deb95632f87813fa5f58ecc0fc2ab4ea86e56e7f Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Fri, 16 Nov 2018 12:48:42 -0500 Subject: [PATCH] p2p/protocols: first version of accounting metrics persistance --- p2p/protocols/accounting.go | 41 ++++++++++-------- p2p/protocols/reporter.go | 77 ++++++++++++++++++++++++++-------- p2p/protocols/reporter_test.go | 55 ++++++++++-------------- swarm/swarm.go | 1 + 4 files changed, 105 insertions(+), 69 deletions(-) diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go index 1266fbb116..1e0f2674bf 100644 --- a/p2p/protocols/accounting.go +++ b/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 -} diff --git a/p2p/protocols/reporter.go b/p2p/protocols/reporter.go index 11663a9bba..91daf2a0d6 100644 --- a/p2p/protocols/reporter.go +++ b/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 + reg metrics.Registry + interval time.Duration + 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{ - reg: r, - interval: d, - stateStore: stateStore, + 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))) } - rep.run() + reg := &reporter{ + reg: r, + interval: d, + db: db, + } + + 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) } }) diff --git a/p2p/protocols/reporter_test.go b/p2p/protocols/reporter_test.go index c12455095c..8f886df43f 100644 --- a/p2p/protocols/reporter_test.go +++ b/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()) } } diff --git a/swarm/swarm.go b/swarm/swarm.go index dc3756d3af..e272a76813 100644 --- a/swarm/swarm.go +++ b/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