chore: clean up logging

Author: adg-flare

internal/boff/backoff.go

@@ -32,15 +32,19 @@ func RetryNoReturn(ctx context.Context, operation func() error, name string) err
 }
 
 func retry[T any](ctx context.Context, operation func() (T, error), name string, maxElapsedTime time.Duration) (T, error) {
-	return backoff.Retry(
+	result, err := backoff.Retry(
 		ctx,
 		operation,
 		backoff.WithBackOff(backoff.NewExponentialBackOff()),
 		backoff.WithMaxElapsedTime(maxElapsedTime),
 		backoff.WithNotify(
 			func(err error, d time.Duration) {
-				logger.Debugf("%s error: %s - retrying after %v", name, err, d)
+				logger.Debugf("Backoff retry: op=%s, error=%s, delay=%v", name, err, d)
 			},
 		),
 	)
+	if err != nil && ctx.Err() == nil {
+		logger.Warnf("Backoff exhausted: op=%s, error=%s", name, err)
+	}
+	return result, err
 }

internal/chain/block_search.go

@@ -21,7 +21,7 @@ func GetNearestBlockByTimestampFromChain(
 	endBlockNumber uint64,
 ) (uint64, error) {
 	logger.Debugf(
-		"Getting nearest block by timestamp from chain: search timestamp: %d, start block: %d, end block: %d",
+		"Block search starting: search_timestamp=%d, start_block=%d, end_block=%d",
 		searchTimestamp, startBlockNumber, endBlockNumber,
 	)
 
@@ -46,7 +46,7 @@ func GetNearestBlockByTimestampFromChain(
 
 		searchStartBlockNumber = startCandidate
 		searchEndBlockNumber = startCandidate + searchWindowBlocks
-		logger.Debugf("Search block window narrowed down to: %d-%d", searchStartBlockNumber, searchEndBlockNumber)
+		logger.Debugf("Block search window narrowed: from=%d, to=%d", searchStartBlockNumber, searchEndBlockNumber)
 	}
 
 	blockNumber, err := binarySearchBlockByTimestamp(
@@ -56,7 +56,7 @@ func GetNearestBlockByTimestampFromChain(
 		return 0, errors.Wrap(err, "GetNearestBlockByTimestampFromChain")
 	}
 
-	logger.Debugf("Found nearest block by timestamp from chain: block number: %d", blockNumber)
+	logger.Debugf("Block search complete: block=%d", blockNumber)
 	return blockNumber, nil
 }
 

internal/core/engine.go

@@ -167,11 +167,11 @@ func (ci *Engine) IndexHistory(ctx context.Context, startIndex uint64) (uint64,
 		return 0, err
 	}
 
-	logger.Infof("Starting to index blocks from %d to %d", ixRange.start, ixRange.end)
+	logger.Infof("Starting history indexing: from=%d, to=%d", ixRange.start, ixRange.end)
 
 	for i := ixRange.start; i <= ixRange.end; i = i + ci.params.BatchSize {
 		if err := ci.indexBatch(ctx, states, i, ixRange); err != nil {
-			return 0, err
+			return 0, errors.Wrapf(err, "indexBatch: from=%d, to=%d", i, min(i+ci.params.BatchSize-1, ixRange.end))
 		}
 
 		// in the second to last run of the loop update lastIndex to get the blocks
@@ -190,6 +190,7 @@ func (ci *Engine) IndexHistory(ctx context.Context, startIndex uint64) (uint64,
 func (ci *Engine) indexBatch(
 	ctx context.Context, states *database.DBStates, batchIx uint64, ixRange *indexRange,
 ) error {
+	batchStart := time.Now()
 	lastBlockNumInRound := min(batchIx+ci.params.BatchSize-1, ixRange.end)
 
 	bBatch, err := ci.obtainBlocksBatch(ctx, batchIx, lastBlockNumInRound)
@@ -217,6 +218,7 @@ func (ci *Engine) indexBatch(
 		batchIx,
 		lastBlockNumInRound,
 		lastForDBTimestamp,
+		batchStart,
 	)
 }
 
@@ -255,8 +257,8 @@ func (ci *Engine) obtainBlocksBatch(ctx context.Context, firstBlockNumber uint64
 		return nil, err
 	}
 
-	logger.Infof(
-		"Successfully obtained blocks %d to %d in %d milliseconds",
+	logger.Debugf(
+		"Fetched blocks: from=%d, to=%d, duration_ms=%d",
 		firstBlockNumber, lastBlockNumInRound, time.Since(startTime).Milliseconds(),
 	)
 
@@ -268,8 +270,8 @@ func (ci *Engine) processBlocksBatch(bBatch *blockBatch) *transactionsBatch {
 	txBatch := new(transactionsBatch)
 
 	ci.processBlocks(bBatch, txBatch)
-	logger.Infof(
-		"Successfully extracted %d transactions in %d milliseconds",
+	logger.Debugf(
+		"Extracted transactions: count=%d, duration_ms=%d",
 		len(txBatch.transactions), time.Since(startTime).Milliseconds(),
 	)
 
@@ -297,8 +299,8 @@ func (ci *Engine) processTransactionsBatch(
 		return err
 	}
 
-	logger.Infof(
-		"Checked receipts of %d transactions in %d milliseconds",
+	logger.Debugf(
+		"Checked receipts: count=%d, duration_ms=%d",
 		countReceipts(txBatch), time.Since(startTime).Milliseconds(),
 	)
 
@@ -337,8 +339,8 @@ func (ci *Engine) obtainLogsBatch(
 		}
 	}
 
-	logger.Infof(
-		"Obtained %d logs by request in %d milliseconds",
+	logger.Debugf(
+		"Fetched logs: count=%d, duration_ms=%d",
 		len(lgBatch.logs), time.Since(startTime).Milliseconds(),
 	)
 
@@ -393,9 +395,8 @@ func (ci *Engine) processAndSave(
 	lgBatch *logsBatch,
 	states *database.DBStates,
 	firstBlockNum, lastDBIndex, lastDBTimestamp uint64,
+	batchStart time.Time,
 ) error {
-	startTime := time.Now()
-
 	data := newDatabaseStructData()
 	data.Blocks = ci.convertBlocksToDB(bBatch)
 
@@ -408,25 +409,26 @@ func (ci *Engine) processAndSave(
 		return errors.Wrap(err, "ci.processLogs")
 	}
 
-	logger.Infof(
-		"Processed %d blocks with %d transactions and extracted %d logs from receipts and %d new logs from requests in %d milliseconds",
-		len(data.Blocks),
-		len(txBatch.transactions),
-		numLogsFromReceipts,
-		len(data.Logs)-numLogsFromReceipts,
-		time.Since(startTime).Milliseconds(),
-	)
-
-	// Push transactions and logs in the database
-	startTime = time.Now()
+	saveStart := time.Now()
 	if err := ci.saveData(data, states, lastDBIndex, lastDBTimestamp); err != nil {
 		return errors.Wrap(err, "ci.saveData")
 	}
 
-	logger.Infof(
-		"Saved %d transactions and %d logs in the DB in %d milliseconds",
+	logger.Debugf(
+		"Saved batch: transactions=%d, logs=%d, duration_ms=%d",
 		len(data.Transactions), len(data.Logs),
-		time.Since(startTime).Milliseconds(),
+		time.Since(saveStart).Milliseconds(),
+	)
+
+	logger.Infof(
+		"Processed batch: from=%d, to=%d, blocks=%d, transactions=%d, receipt_logs=%d, filter_logs=%d, duration_ms=%d",
+		firstBlockNum,
+		lastDBIndex,
+		len(data.Blocks),
+		len(txBatch.transactions),
+		numLogsFromReceipts,
+		len(data.Logs)-numLogsFromReceipts,
+		time.Since(batchStart).Milliseconds(),
 	)
 
 	return nil
@@ -450,7 +452,7 @@ func (ci *Engine) updateLastIndexHistory(
 
 	if lastChainIndex > ixRange.end && ci.params.StopIndex > ixRange.end {
 		ixRange.end = min(lastChainIndex, ci.params.StopIndex)
-		logger.Infof("Updating the last block to %d", ixRange.end)
+		logger.Infof("Extending history range: last_block=%d", ixRange.end)
 	}
 
 	return ixRange, nil
@@ -467,14 +469,13 @@ func (ci *Engine) IndexContinuous(ctx context.Context, startIndex uint64) error
 		return errors.Wrap(err, "ci.getIndexRange")
 	}
 
-	logger.Infof("Continuously indexing blocks from %d", ixRange.start)
+	logger.Infof("Starting continuous indexing: from=%d", ixRange.start)
 
 	// Request blocks one by one
 	blockNum := ixRange.start
 	lastProcessedBlockTime := [2]time.Time{time.Now(), time.Now()}
 	for blockNum <= ci.params.StopIndex {
 		if blockNum > ixRange.end {
-			logger.Debugf("Up to date, last block %d", states.States[database.LastChainIndexState].Index)
 			time.Sleep(time.Millisecond * time.Duration(ci.params.NewBlockCheckMillis))
 
 			ixRange, err = ci.updateLastIndexContinuous(ctx, states, ixRange)
@@ -485,7 +486,7 @@ func (ci *Engine) IndexContinuous(ctx context.Context, startIndex uint64) error
 			elapsed := time.Since(lastProcessedBlockTime[0]).Seconds()
 			delay := ci.params.NoNewBlocksDelayWarning
 			if delay != 0 && elapsed > delay {
-				logger.Warnf("Warning: %.2f seconds have elapsed since the last block was processed.", time.Since(lastProcessedBlockTime[1]).Seconds())
+				logger.Warnf("No new blocks: elapsed_seconds=%.2f", time.Since(lastProcessedBlockTime[1]).Seconds())
 				lastProcessedBlockTime[0] = time.Now()
 			}
 
@@ -501,15 +502,16 @@ func (ci *Engine) IndexContinuous(ctx context.Context, startIndex uint64) error
 		blockNum++
 	}
 
-	logger.Debugf("Stopping the indexer at block %d", blockNum)
+	logger.Debugf("Stopping continuous indexing: block=%d", blockNum)
 
 	return nil
 }
 
 func (ci *Engine) indexContinuousIteration(ctx context.Context, states *database.DBStates, index uint64) error {
+	startTime := time.Now()
 	block, err := ci.fetchBlock(ctx, &index)
 	if err != nil {
-		return errors.Wrap(err, "ci.fetchBlock")
+		return errors.Wrapf(err, "fetchBlock: block=%d", index)
 	}
 
 	bBatch := &blockBatch{blocks: []*chain.Block{block}}
@@ -519,36 +521,41 @@ func (ci *Engine) indexContinuousIteration(ctx context.Context, states *database
 
 	err = ci.getTransactionsReceipt(ctx, txBatch, 0, len(txBatch.transactions))
 	if err != nil {
-		return errors.Wrap(err, "ci.getTransactionsReceipt")
+		return errors.Wrapf(err, "getTransactionsReceipt: block=%d", index)
 	}
 
 	logsBatch := new(logsBatch)
 	for _, logInfo := range ci.params.CollectLogs {
 		err = ci.requestLogs(ctx, logsBatch, logInfo, index, index+1, index)
 		if err != nil {
-			return fmt.Errorf("IndexContinuous: %w", err)
+			return errors.Wrapf(err, "requestLogs: block=%d", index)
 		}
 	}
 
 	data := newDatabaseStructData()
 	data.Blocks = ci.convertBlocksToDB(bBatch)
 
 	if err := ci.processTransactions(txBatch, data); err != nil {
-		return fmt.Errorf("IndexContinuous: %w", err)
+		return errors.Wrapf(err, "processTransactions: block=%d", index)
 	}
 
 	err = ci.processLogs(logsBatch, bBatch, index, data)
 	if err != nil {
-		return fmt.Errorf("IndexContinuous: %w", err)
+		return errors.Wrapf(err, "processLogs: block=%d", index)
 	}
 
 	indexTimestamp := bBatch.blocks[0].Time()
 	if err := ci.saveData(data, states, index, indexTimestamp); err != nil {
-		return fmt.Errorf("IndexContinuous: %w", err)
+		return errors.Wrapf(err, "saveData: block=%d", index)
 	}
 
+	logger.Debugf(
+		"Indexed block: block=%d, transactions=%d, logs=%d, duration_ms=%d",
+		index, len(data.Transactions), len(data.Logs), time.Since(startTime).Milliseconds(),
+	)
+
 	if index%1000 == 0 {
-		logger.Infof("Indexer at block %d", index)
+		logger.Infof("Continuous progress: block=%d", index)
 	}
 
 	return nil

internal/database/history_drop.go

@@ -25,15 +25,14 @@ func DropHistory(
 	client *chain.Client,
 ) {
 	for {
-		logger.Infof("starting DropHistory iteration")
+		logger.Infof("Starting history drop iteration")
 
 		startTime := time.Now()
 		err := dropHistoryIteration(ctx, db, intervalSeconds, client)
 		if err == nil {
-			duration := time.Since(startTime)
-			logger.Infof("finished DropHistory iteration in %v", duration)
+			logger.Infof("Finished history drop iteration: duration_ms=%d", time.Since(startTime).Milliseconds())
 		} else {
-			logger.Errorf("DropHistory error: %s", err)
+			logger.Errorf("History drop error: %s", err)
 		}
 
 		time.Sleep(time.Duration(checkInterval) * time.Second)
@@ -134,7 +133,7 @@ func DeleteInBatches(db *gorm.DB, deleteStartTime uint64, entity interface{}) er
 		// Take a rest every so often to avoid locking up the database too much
 		batchCount++
 		if batchCount%deleteBatchesPauseAfter == 0 {
-			logger.Debugf("Deleted %d rows of %T so far", batchCount*deleteBatchSize, entity)
+			logger.Debugf("History drop progress: entity=%T, deleted=%d", entity, batchCount*deleteBatchSize)
 			time.Sleep(deleteBatchesPauseDuration)
 		}
 	}

internal/diagnostics/collection.go

@@ -21,7 +21,7 @@ func LogIndexerPolicy(cfg config.IndexerConfig) {
 	for i := range cfg.CollectTransactions {
 		tx := &cfg.CollectTransactions[i]
 		logger.Infof(
-			"  tx_filter: contract=%s func_sig=%s status=%t collect_events=%t",
+			"  tx_filter: contract=%s, func_sig=%s, status=%t, collect_events=%t",
 			contractRef(tx.ContractName, tx.ContractAddress),
 			formatHexOrAny(tx.FuncSig),
 			tx.Status,
@@ -31,7 +31,7 @@ func LogIndexerPolicy(cfg config.IndexerConfig) {
 	for i := range cfg.CollectLogs {
 		lg := &cfg.CollectLogs[i]
 		logger.Infof(
-			"  log_filter: contract=%s topic=%s",
+			"  log_filter: contract=%s, topic=%s",
 			contractRef(lg.ContractName, lg.ContractAddress),
 			formatHexOrAny(lg.Topic),
 		)
@@ -45,7 +45,7 @@ func LogFspEventFilter(logs []config.LogInfo) {
 	for i := range logs {
 		lg := &logs[i]
 		logger.Infof(
-			"  fsp_event_filter: contract=%s topic=%s",
+			"  fsp_event_filter: contract=%s, topic=%s",
 			contractRef(lg.ContractName, lg.ContractAddress),
 			formatHexOrAny(lg.Topic),
 		)

internal/fsp/backfill_events.go

@@ -34,7 +34,7 @@ func backfillEventRangesLogs(
 		eventRangeInserted := 0
 
 		logger.Infof(
-			"FSP event indexing started: %d/%d from=%d to=%d",
+			"FSP event indexing started: progress=%d/%d, from=%d, to=%d",
 			eventRangeIx+1,
 			len(eventRanges),
 			eventRange.from,
@@ -61,7 +61,7 @@ func backfillEventRangesLogs(
 		}
 
 		logger.Infof(
-			"FSP event indexing completed: %d/%d from=%d to=%d inserted=%d duration_ms=%d",
+			"FSP event indexing completed: progress=%d/%d, from=%d, to=%d, inserted=%d, duration_ms=%d",
 			eventRangeIx+1,
 			len(eventRanges),
 			eventRange.from,

internal/fsp/runtime.go

@@ -53,7 +53,7 @@ func RunIndexer(
 
 	historyDropSeconds := historyDropHeuristicSeconds(chainID, cfg.Indexer.HistoryEpochs)
 	logger.Infof(
-		"Using FSP history drop: chain_id=%s, history_epochs=%d, derived retention=%ds (%.2f days)",
+		"Using FSP history drop: chain_id=%s, history_epochs=%d, derived_retention_seconds=%d, derived_retention_days=%.2f",
 		chainID,
 		cfg.Indexer.HistoryEpochs,
 		historyDropSeconds,

internal/fsp/startup.go

@@ -68,7 +68,7 @@ func IndexStartup(ctx context.Context, ci *core.Engine) (uint64, error) {
 	backfillEventRanges := !database.IsSet(firstFspEvent) || firstFspEvent.Index > eventStartBlock
 
 	logger.Infof(
-		"FSP startup plan: catchup blocks from=%d, latest confirmed=%d, backfill FSP event ranges=%t, ranges=%+v",
+		"FSP startup plan: catchup_from=%d, latest_confirmed=%d, backfill_event_ranges=%t, ranges=%+v",
 		catchupFromBlock,
 		latestConfirmedNumber,
 		backfillEventRanges,
@@ -102,14 +102,14 @@ func IndexStartup(ctx context.Context, ci *core.Engine) (uint64, error) {
 		}
 	} else {
 		logger.Infof(
-			"Skipping FSP catchup block backfill: start=%d is above latest confirmed=%d",
+			"Skipping FSP catchup block backfill: start=%d, latest_confirmed=%d",
 			catchupFromBlock,
 			latestConfirmedNumber,
 		)
 	}
 
 	logger.Infof(
-		"FSP startup backfill complete: targetFullStart=%d targetEventStart=%d lastIndexed=%d",
+		"FSP startup backfill complete: target_full_start=%d, target_event_start=%d, last_indexed=%d",
 		fullStartBlock,
 		eventStartBlock,
 		lastIndexed,