poanetwork
/
istanbul-tools
mirror of https://github.com/poanetwork/istanbul-tools.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #86 from getamis/test/measure-tx-latency 

Add metrics blockchain
This commit is contained in:
Miya Chen 2017-10-19 04:19:54 -05:00 committed by GitHub
parent 952d49100a 67794ca1cf
commit 92c71b747b
21 changed files with 1092 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
charts/utils.go
View File

@ -50,9 +50,9 @@ func installRelease(name string, args []string, path string, debug bool) error {
cmd.Args = append(cmd.Args, "--dry-run")
cmd.Args = append(cmd.Args, "--debug")
} else {
cmd.Args = append(cmd.Args, "--wait")
cmd.Args = append(cmd.Args, "--timeout")
cmd.Args = append(cmd.Args, "600")
// cmd.Args = append(cmd.Args, "--wait")
// cmd.Args = append(cmd.Args, "--timeout")
// cmd.Args = append(cmd.Args, "120")
}
if debug {

71
client/api.go Normal file
View File

@ -0,0 +1,71 @@
// 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 client
import (
"context"
"math/big"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/p2p"
)
type Client interface {
Close()
AddPeer(ctx context.Context, nodeURL string) error
AdminPeers(ctx context.Context) ([]*p2p.PeerInfo, error)
NodeInfo(ctx context.Context) (*p2p.PeerInfo, error)
BlockNumber(ctx context.Context) (*big.Int, error)
StartMining(ctx context.Context) error
StopMining(ctx context.Context) error
SendTransaction(ctx context.Context, from, to common.Address, value *big.Int) (string, error)
CreateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int) (string, error)
CreatePrivateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int, privateFor []string) (string, error)
ProposeValidator(ctx context.Context, address common.Address, auth bool) error
GetValidators(ctx context.Context, blockNumbers *big.Int) ([]common.Address, error)
// eth client
BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error)
HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error)
HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
TransactionByHash(ctx context.Context, hash common.Hash) (*types.Transaction, bool, error)
TransactionCount(ctx context.Context, blockHash common.Hash) (uint, error)
TransactionInBlock(ctx context.Context, blockHash common.Hash, index uint) (*types.Transaction, error)
TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error)
SyncProgress(ctx context.Context) (*ethereum.SyncProgress, error)
SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error)
NetworkID(ctx context.Context) (*big.Int, error)
BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error)
StorageAt(ctx context.Context, account common.Address, key common.Hash, blockNumber *big.Int) ([]byte, error)
CodeAt(ctx context.Context, account common.Address, blockNumber *big.Int) ([]byte, error)
NonceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (uint64, error)
FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error)
PendingBalanceAt(ctx context.Context, account common.Address) (*big.Int, error)
PendingStorageAt(ctx context.Context, account common.Address, key common.Hash) ([]byte, error)
PendingCodeAt(ctx context.Context, account common.Address) ([]byte, error)
PendingNonceAt(ctx context.Context, account common.Address) (uint64, error)
PendingTransactionCount(ctx context.Context) (uint, error)
CallContract(ctx context.Context, msg ethereum.CallMsg, blockNumber *big.Int) ([]byte, error)
PendingCallContract(ctx context.Context, msg ethereum.CallMsg) ([]byte, error)
SuggestGasPrice(ctx context.Context) (*big.Int, error)
EstimateGas(ctx context.Context, msg ethereum.CallMsg) (*big.Int, error)
SendRawTransaction(ctx context.Context, tx *types.Transaction) error
}

30
client/client.go
View File

