refactor(logger): format logging (#12)

* feat(logger): format main logger * feat(logger): use duration ms for gcp output * refactor(logger): bitcoin node logger * refactor(logger): indexer logger * refactor(logger): fix cmd logger * refactor(logger): logger in config pacakge * refactor(logger): set pgx error log level debug * refactor(logger): btcclient datasource * refactor: processor name * refactor(logger): runese logger * refactor(logger): update logger * fix(runes): wrong btc db datasource * refactor(logger): remove unnecessary debug log * refactor: update logger in indexer * fix(logger): deadlock in load() * fix: remove unused --------- Co-authored-by: Gaze <gazenw@users.noreply.github.com>
2026-04-30 20:52:01 +08:00 · 2024-04-26 23:23:37 +07:00
parent 69ab16c35a
commit fe6988627a
17 changed files with 245 additions and 156 deletions
--- a/modules/bitcoin/btcclient/client_db.go
+++ b/modules/bitcoin/btcclient/client_db.go
@@ -37,13 +37,13 @@ func NewClientDatabase(bitcoinDg datagateway.BitcoinDataGateway) *ClientDatabase
 	}
 }

-func (c ClientDatabase) Name() string {
-	return "BitcoinDatabase"
+func (d ClientDatabase) Name() string {
+	return "bitcoin_database"
 }

-func (c *ClientDatabase) Fetch(ctx context.Context, from, to int64) ([]*types.Block, error) {
+func (d *ClientDatabase) Fetch(ctx context.Context, from, to int64) ([]*types.Block, error) {
 	ch := make(chan []*types.Block)
-	subscription, err := c.FetchAsync(ctx, from, to, ch)
+	subscription, err := d.FetchAsync(ctx, from, to, ch)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -73,8 +73,13 @@ func (c *ClientDatabase) Fetch(ctx context.Context, from, to int64) ([]*types.Bl
 	}
 }

-func (c *ClientDatabase) FetchAsync(ctx context.Context, from, to int64, ch chan<- []*types.Block) (*subscription.ClientSubscription[[]*types.Block], error) {
-	from, to, skip, err := c.prepareRange(ctx, from, to)
+func (d *ClientDatabase) FetchAsync(ctx context.Context, from, to int64, ch chan<- []*types.Block) (*subscription.ClientSubscription[[]*types.Block], error) {
+	ctx = logger.WithContext(ctx,
+		slogx.String("package", "datasources"),
+		slogx.String("datasource", d.Name()),
+	)
+
+	from, to, skip, err := d.prepareRange(ctx, from, to)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to prepare fetch range")
 	}
@@ -129,10 +134,10 @@ func (c *ClientDatabase) FetchAsync(ctx context.Context, from, to int64, ch chan
 					if errors.Is(err, errs.Closed) {
 						return
 					}
-					logger.WarnContext(ctx, "failed while dispatch block",
-						slogx.Error(err),
+					logger.WarnContext(ctx, "Failed to send bitcoin blocks to subscription client",
 						slogx.Int64("start", data[0].Header.Height),
 						slogx.Int64("end", data[len(data)-1].Header.Height),
+						slogx.Error(err),
 					)
 				}
 			case <-ctx.Done():
@@ -159,16 +164,26 @@ func (c *ClientDatabase) FetchAsync(ctx context.Context, from, to int64, ch chan
 					continue
 				}
 				stream.Go(func() []*types.Block {
+					startAt := time.Now()
+					defer func() {
+						logger.DebugContext(ctx, "Fetched chunk of blocks from Bitcoin node",
+							slogx.Int("total_blocks", len(chunk)),
+							slogx.Int64("from", chunk[0]),
+							slogx.Int64("to", chunk[len(chunk)-1]),
+							slogx.Duration("duration", time.Since(startAt)),
+						)
+					}()
+
 					fromHeight, toHeight := chunk[0], chunk[len(chunk)-1]
-					blocks, err := c.bitcoinDg.GetBlocksByHeightRange(ctx, fromHeight, toHeight)
+					blocks, err := d.bitcoinDg.GetBlocksByHeightRange(ctx, fromHeight, toHeight)
 					if err != nil {
-						logger.ErrorContext(ctx, "failed to get blocks",
+						logger.ErrorContext(ctx, "Can't get block data from Bitcoin database",
 							slogx.Error(err),
-							slogx.Int64("from_height", fromHeight),
-							slogx.Int64("to_height", toHeight),
+							slogx.Int64("from", fromHeight),
+							slogx.Int64("to", toHeight),
 						)
 						if err := subscription.SendError(ctx, errors.Wrapf(err, "failed to get blocks: from_height: %d, to_height: %d", fromHeight, toHeight)); err != nil {
-							logger.ErrorContext(ctx, "failed to send error", slogx.Error(err))
+							logger.WarnContext(ctx, "Failed to send datasource error to subscription client", slogx.Error(err))
 						}
 						return nil
 					}
--- a/modules/bitcoin/processor.go
+++ b/modules/bitcoin/processor.go
@@ -29,7 +29,7 @@ func NewProcessor(config config.Config, bitcoinDg datagateway.BitcoinDataGateway
 }

 func (p Processor) Name() string {
-	return "Bitcoin"
+	return "bitcoin"
 }

 func (p *Processor) Process(ctx context.Context, inputs []*types.Block) error {
--- a/modules/runes/processor.go
+++ b/modules/runes/processor.go
@@ -16,6 +16,7 @@ import (
 	"github.com/gaze-network/indexer-network/modules/runes/internal/entity"
 	"github.com/gaze-network/indexer-network/modules/runes/runes"
 	"github.com/gaze-network/indexer-network/pkg/logger"
+	"github.com/gaze-network/indexer-network/pkg/logger/slogx"
 	"github.com/gaze-network/indexer-network/pkg/reportingclient"
 	"github.com/gaze-network/uint128"
 	"github.com/samber/lo"
@@ -158,7 +159,7 @@ func (p *Processor) ensureGenesisRune(ctx context.Context) error {
 }

 func (p *Processor) Name() string {
-	return "Runes"
+	return "runes"
 }

 func (p *Processor) CurrentBlock(ctx context.Context) (types.BlockHeader, error) {
@@ -192,7 +193,10 @@ func (p *Processor) RevertData(ctx context.Context, from int64) error {
 	}
 	defer func() {
 		if err := runesDgTx.Rollback(ctx); err != nil {
-			logger.ErrorContext(ctx, "failed to rollback transaction", err)
+			logger.WarnContext(ctx, "failed to rollback transaction",
+				slogx.Error(err),
+				slogx.String("event", "rollback_runes_revert"),
+			)
 		}
 	}()

--- a/modules/runes/processor_process.go
+++ b/modules/runes/processor_process.go
@@ -25,17 +25,20 @@ import (

 func (p *Processor) Process(ctx context.Context, blocks []*types.Block) error {
 	for _, block := range blocks {
-		ctx := logger.WithContext(ctx, slog.Int("block_height", int(block.Header.Height)))
-		logger.DebugContext(ctx, "[RunesProcessor] Processing block", slog.Int("txs", len(block.Transactions)))
+		ctx := logger.WithContext(ctx, slog.Int64("height", block.Header.Height))
+		logger.DebugContext(ctx, "Processing new block", slog.Int("txs", len(block.Transactions)))

 		for _, tx := range block.Transactions {
 			if err := p.processTx(ctx, tx, block.Header); err != nil {
 				return errors.Wrap(err, "failed to process tx")
 			}
 		}
+
 		if err := p.flushBlock(ctx, block.Header); err != nil {
 			return errors.Wrap(err, "failed to flush block")
 		}
+
+		logger.DebugContext(ctx, "Inserted new block")
 	}
 	return nil
 }
@@ -66,13 +69,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 		for runeId, balance := range balances {
 			unallocated[runeId] = unallocated[runeId].Add(balance.Amount)
 			p.newSpendOutPoints = append(p.newSpendOutPoints, balance.OutPoint)
-			logger.DebugContext(ctx, "[RunesProcessor] Found runes in tx input",
-				slogx.Any("runeId", runeId),
-				slogx.Stringer("amount", balance.Amount),
-				slogx.Stringer("txHash", balance.OutPoint.Hash),
-				slog.Int("txOutIndex", int(balance.OutPoint.Index)),
-				slog.Int("blockHeight", int(tx.BlockHeight)),
-			)
 		}
 	}

@@ -92,13 +88,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 		}
 		allocated[output][runeId] = allocated[output][runeId].Add(amount)
 		unallocated[runeId] = unallocated[runeId].Sub(amount)
-		logger.DebugContext(ctx, "[RunesProcessor] Allocated runes to tx output",
-			slogx.Any("runeId", runeId),
-			slogx.Stringer("amount", amount),
-			slog.Int("output", output),
-			slogx.Stringer("txHash", tx.TxHash),
-			slog.Int("blockHeight", int(tx.BlockHeight)),
-		)
 	}

 	mints := make(map[runes.RuneId]uint128.Uint128)
@@ -131,7 +120,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 				if !premine.IsZero() {
 					unallocated[etchedRuneId] = unallocated[etchedRuneId].Add(premine)
 					mints[etchedRuneId] = mints[etchedRuneId].Add(premine)
-					logger.DebugContext(ctx, "[RunesProcessor] Minted premine", slogx.Any("runeId", etchedRuneId), slogx.Stringer("amount", premine))
 				}
 			}

@@ -206,7 +194,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 		// all input runes and minted runes in a tx with cenotaph are burned
 		for runeId, amount := range unallocated {
 			burns[runeId] = burns[runeId].Add(amount)
-			logger.DebugContext(ctx, "[RunesProcessor] Burned runes in cenotaph", slogx.Any("runeId", runeId), slogx.Stringer("amount", amount))
 		}
 	} else {
 		// assign all un-allocated runes to the default output (pointer), or the first non
@@ -236,7 +223,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 			// if pointer is still nil, then no output is available. Burn all unallocated runes.
 			for runeId, amount := range unallocated {
 				burns[runeId] = burns[runeId].Add(amount)
-				logger.DebugContext(ctx, "[RunesProcessor] Burned runes to no output", slogx.Any("runeId", runeId), slogx.Stringer("amount", amount))
 			}
 		}
 	}
@@ -247,7 +233,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 			// burn all allocated runes to OP_RETURN outputs
 			for runeId, amount := range balances {
 				burns[runeId] = burns[runeId].Add(amount)
-				logger.DebugContext(ctx, "[RunesProcessor] Burned runes to OP_RETURN output", slogx.Any("runeId", runeId), slogx.Stringer("amount", amount))
 			}
 			continue
 		}
@@ -316,7 +301,6 @@ func (p *Processor) processTx(ctx context.Context, tx *types.Transaction, blockH
 		}
 	}
 	p.newRuneTxs = append(p.newRuneTxs, &runeTx)
-	logger.DebugContext(ctx, "[RunesProcessor] created RuneTransaction", slogx.Any("runeTx", runeTx))
 	return nil
 }

@@ -413,7 +397,6 @@ func (p *Processor) mint(ctx context.Context, runeId runes.RuneId, blockHeader t
 	if err := p.incrementMintCount(ctx, runeId, blockHeader); err != nil {
 		return uint128.Zero, errors.Wrap(err, "failed to increment mint count")
 	}
-	logger.DebugContext(ctx, "[RunesProcessor] Minted rune", slogx.Any("runeId", runeId), slogx.Stringer("amount", amount), slogx.Stringer("mintCount", runeEntry.Mints), slogx.Stringer("cap", lo.FromPtr(runeEntry.Terms.Cap)))
 	return amount, nil
 }

@@ -425,11 +408,9 @@ func (p *Processor) getEtchedRune(ctx context.Context, tx *types.Transaction, ru
 	if rune != nil {
 		minimumRune := runes.MinimumRuneAtHeight(p.network, uint64(tx.BlockHeight))
 		if rune.Cmp(minimumRune) < 0 {
-			logger.DebugContext(ctx, "invalid etching: rune is lower than minimum rune at this height", slogx.Any("rune", rune), slogx.Any("minimumRune", minimumRune))
 			return nil, runes.RuneId{}, runes.Rune{}, nil
 		}
 		if rune.IsReserved() {
-			logger.DebugContext(ctx, "invalid etching: rune is reserved", slogx.Any("rune", rune))
 			return nil, runes.RuneId{}, runes.Rune{}, nil
 		}

@@ -438,7 +419,6 @@ func (p *Processor) getEtchedRune(ctx context.Context, tx *types.Transaction, ru
 			return nil, runes.RuneId{}, runes.Rune{}, errors.Wrap(err, "error during check rune existence")
 		}
 		if ok {
-			logger.DebugContext(ctx, "invalid etching: rune already exists", slogx.Any("rune", rune))
 			return nil, runes.RuneId{}, runes.Rune{}, nil
 		}

@@ -448,7 +428,6 @@ func (p *Processor) getEtchedRune(ctx context.Context, tx *types.Transaction, ru
 			return nil, runes.RuneId{}, runes.Rune{}, errors.Wrap(err, "error during check tx commits to rune")
 		}
 		if !commit {
-			logger.DebugContext(ctx, "invalid etching: tx does not commit to the rune", slogx.Any("rune", rune))
 			return nil, runes.RuneId{}, runes.Rune{}, nil
 		}
 	} else {
@@ -464,7 +443,7 @@ func (p *Processor) getEtchedRune(ctx context.Context, tx *types.Transaction, ru

 func (p *Processor) txCommitsToRune(ctx context.Context, tx *types.Transaction, rune runes.Rune) (bool, error) {
 	commitment := rune.Commitment()
-	for _, txIn := range tx.TxIn {
+	for i, txIn := range tx.TxIn {
 		tapscript, ok := extractTapScript(txIn.Witness)
 		if !ok {
 			continue
@@ -492,8 +471,7 @@ func (p *Processor) txCommitsToRune(ctx context.Context, tx *types.Transaction,
 				continue
 			}
 			if err != nil {
-				logger.ErrorContext(ctx, "failed to get pk script at out point", err)
-				continue
+				return false, errors.Wrapf(err, "can't get previous txout for txin `%v:%v`", tx.TxHash.String(), i)
 			}
 			pkScript := prevTx.TxOut[txIn.PreviousOutIndex].PkScript
 			// input utxo must be P2TR
@@ -576,7 +554,6 @@ func (p *Processor) createRuneEntry(ctx context.Context, runestone *runes.Runest
 	}
 	p.newRuneEntries[runeId] = runeEntry
 	p.newRuneEntryStates[runeId] = runeEntry
-	logger.DebugContext(ctx, "[RunesProcessor] created RuneEntry", slogx.Any("runeEntry", runeEntry))
 	return nil
 }

@@ -630,7 +607,6 @@ func (p *Processor) incrementBurnedAmount(ctx context.Context, burned map[runes.
 			continue
 		}
 		runeEntry.BurnedAmount = runeEntry.BurnedAmount.Add(amount)
-		logger.DebugContext(ctx, "[RunesProcessor] burned amount incremented", slogx.Any("runeId", runeId), slogx.Any("amount", amount))
 		p.newRuneEntryStates[runeId] = runeEntry
 	}
 	return nil
@@ -698,7 +674,10 @@ func (p *Processor) flushBlock(ctx context.Context, blockHeader types.BlockHeade
 	}
 	defer func() {
 		if err := runesDgTx.Rollback(ctx); err != nil {
-			logger.ErrorContext(ctx, "[RunesProcessor] failed to rollback runes tx", err)
+			logger.WarnContext(ctx, "failed to rollback transaction",
+				slogx.Error(err),
+				slogx.String("event", "rollback_runes_insertion"),
+			)
 		}
 	}()

@@ -824,6 +803,5 @@ func (p *Processor) flushBlock(ctx context.Context, blockHeader types.BlockHeade
 			return errors.Wrap(err, "failed to submit block report")
 		}
 	}
-	logger.InfoContext(ctx, "[RunesProcessor] block flushed")
 	return nil
 }
--- a/modules/runes/repository/postgres/tx.go
+++ b/modules/runes/repository/postgres/tx.go
@@ -55,7 +55,7 @@ func (r *Repository) Rollback(ctx context.Context) error {
 		return errors.Wrap(err, "failed to rollback transaction")
 	}
 	if err == nil {
-		logger.InfoContext(ctx, "rolled back transaction")
+		logger.DebugContext(ctx, "rolled back transaction")
 	}
 	r.tx = nil
 	return nil