metrics, tests: add metrics for load testing

This commit is contained in:
Miya Chen 2017-09-28 11:16:21 +08:00
parent 0fddd0de2f
commit 9eb0fdfa8a
8 changed files with 668 additions and 1 deletions

View File

@ -29,7 +29,7 @@ import (
var (
DefaultGasPrice int64 = 20000000000
DefaultGasLimit int64 = 22000 // the gas of ether tx should be 21000
DefaultGasLimit int64 = 21000 // the gas of ether tx should be 21000
)
func SendEther(client client.Client, from *ecdsa.PrivateKey, to common.Address, amount *big.Int, nonce uint64) error {

338
metrics/blockchain.go Normal file
View File

@ -0,0 +1,338 @@
// Copyright 2017 AMIS Technologies
// 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 <http://www.gnu.org/licenses/>.
package metrics
import (
"context"
"math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
ethtypes "github.com/ethereum/go-ethereum/core/types"
"github.com/rcrowley/go-metrics"
"github.com/getamis/istanbul-tools/client"
"github.com/getamis/istanbul-tools/container"
)
type metricsManager struct {
registry *DefaultRegistry
SentTxCounter metrics.Counter
TxErrCounter metrics.Counter
ReqMeter metrics.Meter
ExcutedTxCounter metrics.Counter
RespMeter metrics.Meter
UnknownTxCounter metrics.Counter
TxLatencyTimer metrics.Timer
BlockPeriodTimer metrics.Timer
BlockLatencyTimer metrics.Timer
TPSBlockHistogram metrics.Histogram
}
func newMetricsManager() *metricsManager {
r := NewRegistry()
return &metricsManager{
registry: r,
SentTxCounter: r.NewCounter("tx/sent"),
TxErrCounter: r.NewCounter("tx/error"),
ExcutedTxCounter: r.NewCounter("tx/excuted"),
UnknownTxCounter: r.NewCounter("tx/unknown"),
ReqMeter: r.NewMeter("tx/rps"),
RespMeter: r.NewMeter("tx/tps/response"),
TPSBlockHistogram: r.NewHistogram("tx/tps/block"),
TxLatencyTimer: r.NewTimer("tx/latency"),
BlockPeriodTimer: r.NewTimer("block/period"),
BlockLatencyTimer: r.NewTimer("block/latency"),
}
}
func (m *metricsManager) Export() {
m.registry.Export()
}
type metricChain struct {
container.Blockchain
eths []container.Ethereum
headCh chan *ethtypes.Header
headSubs []ethereum.Subscription
txStartCh chan *txInfo
txDoneCh chan *txInfo
metricsMgr *metricsManager
wg sync.WaitGroup
quit chan struct{}
}
func NewMetricChain(blockchain container.Blockchain) container.Blockchain {
if blockchain == nil {
return nil
}
mc := &metricChain{
Blockchain: blockchain,
headCh: make(chan *ethtypes.Header),
txStartCh: make(chan *txInfo, 10000),
txDoneCh: make(chan *txInfo, 10000),
quit: make(chan struct{}),
metricsMgr: newMetricsManager(),
}
mc.eths = mc.getMetricEthereum(mc.Blockchain.Validators())
return mc
}
func (mc *metricChain) AddValidators(numOfValidators int) ([]container.Ethereum, error) {
vals, err := mc.Blockchain.AddValidators(numOfValidators)
if err != nil {
return nil, err
}
mc.eths = mc.getMetricEthereum(vals)
return mc.eths, nil
}
func (mc *metricChain) RemoveValidators(candidates []container.Ethereum, t time.Duration) error {
err := mc.Blockchain.RemoveValidators(candidates, t)
mc.eths = mc.getMetricEthereum(mc.Blockchain.Validators())
return err
}
func (mc *metricChain) Start(strong bool) error {
err := mc.Blockchain.Start(strong)
if err != nil {
return err
}
for _, eth := range mc.eths {
cli := eth.NewClient()
sub, err := cli.SubscribeNewHead(context.Background(), mc.headCh)
if err != nil {
log.Error("Failed to subscribe new head", "err", err)
return err
}
mc.headSubs = append(mc.headSubs, sub)
}
mc.wg.Add(2)
go mc.handleNewHeadEvent()
go mc.updateTxInfo()
return nil
}
func (mc *metricChain) Stop(strong bool) error {
close(mc.quit)
for _, sub := range mc.headSubs {
sub.Unsubscribe()
}
mc.wg.Wait()
mc.metricsMgr.Export()
return mc.Blockchain.Stop(strong)
}
func (mc *metricChain) Validators() []container.Ethereum {
return mc.eths
}
func (mc *metricChain) getMetricEthereum(eths []container.Ethereum) []container.Ethereum {
meths := make([]container.Ethereum, len(eths))
for i, eth := range eths {
meths[i] = &metricEthereum{
Ethereum: eth,
txStartCh: mc.txStartCh,
metricsMgr: mc.metricsMgr,
}
}
return meths
}
func (mc *metricChain) handleNewHeadEvent() {
defer mc.wg.Done()
mutex := sync.Mutex{}
var preBlockTime = time.Now()
handledHeads := map[string]*ethtypes.Header{}
for {
select {
case header := <-mc.headCh:
now := time.Now()
go func(header *ethtypes.Header, now time.Time) {
log.Info("New head", "number", header.Number.Int64(), "hash", header.Hash().TerminalString(), "time", header.Time)
hash := header.Hash().String()
// lock hash first
var wasHandled bool
var preBlock *ethtypes.Header
mutex.Lock()
_, wasHandled = handledHeads[hash]
if !wasHandled {
handledHeads[hash] = header
}
preBlock, _ = handledHeads[header.ParentHash.String()]
mutex.Unlock()
if wasHandled {
return
}
var blockPeriod int64
if header.Number.Int64() > 2 && preBlock != nil {
blockPeriod = new(big.Int).Sub(header.Time, preBlock.Time).Int64()
mc.metricsMgr.BlockPeriodTimer.Update(time.Duration(blockPeriod) * time.Second)
}
mc.metricsMgr.BlockLatencyTimer.Update(now.Sub(preBlockTime))
preBlockTime = now
// check counts of txs
if header.TxHash == ethtypes.DeriveSha(ethtypes.Transactions{}) && header.UncleHash == types.CalcUncleHash([]*ethtypes.Header{}) {
mc.metricsMgr.ExcutedTxCounter.Inc(0)
mc.metricsMgr.RespMeter.Mark(0)
mc.metricsMgr.TPSBlockHistogram.Update(0)
return
}
// get block
blockCh := make(chan *ethtypes.Block, len(mc.eths))
ctx, cancel := context.WithCancel(context.Background())
for _, eth := range mc.eths {
cli := eth.NewClient()
go getBlock(ctx, cli, header.Hash(), blockCh)
}
// wait for right block
var headBlock *ethtypes.Block
for i := 0; i < len(mc.eths); i++ {
headBlock = <-blockCh
if headBlock != nil {
break
}
}
// cancel other requests
cancel()
mc.metricsMgr.ExcutedTxCounter.Inc(int64(len(headBlock.Transactions())))
mc.metricsMgr.RespMeter.Mark(int64(len(headBlock.Transactions())))
if blockPeriod > 0 {
mc.metricsMgr.TPSBlockHistogram.Update(int64(len(headBlock.Transactions())) / blockPeriod)
}
// update tx info
for _, tx := range headBlock.Transactions() {
go func() {
mc.txDoneCh <- &txInfo{
Hash: tx.Hash().String(),
Time: now,
}
}()
}
}(header, now)
case <-mc.quit:
return
}
}
}
func (mc *metricChain) updateTxInfo() {
defer mc.wg.Done()
// TODO: the completed tx should be deleted from map
// given large space is workaround beacause the some problem between deleting and updating map
txStartMap := make(map[string]time.Time, 0)
txDoneMap := make(map[string]time.Time, 0)
defer func() {
// TODO: debug metric to check incomplete tx
// for _ = range txStartMap {
// mc.metricsMgr.UnknownTxCounter.Inc(1)
// }
// for _ = range txDoneMap {
// mc.metricsMgr.UnknownTxCounter.Inc(1)
// }
}()
updateTxStart := func(hash string, startTime time.Time) {
if done, ok := txDoneMap[hash]; ok {
mc.metricsMgr.TxLatencyTimer.Update(done.Sub(startTime))
return
//delete(txDoneMap, hash)
}
txStartMap[hash] = startTime
}
updateTxDone := func(hash string, doneTime time.Time) {
if start, ok := txStartMap[hash]; ok {
mc.metricsMgr.TxLatencyTimer.Update(doneTime.Sub(start))
return
//delete(txStartMap, hash)
}
txDoneMap[hash] = doneTime
}
for {
select {
case txStart := <-mc.txStartCh:
updateTxStart(txStart.Hash, txStart.Time)
case txDone := <-mc.txDoneCh:
updateTxDone(txDone.Hash, txDone.Time)
case <-mc.quit:
// clear tx start
TX_START:
for {
select {
case txStart := <-mc.txStartCh:
updateTxStart(txStart.Hash, txStart.Time)
default:
break TX_START
}
}
// clear tx done
TX_DONE:
for {
select {
case txDone := <-mc.txDoneCh:
updateTxDone(txDone.Hash, txDone.Time)
default:
break TX_DONE
}
}
return
}
}
}
func getBlock(ctx context.Context, cli client.Client, hash common.Hash, blockCh chan<- *ethtypes.Block) {
resp := make(chan *ethtypes.Block)
go func() {
block, err := cli.BlockByHash(ctx, hash)
if err != nil {
resp <- nil
}
resp <- block
}()
select {
case <-ctx.Done():
// Wait for client.BlockByHash
<-resp
// someone cancelled the request
blockCh <- nil
case r := <-resp:
blockCh <- r
}
}

110
metrics/client.go Normal file
View File

@ -0,0 +1,110 @@
// Copyright 2017 AMIS Technologies
// 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 <http://www.gnu.org/licenses/>.
package metrics
import (
"context"
"math/big"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/getamis/istanbul-tools/client"
)
type metricClient struct {
client.Client
txStartCh chan *txInfo
metricsMgr *metricsManager
}
func (c *metricClient) SendTransaction(ctx context.Context, from, to common.Address, value *big.Int) (hash string, err error) {
defer func() {
sendTime := time.Now()
if err != nil {
c.metricsMgr.TxErrCounter.Inc(1)
} else {
c.metricsMgr.SentTxCounter.Inc(1)
c.metricsMgr.ReqMeter.Mark(1)
go func() {
c.txStartCh <- &txInfo{
Hash: hash,
Time: sendTime,
}
}()
}
}()
return c.Client.SendTransaction(ctx, from, to, value)
}
func (c *metricClient) CreateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int) (hash string, err error) {
defer func() {
sendTime := time.Now()
if err != nil {
c.metricsMgr.TxErrCounter.Inc(1)
} else {
c.metricsMgr.SentTxCounter.Inc(1)
c.metricsMgr.ReqMeter.Mark(1)
go func() {
c.txStartCh <- &txInfo{
Hash: hash,
Time: sendTime,
}
}()
}
}()
return c.Client.CreateContract(ctx, from, bytecode, gas)
}
func (c *metricClient) CreatePrivateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int, privateFor []string) (hash string, err error) {
defer func() {
sendTime := time.Now()
if err != nil {
c.metricsMgr.TxErrCounter.Inc(1)
} else {
c.metricsMgr.SentTxCounter.Inc(1)
c.metricsMgr.ReqMeter.Mark(1)
go func() {
c.txStartCh <- &txInfo{
Hash: hash,
Time: sendTime,
}
}()
}
}()
return c.Client.CreatePrivateContract(ctx, from, bytecode, gas, privateFor)
}
func (c *metricClient) SendRawTransaction(ctx context.Context, tx *types.Transaction) (err error) {
defer func() {
sendTime := time.Now()
if err != nil {
c.metricsMgr.TxErrCounter.Inc(1)
} else {
c.metricsMgr.SentTxCounter.Inc(1)
c.metricsMgr.ReqMeter.Mark(1)
go func() {
c.txStartCh <- &txInfo{
Hash: tx.Hash().String(),
Time: sendTime,
}
}()
}
}()
return c.Client.SendRawTransaction(ctx, tx)
}

