Changed badger cleaner to run on interval basis and run for all node types

Author: durkmurder

cmd/consensus/main.go

@@ -685,31 +685,11 @@ func main() {
 			return hot, nil
 		}).
 		Component("consensus compliance engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			// initialize the entity database accessors
-			cleaner := bstorage.NewCleaner(node.Logger, node.DB, node.Metrics.CleanCollector, flow.DefaultValueLogGCFrequency)
-
 			// initialize the pending blocks cache
 			proposals := buffer.NewPendingBlocks()
 
 			logger := createLogger(node.Logger, node.RootChainID)
-			complianceCore, err := compliance.NewCore(logger,
-				node.Metrics.Engine,
-				node.Metrics.Mempool,
-				mainMetrics,
-				node.Metrics.Compliance,
-				node.Tracer,
-				cleaner,
-				node.Storage.Headers,
-				node.Storage.Payloads,
-				mutableState,
-				proposals,
-				syncCore,
-				hotstuffModules.Validator,
-				hot,
-				hotstuffModules.VoteAggregator,
-				hotstuffModules.TimeoutAggregator,
-				modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold),
-			)
+			complianceCore, err := compliance.NewCore(logger, node.Metrics.Engine, node.Metrics.Mempool, mainMetrics, node.Metrics.Compliance, node.Tracer, node.Storage.Headers, node.Storage.Payloads, mutableState, proposals, syncCore, hotstuffModules.Validator, hot, hotstuffModules.VoteAggregator, hotstuffModules.TimeoutAggregator, modulecompliance.WithSkipNewProposalsThreshold(node.ComplianceConfig.SkipNewProposalsThreshold))
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize compliance core: %w", err)
 			}

cmd/scaffold.go

@@ -873,6 +873,10 @@ func (fnb *FlowNodeBuilder) initDB() error {
 		return nil
 	})
 
+	fnb.Component("badger log cleaner", func(node *NodeConfig) (module.ReadyDoneAware, error) {
+		return bstorage.NewCleaner(node.Logger, node.DB, node.Metrics.CleanCollector, flow.DefaultValueLogGCWaitDuration), nil
+	})
+
 	return nil
 }
 

engine/consensus/compliance/core.go

