recon: add debug logs in ledger

FAB-13356 #done

Change-Id: I8359ae52b6a0c0bcf203c6f4be4faa27563f36ad
Signed-off-by: senthil <cendhu@gmail.com>
Signed-off-by: manish <manish.sethi@gmail.com>

core/ledger/kvledger/hashcheck_pvtdata.go

@@ -48,12 +48,15 @@ func findValidAndInvalidBlockPvtData(blockPvtData *ledger.BlockPvtData, blockSto
 	var invalidPvtData []*ledger.PvtdataHashMismatch
 	for _, txPvtData := range blockPvtData.WriteSets {
 		// (1) retrieve the txrwset from the blockstore
+		logger.Debugf("Retrieving rwset of blockNum:[%d], txNum:[%d]", blockPvtData.BlockNum, txPvtData.SeqInBlock)
 		txRWSet, err := retrieveRwsetForTx(blockPvtData.BlockNum, txPvtData.SeqInBlock, blockStore)
 		if err != nil {
 			return nil, nil, err
 		}
 
 		// (2) validate passed pvtData against the pvtData hash in the tx rwset.
+		logger.Debugf("Constructing valid and invalid pvtData using rwset of blockNum:[%d], txNum:[%d]",
+			blockPvtData.BlockNum, txPvtData.SeqInBlock)
 		validData, invalidData := findValidAndInvalidTxPvtData(txPvtData, txRWSet, blockPvtData.BlockNum)
 
 		// (3) append validData to validPvtDataPvt list of this block and

core/ledger/kvledger/kv_ledger.go

@@ -403,23 +403,26 @@ func (l *kvLedger) GetConfigHistoryRetriever() (ledger.ConfigHistoryRetriever, e
 }
 
 func (l *kvLedger) CommitPvtDataOfOldBlocks(pvtData []*ledger.BlockPvtData) ([]*ledger.PvtdataHashMismatch, error) {
+	logger.Debugf("[%s:] Comparing pvtData of [%d] old blocks against the hashes in transaction's rwset to find valid and invalid data",
+		l.ledgerID, len(pvtData))
 	validPvtData, hashMismatches, err := ConstructValidAndInvalidPvtData(pvtData, l.blockStore)
 	if err != nil {
 		return nil, err
 	}
 
+	logger.Debugf("[%s:] Committing pvtData of [%d] old blocks to the pvtdatastore", l.ledgerID, len(pvtData))
 	err = l.blockStore.CommitPvtDataOfOldBlocks(validPvtData)
 	if err != nil {
 		return nil, err
 	}
 
-	// (2) remove stale pvt data of old blocks and then commit to the stateDB
+	logger.Debugf("[%s:] Committing pvtData of [%d] old blocks to the stateDB", l.ledgerID, len(pvtData))
 	err = l.txtmgmt.RemoveStaleAndCommitPvtDataOfOldBlocks(validPvtData)
 	if err != nil {
 		return nil, err
 	}
 
-	// (3) reset the lastUpdatedOldBlockList from the pvtstore
+	logger.Debugf("[%s:] Clearing the bookkeeping information from pvtdatastore", l.ledgerID)
 	if err := l.blockStore.ResetLastUpdatedOldBlocksList(); err != nil {
 		return nil, err
 	}

core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_txmgr.go

@@ -118,6 +118,7 @@ func (txmgr *LockBasedTxMgr) ValidateAndPrepare(blockAndPvtdata *ledger.BlockAnd
 	txmgr.pvtdataPurgeMgr.WaitForPrepareToFinish()
 	txmgr.oldBlockCommit.Lock()
 	defer txmgr.oldBlockCommit.Unlock()
+	logger.Debug("lock acquired on oldBlockCommit for validating read set version against the committed version")
 
 	block := blockAndPvtdata.Block
 	logger.Debugf("Validating new block with num trans = [%d]", len(block.Data.Data))
@@ -167,38 +168,44 @@ func (txmgr *LockBasedTxMgr) RemoveStaleAndCommitPvtDataOfOldBlocks(blocksPvtDat
 	// these three functions to execute parallely. However, we cannot remove
 	// the lock on oldBlockCommit as it is also used to avoid interleaving
 	// between Commit() and execution of this function for the correctness.
+	logger.Debug("Waiting for purge mgr to finish the background job of computing expirying keys for the block")
 	txmgr.pvtdataPurgeMgr.WaitForPrepareToFinish()
 	txmgr.oldBlockCommit.Lock()
 	defer txmgr.oldBlockCommit.Unlock()
+	logger.Debug("lock acquired on oldBlockCommit for committing pvtData of old blocks to state database")
 
 	// (1) as the blocksPvtData can contain multiple versions of pvtData for
 	// a given <ns, coll, key>, we need to find duplicate tuples with different
 	// versions and use the one with the higher version
+	logger.Debug("Constructing unique pvtData by removing duplicate entries")
 	uniquePvtData, err := constructUniquePvtData(blocksPvtData)
 	if len(uniquePvtData) == 0 || err != nil {
 		return err
 	}
 
 	// (3) remove the pvt data which does not matches the hashed
 	// value stored in the public state
+	logger.Debug("Finding and removing stale pvtData")
 	if err := uniquePvtData.findAndRemoveStalePvtData(txmgr.db); err != nil {
 		return err
 	}
 
 	// (4) create the update batch from the uniquePvtData
 	batch := uniquePvtData.transformToUpdateBatch()
 
-	// (5) update booking in the purge manager and update toPurgeList
+	// (5) update bookkeeping in the purge manager and update toPurgeList
 	// (i.e., the list of expiry keys). As the expiring keys would have
 	// been constructed during last PrepareExpiringKeys from commit, we need
 	// to update the list. This is because RemoveStaleAndCommitPvtDataOfOldBlocks
 	// may have added new data which might be eligible for expiry during the
 	// next regular block commit.
+	logger.Debug("Updating bookkeeping info in the purge manager")
 	if err := txmgr.pvtdataPurgeMgr.UpdateBookkeepingForPvtDataOfOldBlocks(batch.PvtUpdates); err != nil {
 		return err
 	}
 
 	// (6) commit the pvt data to the stateDB
+	logger.Debug("Committing updates to state database")
 	if err := txmgr.db.ApplyPrivacyAwareUpdates(batch, nil); err != nil {
 		return err
 	}
@@ -442,6 +449,7 @@ func (txmgr *LockBasedTxMgr) Commit() error {
 	//     end up with an incorrect update batch.
 	txmgr.oldBlockCommit.Lock()
 	defer txmgr.oldBlockCommit.Unlock()
+	logger.Debug("lock acquired on oldBlockCommit for committing regular updates to state database")
 
 	// When using the purge manager for the first block commit after peer start, the asynchronous function
 	// 'PrepareForExpiringKeys' is invoked in-line. However, for the subsequent blocks commits, this function is invoked

core/ledger/pvtdatastorage/store_impl.go

@@ -286,18 +286,21 @@ func (s *store) CommitPvtDataOfOldBlocks(blocksPvtData map[uint64][]*ledger.TxPv
 	dataEntries := constructDataEntriesFromBlocksPvtData(blocksPvtData)
 
 	// (2) construct update entries (i.e., dataEntries, expiryEntries, missingDataEntries) from the above created data entries
+	logger.Debugf("Constructing pvtdatastore entries for pvtData of [%d] old blocks", len(blocksPvtData))
 	updateEntries, err := s.constructUpdateEntriesFromDataEntries(dataEntries)
 	if err != nil {
 		return err
 	}
 
 	// (3) create a db update batch from the update entries
+	logger.Debug("Constructing update batch from pvtdatastore entries")
 	batch, err := constructUpdateBatchFromUpdateEntries(updateEntries)
 	if err != nil {
 		return err
 	}
 
 	// (4) commit the update entries to the pvtStore
+	logger.Debug("Committing the update batch to pvtdatastore")
 	if err := s.commitBatch(batch); err != nil {
 		return err
 	}