Add ActiveLogs() API to commitlog and use it in the CleanupManager

docs/m3db/architecture/engine.md

@@ -181,7 +181,7 @@ The ticking process runs continously in the background and is responsible for a
 
 #### Merging all encoders
 
-M3TSZ is designed for compressing time series data in which each datapoint has a timestamp that is larger than the last encoded datapoint. For monitoring workloads this works very well because every subsequent datapoint is almost always larger than the previous one. However, real world systems are messy and occassionally out of order writes will be received. When this happens, M3DB will allocate a new encoder for the out of order datapoints. The multiple encoders need to be merged before flushing the data to disk, but to prevent huge memory spikes during the flushing process we continuously merge out of order encoders in the background.
+M3TSZ is designed for compressing time series data in which each datapoint has a timestamp that is larger than the last encoded datapoint. For monitoring workloads this works very well because every subsequent datapoint is almost always chronologically after the previous one. However, real world systems are messy and occasionally out of order writes will be received. When this happens, M3DB will allocate a new encoder for the out of order datapoints. The multiple encoders need to be merged before flushing the data to disk, but to prevent huge memory spikes during the flushing process we continuously merge out of order encoders in the background.
 
 #### Removing expired / flushed series and blocks from memory
 

src/dbnode/integration/disk_cleanup_multi_ns_test.go

@@ -154,6 +154,16 @@ func TestDiskCleanupMultipleNamespace(t *testing.T) {
 	// Move now forward by 12 hours, and see if the expected files have been deleted
 	testSetup.setNowFn(end)
 
+	// This isn't great, but right now the commitlog will only ever rotate when writes
+	// are received, so we need to issue a write after changing the time to force the
+	// commitlog rotation. This won't be required once we tie commitlog rotation into
+	// the snapshotting process.
+	testSetup.writeBatch(testNamespaces[0], generate.Block(generate.BlockConfig{
+		IDs:       []string{"foo"},
+		NumPoints: 1,
+		Start:     end,
+	}))
+
 	// Check if expected files have been deleted
 	log.Infof("waiting until data is cleaned up")
 	waitTimeout := 60 * time.Second

src/dbnode/integration/disk_cleanup_test.go

@@ -76,7 +76,6 @@ func TestDiskCleanup(t *testing.T) {
 	}
 	writeDataFileSetFiles(t, testSetup.storageOpts, md, shard, 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,
 		}
@@ -89,6 +88,15 @@ func TestDiskCleanup(t *testing.T) {
 	// and commit logs at now will be deleted
 	newNow := now.Add(retentionPeriod).Add(2 * blockSize)
 	testSetup.setNowFn(newNow)
+	// This isn't great, but right now the commitlog will only ever rotate when writes
+	// are received, so we need to issue a write after changing the time to force the
+	// commitlog rotation. This won't be required once we tie commitlog rotation into
+	// the snapshotting process.
+	testSetup.writeBatch(testNamespaces[0], generate.Block(generate.BlockConfig{
+		IDs:       []string{"foo"},
+		NumPoints: 1,
+		Start:     newNow,
+	}))
 
 	// Check if files have been deleted
 	waitTimeout := 30 * time.Second