log counters for mempool and trackers for tx finalization

mempool/clist_mempool.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"os"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -55,6 +56,12 @@ type CListMempool struct {
 
 	logger  log.Logger
 	metrics *Metrics
+
+	// counter
+	addedTx    uint32
+	rejectedTx uint32
+	// File for logging counters
+	counterFile *os.File
 }
 
 var _ Mempool = &CListMempool{}
@@ -92,9 +99,47 @@ func NewCListMempool(
 		option(mp)
 	}
 
+	var err error
+	homeDir, err := os.UserHomeDir()
+	if err != nil {
+		mp.logger.Error("Failed to get home directory", "err", err)
+	}
+	mp.counterFile, err = os.OpenFile(homeDir+"/customlogs/comet_mempool-counters.log", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		mp.logger.Error("Failed to open counters.log", "err", err)
+	}
+
+	go mp.logCounters()
 	return mp
 }
 
+// logCounters logs the counter values every 1 second
+func (mem *CListMempool) logCounters() {
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			// Atomically load and reset counters
+			addedTx := atomic.SwapUint32(&mem.addedTx, 0)
+			rejectedTx := atomic.SwapUint32(&mem.rejectedTx, 0)
+
+			// Get the current timestamp
+			timestamp := time.Now().UTC().UnixNano()
+
+			// Create a CSV line
+			logLine := fmt.Sprintf("%d,%d,%d\n",
+				timestamp, addedTx, rejectedTx)
+
+			// Append the line to the file
+			if _, err := mem.counterFile.WriteString(logLine); err != nil {
+				fmt.Printf("failed to write counter log: %v\n", err)
+			}
+		}
+	}
+}
+
 func (mem *CListMempool) getCElement(txKey types.TxKey) (*clist.CElement, bool) {
 	if e, ok := mem.txsMap.Load(txKey); ok {
 		return e.(*clist.CElement), true
@@ -443,6 +488,7 @@ func (mem *CListMempool) resCbFirstTime(
 			}
 			memTx.addSender(txInfo.SenderID)
 			mem.addTx(memTx)
+			atomic.AddUint32(&mem.addedTx, 1)
 			mem.logger.Debug(
 				"added good transaction",
 				"tx", types.Tx(tx).Hash(),
@@ -452,6 +498,7 @@ func (mem *CListMempool) resCbFirstTime(
 			)
 			mem.notifyTxsAvailable()
 		} else {
+			atomic.AddUint32(&mem.rejectedTx, 1)
 			// ignore bad transaction
 			mem.logger.Debug(
 				"rejected bad transaction",

state/execution.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	abci "github.com/cometbft/cometbft/abci/types"
@@ -43,6 +44,9 @@ type BlockExecutor struct {
 	logger log.Logger
 
 	metrics *Metrics
+
+	// timeTrackingLog is a file to log the time taken for each block processing
+	timeTrackingLog *os.File
 }
 
 type BlockExecutorOption func(executor *BlockExecutor)
@@ -79,6 +83,12 @@ func NewBlockExecutor(
 		option(res)
 	}
 
+	var err error
+	res.timeTrackingLog, err = os.OpenFile("/tmp/blockExec-timeTracking.log", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		res.logger.Error("failed to open time tracking log", "err", err)
+		os.Exit(1)
+	}
 	return res
 }
 
@@ -319,6 +329,11 @@ func (blockExec *BlockExecutor) applyBlock(state State, blockID types.BlockID, b
 	// NOTE: if we crash between Commit and Save, events wont be fired during replay
 	fireEvents(blockExec.logger, blockExec.eventBus, block, blockID, abciResponse, validatorUpdates)
 
+	logLine := fmt.Sprintf("applyBlock(%d):: FinishedAt %d\n", block.Height, time.Now().UTC().UnixNano())
+	if _, err := blockExec.timeTrackingLog.WriteString(logLine); err != nil {
+		blockExec.logger.Error("failed to write to time tracking log", "err", err)
+	}
+
 	return state, nil
 }