Implementing Size-Based Log Retention

Introduction to the Lesson

An unmanaged log will grow indefinitely, consuming all available disk space and eventually crashing the system. This lesson tackles this problem head-on by implementing a fundamental feature of any production-grade message broker: a log retention policy.

This lesson directly addresses the learning outcome: Implement a size-based log retention policy to clean up old log segments.

Instead of the inefficient process of deleting individual records from a file, we will adopt the architectural pattern used by systems like Apache Kafka. We will partition our log into multiple files, called segments, and implement a policy that deletes entire old segments once a size limit is exceeded.

We will cover:

1. The concept of log segmentation as an enabler for efficient data management.
2. The design of a size-based retention strategy.
3. The Go implementation of a background process to clean up old log segments.

By the end of this session, our broker will be able to manage its own storage footprint, making it a much more robust and sustainable system.

1. The Problem of Infinite Logs and the Solution of Segmentation

As you know from our work so far, our broker's log is an append-only, immutable sequence of messages. While immutability simplifies writes and reads, it presents a major operational challenge: the log will grow forever. We need a way to discard old data.

As the article notes, policies can be based on time, message count, or size. Our focus today is on size.

Trying to delete old messages from a single, massive log file would be extremely inefficient, requiring costly I/O operations to rewrite the file. The standard solution is log segmentation.

To summarize the key points:

• A topic partition's log is not a single file, but a collection of ordered segment files.
• Writes only ever go to the single, most recent active segment.
• When the active segment reaches a configured size (or age), it is "rolled over": it becomes a read-only, inactive segment, and a new empty active segment is created.
• Each segment file is named after the offset of the first message it contains (e.g., 00000000000000000000.log, 00000000000001524100.log, etc.). This allows for a fast binary search over the segments to find the file containing a specific offset.

This segmentation strategy makes retention incredibly efficient: instead of deleting records, we just delete entire files.

2. Designing the Size-Based Retention Mechanism

With segmentation in place, the logic for size-based retention becomes straightforward. This is precisely how Apache Kafka handles it.

Based on this, we can define a clear algorithm for a background "cleaner" process:

1. Run periodically (e.g., every minute).
2. For each partition in the broker:
   a. Calculate the total size of all its log segments.
   b. Check if Total Partition Size > Max Partition Size (a configured value).
   c. If it is, find the oldest segment (the one with the lowest base offset, which will always be the first in our sorted list of segments).
   d. Delete the file(s) associated with that segment.
   e. Remove the segment from the partition's list of active segments.
   f. Repeat from step (a) until the partition size is back within its limit.

This loop ensures that if a single large segment's deletion isn't enough, the process continues with the next-oldest segment.

3. Implementation in Go

Let's translate this design into Go code. This will involve refactoring our file-based storage implementation from a single file to a collection of segments.

Step 1: Update the Storage Structures

Previously, your Partition struct might have managed a single *os.File. Now, we need to manage a collection of Segments.

Let's start in your storage or log package.

// storage/segment.go

package storage

import (
	"fmt"
	"os"
	"path/filepath"
)

const fileExtension = ".log"

// Segment represents a single log file in a partition.
type Segment struct {
	file       *os.File
	BaseOffset int64
	NextOffset int64
}

// NewSegment creates a new segment file.
func NewSegment(dir string, baseOffset int64) (*Segment, error) {
	path := filepath.Join(dir, fmt.Sprintf("%020d%s", baseOffset, fileExtension))
	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
	if err != nil {
		return nil, err
	}

	return &Segment{
		file:       f,
		BaseOffset: baseOffset,
		NextOffset: baseOffset,
	}, nil
}

func (s *Segment) Close() error {
	return s.file.Close()
}

func (s *Segment) Name() string {
    return s.file.Name()
}

Your Partition (or equivalent) struct now needs to manage these segments.

// storage/partition.go

type Partition struct {
	// ... other fields
	dir      string
	segments []*Segment
	
	// Configuration
	maxPartitionBytes int64 // Our new retention setting
	maxSegmentBytes   int64 // To trigger segment rolling
}

// In your Partition's 'Append' method, you'll need to check if you must roll the segment.
func (p *Partition) Append(data []byte) (int64, error) {
    activeSegment := p.segments[len(p.segments)-1]
    
    // Check if we need to roll to a new segment.
    if activeSegment.NextOffset - activeSegment.BaseOffset >= p.maxSegmentBytes {
        newSegment, err := NewSegment(p.dir, activeSegment.NextOffset)
        if err != nil {
            return -1, err
        }
        p.segments = append(p.segments, newSegment)
        activeSegment = newSegment
    }

    // ... continue with writing to the activeSegment.file ...
    // Remember to update activeSegment.NextOffset after the write.
}