@@ -43,7 +43,6 @@ type Core struct {
 	hotstuffMetrics   module.HotstuffMetrics
 	complianceMetrics module.ComplianceMetrics
 	tracer            module.Tracer
-	cleaner           storage.Cleaner
 	headers           storage.Headers
 	payloads          storage.Payloads
 	state             protocol.ParticipantState
@@ -66,7 +65,6 @@ func NewCore(
 	hotstuffMetrics module.HotstuffMetrics,
 	complianceMetrics module.ComplianceMetrics,
 	tracer module.Tracer,
-	cleaner storage.Cleaner,
 	headers storage.Headers,
 	payloads storage.Payloads,
 	state protocol.ParticipantState,
@@ -92,7 +90,6 @@ func NewCore(
 		mempoolMetrics:    mempool,
 		hotstuffMetrics:   hotstuffMetrics,
 		complianceMetrics: complianceMetrics,
-		cleaner:           cleaner,
 		headers:           headers,
 		payloads:          payloads,
 		state:             state,
@@ -238,12 +235,6 @@ func (c *Core) OnBlockProposal(originID flow.Identifier, proposal *messages.Bloc
 		return fmt.Errorf("could not process block proposal: %w", err)
 	}
 
-	// most of the heavy database checks are done at this point, so this is a
-	// good moment to potentially kick-off a garbage collection of the DB
-	// NOTE: this is only effectively run every 1000th calls, which corresponds
-	// to every 1000th successfully processed block
-	c.cleaner.RunGC()
-
 	return nil
 }
 

model/flow/constants.go

@@ -40,9 +40,8 @@ const DefaultMaxCollectionTotalGas = 10_000_000 // 10M
 // DefaultMaxCollectionSize is the default maximum number of transactions allowed inside a collection.
 const DefaultMaxCollectionSize = 100
 
-// DefaultValueLogGCFrequency is the default frequency in blocks that we call the
-// badger value log GC. Equivalent to 10 mins for a 1 second block time
-const DefaultValueLogGCFrequency = 10 * 60
+// DefaultValueLogGCWaitDuration is the default wait duration before we repeatedly call the badger value log GC.
+const DefaultValueLogGCWaitDuration time.Duration = 10 * time.Minute
 
 // DefaultRequiredApprovalsForSealConstruction is the default number of approvals required to construct a candidate seal
 // for subsequent inclusion in block.

storage/badger/cleaner.go

@@ -3,6 +3,8 @@
 package badger
 
 import (
+	"github.com/onflow/flow-go/module/component"
+	"github.com/onflow/flow-go/module/irrecoverable"
 	"math/rand"
 	"time"
 
@@ -12,76 +14,91 @@ import (
 	"github.com/onflow/flow-go/module"
 )
 
+// Cleaner uses component.ComponentManager to implement module.Startable and module.ReadyDoneAware
+// to run an internal goroutine which run badger value log garbage collection on timely basis.
 type Cleaner struct {
-	log     zerolog.Logger
-	db      *badger.DB
-	metrics module.CleanerMetrics
-	enabled bool
-	ratio   float64
-	freq    int
-	calls   int
+	*component.ComponentManager
+	log      zerolog.Logger
+	db       *badger.DB
+	metrics  module.CleanerMetrics
+	ratio    float64
+	interval time.Duration
 }
 
-// NewCleaner returns a cleaner that runs the badger value log garbage collection once every `frequency` calls
-// if a frequency of zero is passed in, we will not run the GC at all
-func NewCleaner(log zerolog.Logger, db *badger.DB, metrics module.CleanerMetrics, frequency int) *Cleaner {
+var _ component.Component = (*Cleaner)(nil)
+
+// NewCleaner returns a cleaner that runs the badger value log garbage collection once every `interval` duration
+// if an interval of zero is passed in, we will not run the GC at all.
+func NewCleaner(log zerolog.Logger, db *badger.DB, metrics module.CleanerMetrics, interval time.Duration) *Cleaner {
 	// NOTE: we run garbage collection frequently at points in our business
 	// logic where we are likely to have a small breather in activity; it thus
 	// makes sense to run garbage collection often, with a smaller ratio, rather
 	// than running it rarely and having big rewrites at once
 	c := &Cleaner{
-		log:     log.With().Str("component", "cleaner").Logger(),
-		db:      db,
-		metrics: metrics,
-		ratio:   0.2,
-		freq:    frequency,
-		enabled: frequency > 0, // Disable if passed in 0 as frequency
+		log:      log.With().Str("component", "cleaner").Logger(),
+		db:       db,
+		metrics:  metrics,
+		ratio:    0.2,
+		interval: interval,
 	}
-	// we don't want the entire network to run GC at the same time, so
-	// distribute evenly over time
-	if c.enabled {
-		c.calls = rand.Intn(c.freq)
+
+	cmBuilder := component.NewComponentManagerBuilder()
+
+	// Disable if passed in 0 as interval
+	if c.interval > 0 {
+		cmBuilder.AddWorker(c.gcWorkerRoutine)
 	}
+
+	c.ComponentManager = cmBuilder.Build()
 	return c
 }
 
-func (c *Cleaner) RunGC() {
-	if !c.enabled {
+// gcWorkerRoutine runs badger GC on timely basis.
+func (c *Cleaner) gcWorkerRoutine(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) {
+	ready()
+	ticker := time.NewTicker(c.nextWaitDuration())
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			c.runGC()
+
+			// reset the ticker with a new interval and random jitter
+			ticker.Reset(c.nextWaitDuration())
+		}
+	}
+}
+
+// nextWaitDuration calculates next duration for Cleaner to wait before attempting to run GC.
+// We add 20% jitter into the interval, so that we don't risk nodes syncing
+// up on their GC calls over time.
+func (c *Cleaner) nextWaitDuration() time.Duration {
+	return time.Duration(c.interval.Milliseconds() + rand.Int63n(c.interval.Milliseconds()/5))
+}
+
+// runGC runs garbage collection for badger DB, handles sentinel errors and reports metrics.
+func (c *Cleaner) runGC() {
+	started := time.Now()
+	err := c.db.RunValueLogGC(c.ratio)
+	if err == badger.ErrRejected {
+		// NOTE: this happens when a GC call is already running
+		c.log.Warn().Msg("garbage collection on value log already running")
 		return
 	}
-	// only actually run approximately every frequency number of calls
-	c.calls++
-	if c.calls < c.freq {
+	if err == badger.ErrNoRewrite {
+		// NOTE: this happens when no files have any garbage to drop
+		c.log.Debug().Msg("garbage collection on value log unnecessary")
+		return
+	}
+	if err != nil {
+		c.log.Error().Err(err).Msg("garbage collection on value log failed")
 		return
 	}
 
-	// we add 20% jitter into the interval, so that we don't risk nodes syncing
-	// up on their GC calls over time
-	c.calls = rand.Intn(c.freq / 5)
-
-	// run the garbage collection in own goroutine and handle sentinel errors
-	go func() {
-		started := time.Now()
-		err := c.db.RunValueLogGC(c.ratio)
-		if err == badger.ErrRejected {
-			// NOTE: this happens when a GC call is already running
-			c.log.Warn().Msg("garbage collection on value log already running")
-			return
-		}
-		if err == badger.ErrNoRewrite {
-			// NOTE: this happens when no files have any garbage to drop
-			c.log.Debug().Msg("garbage collection on value log unnecessary")
-			return
-		}
-		if err != nil {
-			c.log.Error().Err(err).Msg("garbage collection on value log failed")
-			return
-		}
-
-		runtime := time.Since(started)
-		c.log.Debug().
-			Dur("gc_duration", runtime).
-			Msg("garbage collection on value log executed")
-		c.metrics.RanGC(runtime)
-	}()
+	runtime := time.Since(started)
+	c.log.Debug().
+		Dur("gc_duration", runtime).
+		Msg("garbage collection on value log executed")
+	c.metrics.RanGC(runtime)
 }