Handle commit log files with corrupt info headers during cleanup and bootstrap

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

@@ -56,6 +56,9 @@ type BootstrapConfiguration struct {
 	// Peers bootstrapper configuration.
 	Peers *BootstrapPeersConfiguration `yaml:"peers"`
 
+	// Commitlog bootstrapper configuration.
+	Commitlog *BootstrapCommitlogConfiguration `yaml:"commitlog"`
+
 	// CacheSeriesMetadata determines whether individual bootstrappers cache
 	// series metadata across all calls (namespaces / shards / blocks).
 	CacheSeriesMetadata *bool `yaml:"cacheSeriesMetadata"`
@@ -91,6 +94,17 @@ type BootstrapPeersConfiguration struct {
 	FetchBlocksMetadataEndpointVersion client.FetchBlocksMetadataEndpointVersion `yaml:"fetchBlocksMetadataEndpointVersion"`
 }
 
+// BootstrapCommitlogConfiguration specifies config for the commitlog bootstrapper.
+type BootstrapCommitlogConfiguration struct {
+	// ReturnUnfulfilledForCorruptCommitlogFiles controls whether the commitlog bootstrapper
+	// will return unfulfilled for all shard time ranges when it encounters a corrupt commit
+	// file. Note that regardless of this value, the commitlog bootstrapper will still try and
+	// read all the uncorrupted commitlog files and return as much data as it can, but setting
+	// this to true allows the node to attempt a repair if the peers bootstrapper is configured
+	// after the commitlog bootstrapper.
+	ReturnUnfulfilledForCorruptCommitlogFiles bool `yaml:"returnUnfulfilledForCorruptCommitlogFiles"`
+}
+
 // New creates a bootstrap process based on the bootstrap configuration.
 func (bsc BootstrapConfiguration) New(
 	opts storage.Options,
@@ -140,7 +154,8 @@ func (bsc BootstrapConfiguration) New(
 		case commitlog.CommitLogBootstrapperName:
 			cOpts := commitlog.NewOptions().
 				SetResultOptions(rsOpts).
-				SetCommitLogOptions(opts.CommitLogOptions())
+				SetCommitLogOptions(opts.CommitLogOptions()).
+				SetRuntimeOptionsManager(opts.RuntimeOptionsManager())
 
 			inspection, err := fs.InspectFilesystem(fsOpts)
 			if err != nil {

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

@@ -368,6 +368,7 @@ db:
     fs:
       numProcessorsPerCPU: 0.125
     peers: null
+    commitlog: null
     cacheSeriesMetadata: null
   blockRetrieve: null
   cache:

src/dbnode/integration/bootstrap_after_buffer_rotation_regression_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/integration/generate"
 	"github.com/m3db/m3/src/dbnode/retention"
+	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap"
 	bcl "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
@@ -108,7 +109,8 @@ func TestBootstrapAfterBufferRotation(t *testing.T) {
 	bootstrapOpts := newDefaulTestResultOptions(setup.storageOpts)
 	bootstrapCommitlogOpts := bcl.NewOptions().
 		SetResultOptions(bootstrapOpts).
-		SetCommitLogOptions(commitLogOpts)
+		SetCommitLogOptions(commitLogOpts).
+		SetRuntimeOptionsManager(runtime.NewOptionsManager())
 	fsOpts := setup.storageOpts.CommitLogOptions().FilesystemOptions()
 	commitlogBootstrapperProvider, err := bcl.NewCommitLogBootstrapperProvider(
 		bootstrapCommitlogOpts, mustInspectFilesystem(fsOpts), nil)

src/dbnode/integration/bootstrap_before_buffer_rotation_no_tick_regression_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/integration/generate"
 	"github.com/m3db/m3/src/dbnode/retention"
+	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap"
 	bcl "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
@@ -122,7 +123,8 @@ func TestBootstrapBeforeBufferRotationNoTick(t *testing.T) {
 	bootstrapOpts := newDefaulTestResultOptions(setup.storageOpts)
 	bootstrapCommitlogOpts := bcl.NewOptions().
 		SetResultOptions(bootstrapOpts).
-		SetCommitLogOptions(commitLogOpts)
+		SetCommitLogOptions(commitLogOpts).
+		SetRuntimeOptionsManager(runtime.NewOptionsManager())
 	fsOpts := setup.storageOpts.CommitLogOptions().FilesystemOptions()
 	commitlogBootstrapperProvider, err := bcl.NewCommitLogBootstrapperProvider(
 		bootstrapCommitlogOpts, mustInspectFilesystem(fsOpts), nil)

src/dbnode/integration/bootstrap_helpers.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/persist/fs"
 	"github.com/m3db/m3/src/dbnode/persist/fs/commitlog"
+	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper"
 	bcl "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
@@ -173,7 +174,8 @@ func setupCommitLogBootstrapperWithFSInspection(
 	bsOpts := newDefaulTestResultOptions(setup.storageOpts)
 	bclOpts := bcl.NewOptions().
 		SetResultOptions(bsOpts).
-		SetCommitLogOptions(commitLogOpts)
+		SetCommitLogOptions(commitLogOpts).
+		SetRuntimeOptionsManager(runtime.NewOptionsManager())
 	fsOpts := setup.storageOpts.CommitLogOptions().FilesystemOptions()
 	bs, err := bcl.NewCommitLogBootstrapperProvider(
 		bclOpts, mustInspectFilesystem(fsOpts), noOpAll)

src/dbnode/integration/commitlog_bootstrap_merge_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/integration/generate"
 	persistfs "github.com/m3db/m3/src/dbnode/persist/fs"
 	"github.com/m3db/m3/src/dbnode/retention"
+	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper"
 	bcl "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
@@ -124,7 +125,8 @@ func TestCommitLogAndFSMergeBootstrap(t *testing.T) {
 	bsOpts := newDefaulTestResultOptions(setup.storageOpts)
 	bclOpts := bcl.NewOptions().
 		SetResultOptions(bsOpts).
-		SetCommitLogOptions(commitLogOpts)
+		SetCommitLogOptions(commitLogOpts).
+		SetRuntimeOptionsManager(runtime.NewOptionsManager())
 	fsOpts := setup.storageOpts.CommitLogOptions().FilesystemOptions()
 
 	commitLogBootstrapper, err := bcl.NewCommitLogBootstrapperProvider(

src/dbnode/integration/fs_commitlog_mixed_mode_read_write_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/integration/generate"
 	persistfs "github.com/m3db/m3/src/dbnode/persist/fs"
 	"github.com/m3db/m3/src/dbnode/retention"
+	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper"
 	bcl "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
@@ -225,7 +226,8 @@ func setCommitLogAndFilesystemBootstrapper(t *testing.T, opts testOptions, setup
 	bsOpts := newDefaulTestResultOptions(setup.storageOpts)
 	bclOpts := bcl.NewOptions().
 		SetResultOptions(bsOpts).
-		SetCommitLogOptions(commitLogOpts)
+		SetCommitLogOptions(commitLogOpts).
+		SetRuntimeOptionsManager(runtime.NewOptionsManager())
 
 	commitLogBootstrapper, err := bcl.NewCommitLogBootstrapperProvider(
 		bclOpts, mustInspectFilesystem(fsOpts), noOpAll)

src/dbnode/persist/fs/commitlog/commit_log_test.go

@@ -302,8 +302,9 @@ func assertCommitLogWritesByIterating(t *testing.T, l *commitLog, writes []testW
 		FileFilterPredicate:   ReadAllPredicate(),
 		SeriesFilterPredicate: ReadAllSeriesPredicate(),
 	}
-	iter, err := NewIterator(iterOpts)
+	iter, corruptFiles, err := NewIterator(iterOpts)
 	require.NoError(t, err)
+	require.Equal(t, 0, len(corruptFiles))
 	defer iter.Close()
 
 	// Convert the writes to be in-order, but keyed by series ID because the
@@ -425,8 +426,10 @@ func TestReadCommitLogMissingMetadata(t *testing.T) {
 		FileFilterPredicate:   ReadAllPredicate(),
 		SeriesFilterPredicate: ReadAllSeriesPredicate(),
 	}
-	iter, err := NewIterator(iterOpts)
+	iter, corruptFiles, err := NewIterator(iterOpts)
 	require.NoError(t, err)
+	require.Equal(t, 0, len(corruptFiles))
+
 	for iter.Next() {
 		require.NoError(t, iter.Err())
 	}
@@ -526,8 +529,10 @@ func TestCommitLogIteratorUsesPredicateFilter(t *testing.T) {
 		FileFilterPredicate:   commitLogPredicate,
 		SeriesFilterPredicate: ReadAllSeriesPredicate(),
 	}
-	iter, err := NewIterator(iterOpts)
+	iter, corruptFiles, err := NewIterator(iterOpts)
 	require.NoError(t, err)
+	require.Equal(t, 0, len(corruptFiles))
+
 	iterStruct := iter.(*iterator)
 	require.True(t, len(iterStruct.files) == 2)
 }

src/dbnode/persist/fs/commitlog/files.go

@@ -50,7 +50,7 @@ func ReadLogInfo(filePath string, opts Options) (time.Time, time.Duration, int64
 
 	fd, err = os.Open(filePath)
 	if err != nil {
-		return time.Time{}, 0, 0, err
+		return time.Time{}, 0, 0, fsError{err}
 	}
 
 	chunkReader := newChunkReader(opts.FlushSize())
@@ -72,27 +72,44 @@ func ReadLogInfo(filePath string, opts Options) (time.Time, time.Duration, int64
 	err = fd.Close()
 	fd = nil
 	if err != nil {
-		return time.Time{}, 0, 0, err
+		return time.Time{}, 0, 0, fsError{err}
 	}
 
 	return time.Unix(0, logInfo.Start), time.Duration(logInfo.Duration), logInfo.Index, decoderErr
 }
 
 // Files returns a slice of all available commit log files on disk along with
 // their associated metadata.
-func Files(opts Options) ([]File, error) {
+func Files(opts Options) ([]File, []ErrorWithPath, error) {
 	commitLogsDir := fs.CommitLogsDirPath(
 		opts.FilesystemOptions().FilePathPrefix())
 	filePaths, err := fs.SortedCommitLogFiles(commitLogsDir)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	commitLogFiles := make([]File, 0, len(filePaths))
+	errorsWithPath := make([]ErrorWithPath, 0)
 	for _, filePath := range filePaths {
+		file := File{
+			FilePath: filePath,
+		}
+
 		start, duration, index, err := ReadLogInfo(filePath, opts)
+		if _, ok := err.(fsError); ok {
+			return nil, nil, err
+		}
+
 		if err != nil {
-			return nil, err
+			errorsWithPath = append(errorsWithPath, NewErrorWithPath(
+				err, filePath))
+			continue
+		}
+
+		if err == nil {
+			file.Start = start
+			file.Duration = duration
+			file.Index = index
 		}
 
 		commitLogFiles = append(commitLogFiles, File{
@@ -104,8 +121,42 @@ func Files(opts Options) ([]File, error) {
 	}
 
 	sort.Slice(commitLogFiles, func(i, j int) bool {
+		// Sorting is best effort here since we may not know the start.
 		return commitLogFiles[i].Start.Before(commitLogFiles[j].Start)
 	})
 
-	return commitLogFiles, nil
+	return commitLogFiles, errorsWithPath, nil
+}
+
+// ErrorWithPath is an error that includes the path of the file that
+// had the error.
+type ErrorWithPath struct {
+	err  error
+	path string
+}
+
+// Error returns the error.
+func (e ErrorWithPath) Error() string {
+	return e.err.Error()
+}
+
+// Path returns the path of hte file that the error is associated with.
+func (e ErrorWithPath) Path() string {
+	return e.path
+}
+
+// NewErrorWithPath creates a new ErrorWithPath.
+func NewErrorWithPath(err error, path string) ErrorWithPath {
+	return ErrorWithPath{
+		err:  err,
+		path: path,
+	}
+}
+
+type fsError struct {
+	err error
+}
+
+func (e fsError) Error() string {
+	return e.err.Error()
 }

src/dbnode/persist/fs/commitlog/files_test.go

@@ -38,23 +38,24 @@ import (
 )
 
 func TestFiles(t *testing.T) {
-	// TODO(r): Find some time/people to help investigate this flakey test.
-	t.Skip()
-
 	dir, err := ioutil.TempDir("", "commitlogs")
 	require.NoError(t, err)
 	defer os.RemoveAll(dir)
 
 	createTestCommitLogFiles(t, dir, 10*time.Minute, 5)
 
-	opts := NewOptions()
+	var (
+		minNumBlocks = 5
+		opts         = NewOptions()
+	)
 	opts = opts.SetFilesystemOptions(
 		opts.FilesystemOptions().
 			SetFilePathPrefix(dir),
 	)
-	files, err := Files(opts)
+	files, corruptFiles, err := Files(opts)
 	require.NoError(t, err)
-	require.Equal(t, 5, len(files))
+	require.True(t, len(corruptFiles) == 0)
+	require.True(t, len(files) >= minNumBlocks)
 
 	// Make sure its sorted
 	var lastFileStart time.Time
@@ -71,12 +72,12 @@ func TestFiles(t *testing.T) {
 	}
 }
 
-// createTestCommitLogFiles creates the specified number of commit log files
-// on disk with the appropriate block size. Commit log files will be valid
-// and contain readable metadata.
+// createTestCommitLogFiles creates at least the specified number of commit log files
+// on disk with the appropriate block size. Commit log files will be valid and contain
+// readable metadata.
 func createTestCommitLogFiles(
-	t *testing.T, filePathPrefix string, blockSize time.Duration, numBlocks int) {
-	require.True(t, numBlocks >= 2)
+	t *testing.T, filePathPrefix string, blockSize time.Duration, minNumBlocks int) {
+	require.True(t, minNumBlocks >= 2)
 
 	var (
 		nowLock = sync.RWMutex{}
@@ -109,13 +110,12 @@ func createTestCommitLogFiles(
 	}
 	// Commit log writer is asynchronous and performs batching so getting the exact number
 	// of files that we want is tricky. The implementation below loops infinitely, writing
-	// a single datapoint and increasing the time after each iteration until numBlocks -1
-	// files are on disk. After that, it terminates, and the final batch flush from calling
-	// commitlog.Close() will generate the last file.
+	// a single datapoint and increasing the time after each iteration until minNumBlocks
+	// files are on disk.
 	for {
 		files, err := fs.SortedCommitLogFiles(commitLogsDir)
 		require.NoError(t, err)
-		if len(files) == numBlocks-1 {
+		if len(files) >= minNumBlocks {
 			break
 		}
 		err = commitLog.Write(context.NewContext(), series, ts.Datapoint{}, xtime.Second, nil)
@@ -126,5 +126,5 @@ func createTestCommitLogFiles(
 	require.NoError(t, commitLog.Close())
 	files, err := fs.SortedCommitLogFiles(commitLogsDir)
 	require.NoError(t, err)
-	require.Equal(t, numBlocks, len(files))
+	require.True(t, len(files) >= minNumBlocks)
 }

src/dbnode/persist/fs/commitlog/iterator.go

@@ -69,14 +69,14 @@ func ReadAllPredicate() FileFilterPredicate {
 }
 
 // NewIterator creates a new commit log iterator
-func NewIterator(iterOpts IteratorOpts) (Iterator, error) {
+func NewIterator(iterOpts IteratorOpts) (iter Iterator, corruptFiles []ErrorWithPath, err error) {
 	opts := iterOpts.CommitLogOptions
 	iops := opts.InstrumentOptions()
 	iops = iops.SetMetricsScope(iops.MetricsScope().SubScope("iterator"))
 
-	files, err := Files(opts)
+	files, corruptFiles, err := Files(opts)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	filteredFiles := filterFiles(opts, files, iterOpts.FileFilterPredicate)
 
@@ -90,7 +90,7 @@ func NewIterator(iterOpts IteratorOpts) (Iterator, error) {
 		log:        iops.Logger(),
 		files:      filteredFiles,
 		seriesPred: iterOpts.SeriesFilterPredicate,
-	}, nil
+	}, corruptFiles, nil
 }
 
 func (i *iterator) Next() bool {
@@ -191,13 +191,13 @@ func (i *iterator) nextReader() bool {
 }
 
 func filterFiles(opts Options, files []File, predicate FileFilterPredicate) []File {
-	filteredFiles := make([]File, 0, len(files))
+	filtered := make([]File, 0, len(files))
 	for _, f := range files {
 		if predicate(f) {
-			filteredFiles = append(filteredFiles, f)
+			filtered = append(filtered, f)
 		}
 	}
-	return filteredFiles
+	return filtered
 }
 
 func (i *iterator) closeAndResetReader() error {