From 798fbd054f5fce7242854b89485e0dc106223ad8 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Fri, 16 Nov 2018 14:38:43 -0500 Subject: [PATCH] p2p/protocols: added cleanup for DB and comments --- p2p/protocols/accounting.go | 8 ++++++- p2p/protocols/reporter.go | 28 ++++++++++++++++++---- p2p/protocols/reporter_test.go | 23 ++++++++++++++---- swarm/swap/swap.go | 5 ++++ swarm/swarm.go | 44 +++++++++++++++++++++------------- 5 files changed, 83 insertions(+), 25 deletions(-) diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go index 1e0f2674bf..f366bd6aa6 100644 --- a/p2p/protocols/accounting.go +++ b/p2p/protocols/accounting.go @@ -109,8 +109,14 @@ func NewAccounting(balance Balance, po Prices) *Accounting { return ah } +//SetupAccountingMetrics creates a separate registry for p2p accounting metrics; +//this registry should be independent of any other metrics as it persists at different endpoints. +//It also instantiates the given metrics and starts the persisting go-routine which +//at the passed interval writes the metrics to a LevelDB func SetupAccountingMetrics(reportInterval time.Duration, path string) *leveldb.DB { + //create an empty registry registry := metrics.NewRegistry() + //instantiate the metrics mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", registry) mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", registry) mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", registry) @@ -119,7 +125,7 @@ func SetupAccountingMetrics(reportInterval time.Duration, path string) *leveldb. mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", registry) mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", registry) mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", registry) - + //create the DB and start persisting return NewMetricsDB(registry, reportInterval, path) } diff --git a/p2p/protocols/reporter.go b/p2p/protocols/reporter.go index 91daf2a0d6..eea17b50cc 100644 --- a/p2p/protocols/reporter.go +++ b/p2p/protocols/reporter.go @@ -26,22 +26,30 @@ import ( "github.com/syndtr/goleveldb/leveldb" ) +//reporter is an internal structure used to write p2p accounting related +//metrics to a LevelDB. It will periodically write the accrued metrics to the DB. type reporter struct { - reg metrics.Registry - interval time.Duration - db *leveldb.DB + reg metrics.Registry //the registry for these metrics (independent of other metrics) + interval time.Duration //duration at which the reporter will persist metrics + db *leveldb.DB //the actual DB } +//NewMetricsDB creates a new LevelDB instance used to persist metrics defined +//inside p2p/protocols/accounting.go func NewMetricsDB(r metrics.Registry, d time.Duration, path string) *leveldb.DB { var val = make([]byte, 8) var err error + //Create the LevelDB db, err := leveldb.OpenFile(path, nil) if err != nil { log.Error(err.Error()) return nil } + //Check for all defined metrics that there is a value in the DB + //If there is, assign it to the metric. This means that the node + //has been running before and that metrics have been persisted. val, err = db.Get([]byte("account.balance.credit"), nil) if err == nil { mBalanceCredit.Inc(int64(binary.BigEndian.Uint64(val))) @@ -75,40 +83,52 @@ func NewMetricsDB(r metrics.Registry, d time.Duration, path string) *leveldb.DB mSelfDrops.Inc(int64(binary.BigEndian.Uint64(val))) } + //create the reporter reg := &reporter{ reg: r, interval: d, db: db, } + //run the go routine go reg.run() return db } +//run is the go routine which periodically sends the metrics to the configued LevelDB func (r *reporter) run() { intervalTicker := time.NewTicker(r.interval) for _ = range intervalTicker.C { + //at each tick send the metrics if err := r.send(); err != nil { log.Error("unable to send metrics to LevelDB. err=%v", "err", err) + //If there is an error in writing, exit the routine; we assume here that the error is + //severe and don't attempt to write again. + //Also, this should prevent leaking when the node is stopped return } } } +//send the metrics to the DB func (r *reporter) send() error { var err error + //for each metric in the registry (which is independent)... r.reg.Each(func(name string, i interface{}) { switch metric := i.(type) { + //assuming every metric here to be a Counter (separate registry) case metrics.Counter: + //...create a snapshot... ms := metric.Snapshot() byteVal := make([]byte, 8) binary.BigEndian.PutUint64(byteVal, uint64(ms.Count())) + //...and save the value to the DB err = r.db.Put([]byte(name), byteVal, nil) + default: } }) - return err } diff --git a/p2p/protocols/reporter_test.go b/p2p/protocols/reporter_test.go index 8f886df43f..8190ec8714 100644 --- a/p2p/protocols/reporter_test.go +++ b/p2p/protocols/reporter_test.go @@ -17,38 +17,53 @@ package protocols import ( + "io/ioutil" "os" + "path/filepath" "testing" "time" "github.com/ethereum/go-ethereum/log" ) +//TestReporter tests that the metrics being collected for p2p accounting +//are being persisted and available after restart of a node. +//It simulates restarting by just recreating the DB as if the node had restarted. func TestReporter(t *testing.T) { - dir := os.TempDir() + //create a test directory + dir, err := ioutil.TempDir("", "reporter-test") + if err != nil { + t.Fatal(err) + } defer os.RemoveAll(dir) + //setup the metrics log.Debug("Setting up metrics first time") reportInterval := 100 * time.Millisecond - db := SetupAccountingMetrics(reportInterval, dir+"/test.db") + db := SetupAccountingMetrics(reportInterval, filepath.Join(dir, "test.db")) log.Debug("Done.") + //do some metrics mBalanceCredit.Inc(12) mBytesCredit.Inc(34) mMsgDebit.Inc(9) - //give the reporter time to write to DB + //give the reporter time to write the metrics to DB time.Sleep(500 * time.Millisecond) + //set the metrics to nil - this effectively simulates the node having shut down... mBalanceCredit = nil mBytesCredit = nil mMsgDebit = nil + //close the DB also, or we can't create a new one db.Close() + //setup the metrics again log.Debug("Setting up metrics second time") - SetupAccountingMetrics(reportInterval, dir+"/test.db") + SetupAccountingMetrics(reportInterval, filepath.Join(dir, "test.db")) log.Debug("Done.") + //now check the metrics, they should have the same value as before "shutdown" if mBalanceCredit.Count() != 12 { t.Fatalf("Expected counter to be %d, but is %d", 12, mBalanceCredit.Count()) } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 137eb141d9..5d636dc205 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -91,3 +91,8 @@ func (s *Swap) loadState(peer *protocols.Peer) (err error) { } return } + +//Clean up Swap +func (swap *Swap) Close() { + swap.stateStore.Close() +} diff --git a/swarm/swarm.go b/swarm/swarm.go index e272a76813..b20c652f59 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -1,4 +1,4 @@ -// Copyright 2016 The go-ethereum Authors +// 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 @@ -53,6 +53,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/ethereum/go-ethereum/swarm/swap" "github.com/ethereum/go-ethereum/swarm/tracing" + "github.com/syndtr/goleveldb/leveldb" ) var ( @@ -66,20 +67,22 @@ var ( // the swarm stack type Swarm struct { - config *api.Config // swarm configuration - api *api.API // high level api layer (fs/manifest) - dns api.Resolver // DNS registrar - fileStore *storage.FileStore // distributed preimage archive, the local API to the storage with document level storage/retrieval support - streamer *stream.Registry - bzz *network.Bzz // the logistic manager - backend chequebook.Backend // simple blockchain Backend - privateKey *ecdsa.PrivateKey - corsString string - swapEnabled bool - netStore *storage.NetStore - sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit - ps *pss.Pss - swap *swap.Swap + config *api.Config // swarm configuration + api *api.API // high level api layer (fs/manifest) + dns api.Resolver // DNS registrar + fileStore *storage.FileStore // distributed preimage archive, the local API to the storage with document level storage/retrieval support + streamer *stream.Registry + bzz *network.Bzz // the logistic manager + backend chequebook.Backend // simple blockchain Backend + privateKey *ecdsa.PrivateKey + corsString string + swapEnabled bool + netStore *storage.NetStore + sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit + ps *pss.Pss + swap *swap.Swap + stateStore *state.DBStore + metricsStore *leveldb.DB tracerClose io.Closer } @@ -179,7 +182,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")) + self.metricsStore = protocols.SetupAccountingMetrics(10*time.Second, filepath.Join(config.Path, "metrics.db")) } var nodeID enode.ID @@ -447,6 +450,15 @@ func (self *Swarm) Stop() error { ch.Stop() ch.Save() } + if self.swap != nil { + self.swap.Close() + } + if self.metricsStore != nil { + self.metricsStore.Close() + } + if self.stateStore != nil { + self.stateStore.Close() + } if self.netStore != nil { self.netStore.Close()