processes.go

package core

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/hyperledger/burrow/bcm"

	"github.com/hyperledger/burrow/execution"

	"github.com/hyperledger/burrow/consensus/abci"

	"github.com/hyperledger/burrow/keys"
	"github.com/hyperledger/burrow/logging/structure"
	"github.com/hyperledger/burrow/process"
	"github.com/hyperledger/burrow/project"
	"github.com/hyperledger/burrow/rpc"
	"github.com/hyperledger/burrow/rpc/metrics"
	"github.com/hyperledger/burrow/rpc/rpcdump"
	"github.com/hyperledger/burrow/rpc/rpcevents"
	"github.com/hyperledger/burrow/rpc/rpcinfo"
	"github.com/hyperledger/burrow/rpc/rpcquery"
	"github.com/hyperledger/burrow/rpc/rpctransact"
	"github.com/hyperledger/burrow/txs"
	"github.com/tendermint/tendermint/version"
	"github.com/tmthrgd/go-hex"
)

const (
	ProfilingProcessName   = "Profiling"
	DatabaseProcessName    = "Database"
	NoConsensusProcessName = "NoConsensusExecution"
	TendermintProcessName  = "Tendermint"
	StartupProcessName     = "StartupAnnouncer"
	InfoProcessName        = "rpcConfig/info"
	GRPCProcessName        = "rpcConfig/GRPC"
	MetricsProcessName     = "rpcConfig/metrics"
)

func DefaultProcessLaunchers(kern *Kernel, rpcConfig *rpc.RPCConfig, keysConfig *keys.KeysConfig) []process.Launcher {
	// Run announcer after Tendermint so it can get some details
	return []process.Launcher{
		ProfileLauncher(kern, rpcConfig.Profiler),
		DatabaseLauncher(kern),
		NoConsensusLauncher(kern),
		TendermintLauncher(kern),
		StartupLauncher(kern),
		InfoLauncher(kern, rpcConfig.Info),
		MetricsLauncher(kern, rpcConfig.Metrics),
		GRPCLauncher(kern, rpcConfig.GRPC, keysConfig),
	}
}

func ProfileLauncher(kern *Kernel, conf *rpc.ServerConfig) process.Launcher {
	return process.Launcher{
		Name:    ProfilingProcessName,
		Enabled: conf.Enabled,
		Launch: func() (process.Process, error) {
			debugServer := &http.Server{
				Addr: conf.ListenAddress,
			}
			go func() {
				err := debugServer.ListenAndServe()
				if err != nil {
					kern.Logger.InfoMsg("Error from pprof debug server", structure.ErrorKey, err)
				}
			}()
			return debugServer, nil
		},
	}
}

func DatabaseLauncher(kern *Kernel) process.Launcher {
	return process.Launcher{
		Name:    DatabaseProcessName,
		Enabled: true,
		Launch: func() (process.Process, error) {
			// Just close database
			return process.ShutdownFunc(func(ctx context.Context) error {
				kern.database.Close()
				return nil
			}), nil
		},
	}
}

// Run a single uncoordinated local state
func NoConsensusLauncher(kern *Kernel) process.Launcher {
	return process.Launcher{
		Name:    NoConsensusProcessName,
		Enabled: kern.Node == nil,
		Launch: func() (process.Process, error) {
			accountState := kern.State
			nameRegState := kern.State
			kern.Service = rpc.NewService(accountState, nameRegState, kern.Blockchain, kern.State, nil, kern.Logger)
			// TimeoutFactor scales in units of seconds
			blockDuration := time.Duration(kern.timeoutFactor * float64(time.Second))
			//proc := abci.NewProcess(kern.checker, kern.committer, kern.Blockchain, kern.txCodec, blockDuration, kern.Panic)
			proc := abci.NewProcess(kern.committer, kern.Blockchain, kern.txCodec, blockDuration, kern.Panic)
			// Provide execution accounts against backend state since we will commit immediately
			accounts := execution.NewAccounts(kern.committer, kern.keyClient, AccountsRingMutexCount)
			// Elide consensus and use a CheckTx function that immediately commits any valid transaction
			kern.Transactor = execution.NewTransactor(kern.Blockchain, kern.Emitter, accounts, proc.CheckTx, kern.txCodec,
				kern.Logger)
			return proc, nil
		},
	}
}

func TendermintLauncher(kern *Kernel) process.Launcher {
	return process.Launcher{
		Name:    TendermintProcessName,
		Enabled: kern.Node != nil,
		Launch: func() (process.Process, error) {
			const errHeader = "TendermintLauncher():"
			nodeView, err := kern.GetNodeView()
			if err != nil {
				return nil, fmt.Errorf("%s cannot get NodeView %v", errHeader, err)
			}

			accountState := kern.State
			nameRegState := kern.State
			kern.Service = rpc.NewService(accountState, nameRegState, kern.Blockchain, kern.State, nodeView, kern.Logger)

			kern.Blockchain.SetBlockStore(bcm.NewBlockStore(nodeView.BlockStore()))
			// Provide execution accounts against checker state so that we can assign sequence numbers
			accounts := execution.NewAccounts(kern.checker, kern.keyClient, AccountsRingMutexCount)
			// Pass transactions to Tendermint's CheckTx function for broadcast and consensus
			checkTx := kern.Node.MempoolReactor().Mempool.CheckTx
			kern.Transactor = execution.NewTransactor(kern.Blockchain, kern.Emitter, accounts, checkTx, kern.txCodec,
				kern.Logger)

			if err := kern.Node.Start(); err != nil {
				return nil, fmt.Errorf("%s error starting Tendermint node: %v", errHeader, err)
			}

			return process.ShutdownFunc(func(ctx context.Context) error {
				err := kern.Node.Stop()
				// Close tendermint database connections using our wrapper
				defer kern.Node.Close()
				if err != nil {
					return err
				}
				select {
				case <-ctx.Done():
					return ctx.Err()
				case <-kern.Node.Quit():
					kern.Logger.InfoMsg("Tendermint Node has quit, closing DB connections...")
					return nil
				}
			}), nil
		},
	}
}

