mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-01 05:37:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

proxy: improve ABCI app connection handling (#5078)

Browse Source 
Closes #5074

Old code does not work when --consensus.create_empty_blocks=false
(because it only calls tmos.Kill when ApplyBlock fails). New code is
listening ABCI clients for Quit and kills TM process if there were any
errors.
This commit is contained in:
Anton Kaliaev
2020-12-23 18:44:59 +04:00
parent 0aaaa34038
commit 07113abd9e
7 changed files with 810 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/client.go
+10 -4
View File
@@ -11,8 +11,9 @@ import (
"github.com/tendermint/tendermint/abci/types"
)
// NewABCIClient returns newly connected client
// ClientCreator creates new ABCI clients.
type ClientCreator interface {
// NewABCIClient returns a new ABCI client.
NewABCIClient() (abcicli.Client, error)
}
@@ -24,6 +25,8 @@ type localClientCreator struct {
app types.Application
}
// NewLocalClientCreator returns a ClientCreator for the given app,
// which will be running locally.
func NewLocalClientCreator(app types.Application) ClientCreator {
return &localClientCreator{
mtx: new(sync.Mutex),
@@ -44,6 +47,9 @@ type remoteClientCreator struct {
mustConnect bool
}
// NewRemoteClientCreator returns a ClientCreator for the given address (e.g.
// "192.168.0.1") and transport (e.g. "tcp"). Set mustConnect to true if you
// want the client to connect before reporting success.
func NewRemoteClientCreator(addr, transport string, mustConnect bool) ClientCreator {
return &remoteClientCreator{
addr: addr,
@@ -60,9 +66,9 @@ func (r *remoteClientCreator) NewABCIClient() (abcicli.Client, error) {
return remoteApp, nil
}
//-----------------------------------------------------------------
// default
// DefaultClientCreator returns a default ClientCreator, which will create a
// local client if addr is one of: 'counter', 'counter_serial', 'kvstore',
// 'persistent_kvstore' or 'noop', otherwise - a remote client.
func DefaultClientCreator(addr, transport, dbDir string) ClientCreator {
switch addr {
case "counter":
proxy/mocks/client_creator.go
+36
View File
@@ -0,0 +1,36 @@
// Code generated by mockery v2.3.0. DO NOT EDIT.
package mocks
import (
mock "github.com/stretchr/testify/mock"
abcicli "github.com/tendermint/tendermint/abci/client"
)
// ClientCreator is an autogenerated mock type for the ClientCreator type
type ClientCreator struct {
mock.Mock
}
// NewABCIClient provides a mock function with given fields:
func (_m *ClientCreator) NewABCIClient() (abcicli.Client, error) {
ret := _m.Called()
var r0 abcicli.Client
if rf, ok := ret.Get(0).(func() abcicli.Client); ok {
r0 = rf()
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(abcicli.Client)
}
}
var r1 error
if rf, ok := ret.Get(1).(func() error); ok {
r1 = rf()
} else {
r1 = ret.Error(1)
}
return r0, r1
}
proxy/multi_app_conn.go
+97 -37
View File
@@ -1,43 +1,58 @@
package proxy
import (
"github.com/pkg/errors"
"fmt"
abcicli "github.com/tendermint/tendermint/abci/client"
tmlog "github.com/tendermint/tendermint/libs/log"
tmos "github.com/tendermint/tendermint/libs/os"
"github.com/tendermint/tendermint/libs/service"
)
//-----------------------------
const (
connConsensus = "consensus"
connMempool = "mempool"
connQuery = "query"
)
// Tendermint's interface to the application consists of multiple connections
// AppConns is the Tendermint's interface to the application that consists of
// multiple connections.
type AppConns interface {
service.Service
// Mempool connection
Mempool() AppConnMempool
// Consensus connection
Consensus() AppConnConsensus
// Query connection
Query() AppConnQuery
}
// NewAppConns calls NewMultiAppConn.
func NewAppConns(clientCreator ClientCreator) AppConns {
return NewMultiAppConn(clientCreator)
}
//-----------------------------
// multiAppConn implements AppConns
// a multiAppConn is made of a few appConns (mempool, consensus, query)
// and manages their underlying abci clients
// multiAppConn implements AppConns.
//
// A multiAppConn is made of a few appConns and manages their underlying abci
// clients.
// TODO: on app restart, clients must reboot together
type multiAppConn struct {
service.BaseService
mempoolConn AppConnMempool
consensusConn AppConnConsensus
mempoolConn AppConnMempool
queryConn AppConnQuery
consensusConnClient abcicli.Client
mempoolConnClient abcicli.Client
queryConnClient abcicli.Client
clientCreator ClientCreator
}
// Make all necessary abci connections to the application
// NewMultiAppConn makes all necessary abci connections to the application.
func NewMultiAppConn(clientCreator ClientCreator) AppConns {
multiAppConn := &multiAppConn{
clientCreator: clientCreator,
@@ -46,54 +61,99 @@ func NewMultiAppConn(clientCreator ClientCreator) AppConns {
return multiAppConn
}
// Returns the mempool connection
func (app *multiAppConn) Mempool() AppConnMempool {
return app.mempoolConn
}
// Returns the consensus Connection
func (app *multiAppConn) Consensus() AppConnConsensus {
return app.consensusConn
}
// Returns the query Connection
func (app *multiAppConn) Query() AppConnQuery {
return app.queryConn
}
func (app *multiAppConn) OnStart() error {
// query connection
querycli, err := app.clientCreator.NewABCIClient()
c, err := app.abciClientFor(connQuery)
if err != nil {
return errors.Wrap(err, "Error creating ABCI client (query connection)")
return err
}
querycli.SetLogger(app.Logger.With("module", "abci-client", "connection", "query"))
if err := querycli.Start(); err != nil {
return errors.Wrap(err, "Error starting ABCI client (query connection)")
}
app.queryConn = NewAppConnQuery(querycli)
app.queryConnClient = c
app.queryConn = NewAppConnQuery(c)
// mempool connection
memcli, err := app.clientCreator.NewABCIClient()
c, err = app.abciClientFor(connMempool)
if err != nil {
return errors.Wrap(err, "Error creating ABCI client (mempool connection)")
app.stopAllClients()
return err
}
memcli.SetLogger(app.Logger.With("module", "abci-client", "connection", "mempool"))
if err := memcli.Start(); err != nil {
return errors.Wrap(err, "Error starting ABCI client (mempool connection)")
}
app.mempoolConn = NewAppConnMempool(memcli)
app.mempoolConnClient = c
app.mempoolConn = NewAppConnMempool(c)
// consensus connection
concli, err := app.clientCreator.NewABCIClient()
c, err = app.abciClientFor(connConsensus)
if err != nil {
return errors.Wrap(err, "Error creating ABCI client (consensus connection)")
app.stopAllClients()
return err
}
concli.SetLogger(app.Logger.With("module", "abci-client", "connection", "consensus"))
if err := concli.Start(); err != nil {
return errors.Wrap(err, "Error starting ABCI client (consensus connection)")
}
app.consensusConn = NewAppConnConsensus(concli)
app.consensusConnClient = c
app.consensusConn = NewAppConnConsensus(c)
// Kill Tendermint if the ABCI application crashes.
go app.killTMOnClientError()
return nil
}
func (app *multiAppConn) OnStop() {
app.stopAllClients()
}
func (app *multiAppConn) killTMOnClientError() {
killFn := func(conn string, err error, logger tmlog.Logger) {
logger.Error(
fmt.Sprintf("%s connection terminated. Did the application crash? Please restart tendermint", conn),
"err", err)
killErr := tmos.Kill()
if killErr != nil {
logger.Error("Failed to kill this process - please do so manually", "err", killErr)
}
}
select {
case <-app.consensusConnClient.Quit():
if err := app.consensusConnClient.Error(); err != nil {
killFn(connConsensus, err, app.Logger)
}
case <-app.mempoolConnClient.Quit():
if err := app.mempoolConnClient.Error(); err != nil {
killFn(connMempool, err, app.Logger)
}
case <-app.queryConnClient.Quit():
if err := app.queryConnClient.Error(); err != nil {
killFn(connQuery, err, app.Logger)
}
}
}
func (app *multiAppConn) stopAllClients() {
if app.consensusConnClient != nil {
app.consensusConnClient.Stop()
}
if app.mempoolConnClient != nil {
app.mempoolConnClient.Stop()
}
if app.queryConnClient != nil {
app.queryConnClient.Stop()
}
}
func (app *multiAppConn) abciClientFor(conn string) (abcicli.Client, error) {
c, err := app.clientCreator.NewABCIClient()
if err != nil {
return nil, fmt.Errorf("error creating ABCI client (%s connection): %w", conn, err)
}
c.SetLogger(app.Logger.With("module", "abci-client", "connection", conn))
if err := c.Start(); err != nil {
return nil, fmt.Errorf("error starting ABCI client (%s connection): %w", conn, err)
}
return c, nil
}
proxy/multi_app_conn_test.go
+85
View File
@@ -0,0 +1,85 @@
package proxy
import (
"errors"
"os"
"os/signal"
"syscall"
"testing"
"time"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
abcimocks "github.com/tendermint/tendermint/abci/client/mocks"
"github.com/tendermint/tendermint/proxy/mocks"
)
func TestAppConns_Start_Stop(t *testing.T) {
quitCh := make(<-chan struct{})
clientCreatorMock := &mocks.ClientCreator{}
clientMock := &abcimocks.Client{}
clientMock.On("SetLogger", mock.Anything).Return().Times(3)
clientMock.On("Start").Return(nil).Times(3)
clientMock.On("Stop").Return(nil).Times(3)
clientMock.On("Quit").Return(quitCh).Times(3)
clientCreatorMock.On("NewABCIClient").Return(clientMock, nil).Times(3)
appConns := NewAppConns(clientCreatorMock)
err := appConns.Start()
require.NoError(t, err)
time.Sleep(100 * time.Millisecond)
appConns.Stop()
clientMock.AssertExpectations(t)
}
// Upon failure, we call tmos.Kill
func TestAppConns_Failure(t *testing.T) {
ok := make(chan struct{})
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGTERM)
go func() {
for range c {
close(ok)
}
}()
quitCh := make(chan struct{})
var recvQuitCh <-chan struct{} // nolint:gosimple
recvQuitCh = quitCh
clientCreatorMock := &mocks.ClientCreator{}
clientMock := &abcimocks.Client{}
clientMock.On("SetLogger", mock.Anything).Return()
clientMock.On("Start").Return(nil)
clientMock.On("Stop").Return(nil)
clientMock.On("Quit").Return(recvQuitCh)
clientMock.On("Error").Return(errors.New("EOF")).Once()
clientCreatorMock.On("NewABCIClient").Return(clientMock, nil)
appConns := NewAppConns(clientCreatorMock)
err := appConns.Start()
require.NoError(t, err)
defer appConns.Stop()
// simulate failure
close(quitCh)
select {
case <-ok:
t.Log("SIGTERM successfully received")
case <-time.After(5 * time.Second):
t.Fatal("expected process to receive SIGTERM signal")
}
}