diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go index f366bd6aa6..770406a27e 100644 --- a/p2p/protocols/accounting.go +++ b/p2p/protocols/accounting.go @@ -20,7 +20,6 @@ import ( "time" "github.com/ethereum/go-ethereum/metrics" - "github.com/syndtr/goleveldb/leveldb" ) //define some metrics @@ -113,7 +112,7 @@ func NewAccounting(balance Balance, po Prices) *Accounting { //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 { +func SetupAccountingMetrics(reportInterval time.Duration, path string) *AccountingMetrics { //create an empty registry registry := metrics.NewRegistry() //instantiate the metrics @@ -126,7 +125,7 @@ func SetupAccountingMetrics(reportInterval time.Duration, path string) *leveldb. mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", registry) mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", registry) //create the DB and start persisting - return NewMetricsDB(registry, reportInterval, path) + return NewAccountingMetrics(registry, reportInterval, path) } //Implement Hook.Send diff --git a/p2p/protocols/accounting_simulation_test.go b/p2p/protocols/accounting_simulation_test.go index 65b737abed..e90a1d81d2 100644 --- a/p2p/protocols/accounting_simulation_test.go +++ b/p2p/protocols/accounting_simulation_test.go @@ -20,7 +20,10 @@ import ( "context" "flag" "fmt" + "io/ioutil" "math/rand" + "os" + "path/filepath" "reflect" "sync" "testing" @@ -66,6 +69,13 @@ func init() { func TestAccountingSimulation(t *testing.T) { //setup the balances objects for every node bal := newBalances(*nodes) + //setup the metrics system or tests will fail trying to write metrics + dir, err := ioutil.TempDir("", "account-sim") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + SetupAccountingMetrics(1*time.Second, filepath.Join(dir, "metrics.db")) //define the node.Service for this test services := adapters.Services{ "accounting": func(ctx *adapters.ServiceContext) (node.Service, error) { diff --git a/p2p/protocols/reporter.go b/p2p/protocols/reporter.go index eea17b50cc..28f64b1814 100644 --- a/p2p/protocols/reporter.go +++ b/p2p/protocols/reporter.go @@ -26,17 +26,32 @@ import ( "github.com/syndtr/goleveldb/leveldb" ) +//AccountMetrics abstracts away the metrics DB and +//the reporter to persist metrics +type AccountingMetrics struct { + metricsStore *leveldb.DB + reporter *reporter +} + +//Close will be called when the node is being shutdown +//for a graceful cleanup +func (am *AccountingMetrics) Close() { + am.reporter.quit <- struct{}{} + am.metricsStore.Close() +} + //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 //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 + quit chan struct{} //quit the reporter loop } //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 { +func NewAccountingMetrics(r metrics.Registry, d time.Duration, path string) *AccountingMetrics { var val = make([]byte, 8) var err error @@ -50,64 +65,64 @@ func NewMetricsDB(r metrics.Registry, d time.Duration, path string) *leveldb.DB //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))) + metricsMap := map[string]metrics.Counter{ + "account.balance.credit": mBalanceCredit, + "account.balance.debit": mBalanceDebit, + "account.bytes.credit": mBytesCredit, + "account.bytes.debit": mBytesDebit, + "account.msg.credit": mMsgCredit, + "account.msg.debit": mMsgDebit, + "account.peerdrops": mPeerDrops, + "account.selfdrops": mSelfDrops, } - 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))) + //iterate the map and get the values + for key, metric := range metricsMap { + val, err = db.Get([]byte(key), nil) + //until the first time a value is being written, + //this will return an error. + //it could be beneficial though to log errors later, + //but that would require a different logic + if err == nil { + metric.Inc(int64(binary.BigEndian.Uint64(val))) + } } //create the reporter - reg := &reporter{ + rep := &reporter{ reg: r, interval: d, db: db, + quit: make(chan struct{}), } //run the go routine - go reg.run() + go rep.run() - return db + m := &AccountingMetrics{ + metricsStore: db, + reporter: rep, + } + return m } //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 + for { + select { + case <-intervalTicker.C: + //at each tick send the metrics + if err := r.send(); err != nil { + log.Error("unable to send metrics to LevelDB", "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 + } + case <-r.quit: + //graceful shutdown return } } diff --git a/p2p/protocols/reporter_test.go b/p2p/protocols/reporter_test.go index 8190ec8714..b9f06e6744 100644 --- a/p2p/protocols/reporter_test.go +++ b/p2p/protocols/reporter_test.go @@ -39,8 +39,8 @@ func TestReporter(t *testing.T) { //setup the metrics log.Debug("Setting up metrics first time") - reportInterval := 100 * time.Millisecond - db := SetupAccountingMetrics(reportInterval, filepath.Join(dir, "test.db")) + reportInterval := 5 * time.Millisecond + metrics := SetupAccountingMetrics(reportInterval, filepath.Join(dir, "test.db")) log.Debug("Done.") //do some metrics @@ -49,18 +49,19 @@ func TestReporter(t *testing.T) { mMsgDebit.Inc(9) //give the reporter time to write the metrics to DB - time.Sleep(500 * time.Millisecond) + time.Sleep(20 * 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() + metrics.Close() //setup the metrics again log.Debug("Setting up metrics second time") - SetupAccountingMetrics(reportInterval, filepath.Join(dir, "test.db")) + metrics = SetupAccountingMetrics(reportInterval, filepath.Join(dir, "test.db")) + defer metrics.Close() log.Debug("Done.") //now check the metrics, they should have the same value as before "shutdown" diff --git a/swarm/swarm.go b/swarm/swarm.go index b20c652f59..28e88d5ca0 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -53,7 +53,6 @@ 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 ( @@ -67,22 +66,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 - stateStore *state.DBStore - metricsStore *leveldb.DB + 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 + accountingMetrics *protocols.AccountingMetrics tracerClose io.Closer } @@ -182,7 +181,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e return nil, err } self.swap = swap.New(balancesStore) - self.metricsStore = protocols.SetupAccountingMetrics(10*time.Second, filepath.Join(config.Path, "metrics.db")) + self.accountingMetrics = protocols.SetupAccountingMetrics(10*time.Second, filepath.Join(config.Path, "metrics.db")) } var nodeID enode.ID @@ -453,8 +452,8 @@ func (self *Swarm) Stop() error { if self.swap != nil { self.swap.Close() } - if self.metricsStore != nil { - self.metricsStore.Close() + if self.accountingMetrics != nil { + self.accountingMetrics.Close() } if self.stateStore != nil { self.stateStore.Close()