storage: use IngestExternalFile instead of WriteBatches for applying snapshots

[petermattis]

Sending a snapshot of a range currently involves generating a series of 256KB WriteBatches which are then applied on the recipient. Per the recent work to use the RocksDB IngestExternalFile functionality for restore, it would likely be significantly faster on the recipient to send an sstable instead. Using IngestExternalFile for applying snapshots should be straightforward in comparison to the usage within restore as we don't have any of the complexity of Raft involved. The sender can create the sstable locally and then stream it to the recipient. And because we send a header before streaming the snapshot, we can probe whether the recipient supports receiving an sstable snapshot which will make the migration story easier.

There is no particular urgency here as applying snapshots is not currently a bottleneck.

@danhhz Do you have benchmark numbers readily available showing the performance difference between IngestExternalFile and applying the equivalent WriteBatches?

[tbg]

One "problem" is that we always put a range tombstone down right before the ingestion, which would force the SSTable into higher layers. That said, it's probably still faster than what we have today.

[bdarnell]

Since the receiving side could stream the sstable straight to disk, we could also avoid holding the entire snapshot in memory at once on the recipient. This would be an important step towards being able to increase the default range size from 64MB.

[tbg]

I looked into this in light of a privately reported user issue (the fundamental problem in which is that we at some point added a check that prevents large snapshots from being sent).

I was initially hopeful that this was fully straightforward, but now it seems that I've written an RFC draft and think this needs some care. This whole thing seems workable though and worth tackling, though it definitely merits some more preliminary discussion here and then an RFC.

Ok, here we go:

The receiving side will always end up with an SSTable on disk which can then be ingested. On the sender, there are two options:

1. construct an SSTable on and stream that over. Pro: less bandwidth used because SSTable blocks can be compressed.
2. stream the keys in some other format and write them into an SSTable at the receiver. Pro: a little more aligned with the current code, and thus a smaller change.

I'll assume 1. because it seems like the right thing to do™. We have to do some experimentation to decide whether we need to preserve both paths, but I hope not. We can always stream an SSTable, but perhaps we a) avoid flushing it to disk unless it grows large enough and b) avoid ingesting it if it's too small, reading it into a WriteBatch instead. At the end of this, we may also consider making the default range size closer to our target L6 SSTable size (128mb?).

We need to extend RocksDBSstFileWriter so that it allows us to access the SSTable as it's being written (right now, it holds it all in memory and only gives it back on Finish()). Instead, we want an API like this:

// Add adds the supplied key-value pair into the SSTable. The returned
// boolean is true if the caller may retrieve the next chunk of the SSTable
// via `Fetch` if they desire to do so.
func (fw *RocksDBSstFileWriter) Add(kv MVCCKeyValue) (bool, error)

// Fetch retrieves the data that has been flushed into the SSTable
// but not retrieved yet (by an earlier invocation of Fetch). By calling `Fetch`
// every time `Add` returns `true`, the caller can retrieve the constructed SSTable
// piecemeal.
func (fw *RocksDBSstFileWriter) Fetch() ([]byte, error)

// FinishAndFetch finalizes the SSTable and calls `Fetch`. `Close` must still
// be called, but the instance must not be used other than that.
func (fw *RocksDBSstFileWriter) FinishAndFetch() ([]byte, error)

// Close idempotently releases the underlying resources attached.
func (fw *RocksDBSstFileWriter) Close()

Assuming we have that, the sending side uses an RocksDBSstFileWriter in its iteration over the user-space keys and streams straight to the connection (keeping the rate limiting intact).

There's a caveat here, which is that we use a ReplicaDataIterator but we really only want to put the main chunk of userspace keys into an SSTable (holes are bad). We should augment the iterator suitably so that we can detect when we're jumping a gap. Other chunks may be large, too, so the most straightforward solution might be to just make the code adaptive from the start: always stream an SSTable, but send a marker RPC every time the chunk changes, so that the recipient can decide whether the current chunk should be ingested or loaded via a WriteBatch.

Log entries deserve a mention. These will have to keep special treatment as log entries will have inlined sideloading payloads which applySnapshot sideloads again, so we don't ever want to ingest these. Log entries are already plenty special, as they're not rate limited on the [sendsnapshot][sender] and could be quite large due to inlined payloads. As a first pass, it seems reasonable to leave them alone.

As a corollary caveat (and I think the largest complexity with this approach), the recipient has an atomicity problem: we'll definitely end up with something to ingest (the ~64mb-\infty of user data) and something to apply via WriteBatch. This implies two constraints, namely ordering: we must only make the range visible when it's been written completely (this implies that the RangeDescriptor must be written last, etc) but it also requires us to have a way to recover from an ill-timed crash or I/O error during snapshot application.

