forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-26 06:36:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

internal/cli: add block tracing (#397)

Browse source 
* bad block tracing

* Add debug env framework

* add implementation of custom block tracing

* add few comments

* use tar name fn

* wip: add unit tests for block tracing

* complete test for block tracer

* fix: close grpc server

* refactor cli tests

* fix: change condition for parsing args

* Fix port binding for test server

* add helper for creating and closing mock server for tests

* consume mock server in tests

* fixes due to geth merge

* fix: handle port selection for http server

* update help and markdown for debug command

* update docs

* update debug synopsis

* fix: use chunked encoder to handle large data over grpc

* fix prints

* lint

* rm unused function, rename fn to TraceBorBlock

Co-authored-by: Ferran Borreguero <ferranbt@protonmail.com>
This commit is contained in:
Manav Darji 2022-06-06 14:54:05 +05:30 committed by GitHub
parent 4ffdbfe60d
commit 0d2b1d0630
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 1570 additions and 653 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/cli/README.md
View file

@ -22,6 +22,10 @@
- [```debug```](./debug.md)
- [```debug block```](./debug_block.md)
- [```debug pprof```](./debug_pprof.md)
- [```fingerprint```](./fingerprint.md)
- [```peers```](./peers.md)

8
docs/cli/debug.md
View file

@ -2,13 +2,9 @@
The ```bor debug``` command takes a debug dump of the running client.
## Options
- [```bor debug pprof```](./debug_pprof.md): Dumps bor pprof traces.
- ```address```: Address of the grpc endpoint
- ```seconds```: seconds to trace
- ```output```: Output directory
- [```bor debug block <number>```](./debug_block.md): Dumps bor block traces.
## Examples

9
docs/cli/debug_block.md Normal file
View file

@ -0,0 +1,9 @@
# Debug trace
The ```bor debug block <number>``` command will create an archive containing traces of a bor block.
## Options
- ```address```: Address of the grpc endpoint
- ```output```: Output directory

11
docs/cli/debug_pprof.md Normal file
View file

@ -0,0 +1,11 @@
# Debug Pprof
The ```debug pprof <enode>``` command will create an archive containing bor pprof traces.
## Options
- ```address```: Address of the grpc endpoint
- ```seconds```: seconds to trace
- ```output```: Output directory

143
eth/tracers/api_bor.go Normal file
View file

@ -0,0 +1,143 @@
package tracers
import (
"context"
"fmt"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/eth/tracers/logger"
"github.com/ethereum/go-ethereum/internal/ethapi"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rpc"
)
type BlockTraceResult struct {
// Trace of each transaction executed
Transactions []*TxTraceResult `json:"transactions,omitempty"`
// Block that we are executing on the trace
Block interface{} `json:"block"`
}
type TxTraceResult struct {
// Trace results produced by the tracer
Result interface{} `json:"result,omitempty"`
// Trace failure produced by the tracer
Error string `json:"error,omitempty"`
// IntermediateHash of the execution if succeeds
IntermediateHash common.Hash `json:"intermediatehash"`
}
func (api *API) traceBorBlock(ctx context.Context, block *types.Block, config *TraceConfig) (*BlockTraceResult, error) {
if block.NumberU64() == 0 {
return nil, fmt.Errorf("genesis is not traceable")
}
res := &BlockTraceResult{
Block: block,
}
// block object cannot be converted to JSON since much of the fields are non-public
blockFields, err := ethapi.RPCMarshalBlock(block, true, true, api.backend.ChainConfig())
if err != nil {
return nil, err
}
res.Block = blockFields
parent, err := api.blockByNumberAndHash(ctx, rpc.BlockNumber(block.NumberU64()-1), block.ParentHash())
if err != nil {
return nil, err
}
reexec := defaultTraceReexec
if config != nil && config.Reexec != nil {
reexec = *config.Reexec
}
// TODO: discuss consequences of setting preferDisk false.
statedb, err := api.backend.StateAtBlock(ctx, parent, reexec, nil, true, false)
if err != nil {
return nil, err
}
// Execute all the transaction contained within the block concurrently
var (
signer = types.MakeSigner(api.backend.ChainConfig(), block.Number())
txs = block.Transactions()
deleteEmptyObjects = api.backend.ChainConfig().IsEIP158(block.Number())
)
blockCtx := core.NewEVMBlockContext(block.Header(), api.chainContext(ctx), nil)
traceTxn := func(indx int, tx *types.Transaction) *TxTraceResult {
message, _ := tx.AsMessage(signer, block.BaseFee())
txContext := core.NewEVMTxContext(message)
tracer := logger.NewStructLogger(config.Config)
// Run the transaction with tracing enabled.
vmenv := vm.NewEVM(blockCtx, txContext, statedb, api.backend.ChainConfig(), vm.Config{Debug: true, Tracer: tracer, NoBaseFee: true})
// Call Prepare to clear out the statedb access list
// Not sure if we need to do this
statedb.Prepare(tx.Hash(), indx)
execRes, err := core.ApplyMessage(vmenv, message, new(core.GasPool).AddGas(message.Gas()))
if err != nil {
return &TxTraceResult{
Error: err.Error(),
}
}
returnVal := fmt.Sprintf("%x", execRes.Return())
if len(execRes.Revert()) > 0 {
returnVal = fmt.Sprintf("%x", execRes.Revert())
}
result := &ethapi.ExecutionResult{
Gas: execRes.UsedGas,
Failed: execRes.Failed(),
ReturnValue: returnVal,
StructLogs: ethapi.FormatLogs(tracer.StructLogs()),
}
res := &TxTraceResult{
Result: result,
IntermediateHash: statedb.IntermediateRoot(deleteEmptyObjects),
}
return res
}
for indx, tx := range txs {
res.Transactions = append(res.Transactions, traceTxn(indx, tx))
}
return res, nil
}
type TraceBlockRequest struct {
Number int64
Hash string
IsBadBlock bool
Config *TraceConfig
}
// If you use context as first parameter this function gets exposed automaticall on rpc endpoint
func (api *API) TraceBorBlock(req *TraceBlockRequest) (*BlockTraceResult, error) {
ctx := context.Background()
var blockNumber rpc.BlockNumber
if req.Number == -1 {
blockNumber = rpc.LatestBlockNumber
} else {
blockNumber = rpc.BlockNumber(req.Number)
}
log.Debug("Tracing Bor Block", "block number", blockNumber)
block, err := api.blockByNumber(ctx, blockNumber)
if err != nil {
return nil, err
}
return api.traceBorBlock(ctx, block, req.Config)
}

10
internal/cli/command.go
View file

@ -84,6 +84,16 @@ func Commands() map[string]MarkDownCommandFactory {
},
"debug": func() (MarkDownCommand, error) {
return &DebugCommand{
UI: ui,
}, nil
},
"debug pprof": func() (MarkDownCommand, error) {
return &DebugPprofCommand{
Meta2: meta2,
}, nil
},
"debug block": func() (MarkDownCommand, error) {
return &DebugBlockCommand{
Meta2: meta2,
}, nil
},

250
internal/cli/debug.go
View file

@ -1,11 +1,8 @@
package cli
// Based on https://github.com/hashicorp/nomad/blob/main/command/operator_debug.go
import (
"archive/tar"
"compress/gzip"
"context"
"fmt"
"io"
"io/ioutil"
@ -16,20 +13,20 @@ import (
"syscall"
"time"
"github.com/ethereum/go-ethereum/internal/cli/flagset"
"github.com/mitchellh/cli"
"github.com/ethereum/go-ethereum/internal/cli/server/proto"
"github.com/golang/protobuf/jsonpb" // nolint:staticcheck
gproto "github.com/golang/protobuf/proto" // nolint:staticcheck
"github.com/golang/protobuf/ptypes/empty"
grpc_net_conn "github.com/mitchellh/go-grpc-net-conn"
"google.golang.org/grpc"
"google.golang.org/protobuf/runtime/protoiface"
)
// DebugCommand is the command to group the peers commands
type DebugCommand struct {
*Meta2
seconds uint64
output string
UI cli.Ui
}
// MarkDown implements cli.MarkDown interface
@ -53,7 +50,8 @@ func (d *DebugCommand) MarkDown() string {
items := []string{
"# Debug",
"The ```bor debug``` command takes a debug dump of the running client.",
d.Flags().MarkDown(),
"- [```bor debug pprof```](./debug_pprof.md): Dumps bor pprof traces.",
"- [```bor debug block <number>```](./debug_block.md): Dumps bor block traces.",
}
items = append(items, examples...)
@ -61,185 +59,141 @@ func (d *DebugCommand) MarkDown() string {
}
// Help implements the cli.Command interface
func (d *DebugCommand) Help() string {
return `Usage: bor debug
func (c *DebugCommand) Help() string {
return `Usage: bor debug <subcommand>
Build an archive containing Bor pprof traces
This command takes a debug dump of the running client.
Get the pprof traces:
` + d.Flags().Help()
}
$ bor debug pprof <enode>
func (d *DebugCommand) Flags() *flagset.Flagset {
flags := d.NewFlagSet("debug")
Get the block traces:
flags.Uint64Flag(&flagset.Uint64Flag{
Name: "seconds",
Usage: "seconds to trace",
Value: &d.seconds,
Default: 2,
})
flags.StringFlag(&flagset.StringFlag{
Name: "output",
Value: &d.output,
Usage: "Output directory",
})
return flags
$ bor debug block <number>`
}
// Synopsis implements the cli.Command interface
func (d *DebugCommand) Synopsis() string {
return "Build an archive containing Bor pprof traces"
func (c *DebugCommand) Synopsis() string {
return "Get traces of the running client"
}
// Run implements the cli.Command interface
func (d *DebugCommand) Run(args []string) int {
flags := d.Flags()
if err := flags.Parse(args); err != nil {
d.UI.Error(err.Error())
return 1
}
func (c *DebugCommand) Run(args []string) int {
return cli.RunResultHelp
}
clt, err := d.BorConn()
if err != nil {
d.UI.Error(err.Error())
return 1
}
type debugEnv struct {
output string
prefix string
stamped := "bor-debug-" + time.Now().UTC().Format("2006-01-02-150405Z")
name string
dst string
}
func (d *debugEnv) init() error {
d.name = d.prefix + time.Now().UTC().Format("2006-01-02-150405Z")
var err error
// Create the output directory
var tmp string
if d.output != "" {
// User specified output directory
tmp = filepath.Join(d.output, stamped)
tmp = filepath.Join(d.output, d.name)
_, err := os.Stat(tmp)
if !os.IsNotExist(err) {
d.UI.Error("Output directory already exists")
return 1
return fmt.Errorf("output directory already exists")
}
} else {
// Generate temp directory
tmp, err = ioutil.TempDir(os.TempDir(), stamped)
tmp, err = ioutil.TempDir(os.TempDir(), d.name)
if err != nil {
d.UI.Error(fmt.Sprintf("Error creating tmp directory: %s", err.Error()))
return 1
return fmt.Errorf("error creating tmp directory: %s", err.Error())
}
defer os.RemoveAll(tmp)
}
d.UI.Output("Starting debugger...")
d.UI.Output("")
// ensure destine folder exists
if err := os.MkdirAll(tmp, os.ModePerm); err != nil {
d.UI.Error(fmt.Sprintf("failed to create parent directory: %v", err))
return 1
return fmt.Errorf("failed to create parent directory: %v", err)
}
pprofProfile := func(ctx context.Context, profile string, filename string) error {
req := &proto.PprofRequest{
Seconds: int64(d.seconds),
}
d.dst = tmp
switch profile {
case "cpu":
req.Type = proto.PprofRequest_CPU
case "trace":
req.Type = proto.PprofRequest_TRACE
default:
req.Type = proto.PprofRequest_LOOKUP
req.Profile = profile
}
return nil
}
stream, err := clt.Pprof(ctx, req)
if err != nil {
return err
}
// wait for open request
msg, err := stream.Recv()
if err != nil {
return err
}
if _, ok := msg.Event.(*proto.PprofResponse_Open_); !ok {
return fmt.Errorf("expected open message")
}
// create the stream
conn := &grpc_net_conn.Conn{
Stream: stream,
Response: &proto.PprofResponse_Input{},
Decode: grpc_net_conn.SimpleDecoder(func(msg gproto.Message) *[]byte {
return &msg.(*proto.PprofResponse_Input).Data
}),
}
file, err := os.OpenFile(filepath.Join(tmp, filename+".prof"), os.O_RDWR|os.O_CREATE, 0644)
if err != nil {
return err
}
defer file.Close()
if _, err := io.Copy(file, conn); err != nil {
return err
}
func (d *debugEnv) tarName() string {
return d.name + ".tar.gz"
}
func (d *debugEnv) finish() error {
// Exit before archive if output directory was specified
if d.output != "" {
return nil
}
ctx, cancelFn := context.WithCancel(context.Background())
trapSignal(cancelFn)
profiles := map[string]string{
"heap": "heap",
"cpu": "cpu",
"trace": "trace",
}
for profile, filename := range profiles {
if err := pprofProfile(ctx, profile, filename); err != nil {
d.UI.Error(fmt.Sprintf("Error creating profile '%s': %v", profile, err))
return 1
}
}
// append the status
{
statusResp, err := clt.Status(ctx, &empty.Empty{})
if err != nil {
d.UI.Output(fmt.Sprintf("Failed to get status: %v", err))
return 1
}
m := jsonpb.Marshaler{}
data, err := m.MarshalToString(statusResp)
if err != nil {
d.UI.Output(err.Error())
return 1
}
if err := ioutil.WriteFile(filepath.Join(tmp, "status.json"), []byte(data), 0600); err != nil {
d.UI.Output(fmt.Sprintf("Failed to write status: %v", err))
return 1
}
}
// Exit before archive if output directory was specified
if d.output != "" {
d.UI.Output(fmt.Sprintf("Created debug directory: %s", tmp))
return 0
}
// Create archive tarball
archiveFile := stamped + ".tar.gz"
if err = tarCZF(archiveFile, tmp, stamped); err != nil {
d.UI.Error(fmt.Sprintf("Error creating archive: %s", err.Error()))
return 1
archiveFile := d.tarName()
if err := tarCZF(archiveFile, d.dst, d.name); err != nil {
return fmt.Errorf("error creating archive: %s", err.Error())
}
d.UI.Output(fmt.Sprintf("Created debug archive: %s", archiveFile))
return nil
}
return 0
type debugStream interface {
Recv() (*proto.DebugFileResponse, error)
grpc.ClientStream
}
func (d *debugEnv) writeFromStream(name string, stream debugStream) error {
// wait for open request
msg, err := stream.Recv()
if err != nil {
return err
}
if _, ok := msg.Event.(*proto.DebugFileResponse_Open_); !ok {
return fmt.Errorf("expected open message")
}
// create the stream
conn := &grpc_net_conn.Conn{
Stream: stream,
Response: &proto.DebugFileResponse_Input{},
Decode: grpc_net_conn.SimpleDecoder(func(msg gproto.Message) *[]byte {
return &msg.(*proto.DebugFileResponse_Input).Data
}),
}
file, err := os.OpenFile(filepath.Join(d.dst, name), os.O_RDWR|os.O_CREATE, 0644)
if err != nil {
return err
}
defer file.Close()
if _, err := io.Copy(file, conn); err != nil {
return err
}
return nil
}
func (d *debugEnv) writeJSON(name string, msg protoiface.MessageV1) error {
m := jsonpb.Marshaler{}
data, err := m.MarshalToString(msg)
if err != nil {
return err
}
if err := ioutil.WriteFile(filepath.Join(d.dst, name), []byte(data), 0600); err != nil {
return fmt.Errorf("failed to write status: %v", err)
}
return nil
}
func trapSignal(cancel func()) {

126
internal/cli/debug_block.go Normal file
View file

@ -0,0 +1,126 @@
package cli
import (
"context"
"fmt"
"strconv"
"strings"
"github.com/ethereum/go-ethereum/internal/cli/flagset"
"github.com/ethereum/go-ethereum/internal/cli/server/proto"
)
// DebugBlockCommand is the command to group the peers commands
type DebugBlockCommand struct {
*Meta2
output string
}
func (p *DebugBlockCommand) MarkDown() string {
items := []string{
"# Debug trace",
"The ```bor debug block <number>``` command will create an archive containing traces of a bor block.",
p.Flags().MarkDown(),
}
return strings.Join(items, "\n\n")
}
// Help implements the cli.Command interface
func (c *DebugBlockCommand) Help() string {
return `Usage: bor debug block <number>
This command is used get traces of a bor block`
}
func (c *DebugBlockCommand) Flags() *flagset.Flagset {
flags := c.NewFlagSet("trace")
flags.StringFlag(&flagset.StringFlag{
Name: "output",
Value: &c.output,
Usage: "Output directory",
})
return flags
}
// Synopsis implements the cli.Command interface
func (c *DebugBlockCommand) Synopsis() string {
return "Get trace of a bor block"
}
// Run implements the cli.Command interface
func (c *DebugBlockCommand) Run(args []string) int {
flags := c.Flags()
var number *int64 = nil
// parse the block number (if available)
if len(args)%2 != 0 {
num, err := strconv.ParseInt(args[0], 10, 64)
if err == nil {
number = &num
}
args = args[1:]
}
// parse output directory
if err := flags.Parse(args); err != nil {
c.UI.Error(err.Error())
return 1
}
borClt, err := c.BorConn()
if err != nil {
c.UI.Error(err.Error())
return 1
}
dEnv := &debugEnv{
output: c.output,
prefix: "bor-block-trace-",
}
if err := dEnv.init(); err != nil {
c.UI.Error(err.Error())
return 1
}
c.UI.Output("Starting block tracer...")
c.UI.Output("")
// create a debug block request
var debugRequest *proto.DebugBlockRequest = &proto.DebugBlockRequest{}
if number != nil {
debugRequest.Number = *number
} else {
debugRequest.Number = -1
}
// send the request
// receives a grpc stream of debug block response
stream, err := borClt.DebugBlock(context.Background(), debugRequest)
if err != nil {
c.UI.Error(err.Error())
return 1
}
if err := dEnv.writeFromStream("block.json", stream); err != nil {
c.UI.Error(err.Error())
return 1
}
if err := dEnv.finish(); err != nil {
c.UI.Error(err.Error())
return 1
}
if c.output != "" {
c.UI.Output(fmt.Sprintf("Created debug directory: %s", dEnv.dst))
} else {
c.UI.Output(fmt.Sprintf("Created block trace archive: %s", dEnv.tarName()))
}
return 0
}

159
internal/cli/debug_pprof.go Normal file
View file

@ -0,0 +1,159 @@
package cli
// Based on https://github.com/hashicorp/nomad/blob/main/command/operator_debug.go
import (
"context"
"fmt"
"strings"
"github.com/golang/protobuf/ptypes/empty"
"github.com/ethereum/go-ethereum/internal/cli/flagset"
"github.com/ethereum/go-ethereum/internal/cli/server/proto"
)
type DebugPprofCommand struct {
*Meta2
seconds uint64
output string
}
func (p *DebugPprofCommand) MarkDown() string {
items := []string{
"# Debug Pprof",
"The ```debug pprof <enode>``` command will create an archive containing bor pprof traces.",
p.Flags().MarkDown(),
}
return strings.Join(items, "\n\n")
}
// Help implements the cli.Command interface
func (d *DebugPprofCommand) Help() string {
return `Usage: bor debug
Build an archive containing Bor pprof traces
` + d.Flags().Help()
}
func (d *DebugPprofCommand) Flags() *flagset.Flagset {
flags := d.NewFlagSet("debug")
flags.Uint64Flag(&flagset.Uint64Flag{
Name: "seconds",
Usage: "seconds to trace",
Value: &d.seconds,
Default: 2,
})
flags.StringFlag(&flagset.StringFlag{
Name: "output",
Value: &d.output,
Usage: "Output directory",
})
return flags
}
// Synopsis implements the cli.Command interface
func (d *DebugPprofCommand) Synopsis() string {
return "Build an archive containing Bor pprof traces"
}
// Run implements the cli.Command interface
func (d *DebugPprofCommand) Run(args []string) int {
flags := d.Flags()
if err := flags.Parse(args); err != nil {
d.UI.Error(err.Error())
return 1
}
clt, err := d.BorConn()
if err != nil {
d.UI.Error(err.Error())
return 1
}
dEnv := &debugEnv{
output: d.output,
prefix: "bor-debug-",
}
if err := dEnv.init(); err != nil {
d.UI.Error(err.Error())
return 1
}
d.UI.Output("Starting debugger...")
d.UI.Output("")
pprofProfile := func(ctx context.Context, profile string, filename string) error {
req := &proto.DebugPprofRequest{
Seconds: int64(d.seconds),
}
switch profile {
case "cpu":
req.Type = proto.DebugPprofRequest_CPU
case "trace":
req.Type = proto.DebugPprofRequest_TRACE
default:
req.Type = proto.DebugPprofRequest_LOOKUP
req.Profile = profile
}
stream, err := clt.DebugPprof(ctx, req)
if err != nil {
return err
}
if err := dEnv.writeFromStream(filename+".prof", stream); err != nil {
return err
}
return nil
}
ctx, cancelFn := context.WithCancel(context.Background())
trapSignal(cancelFn)
profiles := map[string]string{
"heap": "heap",
"cpu": "cpu",
"trace": "trace",
}
for profile, filename := range profiles {
if err := pprofProfile(ctx, profile, filename); err != nil {
d.UI.Error(fmt.Sprintf("Error creating profile '%s': %v", profile, err))
return 1
}
}
// append the status
{
statusResp, err := clt.Status(ctx, &empty.Empty{})
if err != nil {
d.UI.Output(fmt.Sprintf("Failed to get status: %v", err))
return 1
}
if err := dEnv.writeJSON("status.json", statusResp); err != nil {
d.UI.Error(err.Error())
return 1
}
}
if err := dEnv.finish(); err != nil {
d.UI.Error(err.Error())
return 1
}
if d.output != "" {
d.UI.Output(fmt.Sprintf("Created debug directory: %s", dEnv.dst))
} else {
d.UI.Output(fmt.Sprintf("Created debug archive: %s", dEnv.tarName()))
}
return 0
}

112
internal/cli/debug_test.go Normal file
View file

@ -0,0 +1,112 @@
package cli
import (
"os"
"path"
"strconv"
"testing"
"time"
"github.com/mitchellh/cli"
"github.com/stretchr/testify/assert"
"github.com/ethereum/go-ethereum/internal/cli/server"
)
var currentDir string = ""
func TestCommand_DebugBlock(t *testing.T) {
t.Parallel()
// Start a blockchain in developer mode and get trace of block
config := server.DefaultConfig()
// enable developer mode
config.Developer.Enabled = true
config.Developer.Period = 2 // block time
// enable archive mode for getting traces of ancient blocks
config.GcMode = "archive"
// start the mock server
srv, err := server.CreateMockServer(config)
assert.NoError(t, err)
defer server.CloseMockServer(srv)
// get the grpc port
port := srv.GetGrpcAddr()
// wait for 4 seconds to mine a 2 blocks
time.Sleep(2 * time.Duration(config.Developer.Period) * time.Second)
// add prefix for debug trace
prefix := "bor-block-trace-"
// output dir
output := "debug_block_test"
// set current directory
currentDir, _ = os.Getwd()
// trace 1st block
start := time.Now()
dst1 := path.Join(output, prefix+time.Now().UTC().Format("2006-01-02-150405Z"), "block.json")
res := traceBlock(port, 1, output)
assert.Equal(t, 0, res)
t.Logf("Completed trace of block %d in %d ms at %s", 1, time.Since(start).Milliseconds(), dst1)
// adding this to avoid debug directory name conflicts
time.Sleep(time.Second)
// trace last/recent block
start = time.Now()
latestBlock := srv.GetLatestBlockNumber().Int64()
dst2 := path.Join(output, prefix+time.Now().UTC().Format("2006-01-02-150405Z"), "block.json")
res = traceBlock(port, latestBlock, output)
assert.Equal(t, 0, res)
t.Logf("Completed trace of block %d in %d ms at %s", latestBlock, time.Since(start).Milliseconds(), dst2)
// verify if the trace files are created
done := verify(dst1)
assert.Equal(t, true, done)
done = verify(dst2)
assert.Equal(t, true, done)
// delete the traces
deleteTraces(output)
}
// traceBlock calls the cli command to trace a block
func traceBlock(port string, number int64, output string) int {
ui := cli.NewMockUi()
command := &DebugBlockCommand{
Meta2: &Meta2{
UI: ui,
addr: "127.0.0.1:" + port,
},
}
// run trace (by explicitly passing the output directory and grpc address)
return command.Run([]string{strconv.FormatInt(number, 10), "--output", output, "--address", command.Meta2.addr})
}
// verify checks if the trace file is created at the destination
// directory or not
func verify(dst string) bool {
dst = path.Join(currentDir, dst)
if file, err := os.Stat(dst); err == nil {
// check if the file has content
if file.Size() > 0 {
return true
}
}
return false
}
// deleteTraces removes the traces created during the test
func deleteTraces(dst string) {
dst = path.Join(currentDir, dst)
os.RemoveAll(dst)
}

72
internal/cli/server/helper.go Normal file
View file

@ -0,0 +1,72 @@
package server
import (
"fmt"
"io/ioutil"
"math/rand"
"net"
"os"
"sync/atomic"
"time"
)
var maxPortCheck int32 = 100
// findAvailablePort returns the next available port starting from `from`
func findAvailablePort(from int32, count int32) (int32, error) {
if count == maxPortCheck {
return 0, fmt.Errorf("no available port found")
}
port := atomic.AddInt32(&from, 1)
addr := fmt.Sprintf("localhost:%d", port)
lis, err := net.Listen("tcp", addr)
count++
if err == nil {
lis.Close()
return port, nil
} else {
return findAvailablePort(from, count)
}
}
func CreateMockServer(config *Config) (*Server, error) {
if config == nil {
config = DefaultConfig()
}
// find available port for grpc server
rand.Seed(time.Now().UnixNano())
var from int32 = 60000 // the min port to start checking from
var to int32 = 61000 // the max port to start checking from
port, err := findAvailablePort(rand.Int31n(to-from+1)+from, 0)
if err != nil {
return nil, err
}
// grpc port
config.GRPC.Addr = fmt.Sprintf(":%d", port)
// datadir
datadir, _ := ioutil.TempDir("/tmp", "bor-cli-test")
config.DataDir = datadir
// find available port for http server
from = 8545
to = 9545
port, err = findAvailablePort(rand.Int31n(to-from+1)+from, 0)
if err != nil {
return nil, err
}
config.JsonRPC.Http.Port = uint64(port)
// start the server
return NewServer(config)
}
func CloseMockServer(server *Server) {
// remove the contents of temp data dir
os.RemoveAll(server.config.DataDir)
// close the server
server.Stop()
}

956
internal/cli/server/proto/server.pb.go
View file

File diff suppressed because it is too large Load diff

25
internal/cli/server/proto/server.proto
View file

@ -7,8 +7,6 @@ import "google/protobuf/empty.proto";
option go_package = "/internal/cli/server/proto";
service Bor {
rpc Pprof(PprofRequest) returns (stream PprofResponse);
rpc PeersAdd(PeersAddRequest) returns (PeersAddResponse);
rpc PeersRemove(PeersRemoveRequest) returns (PeersRemoveResponse);
@ -20,8 +18,20 @@ service Bor {
rpc ChainSetHead(ChainSetHeadRequest) returns (ChainSetHeadResponse);
rpc Status(google.protobuf.Empty) returns (StatusResponse);
rpc ChainWatch(ChainWatchRequest) returns (stream ChainWatchResponse);
rpc DebugPprof(DebugPprofRequest) returns (stream DebugFileResponse);
rpc DebugBlock(DebugBlockRequest) returns (stream DebugFileResponse);
}
message TraceRequest {
int64 number = 1;
}
message TraceResponse {
}
message ChainWatchRequest {
@ -113,7 +123,7 @@ message Header {
uint64 number = 2;
}
message PprofRequest {
message DebugPprofRequest {
Type type = 1;
string profile = 2;
@ -127,7 +137,11 @@ message PprofRequest {
}
}
message PprofResponse {
message DebugBlockRequest {
int64 number = 1;
}
message DebugFileResponse {
oneof event {
Open open = 1;
Input input = 2;
@ -136,7 +150,6 @@ message PprofResponse {
message Open {
map<string, string> headers = 1;
int64 size = 2;
}
message Input {

193
internal/cli/server/proto/server_grpc.pb.go
View file

@ -1,4 +1,8 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.2.0
// - protoc v3.19.3
// source: internal/cli/server/proto/server.proto
package proto
@ -19,7 +23,6 @@ const _ = grpc.SupportPackageIsVersion7
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
type BorClient interface {
Pprof(ctx context.Context, in *PprofRequest, opts ...grpc.CallOption) (Bor_PprofClient, error)
PeersAdd(ctx context.Context, in *PeersAddRequest, opts ...grpc.CallOption) (*PeersAddResponse, error)
PeersRemove(ctx context.Context, in *PeersRemoveRequest, opts ...grpc.CallOption) (*PeersRemoveResponse, error)
PeersList(ctx context.Context, in *PeersListRequest, opts ...grpc.CallOption) (*PeersListResponse, error)
@ -27,6 +30,8 @@ type BorClient interface {
ChainSetHead(ctx context.Context, in *ChainSetHeadRequest, opts ...grpc.CallOption) (*ChainSetHeadResponse, error)
Status(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*StatusResponse, error)
ChainWatch(ctx context.Context, in *ChainWatchRequest, opts ...grpc.CallOption) (Bor_ChainWatchClient, error)
DebugPprof(ctx context.Context, in *DebugPprofRequest, opts ...grpc.CallOption) (Bor_DebugPprofClient, error)
DebugBlock(ctx context.Context, in *DebugBlockRequest, opts ...grpc.CallOption) (Bor_DebugBlockClient, error)
}
type borClient struct {
@ -37,38 +42,6 @@ func NewBorClient(cc grpc.ClientConnInterface) BorClient {
return &borClient{cc}
}
func (c *borClient) Pprof(ctx context.Context, in *PprofRequest, opts ...grpc.CallOption) (Bor_PprofClient, error) {
stream, err := c.cc.NewStream(ctx, &Bor_ServiceDesc.Streams[0], "/proto.Bor/Pprof", opts...)
if err != nil {
return nil, err
}
x := &borPprofClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
type Bor_PprofClient interface {
Recv() (*PprofResponse, error)
grpc.ClientStream
}
type borPprofClient struct {
grpc.ClientStream
}
func (x *borPprofClient) Recv() (*PprofResponse, error) {
m := new(PprofResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func (c *borClient) PeersAdd(ctx context.Context, in *PeersAddRequest, opts ...grpc.CallOption) (*PeersAddResponse, error) {
out := new(PeersAddResponse)
err := c.cc.Invoke(ctx, "/proto.Bor/PeersAdd", in, out, opts...)
@ -124,7 +97,7 @@ func (c *borClient) Status(ctx context.Context, in *emptypb.Empty, opts ...grpc.
}
func (c *borClient) ChainWatch(ctx context.Context, in *ChainWatchRequest, opts ...grpc.CallOption) (Bor_ChainWatchClient, error) {
stream, err := c.cc.NewStream(ctx, &Bor_ServiceDesc.Streams[1], "/proto.Bor/ChainWatch", opts...)
stream, err := c.cc.NewStream(ctx, &Bor_ServiceDesc.Streams[0], "/proto.Bor/ChainWatch", opts...)
if err != nil {
return nil, err
}
@ -155,11 +128,74 @@ func (x *borChainWatchClient) Recv() (*ChainWatchResponse, error) {
return m, nil
}
func (c *borClient) DebugPprof(ctx context.Context, in *DebugPprofRequest, opts ...grpc.CallOption) (Bor_DebugPprofClient, error) {
stream, err := c.cc.NewStream(ctx, &Bor_ServiceDesc.Streams[1], "/proto.Bor/DebugPprof", opts...)
if err != nil {
return nil, err
}
x := &borDebugPprofClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
type Bor_DebugPprofClient interface {
Recv() (*DebugFileResponse, error)
grpc.ClientStream
}
type borDebugPprofClient struct {
grpc.ClientStream
}
func (x *borDebugPprofClient) Recv() (*DebugFileResponse, error) {
m := new(DebugFileResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func (c *borClient) DebugBlock(ctx context.Context, in *DebugBlockRequest, opts ...grpc.CallOption) (Bor_DebugBlockClient, error) {
stream, err := c.cc.NewStream(ctx, &Bor_ServiceDesc.Streams[2], "/proto.Bor/DebugBlock", opts...)
if err != nil {
return nil, err
}
x := &borDebugBlockClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
type Bor_DebugBlockClient interface {
Recv() (*DebugFileResponse, error)
grpc.ClientStream
}
type borDebugBlockClient struct {
grpc.ClientStream
}
func (x *borDebugBlockClient) Recv() (*DebugFileResponse, error) {
m := new(DebugFileResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
// BorServer is the server API for Bor service.
// All implementations must embed UnimplementedBorServer
// for forward compatibility
type BorServer interface {
Pprof(*PprofRequest, Bor_PprofServer) error
PeersAdd(context.Context, *PeersAddRequest) (*PeersAddResponse, error)
PeersRemove(context.Context, *PeersRemoveRequest) (*PeersRemoveResponse, error)
PeersList(context.Context, *PeersListRequest) (*PeersListResponse, error)
@ -167,6 +203,8 @@ type BorServer interface {
ChainSetHead(context.Context, *ChainSetHeadRequest) (*ChainSetHeadResponse, error)
Status(context.Context, *emptypb.Empty) (*StatusResponse, error)
ChainWatch(*ChainWatchRequest, Bor_ChainWatchServer) error
DebugPprof(*DebugPprofRequest, Bor_DebugPprofServer) error
DebugBlock(*DebugBlockRequest, Bor_DebugBlockServer) error
mustEmbedUnimplementedBorServer()
}
@ -174,9 +212,6 @@ type BorServer interface {
type UnimplementedBorServer struct {
}
func (UnimplementedBorServer) Pprof(*PprofRequest, Bor_PprofServer) error {
return status.Errorf(codes.Unimplemented, "method Pprof not implemented")
}
func (UnimplementedBorServer) PeersAdd(context.Context, *PeersAddRequest) (*PeersAddResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method PeersAdd not implemented")
}
@ -198,6 +233,12 @@ func (UnimplementedBorServer) Status(context.Context, *emptypb.Empty) (*StatusRe
func (UnimplementedBorServer) ChainWatch(*ChainWatchRequest, Bor_ChainWatchServer) error {
return status.Errorf(codes.Unimplemented, "method ChainWatch not implemented")
}
func (UnimplementedBorServer) DebugPprof(*DebugPprofRequest, Bor_DebugPprofServer) error {
return status.Errorf(codes.Unimplemented, "method DebugPprof not implemented")
}
func (UnimplementedBorServer) DebugBlock(*DebugBlockRequest, Bor_DebugBlockServer) error {
return status.Errorf(codes.Unimplemented, "method DebugBlock not implemented")
}
func (UnimplementedBorServer) mustEmbedUnimplementedBorServer() {}
// UnsafeBorServer may be embedded to opt out of forward compatibility for this service.
@ -211,27 +252,6 @@ func RegisterBorServer(s grpc.ServiceRegistrar, srv BorServer) {
s.RegisterService(&Bor_ServiceDesc, srv)
}
func _Bor_Pprof_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(PprofRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(BorServer).Pprof(m, &borPprofServer{stream})
}
type Bor_PprofServer interface {
Send(*PprofResponse) error
grpc.ServerStream
}
type borPprofServer struct {
grpc.ServerStream
}
func (x *borPprofServer) Send(m *PprofResponse) error {
return x.ServerStream.SendMsg(m)
}
func _Bor_PeersAdd_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(PeersAddRequest)
if err := dec(in); err != nil {
@ -361,6 +381,48 @@ func (x *borChainWatchServer) Send(m *ChainWatchResponse) error {
return x.ServerStream.SendMsg(m)
}
func _Bor_DebugPprof_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(DebugPprofRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(BorServer).DebugPprof(m, &borDebugPprofServer{stream})
}
type Bor_DebugPprofServer interface {
Send(*DebugFileResponse) error
grpc.ServerStream
}
type borDebugPprofServer struct {
grpc.ServerStream
}
func (x *borDebugPprofServer) Send(m *DebugFileResponse) error {
return x.ServerStream.SendMsg(m)
}
func _Bor_DebugBlock_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(DebugBlockRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(BorServer).DebugBlock(m, &borDebugBlockServer{stream})
}
type Bor_DebugBlockServer interface {
Send(*DebugFileResponse) error
grpc.ServerStream
}
type borDebugBlockServer struct {
grpc.ServerStream
}
func (x *borDebugBlockServer) Send(m *DebugFileResponse) error {
return x.ServerStream.SendMsg(m)
}
// Bor_ServiceDesc is the grpc.ServiceDesc for Bor service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@ -395,13 +457,18 @@ var Bor_ServiceDesc = grpc.ServiceDesc{
},
Streams: []grpc.StreamDesc{
{
StreamName: "Pprof",
Handler: _Bor_Pprof_Handler,
StreamName: "ChainWatch",
Handler: _Bor_ChainWatch_Handler,
ServerStreams: true,
},
{
StreamName: "ChainWatch",
Handler: _Bor_ChainWatch_Handler,
StreamName: "DebugPprof",
Handler: _Bor_DebugPprof_Handler,
ServerStreams: true,
},
{
StreamName: "DebugBlock",
Handler: _Bor_DebugBlock_Handler,
ServerStreams: true,
},
},

14
internal/cli/server/server.go
View file

@ -4,6 +4,7 @@ import (
"context"
"fmt"
"io"
"math/big"
"net"
"net/http"
"os"
@ -43,6 +44,9 @@ type Server struct {
grpcServer *grpc.Server
tracer *sdktrace.TracerProvider
config *Config
// tracerAPI to trace block executions
tracerAPI *tracers.API
}
func NewServer(config *Config) (*Server, error) {
@ -162,6 +166,7 @@ func NewServer(config *Config) (*Server, error) {
// debug tracing is enabled by default
stack.RegisterAPIs(tracers.APIs(srv.backend.APIBackend))
srv.tracerAPI = tracers.NewAPI(srv.backend.APIBackend)
// graphql is started from another place
if config.JsonRPC.Graphql.Enabled {
@ -200,6 +205,7 @@ func NewServer(config *Config) (*Server, error) {
func (s *Server) Stop() {
s.node.Close()
s.grpcServer.Stop()
// shutdown the tracer
if s.tracer != nil {
@ -354,3 +360,11 @@ func setupLogger(logLevel string) {
}
log.Root().SetHandler(glogger)
}
func (s *Server) GetLatestBlockNumber() *big.Int {
return s.backend.BlockChain().CurrentBlock().Number()
}
func (s *Server) GetGrpcAddr() string {
return s.config.GRPC.Addr[1:]
}

35
internal/cli/server/server_test.go
View file

@ -1,14 +1,33 @@
package server
import (
"fmt"
"net"
"sync/atomic"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/stretchr/testify/assert"
)
func TestServer_DeveloperMode(t *testing.T) {
var initialPort uint64 = 61000
// nextPort gives the next available port starting from 60000
func nextPort() uint64 {
log.Info("Checking for new port", "current", initialPort)
port := atomic.AddUint64(&initialPort, 1)
addr := fmt.Sprintf("localhost:%d", port)
lis, err := net.Listen("tcp", addr)
if err == nil {
lis.Close()
return port
} else {
return nextPort()
}
}
func TestServer_DeveloperMode(t *testing.T) {
// get the default config
config := DefaultConfig()
@ -16,17 +35,16 @@ func TestServer_DeveloperMode(t *testing.T) {
config.Developer.Enabled = true
config.Developer.Period = 2 // block time
// start the server
server, err1 := NewServer(config)
if err1 != nil {
t.Fatalf("failed to start server: %v", err1)
}
// start the mock server
server, err := CreateMockServer(config)
assert.NoError(t, err)
defer CloseMockServer(server)
// record the initial block number
blockNumber := server.backend.BlockChain().CurrentBlock().Header().Number.Int64()
var i int64 = 0
for i = 0; i < 10; i++ {
for i = 0; i < 3; i++ {
// We expect the node to mine blocks every `config.Developer.Period` time period
time.Sleep(time.Duration(config.Developer.Period) * time.Second)
currBlock := server.backend.BlockChain().CurrentBlock().Header().Number.Int64()
@ -35,7 +53,4 @@ func TestServer_DeveloperMode(t *testing.T) {
break
}
}
// stop the server
server.Stop()
}

96
internal/cli/server/service.go
View file

@ -2,6 +2,7 @@ package server
import (
"context"
"encoding/json"
"fmt"
"math/big"
"reflect"
@ -9,6 +10,8 @@ import (
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/tracers"
"github.com/ethereum/go-ethereum/eth/tracers/logger"
"github.com/ethereum/go-ethereum/internal/cli/server/pprof"
"github.com/ethereum/go-ethereum/internal/cli/server/proto"
"github.com/ethereum/go-ethereum/p2p"
@ -18,30 +21,14 @@ import (
grpc_net_conn "github.com/mitchellh/go-grpc-net-conn"
)
func (s *Server) Pprof(req *proto.PprofRequest, stream proto.Bor_PprofServer) error {
var payload []byte
var headers map[string]string
var err error
ctx := context.Background()
switch req.Type {
case proto.PprofRequest_CPU:
payload, headers, err = pprof.CPUProfile(ctx, int(req.Seconds))
case proto.PprofRequest_TRACE:
payload, headers, err = pprof.Trace(ctx, int(req.Seconds))
case proto.PprofRequest_LOOKUP:
payload, headers, err = pprof.Profile(req.Profile, 0, 0)
}
if err != nil {
return err
}
const chunkSize = 1024 * 1024 * 1024
func sendStreamDebugFile(stream proto.Bor_DebugPprofServer, headers map[string]string, data []byte) error {
// open the stream and send the headers
err = stream.Send(&proto.PprofResponse{
Event: &proto.PprofResponse_Open_{
Open: &proto.PprofResponse_Open{
err := stream.Send(&proto.DebugFileResponse{
Event: &proto.DebugFileResponse_Open_{
Open: &proto.DebugFileResponse_Open{
Headers: headers,
Size: int64(len(payload)),
},
},
})
@ -50,24 +37,51 @@ func (s *Server) Pprof(req *proto.PprofRequest, stream proto.Bor_PprofServer) er
}
// Wrap our conn around the response.
encoder := grpc_net_conn.SimpleEncoder(func(msg gproto.Message) *[]byte {
return &msg.(*proto.DebugFileResponse_Input).Data
})
conn := &grpc_net_conn.Conn{
Stream: stream,
Request: &proto.PprofResponse_Input{},
Encode: grpc_net_conn.SimpleEncoder(func(msg gproto.Message) *[]byte {
return &msg.(*proto.PprofResponse_Input).Data
}),
Request: &proto.DebugFileResponse_Input{},
Encode: grpc_net_conn.ChunkedEncoder(encoder, chunkSize),
}
if _, err := conn.Write(payload); err != nil {
if _, err := conn.Write(data); err != nil {
return err
}
// send the eof
err = stream.Send(&proto.PprofResponse{
Event: &proto.PprofResponse_Eof{},
err = stream.Send(&proto.DebugFileResponse{
Event: &proto.DebugFileResponse_Eof{},
})
if err != nil {
return err
}
return nil
}
func (s *Server) DebugPprof(req *proto.DebugPprofRequest, stream proto.Bor_DebugPprofServer) error {
var payload []byte
var headers map[string]string
var err error
ctx := context.Background()
switch req.Type {
case proto.DebugPprofRequest_CPU:
payload, headers, err = pprof.CPUProfile(ctx, int(req.Seconds))
case proto.DebugPprofRequest_TRACE:
payload, headers, err = pprof.Trace(ctx, int(req.Seconds))
case proto.DebugPprofRequest_LOOKUP:
payload, headers, err = pprof.Profile(req.Profile, 0, 0)
}
if err != nil {
return err
}
// send the file on a grpc stream
if err := sendStreamDebugFile(stream, headers, payload); err != nil {
return err
}
return nil
}
@ -169,6 +183,32 @@ func headerToProtoHeader(h *types.Header) *proto.Header {
}
}
func (s *Server) DebugBlock(req *proto.DebugBlockRequest, stream proto.Bor_DebugBlockServer) error {
traceReq := &tracers.TraceBlockRequest{
Number: req.Number,
Config: &tracers.TraceConfig{
Config: &logger.Config{
EnableMemory: true,
},
},
}
res, err := s.tracerAPI.TraceBorBlock(traceReq)
if err != nil {
return err
}
// this is memory heavy
data, err := json.Marshal(res)
if err != nil {
return err
}
if err := sendStreamDebugFile(stream, map[string]string{}, data); err != nil {
return err
}
return nil
}
var bigIntT = reflect.TypeOf(new(big.Int)).Kind()
// gatherForks gathers all the fork numbers via reflection