From 5dc09fdf132b2c4c9623cfddd93adce651a1a188 Mon Sep 17 00:00:00 2001 From: akhilkumarpilli Date: Thu, 3 Mar 2022 14:21:05 +0530 Subject: [PATCH 01/19] add benchmark test --- Makefile | 6 +++ tracelistener/benchmark_test.go | 75 +++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+) create mode 100644 tracelistener/benchmark_test.go diff --git a/Makefile b/Makefile index 2e46080a..5279e5e9 100644 --- a/Makefile +++ b/Makefile @@ -10,6 +10,7 @@ BUILD_VERSIONS_DEBUG = $(shell jq -r '.versions|map("build-\(.)-debug")[]' ${TAR STORE_MOD_VERSIONS = $(shell jq -r '.versions|map("store-mod-\(.)")[]' ${TARGETS}) TEST_VERSIONS = $(shell jq -r '.versions|map("test-\(.)")[]' ${TARGETS}) COVERAGE_VERSIONS = $(shell jq -r '.versions|map("coverage-\(.)")[]' ${TARGETS}) +BENCHMARK_VERSIONS = $(shell jq -r '.versions|map("benchmark-\(.)")[]' ${TARGETS}) BRANCH := $(shell git rev-parse --abbrev-ref HEAD) COMMIT := $(shell git log -1 --format='%H') @@ -56,6 +57,11 @@ $(TEST_VERSIONS): go test -v -failfast -race -count=1 \ -tags $(shell echo $@ | sed -e 's/test-/sdk_/g' -e 's/-/_/g'),muslc \ ./... + +$(BENCHMARK_VERSIONS): + go test -v -failfast -bench=. -run=^# -benchmem -count=1 \ + -tags $(shell echo $@ | sed -e 's/benchmark-/sdk_/g' -e 's/-/_/g'),muslc \ + ./... $(COVERAGE_VERSIONS): go test -v -failfast -coverprofile=coverage.out -covermode=atomic -count=1\ diff --git a/tracelistener/benchmark_test.go b/tracelistener/benchmark_test.go new file mode 100644 index 00000000..09ee8dd0 --- /dev/null +++ b/tracelistener/benchmark_test.go @@ -0,0 +1,75 @@ +package tracelistener_test + +import ( + "context" + "encoding/json" + "os" + "syscall" + "testing" + + "github.com/allinbits/tracelistener/tracelistener" + "github.com/containerd/fifo" + "go.uber.org/zap" +) + +func BenchmarkTraceListener(b *testing.B) { + f, err := os.CreateTemp("", "test_data") + if err != nil { + panic(err) + } + + err = f.Close() + if err != nil { + panic(err) + } + defer os.Remove(f.Name()) + + dataChan := make(chan tracelistener.TraceOperation) + errChan := make(chan error) + l, _ := zap.NewDevelopment() + tw := tracelistener.TraceWatcher{ + DataSourcePath: f.Name(), + WatchedOps: []tracelistener.Operation{ + tracelistener.WriteOp, + tracelistener.DeleteOp, + }, + DataChan: dataChan, + ErrorChan: errChan, + Logger: l.Sugar(), + } + + go func() { + tw.Watch() + }() + + for i := 0; i < b.N; i++ { + err := loadTest(i, f.Name()) + if err != nil { + panic(err) + } + } +} + +func loadTest(height int, file string) error { + ff, err := fifo.OpenFifo(context.Background(), file, syscall.O_WRONLY, 0655) + if err != nil { + return err + } + trace := tracelistener.TraceOperation{ + Operation: string(tracelistener.WriteOp), + Key: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, + Value: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, + BlockHeight: uint64(height), + TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + } + data, err := json.Marshal(trace) + if err != nil { + return err + } + ff.Write(data) + err = ff.Close() + if err != nil { + return err + } + return nil +} From e7fa4d36c35586985463eb74f392a1ab8a47d6fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Fri, 4 Mar 2022 13:59:43 +0100 Subject: [PATCH 02/19] chore(tracelistener): first perf impro on ingest --- tracelistener/benchmark_test.go | 80 ++++++++++++++-------- tracelistener/trace.go | 63 +++++++++++++----- tracelistener/tracelistener.go | 114 ++++++++++++++++++-------------- 3 files changed, 162 insertions(+), 95 deletions(-) diff --git a/tracelistener/benchmark_test.go b/tracelistener/benchmark_test.go index 09ee8dd0..a6379ed0 100644 --- a/tracelistener/benchmark_test.go +++ b/tracelistener/benchmark_test.go @@ -2,7 +2,8 @@ package tracelistener_test import ( "context" - "encoding/json" + "fmt" + "io" "os" "syscall" "testing" @@ -12,7 +13,8 @@ import ( "go.uber.org/zap" ) -func BenchmarkTraceListener(b *testing.B) { +func runBenchmark(b *testing.B, amount int, kind string) { + b.Helper() f, err := os.CreateTemp("", "test_data") if err != nil { panic(err) @@ -26,7 +28,7 @@ func BenchmarkTraceListener(b *testing.B) { dataChan := make(chan tracelistener.TraceOperation) errChan := make(chan error) - l, _ := zap.NewDevelopment() + l := zap.NewNop() tw := tracelistener.TraceWatcher{ DataSourcePath: f.Name(), WatchedOps: []tracelistener.Operation{ @@ -42,34 +44,60 @@ func BenchmarkTraceListener(b *testing.B) { tw.Watch() }() - for i := 0; i < b.N; i++ { - err := loadTest(i, f.Name()) + ff, err := fifo.OpenFifo(context.Background(), f.Name(), syscall.O_WRONLY, 0655) + if err != nil { + panic(err) + } + for i := 0; i < amount; i++ { + err := loadTest(b, i, ff, kind) if err != nil { panic(err) } } + + ff.Close() } -func loadTest(height int, file string) error { - ff, err := fifo.OpenFifo(context.Background(), file, syscall.O_WRONLY, 0655) - if err != nil { - return err - } - trace := tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, - Value: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, - BlockHeight: uint64(height), - TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", - } - data, err := json.Marshal(trace) - if err != nil { - return err - } - ff.Write(data) - err = ff.Close() - if err != nil { - return err - } +func BenchmarkTraceListenerKindWrite(b *testing.B) { + runBenchmark(b, b.N, "write") +} + +func BenchmarkTraceListener100KKindWrite(b *testing.B) { + runBenchmark(b, 100000, "write") +} + +func BenchmarkTraceListener1MKindWrite(b *testing.B) { + runBenchmark(b, 1000000, "write") +} + +func BenchmarkTraceListener1MKindIterRange(b *testing.B) { + runBenchmark(b, 1000000, "IterRange") +} + +func BenchmarkTraceListener10MKindWrite(b *testing.B) { + runBenchmark(b, 10000000, "write") +} + +func loadTest(b *testing.B, height int, ff io.Writer, kind string) error { + b.Helper() + + // trace := tracelistener.TraceOperation{ + // Operation: string(tracelistener.WriteOp), + // Key: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, + // Value: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, + // BlockHeight: uint64(height), + // TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + // } + // data, err := json.Marshal(trace) + // if err != nil { + // return err + // } + + // println(string(data)) + + s := `{"operation":"%s","key":"aGVsbG8K","value":"aGVsbG8K","block_height":158284,"tx_hash":"A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB","SuggestedProcessor":""}` + + fmt.Fprintf(ff, s+"\n", kind) + return nil } diff --git a/tracelistener/trace.go b/tracelistener/trace.go index ba42333b..a55c0c4b 100644 --- a/tracelistener/trace.go +++ b/tracelistener/trace.go @@ -1,8 +1,10 @@ package tracelistener import ( - "encoding/json" "fmt" + "sync" + + "github.com/goccy/go-json" ) const ( @@ -10,6 +12,12 @@ const ( metadataTxHash = "txHash" ) +var toiPool = sync.Pool{ + New: func() interface{} { + return &traceOperationInter{} + }, +} + type TraceOperation struct { Operation string `json:"operation"` Key []byte `json:"key"` @@ -22,39 +30,58 @@ type TraceOperation struct { SuggestedProcessor SDKModuleName } +func (to *TraceOperation) Copy() TraceOperation { + ret := TraceOperation{} + ret = *to + return ret +} + +func (to *TraceOperation) Reset() { + to.Operation = "" + to.Key = to.Key[:0] + to.Value = to.Value[:0] + to.BlockHeight = 0 + to.TxHash = "" + to.SuggestedProcessor = "" +} + func (t TraceOperation) String() string { return fmt.Sprintf(`[%s] "%v" -> "%v"`, t.Operation, string(t.Key), string(t.Value)) } type traceOperationInter struct { - Operation string `json:"operation"` - Key []byte `json:"key"` - Value []byte `json:"value"` - Metadata map[string]interface{} `json:"metadata"` + Operation string `json:"operation"` + Key []byte `json:"key"` + Value []byte `json:"value"` + Metadata struct { + BlockHeight uint64 `json:"blockHeight"` + TxHash string `json:"txHash"` + } `json:"metadata"` +} + +func (toi *traceOperationInter) Reset() { + toi.Operation = "" + toi.Key = toi.Key[:0] + toi.Value = toi.Value[:0] + toi.Metadata.BlockHeight = 0 + toi.Metadata.TxHash = "" } func (t *TraceOperation) UnmarshalJSON(bytes []byte) error { - toi := traceOperationInter{} + toi := toiPool.Get().(*traceOperationInter) + toi.Reset() if err := json.Unmarshal(bytes, &toi); err != nil { return err } - if toi.Metadata == nil { - t.BlockHeight = 0 - } else { - if data, ok := toi.Metadata[metadataBlockHeight]; ok { - t.BlockHeight = uint64(data.(float64)) - } - - if data, ok := toi.Metadata[metadataTxHash]; ok { - t.TxHash = data.(string) - } - } - + t.BlockHeight = toi.Metadata.BlockHeight + t.TxHash = toi.Metadata.TxHash t.Operation = toi.Operation t.Key = toi.Key t.Value = toi.Value + toiPool.Put(toi) + return nil } diff --git a/tracelistener/tracelistener.go b/tracelistener/tracelistener.go index b9100412..dba7d917 100644 --- a/tracelistener/tracelistener.go +++ b/tracelistener/tracelistener.go @@ -1,12 +1,14 @@ package tracelistener import ( - "bytes" - "encoding/json" "fmt" "math" "reflect" + "sync" "time" + "unsafe" + + "github.com/goccy/go-json" models "github.com/allinbits/demeris-backend-models/tracelistener" "github.com/nxadm/tail" @@ -59,7 +61,7 @@ var SupportedSDKModuleList = map[SDKModuleName]struct{}{ const dbPlaceholderLimit = 65535 // Operation is a kind of operations a TraceWatcher observes. -type Operation []byte +type Operation string // String implements fmt.Stringer on Operation. func (o Operation) String() string { @@ -68,16 +70,21 @@ func (o Operation) String() string { var ( // WriteOp is a write trace operation - WriteOp Operation = []byte("write") + WriteOp Operation = Operation(writeOpStr) // DeleteOp is a write trace operation - DeleteOp Operation = []byte("delete") + DeleteOp Operation = Operation(deleteOpStr) // ReadOp is a write trace operation - ReadOp Operation = []byte("read") + ReadOp Operation = Operation(readOpStr) // IterRangeOp is a write trace operation - IterRangeOp Operation = []byte("iterRange") + IterRangeOp Operation = Operation(iterRangeOp) + + writeOpStr = "write" + deleteOpStr = "delete" + readOpStr = "read" + iterRangeOp = "iterRange" ) // WritebackOp represents a unit of database writeback operated by a processor. @@ -243,9 +250,17 @@ type TraceWatcher struct { DataChan chan<- TraceOperation ErrorChan chan<- error Logger *zap.SugaredLogger + + toPool sync.Pool } func (tr *TraceWatcher) Watch() { + tr.toPool = sync.Pool{ + New: func() interface{} { + return &TraceOperation{} + }, + } + errorHappened := false for { // infinite cycle, if something goes wrong in reading the fifo we restart the cycle if errorHappened { @@ -263,44 +278,55 @@ func (tr *TraceWatcher) Watch() { } for line := range t.Lines { - if line.Err != nil { - tr.ErrorChan <- fmt.Errorf("line reading error, line %v, error %w", line, err) - break // restart the reading loop - } + tr.handleLine(line) + } + } +} - tr.Logger.Debugw("new line read from reader", "line", line.Text) +func unsafeGetBytes(s string) []byte { + return (*[0x7fff0000]byte)(unsafe.Pointer( + (*reflect.StringHeader)(unsafe.Pointer(&s)).Data), + )[:len(s):len(s)] +} - lineBytes := []byte(line.Text) +func (tr *TraceWatcher) handleLine(line *tail.Line) { + if line.Err != nil { + tr.ErrorChan <- fmt.Errorf("line reading error, line %v, error %w", line, line.Err) + return + } - // Log line used to trigger Grafana alerts. - // Do not modify or remove without changing the corresponding dashboards - tr.Logger.Infow("Probe", "c", "trace", "s", len(lineBytes)) + tr.Logger.Debugw("new line read from reader", "line", line.Text) - if !tr.mustConsiderData(lineBytes) { - continue - } + lineBytes := unsafeGetBytes(line.Text) - to := TraceOperation{} - if err := json.Unmarshal(lineBytes, &to); err != nil { - tr.ErrorChan <- fmt.Errorf("failed unmarshaling, %w, data: %s", err, line.Text) - continue - } + // Log line used to trigger Grafana alerts. + // Do not modify or remove without changing the corresponding dashboards + tr.Logger.Infow("Probe", "c", "trace", "s", len(lineBytes)) - if err := tr.ParseOperation(to); err != nil { - tr.ErrorChan <- fmt.Errorf("failed parsing operation, %w, data: %s", err, line.Text) - continue - } + to := tr.toPool.Get().(*TraceOperation) + to.Reset() + defer func() { + tr.toPool.Put(to) + }() - tr.Logger.Infow("trace processed", - "kind", to.Operation, - "block_height", to.BlockHeight, - "tx_hash", to.TxHash, - ) - } + if err := json.Unmarshal(lineBytes, &to); err != nil { + tr.ErrorChan <- fmt.Errorf("failed unmarshaling, %w, data: %s", err, line.Text) + return } + + if err := tr.ParseOperation(to); err != nil { + tr.ErrorChan <- fmt.Errorf("failed parsing operation, %w, data: %s", err, line.Text) + return + } + + tr.Logger.Infow("trace processed", + "kind", to.Operation, + "block_height", to.BlockHeight, + "tx_hash", to.TxHash, + ) } -func (tr *TraceWatcher) ParseOperation(data TraceOperation) error { +func (tr *TraceWatcher) ParseOperation(data *TraceOperation) error { if !tr.mustConsiderOperation(data) { return nil } @@ -311,27 +337,13 @@ func (tr *TraceWatcher) ParseOperation(data TraceOperation) error { } go func() { - tr.DataChan <- data + tr.DataChan <- data.Copy() }() return nil } -func (tr *TraceWatcher) mustConsiderData(b []byte) bool { - if tr.WatchedOps == nil || len(tr.WatchedOps) == 0 { - return true - } - - for _, op := range tr.WatchedOps { - if bytes.Contains(b, op) { - return true - } - } - - return false -} - -func (tr *TraceWatcher) mustConsiderOperation(op TraceOperation) bool { +func (tr *TraceWatcher) mustConsiderOperation(op *TraceOperation) bool { if tr.WatchedOps == nil || len(tr.WatchedOps) == 0 { return true } From 3b344270eced04d95672a483e4cc291804b81684 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Fri, 4 Mar 2022 14:05:26 +0100 Subject: [PATCH 03/19] chore(benchmark): drain data channel for good measure --- tracelistener/benchmark_test.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tracelistener/benchmark_test.go b/tracelistener/benchmark_test.go index a6379ed0..beb5e32d 100644 --- a/tracelistener/benchmark_test.go +++ b/tracelistener/benchmark_test.go @@ -40,6 +40,12 @@ func runBenchmark(b *testing.B, amount int, kind string) { Logger: l.Sugar(), } + go func() { + // drain data channel + for range dataChan { + } + }() + go func() { tw.Watch() }() From f877695ccf068c67df9923d4db57088bdb7b0332 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Fri, 4 Mar 2022 14:39:46 +0100 Subject: [PATCH 04/19] chore(benchmark): add func to read traces off a single file --- tracelistener/benchmark_test.go | 58 ++++++++++++++++++++++++++++++++- 1 file changed, 57 insertions(+), 1 deletion(-) diff --git a/tracelistener/benchmark_test.go b/tracelistener/benchmark_test.go index beb5e32d..07cac123 100644 --- a/tracelistener/benchmark_test.go +++ b/tracelistener/benchmark_test.go @@ -1,6 +1,7 @@ package tracelistener_test import ( + "bufio" "context" "fmt" "io" @@ -13,7 +14,7 @@ import ( "go.uber.org/zap" ) -func runBenchmark(b *testing.B, amount int, kind string) { +func setup(b *testing.B) io.ReadWriteCloser { b.Helper() f, err := os.CreateTemp("", "test_data") if err != nil { @@ -54,6 +55,15 @@ func runBenchmark(b *testing.B, amount int, kind string) { if err != nil { panic(err) } + + return ff +} + +func runBenchmark(b *testing.B, amount int, kind string) { + ff := setup(b) + + b.ResetTimer() + for i := 0; i < amount; i++ { err := loadTest(b, i, ff, kind) if err != nil { @@ -64,6 +74,23 @@ func runBenchmark(b *testing.B, amount int, kind string) { ff.Close() } +func BenchmarkTracelistenerRealTraces(b *testing.B) { + b.Log("reading test traces file...") + lines, err := loadTestFile(b) + if err != nil { + b.Fatal(err) + } + b.Log("finished reading test traces file!") + + ff := setup(b) + + b.ResetTimer() + + for _, line := range lines { + fmt.Fprintf(ff, line+"\n") + } +} + func BenchmarkTraceListenerKindWrite(b *testing.B) { runBenchmark(b, b.N, "write") } @@ -107,3 +134,32 @@ func loadTest(b *testing.B, height int, ff io.Writer, kind string) error { return nil } + +func loadTestFile(b *testing.B) ([]string, error) { + b.Helper() + + fname := os.Getenv("TRACELISTENER_BENCH_TRACEFILE") + if fname == "" { + return nil, fmt.Errorf("TRACELISTENER_BENCH_TRACEFILE environment variable not defined") + } + + file, err := os.Open(fname) + if err != nil { + return nil, fmt.Errorf("cannot open file %s, %w", fname, err) + } + + scanner := bufio.NewScanner(file) + buf := make([]byte, 1000000) // a very high capacity + scanner.Buffer(buf, 1000000) + + ret := []string{} + for scanner.Scan() { + ret = append(ret, scanner.Text()) + } + + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("scanning error, %w", err) + } + + return ret, nil +} From 1f1643a916daa14f389c93868bf008a5b7dc107e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Fri, 4 Mar 2022 14:40:14 +0100 Subject: [PATCH 05/19] chore(tracelistener): obsolete 2-stage unmarshal --- tracelistener/processor/auth_test.go | 16 +-- tracelistener/processor/bank_test.go | 8 +- .../processor/datamarshaler/impl_v44.go | 26 ++--- tracelistener/processor/delegation_test.go | 19 ++-- tracelistener/processor/delegations.go | 2 +- tracelistener/processor/processor.go | 4 +- tracelistener/processor/processor_test.go | 40 +++++--- .../processor/unbonding_delegation_test.go | 20 ++-- .../unbonding_delegation_v44_test.go | 16 +-- tracelistener/trace.go | 99 ++++++++++--------- tracelistener/trace_test.go | 14 ++- tracelistener/tracelistener.go | 7 +- 12 files changed, 152 insertions(+), 119 deletions(-) diff --git a/tracelistener/processor/auth_test.go b/tracelistener/processor/auth_test.go index 1dc4e93f..f4c32d5a 100644 --- a/tracelistener/processor/auth_test.go +++ b/tracelistener/processor/auth_test.go @@ -77,9 +77,11 @@ func TestAuthProcess(t *testing.T) { Sequence: 11, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("cosmos1xrnner9s783446"), - BlockHeight: 1, + Operation: string(tracelistener.WriteOp), + Key: []byte("cosmos1xrnner9s783446"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + }, }, false, 1, @@ -92,9 +94,11 @@ func TestAuthProcess(t *testing.T) { Sequence: 11, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("cosmos1xrnner9s783446"), - BlockHeight: 1, + Operation: string(tracelistener.WriteOp), + Key: []byte("cosmos1xrnner9s783446"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + }, }, true, 0, diff --git a/tracelistener/processor/bank_test.go b/tracelistener/processor/bank_test.go index e989c79f..82af4bca 100644 --- a/tracelistener/processor/bank_test.go +++ b/tracelistener/processor/bank_test.go @@ -75,9 +75,11 @@ func TestBankProcess(t *testing.T) { Amount: 500, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("cosmos1xrnner9s783446yz3hhshpr5fpz6wzcwkvwv5j"), - BlockHeight: 101, + Operation: string(tracelistener.WriteOp), + Key: []byte("cosmos1xrnner9s783446yz3hhshpr5fpz6wzcwkvwv5j"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 101, + }, }, false, 1, diff --git a/tracelistener/processor/datamarshaler/impl_v44.go b/tracelistener/processor/datamarshaler/impl_v44.go index 73ee7f45..221632cd 100644 --- a/tracelistener/processor/datamarshaler/impl_v44.go +++ b/tracelistener/processor/datamarshaler/impl_v44.go @@ -93,15 +93,15 @@ func (d DataMarshaler) Bank(data tracelistener.TraceOperation) (models.BalanceRo "operation", data.Operation, "address", hAddr, "new_balance", coins.String(), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) return models.BalanceRow{ Address: hAddr, Amount: coins.String(), Denom: coins.Denom, - BlockHeight: data.BlockHeight, + BlockHeight: data.Metadata.BlockHeight, }, nil } @@ -160,8 +160,8 @@ func (d DataMarshaler) Auth(data tracelistener.TraceOperation) (models.AuthRow, "address", hAddr, "sequence_number", acc.GetSequence(), "account_number", acc.GetAccountNumber(), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) return models.AuthRow{ @@ -219,15 +219,15 @@ func (d DataMarshaler) Delegations(data tracelistener.TraceOperation) (models.De "delegator", delegator, "validator", validator, "amount", delegation.Shares.String(), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) return models.DelegationRow{ Delegator: delegator, Validator: validator, Amount: delegation.Shares.String(), - BlockHeight: data.BlockHeight, + BlockHeight: data.Metadata.BlockHeight, }, nil } @@ -421,8 +421,8 @@ func (d DataMarshaler) UnbondingDelegations(data tracelistener.TraceOperation) ( "delegator", delegator, "validator", validator, "entries", string(entries), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) var entriesStore models.UnbondingDelegationEntries @@ -469,9 +469,9 @@ func (d DataMarshaler) Validators(data tracelistener.TraceOperation) (models.Val d.l.Debugw("new validator write", "operator_address", v.OperatorAddress, - "height", data.BlockHeight, - "txHash", data.TxHash, - "cons pub key type", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, + "cons pub key type", data.Metadata.TxHash, "cons pub key", val, "key", k, ) diff --git a/tracelistener/processor/delegation_test.go b/tracelistener/processor/delegation_test.go index 8f2371a5..919f28bb 100644 --- a/tracelistener/processor/delegation_test.go +++ b/tracelistener/processor/delegation_test.go @@ -89,10 +89,12 @@ func TestDelegationProcess(t *testing.T) { Shares: 100, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), - BlockHeight: 1, - TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + Operation: string(tracelistener.WriteOp), + Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + }, }, false, 1, @@ -103,10 +105,11 @@ func TestDelegationProcess(t *testing.T) { Shares: 100, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), - BlockHeight: 1, - TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + Operation: string(tracelistener.WriteOp), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + }, }, true, 0, diff --git a/tracelistener/processor/delegations.go b/tracelistener/processor/delegations.go index 8155c652..cc1df5d0 100644 --- a/tracelistener/processor/delegations.go +++ b/tracelistener/processor/delegations.go @@ -104,7 +104,7 @@ func (b *delegationsProcessor) Process(data tracelistener.TraceOperation) error Delegator: res.Delegator, Validator: res.Validator, Amount: res.Amount, - BlockHeight: data.BlockHeight, + BlockHeight: data.Metadata.BlockHeight, } return nil diff --git a/tracelistener/processor/processor.go b/tracelistener/processor/processor.go index 4283e565..a239a0a3 100644 --- a/tracelistener/processor/processor.go +++ b/tracelistener/processor/processor.go @@ -196,7 +196,7 @@ func (p *Processor) Flush() error { func (p *Processor) lifecycle() { for data := range p.writeChan { - if data.BlockHeight != p.lastHeight && data.BlockHeight != 0 { + if data.Metadata.BlockHeight != p.lastHeight && data.Metadata.BlockHeight != 0 { if err := p.Flush(); err != nil { p.errorsChan <- fmt.Errorf("error while flushing caches, %w", err) continue @@ -204,7 +204,7 @@ func (p *Processor) lifecycle() { p.l.Infow("processed new block", "height", p.lastHeight) - p.lastHeight = data.BlockHeight + p.lastHeight = data.Metadata.BlockHeight } processorList := p.moduleProcessors diff --git a/tracelistener/processor/processor_test.go b/tracelistener/processor/processor_test.go index 9a9a9764..9944c411 100644 --- a/tracelistener/processor/processor_test.go +++ b/tracelistener/processor/processor_test.go @@ -120,10 +120,12 @@ func TestLifecycle(t *testing.T) { "no error when queueing new message accepted by the processor", nil, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("key"), - Value: []byte("key"), - BlockHeight: 0, + Operation: string(tracelistener.WriteOp), + Key: []byte("key"), + Value: []byte("key"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + }, }, func(_ tracelistener.TraceOperation) error { return nil @@ -135,10 +137,12 @@ func TestLifecycle(t *testing.T) { "error when queueing new message accepted by the processor", nil, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("key"), - Value: []byte("key"), - BlockHeight: 0, + Operation: string(tracelistener.WriteOp), + Key: []byte("key"), + Value: []byte("key"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 0, + }, }, func(_ tracelistener.TraceOperation) error { return fmt.Errorf("oh no, error") @@ -150,17 +154,21 @@ func TestLifecycle(t *testing.T) { "new message, block different re: last height", []tracelistener.TraceOperation{ { - Operation: string(tracelistener.WriteOp), - Key: []byte("key"), - Value: []byte("key"), - BlockHeight: 0, + Operation: string(tracelistener.WriteOp), + Key: []byte("key"), + Value: []byte("key"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 0, + }, }, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("key"), - Value: []byte("key"), - BlockHeight: 1, + Operation: string(tracelistener.WriteOp), + Key: []byte("key"), + Value: []byte("key"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + }, }, func(_ tracelistener.TraceOperation) error { return nil diff --git a/tracelistener/processor/unbonding_delegation_test.go b/tracelistener/processor/unbonding_delegation_test.go index 2442f19e..5a329180 100644 --- a/tracelistener/processor/unbonding_delegation_test.go +++ b/tracelistener/processor/unbonding_delegation_test.go @@ -83,10 +83,12 @@ func TestUnbondingDelegationProcess(t *testing.T) { }, }, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), - BlockHeight: 1, - TxHash: "066050E449C3450F943FC6227F155C19EF5C14653F268E9BAFEFE93DF9B3EDAD", + Operation: string(tracelistener.WriteOp), + Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + TxHash: "066050E449C3450F943FC6227F155C19EF5C14653F268E9BAFEFE93DF9B3EDAD", + }, }, false, 1, @@ -95,10 +97,12 @@ func TestUnbondingDelegationProcess(t *testing.T) { "Invalid addresses - error", datamarshaler.TestUnbondingDelegation{}, tracelistener.TraceOperation{ - Operation: string(tracelistener.WriteOp), - Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), - BlockHeight: 1, - TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + Operation: string(tracelistener.WriteOp), + Key: []byte("AtdlV8qD6o6J2shsj9acpI+9Opd/e5uTqZIi7NK5i3y9"), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 1, + TxHash: "A5CF62609D62ADDE56816681B6191F5F0252D2800FC2C312EB91D962AB7A97CB", + }, }, true, 0, diff --git a/tracelistener/processor/unbonding_delegation_v44_test.go b/tracelistener/processor/unbonding_delegation_v44_test.go index 67e3185d..2e871384 100644 --- a/tracelistener/processor/unbonding_delegation_v44_test.go +++ b/tracelistener/processor/unbonding_delegation_v44_test.go @@ -29,10 +29,12 @@ func versionSpecificUnbondingDelegationsProcessTests() []unbondingDelegationsPro Validator: "cosmosvaloper19xawgvgn887e9gef5vkzkemwh33mtgwa6haa7s", }, tracelistener.TraceOperation{ - Operation: string(tracelistener.DeleteOp), - Key: []byte("QXRkbFY4cUQ2bzZKMnNoc2o5YWNwSSs5T3BkL2U1dVRxWklpN05LNWkzeTk="), - Value: []byte{}, - BlockHeight: 0, + Operation: string(tracelistener.DeleteOp), + Key: []byte("QXRkbFY4cUQ2bzZKMnNoc2o5YWNwSSs5T3BkL2U1dVRxWklpN05LNWkzeTk="), + Value: []byte{}, + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 0, + }, }, false, 0, @@ -49,8 +51,10 @@ func versionSpecificUnbondingDelegationsProcessTests() []unbondingDelegationsPro 0x33, // prefix 9, 118, 97, 108, 105, 100, 97, 116, 111, 114, 9, 100, 101, 108, 101, 103, 97, 116, 111, 114, }, - Value: []byte{}, - BlockHeight: 0, + Value: []byte{}, + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 0, + }, }, false, 1, diff --git a/tracelistener/trace.go b/tracelistener/trace.go index a55c0c4b..f0946864 100644 --- a/tracelistener/trace.go +++ b/tracelistener/trace.go @@ -2,9 +2,6 @@ package tracelistener import ( "fmt" - "sync" - - "github.com/goccy/go-json" ) const ( @@ -12,23 +9,23 @@ const ( metadataTxHash = "txHash" ) -var toiPool = sync.Pool{ - New: func() interface{} { - return &traceOperationInter{} - }, -} +// var toiPool = sync.Pool{ +// New: func() interface{} { +// return &traceOperationInter{} +// }, +// } -type TraceOperation struct { - Operation string `json:"operation"` - Key []byte `json:"key"` - Value []byte `json:"value"` - BlockHeight uint64 `json:"block_height"` - TxHash string `json:"tx_hash"` +// type TraceOperation struct { +// Operation string `json:"operation"` +// Key []byte `json:"key"` +// Value []byte `json:"value"` +// BlockHeight uint64 `json:"block_height"` +// TxHash string `json:"tx_hash"` - // SuggestedProcessor signals to the trace processor that - // what SDK module this trace comes from. - SuggestedProcessor SDKModuleName -} +// // SuggestedProcessor signals to the trace processor that +// // what SDK module this trace comes from. +// SuggestedProcessor SDKModuleName +// } func (to *TraceOperation) Copy() TraceOperation { ret := TraceOperation{} @@ -40,8 +37,8 @@ func (to *TraceOperation) Reset() { to.Operation = "" to.Key = to.Key[:0] to.Value = to.Value[:0] - to.BlockHeight = 0 - to.TxHash = "" + to.Metadata.BlockHeight = 0 + to.Metadata.TxHash = "" to.SuggestedProcessor = "" } @@ -49,39 +46,45 @@ func (t TraceOperation) String() string { return fmt.Sprintf(`[%s] "%v" -> "%v"`, t.Operation, string(t.Key), string(t.Value)) } -type traceOperationInter struct { - Operation string `json:"operation"` - Key []byte `json:"key"` - Value []byte `json:"value"` - Metadata struct { - BlockHeight uint64 `json:"blockHeight"` - TxHash string `json:"txHash"` - } `json:"metadata"` +type TraceMetadata struct { + BlockHeight uint64 `json:"blockHeight"` + TxHash string `json:"txHash"` } -func (toi *traceOperationInter) Reset() { - toi.Operation = "" - toi.Key = toi.Key[:0] - toi.Value = toi.Value[:0] - toi.Metadata.BlockHeight = 0 - toi.Metadata.TxHash = "" +type TraceOperation struct { + Operation string `json:"operation"` + Key []byte `json:"key"` + Value []byte `json:"value"` + Metadata TraceMetadata `json:"metadata"` + + // SuggestedProcessor signals to the trace processor that + // what SDK module this trace comes from. + SuggestedProcessor SDKModuleName } -func (t *TraceOperation) UnmarshalJSON(bytes []byte) error { - toi := toiPool.Get().(*traceOperationInter) - toi.Reset() +// func (toi *traceOperationInter) Reset() { +// toi.Operation = "" +// toi.Key = toi.Key[:0] +// toi.Value = toi.Value[:0] +// toi.Metadata.BlockHeight = 0 +// toi.Metadata.TxHash = "" +// } - if err := json.Unmarshal(bytes, &toi); err != nil { - return err - } +// func (t *TraceOperation) UnmarshalJSON(bytes []byte) error { +// toi := toiPool.Get().(*traceOperationInter) +// toi.Reset() - t.BlockHeight = toi.Metadata.BlockHeight - t.TxHash = toi.Metadata.TxHash - t.Operation = toi.Operation - t.Key = toi.Key - t.Value = toi.Value +// if err := json.Unmarshal(bytes, &toi); err != nil { +// return err +// } - toiPool.Put(toi) +// t.BlockHeight = toi.Metadata.BlockHeight +// t.TxHash = toi.Metadata.TxHash +// t.Operation = toi.Operation +// t.Key = toi.Key +// t.Value = toi.Value - return nil -} +// toiPool.Put(toi) + +// return nil +// } diff --git a/tracelistener/trace_test.go b/tracelistener/trace_test.go index e389bb60..d4afffbd 100644 --- a/tracelistener/trace_test.go +++ b/tracelistener/trace_test.go @@ -84,10 +84,12 @@ func TestTraceOperation_UnmarshalJSON(t1 *testing.T) { "operation with block height", opWithBlockHeight, tracelistener.TraceOperation{ - Operation: "write", - Key: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, - Value: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, - BlockHeight: 42, + Operation: "write", + Key: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, + Value: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 42, + }, }, false, }, @@ -98,7 +100,9 @@ func TestTraceOperation_UnmarshalJSON(t1 *testing.T) { Operation: "write", Key: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, Value: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f, 0xa}, - TxHash: "hash", + Metadata: tracelistener.TraceMetadata{ + TxHash: "hash", + }, }, false, }, diff --git a/tracelistener/tracelistener.go b/tracelistener/tracelistener.go index dba7d917..72b26643 100644 --- a/tracelistener/tracelistener.go +++ b/tracelistener/tracelistener.go @@ -1,6 +1,7 @@ package tracelistener import ( + "encoding/json" "fmt" "math" "reflect" @@ -8,7 +9,7 @@ import ( "time" "unsafe" - "github.com/goccy/go-json" + //"github.com/goccy/go-json" models "github.com/allinbits/demeris-backend-models/tracelistener" "github.com/nxadm/tail" @@ -321,8 +322,8 @@ func (tr *TraceWatcher) handleLine(line *tail.Line) { tr.Logger.Infow("trace processed", "kind", to.Operation, - "block_height", to.BlockHeight, - "tx_hash", to.TxHash, + "block_height", to.Metadata.BlockHeight, + "tx_hash", to.Metadata.TxHash, ) } From ac508c306e2ca75e0b81647dabaf75d71f0637bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Fri, 4 Mar 2022 14:50:51 +0100 Subject: [PATCH 06/19] chore(benchmark): actually write data --- tracelistener/benchmark_test.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tracelistener/benchmark_test.go b/tracelistener/benchmark_test.go index 07cac123..46a3d77c 100644 --- a/tracelistener/benchmark_test.go +++ b/tracelistener/benchmark_test.go @@ -14,7 +14,7 @@ import ( "go.uber.org/zap" ) -func setup(b *testing.B) io.ReadWriteCloser { +func setup(b *testing.B) (io.ReadWriteCloser, string) { b.Helper() f, err := os.CreateTemp("", "test_data") if err != nil { @@ -25,7 +25,6 @@ func setup(b *testing.B) io.ReadWriteCloser { if err != nil { panic(err) } - defer os.Remove(f.Name()) dataChan := make(chan tracelistener.TraceOperation) errChan := make(chan error) @@ -56,11 +55,11 @@ func setup(b *testing.B) io.ReadWriteCloser { panic(err) } - return ff + return ff, f.Name() } func runBenchmark(b *testing.B, amount int, kind string) { - ff := setup(b) + ff, fifoName := setup(b) b.ResetTimer() @@ -71,6 +70,8 @@ func runBenchmark(b *testing.B, amount int, kind string) { } } + os.Remove(fifoName) + ff.Close() } @@ -82,13 +83,17 @@ func BenchmarkTracelistenerRealTraces(b *testing.B) { } b.Log("finished reading test traces file!") - ff := setup(b) + ff, fifoName := setup(b) b.ResetTimer() for _, line := range lines { fmt.Fprintf(ff, line+"\n") } + + os.Remove(fifoName) + + ff.Close() } func BenchmarkTraceListenerKindWrite(b *testing.B) { From 823aef402097c100e15042714021d221c0280aa3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 11:33:37 +0100 Subject: [PATCH 07/19] feat(tracestats): add tracestats program `tracestats` reads cosmos-sdk traces written to a file by the`--trace-store` command flag, and retrieves - block height - key length - value length - total trace length off each and every trace line, then creates a CSV file with this data. Useful for data analysis purposes. --- cmd/tracestats/main.go | 108 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 cmd/tracestats/main.go diff --git a/cmd/tracestats/main.go b/cmd/tracestats/main.go new file mode 100644 index 00000000..4057a727 --- /dev/null +++ b/cmd/tracestats/main.go @@ -0,0 +1,108 @@ +package main + +import ( + "bufio" + "encoding/csv" + "encoding/json" + "fmt" + "os" + "strconv" + + "github.com/allinbits/tracelistener/tracelistener" +) + +type traceInfo struct { + BlockHeight uint64 + KeyLength uint64 + ValueLength uint64 + Length uint64 +} + +type traceInfos []traceInfo + +func (ti traceInfos) CSV() [][]string { + ret := make([][]string, 0, 1+len(ti)) // add 1 row for title + + ret = append(ret, []string{"block_height", "key_lengt", "value_length", "length"}) + + for _, t := range ti { + ret = append(ret, []string{ + strconv.FormatUint(t.BlockHeight, 10), + strconv.FormatUint(t.KeyLength, 10), + strconv.FormatUint(t.ValueLength, 10), + strconv.FormatUint(t.Length, 10), + }) + } + + return ret +} + +func main() { + fname := os.Args[1] + + rows, err := loadTestFile(fname) + if err != nil { + panic(err) + } + + ti, err := getTraceInfo(rows) + if err != nil { + panic(err) + } + + o, err := os.OpenFile("tracestats.csv", os.O_CREATE|os.O_APPEND|os.O_RDWR, 0755) + if err != nil { + panic(err) + } + + w := csv.NewWriter(o) + for _, record := range ti.CSV() { + if err := w.Write(record); err != nil { + panic(err) + } + } + + o.Close() +} + +func getTraceInfo(traces []string) (traceInfos, error) { + ret := make([]traceInfo, 0, len(traces)) + + for _, t := range traces { + tr := tracelistener.TraceOperation{} + if err := json.Unmarshal([]byte(t), &tr); err != nil { + return nil, err + } + + ret = append(ret, traceInfo{ + BlockHeight: tr.Metadata.BlockHeight, + KeyLength: uint64(len(tr.Key)), + ValueLength: uint64(len(tr.Value)), + Length: uint64(len(t)), + }) + } + + return ret, nil +} + +func loadTestFile(fname string) ([]string, error) { + file, err := os.Open(fname) + if err != nil { + return nil, fmt.Errorf("cannot open file %s, %w", fname, err) + } + + scanner := bufio.NewScanner(file) + buf := make([]byte, 1000000) // a very high capacity + scanner.Buffer(buf, 1000000) + + ret := []string{} + for scanner.Scan() { + ret = append(ret, scanner.Text()) + } + + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("scanning error, %w", err) + } + + return ret, nil +} From a57cf13444117d3aa7c079add40bf69ea74781cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 11:34:36 +0100 Subject: [PATCH 08/19] chore(tracelistener): revert back to encoding/json --- tracelistener/tracelistener.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/tracelistener/tracelistener.go b/tracelistener/tracelistener.go index 72b26643..bb78fb40 100644 --- a/tracelistener/tracelistener.go +++ b/tracelistener/tracelistener.go @@ -9,8 +9,6 @@ import ( "time" "unsafe" - //"github.com/goccy/go-json" - models "github.com/allinbits/demeris-backend-models/tracelistener" "github.com/nxadm/tail" From 7425a4153f6f0d95200011f412b3f70f67d577b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 11:34:58 +0100 Subject: [PATCH 09/19] chore(bulkimport): change TraceOperation interface Make it work with the new no-middleman type. --- tracelistener/bulk/bulkimport.go | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tracelistener/bulk/bulkimport.go b/tracelistener/bulk/bulkimport.go index bef3481e..23d32cc4 100644 --- a/tracelistener/bulk/bulkimport.go +++ b/tracelistener/bulk/bulkimport.go @@ -155,14 +155,16 @@ func (i *Importer) Do() error { for ; ii.Valid(); ii.Next() { to := tracelistener.TraceOperation{ - Operation: tracelistener.WriteOp.String(), - Key: ii.Key(), - Value: ii.Value(), - BlockHeight: uint64(latestBlockHeight), + Operation: tracelistener.WriteOp.String(), + Key: ii.Key(), + Value: ii.Value(), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: uint64(latestBlockHeight), + }, SuggestedProcessor: tracelistener.SDKModuleName(key.Name()), } - if err := i.TraceWatcher.ParseOperation(to); err != nil { + if err := i.TraceWatcher.ParseOperation(&to); err != nil { return fmt.Errorf("cannot parse operation %v, %w", to, err) } From 3cd4400d598b2aaaf60a2ad7c47b99a2b97ea4cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 13:48:03 +0100 Subject: [PATCH 10/19] fix(tracestats): defer file close --- cmd/tracestats/main.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/cmd/tracestats/main.go b/cmd/tracestats/main.go index 4057a727..96ad0a9b 100644 --- a/cmd/tracestats/main.go +++ b/cmd/tracestats/main.go @@ -55,14 +55,18 @@ func main() { panic(err) } + defer func() { + if err := o.Close(); err != nil { + panic(err) + } + }() + w := csv.NewWriter(o) for _, record := range ti.CSV() { if err := w.Write(record); err != nil { panic(err) } } - - o.Close() } func getTraceInfo(traces []string) (traceInfos, error) { From bdeb54e54672f0e34d64d1e0032f9bdd843f8a14 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 13:54:55 +0100 Subject: [PATCH 11/19] chore(benchmark): address readability concerns --- tracelistener/benchmark_test.go | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/tracelistener/benchmark_test.go b/tracelistener/benchmark_test.go index 46a3d77c..5ac11657 100644 --- a/tracelistener/benchmark_test.go +++ b/tracelistener/benchmark_test.go @@ -59,20 +59,23 @@ func setup(b *testing.B) (io.ReadWriteCloser, string) { } func runBenchmark(b *testing.B, amount int, kind string) { - ff, fifoName := setup(b) + fileWriter, fifoName := setup(b) + defer func() { + if err := fileWriter.Close(); err != nil { + panic(err) + } + }() b.ResetTimer() for i := 0; i < amount; i++ { - err := loadTest(b, i, ff, kind) + err := loadTest(b, i, fileWriter, kind) if err != nil { panic(err) } } os.Remove(fifoName) - - ff.Close() } func BenchmarkTracelistenerRealTraces(b *testing.B) { @@ -83,17 +86,20 @@ func BenchmarkTracelistenerRealTraces(b *testing.B) { } b.Log("finished reading test traces file!") - ff, fifoName := setup(b) + fileWriter, fifoName := setup(b) + defer func() { + if err := fileWriter.Close(); err != nil { + panic(err) + } + }() b.ResetTimer() for _, line := range lines { - fmt.Fprintf(ff, line+"\n") + fmt.Fprintf(fileWriter, line+"\n") } os.Remove(fifoName) - - ff.Close() } func BenchmarkTraceListenerKindWrite(b *testing.B) { From 6f17b2e52b28972dd2a98a6b6ab5d090c9175cae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 13:56:38 +0100 Subject: [PATCH 12/19] chore(tracelistener): address Operation godoc readability concerns. --- tracelistener/tracelistener.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tracelistener/tracelistener.go b/tracelistener/tracelistener.go index bb78fb40..0e5c2466 100644 --- a/tracelistener/tracelistener.go +++ b/tracelistener/tracelistener.go @@ -59,7 +59,7 @@ var SupportedSDKModuleList = map[SDKModuleName]struct{}{ // Info: https://github.com/cockroachdb/cockroach/issues/49256 const dbPlaceholderLimit = 65535 -// Operation is a kind of operations a TraceWatcher observes. +// Operation represents the kind of Cosmos SDK store operation a TraceWatcher observes. type Operation string // String implements fmt.Stringer on Operation. From 35626c374f13bbbdc94c52b72b2aea9a68d9bc41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 14:00:40 +0100 Subject: [PATCH 13/19] fix(tracelistener): address potential data race in TraceOperation writeback --- tracelistener/tracelistener.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tracelistener/tracelistener.go b/tracelistener/tracelistener.go index 0e5c2466..8bd3b4ed 100644 --- a/tracelistener/tracelistener.go +++ b/tracelistener/tracelistener.go @@ -335,9 +335,11 @@ func (tr *TraceWatcher) ParseOperation(data *TraceOperation) error { return nil } - go func() { - tr.DataChan <- data.Copy() - }() + // Happy path has been taken, locally copy data contents and pass + // them to the database writing goroutine. + go func(op TraceOperation) { + tr.DataChan <- op + }(data.Copy()) return nil } From 59056b3800c0c48a628ae404852a3a4ae0ded0d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 14:32:13 +0100 Subject: [PATCH 14/19] chore(tracelistener,processor): fix v42 tests --- .../processor/datamarshaler/impl_v42.go | 26 +++++++++---------- .../unbonding_delegation_v42_test.go | 10 ++++--- 2 files changed, 19 insertions(+), 17 deletions(-) diff --git a/tracelistener/processor/datamarshaler/impl_v42.go b/tracelistener/processor/datamarshaler/impl_v42.go index 7c10a832..df330022 100644 --- a/tracelistener/processor/datamarshaler/impl_v42.go +++ b/tracelistener/processor/datamarshaler/impl_v42.go @@ -64,14 +64,14 @@ func (d DataMarshaler) Bank(data tracelistener.TraceOperation) (models.BalanceRo "operation", data.Operation, "address", hAddr, "new_balance", coins.String(), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) return models.BalanceRow{ Address: hAddr, Amount: coins.String(), Denom: coins.Denom, - BlockHeight: data.BlockHeight, + BlockHeight: data.Metadata.BlockHeight, }, nil } @@ -129,8 +129,8 @@ func (d DataMarshaler) Auth(data tracelistener.TraceOperation) (models.AuthRow, "address", hAddr, "sequence_number", acc.GetSequence(), "account_number", acc.GetAccountNumber(), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) return models.AuthRow{ @@ -176,15 +176,15 @@ func (d DataMarshaler) Delegations(data tracelistener.TraceOperation) (models.De "delegator", delegator, "validator", validator, "amount", delegation.Shares.String(), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) return models.DelegationRow{ Delegator: delegator, Validator: validator, Amount: delegation.Shares.String(), - BlockHeight: data.BlockHeight, + BlockHeight: data.Metadata.BlockHeight, }, nil } @@ -356,8 +356,8 @@ func (d DataMarshaler) UnbondingDelegations(data tracelistener.TraceOperation) ( "delegator", delegator, "validator", validator, "entries", string(entries), - "height", data.BlockHeight, - "txHash", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, ) var entriesStore models.UnbondingDelegationEntries @@ -402,9 +402,9 @@ func (d DataMarshaler) Validators(data tracelistener.TraceOperation) (models.Val d.l.Debugw("new validator write", "operator_address", v.OperatorAddress, - "height", data.BlockHeight, - "txHash", data.TxHash, - "cons pub key type", data.TxHash, + "height", data.Metadata.BlockHeight, + "txHash", data.Metadata.TxHash, + "cons pub key type", data.Metadata.TxHash, "cons pub key", val, "key", k, ) diff --git a/tracelistener/processor/unbonding_delegation_v42_test.go b/tracelistener/processor/unbonding_delegation_v42_test.go index 1c0b9cf6..a380ddf0 100644 --- a/tracelistener/processor/unbonding_delegation_v42_test.go +++ b/tracelistener/processor/unbonding_delegation_v42_test.go @@ -22,10 +22,12 @@ func versionSpecificUnbondingDelegationsProcessTests() []unbondingDelegationsPro Validator: "cosmosvaloper19xawgvgn887e9gef5vkzkemwh33mtgwa6haa7s", }, tracelistener.TraceOperation{ - Operation: string(tracelistener.DeleteOp), - Key: []byte("QXRkbFY4cUQ2bzZKMnNoc2o5YWNwSSs5T3BkL2U1dVRxWklpN05LNWkzeTk="), - Value: []byte("Ci1jb3Ntb3MxeHJubmVyOXM3ODM0NDZ5ejNoaHNocHI1ZnB6Nnd6Y3drdnd2NWoSNGNvc21vc3ZhbG9wZXIxOXhhd2d2Z244ODdlOWdlZjV2a3prZW13aDMzbXRnd2E2aGFhN3MaHAiYIBILCICSuMOY/v///wEaBDEwMDAiBDExMDA="), - BlockHeight: 0, + Operation: string(tracelistener.DeleteOp), + Key: []byte("QXRkbFY4cUQ2bzZKMnNoc2o5YWNwSSs5T3BkL2U1dVRxWklpN05LNWkzeTk="), + Value: []byte("Ci1jb3Ntb3MxeHJubmVyOXM3ODM0NDZ5ejNoaHNocHI1ZnB6Nnd6Y3drdnd2NWoSNGNvc21vc3ZhbG9wZXIxOXhhd2d2Z244ODdlOWdlZjV2a3prZW13aDMzbXRnd2E2aGFhN3MaHAiYIBILCICSuMOY/v///wEaBDEwMDAiBDExMDA="), + Metadata: tracelistener.TraceMetadata{ + BlockHeight: 0, + }, }, false, 1, From 60a0c39731081af139a417dbe6f6d3de9afbecfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 14:47:08 +0100 Subject: [PATCH 15/19] chore(tracelistener): cleanup trace.go --- tracelistener/trace.go | 50 ------------------------------------------ 1 file changed, 50 deletions(-) diff --git a/tracelistener/trace.go b/tracelistener/trace.go index f0946864..2a8c654b 100644 --- a/tracelistener/trace.go +++ b/tracelistener/trace.go @@ -4,29 +4,6 @@ import ( "fmt" ) -const ( - metadataBlockHeight = "blockHeight" - metadataTxHash = "txHash" -) - -// var toiPool = sync.Pool{ -// New: func() interface{} { -// return &traceOperationInter{} -// }, -// } - -// type TraceOperation struct { -// Operation string `json:"operation"` -// Key []byte `json:"key"` -// Value []byte `json:"value"` -// BlockHeight uint64 `json:"block_height"` -// TxHash string `json:"tx_hash"` - -// // SuggestedProcessor signals to the trace processor that -// // what SDK module this trace comes from. -// SuggestedProcessor SDKModuleName -// } - func (to *TraceOperation) Copy() TraceOperation { ret := TraceOperation{} ret = *to @@ -61,30 +38,3 @@ type TraceOperation struct { // what SDK module this trace comes from. SuggestedProcessor SDKModuleName } - -// func (toi *traceOperationInter) Reset() { -// toi.Operation = "" -// toi.Key = toi.Key[:0] -// toi.Value = toi.Value[:0] -// toi.Metadata.BlockHeight = 0 -// toi.Metadata.TxHash = "" -// } - -// func (t *TraceOperation) UnmarshalJSON(bytes []byte) error { -// toi := toiPool.Get().(*traceOperationInter) -// toi.Reset() - -// if err := json.Unmarshal(bytes, &toi); err != nil { -// return err -// } - -// t.BlockHeight = toi.Metadata.BlockHeight -// t.TxHash = toi.Metadata.TxHash -// t.Operation = toi.Operation -// t.Key = toi.Key -// t.Value = toi.Value - -// toiPool.Put(toi) - -// return nil -// } From 0918f2e2220a153d590c082fb4326ac06f7a5102 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 14:50:06 +0100 Subject: [PATCH 16/19] chore(bulkimport): fix potential failure of sync.Pool Since tracewatcher now holds a sync.Pool, we can't just copy it around anymore, needs pointers. --- cmd/tracelistener/main.go | 2 +- tracelistener/bulk/bulkimport.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/tracelistener/main.go b/cmd/tracelistener/main.go index a456cc69..52467f20 100644 --- a/cmd/tracelistener/main.go +++ b/cmd/tracelistener/main.go @@ -82,7 +82,7 @@ func main() { if ca.existingDatabasePath != "" { importer := bulk.Importer{ Path: ca.existingDatabasePath, - TraceWatcher: watcher, + TraceWatcher: &watcher, Processor: dpi, Logger: logger, Database: di, diff --git a/tracelistener/bulk/bulkimport.go b/tracelistener/bulk/bulkimport.go index 23d32cc4..156a87e5 100644 --- a/tracelistener/bulk/bulkimport.go +++ b/tracelistener/bulk/bulkimport.go @@ -25,7 +25,7 @@ import ( type Importer struct { Path string - TraceWatcher tracelistener.TraceWatcher + TraceWatcher *tracelistener.TraceWatcher Processor tracelistener.DataProcessor Logger *zap.SugaredLogger Database *database.Instance @@ -41,7 +41,7 @@ func ImportableModulesList() []string { return ml } -func (i Importer) validateModulesList() error { +func (i *Importer) validateModulesList() error { for _, m := range i.Modules { if _, ok := tracelistener.SupportedSDKModuleList[tracelistener.SDKModuleName(m)]; !ok { return fmt.Errorf("unknown bulk import module %s", m) From 3b98d40738b60d76e56b0c1a81fc938e55382ec4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 14:50:19 +0100 Subject: [PATCH 17/19] chore(tracelistener): make the linter happy --- tracelistener/trace.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tracelistener/trace.go b/tracelistener/trace.go index 2a8c654b..d73b5737 100644 --- a/tracelistener/trace.go +++ b/tracelistener/trace.go @@ -5,8 +5,7 @@ import ( ) func (to *TraceOperation) Copy() TraceOperation { - ret := TraceOperation{} - ret = *to + ret := *to return ret } @@ -19,7 +18,7 @@ func (to *TraceOperation) Reset() { to.SuggestedProcessor = "" } -func (t TraceOperation) String() string { +func (to TraceOperation) String() string { return fmt.Sprintf(`[%s] "%v" -> "%v"`, t.Operation, string(t.Key), string(t.Value)) } From 7cd1033c8005744f63379f040cc58d0f88775f30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Mon, 14 Mar 2022 14:50:45 +0100 Subject: [PATCH 18/19] chore(tracelistener): forgot to rename variables Whoops! --- tracelistener/trace.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tracelistener/trace.go b/tracelistener/trace.go index d73b5737..3246f093 100644 --- a/tracelistener/trace.go +++ b/tracelistener/trace.go @@ -19,7 +19,7 @@ func (to *TraceOperation) Reset() { } func (to TraceOperation) String() string { - return fmt.Sprintf(`[%s] "%v" -> "%v"`, t.Operation, string(t.Key), string(t.Value)) + return fmt.Sprintf(`[%s] "%v" -> "%v"`, to.Operation, string(to.Key), string(to.Value)) } type TraceMetadata struct { From 80d1858b7135ba39f11814d8675b877fa30c38ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gianguido=20Sor=C3=A0?= Date: Tue, 15 Mar 2022 10:28:30 +0100 Subject: [PATCH 19/19] fix(tracelistener): address struct copying concern Also added missing godoc strings, removed `TraceOperation.String()` method, it wasn't needed anymore. --- tracelistener/trace.go | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/tracelistener/trace.go b/tracelistener/trace.go index 3246f093..5139b3d9 100644 --- a/tracelistener/trace.go +++ b/tracelistener/trace.go @@ -1,14 +1,23 @@ package tracelistener -import ( - "fmt" -) - +// Copy deep-copies to to a new instances of TraceOperation, +// useful when sending over data down the processing pipeline. func (to *TraceOperation) Copy() TraceOperation { ret := *to + + // Explicitly copy key and value slices to new + // slice instances to avoid aliasing. + ret.Key = make([]byte, len(to.Key)) + copy(ret.Key, to.Key) + + ret.Value = make([]byte, len(to.Value)) + copy(ret.Value, to.Value) + return ret } +// Reset resets to to an empty state. +// Useful when storing it in a sync.Pool. func (to *TraceOperation) Reset() { to.Operation = "" to.Key = to.Key[:0] @@ -18,15 +27,16 @@ func (to *TraceOperation) Reset() { to.SuggestedProcessor = "" } -func (to TraceOperation) String() string { - return fmt.Sprintf(`[%s] "%v" -> "%v"`, to.Operation, string(to.Key), string(to.Value)) -} - +// TraceMetadata holds circumstantial information about a trace, +// like the block height at which it was generated, and optionally a +// the block hash that generated it. type TraceMetadata struct { BlockHeight uint64 `json:"blockHeight"` TxHash string `json:"txHash"` } +// TraceOperation represents a Cosmos SDK store operation, parsed from +// JSON lines produced by the SDK's "--trace-store" CLI flag. type TraceOperation struct { Operation string `json:"operation"` Key []byte `json:"key"`