From 17f65cd1e5e0fea6e4f7b96c60767aaa0ada366d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 25 Aug 2015 13:57:49 +0300 Subject: eth: update metrics collection to handle eth/62 algos --- eth/downloader/downloader.go | 44 +++++++++++++++++++++++++++++++++++++++---- eth/downloader/metrics.go | 45 ++++++++++++++++++++++++++++++++++++++++++++ eth/downloader/queue.go | 12 ++++++++++++ 3 files changed, 97 insertions(+), 4 deletions(-) create mode 100644 eth/downloader/metrics.go (limited to 'eth/downloader') diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 0e8529756..574f2ba15 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -526,6 +526,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error { glog.V(logger.Debug).Infof("%v: downloading hashes from #%d", p, from) // Create a timeout timer, and the associated hash fetcher + request := time.Now() // time of the last fetch request timeout := time.NewTimer(0) // timer to dump a non-responsive active peer <-timeout.C // timeout channel should be initially empty defer timeout.Stop() @@ -534,6 +535,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error { glog.V(logger.Detail).Infof("%v: fetching %d hashes from #%d", p, MaxHashFetch, from) go p.getAbsHashes(from, MaxHashFetch) + request = time.Now() timeout.Reset(hashTTL) } // Start pulling hashes, until all are exhausted @@ -557,6 +559,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error { glog.V(logger.Debug).Infof("Received hashes from incorrect peer(%s)", hashPack.peerId) break } + hashReqTimer.UpdateSince(request) timeout.Stop() // If no more hashes are inbound, notify the block fetcher and return @@ -609,6 +612,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error { case <-timeout.C: glog.V(logger.Debug).Infof("%v: hash request timed out", p) + hashTimeoutMeter.Mark(1) return errTimeout } } @@ -896,6 +900,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error { defer glog.V(logger.Debug).Infof("%v: header download terminated", p) // Create a timeout timer, and the associated hash fetcher + request := time.Now() // time of the last fetch request timeout := time.NewTimer(0) // timer to dump a non-responsive active peer <-timeout.C // timeout channel should be initially empty defer timeout.Stop() @@ -904,6 +909,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error { glog.V(logger.Detail).Infof("%v: fetching %d headers from #%d", p, MaxHeaderFetch, from) go p.getAbsHeaders(from, MaxHeaderFetch, 0, false) + request = time.Now() timeout.Reset(headerTTL) } // Start pulling headers, until all are exhausted @@ -927,6 +933,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error { glog.V(logger.Debug).Infof("Received headers from incorrect peer (%s)", headerPack.peerId) break } + headerReqTimer.UpdateSince(request) timeout.Stop() // If no more headers are inbound, notify the body fetcher and return @@ -980,6 +987,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error { case <-timeout.C: // Header retrieval timed out, consider the peer bad and drop glog.V(logger.Debug).Infof("%v: header request timed out", p) + headerTimeoutMeter.Mark(1) d.dropPeer(p.id) // Finish the sync gracefully instead of dumping the gathered data though @@ -1244,7 +1252,14 @@ func (d *Downloader) process() { // DeliverHashes61 injects a new batch of hashes received from a remote node into // the download schedule. This is usually invoked through the BlockHashesMsg by // the protocol handler. -func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) error { +func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) (err error) { + // Update the delivery metrics for both good and failed deliveries + hashInMeter.Mark(int64(len(hashes))) + defer func() { + if err != nil { + hashDropMeter.Mark(int64(len(hashes))) + } + }() // Make sure the downloader is active if atomic.LoadInt32(&d.synchronising) == 0 { return errNoSyncActive @@ -1265,7 +1280,14 @@ func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) error { // DeliverBlocks61 injects a new batch of blocks received from a remote node. // This is usually invoked through the BlocksMsg by the protocol handler. -func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) error { +func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) (err error) { + // Update the delivery metrics for both good and failed deliveries + blockInMeter.Mark(int64(len(blocks))) + defer func() { + if err != nil { + blockDropMeter.Mark(int64(len(blocks))) + } + }() // Make sure the downloader is active if atomic.LoadInt32(&d.synchronising) == 0 { return errNoSyncActive @@ -1286,7 +1308,14 @@ func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) error { // DeliverHeaders injects a new batch of blck headers received from a remote // node into the download schedule. -func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error { +func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) (err error) { + // Update the delivery metrics for both good and failed deliveries + headerInMeter.Mark(int64(len(headers))) + defer func() { + if err != nil { + headerDropMeter.Mark(int64(len(headers))) + } + }() // Make sure the downloader is active if atomic.LoadInt32(&d.synchronising) == 0 { return errNoSyncActive @@ -1306,7 +1335,14 @@ func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error { } // DeliverBodies injects a new batch of block bodies received from a remote node. -func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) error { +func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) (err error) { + // Update the delivery metrics for both good and failed deliveries + bodyInMeter.Mark(int64(len(transactions))) + defer func() { + if err != nil { + bodyDropMeter.Mark(int64(len(transactions))) + } + }() // Make sure the downloader is active if atomic.LoadInt32(&d.synchronising) == 0 { return errNoSyncActive diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go new file mode 100644 index 000000000..fd926affd --- /dev/null +++ b/eth/downloader/metrics.go @@ -0,0 +1,45 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Contains the metrics collected by the downloader. + +package downloader + +import ( + "github.com/ethereum/go-ethereum/metrics" +) + +var ( + hashInMeter = metrics.NewMeter("eth/downloader/hashes/in") + hashReqTimer = metrics.NewTimer("eth/downloader/hashes/req") + hashDropMeter = metrics.NewMeter("eth/downloader/hashes/drop") + hashTimeoutMeter = metrics.NewMeter("eth/downloader/hashes/timeout") + + blockInMeter = metrics.NewMeter("eth/downloader/blocks/in") + blockReqTimer = metrics.NewTimer("eth/downloader/blocks/req") + blockDropMeter = metrics.NewMeter("eth/downloader/blocks/drop") + blockTimeoutMeter = metrics.NewMeter("eth/downloader/blocks/timeout") + + headerInMeter = metrics.NewMeter("eth/downloader/headers/in") + headerReqTimer = metrics.NewTimer("eth/downloader/headers/req") + headerDropMeter = metrics.NewMeter("eth/downloader/headers/drop") + headerTimeoutMeter = metrics.NewMeter("eth/downloader/headers/timeout") + + bodyInMeter = metrics.NewMeter("eth/downloader/bodies/in") + bodyReqTimer = metrics.NewTimer("eth/downloader/bodies/req") + bodyDropMeter = metrics.NewMeter("eth/downloader/bodies/drop") + bodyTimeoutMeter = metrics.NewMeter("eth/downloader/bodies/timeout") +) diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index a527414ff..7db78327b 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -397,9 +397,19 @@ func (q *queue) Expire(timeout time.Duration) []string { peers := []string{} for id, request := range q.pendPool { if time.Since(request.Time) > timeout { + // Update the metrics with the timeout + if len(request.Hashes) > 0 { + blockTimeoutMeter.Mark(1) + } else { + bodyTimeoutMeter.Mark(1) + } + // Return any non satisfied requests to the pool for hash, index := range request.Hashes { q.hashQueue.Push(hash, float32(index)) } + for _, header := range request.Headers { + q.headerQueue.Push(header, -float32(header.Number.Uint64())) + } peers = append(peers, id) } } @@ -420,6 +430,7 @@ func (q *queue) Deliver61(id string, blocks []*types.Block) (err error) { if request == nil { return errNoFetchesPending } + blockReqTimer.UpdateSince(request.Time) delete(q.pendPool, id) // If no blocks were retrieved, mark them as unavailable for the origin peer @@ -468,6 +479,7 @@ func (q *queue) Deliver(id string, txLists [][]*types.Transaction, uncleLists [] if request == nil { return errNoFetchesPending } + bodyReqTimer.UpdateSince(request.Time) delete(q.pendPool, id) // If no block bodies were retrieved, mark them as unavailable for the origin peer -- cgit v1.2.3