[dbnode] Background cold flush process

[notbdu]

What this PR does / why we need it:

Background TSDB cold flush process so it doesn't block data snapshotting/commit log rotation.

After backgrounding cold flush, we need to ensure data consistency during node restart/bootstrap. Added an integration test to guarantee that we can bootstrap TSDB data from snapshots for all block starts. Also tests that we can cold flush the data afterwards and that the persisted snapshots ultimately end up being empty on disk.

Special notes for your reviewer:

Does this PR introduce a user-facing and/or backwards incompatible change?:

NONE

Does this PR require updating code package or user-facing documentation?:

NONE

[notbdu]

Refactored bootstrapper initializing to be mostly done in a centralized place.

[robskillington]

+1

[notbdu]

There were two places where the commit log bootstrapper was getting skipped. This was the first one.

[robskillington]

Ah, TY for pointing that out.

[notbdu]

Changed commitlog bootstrapper here to use the TargetShardTimeRanges which covers the entire bootstrappable range.

[robskillington]

👍

[notbdu]

This is the additional option I added to capture the original target time ranges for shards that does not change per bootstrapper pass.

[robskillington]

Hm, I've always found sleeping to be quite flakey, is there no way to look at metrics emitted using a test scope perhaps?

e.g. using setup.Scope().Snapshot().Counters():
https://github.com/m3db/m3/blob/c373017b19293f9ce24164d4757431f9141906d4/src/dbnode/integration/index_single_node_high_concurrency_test.go#L263:L286

[robskillington]

Also if we don't use metrics to wait until the snapshot is done, I also feel like we probably can't guarantee these files were actually snapshotted? It's kind of a race.

Could we perhaps A) use the metric to work out when snapshot has done, B) stop the server, then use the code you already have below to make sure there are entries in the snapshot files? That will guarantee cold snapshots actually occurred. Otherwise it's possible snapshots had not have happened yet, data was in commit logs and actually restored from commit logs not snapshots, hence we wouldn't be testing the lifecycle of cold snapshots. I know that's maybe only some % chance of possibility, but we don't want to accidentally slide into thinking we have this covered and find out later this wasn't actually catching a regression when a later regression occurs.

[robskillington]

Similar mention here.

[robskillington]

nit: Add endline after this for between the methods?

[robskillington]

LGTM

[ryanhall07]

@notbdu curious about this change. it seems to have introduced a regression where we read the commit log even if there are no time ranges to fulfill. example of taking 11 mins to read a commit log without any range to fulfill.

{"level":"info","ts":1599641830.4700596,"msg":"bootstrap from source requested","bootstrapper":"commitlog","namespace":"default","numShards":85,"dataRange":0,"indexRange":0}
{"level":"info","ts":1599641830.4701018,"msg":"bootstrap from source ready after availability query","bootstrapper":"commitlog","namespace":"default","numShards":85,"dataRange":0,"indexRange":0}
{"level":"info","ts":1599642651.0040348,"msg":"bootstrapping from source completed successfully","bootstrapper":"commitlog","took":817.729679365,"namespace":"default","numShards":85,"dataRangeRequested":0,"dataRangeFulfilled":0,"indexRangeRequested":0,"indexRangeFulfilled":0,"numIndexBlocks":0}

cc: @robskillington

[notbdu]

We made this change so that the commit log bootstrapper would not be dependent on prior bootstrappers. The commit log bootstrapper will bootstrap the target shard time ranges irrespective of what has been previously fulfilled.

// NB(r): Combine all shard time ranges across data and index
// so we can do in one go.
shardTimeRanges := result.NewShardTimeRanges()
// NB(bodu): Use TargetShardTimeRanges which covers the entire original target shard range
// since the commitlog bootstrapper should run for the entire bootstrappable range per shard.
shardTimeRanges.AddRanges(ns.DataRunOptions.TargetShardTimeRanges)
if ns.Metadata.Options().IndexOptions().Enabled() {
        shardTimeRanges.AddRanges(ns.IndexRunOptions.TargetShardTimeRanges)
}

// NamespaceRunOptions are the run options for a bootstrap process run.
type NamespaceRunOptions struct {
        // ShardTimeRanges are the time ranges for the shards that should be fulfilled
        // by the bootstrapper. This changes each bootstrapper pass as time ranges are fulfilled.
        ShardTimeRanges result.ShardTimeRanges
        // TargetShardTimeRanges are the original target time ranges for shards and does not change
        // each bootstrapper pass.
        // NB(bodu): This is used by the commit log bootstrapper as it needs to run for the entire original
        // target shard time ranges.
        TargetShardTimeRanges result.ShardTimeRanges
        // RunOptions are the run options for the bootstrap run.
        RunOptions RunOptions
}