Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: better logging for tx, provider, query, processor #120

Merged
merged 3 commits into from
Aug 9, 2023
Merged

feat: better logging for tx, provider, query, processor #120

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fb1ec28
feat: better logging for tx, provider, query, processor
Aug 1, 2023
c181859
Merge branch 'main' into feat/improve-logging
izyak Aug 2, 2023
77166fd
Merge branch 'main' into feat/improve-logging
izyak Aug 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 57 additions & 29 deletions relayer/chains/icon/icon_chain_processor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ type Verifier struct {

func NewIconChainProcessor(log *zap.Logger, provider *IconProvider, metrics *processor.PrometheusMetrics) *IconChainProcessor {
return &IconChainProcessor{
log: log.With(zap.String("chain_name", "Icon")),
log: log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
chainProvider: provider,
latestClientState: make(latestClientState),
connectionStateCache: make(processor.ConnectionStateCache),
Expand Down Expand Up @@ -149,7 +149,7 @@ func (icp *IconChainProcessor) Run(ctx context.Context, initialBlockHistory uint
}

// start_query_cycle
icp.log.Debug(" **************** Entering main query loop **************** ")
icp.log.Debug("Starting query cycle")
err := icp.monitoring(ctx, &persistence)
return err
}
Expand All @@ -168,6 +168,14 @@ func (icp *IconChainProcessor) StartFromHeight(ctx context.Context) int {
return snapshotHeight
}

func (icp *IconChainProcessor) getLastSavedHeight() int {
snapshotHeight, err := rlycommon.LoadSnapshotHeight(icp.Provider().ChainId())
if err != nil || snapshotHeight < 0 {
return 0
}
return snapshotHeight
}

func (icp *IconChainProcessor) initializeConnectionState(ctx context.Context) error {
// TODO: review
ctx, cancel := context.WithTimeout(ctx, queryTimeout)
Expand All @@ -187,9 +195,9 @@ func (icp *IconChainProcessor) initializeConnectionState(ctx context.Context) er
CounterpartyClientID: c.Counterparty.ClientId,
}] = c.State == conntypes.OPEN

icp.log.Info("found connection",
zap.String("ClientId ", c.ClientId),
zap.String("ConnectionID ", c.Id),
icp.log.Debug("Found open connection",
zap.String("client-id ", c.ClientId),
zap.String("connection-id ", c.Id),
)
}
return nil
Expand Down Expand Up @@ -221,11 +229,11 @@ func (icp *IconChainProcessor) initializeChannelState(ctx context.Context) error
CounterpartyPortID: ch.Counterparty.PortId,
}] = ch.State == chantypes.OPEN

icp.log.Info("Found channel",
zap.String("channelID", ch.ChannelId),
zap.String("Port id ", ch.PortId))
zap.String("Counterparty Channel Id ", ch.Counterparty.ChannelId)
zap.String("Counterparty Port Id", ch.Counterparty.PortId)
icp.log.Debug("Found open channel",
zap.String("channel-id", ch.ChannelId),
zap.String("port-id ", ch.PortId),
zap.String("counterparty-channel-id", ch.Counterparty.ChannelId),
zap.String("counterparty-port-id", ch.Counterparty.PortId))
}

return nil
Expand Down Expand Up @@ -276,7 +284,7 @@ func (icp *IconChainProcessor) monitoring(ctx context.Context, persistence *quer
}
// }

icp.log.Debug("Start to query from height", zap.Int64("height", processedheight))
icp.log.Info("Start to query from height", zap.Int64("height", processedheight))
// subscribe to monitor block
ctxMonitorBlock, cancelMonitorBlock := context.WithCancel(ctx)
reconnect()
Expand Down Expand Up @@ -304,7 +312,7 @@ loop:

