p2p/protocols: graceful shutdown of accounting metrics and fixes

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

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) {

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
 		}
 	}

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"

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()