feat: Logging and fsync delay for retention deletion

tsdb/series_file.go

@@ -203,6 +203,30 @@ func (f *SeriesFile) DeleteSeriesID(id uint64) error {
 	return p.DeleteSeriesID(id)
 }
 
+// DeleteSeriesIDNoFlush flags a series as permanently deleted.
+// This method returns a SeriesPartition that can be then used later
+// for fsync operations.
+func (f *SeriesFile) DeleteSeriesIDNoFlush(id uint64) (*SeriesPartition, error) {
+	p := f.SeriesIDPartition(id)
+	if p == nil {
+		return nil, ErrInvalidSeriesPartitionID
+	}
+	return p, p.DeleteSeriesIDNoFlush(id)
+}
+
+// FlushSegments flushes a group of partitions by id
+func (f *SeriesFile) FlushSegments(partitionIDs map[int]struct{}) error {
+	for id := range partitionIDs {
+		p := f.partitions[id]
+		if segment := p.activeSegment(); segment != nil {
+			if err := segment.Flush(); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // IsDeleted returns true if the ID has been deleted before.
 func (f *SeriesFile) IsDeleted(id uint64) bool {
 	p := f.SeriesIDPartition(id)

tsdb/series_partition.go

@@ -328,6 +328,50 @@ func (p *SeriesPartition) Compacting() bool {
 	return p.compacting
 }
 
+// DeleteSeriesIDNoFlush flags a series as permanently deleted.
+// This operation does not incur an fsync.
+// If the series is reintroduced later then it must create a new id.
+func (p *SeriesPartition) DeleteSeriesIDNoFlush(id uint64) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.closed {
+		return ErrSeriesPartitionClosed
+	}
+
+	// Already tombstoned, ignore.
+	if p.index.IsDeleted(id) {
+		return nil
+	}
+
+	// Write tombstone entry.
+	_, err := p.writeLogEntry(AppendSeriesEntry(nil, SeriesEntryTombstoneFlag, id, nil))
+	if err != nil {
+		return err
+	}
+
+	// Mark tombstone in memory.
+	p.index.Delete(id)
+
+	return nil
+}
+
+// FlushSegment fsyncs a segment to disk
+func (p *SeriesPartition) FlushSegment() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.closed {
+		return ErrSeriesPartitionClosed
+	}
+
+	if segment := p.activeSegment(); segment != nil {
+		if err := segment.Flush(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // DeleteSeriesID flags a series as permanently deleted.
 // If the series is reintroduced later then it must create a new id.
 func (p *SeriesPartition) DeleteSeriesID(id uint64) error {

tsdb/store.go

@@ -1073,15 +1073,39 @@ func (s *Store) DeleteShard(shardID uint64) error {
 				}
 			}
 
+			const DeleteLogTrigger = 10_000
+			seriesCount := ss.Cardinality()
+			deleteStart := time.Now()
+			var deletedCount atomic.Uint64
+			var partitionIDs = make(map[int]struct{})
+
 			ss.ForEach(func(id uint64) {
-				if err := sfile.DeleteSeriesID(id); err != nil {
+				part, err := sfile.DeleteSeriesIDNoFlush(id)
+				if err != nil {
 					sfile.Logger.Error(
 						"cannot delete series in shard",
 						zap.Uint64("series_id", id),
 						zap.Uint64("shard_id", shardID),
 						zap.Error(err))
+				} else {
+					partitionIDs[part.id] = struct{}{}
+					deleted := deletedCount.Add(1)
+
+					if deleted%DeleteLogTrigger == 0 {
+						s.Logger.Info(fmt.Sprintf("DeleteShard: %d series deleted", DeleteLogTrigger),
+							zap.String("db", db),
+							zap.Uint64("shard_id", shardID),
+							zap.Uint64("deleted", deleted),
+							zap.Uint64("remaining", seriesCount-deleted),
+							zap.Uint64("total", seriesCount),
+							zap.Duration("elapsed", time.Since(deleteStart)))
+					}
 				}
 			})
+
+			if err := sfile.FlushSegments(partitionIDs); err != nil {
+				return err
+			}
 		}
 	}