storage: use engine for os-level operations.

This is important when encryption is enabled as we need to make sure we
handle file metadata properly.
While `os.Remove` will leave cruft behind, `os.Link` won't copy the file
encryption settings from the original and will therefore make it
unreadable.

This fixes `RESTORE` which writes file to local disk using local
encryption settings then ingests them. The link bypassed the rocksdb Env
leaving the ingested file without any encryption settings attached (aka:
plaintext).

Release note: None

Author: marc

c-deps/libroach/batch.cc

@@ -535,6 +535,8 @@ DBStatus DBBatch::EnvDeleteFile(DBSlice path) { return FmtStatus("unsupported");
 
 DBStatus DBBatch::EnvDeleteDirAndFiles(DBSlice dir) { return FmtStatus("unsupported"); }
 
+DBStatus DBBatch::EnvLinkFile(DBSlice oldname, DBSlice newname) { return FmtStatus("unsupported"); }
+
 DBWriteOnlyBatch::DBWriteOnlyBatch(DBEngine* db) : DBEngine(db->rep, db->iters), updates(0) {}
 
 DBWriteOnlyBatch::~DBWriteOnlyBatch() {}
@@ -628,6 +630,8 @@ DBStatus DBWriteOnlyBatch::EnvDeleteFile(DBSlice path) { return FmtStatus("unsup
 
 DBStatus DBWriteOnlyBatch::EnvDeleteDirAndFiles(DBSlice dir) { return FmtStatus("unsupported"); }
 
+DBStatus DBWriteOnlyBatch::EnvLinkFile(DBSlice oldname, DBSlice newname) { return FmtStatus("unsupported"); }
+
 rocksdb::WriteBatch::Handler* GetDBBatchInserter(::rocksdb::WriteBatchBase* batch) {
   return new DBBatchInserter(batch);
 }

c-deps/libroach/batch.h

@@ -49,6 +49,7 @@ struct DBBatch : public DBEngine {
   virtual DBStatus EnvCloseFile(rocksdb::WritableFile* file);
   virtual DBStatus EnvDeleteFile(DBSlice path);
   virtual DBStatus EnvDeleteDirAndFiles(DBSlice dir);
+  virtual DBStatus EnvLinkFile(DBSlice oldname, DBSlice newname);
 };
 
 struct DBWriteOnlyBatch : public DBEngine {
@@ -78,6 +79,7 @@ struct DBWriteOnlyBatch : public DBEngine {
   virtual DBStatus EnvCloseFile(rocksdb::WritableFile* file);
   virtual DBStatus EnvDeleteFile(DBSlice path);
   virtual DBStatus EnvDeleteDirAndFiles(DBSlice dir);
+  virtual DBStatus EnvLinkFile(DBSlice oldname, DBSlice newname);
 };
 
 // GetDBBatchInserter returns a WriteBatch::Handler that operates on a

c-deps/libroach/db.cc

@@ -477,6 +477,8 @@ DBStatus DBEnvDeleteFile(DBEngine* db, DBSlice path) { return db->EnvDeleteFile(
 
 DBStatus DBEnvDeleteDirAndFiles(DBEngine* db, DBSlice dir) { return db->EnvDeleteDirAndFiles(dir); }
 
+DBStatus DBEnvLinkFile(DBEngine* db, DBSlice oldname, DBSlice newname) { return db->EnvLinkFile(oldname, newname); }
+
 DBIterator* DBNewIter(DBEngine* db, bool prefix, bool stats) {
   rocksdb::ReadOptions opts;
   opts.prefix_same_as_start = prefix;

c-deps/libroach/engine.cc

@@ -435,4 +435,9 @@ DBStatus DBImpl::EnvDeleteDirAndFiles(DBSlice dir) {
   return ToDBStatus(status);
 }
 
+// EnvLinkFile creates 'newname' as a hard link to 'oldname'.
+DBStatus DBImpl::EnvLinkFile(DBSlice oldname, DBSlice newname) {
+  return ToDBStatus(this->rep->GetEnv()->LinkFile(ToString(oldname), ToString(newname)));
+}
+
 }  // namespace cockroach

c-deps/libroach/engine.h

@@ -50,6 +50,7 @@ struct DBEngine {
   virtual DBStatus EnvCloseFile(rocksdb::WritableFile* file) = 0;
   virtual DBStatus EnvDeleteFile(DBSlice path) = 0;
   virtual DBStatus EnvDeleteDirAndFiles(DBSlice dir) = 0;
+  virtual DBStatus EnvLinkFile(DBSlice oldname, DBSlice newname) = 0;
 
   DBSSTable* GetSSTables(int* n);
   DBString GetUserProperties();
@@ -94,6 +95,7 @@ struct DBImpl : public DBEngine {
   virtual DBStatus EnvCloseFile(rocksdb::WritableFile* file);
   virtual DBStatus EnvDeleteFile(DBSlice path);
   virtual DBStatus EnvDeleteDirAndFiles(DBSlice dir);
+  virtual DBStatus EnvLinkFile(DBSlice oldname, DBSlice newname);
 };
 
 }  // namespace cockroach

c-deps/libroach/include/libroach.h

@@ -402,6 +402,9 @@ DBStatus DBEnvDeleteFile(DBEngine* db, DBSlice path);
 // files it contains but not subdirectories in the given engine.
 DBStatus DBEnvDeleteDirAndFiles(DBEngine* db, DBSlice dir);
 
+// DBEnvLinkFile creates 'newname' as a hard link to 'oldname using the given engine.
+DBStatus DBEnvLinkFile(DBEngine* db, DBSlice oldname, DBSlice newname);
+
 // DBFileLock contains various parameters set during DBLockFile and required for DBUnlockFile.
 typedef void* DBFileLock;
 

c-deps/libroach/snapshot.cc

@@ -79,4 +79,6 @@ DBStatus DBSnapshot::EnvDeleteFile(DBSlice path) { return FmtStatus("unsupported
 
 DBStatus DBSnapshot::EnvDeleteDirAndFiles(DBSlice dir) { return FmtStatus("unsupported"); }
 
+DBStatus DBSnapshot::EnvLinkFile(DBSlice oldname, DBSlice newname) { return FmtStatus("unsupported"); }
+
 }  // namespace cockroach

c-deps/libroach/snapshot.h

@@ -46,6 +46,7 @@ struct DBSnapshot : public DBEngine {
   virtual DBStatus EnvCloseFile(rocksdb::WritableFile* file);
   virtual DBStatus EnvDeleteFile(DBSlice path);
   virtual DBStatus EnvDeleteDirAndFiles(DBSlice dir);
+  virtual DBStatus EnvLinkFile(DBSlice oldname, DBSlice newname);
 };
 
 }  // namespace cockroach

pkg/storage/engine/engine.go

@@ -290,6 +290,10 @@ type Engine interface {
 	// not subdirectories from this RocksDB's env. If dir does not exist,
 	// DeleteDirAndFiles returns nil (no error).
 	DeleteDirAndFiles(dir string) error
+	// LinkFile creates 'newname' as a hard link to 'oldname'. This is done using
+	// 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
 }
 
 // WithSSTables extends the Engine interface with a method to get info

pkg/storage/engine/rocksdb.go

@@ -2713,6 +2713,20 @@ func (r *RocksDB) DeleteDirAndFiles(dir string) error {
 	return nil
 }
 
+// LinkFile creates 'newname' as a hard link to 'oldname'. This use the Env responsible for the file
+// which may handle extra logic (eg: copy encryption settings for EncryptedEnv).
+func (r *RocksDB) LinkFile(oldname, newname string) error {
+	if err := statusToError(C.DBEnvLinkFile(r.rdb, goToCSlice([]byte(oldname)), goToCSlice([]byte(newname)))); err != nil {
+		return &os.LinkError{
+			Op:  "link",
+			Old: oldname,
+			New: newname,
+			Err: err,
+		}
+	}
+	return nil
+}
+
 // IsValidSplitKey returns whether the key is a valid split key. Certain key
 // ranges cannot be split (the meta1 span and the system DB span); split keys
 // chosen within any of these ranges are considered invalid. And a split key