Add support for putting commit log bootstrapper before peers

kube/m3dbnode-configmap.yaml

@@ -68,6 +68,8 @@ data:
         bootstrappers:
             - filesystem
             - commitlog
+            - peers
+            - uninitialized_topology
         fs:
             numProcessorsPerCPU: 0.125
 

src/cmd/services/m3dbnode/config/bootstrap.go

@@ -33,6 +33,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
 	bfs "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/fs"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/peers"
+	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/uninitialized"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
 	"github.com/m3db/m3/src/dbnode/storage/index"
 )
@@ -120,7 +121,7 @@ func (bsc BootstrapConfiguration) New(
 		case bootstrapper.NoOpNoneBootstrapperName:
 			bs = bootstrapper.NewNoOpNoneBootstrapperProvider()
 		case bfs.FileSystemBootstrapperName:
-			fsbopts := bfs.NewOptions().
+			fsbOpts := bfs.NewOptions().
 				SetInstrumentOptions(opts.InstrumentOptions()).
 				SetResultOptions(rsOpts).
 				SetFilesystemOptions(fsOpts).
@@ -129,35 +130,40 @@ func (bsc BootstrapConfiguration) New(
 				SetDatabaseBlockRetrieverManager(opts.DatabaseBlockRetrieverManager()).
 				SetRuntimeOptionsManager(opts.RuntimeOptionsManager()).
 				SetIdentifierPool(opts.IdentifierPool())
-			bs, err = bfs.NewFileSystemBootstrapperProvider(fsbopts, bs)
+			bs, err = bfs.NewFileSystemBootstrapperProvider(fsbOpts, bs)
 			if err != nil {
 				return nil, err
 			}
 		case commitlog.CommitLogBootstrapperName:
-			copts := commitlog.NewOptions().
+			cOpts := commitlog.NewOptions().
 				SetResultOptions(rsOpts).
 				SetCommitLogOptions(opts.CommitLogOptions())
 
 			inspection, err := fs.InspectFilesystem(fsOpts)
 			if err != nil {
 				return nil, err
 			}
-			bs, err = commitlog.NewCommitLogBootstrapperProvider(copts, inspection, bs)
+			bs, err = commitlog.NewCommitLogBootstrapperProvider(cOpts, inspection, bs)
 			if err != nil {
 				return nil, err
 			}
 		case peers.PeersBootstrapperName:
-			popts := peers.NewOptions().
+			pOpts := peers.NewOptions().
 				SetResultOptions(rsOpts).
 				SetAdminClient(adminClient).
 				SetPersistManager(opts.PersistManager()).
 				SetDatabaseBlockRetrieverManager(opts.DatabaseBlockRetrieverManager()).
 				SetFetchBlocksMetadataEndpointVersion(bsc.peersFetchBlocksMetadataEndpointVersion()).
 				SetRuntimeOptionsManager(opts.RuntimeOptionsManager())
-			bs, err = peers.NewPeersBootstrapperProvider(popts, bs)
+			bs, err = peers.NewPeersBootstrapperProvider(pOpts, bs)
 			if err != nil {
 				return nil, err
 			}
+		case uninitialized.UninitializedTopologyBootstrapperName:
+			uopts := uninitialized.NewOptions().
+				SetResultOptions(rsOpts).
+				SetInstrumentOptions(opts.InstrumentOptions())
+			bs = uninitialized.NewuninitializedTopologyBootstrapperProvider(uopts, bs)
 		default:
 			return nil, fmt.Errorf("unknown bootstrapper: %s", bsc.Bootstrappers[i])
 		}
@@ -189,12 +195,20 @@ func ValidateBootstrappersOrder(names []string) error {
 		peers.PeersBootstrapperName: []string{
 			// Peers must always appear after filesystem
 			bfs.FileSystemBootstrapperName,
+			// Peers may appear before OR after commitlog
+			commitlog.CommitLogBootstrapperName,
 		},
 		commitlog.CommitLogBootstrapperName: []string{
 			// Commit log bootstrapper may appear after filesystem or peers
 			bfs.FileSystemBootstrapperName,
 			peers.PeersBootstrapperName,
 		},
+		uninitialized.UninitializedTopologyBootstrapperName: []string{
+			// Unintialized bootstrapper may appear after filesystem or peers or commitlog
+			bfs.FileSystemBootstrapperName,
+			commitlog.CommitLogBootstrapperName,
+			peers.PeersBootstrapperName,
+		},
 	}
 
 	validated := make(map[string]struct{})

src/cmd/services/m3dbnode/main/common_test.go

@@ -34,8 +34,10 @@ import (
 	"time"
 
 	"github.com/gogo/protobuf/proto"
+	"github.com/m3db/m3/src/dbnode/client"
 	"github.com/m3db/m3/src/dbnode/retention"
 	"github.com/m3db/m3/src/dbnode/storage/namespace"
+	"github.com/m3db/m3cluster/shard"
 	"github.com/m3db/m3x/ident"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -139,3 +141,35 @@ func newNamespaceWithIndexProtoValue(id string, indexEnabled bool) (proto.Messag
 	}
 	return namespace.ToProto(nsMap), nil
 }
+
+// waitUntilAllShardsAreAvailable continually polls the session checking to see if the topology.Map
+// that the session is currently storing contains a non-zero number of host shard sets, and if so,
+// makes sure that all their shard states are Available.
+func waitUntilAllShardsAreAvailable(t *testing.T, session client.AdminSession) {
+outer:
+	for {
+		time.Sleep(10 * time.Millisecond)
+
+		topoMap, err := session.TopologyMap()
+		require.NoError(t, err)
+
+		var (
+			hostShardSets = topoMap.HostShardSets()
+		)
+
+		if len(hostShardSets) == 0 {
+			// We haven't received an actual topology yet.
+			continue
+		}
+
+		for _, hostShardSet := range hostShardSets {
+			for _, hostShard := range hostShardSet.ShardSet().All() {
+				if hostShard.State() != shard.Available {
+					continue outer
+				}
+			}
+		}
+
+		break
+	}
+}

src/cmd/services/m3dbnode/main/main_index_test.go

@@ -327,6 +327,8 @@ db:
         bootstrappers:
             - filesystem
             - commitlog
+            - peers
+            - uninitialized_topology
         fs:
             numProcessorsPerCPU: 0.125
 

src/cmd/services/m3dbnode/main/main_test.go

@@ -37,7 +37,6 @@ import (
 	"github.com/m3db/m3cluster/integration/etcd"
 	"github.com/m3db/m3cluster/placement"
 	"github.com/m3db/m3cluster/services"
-	"github.com/m3db/m3cluster/shard"
 	xconfig "github.com/m3db/m3x/config"
 	"github.com/m3db/m3x/ident"
 	"github.com/m3db/m3x/instrument"
@@ -488,6 +487,8 @@ db:
         bootstrappers:
             - filesystem
             - commitlog
+            - peers
+            - uninitialized_topology
         fs:
             numProcessorsPerCPU: 0.125
 
@@ -632,35 +633,3 @@ db:
                   endpoint: {{.InitialClusterEndpoint}}
 `
 )
-
-// waitUntilAllShardsAreAvailable continually polls the session checking to see if the topology.Map
-// that the session is currently storing contains a non-zero number of host shard sets, and if so,
-// makes sure that all their shard states are Available.
-func waitUntilAllShardsAreAvailable(t *testing.T, session client.AdminSession) {
-outer:
-	for {
-		time.Sleep(10 * time.Millisecond)
-
-		topoMap, err := session.TopologyMap()
-		require.NoError(t, err)
-
-		var (
-			hostShardSets = topoMap.HostShardSets()
-		)
-
-		if len(hostShardSets) == 0 {
-			// We haven't received an actual topology yet.
-			continue
-		}
-
-		for _, hostShardSet := range hostShardSets {
-			for _, hostShard := range hostShardSet.ShardSet().All() {
-				if hostShard.State() != shard.Available {
-					continue outer
-				}
-			}
-		}
-
-		break
-	}
-}

src/dbnode/config/m3dbnode-local-etcd.yml

@@ -63,7 +63,8 @@ db:
     bootstrappers:
         - filesystem
         - commitlog
-        - noop-none
+        - peers
+        - uninitialized_topology
     fs:
         numProcessorsPerCPU: 0.125
 

src/dbnode/config/m3dbnode-local.yml

@@ -63,7 +63,8 @@ db:
     bootstrappers:
         - filesystem
         - commitlog
-        - noop-none
+        - peers
+        - uninitialized_topology
     fs:
         numProcessorsPerCPU: 0.125
 

src/dbnode/example/m3db-node-config.yaml

@@ -47,6 +47,8 @@ bootstrap:
   bootstrappers:
       - filesystem
       - commitlog
+      - peers
+      - uninitialized_topology
   fs:
       numProcessorsPerCPU: 0.125
 

src/dbnode/integration/bootstrap_after_buffer_rotation_regression_test.go

@@ -118,6 +118,7 @@ func TestBootstrapAfterBufferRotation(t *testing.T) {
 	signalCh := make(chan struct{})
 	bootstrapper, err := commitlogBootstrapperProvider.Provide()
 	require.NoError(t, err)
+
 	test := newTestBootstrapperSource(testBootstrapperSourceOptions{
 		readData: func(
 			_ namespace.Metadata,

src/dbnode/storage/bootstrap/bootstrapper/commitlog/source.go

@@ -36,8 +36,10 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
 	"github.com/m3db/m3/src/dbnode/storage/index/convert"
 	"github.com/m3db/m3/src/dbnode/storage/namespace"
+	"github.com/m3db/m3/src/dbnode/topology"
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/dbnode/x/xio"
+	"github.com/m3db/m3cluster/shard"
 	"github.com/m3db/m3x/checked"
 	"github.com/m3db/m3x/ident"
 	"github.com/m3db/m3x/instrument"
@@ -104,10 +106,7 @@ func (s *commitLogSource) AvailableData(
 	shardsTimeRanges result.ShardTimeRanges,
 	runOpts bootstrap.RunOptions,
 ) result.ShardTimeRanges {
-	// Commit log bootstrapper is a last ditch effort, so fulfill all
-	// time ranges requested even if not enough data, just to succeed
-	// the bootstrap
-	return shardsTimeRanges
+	return s.availability(ns, shardsTimeRanges, runOpts)
 }
 
 // ReadData will read a combination of the available snapshot files and commit log files to
@@ -1249,10 +1248,7 @@ func (s *commitLogSource) AvailableIndex(
 	shardsTimeRanges result.ShardTimeRanges,
 	runOpts bootstrap.RunOptions,
 ) result.ShardTimeRanges {
-	// Commit log bootstrapper is a last ditch effort, so fulfill all
-	// time ranges requested even if not enough data, just to succeed
-	// the bootstrap
-	return shardsTimeRanges
+	return s.availability(ns, shardsTimeRanges, runOpts)
 }
 
 func (s *commitLogSource) ReadIndex(
@@ -1425,6 +1421,76 @@ func (s commitLogSource) maybeAddToIndex(
 	return err
 }
 
+// The commitlog bootstrapper determines availability primarily by checking if the
+// origin host has ever reached the "Available" state for the shard that is being
+// bootstrapped. If not, then it can't provide data for that shard because it doesn't
+// have all of it by definition.
+func (s *commitLogSource) availability(
+	ns namespace.Metadata,
+	shardsTimeRanges result.ShardTimeRanges,
+	runOpts bootstrap.RunOptions,
+) result.ShardTimeRanges {
+	var (
+		topoState                = runOpts.InitialTopologyState()
+		availableShardTimeRanges = result.ShardTimeRanges{}
+	)
+
+	for shardIDUint := range shardsTimeRanges {
+		shardID := topology.ShardID(shardIDUint)
+		hostShardStates, ok := topoState.ShardStates[shardID]
+		if !ok {
+			// This shard was not part of the topology when the bootstrapping
+			// process began.
+			continue
+		}
+
+		originHostShardState, ok := hostShardStates[topology.HostID(topoState.Origin.ID())]
+		if !ok {
+			// TODO(rartoul): Make this a hard error once we refactor the interface to support
+			// returning errors.
+			iOpts := s.opts.CommitLogOptions().InstrumentOptions()
+			invariantLogger := instrument.EmitInvariantViolationAndGetLogger(iOpts)
+			invariantLogger.Errorf(
+				"initial topology state does not contain shard state for origin node and shard: %d", shardIDUint)
+			continue
+		}
+
+		originShardState := originHostShardState.ShardState
+		switch originShardState {
+		// In the Initializing state we have to assume that the commit log
+		// is missing data and can't satisfy the bootstrap request.
+		case shard.Initializing:
+		// In the Leaving and Available case, we assume that the commit log contains
+		// all the data required to satisfy the bootstrap request because the node
+		// had (at some point) been completely bootstrapped for the requested shard.
+		// This doesn't mean that the node can't be missing any data or wasn't down
+		// for some period of time and missing writes in a multi-node deployment, it
+		// only means that technically the node has successfully taken ownership of
+		// the data for this shard and made it to a "bootstrapped" state which is
+		// all that is required to maintain our cluster-level consistency guarantees.
+		case shard.Leaving:
+			fallthrough
+		case shard.Available:
+			// Assume that we can bootstrap any requested time range, which is valid as
+			// long as the FS bootstrapper precedes the commit log bootstrapper.
+			// TODO(rartoul): Once we make changes to the bootstrapping interfaces
+			// to distinguish between "unfulfilled" data and "corrupt" data, then
+			// modify this to only say the commit log bootstrapper can fullfil
+			// "unfulfilled" data, but not corrupt data.
+			availableShardTimeRanges[shardIDUint] = shardsTimeRanges[shardIDUint]
+		case shard.Unknown:
+			fallthrough
+		default:
+			// TODO(rartoul): Make this a hard error once we refactor the interface to support
+			// returning errors.
+			s.log.Errorf("unknown shard state: %v", originShardState)
+			return result.ShardTimeRanges{}
+		}
+	}
+
+	return availableShardTimeRanges
+}
+
 func newReadSeriesPredicate(ns namespace.Metadata) commitlog.SeriesFilterPredicate {
 	nsID := ns.ID()
 	return func(id ident.ID, namespace ident.ID) bool {