There might be a better way to address that, but one I can think off right now is to first write (and sync) a marker key to RocksDB that we delete at the end of a snapshot application; that we make sure to remove all partial state potentially written by an earlier snapshot before writing our own data; and clearing that marker (atomically with or after syncing) the final write which puts down the RangeDescriptor. I haven't thought this through, but potentially Raft tombstones could be augmented to carry a sort of "dirty bit" (which can be cleared by applying a snapshot successfully or GC'ing the replica). Recovering from an I/O error in particular will be tricky. We have to remove the replica from the store and also make sure it isn't used (in meaningful ways) by anyone who might be holding on to it. It's easier to recover from a crash as we'll see the dirty bit and get to clean it up eagerly.

It seems pretty bad to render a replica unusable after a failed snapshot application, but keep in mind that the "failure" (assuming no bugs) would be an I/O error, and we generally crash on those intentionally (to avoid undefined behavior). Additionally, a snapshot is either preemptive (in which case there isn't really a Replica) or Raft-trigger (in which case the Replica is so far behind that it is effectively not participating much in the Range, though it does get to vote).

[tbg]

@petermattis am I correct in assuming that ingesting a "everything-covering" SSTable is a terrible idea? Do you think there's any chance of teaching RocksDB to ingest a set of SSTables plus WriteBatches atomically?

[petermattis]

am I correct in assuming that ingesting a "everything-covering" SSTable is a terrible idea?

Yeah, I'm pretty sure that would be very bad.

Do you think there's any chance of teaching RocksDB to ingest a set of SSTables plus WriteBatches atomically?

I don't know. I think the sstable ingestion is atomic due to writing the manifest. But WriteBatches are atomic due to writing to the WAL. I'm not sure if it if would be possible to connect to the two atomic mechanisms.

[danhhz]

An "everything-covering" sstable effectively just means that it will be ingested at L0, which is fine if it's small, but probably bad if it's big.

RocksDB has assured us that it will be Hard(tm) to make an atomic sstable ingestion + writebatch application, but ingesting 2 sstables is already atomic. Is there any way we can break up what you're doing into a large sstable that affects only a small keyrange and a small sstable that covers everything? Perhaps that could be made to work, though it seems to close enough to the margin of what RocksDB has imagined that I'd try it before committing to that plan

[tbg]

@danhhz using only SSTables might be one way out. I'm not clear how bad a small everything-covering SSTable would be. BTW, are you saying 2 sstables are atomic or any number of SSTables is atomic? I wonder if it'd be better to ingest ~10 small SSTables and one large one if that makes these small ones "gapless".

[danhhz]

any number. IngestExternalFile takes a vector of paths and applies them all atomically

[danhhz]

i'm reading over the code again and just noticed that the files you ingest must themselves be mutually non-overlapping. dunno if that's relevant but fyi

it does appear to place each file individually in the lowest level where it doesn't overlap anything, so you should indeed be able to atomically have small, overlappy ones end up in L0 and big, non-overlappy ones in lower levels

One thing you do have to consider in the snapshot case is whether the range of data you're ingesting has any recent writes. If so, you could still end up in the bad case

[tbg]

@danhhz using only SSTables is a valid idea. That might work! I think the log entries can be streamlined by changing the API: (receiver permitting), send slim entries and send the files separately. The receiver can first populate the sideloaded storage and then do the ingestion.

[tbg]

we'd likely want to send multiple SSTables anyway, right? We have a target size of ~128MB I think in L6, so that seems like the size at which we'd want to start a new one. Or is that not necessary/worse than sending a huge one (for the user data, i.e. without holes)?

Does RocksDB ever have to load SSTables into memory (I hope not)? If so, large SSTables would be problematic anyway. I'm not sure if atomically ingesting N smaller SSTables would be any better, though.

[petermattis]

RocksDB streams sstables into memory while compacting. And it loads/pins sstable index blocks into memory for normal reads, but these index blocks are a small fraction of the sstable size. The only downside to large sstables that I'm aware of are they make certain compactions slower which ties up resources that could be used for other (higher priority) compactions. The downside to smaller sstables is we can either run out of file descriptors, or thrash the cache of open sstables.

[tbg]

@petermattis in light of that, sticking to ~128mb SSTables seems reasonable?

[petermattis]

@petermattis in light of that, sticking to ~128mb SSTables seems reasonable?

Yes. I'm not aware of any compelling reason to adjust that size right now.

[jeffrey-xiao]

Closed by #38932.