go func(ctx context.Context, cancel context.CancelFunc) {
blockReq.Height = types.NewHexInt(processedheight)
icp.log.Debug("Querying Height", zap.Int64("height", processedheight))
icp.log.Debug("Try to reconnect from", zap.Int64("height", processedheight))
err := icp.chainProvider.client.MonitorBlock(ctx, blockReq, func(conn *websocket.Conn, v *types.BlockNotification) error {
if !errors.Is(ctx.Err(), context.Canceled) {
btpBlockNotifCh <- v
Expand All @@ -313,7 +321,10 @@ loop:
}, func(conn *websocket.Conn) {
}, func(conn *websocket.Conn, err error) {})
if err != nil {
icp.SnapshotHeight(int(processedheight) - 5)
ht := icp.getHeightToSave(processedheight)
if ht != icp.getLastSavedHeight() {
icp.SnapshotHeight(ht)
}
if errors.Is(err, context.Canceled) {
return
}
Expand All @@ -329,8 +340,8 @@ loop:
err := icp.verifyBlock(ctx, br.Header)
if err != nil {
reconnect()
icp.log.Warn("failed to Verify BTP Block",
zap.Int64("got", br.Height),
icp.log.Warn("Failed to verify BTP Block",
zap.Int64("height", br.Height),
zap.Error(err),
)
break
Expand All @@ -348,8 +359,7 @@ loop:
}

ibcHeaderCache[uint64(br.Height)] = br.Header
icp.log.Info("Queried Latest height: ",
zap.String("chain id ", icp.chainProvider.ChainId()),
icp.log.Debug("Queried block ",
zap.Int64("height", br.Height))
err = icp.handlePathProcessorUpdate(ctx, br.Header, ibcMessageCache, ibcHeaderCache.Clone())
if err != nil {
Expand All @@ -364,7 +374,10 @@ loop:
if br = nil; len(btpBlockRespCh) > 0 {
br = <-btpBlockRespCh
}
icp.SnapshotHeight(int(icp.latestBlock.Height) - 5)
ht, takeSnapshot := icp.shouldSnapshot(int(icp.latestBlock.Height))
if takeSnapshot {
icp.SnapshotHeight(ht)
}
}
// remove unprocessed blockResponses
for len(btpBlockRespCh) > 0 {
Expand Down Expand Up @@ -455,19 +468,32 @@ loop:
}
}

func (icp *IconChainProcessor) SnapshotHeight(height int) {

func (icp *IconChainProcessor) shouldSnapshot(height int) (int, bool) {
blockInterval := icp.Provider().ProviderConfig().GetBlockInterval()
snapshotThreshold := rlycommon.ONE_HOUR / int(blockInterval)

retryAfter := icp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
snapshotHeight := height - int(retryAfter)
snapshotHeight := icp.getHeightToSave(int64(height))

if snapshotHeight%snapshotThreshold == 0 {
err := rlycommon.SnapshotHeight(icp.Provider().ChainId(), height)
if err != nil {
icp.log.Warn("Failed saving height snapshot for height", zap.Int("height", height))
}
return snapshotHeight, true
}
return 0, false
}

func (icp *IconChainProcessor) getHeightToSave(height int64) int {
retryAfter := icp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
ht := int(height - int64(retryAfter))
if ht < 0 {
return 0
}
return ht
}

func (icp *IconChainProcessor) SnapshotHeight(height int) {
icp.log.Info("Save height for snapshot", zap.Int("height", height))
err := rlycommon.SnapshotHeight(icp.Provider().ChainId(), height)
if err != nil {
icp.log.Warn("Failed saving height snapshot for height", zap.Int("height", height))
}
}

Expand Down Expand Up @@ -526,6 +552,8 @@ func (icp *IconChainProcessor) verifyBlock(ctx context.Context, ibcHeader provid
icp.verifier.nextProofContext = header.Validators
icp.verifier.verifiedHeight = int64(header.Height())
icp.verifier.prevNetworkSectionHash = types.NewNetworkSection(header.Header).Hash()
icp.log.Debug("Verified block ",
zap.Uint64("height", header.Height()))
return nil
}

Expand Down Expand Up @@ -602,8 +630,8 @@ func (icp *IconChainProcessor) handleBTPBlockRequest(
request.err = errors.Wrapf(err, "event.UnmarshalFromBytes: %v", err)
return
}
icp.log.Info("Detected eventlog: ", zap.Int64("Height", request.height),
zap.String("Eventlog", string(el.Indexed[0])))
icp.log.Info("Detected eventlog ", zap.Int64("height", request.height),
zap.String("eventlog", IconCosmosEventMap[string(el.Indexed[0])]))
eventlogs = append(eventlogs, el)
}

Expand All @@ -629,7 +657,7 @@ func (icp *IconChainProcessor) handleBTPBlockRequest(
request.response.IsProcessed = processed
return
}
request.err = errors.Wrapf(err, "failed to get btp header: %v", err)
request.err = errors.Wrapf(err, "Failed to get btp header: %v", err)
return
}
request.response.Header = NewIconIBCHeader(btpHeader, validators, int64(btpHeader.MainHeight))
Expand Down
2 changes: 0 additions & 2 deletions relayer/chains/icon/msg.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,5 @@ func (icp *IconProvider) NewIconMessage(msg interface{}, method string) provider
Method: method,
}

// icp.log.Debug("Icon Message ", zap.String("method", method), zap.Any("message ", msg))

return im
}
3 changes: 1 addition & 2 deletions relayer/chains/icon/provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,6 @@ func (pp *IconProviderConfig) GetFirstRetryBlockAfter() uint64 {
return 8
}

// NewProvider should provide a new Icon provider
// NewProvider should provide a new Icon provider
func (pp *IconProviderConfig) NewProvider(log *zap.Logger, homepath string, debug bool, chainName string) (provider.ChainProvider, error) {

Expand All @@ -118,7 +117,7 @@ func (pp *IconProviderConfig) NewProvider(log *zap.Logger, homepath string, debu
codec := MakeCodec(ModuleBasics, []string{})

return &IconProvider{
log: log.With(zap.String("sys", "chain_client")),
log: log.With(zap.String("chain_id", pp.ChainID)),
client: NewClient(pp.getRPCAddr(), log),
PCfg: pp,
wallet: wallet,
Expand Down
6 changes: 3 additions & 3 deletions relayer/chains/icon/query.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -615,20 +615,20 @@ func (icp *IconProvider) QueryChannels(ctx context.Context) ([]*chantypes.Identi
"portId": portId,
}), &_channel)
if err != nil {
icp.log.Error("unable to fetch channel for ", zap.String("channel id ", channelId), zap.Error(err))
icp.log.Error("unable to fetch channel for ", zap.String("channel-id ", channelId), zap.Error(err))
continue
}

if _channel == "" {
icp.log.Debug("channel not present for ", zap.String("Channel id ", channelId), zap.String("port id ", portId))
icp.log.Debug("Channel not present for ", zap.String("channel-id ", channelId), zap.String("port-id ", portId))
continue
}

var channel chantypes.Channel
_, err = HexBytesToProtoUnmarshal(_channel, &channel)
if err != nil {
icp.log.Info("Unable to unmarshal channel for ",
zap.String("channel id ", channelId), zap.Error(err))
zap.String("channel-id ", channelId), zap.Error(err))
continue
}

Expand Down
7 changes: 3 additions & 4 deletions relayer/chains/wasm/tx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -920,7 +920,7 @@ func (ap *WasmProvider) buildMessages(clientCtx client.Context, txf tx.Factory,
}

txf = txf.WithGas(adjusted)
_, _ = fmt.Fprintf(os.Stderr, "%s\n", tx.GasEstimateResponse{GasEstimate: txf.Gas()})
// _, _ = fmt.Fprintf(os.Stderr, "%s\n", tx.GasEstimateResponse{GasEstimate: txf.Gas()})
}

if clientCtx.Simulate {
Expand Down Expand Up @@ -1003,9 +1003,8 @@ func (ap *WasmProvider) BroadcastTx(

ap.log.Info("Submitted transaction",
zap.String("chain_id", ap.PCfg.ChainID),
zap.String("txHash", res.TxHash),
zap.Int64("Height", res.Height),
zap.Any("Methods called", msgTypesField(msgs)),
zap.String("tx_hash", res.TxHash),
msgTypesField(msgs),
)

if shouldWait {
Expand Down
Loading