51
metrics/ethereum.go Normal file
View File

@ -0,0 +1,51 @@
// Copyright 2017 AMIS Technologies
// 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 <http://www.gnu.org/licenses/>.
package metrics
import (
"errors"
"fmt"
"math/big"
"time"
"github.com/getamis/istanbul-tools/client"
"github.com/getamis/istanbul-tools/container"
"github.com/getamis/istanbul-tools/k8s"
)
type metricEthereum struct {
container.Ethereum
txStartCh chan *txInfo
metricsMgr *metricsManager
}
func (e *metricEthereum) NewClient() client.Client {
return &metricClient{
Client: e.Ethereum.NewClient(),
txStartCh: e.txStartCh,
metricsMgr: e.metricsMgr,
}
}
func (eth *metricEthereum) SendTransactions(client client.Client, amount *big.Int, duration time.Duration) error {
transactor, ok := eth.Ethereum.(k8s.Transactor)
if !ok {
return errors.New("Not support Transactor interface.")
}
fmt.Println("Begin to SendTransactions.")
return transactor.SendTransactions(client, amount, duration)
}

23
metrics/log.go Normal file
View File

@ -0,0 +1,23 @@
// Copyright 2017 AMIS Technologies
// 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 <http://www.gnu.org/licenses/>.
package metrics
import (
logging "github.com/getamis/istanbul-tools/log"
)
var log = logging.New()

