splitstore shed utils

api/api_full.go

@@ -164,6 +164,13 @@ type FullNode interface {
 	// If oldmsgskip is set, messages from before the requested roots are also not included.
 	ChainExport(ctx context.Context, nroots abi.ChainEpoch, oldmsgskip bool, tsk types.TipSetKey) (<-chan []byte, error) //perm:read
 
+	// ChainCheckBlockstore performs an (asynchronous) health check on the chain/state blockstore
+	// if supported by the underlying implementation.
+	ChainCheckBlockstore(context.Context) error //perm:read
+
+	// ChainBlockstoreInfo returns some basic information about the blockstore
+	ChainBlockstoreInfo(context.Context) (map[string]interface{}, error) //perm:read
+
 	// MethodGroup: Beacon
 	// The Beacon method group contains methods for interacting with the random beacon (DRAND)
 

blockstore/badger/blockstore.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
+	"path/filepath"
 	"runtime"
 	"sync"
 
@@ -84,7 +86,8 @@ type Blockstore struct {
 	state   int
 	viewers sync.WaitGroup
 
-	DB *badger.DB
+	DB   *badger.DB
+	opts Options
 
 	prefixing bool
 	prefix    []byte
@@ -95,6 +98,7 @@ var _ blockstore.Blockstore = (*Blockstore)(nil)
 var _ blockstore.Viewer = (*Blockstore)(nil)
 var _ blockstore.BlockstoreIterator = (*Blockstore)(nil)
 var _ blockstore.BlockstoreGC = (*Blockstore)(nil)
+var _ blockstore.BlockstoreSize = (*Blockstore)(nil)
 var _ io.Closer = (*Blockstore)(nil)
 
 // Open creates a new badger-backed blockstore, with the supplied options.
@@ -109,7 +113,7 @@ func Open(opts Options) (*Blockstore, error) {
 		return nil, fmt.Errorf("failed to open badger blockstore: %w", err)
 	}
 
-	bs := &Blockstore{DB: db}
+	bs := &Blockstore{DB: db, opts: opts}
 	if p := opts.Prefix; p != "" {
 		bs.prefixing = true
 		bs.prefix = []byte(p)
@@ -191,6 +195,37 @@ func (b *Blockstore) CollectGarbage() error {
 	return err
 }
 
+// Size returns the aggregate size of the blockstore
+func (b *Blockstore) Size() (int64, error) {
+	if err := b.access(); err != nil {
+		return 0, err
+	}
+	defer b.viewers.Done()
+
+	lsm, vlog := b.DB.Size()
+	size := lsm + vlog
+
+	if size == 0 {
+		// badger reports a 0 size on symlinked directories... sigh
+		dir := b.opts.Dir
+		entries, err := os.ReadDir(dir)
+		if err != nil {
+			return 0, err
+		}
+
+		for _, e := range entries {
+			path := filepath.Join(dir, e.Name())
+			finfo, err := os.Stat(path)
+			if err != nil {
+				return 0, err
+			}
+			size += finfo.Size()
+		}
+	}
+
+	return size, nil
+}
+
 // View implements blockstore.Viewer, which leverages zero-copy read-only
 // access to values.
 func (b *Blockstore) View(cid cid.Cid, fn func([]byte) error) error {

blockstore/blockstore.go

@@ -40,6 +40,11 @@ type BlockstoreGC interface {
 	CollectGarbage() error
 }
 
+// BlockstoreSize is a trait for on-disk blockstores that can report their size
+type BlockstoreSize interface {
+	Size() (int64, error)
+}
+
 // WrapIDStore wraps the underlying blockstore in an "identity" blockstore.
 // The ID store filters out all puts for blocks with CIDs using the "identity"
 // hash function. It also extracts inlined blocks from CIDs using the identity

blockstore/splitstore/README.md

@@ -99,3 +99,17 @@ Compaction works transactionally with the following algorithm:
 ## Garbage Collection
 
 TBD -- see [#6577](https://github.com/filecoin-project/lotus/issues/6577)
+
+## Utilities
+
+`lotus-shed` has a `splitstore` command which provides some utilities:
+
+- `rollback` -- rolls back a splitstore installation.
+  This command copies the hotstore on top of the coldstore, and then deletes the splitstore
+  directory and associated metadata keys.
+  It can also optionally compact/gc the coldstore after the copy (with the `--gc-coldstore` flag)
+  and automatically rewrite the lotus config to disable splitstore (with the `--rewrite-config` flag).
+  Note: the node *must be stopped* before running this command.
+- `check` -- asynchronously runs a basic healthcheck on the splitstore.
+  The results are appended to `<lotus-repo>/datastore/splitstore/check.txt`.
+- `info` -- prints some basic information about the splitstore.

blockstore/splitstore/splitstore.go

@@ -102,7 +102,8 @@ type SplitStore struct {
 	compacting int32 // compaction/prune/warmup in progress
 	closing    int32 // the splitstore is closing
 
-	cfg *Config
+	cfg  *Config
+	path string
 
 	mx          sync.Mutex
 	warmupEpoch abi.ChainEpoch // protected by mx
@@ -169,6 +170,7 @@ func Open(path string, ds dstore.Datastore, hot, cold bstore.Blockstore, cfg *Co
 	// and now we can make a SplitStore
 	ss := &SplitStore{
 		cfg:        cfg,
+		path:       path,
 		ds:         ds,
 		cold:       cold,
 		hot:        hots,

blockstore/splitstore/splitstore_check.go

@@ -0,0 +1,150 @@
+package splitstore
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/xerrors"
+
+	cid "github.com/ipfs/go-cid"
+
+	bstore "github.com/filecoin-project/lotus/blockstore"
+	"github.com/filecoin-project/lotus/chain/types"
+)
+
+// performs an asynchronous health-check on the splitstore; results are appended to
+// <splitstore-path>/check.txt
+func (s *SplitStore) Check() error {
+	s.headChangeMx.Lock()
+	defer s.headChangeMx.Unlock()
+
+	// try to take compaction lock and inhibit compaction while the health-check is running
+	if !atomic.CompareAndSwapInt32(&s.compacting, 0, 1) {
+		return xerrors.Errorf("can't acquire compaction lock; compacting operation in progress")
+	}
+
+	if s.compactionIndex == 0 {
+		atomic.StoreInt32(&s.compacting, 0)
+		return xerrors.Errorf("splitstore hasn't compacted yet; health check is not meaningful")
+	}
+
+	// check if we are actually closing first
+	if err := s.checkClosing(); err != nil {
+		atomic.StoreInt32(&s.compacting, 0)
+		return err
+	}
+
+	curTs := s.chain.GetHeaviestTipSet()
+	go func() {
+		defer atomic.StoreInt32(&s.compacting, 0)
+
+		log.Info("checking splitstore health")
+		start := time.Now()
+
+		err := s.doCheck(curTs)
+		if err != nil {
+			log.Errorf("error checking splitstore health: %s", err)
+			return
+		}
+
+		log.Infow("health check done", "took", time.Since(start))
+	}()
+
+	return nil
+}
+
+func (s *SplitStore) doCheck(curTs *types.TipSet) error {
+	currentEpoch := curTs.Height()
+	boundaryEpoch := currentEpoch - CompactionBoundary
+
+	outputPath := filepath.Join(s.path, "check.txt")
+	output, err := os.OpenFile(outputPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
+	if err != nil {
+		return xerrors.Errorf("error opening check output file %s: %w", outputPath, err)
+	}
+	defer output.Close() //nolint:errcheck
+
+	write := func(format string, args ...interface{}) {
+		_, err := fmt.Fprintf(output, format+"\n", args...)
+		if err != nil {
+			log.Warnf("error writing check output: %s", err)
+		}
+	}
+
+	ts, _ := time.Now().MarshalText()
+	write("---------------------------------------------")
+	write("start check at %s", ts)
+	write("current epoch: %d", currentEpoch)
+	write("boundary epoch: %d", boundaryEpoch)
+	write("compaction index: %d", s.compactionIndex)
+	write("--")
+
+	var coldCnt, missingCnt int64
+	err = s.walkChain(curTs, boundaryEpoch, boundaryEpoch,
+		func(c cid.Cid) error {
+			if isUnitaryObject(c) {
+				return errStopWalk
+			}
+
+			has, err := s.hot.Has(c)
+			if err != nil {
+				return xerrors.Errorf("error checking hotstore: %w", err)
+			}
+
+			if has {
+				return nil
+			}
+
+			has, err = s.cold.Has(c)
+			if err != nil {
+				return xerrors.Errorf("error checking coldstore: %w", err)
+			}
+
+			if has {
+				coldCnt++
+				write("cold object reference: %s", c)
+			} else {
+				missingCnt++
+				write("missing object reference: %s", c)
+				return errStopWalk
+			}
+
+			return nil
+		})
+
+	if err != nil {
+		err = xerrors.Errorf("error walking chain: %w", err)
+		write("ERROR: %s", err)
+		return err
+	}
+
+	log.Infow("check done", "cold", coldCnt, "missing", missingCnt)
+	write("--")
+	write("cold: %d missing: %d", coldCnt, missingCnt)
+	write("DONE")
+
+	return nil
+}
+
+// provides some basic information about the splitstore
+func (s *SplitStore) Info() map[string]interface{} {
+	info := make(map[string]interface{})
+	info["base epoch"] = s.baseEpoch
+	info["warmup epoch"] = s.warmupEpoch
+	info["compactions"] = s.compactionIndex
+
+	sizer, ok := s.hot.(bstore.BlockstoreSize)
+	if ok {
+		size, err := sizer.Size()
+		if err != nil {
+			log.Warnf("error getting hotstore size: %s", err)
+		} else {
+			info["hotstore size"] = size
+		}
+	}
+
+	return info
+}

cmd/lotus-shed/main.go

@@ -60,6 +60,7 @@ func main() {
 		actorCmd,
 		minerTypesCmd,
 		minerMultisigsCmd,
+		splitstoreCmd,
 	}
 
 	app := &cli.App{