Implementing Time-Based Log Retention

Introduction to the Lesson

While size-based retention is crucial for managing disk space, a more common and intuitive requirement is to retain data for a specific period, like "the last 7 days." This lesson focuses on exactly that.

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

We will build directly upon the LogCleaner and segmentation logic you created previously. Instead of just checking the total size of a partition, we will now also check the age of each segment and delete those that have expired.

We will cover:

1. The mechanics of time-based retention in log-based systems.
2. How to use Go's standard library to handle time and file metadata.
3. Integrating the time-based policy into our existing LogCleaner.

By the end of this session, our broker will support both size- and time-based retention, giving us flexible and powerful control over the data lifecycle.

1. Conceptual Framework: How Time-Based Retention Works

At a high level, the idea is simple: delete data older than a certain age. However, the implementation detail is important. As we established, deleting individual messages is inefficient. Instead, retention policies operate on whole segments.

As the article explains, a segment is only considered for deletion after it has been "rolled over" and is no longer the active segment for writes. The broker's cleanup process periodically checks these inactive segments.

A key implementation detail in Kafka is that a segment is deleted only when the timestamp of the last message within it is older than the retention period. For our system, we can use a simpler but equally effective proxy: the segment file's last modification time. When a segment is rolled and becomes inactive, its modification time is fixed. Our LogCleaner can then check this timestamp to determine the segment's age.

2. Designing the Implementation in Go

Our goal is to extend the LogCleaner you built in the previous lesson. The core structure, a time.Ticker that periodically triggers a cleanup function, is already perfect. We just need to enhance the logic within the cleanup cycle.

To do this, we'll leverage two key parts of Go's standard library:

• os.Stat(): This function returns a FileInfo struct, which contains metadata about a file, including its last modification time.
• time: The time package provides functions like time.Now(), time.Since(), and time.Duration to compare times and represent age.

Let's see a practical pattern for this kind of cleanup task.

The algorithm for our time-based policy will be:

1. Run periodically via the LogCleaner's ticker.
2. For each partition:
   a. Get the list of segments, making sure to exclude the active segment from deletion.
   b. For each inactive segment, get its file information using os.Stat().
   c. Retrieve its last modification time: fileInfo.ModTime().
   d. Check if the segment has expired: modTime.Before(time.Now().Add(-maxRetentionAge)).
   e. If it has expired, delete the segment file and remove it from the partition's list of segments.

3. Integrating with the Existing Log Cleaner

Now, let's write the code. We'll add the new policy to the LogCleaner and the necessary configuration to the Partition struct.

Step 1: Update the Partition Configuration

First, add a new configuration field to your Partition struct in the storage package. A value of zero can signify that time-based retention is disabled.

// storage/partition.go

import "time"

type Partition struct {
	// ... existing fields: dir, segments, etc.
	
	// Configuration
	maxPartitionBytes int64
	maxSegmentBytes   int64
	maxRetentionAge   time.Duration // New retention setting
}

// Add a getter for this new field
func (p *Partition) MaxRetentionAge() time.Duration {
    // You might need to add locking here if config can change at runtime
    return p.maxRetentionAge
}

Step 2: Implement the Time-Based Cleanup Logic

In your broker/cleaner.go, let's add a new function that specifically handles the time-based policy. This keeps the logic clean and separate from the size-based policy.

// broker/cleaner.go

// ... imports

func (c *LogCleaner) applyTimeRetention(p *storage.Partition) {
	maxAge := p.MaxRetentionAge()
	if maxAge <= 0 {
		return // Time-based retention is disabled for this partition
	}

	segments := p.Segments()
	if len(segments) <= 1 {
		return // Don't delete the only segment
	}

	// We only clean up inactive segments. The last segment in the slice is the active one.
	inactiveSegments := segments[:len(segments)-1]
	
	// The cutoff time for deletion
	cutoffTime := time.Now().Add(-maxAge)

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

		if info.ModTime().Before(cutoffTime) {
			log.Printf("Segment %s is older than retention period (%s). Deleting.", segment.Name(), maxAge)
			
			// Close and delete the file.
			if err := segment.Close(); err != nil {
				log.Printf("Failed to close segment %s before deletion: %v", segment.Name(), err)
			}
			if err := os.Remove(segment.Name()); err != nil {
				log.Printf("Failed to delete segment file %s: %v", segment.Name(), err)
				// If we fail to delete, stop trying for this segment in this cycle.
				continue
			}

			// IMPORTANT: This must be a thread-safe operation on the Partition.
			// The partition needs a method to remove a segment by its base offset.
			p.RemoveSegment(segment.BaseOffset)
		} else {
			// Since segments are ordered by age, we can stop checking once we find one
			// that is not expired.
			break
		}
	}
}

Note: The previous lesson's RemoveOldestSegment() is good, but a more flexible RemoveSegment(baseOffset int64) method on your Partition struct would be even better, as it allows removing any specific segment. This method must use a mutex (sync.RWMutex) to safely modify the partition's segment slice, preventing race conditions.

Step 3: Update the Main Cleanup Loop

Finally, call this new function from your main cleanup method. A common strategy, which Kafka employs, is to allow either policy to trigger deletion. We can run the time-based cleanup first, and then run the size-based cleanup.

// broker/cleaner.go

func (c *LogCleaner) cleanup() {
	log.Println("Running log cleanup...")
	for path, partition := range c.partitions {
		log.Printf("Cleaning partition %s", path)
		
		// Apply time-based policy first
		c.applyTimeRetention(partition)
		
		// Then apply size-based policy
		c.applySizeRetention(path, partition) // Rename your previous function if needed
	}
}

// Previously implemented size-based logic, perhaps renamed for clarity.
func (c *LogCleaner) applySizeRetention(partitionPath string, p *storage.Partition) {
	// ... your existing size-based cleanup logic from the last lesson ...
}

By running both, you ensure that even if data volume is low, old data gets purged. Conversely, if data volume is extremely high, the size policy will kick in to protect your disk even if the data is very recent.

Conclusion

Excellent work. You have now implemented a time-based retention policy, making our message broker's data management capabilities significantly more powerful and flexible. You've extended the existing LogCleaner and learned to use Go's os and time packages to make decisions based on file metadata.

Key Takeaways:

• Time-Based Retention is a Core Feature: It provides an intuitive way to manage the data lifecycle, often directly mapping to business requirements.
• File Metadata is Your Friend: The last modification time of an inactive segment file is an effective and simple proxy for the age of its data.
• Go's Standard Library is Powerful: os.Stat() and the time package provide all the necessary primitives for implementing time-based logic.
• Policies are Complementary: Combining time- and size-based retention creates a robust system that protects against both slow, long-term growth and sudden, high-volume bursts.

Next Up

Our broker is now resilient to running out of disk space and can reliably manage its data lifecycle. The focus now shifts from correctness and stability to performance. How fast is our broker? Where are the bottlenecks?

In the next module, we will dive into performance analysis. The first step will be to write a benchmarking suite using testing.B to measure producer throughput under concurrent load, giving us our first hard numbers on the broker's performance.