@ -30,29 +30,29 @@ import (
"github.com/getamis/go-ethereum/ethclient"
)
type Client struct {
type client struct {
c *rpc.Client
ethClient *ethclient.Client
}
func Dial(rawurl string) (*Client, error) {
func Dial(rawurl string) (Client, error) {
c, err := rpc.Dial(rawurl)
if err != nil {
return nil, err
}
return &Client{
return &client{
c: c,
ethClient: ethclient.NewClient(c),
}, nil
}
func (c *Client) Close() {
func (c *client) Close() {
c.c.Close()
}
// ----------------------------------------------------------------------------
func (ic *Client) AddPeer(ctx context.Context, nodeURL string) error {
func (ic *client) AddPeer(ctx context.Context, nodeURL string) error {
var r bool
// TODO: Result needs to be verified
// The response data type are bytes, but we cannot parse...
@ -63,7 +63,7 @@ func (ic *Client) AddPeer(ctx context.Context, nodeURL string) error {
return err
}
func (ic *Client) AdminPeers(ctx context.Context) ([]*p2p.PeerInfo, error) {
func (ic *client) AdminPeers(ctx context.Context) ([]*p2p.PeerInfo, error) {
var r []*p2p.PeerInfo
// The response data type are bytes, but we cannot parse...
err := ic.c.CallContext(ctx, &r, "admin_peers")
@ -73,7 +73,7 @@ func (ic *Client) AdminPeers(ctx context.Context) ([]*p2p.PeerInfo, error) {
return r, err
}
func (ic *Client) NodeInfo(ctx context.Context) (*p2p.PeerInfo, error) {
func (ic *client) NodeInfo(ctx context.Context) (*p2p.PeerInfo, error) {
var r *p2p.PeerInfo
err := ic.c.CallContext(ctx, &r, "admin_nodeInfo")
if err != nil {
@ -83,7 +83,7 @@ func (ic *Client) NodeInfo(ctx context.Context) (*p2p.PeerInfo, error) {
}
// ----------------------------------------------------------------------------
func (ic *Client) BlockNumber(ctx context.Context) (*big.Int, error) {
func (ic *client) BlockNumber(ctx context.Context) (*big.Int, error) {
var r string
err := ic.c.CallContext(ctx, &r, "eth_blockNumber")
if err != nil {
@ -95,7 +95,7 @@ func (ic *Client) BlockNumber(ctx context.Context) (*big.Int, error) {
// ----------------------------------------------------------------------------
func (ic *Client) StartMining(ctx context.Context) error {
func (ic *client) StartMining(ctx context.Context) error {
var r []byte
// TODO: Result needs to be verified
// The response data type are bytes, but we cannot parse...
@ -106,7 +106,7 @@ func (ic *Client) StartMining(ctx context.Context) error {
return err
}
func (ic *Client) StopMining(ctx context.Context) error {
func (ic *client) StopMining(ctx context.Context) error {
err := ic.c.CallContext(ctx, nil, "miner_stop", nil)
if err != nil {
return err
@ -116,7 +116,7 @@ func (ic *Client) StopMining(ctx context.Context) error {
// ----------------------------------------------------------------------------
func (ic *Client) SendTransaction(ctx context.Context, from, to common.Address, value *big.Int) (txHash string, err error) {
func (ic *client) SendTransaction(ctx context.Context, from, to common.Address, value *big.Int) (txHash string, err error) {
var hex hexutil.Bytes
arg := map[string]interface{}{
"from": from,
@ -129,7 +129,7 @@ func (ic *Client) SendTransaction(ctx context.Context, from, to common.Address,
return
}
func (ic *Client) CreateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int) (txHash string, err error) {
func (ic *client) CreateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int) (txHash string, err error) {
var hex hexutil.Bytes
arg := map[string]interface{}{
"from": from,
@ -142,7 +142,7 @@ func (ic *Client) CreateContract(ctx context.Context, from common.Address, bytec
return
}
func (ic *Client) CreatePrivateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int, privateFor []string) (txHash string, err error) {
func (ic *client) CreatePrivateContract(ctx context.Context, from common.Address, bytecode string, gas *big.Int, privateFor []string) (txHash string, err error) {
var hex hexutil.Bytes
arg := map[string]interface{}{
"from": from,
@ -158,7 +158,7 @@ func (ic *Client) CreatePrivateContract(ctx context.Context, from common.Address
// ----------------------------------------------------------------------------
func (ic *Client) ProposeValidator(ctx context.Context, address common.Address, auth bool) error {
func (ic *client) ProposeValidator(ctx context.Context, address common.Address, auth bool) error {
var r []byte
// TODO: Result needs to be verified with other method
// The response data type are bytes, but we cannot parse...
@ -183,7 +183,7 @@ func (addrs addresses) Swap(i, j int) {
addrs[i], addrs[j] = addrs[j], addrs[i]
}
func (ic *Client) GetValidators(ctx context.Context, blockNumbers *big.Int) ([]common.Address, error) {
func (ic *client) GetValidators(ctx context.Context, blockNumbers *big.Int) ([]common.Address, error) {
var r []common.Address
err := ic.c.CallContext(ctx, &r, "istanbul_getValidators", toNumArg(blockNumbers))
if err == nil && r == nil {

54
client/ethclient.go
View File

@ -31,7 +31,7 @@ import (
//
// Note that loading full blocks requires two requests. Use HeaderByHash
// if you don't need all transactions or uncle headers.
func (c *Client) BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) {
func (c *client) BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) {
return c.ethClient.BlockByHash(ctx, hash)
}
@ -40,122 +40,122 @@ func (c *Client) BlockByHash(ctx context.Context, hash common.Hash) (*types.Bloc
//
// Note that loading full blocks requires two requests. Use HeaderByNumber
// if you don't need all transactions or uncle headers.
func (c *Client) BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error) {
func (c *client) BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error) {
return c.ethClient.BlockByNumber(ctx, number)
}
// HeaderByHash returns the block header with the given hash.
func (c *Client) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
func (c *client) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
return c.ethClient.HeaderByHash(ctx, hash)
}
// HeaderByNumber returns a block header from the current canonical chain. If number is
// nil, the latest known header is returned.
func (c *Client) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
func (c *client) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
return c.ethClient.HeaderByNumber(ctx, number)
}
// TransactionByHash returns the transaction with the given hash.
func (c *Client) TransactionByHash(ctx context.Context, hash common.Hash) (tx *types.Transaction, isPending bool, err error) {
func (c *client) TransactionByHash(ctx context.Context, hash common.Hash) (tx *types.Transaction, isPending bool, err error) {
return c.ethClient.TransactionByHash(ctx, hash)
}
// TransactionCount returns the total number of transactions in the given block.
func (c *Client) TransactionCount(ctx context.Context, blockHash common.Hash) (uint, error) {
func (c *client) TransactionCount(ctx context.Context, blockHash common.Hash) (uint, error) {
return c.ethClient.TransactionCount(ctx, blockHash)
}
// TransactionInBlock returns a single transaction at index in the given block.
func (c *Client) TransactionInBlock(ctx context.Context, blockHash common.Hash, index uint) (*types.Transaction, error) {
func (c *client) TransactionInBlock(ctx context.Context, blockHash common.Hash, index uint) (*types.Transaction, error) {
return c.ethClient.TransactionInBlock(ctx, blockHash, index)
}
// TransactionReceipt returns the receipt of a transaction by transaction hash.
// Note that the receipt is not available for pending transactions.
func (c *Client) TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
func (c *client) TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
return c.ethClient.TransactionReceipt(ctx, txHash)
}
// SyncProgress retrieves the current progress of the sync algorithm. If there's
// no sync currently running, it returns nil.
func (c *Client) SyncProgress(ctx context.Context) (*ethereum.SyncProgress, error) {
func (c *client) SyncProgress(ctx context.Context) (*ethereum.SyncProgress, error) {
return c.ethClient.SyncProgress(ctx)
}
// SubscribeNewHead subscribes to notifications about the current blockchain head
// on the given channel.
func (c *Client) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
func (c *client) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
return c.ethClient.SubscribeNewHead(ctx, ch)
}
// State Access
// NetworkID returns the network ID (also known as the chain ID) for this chain.
func (c *Client) NetworkID(ctx context.Context) (*big.Int, error) {
func (c *client) NetworkID(ctx context.Context) (*big.Int, error) {
return c.ethClient.NetworkID(ctx)
}
// BalanceAt returns the wei balance of the given account.
// The block number can be nil, in which case the balance is taken from the latest known block.
func (c *Client) BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error) {
func (c *client) BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error) {
return c.ethClient.BalanceAt(ctx, account, blockNumber)
}
// StorageAt returns the value of key in the contract storage of the given account.
// The block number can be nil, in which case the value is taken from the latest known block.
func (c *Client) StorageAt(ctx context.Context, account common.Address, key common.Hash, blockNumber *big.Int) ([]byte, error) {
func (c *client) StorageAt(ctx context.Context, account common.Address, key common.Hash, blockNumber *big.Int) ([]byte, error) {
return c.ethClient.StorageAt(ctx, account, key, blockNumber)
}
// CodeAt returns the contract code of the given account.
// The block number can be nil, in which case the code is taken from the latest known block.
func (c *Client) CodeAt(ctx context.Context, account common.Address, blockNumber *big.Int) ([]byte, error) {
func (c *client) CodeAt(ctx context.Context, account common.Address, blockNumber *big.Int) ([]byte, error) {
return c.ethClient.CodeAt(ctx, account, blockNumber)
}
// NonceAt returns the account nonce of the given account.
// The block number can be nil, in which case the nonce is taken from the latest known block.
func (c *Client) NonceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (uint64, error) {
func (c *client) NonceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (uint64, error) {
return c.ethClient.NonceAt(ctx, account, blockNumber)
}
// Filters
// FilterLogs executes a filter query.
func (c *Client) FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) {
func (c *client) FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) {
return c.ethClient.FilterLogs(ctx, q)
}
// SubscribeFilterLogs subscribes to the results of a streaming filter query.
func (c *Client) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error) {
func (c *client) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error) {
return c.ethClient.SubscribeFilterLogs(ctx, q, ch)
}
// Pending State
// PendingBalanceAt returns the wei balance of the given account in the pending state.
func (c *Client) PendingBalanceAt(ctx context.Context, account common.Address) (*big.Int, error) {
func (c *client) PendingBalanceAt(ctx context.Context, account common.Address) (*big.Int, error) {
return c.ethClient.PendingBalanceAt(ctx, account)
}
// PendingStorageAt returns the value of key in the contract storage of the given account in the pending state.
func (c *Client) PendingStorageAt(ctx context.Context, account common.Address, key common.Hash) ([]byte, error) {
func (c *client) PendingStorageAt(ctx context.Context, account common.Address, key common.Hash) ([]byte, error) {
return c.ethClient.PendingStorageAt(ctx, account, key)
}
// PendingCodeAt returns the contract code of the given account in the pending state.
func (c *Client) PendingCodeAt(ctx context.Context, account common.Address) ([]byte, error) {
func (c *client) PendingCodeAt(ctx context.Context, account common.Address) ([]byte, error) {
return c.ethClient.PendingCodeAt(ctx, account)
}
// PendingNonceAt returns the account nonce of the given account in the pending state.
// This is the nonce that should be used for the next transaction.
func (c *Client) PendingNonceAt(ctx context.Context, account common.Address) (uint64, error) {
func (c *client) PendingNonceAt(ctx context.Context, account common.Address) (uint64, error) {
return c.ethClient.PendingNonceAt(ctx, account)
}
// PendingTransactionCount returns the total number of transactions in the pending state.
func (c *Client) PendingTransactionCount(ctx context.Context) (uint, error) {
func (c *client) PendingTransactionCount(ctx context.Context) (uint, error) {
return c.ethClient.PendingTransactionCount(ctx)
}
@ -167,19 +167,19 @@ func (c *Client) PendingTransactionCount(ctx context.Context) (uint, error) {
// blockNumber selects the block height at which the call runs. It can be nil, in which
// case the code is taken from the latest known block. Note that state from very old
// blocks might not be available.
func (c *Client) CallContract(ctx context.Context, msg ethereum.CallMsg, blockNumber *big.Int) ([]byte, error) {
func (c *client) CallContract(ctx context.Context, msg ethereum.CallMsg, blockNumber *big.Int) ([]byte, error) {
return c.ethClient.CallContract(ctx, msg, blockNumber)
}
// PendingCallContract executes a message call transaction using the EVM.
// The state seen by the contract call is the pending state.
func (c *Client) PendingCallContract(ctx context.Context, msg ethereum.CallMsg) ([]byte, error) {
func (c *client) PendingCallContract(ctx context.Context, msg ethereum.CallMsg) ([]byte, error) {
return c.ethClient.PendingCallContract(ctx, msg)
}
// SuggestGasPrice retrieves the currently suggested gas price to allow a timely
// execution of a transaction.
func (c *Client) SuggestGasPrice(ctx context.Context) (*big.Int, error) {
func (c *client) SuggestGasPrice(ctx context.Context) (*big.Int, error) {
return c.ethClient.SuggestGasPrice(ctx)
}
@ -187,7 +187,7 @@ func (c *Client) SuggestGasPrice(ctx context.Context) (*big.Int, error) {
// the current pending state of the backend blockchain. There is no guarantee that this is
// the true gas limit requirement as other transactions may be added or removed by miners,
// but it should provide a basis for setting a reasonable default.
func (c *Client) EstimateGas(ctx context.Context, msg ethereum.CallMsg) (*big.Int, error) {
func (c *client) EstimateGas(ctx context.Context, msg ethereum.CallMsg) (*big.Int, error) {
return c.ethClient.EstimateGas(ctx, msg)
}
@ -195,6 +195,6 @@ func (c *Client) EstimateGas(ctx context.Context, msg ethereum.CallMsg) (*big.In
//
// If the transaction was a contract creation use the TransactionReceipt method to get the
// contract address after the transaction has been mined.
func (c *Client) SendRawTransaction(ctx context.Context, tx *types.Transaction) error {
func (c *client) SendRawTransaction(ctx context.Context, tx *types.Transaction) error {
return c.ethClient.SendTransaction(ctx, tx)
}

4
common/transactions.go
View File

@ -29,10 +29,10 @@ 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 {
func SendEther(client client.Client, from *ecdsa.PrivateKey, to common.Address, amount *big.Int, nonce uint64) error {
tx := types.NewTransaction(nonce, to, amount, big.NewInt(DefaultGasLimit), big.NewInt(DefaultGasPrice), []byte{})
signedTx, err := types.SignTx(tx, types.NewEIP155Signer(big.NewInt(2017)), from)
if err != nil {

4
container/ethereum.go
View File

@ -67,7 +67,7 @@ type Ethereum interface {
ContainerID() string
Host() string
NewClient() *client.Client
NewClient() client.Client
ConsensusMonitor(err chan<- error, quit chan struct{})
WaitForProposed(expectedAddress common.Address, t time.Duration) error
@ -356,7 +356,7 @@ func (eth *ethereum) Running() bool {
return false
}
func (eth *ethereum) NewClient() *client.Client {
func (eth *ethereum) NewClient() client.Client {
var scheme, port string
if eth.rpcPort != "" {

5
infra/load-testing/azure.tf
View File

@ -16,7 +16,7 @@ resource "random_id" "namer" {
}
variable "username" {
type = "string"
type = "string"
default = "amis"
}
@ -67,8 +67,9 @@ resource "azurerm_container_service" "test" {
resource "null_resource" "kubeconfig" {
provisioner "local-exec" {
command = "sleep 10 && scp -o StrictHostKeyChecking=no ${var.username}@istanbul-${random_id.namer.hex}.${azurerm_resource_group.test.location}.cloudapp.azure.com:~/.kube/config ~/.kube/config"
command = "scp -o StrictHostKeyChecking=no ${var.username}@istanbul-${random_id.namer.hex}.${azurerm_resource_group.test.location}.cloudapp.azure.com:~/.kube/config ~/.kube/config"
interpreter = ["bash", "-c"]
}
depends_on = ["azurerm_container_service.test"]
}

8
infra/load-testing/kubernetes.tf
View File

@ -7,9 +7,11 @@ resource "kubernetes_service" "validator-svc" {
selector {
app = "validator-${count.index}"
}
type = "LoadBalancer"
port {
port = 8546
port = 8546
target_port = 8546
}
@ -17,4 +19,6 @@ resource "kubernetes_service" "validator-svc" {
}
count = "${length(var.svcs)}"
}
depends_on = ["null_resource.kubeconfig"]
}

2
k8s/blockchain_example.go
View File

@ -28,7 +28,7 @@ func ExampleK8SBlockchain() {
ImageRepository("quay.io/amis/geth"),
ImageTag("istanbul_develop"),
ServiceType("LoadBalancer"),
Mine(),
Mine(true),
)
defer chain.Finalize()

20
k8s/ethereum.go
View File

@ -78,6 +78,8 @@ func (eth *ethereum) Start() error {
return err
}
<-time.After(3 * time.Minute)
eth.k8sClient = k8sClient(eth.chart.Name() + "-0")
return nil
}
@ -106,7 +108,7 @@ func (eth *ethereum) DockerBinds() []string {
return nil
}
func (eth *ethereum) NewClient() *client.Client {
func (eth *ethereum) NewClient() client.Client {
for i := 0; i < healthCheckRetryCount; i++ {
client, err := client.Dial("ws://" + eth.Host() + ":8546")
if err != nil {
@ -344,11 +346,23 @@ func (eth *ethereum) AddPeer(address string) error {
}
func (eth *ethereum) StartMining() error {
return nil
client := eth.NewClient()
if client == nil {
return errors.New("failed to retrieve client")
}
defer client.Close()
return client.StartMining(context.Background())
}
func (eth *ethereum) StopMining() error {
return nil
client := eth.NewClient()
if client == nil {
return errors.New("failed to retrieve client")
}
defer client.Close()
return client.StopMining(context.Background())
}
func (eth *ethereum) Accounts() (addrs []common.Address) {

4
k8s/option.go
View File

@ -62,9 +62,9 @@ func NetworkID(networkID string) Option {
}
}
func Mine() Option {
func Mine(mine bool) Option {
return func(eth *ethereum) {
eth.args = append(eth.args, "ethereum.mining.enabled=true")
eth.args = append(eth.args, fmt.Sprintf("ethereum.mining.enabled=%v", mine))
}
}

75
k8s/transactor.go
View File

@ -18,7 +18,9 @@ package k8s
import (
"context"
"crypto/ecdsa"
"math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum/crypto"
@ -28,19 +30,30 @@ import (
)
type Transactor interface {
SendTransactions(*client.Client, *big.Int, time.Duration) error
AccountKeys() []*ecdsa.PrivateKey
SendTransactions(client.Client, []*ecdsa.PrivateKey, *big.Int, time.Duration, time.Duration) error
PreloadTransactions(client.Client, []*ecdsa.PrivateKey, *big.Int, int) error
}
func (eth *ethereum) SendTransactions(client *client.Client, amount *big.Int, duration time.Duration) error {
func (eth *ethereum) AccountKeys() []*ecdsa.PrivateKey {
return eth.accounts
}
// SendTransactions is to send a lot of transactions by each account in geth.
// duration: the period of sending transactions
// rate: total tx per second
func (eth *ethereum) SendTransactions(client client.Client, accounts []*ecdsa.PrivateKey, amount *big.Int, duration, frequnce time.Duration) error {
var fns []func() error
for i, key := range eth.accounts {
for i, key := range accounts {
i := i
key := key
fn := func() error {
fromAddr := crypto.PubkeyToAddress(key.PublicKey)
toAddr := crypto.PubkeyToAddress(eth.accounts[(i+1)%len(eth.accounts)].PublicKey)
toAddr := crypto.PubkeyToAddress(accounts[(i+1)%len(accounts)].PublicKey)
timeout := time.After(duration)
ticker := time.NewTicker(frequnce)
defer ticker.Stop()
nonce, err := client.NonceAt(context.Background(), fromAddr, nil)
if err != nil {
@ -48,14 +61,28 @@ func (eth *ethereum) SendTransactions(client *client.Client, amount *big.Int, du
return err
}
var wg sync.WaitGroup
defer wg.Wait()
errCh := make(chan error)
for {
select {
case <-timeout:
return nil
default:
if err := istcommon.SendEther(client, key, toAddr, amount, nonce); err != nil {
return err
}
case err := <-errCh:
log.Error("Failed to SendEther", "addr", fromAddr, "to", toAddr, "err", err)
return err
case <-ticker.C:
wg.Add(1)
go func(nonce uint64, wg *sync.WaitGroup) {
if err := istcommon.SendEther(client, key, toAddr, amount, nonce); err != nil {
select {
case errCh <- err:
default:
}
}
wg.Done()
}(nonce, &wg)
nonce++
}
}
@ -66,3 +93,35 @@ func (eth *ethereum) SendTransactions(client *client.Client, amount *big.Int, du
return executeInParallel(fns...)
}
func (eth *ethereum) PreloadTransactions(client client.Client, accounts []*ecdsa.PrivateKey, amount *big.Int, txCount int) error {
eachCount := txCount / len(accounts)
var fns []func() error
for i, key := range accounts {
i := i
key := key
fn := func() error {
fromAddr := crypto.PubkeyToAddress(key.PublicKey)
toAddr := crypto.PubkeyToAddress(accounts[(i+1)%len(accounts)].PublicKey)
nonce, err := client.NonceAt(context.Background(), fromAddr, nil)
if err != nil {
log.Error("Failed to get nonce", "addr", fromAddr, "err", err)
return err
}
for i := 0; i < eachCount; i++ {
if err := istcommon.SendEther(client, key, toAddr, amount, nonce); err != nil {
return err
}
nonce++
}
return nil
}
fns = append(fns, fn)
}
return executeInParallel(fns...)
}

353
metrics/blockchain.go Normal file
View File

@ -0,0 +1,353 @@
// 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"
"fmt"
"math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
ethtypes "github.com/ethereum/go-ethereum/core/types"
"github.com/getamis/istanbul-tools/client"
"github.com/getamis/istanbul-tools/container"
)
type SnapshotStopper func()
type metricsManager struct {
registry *DefaultRegistry
SentTxCounter *Counter
TxErrCounter *Counter
ExcutedTxCounter *Counter
UnknownTxCounter *Counter
ReqMeter *Meter
RespMeter *Meter
TxLatencyTimer *Timer
BlockPeriodTimer *Timer
BlockLatencyTimer *Timer
}
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"),
TxLatencyTimer: r.NewTimer("tx/latency"),
BlockPeriodTimer: r.NewTimer("block/period"),
BlockLatencyTimer: r.NewTimer("block/latency"),
}
}
func (m *metricsManager) Export() {
m.registry.Export()
}
func (m *metricsManager) SnapshotMeter(meter *Meter, name string, d time.Duration) SnapshotStopper {
mname := fmt.Sprintf("%s/%s", meter.Name(), name)
stop := make(chan struct{})
stopFn := func() {
fmt.Println("Stop snapshot", mname)
close(stop)
}
go func() {
ticker := time.NewTicker(d)
defer ticker.Stop()
for {
select {
case <-ticker.C:
snapshot := meter.Snapshot()
his := m.registry.NewHistogram(mname)
his.Update(int64(snapshot.Rate1()))
case <-stop:
return
}
}
}()
return stopFn
}
// --------------------------------------------------------------------------------------------------
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, 1000),
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.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
// 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())))
// 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)
}

67
metrics/ethereum.go Normal file
View File

@ -0,0 +1,67 @@
// 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 (
"crypto/ecdsa"
"errors"
"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) AccountKeys() []*ecdsa.PrivateKey {
transactor, ok := eth.Ethereum.(k8s.Transactor)
if !ok {
return nil
}
return transactor.AccountKeys()
}
func (eth *metricEthereum) SendTransactions(client client.Client, accounts []*ecdsa.PrivateKey, amount *big.Int, duration, frequnce time.Duration) error {
transactor, ok := eth.Ethereum.(k8s.Transactor)
if !ok {
return errors.New("Not support Transactor interface.")
}
return transactor.SendTransactions(client, accounts, amount, duration, frequnce)
}
func (eth *metricEthereum) PreloadTransactions(client client.Client, accounts []*ecdsa.PrivateKey, amount *big.Int, txCount int) error {
transactor, ok := eth.Ethereum.(k8s.Transactor)
if !ok {
return errors.New("Not support Transactor interface.")
}
return transactor.PreloadTransactions(client, accounts, amount, txCount)
}

55
metrics/exporter.go Normal file
View File

@ -0,0 +1,55 @@
// 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 Exporter interface {
Export()
SentTxCount() int64
ExcutedTxCount() int64
SnapshotTxReqMeter(name string) SnapshotStopper
SnapshotTxRespMeter(name string) SnapshotStopper
}
func (mc *metricChain) Export() {
mc.metricsMgr.Export()
}
func (mc *metricChain) SentTxCount() int64 {
return mc.metricsMgr.SentTxCounter.Snapshot().Count()
}
func (mc *metricChain) ExcutedTxCount() int64 {
return mc.metricsMgr.ExcutedTxCounter.Snapshot().Count()
}
func (mc *metricChain) SnapshotTxReqMeter(name string) SnapshotStopper {
if name == "" {
name = "snapshot"
}
return mc.metricsMgr.SnapshotMeter(mc.metricsMgr.ReqMeter, name, 5*time.Second)
}
func (mc *metricChain) SnapshotTxRespMeter(name string) SnapshotStopper {
if name == "" {
name = "snapshot"
}
return mc.metricsMgr.SnapshotMeter(mc.metricsMgr.RespMeter, name, 5*time.Second)
}

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()

140
metrics/metrics.go Normal file
View File

@ -0,0 +1,140 @@
// 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 {
metrics.Registry
}
func NewRegistry() *DefaultRegistry {
r := metrics.NewRegistry()
return &DefaultRegistry{r}
}
func (r *DefaultRegistry) NewCounter(name string) *Counter {
return &Counter{metrics.GetOrRegisterCounter(name, r.Registry), name}
}
func (r *DefaultRegistry) NewMeter(name string) *Meter {
return &Meter{metrics.GetOrRegisterMeter(name, r.Registry), name}
}
func (r *DefaultRegistry) NewTimer(name string) *Timer {
return &Timer{metrics.GetOrRegisterTimer(name, r.Registry), name}
}
func (r *DefaultRegistry) NewHistogram(name string) *Histogram {
return &Histogram{metrics.GetOrRegisterHistogram(name, r.Registry, metrics.NewExpDecaySample(1028, 0.015)), name}
}
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.90, 0.95, 0.99})
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(" 90%%: %e\n", ps[2])
fmt.Printf(" 95%%: %e\n", ps[3])
fmt.Printf(" 99%%: %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.90, 0.95, 0.99})
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(" 90%%: %e\n", ps[2])
fmt.Printf(" 95%%: %e\n", ps[3])
fmt.Printf(" 99%%: %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())
}
})
}
type Counter struct {
metrics.Counter
name string
}
func (c *Counter) Name() string { return c.name }
type Meter struct {
metrics.Meter
name string
}
func (m *Meter) Name() string { return m.name }
type Timer struct {
metrics.Timer
name string
}
func (t *Timer) Name() string { return t.name }
type Histogram struct {
metrics.Histogram
name string
}
func (h *Histogram) Name() string { return h.name }

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
}

109
tests/load/load_test.go
View File

@ -17,6 +17,7 @@
package load
import (
"fmt"
"math/big"
"sync"
"testing"
@ -26,8 +27,10 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
istcommon "github.com/getamis/istanbul-tools/common"
"github.com/getamis/istanbul-tools/container"
"github.com/getamis/istanbul-tools/k8s"
"github.com/getamis/istanbul-tools/metrics"
"github.com/getamis/istanbul-tools/tests"
)
@ -38,14 +41,20 @@ var _ = Describe("TPS-01: Large amount of transactions", func() {
func(gaslimit int) {
tests.CaseTable("with txpool size",
func(txpoolSize int) {
runTests(numberOfValidators, gaslimit, txpoolSize)
tests.CaseTable("with tx send rate",
func(rate int) {
runTests(numberOfValidators, gaslimit, txpoolSize, rate)
},
// only preload txs if send rare is 0
tests.Case("150ms", 150),
)
},
tests.Case("10240", 10240),
tests.Case("20480", 20480),
)
},
tests.Case("21000*3000", 21000*3000),
tests.Case("21000*1500", 21000*1500),
)
},
@ -54,14 +63,20 @@ var _ = Describe("TPS-01: Large amount of transactions", func() {
)
})
func runTests(numberOfValidators int, gaslimit int, txpoolSize int) {
func runTests(numberOfValidators int, gaslimit int, txpoolSize int, sendRate int) {
Describe("", func() {
const (
preloadAccounts = 10
sendAccount = 20
)
var (
blockchain container.Blockchain
sendEtherAddrs map[common.Address]common.Address
duration = 10 * time.Minute
accountsPerGeth = 30
duration = 5 * time.Minute
accountsPerGeth = preloadAccounts + sendAccount
allTPSSanpshotStopper metrics.SnapshotStopper
)
BeforeEach(func() {
@ -71,9 +86,10 @@ func runTests(numberOfValidators int, gaslimit int, txpoolSize int) {
uint64(gaslimit),
k8s.ImageRepository("quay.io/amis/geth"),
k8s.ImageTag("istanbul_develop"),
k8s.Mine(),
k8s.Mine(false),
k8s.TxPoolSize(txpoolSize),
)
blockchain = metrics.NewMetricChain(blockchain)
Expect(blockchain).NotTo(BeNil())
Expect(blockchain.Start(true)).To(BeNil())
@ -82,23 +98,34 @@ func runTests(numberOfValidators int, gaslimit int, txpoolSize int) {
for i, v := range blockchain.Validators() {
sendEtherAddrs[v.Address()] = blockchain.Validators()[(i+1)%num].Address()
}
if metricsExport, ok := blockchain.(metrics.Exporter); ok {
allTPSSanpshotStopper = metricsExport.SnapshotTxRespMeter("all")
}
})
AfterEach(func() {
Expect(blockchain).NotTo(BeNil())
if allTPSSanpshotStopper != nil {
allTPSSanpshotStopper()
}
fmt.Println("Begin to Stop blockchain")
Expect(blockchain.Stop(true)).To(BeNil())
fmt.Println("End to Stop blockchain")
blockchain.Finalize()
})
It("", func() {
By("Wait for p2p connection", func() {
tests.WaitFor(blockchain.Validators(), func(geth container.Ethereum, wg *sync.WaitGroup) {
fmt.Println("Start p2p")
Expect(geth.WaitForPeersConnected(numberOfValidators - 1)).To(BeNil())
wg.Done()
fmt.Println("Done p2p")
})
})
By("Send transactions", func() {
By("Preload transactions", func() {
tests.WaitFor(blockchain.Validators(), func(geth container.Ethereum, wg *sync.WaitGroup) {
transactor, ok := geth.(k8s.Transactor)
Expect(ok).To(BeTrue())
@ -106,14 +133,78 @@ func runTests(numberOfValidators int, gaslimit int, txpoolSize int) {
client := geth.NewClient()
Expect(client).NotTo(BeNil())
accounts := transactor.AccountKeys()[:preloadAccounts]
preloadCnt := txpoolSize
Expect(transactor.PreloadTransactions(
client,
accounts,
new(big.Int).Exp(big.NewInt(10), big.NewInt(3), nil),
preloadCnt)).To(BeNil())
wg.Done()
})
})
By("Start mining", func() {
tests.WaitFor(blockchain.Validators(), func(geth container.Ethereum, wg *sync.WaitGroup) {
fmt.Println("Start mining")
Expect(geth.StartMining()).To(BeNil())
wg.Done()
})
})
By("Send transactions with specific rate", func() {
if sendRate == 0 {
fmt.Println("Skip to send tx")
return
}
fmt.Println("Start to send tx")
if metricsExport, ok := blockchain.(metrics.Exporter); ok {
mname := fmt.Sprintf("rate%dms", sendRate)
rpsSanpshotStopper := metricsExport.SnapshotTxReqMeter(mname)
defer rpsSanpshotStopper()
tpsSanpshotStopper := metricsExport.SnapshotTxRespMeter(mname)
defer tpsSanpshotStopper()
}
tests.WaitFor(blockchain.Validators(), func(geth container.Ethereum, wg *sync.WaitGroup) {
transactor, ok := geth.(k8s.Transactor)
Expect(ok).To(BeTrue())
client := geth.NewClient()
Expect(client).NotTo(BeNil())
accounts := transactor.AccountKeys()[preloadAccounts:]
rate := time.Duration(sendRate) * time.Millisecond
Expect(transactor.SendTransactions(
client,
accounts,
new(big.Int).Exp(big.NewInt(10), big.NewInt(3), nil),
duration)).To(BeNil())
duration,
rate)).To(BeNil())
wg.Done()
})
})
By("Wait for txs consuming", func() {
var blocksCnt int = 5
metricsExport, ok := blockchain.(metrics.Exporter)
if ok {
blockSize := gaslimit / int(istcommon.DefaultGasLimit)
blocksCnt = int(int(metricsExport.SentTxCount()-metricsExport.ExcutedTxCount())/blockSize/7*10) + 5
fmt.Println("blockSize", blockSize, "sendTx", metricsExport.SentTxCount(), "excutedTx", metricsExport.ExcutedTxCount(), "waitFor", blocksCnt)
tpsSanpshotStopper := metricsExport.SnapshotTxRespMeter("final")
defer tpsSanpshotStopper()
}
tests.WaitFor(blockchain.Validators(), func(geth container.Ethereum, wg *sync.WaitGroup) {
Expect(geth.WaitForBlocks(blocksCnt)).To(BeNil())
wg.Done()
})
})
})
})
}

4
tests/quorum/functional/private_transaction_test.go
View File

@ -190,7 +190,7 @@ func genByteCodeWithValue(v int) string {
return fmt.Sprintf("%s%064x", testBaseByteCode, v)
}
func getTxReceipt(ethClient *client.Client, txHash common.Hash, timeout time.Duration) (*types.Receipt, error) {
func getTxReceipt(ethClient client.Client, txHash common.Hash, timeout time.Duration) (*types.Receipt, error) {
timer := time.After(timeout)
ticker := time.NewTicker(3 * time.Second)
defer ticker.Stop()
@ -207,7 +207,7 @@ func getTxReceipt(ethClient *client.Client, txHash common.Hash, timeout time.Dur
}
}
func checkContractValue(ethClient *client.Client, txHash common.Hash, expValue int) error {
func checkContractValue(ethClient client.Client, txHash common.Hash, expValue int) error {
receipt, err := getTxReceipt(ethClient, txHash, 10*time.Second)
if err != nil {
return err