container: add monitor to detect consensus status
This commit is contained in:
parent
5af37f8b17
commit
35f81242b2
|
@ -36,6 +36,7 @@ import (
|
|||
"github.com/docker/docker/client"
|
||||
"github.com/docker/go-connections/nat"
|
||||
"github.com/ethereum/go-ethereum/cmd/utils"
|
||||
ethtypes "github.com/ethereum/go-ethereum/core/types"
|
||||
"github.com/ethereum/go-ethereum/ethclient"
|
||||
"github.com/ethereum/go-ethereum/p2p/discover"
|
||||
|
||||
|
@ -48,6 +49,10 @@ const (
|
|||
healthCheckRetryDelay = 2 * time.Second
|
||||
)
|
||||
|
||||
var (
|
||||
ErrConsensusTimeout = errors.New("consensus timeout")
|
||||
)
|
||||
|
||||
type Ethereum interface {
|
||||
Init(string) error
|
||||
Start() error
|
||||
|
@ -59,6 +64,7 @@ type Ethereum interface {
|
|||
Host() string
|
||||
NewClient() *ethclient.Client
|
||||
NewIstanbulClient() *istclient.Client
|
||||
ConsensusMonitor(err chan<- error, quit chan struct{})
|
||||
}
|
||||
|
||||
func NewEthereum(c *client.Client, options ...Option) *ethereum {
|
||||
|
@ -357,6 +363,47 @@ func (eth *ethereum) NodeAddress() string {
|
|||
return ""
|
||||
}
|
||||
|
||||
func (eth *ethereum) ConsensusMonitor(errCh chan<- error, quit chan struct{}) {
|
||||
cli := eth.NewClient()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
subCh := make(chan *ethtypes.Header)
|
||||
|
||||
sub, err := cli.SubscribeNewHead(ctx, subCh)
|
||||
if err != nil {
|
||||
log.Fatal(fmt.Sprintf("subscribe error:%v", err))
|
||||
errCh <- err
|
||||
return
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
for {
|
||||
latestUpdate := time.Now()
|
||||
select {
|
||||
case err := <-sub.Err():
|
||||
log.Printf("Connection lost: %v", err)
|
||||
errCh <- err
|
||||
return
|
||||
case head := <-subCh:
|
||||
// Ensure that mining is stable.
|
||||
if head.Number.Uint64() < 3 {
|
||||
continue
|
||||
}
|
||||
|
||||
// Block is generated by 2 seconds. We tolerate 1 second delay in consensus.
|
||||
if time.Now().Sub(latestUpdate).Seconds() > 3.0 {
|
||||
errCh <- ErrConsensusTimeout
|
||||
return
|
||||
}
|
||||
latestUpdate = time.Now()
|
||||
case <-quit:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
func (eth *ethereum) showLog(context context.Context) {
|
||||
|
|
|
@ -20,6 +20,7 @@ import (
|
|||
"context"
|
||||
"math/big"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
|
@ -43,7 +44,7 @@ var _ = Describe("4 validators Istanbul", func() {
|
|||
container.DataDir("/data"),
|
||||
container.WebSocket(),
|
||||
container.WebSocketAddress("0.0.0.0"),
|
||||
container.WebSocketAPI("admin,eth,net,web3,personal,miner"),
|
||||
container.WebSocketAPI("admin,eth,net,web3,personal,miner,istanbul"),
|
||||
container.WebSocketOrigin("*"),
|
||||
container.NAT("any"),
|
||||
container.NoDiscover(),
|
||||
|
@ -69,9 +70,32 @@ var _ = Describe("4 validators Istanbul", func() {
|
|||
Expect(err).To(BeNil())
|
||||
Expect(block).NotTo(BeNil())
|
||||
}
|
||||
|
||||
By("Ensure that consensus is working in 30 seconds", func() { ensureConsensusWorking(blockchain.Validators(), 30*time.Second) })
|
||||
})
|
||||
})
|
||||
|
||||
func ensureConsensusWorking(geths []container.Ethereum, t time.Duration) {
|
||||
errCh := make(chan error, len(geths))
|
||||
quitCh := make(chan struct{}, len(geths))
|
||||
for _, geth := range geths {
|
||||
go geth.ConsensusMonitor(errCh, quitCh)
|
||||
}
|
||||
|
||||
timeout := time.NewTimer(t)
|
||||
|
||||
defer timeout.Stop()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
Expect(err).Should(BeNil())
|
||||
case <-timeout.C:
|
||||
for i := 0; i < len(geths); i++ {
|
||||
quitCh <- struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestIstanbul(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Istanbul Test Suite")
|
||||
|
|
Loading…
Reference in New Issue