Node utx async clean

pkg/metrics/metrics.go

@@ -69,8 +69,9 @@
   - Avoid more than one piece of information in one tag
 */
 var (
-	once sync.Once
-	rep  *reporter = nil
+	once              sync.Once
+	rep               *reporter = nil
+	DefaultUtxSampler *RangeSampler
 )
 
 func MicroBlockInv(mb *proto.MicroBlockInv, source string) {
@@ -211,12 +212,13 @@
 	reportBlock(t, f)
 }
 
-func Utx(utxCount int) {
+func UtxCollect() {
 	if rep == nil {
 		return
 	}
+	curV, minV, maxV := DefaultUtxSampler.Sample()
 	t := emptyTags().node().withEvent(eventUtx)
-	f := emptyFields().withUtxCount(utxCount)
+	f := emptyFields().withUtxCount(curV, minV, maxV)
 	reportUtx(t, f)
 }
 
@@ -311,8 +313,10 @@
 	return f
 }
 
-func (f fields) withUtxCount(utxCount int) fields {
-	f["utx_count"] = utxCount
+func (f fields) withUtxCount(currentValue, minValue, maxValue int) fields {
+	f["utx_current"] = currentValue
+	f["utx_min"] = minValue
+	f["utx_max"] = maxValue
 	return f
 }
 
@@ -379,6 +383,7 @@
 			interval:  reportInterval,
 			in:        make(chan *influx.Point, bufferSize),
 		}
+		DefaultUtxSampler = NewRangeSampler(0)
 		go rep.run(ctx)
 	})
 	return nil
@@ -397,6 +402,7 @@
 			}
 			return
 		case <-ticker.C:
+			UtxCollect()
 			err := r.report()
 			if err != nil {
 				zap.S().Warnf("Failed to report metrics: %v", err)
@@ -473,3 +479,39 @@
 func shortID(id proto.BlockID) string {
 	return id.String()[:6]
 }
+
+type RangeSampler struct {
+	sync.Mutex
+	current int
+	min     int
+	max     int
+}
+
+func NewRangeSampler(initial int) *RangeSampler {
+	return &RangeSampler{
+		current: initial,
+		min:     initial,
+		max:     initial,
+	}
+}
+
+func (r *RangeSampler) Update(value int) {
+	r.Lock()
+	defer r.Unlock()
+	r.current = value
+	if value < r.min {
+		r.min = value
+	}
+	if value > r.max {
+		r.max = value
+	}
+}
+
+func (r *RangeSampler) Sample() (int, int, int) {
+	r.Lock()
+	defer r.Unlock()
+	curV, minV, maxV := r.current, r.min, r.max
+	r.min = r.current
+	r.max = r.current
+	return curV, minV, maxV
+}

pkg/miner/utxpool/pool.go

@@ -8,6 +8,7 @@ import (
 	"github.com/mr-tron/base58"
 	"github.com/pkg/errors"
 	"github.com/wavesplatform/gowaves/pkg/crypto"
+	"github.com/wavesplatform/gowaves/pkg/metrics"
 	"github.com/wavesplatform/gowaves/pkg/proto"
 	"github.com/wavesplatform/gowaves/pkg/settings"
 	"github.com/wavesplatform/gowaves/pkg/types"
@@ -74,7 +75,13 @@ func (a *UtxImpl) Add(t proto.Transaction) error {
 	}
 	a.mu.Lock()
 	defer a.mu.Unlock()
-	return a.addWithBytes(t, bts)
+
+	errAdd := a.addWithBytes(t, bts)
+	if errAdd != nil {
+		return errAdd
+	}
+	metrics.DefaultUtxSampler.Update(a.Count())
+	return nil
 }
 
 func (a *UtxImpl) AddBytes(bts []byte) error {

pkg/node/fsm/fsm.go

@@ -92,8 +92,10 @@ func (a *BaseInfo) BroadcastTransaction(t proto.Transaction, receivedFrom peer.P
 }
 
 func (a *BaseInfo) CleanUtx() {
-	utxpool.NewCleaner(a.storage, a.utx, a.tm).Clean()
-	metrics.Utx(a.utx.Count())
+	go func() {
+		utxpool.NewCleaner(a.storage, a.utx, a.tm).Clean()
+		metrics.DefaultUtxSampler.Update(a.utx.Count())
+	}()
 }
 
 // States.

pkg/node/fsm/fsm_common.go

@@ -13,7 +13,6 @@ import (
 
 	"github.com/wavesplatform/gowaves/pkg/libs/signatures"
 	"github.com/wavesplatform/gowaves/pkg/logging"
-	"github.com/wavesplatform/gowaves/pkg/metrics"
 	"github.com/wavesplatform/gowaves/pkg/node/fsm/sync_internal"
 	"github.com/wavesplatform/gowaves/pkg/node/fsm/tasks"
 	"github.com/wavesplatform/gowaves/pkg/p2p/peer"
@@ -152,7 +151,6 @@ func tryBroadcastTransaction(
 		return fsm, nil, err
 	}
 	baseInfo.BroadcastTransaction(t, p)
-	metrics.Utx(baseInfo.utx.Count())
 	return fsm, nil, nil
 }
 

pkg/node/fsm/ng_state.go

@@ -195,7 +195,6 @@ func (a *NGState) Block(peer peer.Peer, block *proto.Block) (State, Async, error
 		return a, nil, a.Errorf(errors.Wrapf(err, "failed to apply block %s", block.BlockID()))
 	}
 	metrics.BlockApplied(block, height+1)
-	metrics.Utx(a.baseInfo.utx.Count())
 	a.blocksCache.Clear()
 	a.blocksCache.AddBlockState(block)
 	a.baseInfo.scheduler.Reschedule()
@@ -235,7 +234,6 @@ func (a *NGState) MinedBlock(
 		return a, nil, a.Errorf(err)
 	}
 	metrics.BlockApplied(block, height+1)
-	metrics.Utx(a.baseInfo.utx.Count())
 	zap.S().Infof("[%s] Generated key block '%s' successfully applied to state", a, block.ID.String())
 
 	a.blocksCache.Clear()
@@ -308,7 +306,6 @@ func (a *NGState) mineMicro(
 	a.blocksCache.AddBlockState(block)
 	a.baseInfo.scheduler.Reschedule()
 	metrics.MicroBlockApplied(micro)
-	metrics.Utx(a.baseInfo.utx.Count())
 	a.baseInfo.CleanUtx()
 	inv := proto.NewUnsignedMicroblockInv(
 		micro.SenderPK,
@@ -392,7 +389,6 @@ func (a *NGState) checkAndAppendMicroBlock(
 		return nil, errors.Wrap(err, "failed to apply created from micro block")
 	}
 	metrics.MicroBlockApplied(micro)
-	metrics.Utx(a.baseInfo.utx.Count())
 	a.baseInfo.CleanUtx()
 	return newBlock, nil
 }

pkg/node/fsm/sync_state.go

@@ -212,7 +212,6 @@ func (a *SyncState) MinedBlock(
 		return a, nil, nil // We've failed to apply mined block, it's not an error
 	}
 	metrics.BlockAppliedFromExtension(block, height+1)
-	metrics.Utx(a.baseInfo.utx.Count())
 	a.baseInfo.scheduler.Reschedule()
 
 	// first we should send block
@@ -285,7 +284,6 @@ func (a *SyncState) applyBlocksWithSnapshots(
 	}
 	for _, b := range blocks {
 		metrics.BlockAppliedFromExtension(b, height+1)
-		metrics.Utx(a.baseInfo.utx.Count())
 		height++
 	}
 	a.baseInfo.scheduler.Reschedule()

pkg/node/fsm/wait_micro_snapshot.go

@@ -199,7 +199,6 @@ func (a *WaitMicroSnapshotState) checkAndAppendMicroBlock(
 		return nil, errors.Wrap(err, "failed to apply created from micro block")
 	}
 	metrics.MicroBlockApplied(micro)
-	metrics.Utx(a.baseInfo.utx.Count())
 	a.baseInfo.CleanUtx()
 	return newBlock, nil
 }

pkg/node/fsm/wait_snapshot_state.go

@@ -122,7 +122,6 @@ func (a *WaitSnapshotState) BlockSnapshot(
 		return processScoreAfterApplyingOrReturnToNG(a, a.baseInfo, a.receivedScores, a.blocksCache)
 	}
 	metrics.SnapshotBlockApplied(a.blockWaitingForSnapshot, height+1)
-	metrics.Utx(a.baseInfo.utx.Count())
 	zap.S().Named(logging.FSMNamespace).Debugf("[%s] Handle received key block message: block '%s' applied to state",
 		a, blockID)