// SUCCESS indicates the rpc calling is successful.
SUCCESS = "success"
// FAIL indicated the rpc calling is failed.
- FAIL = "fail"
- crosscoreRPCPrefix = "/rpc/"
+ FAIL = "fail"
)
// Response describes the response standard.
m.Handle("/get-merkle-proof", jsonHandler(a.getMerkleProof))
handler := latencyHandler(m, walletEnable)
- handler = maxBytesHandler(handler) // TODO(tessr): consider moving this to non-core specific mux
handler = webAssetsHandler(handler)
handler = gzip.Handler{Handler: handler}
a.handler = handler
}
-func maxBytesHandler(h http.Handler) http.Handler {
- const maxReqSize = 1e7 // 10MB
- return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
- // A block can easily be bigger than maxReqSize, but everything
- // else should be pretty small.
- if req.URL.Path != crosscoreRPCPrefix+"signer/sign-block" {
- req.Body = http.MaxBytesReader(w, req.Body, maxReqSize)
- }
- h.ServeHTTP(w, req)
- })
-}
-
// json Handler
func jsonHandler(f interface{}) http.Handler {
h, err := httpjson.Handler(f, errorFormatter.Write)
var (
latencyMu sync.Mutex
latencies = map[string]*metrics.RotatingLatency{}
-
- latencyRange = map[string]time.Duration{
- crosscoreRPCPrefix + "get-block": 20 * time.Second,
- crosscoreRPCPrefix + "signer/sign-block": 5 * time.Second,
- crosscoreRPCPrefix + "get-snapshot": 30 * time.Second,
- // the rest have a default range
- }
)
// latency returns a rotating latency histogram for the given request.
}
// Create a histogram only if the path is legit.
if _, pat := tab.Handler(req); pat == req.URL.Path {
- d, ok := latencyRange[req.URL.Path]
- if !ok {
- d = 100 * time.Millisecond
- }
+ d := 100 * time.Millisecond
l := metrics.NewRotatingLatency(5, d)
latencies[req.URL.Path] = l
metrics.PublishLatency(req.URL.Path, l)
if _, ok := f.msgSet[blockHash]; !ok {
f.msgSet[blockHash] = msg
f.queue.Push(msg, -float32(msg.block.Height))
- log.WithField("block height", msg.block.Height).Debug("fetcher receive mine block")
+ log.WithFields(log.Fields{
+ "module": logModule,
+ "block height": msg.block.Height,
+ "block hash": blockHash.String(),
+ }).Debug("blockFetcher receive mine block")
}
}
}
if err := f.peers.broadcastMinedBlock(msg.block); err != nil {
- log.WithField("err", err).Error("fail on fetcher broadcast new block")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("blockFetcher fail on broadcast new block")
return
}
}
header, err = bk.chain.GetHeaderByHeight(header.Height - step)
}
if err != nil {
- log.WithField("err", err).Error("blockKeeper fail on get blockLocator")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("blockKeeper fail on get blockLocator")
break
}
if peer != nil && checkPoint != nil && peer.Height() >= checkPoint.Height {
bk.syncPeer = peer
if err := bk.fastBlockSync(checkPoint); err != nil {
- log.WithField("err", err).Warning("fail on fastBlockSync")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on fastBlockSync")
bk.peers.errorHandler(peer.ID(), err)
return false
}
}
if err := bk.regularBlockSync(targetHeight); err != nil {
- log.WithField("err", err).Warning("fail on regularBlockSync")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on regularBlockSync")
bk.peers.errorHandler(peer.ID(), err)
return false
}
func (bk *blockKeeper) syncWorker() {
genesisBlock, err := bk.chain.GetBlockByHeight(0)
if err != nil {
- log.WithField("err", err).Error("fail on handleStatusRequestMsg get genesis")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleStatusRequestMsg get genesis")
return
}
syncTicker := time.NewTicker(syncCycle)
block, err := bk.chain.GetBlockByHeight(bk.chain.BestBlockHeight())
if err != nil {
- log.WithField("err", err).Error("fail on syncWorker get best block")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on syncWorker get best block")
}
if err := bk.peers.broadcastMinedBlock(block); err != nil {
- log.WithField("err", err).Error("fail on syncWorker broadcast new block")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on syncWorker broadcast new block")
}
if err = bk.peers.broadcastNewStatus(block, genesisBlock); err != nil {
- log.WithField("err", err).Error("fail on syncWorker broadcast new status")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on syncWorker broadcast new status")
}
}
}
)
const (
+ logModule = "netsync"
maxTxChanSize = 10000
maxFilterAddressSize = 50
maxFilterAddressCount = 1000
func (sm *SyncManager) handleBlocksMsg(peer *peer, msg *BlocksMessage) {
blocks, err := msg.GetBlocks()
if err != nil {
- log.WithField("err", err).Debug("fail on handleBlocksMsg GetBlocks")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Debug("fail on handleBlocksMsg GetBlocks")
return
}
block, err = sm.chain.GetBlockByHash(msg.GetHash())
}
if err != nil {
- log.WithField("err", err).Warning("fail on handleGetBlockMsg get block from chain")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on handleGetBlockMsg get block from chain")
return
}
sm.peers.removePeer(peer.ID())
}
if err != nil {
- log.WithField("err", err).Error("fail on handleGetBlockMsg sentBlock")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleGetBlockMsg sentBlock")
}
}
for _, block := range blocks {
rawData, err := block.MarshalText()
if err != nil {
- log.WithField("err", err).Error("fail on handleGetBlocksMsg marshal block")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleGetBlocksMsg marshal block")
continue
}
sm.peers.removePeer(peer.ID())
}
if err != nil {
- log.WithField("err", err).Error("fail on handleGetBlocksMsg sentBlock")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleGetBlocksMsg sentBlock")
}
}
func (sm *SyncManager) handleGetHeadersMsg(peer *peer, msg *GetHeadersMessage) {
headers, err := sm.blockKeeper.locateHeaders(msg.GetBlockLocator(), msg.GetStopHash())
if err != nil || len(headers) == 0 {
- log.WithField("err", err).Debug("fail on handleGetHeadersMsg locateHeaders")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Debug("fail on handleGetHeadersMsg locateHeaders")
return
}
sm.peers.removePeer(peer.ID())
}
if err != nil {
- log.WithField("err", err).Error("fail on handleGetHeadersMsg sentBlock")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleGetHeadersMsg sentBlock")
}
}
block, err = sm.chain.GetBlockByHash(msg.GetHash())
}
if err != nil {
- log.WithField("err", err).Warning("fail on handleGetMerkleBlockMsg get block from chain")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on handleGetMerkleBlockMsg get block from chain")
return
}
blockHash := block.Hash()
txStatus, err := sm.chain.GetTransactionStatus(&blockHash)
if err != nil {
- log.WithField("err", err).Warning("fail on handleGetMerkleBlockMsg get transaction status")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on handleGetMerkleBlockMsg get transaction status")
return
}
ok, err := peer.sendMerkleBlock(block, txStatus)
if err != nil {
- log.WithField("err", err).Error("fail on handleGetMerkleBlockMsg sentMerkleBlock")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleGetMerkleBlockMsg sentMerkleBlock")
return
}
func (sm *SyncManager) handleHeadersMsg(peer *peer, msg *HeadersMessage) {
headers, err := msg.GetHeaders()
if err != nil {
- log.WithField("err", err).Debug("fail on handleHeadersMsg GetHeaders")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Debug("fail on handleHeadersMsg GetHeaders")
return
}
func (sm *SyncManager) handleMineBlockMsg(peer *peer, msg *MineBlockMessage) {
block, err := msg.GetMineBlock()
if err != nil {
- log.WithField("err", err).Warning("fail on handleMineBlockMsg GetMineBlock")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on handleMineBlockMsg GetMineBlock")
return
}
bestHeader := sm.chain.BestBlockHeader()
genesisBlock, err := sm.chain.GetBlockByHeight(0)
if err != nil {
- log.WithField("err", err).Error("fail on handleStatusRequestMsg get genesis")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleStatusRequestMsg get genesis")
}
genesisHash := genesisBlock.Hash()
if genesisHash := msg.GetGenesisHash(); sm.genesisHash != *genesisHash {
log.WithFields(log.Fields{
+ "module": logModule,
"remote genesis": genesisHash.String(),
"local genesis": sm.genesisHash.String(),
}).Warn("fail hand shake due to differnt genesis")
return
}
+ log.WithFields(log.Fields{
+ "module": logModule,
+ "peer": basePeer.Addr(),
+ "type": reflect.TypeOf(msg),
+ "message": msg.String(),
+ }).Info("receive message from peer")
+
switch msg := msg.(type) {
case *GetBlockMessage:
sm.handleGetBlockMsg(peer, msg)
sm.handleGetMerkleBlockMsg(peer, msg)
default:
- log.Errorf("unknown message type %v", reflect.TypeOf(msg))
+ log.WithFields(log.Fields{
+ "module": logModule,
+ "peer": basePeer.Addr(),
+ "message_type": reflect.TypeOf(msg),
+ }).Error("unhandled message type")
}
}
case blockHash := <-sm.newBlockCh:
block, err := sm.chain.GetBlockByHash(blockHash)
if err != nil {
- log.Errorf("Failed on mined broadcast loop get block %v", err)
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on mined broadcast loop get block")
return
}
if err := sm.peers.broadcastMinedBlock(block); err != nil {
- log.Errorf("Broadcast mine block error. %v", err)
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on broadcast mine block")
return
}
case <-sm.quitSync:
import (
"bytes"
+ "encoding/hex"
"encoding/json"
"errors"
"fmt"
)
//BlockchainMessage is a generic message for this reactor.
-type BlockchainMessage interface{}
+type BlockchainMessage interface {
+ String() string
+}
var _ = wire.RegisterInterface(
struct{ BlockchainMessage }{},
return &hash
}
-//String convert msg to string
func (m *GetBlockMessage) String() string {
if m.Height > 0 {
- return fmt.Sprintf("GetBlockMessage{Height: %d}", m.Height)
+ return fmt.Sprintf("{height: %d}", m.Height)
}
- hash := m.GetHash()
- return fmt.Sprintf("GetBlockMessage{Hash: %s}", hash.String())
+ return fmt.Sprintf("{hash: %s}", hex.EncodeToString(m.RawHash[:]))
}
//BlockMessage response get block msg
return block, nil
}
-//String convert msg to string
func (m *BlockMessage) String() string {
- return fmt.Sprintf("BlockMessage{Size: %d}", len(m.RawBlock))
+ return fmt.Sprintf("{block_size: %d}", len(m.RawBlock))
}
//GetHeadersMessage is one of the bytom msg type
}
//GetBlockLocator return the locator of the msg
-func (msg *GetHeadersMessage) GetBlockLocator() []*bc.Hash {
+func (m *GetHeadersMessage) GetBlockLocator() []*bc.Hash {
blockLocator := []*bc.Hash{}
- for _, rawHash := range msg.RawBlockLocator {
+ for _, rawHash := range m.RawBlockLocator {
hash := bc.NewHash(rawHash)
blockLocator = append(blockLocator, &hash)
}
return blockLocator
}
+func (m *GetHeadersMessage) String() string {
+ return fmt.Sprintf("{stop_hash: %s}", hex.EncodeToString(m.RawStopHash[:]))
+}
+
//GetStopHash return the stop hash of the msg
-func (msg *GetHeadersMessage) GetStopHash() *bc.Hash {
- hash := bc.NewHash(msg.RawStopHash)
+func (m *GetHeadersMessage) GetStopHash() *bc.Hash {
+ hash := bc.NewHash(m.RawStopHash)
return &hash
}
}
//GetHeaders return the headers in the msg
-func (msg *HeadersMessage) GetHeaders() ([]*types.BlockHeader, error) {
+func (m *HeadersMessage) GetHeaders() ([]*types.BlockHeader, error) {
headers := []*types.BlockHeader{}
- for _, data := range msg.RawHeaders {
+ for _, data := range m.RawHeaders {
header := &types.BlockHeader{}
if err := json.Unmarshal(data, header); err != nil {
return nil, err
return headers, nil
}
+func (m *HeadersMessage) String() string {
+ return fmt.Sprintf("{header_length: %d}", len(m.RawHeaders))
+}
+
//GetBlocksMessage is one of the bytom msg type
type GetBlocksMessage struct {
RawBlockLocator [][32]byte
}
//GetBlockLocator return the locator of the msg
-func (msg *GetBlocksMessage) GetBlockLocator() []*bc.Hash {
+func (m *GetBlocksMessage) GetBlockLocator() []*bc.Hash {
blockLocator := []*bc.Hash{}
- for _, rawHash := range msg.RawBlockLocator {
+ for _, rawHash := range m.RawBlockLocator {
hash := bc.NewHash(rawHash)
blockLocator = append(blockLocator, &hash)
}
}
//GetStopHash return the stop hash of the msg
-func (msg *GetBlocksMessage) GetStopHash() *bc.Hash {
- hash := bc.NewHash(msg.RawStopHash)
+func (m *GetBlocksMessage) GetStopHash() *bc.Hash {
+ hash := bc.NewHash(m.RawStopHash)
return &hash
}
+func (m *GetBlocksMessage) String() string {
+ return fmt.Sprintf("{stop_hash: %s}", hex.EncodeToString(m.RawStopHash[:]))
+}
+
//BlocksMessage is one of the bytom msg type
type BlocksMessage struct {
RawBlocks [][]byte
}
//GetBlocks returns the blocks in the msg
-func (msg *BlocksMessage) GetBlocks() ([]*types.Block, error) {
+func (m *BlocksMessage) GetBlocks() ([]*types.Block, error) {
blocks := []*types.Block{}
- for _, data := range msg.RawBlocks {
+ for _, data := range m.RawBlocks {
block := &types.Block{}
if err := json.Unmarshal(data, block); err != nil {
return nil, err
return blocks, nil
}
+func (m *BlocksMessage) String() string {
+ return fmt.Sprintf("{blocks_length: %d}", len(m.RawBlocks))
+}
+
//StatusRequestMessage status request msg
type StatusRequestMessage struct{}
-//String
func (m *StatusRequestMessage) String() string {
- return "StatusRequestMessage"
+ return "{}"
}
//StatusResponseMessage get status response msg
return &hash
}
-//String convert msg to string
func (m *StatusResponseMessage) String() string {
- hash := m.GetHash()
- genesisHash := m.GetGenesisHash()
- return fmt.Sprintf("StatusResponseMessage{Height: %d, Best hash: %s, Genesis hash: %s}", m.Height, hash.String(), genesisHash.String())
+ return fmt.Sprintf("{height: %d, hash: %s}", m.Height, hex.EncodeToString(m.RawHash[:]))
}
//TransactionMessage notify new tx msg
return tx, nil
}
-//String
func (m *TransactionMessage) String() string {
- return fmt.Sprintf("TransactionMessage{Size: %d}", len(m.RawTx))
+ return fmt.Sprintf("{tx_size: %d}", len(m.RawTx))
}
//MineBlockMessage new mined block msg
return block, nil
}
-//String convert msg to string
func (m *MineBlockMessage) String() string {
- return fmt.Sprintf("NewMineBlockMessage{Size: %d}", len(m.RawBlock))
+ return fmt.Sprintf("{block_size: %d}", len(m.RawBlock))
}
//FilterLoadMessage tells the receiving peer to filter the transactions according to address.
Addresses [][]byte
}
+func (m *FilterLoadMessage) String() string {
+ return fmt.Sprintf("{addresses_length: %d}", len(m.Addresses))
+}
+
// FilterAddMessage tells the receiving peer to add address to the filter.
type FilterAddMessage struct {
Address []byte
}
+func (m *FilterAddMessage) String() string {
+ return fmt.Sprintf("{address: %s}", hex.EncodeToString(m.Address))
+}
+
//FilterClearMessage tells the receiving peer to remove a previously-set filter.
type FilterClearMessage struct{}
+func (m *FilterClearMessage) String() string {
+ return "{}"
+}
+
//GetMerkleBlockMessage request merkle blocks from remote peers by height/hash
type GetMerkleBlockMessage struct {
Height uint64
return &hash
}
+func (m *GetMerkleBlockMessage) String() string {
+ if m.Height > 0 {
+ return fmt.Sprintf("{height: %d}", m.Height)
+ }
+ return fmt.Sprintf("{hash: %s}", hex.EncodeToString(m.RawHash[:]))
+}
+
//MerkleBlockMessage return the merkle block to client
type MerkleBlockMessage struct {
RawBlockHeader []byte
Flags []byte
}
-func (msg *MerkleBlockMessage) setRawBlockHeader(bh types.BlockHeader) error {
+func (m *MerkleBlockMessage) setRawBlockHeader(bh types.BlockHeader) error {
rawHeader, err := bh.MarshalText()
if err != nil {
return err
}
- msg.RawBlockHeader = rawHeader
+ m.RawBlockHeader = rawHeader
return nil
}
-func (msg *MerkleBlockMessage) setTxInfo(txHashes []*bc.Hash, txFlags []uint8, relatedTxs []*types.Tx) error {
+func (m *MerkleBlockMessage) setTxInfo(txHashes []*bc.Hash, txFlags []uint8, relatedTxs []*types.Tx) error {
for _, txHash := range txHashes {
- msg.TxHashes = append(msg.TxHashes, txHash.Byte32())
+ m.TxHashes = append(m.TxHashes, txHash.Byte32())
}
for _, tx := range relatedTxs {
rawTxData, err := tx.MarshalText()
return err
}
- msg.RawTxDatas = append(msg.RawTxDatas, rawTxData)
+ m.RawTxDatas = append(m.RawTxDatas, rawTxData)
}
- msg.Flags = txFlags
+ m.Flags = txFlags
return nil
}
-func (msg *MerkleBlockMessage) setStatusInfo(statusHashes []*bc.Hash, relatedStatuses []*bc.TxVerifyResult) error {
+func (m *MerkleBlockMessage) setStatusInfo(statusHashes []*bc.Hash, relatedStatuses []*bc.TxVerifyResult) error {
for _, statusHash := range statusHashes {
- msg.StatusHashes = append(msg.StatusHashes, statusHash.Byte32())
+ m.StatusHashes = append(m.StatusHashes, statusHash.Byte32())
}
for _, status := range relatedStatuses {
return err
}
- msg.RawTxStatuses = append(msg.RawTxStatuses, rawStatusData)
+ m.RawTxStatuses = append(m.RawTxStatuses, rawStatusData)
}
return nil
}
+func (m *MerkleBlockMessage) String() string {
+ return "{}"
+}
+
//NewMerkleBlockMessage construct merkle block message
func NewMerkleBlockMessage() *MerkleBlockMessage {
return &MerkleBlockMessage{}
func (p *peer) addBanScore(persistent, transient uint64, reason string) bool {
score := p.banScore.Increase(persistent, transient)
if score > defaultBanThreshold {
- log.WithFields(log.Fields{"address": p.Addr(), "score": score, "reason": reason}).Errorf("banning and disconnecting")
+ log.WithFields(log.Fields{
+ "module": logModule,
+ "address": p.Addr(),
+ "score": score,
+ "reason": reason,
+ }).Errorf("banning and disconnecting")
return true
}
warnThreshold := defaultBanThreshold >> 1
if score > warnThreshold {
- log.WithFields(log.Fields{"address": p.Addr(), "score": score, "reason": reason}).Warning("ban score increasing")
+ log.WithFields(log.Fields{
+ "module": logModule,
+ "address": p.Addr(),
+ "score": score,
+ "reason": reason,
+ }).Warning("ban score increasing")
}
return false
}
defer p.mtx.Unlock()
if p.filterAdds.Size() >= maxFilterAddressCount {
- log.Warn("the count of filter addresses is greater than limit")
+ log.WithField("module", logModule).Warn("the count of filter addresses is greater than limit")
return
}
if len(address) > maxFilterAddressSize {
- log.Warn("the size of filter address is greater than limit")
+ log.WithField("module", logModule).Warn("the size of filter address is greater than limit")
return
}
p.filterAdds.Add(hex.EncodeToString(address))
return
}
if err := ps.AddBannedPeer(peer.Addr().String()); err != nil {
- log.WithField("err", err).Error("fail on add ban peer")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on add ban peer")
}
ps.removePeer(peerID)
}
ps.peers[peer.ID()] = newPeer(height, hash, peer)
return
}
- log.WithField("ID", peer.ID()).Warning("add existing peer to blockKeeper")
+ log.WithField("module", logModule).Warning("add existing peer to blockKeeper")
}
func (ps *peerSet) bestPeer(flag consensus.ServiceFlag) *peer {
func (pr *ProtocolReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
msgType, msg, err := DecodeMessage(msgBytes)
if err != nil {
- log.WithField("err", err).Errorf("fail on reactor decoding message")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on reactor decoding message")
return
}
select {
case newTx := <-sm.newTxCh:
if err := sm.peers.broadcastTx(newTx); err != nil {
- log.Errorf("Broadcast new tx error. %v", err)
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on broadcast new tx.")
return
}
case <-sm.quitSync:
// Send the pack in the background.
log.WithFields(log.Fields{
- "count": len(sendTxs),
- "bytes": totalSize,
- "peer": msg.peerID,
+ "module": logModule,
+ "count": len(sendTxs),
+ "bytes": totalSize,
+ "peer": msg.peerID,
}).Debug("txSyncLoop sending transactions")
sending = true
go func() {
case err := <-done:
sending = false
if err != nil {
- log.WithField("err", err).Warning("fail on txSyncLoop sending")
+ log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("fail on txSyncLoop sending")
}
if s := pick(); s != nil {
func() {
var n int
wire.ReadBinary(peerNodeInfo, pc.conn, maxNodeInfoSize, &n, &err2)
- log.WithField("peerNodeInfo", peerNodeInfo).Info("Peer handshake")
+ log.WithField("address", peerNodeInfo.ListenAddr).Info("Peer handshake")
})
if err1 != nil {
return peerNodeInfo, errors.Wrap(err1, "Error during handshake/write")
c.index.SetMainChain(node)
c.bestNode = node
- log.WithFields(log.Fields{"height": c.bestNode.Height, "hash": c.bestNode.Hash}).Debug("chain best status has been update")
+ log.WithFields(log.Fields{"height": c.bestNode.Height, "hash": c.bestNode.Hash.String()}).Debug("chain best status has been update")
c.cond.Broadcast()
return nil
}
OSDN Git Service
