[dbnode] Integrate ns cold flusher into tile aggregation.

src/dbnode/storage/namespace.go

@@ -1751,6 +1751,11 @@ func (n *dbNamespace) aggregateTiles(
 		blockReaders = append(blockReaders, reader)
 	}
 
+	onColdFlushNs, err := n.opts.OnColdFlush().ColdFlushNamespace(n)
+	if err != nil {
+		return 0, err
+	}
+
 	var processedTileCount int64
 	for _, targetShard := range targetShards {
 		sourceShard, _, err := sourceNs.ReadableShardAt(targetShard.ID())
@@ -1775,7 +1780,8 @@ func (n *dbNamespace) aggregateTiles(
 		}
 
 		shardProcessedTileCount, err := targetShard.AggregateTiles(
-			sourceNs.ID(), n, sourceShard.ID(), blockReaders, writer, sourceBlockVolumes, opts)
+			sourceNs.ID(), n, sourceShard.ID(), blockReaders, writer, sourceBlockVolumes,
+			onColdFlushNs, opts)
 
 		processedTileCount += shardProcessedTileCount
 		processedShards.Inc(1)
@@ -1784,6 +1790,10 @@ func (n *dbNamespace) aggregateTiles(
 		}
 	}
 
+	if err := onColdFlushNs.Done(); err != nil {
+		return 0, err
+	}
+
 	n.log.Info("finished large tiles aggregation for namespace",
 		zap.String("sourceNs", sourceNs.ID().String()),
 		zap.Time("targetBlockStart", targetBlockStart),

src/dbnode/storage/namespace_test.go

@@ -57,9 +57,7 @@ import (
 	"github.com/uber-go/tally"
 )
 
-var (
-	testShardIDs = sharding.NewShards([]uint32{0, 1}, shard.Available)
-)
+var testShardIDs = sharding.NewShards([]uint32{0, 1}, shard.Available)
 
 type closerFn func()
 
@@ -800,9 +798,7 @@ func newNeedsFlushNamespace(t *testing.T, shardNumbers []uint32) *dbNamespace {
 	shards := sharding.NewShards(shardNumbers, shard.Available)
 	dopts := DefaultTestOptions()
 
-	var (
-		hashFn = func(identifier ident.ID) uint32 { return shards[0].ID() }
-	)
+	hashFn := func(identifier ident.ID) uint32 { return shards[0].ID() }
 	metadata, err := namespace.NewMetadata(defaultTestNs1ID, defaultTestNs1Opts)
 	require.NoError(t, err)
 	ropts := metadata.Options().RetentionOptions()
@@ -902,6 +898,7 @@ func TestNamespaceNeedsFlushRangeMultipleShardConflict(t *testing.T) {
 	assertNeedsFlush(t, ns, t2, t1, false)
 	assertNeedsFlush(t, ns, t2, t0, false)
 }
+
 func TestNamespaceNeedsFlushRangeSingleShardConflict(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -943,11 +940,9 @@ func TestNamespaceNeedsFlushAllSuccess(t *testing.T) {
 	var (
 		shards = sharding.NewShards([]uint32{0, 2, 4}, shard.Available)
 		dopts  = DefaultTestOptions()
-	)
-
-	var (
 		hashFn = func(identifier ident.ID) uint32 { return shards[0].ID() }
 	)
+
 	metadata, err := namespace.NewMetadata(defaultTestNs1ID, defaultTestNs1Opts)
 	require.NoError(t, err)
 	shardSet, err := sharding.NewShardSet(shards, hashFn)
@@ -1435,6 +1430,13 @@ func TestNamespaceAggregateTiles(t *testing.T) {
 	targetRetentionOpts := targetNs.nopts.RetentionOptions().SetBlockSize(targetBlockSize)
 	targetNs.nopts = targetNs.nopts.SetColdWritesEnabled(true).SetRetentionOptions(targetRetentionOpts)
 
+	// Pass in mock cold flusher and expect the cold flush ns process to finish.
+	mockOnColdFlushNs := NewMockOnColdFlushNamespace(ctrl)
+	mockOnColdFlushNs.EXPECT().Done().Return(nil)
+	mockOnColdFlush := NewMockOnColdFlush(ctrl)
+	mockOnColdFlush.EXPECT().ColdFlushNamespace(gomock.Any()).Return(mockOnColdFlushNs, nil)
+	targetNs.opts = targetNs.opts.SetOnColdFlush(mockOnColdFlush)
+
 	sourceShard0 := NewMockdatabaseShard(ctrl)
 	sourceShard1 := NewMockdatabaseShard(ctrl)
 	sourceNs.shards[0] = sourceShard0
@@ -1466,13 +1468,13 @@ func TestNamespaceAggregateTiles(t *testing.T) {
 	targetShard0.EXPECT().
 		AggregateTiles(
 			sourceNsIDMatcher, targetNs, shard0ID, gomock.Len(2), gomock.Any(),
-			sourceBlockVolumes0, opts).
+			sourceBlockVolumes0, gomock.Any(), opts).
 		Return(int64(3), nil)
 
 	targetShard1.EXPECT().
 		AggregateTiles(
 			sourceNsIDMatcher, targetNs, shard1ID, gomock.Len(2), gomock.Any(),
-			sourceBlockVolumes1, opts).
+			sourceBlockVolumes1, gomock.Any(), opts).
 		Return(int64(2), nil)
 
 	processedTileCount, err := targetNs.AggregateTiles(sourceNs, opts)

src/dbnode/storage/options.go

@@ -947,6 +947,7 @@ func (a *noopTileAggregator) AggregateTiles(
 	shardID uint32,
 	readers []fs.DataFileSetReader,
 	writer fs.StreamingWriter,
+	onFlushSeries persist.OnFlushSeries,
 ) (int64, error) {
 	return 0, nil
 }

src/dbnode/storage/shard.go

@@ -2289,7 +2289,7 @@ func (s *dbShard) ColdFlush(
 	flushPreparer persist.FlushPreparer,
 	resources coldFlushReuseableResources,
 	nsCtx namespace.Context,
-	onFlush persist.OnFlushSeries,
+	onFlushSeries persist.OnFlushSeries,
 ) (ShardColdFlush, error) {
 	// We don't flush data when the shard is still bootstrapping.
 	s.RLock()
@@ -2395,7 +2395,8 @@ func (s *dbShard) ColdFlush(
 		}
 
 		nextVersion := coldVersion + 1
-		close, err := merger.Merge(fsID, mergeWithMem, nextVersion, flushPreparer, nsCtx, onFlush)
+		close, err := merger.Merge(fsID, mergeWithMem, nextVersion, flushPreparer, nsCtx,
+			onFlushSeries)
 		if err != nil {
 			multiErr = multiErr.Add(err)
 			continue
@@ -2664,6 +2665,7 @@ func (s *dbShard) AggregateTiles(
 	blockReaders []fs.DataFileSetReader,
 	writer fs.StreamingWriter,
 	sourceBlockVolumes []shardBlockVolume,
+	onFlushSeries persist.OnFlushSeries,
 	opts AggregateTilesOptions,
 ) (int64, error) {
 	if len(blockReaders) != len(sourceBlockVolumes) {
@@ -2737,7 +2739,7 @@ func (s *dbShard) AggregateTiles(
 	var multiErr xerrors.MultiError
 
 	processedTileCount, err := s.tileAggregator.AggregateTiles(
-		opts, targetNs, s.ID(), openBlockReaders, writer)
+		opts, targetNs, s.ID(), openBlockReaders, writer, onFlushSeries)
 	if err != nil {
 		// NB: cannot return on the error here, must finish writing.
 		multiErr = multiErr.Add(err)

src/dbnode/storage/shard_test.go

@@ -1899,14 +1899,17 @@ func TestShardAggregateTiles(t *testing.T) {
 		}),
 		writer.EXPECT().Close(),
 	)
+	noOpColdFlushNs := &persist.NoOpColdFlushNamespace{}
 
 	targetNs := NewMockNamespace(ctrl)
 	aggregator.EXPECT().
-		AggregateTiles(opts, targetNs, sourceShard.ID(), gomock.Len(2), writer).
+		AggregateTiles(opts, targetNs, sourceShard.ID(), gomock.Len(2), writer,
+			noOpColdFlushNs).
 		Return(expectedProcessedTileCount, nil)
 
 	processedTileCount, err := targetShard.AggregateTiles(
-		sourceNsID, targetNs, sourceShard.ID(), blockReaders, writer, sourceBlockVolumes, opts)
+		sourceNsID, targetNs, sourceShard.ID(), blockReaders, writer,
+		sourceBlockVolumes, noOpColdFlushNs, opts)
 	require.NoError(t, err)
 	assert.Equal(t, expectedProcessedTileCount, processedTileCount)
 }
@@ -1929,7 +1932,8 @@ func TestShardAggregateTilesVerifySliceLengths(t *testing.T) {
 	writer := fs.NewMockStreamingWriter(ctrl)
 
 	_, err := targetShard.AggregateTiles(
-		srcNsID, nil, 1, blockReaders, writer, sourceBlockVolumes, AggregateTilesOptions{})
+		srcNsID, nil, 1, blockReaders, writer, sourceBlockVolumes,
+		&persist.NoOpColdFlushNamespace{}, AggregateTilesOptions{})
 	require.EqualError(t, err, "blockReaders and sourceBlockVolumes length mismatch (0 != 1)")
 }
 

src/dbnode/storage/types.go

@@ -653,6 +653,7 @@ type databaseShard interface {
 		blockReaders []fs.DataFileSetReader,
 		writer fs.StreamingWriter,
 		sourceBlockVolumes []shardBlockVolume,
+		onFlushSeries persist.OnFlushSeries,
 		opts AggregateTilesOptions,
 	) (int64, error)
 
@@ -1403,6 +1404,7 @@ type TileAggregator interface {
 		shardID uint32,
 		readers []fs.DataFileSetReader,
 		writer fs.StreamingWriter,
+		onFlushSeries persist.OnFlushSeries,
 	) (int64, error)
 }