diff --git a/baseapp/options.go b/baseapp/options.go index 0401aa84d5f9..1f809e498b9c 100644 --- a/baseapp/options.go +++ b/baseapp/options.go @@ -7,12 +7,10 @@ import ( dbm "github.com/cosmos/cosmos-db" - "cosmossdk.io/log" "cosmossdk.io/store/metrics" pruningtypes "cosmossdk.io/store/pruning/types" "cosmossdk.io/store/snapshots" snapshottypes "cosmossdk.io/store/snapshots/types" - "cosmossdk.io/store/tracekv" storetypes "cosmossdk.io/store/types" "github.com/cosmos/cosmos-sdk/baseapp/oe" @@ -275,15 +273,8 @@ func (app *BaseApp) SetNotSigverifyTx() { } // SetCommitMultiStoreTracer sets the store tracer on the BaseApp's underlying -// CommitMultiStore. If w is nil and the logger is a MemLogger, a TraceWriter -// is automatically created to enable store tracing for the MemLogger. +// CommitMultiStore. func (app *BaseApp) SetCommitMultiStoreTracer(w io.Writer) { - if w == nil { - if _, ok := app.logger.Impl().(*log.MemLogger); ok { - app.logger.Info("Auto-enabled store tracing for MemLogger") - w = tracekv.NewTraceWriter(app.logger) - } - } app.cms.SetTracer(w) } diff --git a/log/mem_filter.go b/log/mem_filter.go index bd6692d9310c..77a6fcae632e 100644 --- a/log/mem_filter.go +++ b/log/mem_filter.go @@ -73,7 +73,6 @@ func storageMsgs() []string { "tx result detail", "hash of tx results", "finalized block", - "store trace set", "store change set", "CONSENSUS FAILURE!!!", } diff --git a/server/start.go b/server/start.go index 71c61e2cd4cd..b85608c0005d 100644 --- a/server/start.go +++ b/server/start.go @@ -436,7 +436,16 @@ func setupTraceWriter(svrCtx *Context) (traceWriter io.WriteCloser, cleanup func switch { case svrCtx.Viper.GetBool(FlagMemLogEnabled): - traceWriter = tracekv.NewTraceWriter(svrCtx.Logger) + baseDir := svrCtx.Viper.GetString(FlagMemLogOutputDir) + if baseDir == "" { + baseDir = svrCtx.Config.RootDir + } + traceDir := filepath.Join(baseDir, "traces") + maxBytes := svrCtx.Viper.GetInt(FlagMemLogMemoryLimit) + traceWriter, err = tracekv.NewTraceFileWriterWithSize(traceDir, maxBytes) + if err != nil { + return nil, cleanup, err + } case svrCtx.Viper.GetString(flagTraceStore) != "": traceWriterFile := svrCtx.Viper.GetString(flagTraceStore) traceWriter, err = openTraceWriter(traceWriterFile) diff --git a/store/tracekv/mem_trace_writer.go b/store/tracekv/mem_trace_writer.go new file mode 100644 index 000000000000..ebd6f9992e82 --- /dev/null +++ b/store/tracekv/mem_trace_writer.go @@ -0,0 +1,238 @@ +package tracekv + +import ( + "bytes" + "compress/gzip" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "sync" +) + +const ( + // DefaultMaxBytes is the maximum buffer size before flushing (1MB). + DefaultMaxBytes = 1 << 20 +) + +var ( + _ io.Writer = (*MemTraceWriter)(nil) + _ io.Closer = (*MemTraceWriter)(nil) +) + +// storeTraceSet is the bundled format expected by cosmos-analyzer. +type storeTraceSet struct { + Msg string `json:"_msg"` + Height int64 `json:"height"` + Count int `json:"count"` + Traces []json.RawMessage `json:"traces"` +} + +// traceWorkItem bundles trace data for async compression. +type traceWorkItem struct { + opsByHeight map[int64][]json.RawMessage + minHeight int64 + maxHeight int64 +} + +// MemTraceWriter writes trace operations to range-based gzipped files. +// Multiple block heights are buffered and written to a single file named +// trace-{minHeight}-{maxHeight}.gz when the buffer reaches the size threshold. +// Compression and file I/O are performed asynchronously in a background goroutine. +type MemTraceWriter struct { + dir string + opsByHeight map[int64][]json.RawMessage + minHeight int64 + maxHeight int64 + currentSize int + maxBytes int + mu sync.Mutex + + // async compression + workCh chan traceWorkItem + wg sync.WaitGroup +} + +// NewTraceFileWriter creates a new MemTraceWriter that writes to the given directory. +func NewTraceFileWriter(dir string) (*MemTraceWriter, error) { + return NewTraceFileWriterWithSize(dir, DefaultMaxBytes) +} + +// NewTraceFileWriterWithSize creates a MemTraceWriter with a custom buffer size. +func NewTraceFileWriterWithSize(dir string, maxBytes int) (*MemTraceWriter, error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("create trace dir: %w", err) + } + if maxBytes <= 0 { + maxBytes = DefaultMaxBytes + } + w := &MemTraceWriter{ + dir: dir, + opsByHeight: make(map[int64][]json.RawMessage), + maxBytes: maxBytes, + workCh: make(chan traceWorkItem, 64), + } + w.wg.Add(1) + go w.compressor() + return w, nil +} + +// Write implements io.Writer. It buffers the trace operation for batch writing. +func (w *MemTraceWriter) Write(p []byte) (n int, err error) { + w.mu.Lock() + + height := extractHeightFromOp(p) + if height == 0 { + w.mu.Unlock() + return len(p), nil + } + + data := make([]byte, len(p)) + copy(data, p) + + if w.minHeight == 0 || height < w.minHeight { + w.minHeight = height + } + if height > w.maxHeight { + w.maxHeight = height + } + + w.opsByHeight[height] = append(w.opsByHeight[height], json.RawMessage(data)) + w.currentSize += len(data) + + var wi traceWorkItem + if w.currentSize >= w.maxBytes { + wi = w.takeBufferLocked() + } + w.mu.Unlock() + + if wi.opsByHeight != nil { + w.workCh <- wi + } + + return len(p), nil +} + +// takeBufferLocked swaps out the current buffer and returns it. Caller must hold w.mu. +func (w *MemTraceWriter) takeBufferLocked() traceWorkItem { + wi := traceWorkItem{ + opsByHeight: w.opsByHeight, + minHeight: w.minHeight, + maxHeight: w.maxHeight, + } + w.opsByHeight = make(map[int64][]json.RawMessage) + w.minHeight = 0 + w.maxHeight = 0 + w.currentSize = 0 + return wi +} + +// compressor is the background goroutine that processes work items. +func (w *MemTraceWriter) compressor() { + defer w.wg.Done() + for wi := range w.workCh { + w.writeTraceFile(wi) + } +} + +// writeTraceFile compresses and writes a work item to disk. +func (w *MemTraceWriter) writeTraceFile(wi traceWorkItem) { + if len(wi.opsByHeight) == 0 { + return + } + + filename := filepath.Join(w.dir, fmt.Sprintf("trace-%d-%d.gz", wi.minHeight, wi.maxHeight)) + f, err := os.Create(filename) + if err != nil { + return + } + defer f.Close() + + gw := gzip.NewWriter(f) + defer gw.Close() + + for height := wi.minHeight; height <= wi.maxHeight; height++ { + ops, ok := wi.opsByHeight[height] + if !ok { + continue + } + + traceSet := storeTraceSet{ + Msg: "store trace set", + Height: height, + Count: len(ops), + Traces: ops, + } + + line, err := json.Marshal(traceSet) + if err != nil { + continue + } + gw.Write(line) + gw.Write([]byte("\n")) + } +} + +var blockHeightKey = []byte(`"blockHeight":`) + +func extractHeightFromOp(data []byte) int64 { + idx := bytes.Index(data, blockHeightKey) + if idx == -1 { + return 0 + } + + start := idx + len(blockHeightKey) + for start < len(data) && (data[start] == ' ' || data[start] == '\t') { + start++ + } + + end := start + for end < len(data) && data[end] >= '0' && data[end] <= '9' { + end++ + } + + if start == end { + return 0 + } + + height, _ := strconv.ParseInt(string(data[start:end]), 10, 64) + return height +} + +// Flush synchronously writes any pending buffer to disk. +func (w *MemTraceWriter) Flush() { + w.mu.Lock() + if len(w.opsByHeight) == 0 { + w.mu.Unlock() + return + } + wi := w.takeBufferLocked() + w.mu.Unlock() + + // Write synchronously for explicit Flush calls + w.writeTraceFile(wi) +} + +// Close stops the background compressor and flushes any remaining data. +func (w *MemTraceWriter) Close() error { + // Enqueue any remaining buffer + w.mu.Lock() + if len(w.opsByHeight) > 0 { + wi := w.takeBufferLocked() + w.mu.Unlock() + w.workCh <- wi + } else { + w.mu.Unlock() + } + + // Stop compressor and wait + close(w.workCh) + w.wg.Wait() + return nil +} + +func (w *MemTraceWriter) Dir() string { + return w.dir +} diff --git a/store/tracekv/writer.go b/store/tracekv/writer.go deleted file mode 100644 index 4f2848a6ae93..000000000000 --- a/store/tracekv/writer.go +++ /dev/null @@ -1,55 +0,0 @@ -package tracekv - -import ( - "bytes" - "encoding/json" - "io" - - "cosmossdk.io/log" -) - -var ( - _ io.Writer = (*TraceWriter)(nil) - _ io.Closer = (*TraceWriter)(nil) -) - -// TraceWriter implements io.Writer and buffers trace operations. -// It logs all buffered operations via logger.Debug() when Flush() is called. -type TraceWriter struct { - logger log.Logger - ops []json.RawMessage -} - -// NewTraceWriter creates a new TraceWriter with the given logger. -func NewTraceWriter(logger log.Logger) *TraceWriter { - return &TraceWriter{logger: logger} -} - -// Write implements io.Writer. It buffers the trace operation (JSON line) for batch logging. -func (w *TraceWriter) Write(p []byte) (n int, err error) { - // Remove trailing newline if present - data := bytes.TrimSuffix(p, []byte("\n")) - if len(data) > 0 { - w.ops = append(w.ops, json.RawMessage(data)) - } - return len(p), nil -} - -// Flush writes all buffered trace operations to the logger and clears the buffer. -func (w *TraceWriter) Flush() { - if w.logger == nil || len(w.ops) == 0 { - return - } - - w.logger.Debug("store trace set", - "count", len(w.ops), - "traces", w.ops, - ) - w.ops = w.ops[:0] -} - -// Close implements io.Closer. It flushes any remaining buffered operations. -func (w *TraceWriter) Close() error { - w.Flush() - return nil -}