func StartupLauncher(kern *Kernel) process.Launcher {
	return process.Launcher{
		Name:    StartupProcessName,
		Enabled: true,
		Launch: func() (process.Process, error) {
			start := time.Now()
			shutdown := process.ShutdownFunc(func(ctx context.Context) error {
				stop := time.Now()
				return kern.Logger.InfoMsg("Burrow is shutting down. Prepare for re-entrancy.",
					"announce", "shutdown",
					"shutdown_time", stop,
					"elapsed_run_time", stop.Sub(start).String())
			})

			if kern.Node == nil {
				return shutdown, nil
			}

			nodeView, err := kern.GetNodeView()
			if err != nil {
				return nil, err
			}

			genesisDoc := kern.Blockchain.GenesisDoc()
			info := kern.Node.NodeInfo()
			logger := kern.Logger.With(
				"launch_time", start,
				"burrow_version", project.FullVersion(),
				"tendermint_version", version.TMCoreSemVer,
				"validator_address", nodeView.ValidatorAddress(),
				"node_id", string(info.ID()),
				"net_address", info.NetAddress().String(),
				"genesis_app_hash", genesisDoc.AppHash.String(),
				"genesis_hash", hex.EncodeUpperToString(genesisDoc.Hash()),
			)

			err = logger.InfoMsg("Burrow is launching. We have marmot-off.", "announce", "startup")
			return shutdown, err
		},
	}
}

func InfoLauncher(kern *Kernel, conf *rpc.ServerConfig) process.Launcher {
	return process.Launcher{
		Name:    InfoProcessName,
		Enabled: conf.Enabled,
		Launch: func() (process.Process, error) {
			listener, err := process.ListenerFromAddress(conf.ListenAddress)
			if err != nil {
				return nil, err
			}
			err = kern.registerListener(InfoProcessName, listener)
			if err != nil {
				return nil, err
			}
			server, err := rpcinfo.StartServer(kern.Service, "/websocket", listener, kern.Logger)
			if err != nil {
				return nil, err
			}
			return server, nil
		},
	}
}

func MetricsLauncher(kern *Kernel, conf *rpc.MetricsConfig) process.Launcher {
	return process.Launcher{
		Name:    MetricsProcessName,
		Enabled: conf.Enabled,
		Launch: func() (process.Process, error) {
			listener, err := process.ListenerFromAddress(conf.ListenAddress)
			if err != nil {
				return nil, err
			}
			err = kern.registerListener(MetricsProcessName, listener)
			if err != nil {
				return nil, err
			}
			server, err := metrics.StartServer(kern.Service, conf.MetricsPath, listener, conf.BlockSampleSize,
				kern.Logger)
			if err != nil {
				return nil, err
			}
			return server, nil
		},
	}
}

func GRPCLauncher(kern *Kernel, conf *rpc.ServerConfig, keyConfig *keys.KeysConfig) process.Launcher {
	return process.Launcher{
		Name:    GRPCProcessName,
		Enabled: conf.Enabled,
		Launch: func() (process.Process, error) {
			nodeView, err := kern.GetNodeView()
			if err != nil {
				return nil, err
			}

			listener, err := process.ListenerFromAddress(conf.ListenAddress)
			if err != nil {
				return nil, err
			}
			err = kern.registerListener(GRPCProcessName, listener)
			if err != nil {
				return nil, err
			}

			grpcServer := rpc.NewGRPCServer(kern.Logger)
			var ks *keys.KeyStore
			if kern.keyStore != nil {
				ks = kern.keyStore
			}
			grpcServer.GetServiceInfo()

			if keyConfig.GRPCServiceEnabled {
				if kern.keyStore == nil {
					ks = keys.NewKeyStore(keyConfig.KeysDirectory, keyConfig.AllowBadFilePermissions)
				}
				keys.RegisterKeysServer(grpcServer, ks)
			}

			nameRegState := kern.State
			proposalRegState := kern.State
			rpcquery.RegisterQueryServer(grpcServer, rpcquery.NewQueryServer(kern.State, nameRegState, proposalRegState,
				kern.Blockchain, kern.State, nodeView, kern.Logger))

			txCodec := txs.NewAminoCodec()
			rpctransact.RegisterTransactServer(grpcServer, rpctransact.NewTransactServer(kern.Transactor, txCodec))

			rpcevents.RegisterExecutionEventsServer(grpcServer, rpcevents.NewExecutionEventsServer(kern.State,
				kern.Emitter, kern.Blockchain, kern.Logger))

			rpcdump.RegisterDumpServer(grpcServer, rpcdump.NewDumpServer(kern.State, kern.Blockchain, kern.Logger))

			// Provides metadata about services registered
			// reflection.Register(grpcServer)

			go grpcServer.Serve(listener)

			return process.ShutdownFunc(func(ctx context.Context) error {
				grpcServer.Stop()
				// listener is closed for us
				return nil
			}), nil
		},
	}
}