implement debug log

blockstore/splitstore/debug.go

@@ -1,6 +1,17 @@
 package splitstore
 
 import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime/debug"
+	"sync"
+	"time"
+
+	"go.uber.org/multierr"
+	"golang.org/x/xerrors"
+
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/chain/types"
 
@@ -9,14 +20,72 @@ import (
 )
 
 type debugLog struct {
+	basePath                      string
+	readPath, writePath, movePath string
+	readMx, writeMx, moveMx       sync.Mutex
+	readLog, writeLog, moveLog    *os.File
+	readCnt, writeCnt, moveCnt    int
 }
 
-func (d *debugLog) LogReadMiss(cid cid.Cid) {
+func openDebugLog(path string) (*debugLog, error) {
+	basePath := filepath.Join(path, "debug")
+	err := os.MkdirAll(basePath, 0755)
+	if err != nil {
+		return nil, xerrors.Errorf("error creating debug log directory: %w", err)
+	}
+
+	readPath := filepath.Join(basePath, "read.log")
+	readFile, err := os.OpenFile(readPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
+	if err != nil {
+		return nil, xerrors.Errorf("error opening read log: %w", err)
+	}
+
+	writePath := filepath.Join(basePath, "write.log")
+	writeFile, err := os.OpenFile(writePath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
+	if err != nil {
+		_ = readFile.Close()
+		return nil, xerrors.Errorf("error opening write log: %w", err)
+	}
+
+	movePath := filepath.Join(basePath, "move.log")
+	moveFile, err := os.OpenFile(movePath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
+	if err != nil {
+		_ = readFile.Close()
+		_ = writeFile.Close()
+		return nil, xerrors.Errorf("error opening move log: %w", err)
+	}
+
+	return &debugLog{
+		basePath:  basePath,
+		readPath:  readPath,
+		writePath: writePath,
+		movePath:  movePath,
+		readLog:   readFile,
+		writeLog:  writeFile,
+		moveLog:   moveFile,
+	}, nil
+}
+
+func (d *debugLog) LogReadMiss(curTs *types.TipSet, cid cid.Cid) {
 	if d == nil {
 		return
 	}
 
-	// TODO
+	stack := debug.Stack()
+	var epoch abi.ChainEpoch
+	if curTs != nil {
+		epoch = curTs.Height()
+	}
+
+	d.readMx.Lock()
+	defer d.readMx.Unlock()
+
+	d.readCnt++
+
+	_, err := fmt.Fprintf(d.readLog, "%d %s\n%s\n", epoch, cid, string(stack))
+	if err != nil {
+		log.Warnf("error writing read log: %s", err)
+	}
 }
 
 func (d *debugLog) LogWrite(curTs *types.TipSet, blk blocks.Block, writeEpoch abi.ChainEpoch) {
@@ -24,7 +93,15 @@ func (d *debugLog) LogWrite(curTs *types.TipSet, blk blocks.Block, writeEpoch ab
 		return
 	}
 
-	// TODO
+	d.writeMx.Lock()
+	defer d.writeMx.Unlock()
+
+	d.writeCnt++
+
+	_, err := fmt.Fprintf(d.writeLog, "%d %s %d\n", curTs.Height(), blk.Cid(), writeEpoch)
+	if err != nil {
+		log.Warnf("error writing write log: %s", err)
+	}
 }
 
 func (d *debugLog) LogWriteMany(curTs *types.TipSet, blks []blocks.Block, writeEpoch abi.ChainEpoch) {
@@ -32,7 +109,18 @@ func (d *debugLog) LogWriteMany(curTs *types.TipSet, blks []blocks.Block, writeE
 		return
 	}
 
-	// TODO
+	d.writeMx.Lock()
+	defer d.writeMx.Unlock()
+
+	d.writeCnt += len(blks)
+
+	for _, blk := range blks {
+		_, err := fmt.Fprintf(d.writeLog, "%d %s %d\n", curTs.Height(), blk.Cid(), writeEpoch)
+		if err != nil {
+			log.Warnf("error writing write log: %s", err)
+			break
+		}
+	}
 }
 
 func (d *debugLog) LogMove(curTs *types.TipSet, cid cid.Cid, writeEpoch abi.ChainEpoch) {
@@ -40,15 +128,106 @@ func (d *debugLog) LogMove(curTs *types.TipSet, cid cid.Cid, writeEpoch abi.Chai
 		return
 	}
 
-	// TODO
+	d.moveMx.Lock()
+	defer d.moveMx.Unlock()
+
+	d.moveCnt++
+
+	_, err := fmt.Fprintf(d.moveLog, "%d %s %d\n", curTs.Height(), cid, writeEpoch)
+	if err != nil {
+		log.Warnf("error writing move log: %s", err)
+	}
 }
 
-func (d *debugLog) FlushMove() {
+func (d *debugLog) Flush() {
 	if d == nil {
 		return
 	}
 
-	// TODO
+	// rotate non-empty logs
+	d.rotateReadLog()
+	d.rotateWriteLog()
+	d.rotateMoveLog()
+}
+
+func (d *debugLog) rotateReadLog() {
+	d.readMx.Lock()
+	defer d.readMx.Unlock()
+
+	if d.readCnt == 0 {
+		return
+	}
+
+	err := d.rotate(d.readLog, d.readPath)
+	if err != nil {
+		log.Errorf("error rotating read log: %s", err)
+	}
+
+	d.readLog, err = os.OpenFile(d.readPath, os.O_WRONLY|os.O_CREATE, 0644)
+	if err != nil {
+		log.Errorf("error opening log file: %s", err)
+	}
+}
+
+func (d *debugLog) rotateWriteLog() {
+	d.writeMx.Lock()
+	defer d.writeMx.Unlock()
+
+	if d.writeCnt == 0 {
+		return
+	}
+
+	err := d.rotate(d.writeLog, d.writePath)
+	if err != nil {
+		log.Errorf("error rotating read log: %s", err)
+	}
+
+	d.writeLog, err = os.OpenFile(d.writePath, os.O_WRONLY|os.O_CREATE, 0644)
+	if err != nil {
+		log.Errorf("error opening log file: %s", err)
+	}
+}
+
+func (d *debugLog) rotateMoveLog() {
+	d.moveMx.Lock()
+	defer d.moveMx.Unlock()
+
+	if d.moveCnt == 0 {
+		return
+	}
+
+	err := d.rotate(d.moveLog, d.movePath)
+	if err != nil {
+		log.Errorf("error rotating read log: %s", err)
+	}
+
+	d.moveLog, err = os.OpenFile(d.movePath, os.O_WRONLY|os.O_CREATE, 0644)
+	if err != nil {
+		log.Errorf("error opening log file: %s", err)
+	}
+}
+
+func (d *debugLog) rotate(f *os.File, path string) error {
+	err := f.Close()
+	if err != nil {
+		return xerrors.Errorf("error closing file: %w", err)
+	}
+
+	arxivPath := fmt.Sprintf("%s-%d", path, time.Now().Unix())
+	err = os.Rename(path, arxivPath)
+	if err != nil {
+		return xerrors.Errorf("error moving file: %w", err)
+	}
+
+	go func() {
+		cmd := exec.Command("gzip", arxivPath)
+		err := cmd.Run()
+		if err != nil {
+			log.Errorf("error compressing log: %s", err)
+		}
+	}()
+
+	return nil
 }
 
 func (d *debugLog) Close() error {
@@ -56,6 +235,17 @@ func (d *debugLog) Close() error {
 		return nil
 	}
 
-	// TODO
+	d.readMx.Lock()
-	return nil
+	err1 := d.readLog.Close()
+	d.readMx.Unlock()
+
+	d.writeMx.Lock()
+	err2 := d.writeLog.Close()
+	d.writeMx.Unlock()
+
+	d.moveMx.Lock()
+	err3 := d.moveLog.Close()
+	d.moveMx.Unlock()
+
+	return multierr.Combine(err1, err2, err3)
 }

blockstore/splitstore/splitstore.go

@@ -74,6 +74,9 @@ var (
 	markSetSizeKey = dstore.NewKey("/splitstore/markSetSize")
 
 	log = logging.Logger("splitstore")
+
+	// set this to true if you are debugging the splitstore to enable debug logging
+	enableDebugLog = false
 )
 
 const (
@@ -173,6 +176,13 @@ func Open(path string, ds dstore.Datastore, hot, cold bstore.Blockstore, cfg *Co
 
 	ss.ctx, ss.cancel = context.WithCancel(context.Background())
 
+	if enableDebugLog {
+		ss.debug, err = openDebugLog(path)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return ss, nil
 }
 
@@ -205,7 +215,10 @@ func (s *SplitStore) Get(cid cid.Cid) (blocks.Block, error) {
 		return blk, nil
 
 	case bstore.ErrNotFound:
-		s.debug.LogReadMiss(cid)
+		s.mx.Lock()
+		curTs := s.curTs
+		s.mx.Unlock()
+		s.debug.LogReadMiss(curTs, cid)
 
 		blk, err = s.cold.Get(cid)
 		if err == nil {
@@ -227,7 +240,10 @@ func (s *SplitStore) GetSize(cid cid.Cid) (int, error) {
 		return size, nil
 
 	case bstore.ErrNotFound:
-		s.debug.LogReadMiss(cid)
+		s.mx.Lock()
+		curTs := s.curTs
+		s.mx.Unlock()
+		s.debug.LogReadMiss(curTs, cid)
 
 		size, err = s.cold.GetSize(cid)
 		if err == nil {
@@ -332,7 +348,10 @@ func (s *SplitStore) View(cid cid.Cid, cb func([]byte) error) error {
 	err := s.hot.View(cid, cb)
 	switch err {
 	case bstore.ErrNotFound:
-		s.debug.LogReadMiss(cid)
+		s.mx.Lock()
+		curTs := s.curTs
+		s.mx.Unlock()
+		s.debug.LogReadMiss(curTs, cid)
 
 		err = s.cold.View(cid, cb)
 		if err == nil {
@@ -806,7 +825,7 @@ func (s *SplitStore) doCompact(curTs *types.TipSet) error {
 		return nil
 	})
 
-	s.debug.FlushMove()
+	s.debug.Flush()
 
 	if err != nil {
 		return xerrors.Errorf("error collecting cold objects: %w", err)