Delete commit log files that are covered by snapshot files

src/dbnode/integration/commitlog_bootstrap_helpers.go

@@ -144,7 +144,8 @@ func writeCommitLogDataBase(
 	}
 
 	// ensure commit log is flushing frequently
-	require.Equal(t, defaultIntegrationTestFlushInterval, opts.FlushInterval())
+	require.Equal(
+		t, defaultIntegrationTestFlushInterval, opts.FlushInterval())
 
 	var (
 		seriesLookup = newCommitLogSeriesStates(data)

src/dbnode/integration/disk_cleanup_helpers.go

@@ -24,7 +24,6 @@ package integration
 
 import (
 	"errors"
-	"os"
 	"testing"
 	"time"
 
@@ -83,14 +82,6 @@ func writeIndexFileSetFiles(t *testing.T, storageOpts storage.Options, md namesp
 	}
 }
 
-func writeCommitLogs(t *testing.T, filePathPrefix string, fileTimes []time.Time) {
-	for _, start := range fileTimes {
-		commitLogFile, _ := fs.NextCommitLogsFile(filePathPrefix, start)
-		_, err := os.Create(commitLogFile)
-		require.NoError(t, err)
-	}
-}
-
 type cleanupTimesCommitLog struct {
 	filePathPrefix string
 	times          []time.Time

src/dbnode/integration/disk_cleanup_multi_ns_test.go

@@ -26,8 +26,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/m3db/m3db/src/dbnode/integration/generate"
 	"github.com/m3db/m3db/src/dbnode/retention"
 	"github.com/m3db/m3db/src/dbnode/storage/namespace"
+	xtime "github.com/m3db/m3x/time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -133,7 +135,21 @@ func TestDiskCleanupMultipleNamespace(t *testing.T) {
 
 	log.Infof("creating commit log and fileset files")
 	shard := uint32(0)
-	writeCommitLogs(t, filePathPrefix, commitLogTimes)
+	for _, clTime := range commitLogTimes {
+		// Need to generate valid commit log files otherwise cleanup will fail.
+		data := map[xtime.UnixNano]generate.SeriesBlock{
+			xtime.ToUnixNano(clTime): nil,
+		}
+		writeCommitLogDataSpecifiedTS(
+			t,
+			testSetup,
+			testSetup.storageOpts.CommitLogOptions().SetFlushInterval(defaultIntegrationTestFlushInterval),
+			data,
+			ns1,
+			clTime,
+			false,
+		)
+	}
 	writeDataFileSetFiles(t, testSetup.storageOpts, ns1, shard, ns1Times)
 	writeDataFileSetFiles(t, testSetup.storageOpts, ns2, shard, ns2Times)
 

src/dbnode/integration/disk_cleanup_test.go

@@ -26,6 +26,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/m3db/m3db/src/dbnode/integration/generate"
+	"github.com/m3db/m3db/src/dbnode/storage/namespace"
+	xtime "github.com/m3db/m3x/time"
 	"github.com/stretchr/testify/require"
 )
 
@@ -61,11 +64,27 @@ func TestDiskCleanup(t *testing.T) {
 	numTimes := 10
 	fileTimes := make([]time.Time, numTimes)
 	now := testSetup.getNowFn()
+	ns1, err := namespace.NewMetadata(testNamespaces[0], namespace.NewOptions())
+	require.NoError(t, err)
 	for i := 0; i < numTimes; i++ {
 		fileTimes[i] = now.Add(time.Duration(i) * blockSize)
 	}
 	writeDataFileSetFiles(t, testSetup.storageOpts, md, shard, fileTimes)
-	writeCommitLogs(t, filePathPrefix, fileTimes)
+	for _, clTime := range fileTimes {
+		// Need to generate valid commit log files otherwise cleanup will fail.
+		data := map[xtime.UnixNano]generate.SeriesBlock{
+			xtime.ToUnixNano(clTime): nil,
+		}
+		writeCommitLogDataSpecifiedTS(
+			t,
+			testSetup,
+			testSetup.storageOpts.CommitLogOptions().SetFlushInterval(defaultIntegrationTestFlushInterval),
+			data,
+			ns1,
+			clTime,
+			false,
+		)
+	}
 
 	// Move now forward by retentionPeriod + 2 * blockSize so fileset files
 	// and commit logs at now will be deleted

src/dbnode/integration/fs_commitlog_mixed_mode_read_write_test.go

@@ -317,6 +317,19 @@ func (d dataPointsInTimeOrder) toSeriesMap(blockSize time.Duration) generate.Ser
 	return nil
 }
 
+// before returns a slice of the dataPointsInTimeOrder that are before the
+// specified time t.
+func (d dataPointsInTimeOrder) before(t time.Time) dataPointsInTimeOrder {
+	var i int
+	for i = range d {
+		if !d[i].time.Before(t) {
+			break
+		}
+	}
+
+	return d[:i]
+}
+
 type idGen struct {
 	baseID string
 }