113
metrics/metrics.go Normal file
View File

@ -0,0 +1,113 @@
// Copyright 2017 AMIS Technologies
// 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 <http://www.gnu.org/licenses/>.
package metrics
import (
"fmt"
"github.com/rcrowley/go-metrics"
)
type DefaultRegistry struct {
registry metrics.Registry
}
func NewRegistry() *DefaultRegistry {
r := metrics.NewRegistry()
return &DefaultRegistry{registry: r}
}
func (r *DefaultRegistry) NewCounter(name string) metrics.Counter {
return metrics.GetOrRegisterCounter(name, r.registry)
}
func (r *DefaultRegistry) NewMeter(name string) metrics.Meter {
return metrics.GetOrRegisterMeter(name, r.registry)
}
func (r *DefaultRegistry) NewTimer(name string) metrics.Timer {
return metrics.GetOrRegisterTimer(name, r.registry)
}
func (r *DefaultRegistry) NewHistogram(name string) metrics.Histogram {
return metrics.GetOrRegisterHistogram(name, r.registry, metrics.NewExpDecaySample(1028, 0.015))
}
// -----------------------------------------------------------------
func (r *DefaultRegistry) Export() {
r.export()
}
func (r *DefaultRegistry) export() {
r.registry.Each(func(name string, i interface{}) {
switch metric := i.(type) {
case metrics.Counter:
fmt.Printf("counter %s\n", name)
fmt.Printf(" count: %9d\n", metric.Count())
case metrics.Gauge:
fmt.Printf("gauge %s\n", name)
fmt.Printf(" value: %9d\n", metric.Value())
case metrics.GaugeFloat64:
fmt.Printf("gauge %s\n", name)
fmt.Printf(" value: %f\n", metric.Value())
case metrics.Healthcheck:
metric.Check()
fmt.Printf("healthcheck %s\n", name)
fmt.Printf(" error: %v\n", metric.Error())
case metrics.Histogram:
h := metric.Snapshot()
ps := h.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
fmt.Printf("histogram %s\n", name)
fmt.Printf(" count: %9d\n", h.Count())
fmt.Printf(" min: %9d\n", h.Min())
fmt.Printf(" max: %9d\n", h.Max())
fmt.Printf(" mean: %e\n", h.Mean())
fmt.Printf(" stddev: %e\n", h.StdDev())
fmt.Printf(" median: %e\n", ps[0])
fmt.Printf(" 75%%: %e\n", ps[1])
fmt.Printf(" 95%%: %e\n", ps[2])
fmt.Printf(" 99%%: %e\n", ps[3])
fmt.Printf(" 99.9%%: %e\n", ps[4])
case metrics.Meter:
m := metric.Snapshot()
fmt.Printf("meter %s\n", name)
fmt.Printf(" count: %9d\n", m.Count())
fmt.Printf(" 1-min rate: %e\n", m.Rate1())
fmt.Printf(" 5-min rate: %e\n", m.Rate5())
fmt.Printf(" 15-min rate: %e\n", m.Rate15())
fmt.Printf(" mean rate: %e\n", m.RateMean())
case metrics.Timer:
t := metric.Snapshot()
ps := t.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
fmt.Printf("timer %s\n", name)
fmt.Printf(" count: %9d\n", t.Count())
fmt.Printf(" min: %e\n", float64(t.Min()))
fmt.Printf(" max: %e\n", float64(t.Max()))
fmt.Printf(" mean: %e\n", t.Mean())
fmt.Printf(" stddev: %e\n", t.StdDev())
fmt.Printf(" median: %e\n", ps[0])
fmt.Printf(" 75%%: %e\n", ps[1])
fmt.Printf(" 95%%: %e\n", ps[2])
fmt.Printf(" 99%%: %e\n", ps[3])
fmt.Printf(" 99.9%%: %e\n", ps[4])
fmt.Printf(" 1-min rate: %e\n", t.Rate1())
fmt.Printf(" 5-min rate: %e\n", t.Rate5())
fmt.Printf(" 15-min rate: %e\n", t.Rate15())
fmt.Printf(" mean rate: %e\n", t.RateMean())
}
})
}