Step 2: Implement the Log Cleaner

Now for the core retention logic. We can create a LogCleaner that runs in its own goroutine.

// broker/cleaner.go

package broker

import (
	"log"
	"os"
	"path/filepath"
	"sort"
	"time"

	"your_project/internal/storage" // assuming your partition struct is here
)

type LogCleaner struct {
	partitions map[string]*storage.Partition // A reference to the broker's partitions
	interval   time.Duration
	stopCh     chan struct{}
}

func NewLogCleaner(partitions map[string]*storage.Partition, interval time.Duration) *LogCleaner {
	return &LogCleaner{
		partitions: partitions,
		interval:   interval,
		stopCh:     make(chan struct{}),
	}
}

func (c *LogCleaner) Start() {
	ticker := time.NewTicker(c.interval)
	go func() {
		for {
			select {
			case <-ticker.C:
				c.cleanup()
			case <-c.stopCh:
				ticker.Stop()
				return
			}
		}
	}()
}

func (c *LogCleaner) Stop() {
	close(c.stopCh)
}

func (c *LogCleaner) cleanup() {
	log.Println("Running log cleanup...")
	for path, partition := range c.partitions {
		c.applyRetentionPolicy(path, partition)
	}
}

Step 3: Implement the Retention Policy Logic

This is the method that executes our algorithm. It needs access to the partition's segments and its size limit.

// broker/cleaner.go

func (c *LogCleaner) applyRetentionPolicy(partitionPath string, p *storage.Partition) {
	// We need to get these values from the partition
	maxBytes := p.MaxPartitionBytes() 
	segments := p.Segments()

	// Ensure segments are sorted by base offset, oldest first.
	sort.Slice(segments, func(i, j int) bool {
		return segments[i].BaseOffset < segments[j].BaseOffset
	})

	var totalSize int64
	for _, s := range segments {
		info, err := os.Stat(s.Name())
		if err != nil {
			log.Printf("Error getting stats for segment %s: %v", s.Name(), err)
			continue
		}
		totalSize += info.Size()
	}

	for totalSize > maxBytes {
		// Must have at least one segment to leave. Don't delete the active one if it's the only one.
		if len(segments) <= 1 {
			log.Printf("Cannot apply retention policy on partition %s: only one segment remains.", partitionPath)
			break
		}

		// Get the oldest segment (but not the active one)
		segmentToDelete := segments[0]

		log.Printf("Partition %s size (%d) exceeds max (%d). Deleting oldest segment %s",
			partitionPath, totalSize, maxBytes, segmentToDelete.Name())

		// Get size before deleting
		info, err := os.Stat(segmentToDelete.Name())
		if err != nil {
			log.Printf("Could not stat segment to delete %s: %v", segmentToDelete.Name(), err)
			break // Stop trying on this partition for this cycle
		}

		// Close and delete the file
		if err := segmentToDelete.Close(); err != nil {
			log.Printf("Failed to close segment %s before deletion: %v", segmentToDelete.Name(), err)
		}
		if err := os.Remove(segmentToDelete.Name()); err != nil {
			log.Printf("Failed to delete segment file %s: %v", segmentToDelete.Name(), err)
			break // Stop trying
		}

		// Update state
		totalSize -= info.Size()
		segments = segments[1:] // Remove from our temporary slice for this loop

		// IMPORTANT: This requires synchronizing with the Partition struct itself.
		// The Partition must expose a method to safely remove its oldest segment.
		p.RemoveOldestSegment() 
	}
}

Note: The p.RemoveOldestSegment() call is crucial. Your Partition struct will need a thread-safe method to update its internal []*Segment slice. This prevents race conditions between the cleaner trying to delete a segment and a reader trying to access it.

Conclusion

Congratulations! You have now implemented a critical feature for operational stability. By moving from a monolithic log file to a segmented log, you've enabled an efficient, size-based retention policy that prevents your broker from running out of disk space.

Key Takeaways:

• Log Retention is Non-Negotiable: For any system intended to run continuously, managing data growth is a mandatory feature, not an optional one.
• Segmentation is Key: Breaking a logical log into physical segment files is the architectural pattern that makes efficient retention possible. Deleting a whole file is vastly superior to modifying one.
• Retention is a Background Process: Log cleanup is a periodic, asynchronous task that runs alongside the main request-serving logic of the broker.
• The Algorithm is a Loop: The cleaner repeatedly removes the oldest segment until the partition's total size falls within the configured limit.

Next Up

We have successfully implemented a retention policy based on size. However, it's often more intuitive to retain data based on time (e.g., "keep the last 7 days of data").

In the next lesson, we will extend the LogCleaner you just built to implement a time-based log retention policy, providing another essential tool for managing your broker's data lifecycle.