engine: allow taking RocksDB CheckPoint

We've repeatedly wanted this to preserve state when finding replica
inconsistencies.

See, for example:

#36861

Release note: None

c-deps/libroach/db.cc

@@ -18,6 +18,7 @@
 #include <rocksdb/perf_context.h>
 #include <rocksdb/sst_file_writer.h>
 #include <rocksdb/table.h>
+#include <rocksdb/utilities/checkpoint.h>
 #include <stdarg.h>
 #include "batch.h"
 #include "cache.h"
@@ -257,6 +258,22 @@ DBStatus DBOpen(DBEngine** db, DBSlice dir, DBOptions db_opts) {
   return kSuccess;
 }
 
+DBStatus DBCreateCheckpoint(DBEngine* db, DBSlice dir) {
+  const std::string cp_dir = ToString(dir);
+
+  rocksdb::Checkpoint* cp_ptr;
+  auto status = rocksdb::Checkpoint::Create(db->rep, &cp_ptr);
+  if (!status.ok()) {
+    return ToDBStatus(status);
+  }
+  // NB: passing 0 for log_size_for_flush forces a WAL sync, i.e. makes sure
+  // that the checkpoint is up to date.
+  status = cp_ptr->CreateCheckpoint(cp_dir, 0 /* log_size_for_flush */);
+  delete(cp_ptr);
+  return ToDBStatus(status);
+}
+
+
 DBStatus DBDestroy(DBSlice dir) {
   rocksdb::Options options;
   return ToDBStatus(rocksdb::DestroyDB(ToString(dir), options));

c-deps/libroach/include/libroach.h

@@ -100,6 +100,12 @@ void DBReleaseCache(DBCache* cache);
 // exist.
 DBStatus DBOpen(DBEngine** db, DBSlice dir, DBOptions options);
 
+// Creates a RocksDB checkpoint in the specified directory (which must not exist).
+// A checkpoint is a logical copy of the database, though it will hardlink the
+// SSTs references by it (when possible), thus avoiding duplication of any of
+// the actual data.
+DBStatus DBCreateCheckpoint(DBEngine* db, DBSlice dir);
+
 // Set a callback to be invoked during DBOpen that can make changes to RocksDB
 // initialization. Used by CCL code to install additional features.
 //

pkg/storage/engine/engine.go

@@ -333,6 +333,10 @@ type Engine interface {
 	// the engine implementation. For RocksDB, this means using the Env responsible for the file
 	// which may handle extra logic (eg: copy encryption settings for EncryptedEnv).
 	LinkFile(oldname, newname string) error
+	// CreateCheckpoint creates a checkpoint of the engine in the given directory,
+	// which must not exist. The directory should be on the same file system so
+	// that hard links can be used.
+	CreateCheckpoint(dir string) error
 }
 
 // MapProvidingEngine is an Engine that also provides facilities for making a

pkg/storage/engine/engine_test.go

@@ -18,18 +18,22 @@ import (
 	"bytes"
 	"context"
 	"math/rand"
+	"path/filepath"
 	"reflect"
 	"sort"
 	"strconv"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
 )
 
 func ensureRangeEqual(
@@ -790,3 +794,36 @@ func insertKeysAndValues(keys []MVCCKey, values [][]byte, engine Engine, t *test
 		}
 	}
 }
+
+func TestCreateCheckpoint(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	dir, cleanup := testutils.TempDir(t)
+	defer cleanup()
+
+	// TODO(tbg): use the method below, but only for on-disk stores.
+	_ = runWithAllEngines
+
+	rocksDB, err := NewRocksDB(
+		RocksDBConfig{
+			Settings: cluster.MakeTestingClusterSettings(),
+			Dir:      dir,
+		},
+		RocksDBCache{},
+	)
+
+	db := Engine(rocksDB) // be impl neutral from now on
+	defer db.Close()
+
+	dir = filepath.Join(dir, "checkpoint")
+
+	assert.NoError(t, err)
+	assert.NoError(t, db.CreateCheckpoint(dir))
+	assert.DirExists(t, dir)
+	m, err := filepath.Glob(dir + "/*")
+	assert.NoError(t, err)
+	assert.True(t, len(m) > 0)
+	if err := db.CreateCheckpoint(dir); !testutils.IsError(err, "exists") {
+		t.Fatal(err)
+	}
+}

pkg/storage/engine/rocksdb.go

@@ -765,6 +765,14 @@ func (r *RocksDB) Close() {
 	r.syncer.Unlock()
 }
 
+// CreateCheckpoint creates a RocksDB checkpoint in the given directory (which
+// must not exist). This directory should be located on the same file system, or
+// copies of all data are used instead of hard links, which is very expensive.
+func (r *RocksDB) CreateCheckpoint(dir string) error {
+	status := C.DBCreateCheckpoint(r.rdb, goToCSlice([]byte(dir)))
+	return errors.Wrap(statusToError(status), "unable to take RocksDB checkpoint")
+}
+
 // Closed returns true if the engine is closed.
 func (r *RocksDB) Closed() bool {
 	return r.rdb == nil