26
metrics/utils.go Normal file
View File

@ -0,0 +1,26 @@
// Copyright 2017 AMIS Technologies
// 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 <http://www.gnu.org/licenses/>.
package metrics
import (
"time"
)
type txInfo struct {
Hash string
Time time.Time
}

View File

@ -28,6 +28,7 @@ import (
"github.com/getamis/istanbul-tools/container"
"github.com/getamis/istanbul-tools/k8s"
"github.com/getamis/istanbul-tools/metrics"
"github.com/getamis/istanbul-tools/tests"
)
@ -74,6 +75,7 @@ func runTests(numberOfValidators int, gaslimit int, txpoolSize int) {
k8s.Mine(),
k8s.TxPoolSize(txpoolSize),
)
blockchain = metrics.NewMetricChain(blockchain)
Expect(blockchain).NotTo(BeNil())
Expect(blockchain.Start(true)).To(BeNil())
@ -86,6 +88,10 @@ func runTests(numberOfValidators int, gaslimit int, txpoolSize int) {
AfterEach(func() {
Expect(blockchain).NotTo(BeNil())
tests.WaitFor(blockchain.Validators(), func(geth container.Ethereum, wg *sync.WaitGroup) {
Expect(geth.WaitForBlocks(10)).To(BeNil())
wg.Done()
})
Expect(blockchain.Stop(true)).To(BeNil())
blockchain.Finalize()
})