Repository: liftbridge-io/liftbridge
Pull request #60 (2019-06-21 03:58:07)

Compaction improvements


Patch:
diff --git a/server/commitlog.go b/server/commitlog.go
index 99531dc..357d427 100644
--- a/server/commitlog.go
+++ b/server/commitlog.go
@@ -44,6 +44,10 @@ type CommitLog interface {
 	// returns the corresponding offsets in the log.
 	AppendMessageSet(ms []byte) ([]int64, error)
 
+	// Clean applies retention and compaction rules against the log, if
+	// applicable.
+	Clean() error
+
 	// Close closes each log segment file and stops the background goroutine
 	// checkpointing the high watermark to disk.
 	Close() error
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index e4a107b..ebeaa53 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -22,9 +22,7 @@ import (
 	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
-var (
-	ErrSegmentNotFound = errors.New("segment not found")
-)
+var ErrSegmentNotFound = errors.New("segment not found")
 
 const (
 	logFileSuffix               = ".log"
@@ -145,7 +143,7 @@ func (l *CommitLog) open() error {
 			_, err := os.Stat(filepath.Join(
 				l.Path, strings.Replace(file.Name(), indexFileSuffix, logFileSuffix, 1)))
 			if os.IsNotExist(err) {
-				if err := os.Remove(file.Name()); err != nil {
+				if err := os.Remove(filepath.Join(l.Path, file.Name())); err != nil {
 					return err
 				}
 			} else if err != nil {
@@ -489,18 +487,8 @@ func (l *CommitLog) split(oldActiveSegment *Segment) error {
 	}
 	l.mu.Lock()
 	segments := append(l.segments, segment)
-	// TODO: This will cause problems if a background clean is running because
-	// of the segments replace.
 	l.segments = segments
 	l.mu.Unlock()
-	cleaned, err := l.clean(segments)
-	if err != nil {
-		l.Logger.Errorf("Failed to clean log %s: %v", l.Path, err)
-	} else {
-		l.mu.Lock()
-		l.segments = cleaned
-		l.mu.Unlock()
-	}
 	return nil
 }
 
@@ -533,30 +521,37 @@ func (l *CommitLog) cleanerLoop() {
 	}
 }
 
+// Clean applies retention and compaction rules against the log, if applicable.
 func (l *CommitLog) Clean() error {
-	cleaned, err := l.clean(l.Segments())
+	l.mu.RLock()
+	oldSegments := l.segments
+	l.mu.RUnlock()
+	cleaned, err := l.clean(oldSegments)
 	if err != nil {
 		return err
 	}
-	// TODO: This will cause problems if the active segment is split
-	// while this is run because of the segments replace.
 	l.mu.Lock()
+	newSegments := l.segments
+	if len(newSegments) > len(oldSegments) {
+		// New segments were added while cleaning. Rebase the new segments onto
+		// the cleaned ones.
+		for _, seg := range newSegments[len(oldSegments):] {
+			cleaned = append(cleaned, seg)
+		}
+	}
 	l.segments = cleaned
 	l.mu.Unlock()
 	return nil
 }
 
 func (l *CommitLog) clean(segments []*Segment) ([]*Segment, error) {
-	// TODO: Only allow one clean to run at a time.
 	cleaned, err := l.deleteCleaner.Clean(segments)
 	if err != nil {
-		l.Logger.Errorf("Failed to clean log %s: %v", l.Path, err)
 		return nil, err
 	}
 	if l.Compact {
-		cleaned, err = l.compactCleaner.Clean(l.HighWatermark(), cleaned)
+		cleaned, err = l.compactCleaner.Compact(l.HighWatermark(), cleaned)
 		if err != nil {
-			l.Logger.Errorf("Failed to clean log %s: %v", l.Path, err)
 			return nil, err
 		}
 	}
diff --git a/server/commitlog/compact_cleaner.go b/server/commitlog/compact_cleaner.go
index 79f4392..e217e05 100644
--- a/server/commitlog/compact_cleaner.go
+++ b/server/commitlog/compact_cleaner.go
@@ -1,6 +1,8 @@
 package commitlog
 
 import (
+	"time"
+
 	"github.com/pkg/errors"
 
 	"github.com/liftbridge-io/liftbridge/server/logger"
@@ -26,19 +28,24 @@ func NewCompactCleaner(opts CompactCleanerOptions) *CompactCleaner {
 	return &CompactCleaner{opts}
 }
 
-// Clean performs log compaction by rewriting segments such that they contain
+// Compact performs log compaction by rewriting segments such that they contain
 // only the last message for a given key. Compaction is applied to all segments
 // up to but excluding the active (last) segment or the provided HW, whichever
 // comes first.
-func (c *CompactCleaner) Clean(hw int64, segments []*Segment) ([]*Segment, error) {
+func (c *CompactCleaner) Compact(hw int64, segments []*Segment) ([]*Segment, error) {
 	if len(segments) <= 1 {
 		return segments, nil
 	}
 
 	c.Logger.Debugf("Compacting log %s", c.Name)
+	before := time.Now()
 	compacted, removed, err := c.compact(hw, segments)
 	if err == nil {
-		c.Logger.Debugf("Finished compacting log %s, removed %d messages", c.Name, removed)
+		c.Logger.Debugf("Finished compacting log %s\n"+
+			"\tMessages Removed: %d\n"+
+			"\tSegments: %d -> %d\n"+
+			"\tDuration: %s",
+			c.Name, removed, len(segments), len(compacted), time.Since(before))
 	}
 
 	return compacted, errors.Wrap(err, "failed to compact log")
diff --git a/server/commitlog/delete_cleaner.go b/server/commitlog/delete_cleaner.go
index 08d7fb1..7111535 100644
--- a/server/commitlog/delete_cleaner.go
+++ b/server/commitlog/delete_cleaner.go
@@ -76,29 +76,30 @@ func (c *DeleteCleaner) Clean(segments []*Segment) ([]*Segment, error) {
 }
 
 func (c DeleteCleaner) applyMessagesLimit(segments []*Segment) ([]*Segment, error) {
+	if len(segments) <= 1 {
+		return segments, nil
+	}
+
 	// We start at the most recent segment and work our way backwards until we
 	// meet the retention size.
 	var (
-		lastSeg         = segments[len(segments)-1]
-		cleanedSegments = []*Segment{lastSeg}
-		totalMessages   = lastSeg.MessageCount()
+		cleanedSegments = make([]*Segment, 0, len(segments))
+		totalMessages   int64
 	)
 
-	if len(segments) > 1 {
-		var i int
-		for i = len(segments) - 2; i > -1; i-- {
-			s := segments[i]
-			totalMessages += s.MessageCount()
-			if totalMessages > c.Retention.Messages {
-				break
-			}
-			cleanedSegments = append([]*Segment{s}, cleanedSegments...)
+	var i int
+	for i = len(segments) - 1; i > -1; i-- {
+		s := segments[i]
+		totalMessages += s.MessageCount()
+		if totalMessages > c.Retention.Messages {
+			break
 		}
-		if i > -1 {
-			for ; i > -1; i-- {
-				if err := segments[i].Delete(); err != nil {
-					return nil, err
-				}
+		cleanedSegments = append([]*Segment{s}, cleanedSegments...)
+	}
+	if i > -1 {
+		for ; i > -1; i-- {
+			if err := segments[i].Delete(); err != nil {
+				return nil, err
 			}
 		}
 	}
diff --git a/server/commitlog/segment.go b/server/commitlog/segment.go
index e28d429..31df9ec 100644
--- a/server/commitlog/segment.go
+++ b/server/commitlog/segment.go
@@ -301,12 +301,12 @@ func (s *Segment) close() error {
 
 // Cleaned creates a cleaned segment for this segment.
 func (s *Segment) Cleaned() (*Segment, error) {
-	return NewSegment(s.path, s.BaseOffset, s.maxBytes, true, cleanedSuffix)
+	return NewSegment(s.path, s.BaseOffset, s.maxBytes, false, cleanedSuffix)
 }
 
 // Truncated creates a truncated segment for this segment.
 func (s *Segment) Truncated() (*Segment, error) {
-	return NewSegment(s.path, s.BaseOffset, s.maxBytes, true, truncatedSuffix)
+	return NewSegment(s.path, s.BaseOffset, s.maxBytes, false, truncatedSuffix)
 }
 
 // Replace replaces the given segment with the callee.
diff --git a/server/config.go b/server/config.go
index d51c2e4..56b2050 100644
--- a/server/config.go
+++ b/server/config.go
@@ -36,7 +36,7 @@ const (
 	defaultMinInsyncReplicas       = 1
 	defaultRetentionMaxAge         = 7 * 24 * time.Hour
 	defaultCleanerInterval         = 5 * time.Minute
-	defaultMaxSegmentBytes         = 1073741824
+	defaultMaxSegmentBytes         = 1073741824 // 1GB
 	defaultLogRollTime             = defaultRetentionMaxAge
 )
 


Test patch:
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 2991d67..80f01a8 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -177,6 +177,9 @@ func TestCleaner(t *testing.T) {
 
 	_, err = l.Append(msgs)
 	require.NoError(t, err)
+
+	l.Clean()
+
 	require.Equal(t, 1, len(l.Segments()))
 	for i, s := range l.Segments() {
 		require.NotEqual(t, s, segments[i])
diff --git a/server/commitlog/compact_cleaner_test.go b/server/commitlog/compact_cleaner_test.go
index 901b33d..e87b9c9 100644
--- a/server/commitlog/compact_cleaner_test.go
+++ b/server/commitlog/compact_cleaner_test.go
@@ -19,16 +19,16 @@ type expectedMsg struct {
 	Msg    *proto.Message
 }
 
-// Ensure Clean is a no-op when there are no segments.
+// Ensure Compact is a no-op when there are no segments.
 func TestCompactCleanerNoSegments(t *testing.T) {
 	opts := CompactCleanerOptions{Name: "foo", Logger: noopLogger()}
 	cleaner := NewCompactCleaner(opts)
-	segments, err := cleaner.Clean(0, nil)
+	segments, err := cleaner.Compact(0, nil)
 	require.NoError(t, err)
 	require.Nil(t, segments)
 }
 
-// Ensure Clean is a no-op when there is one segment.
+// Ensure Compact is a no-op when there is one segment.
 func TestCompactCleanerOneSegment(t *testing.T) {
 	opts := CompactCleanerOptions{Name: "foo", Logger: noopLogger()}
 	cleaner := NewCompactCleaner(opts)
@@ -36,12 +36,12 @@ func TestCompactCleanerOneSegment(t *testing.T) {
 	defer remove(t, dir)
 
 	expected := []*Segment{createSegment(t, dir, 0, 100)}
-	actual, err := cleaner.Clean(0, expected)
+	actual, err := cleaner.Compact(0, expected)
 	require.NoError(t, err)
 	require.Equal(t, expected, actual)
 }
 
-// Ensure Clean retains only the latest message for each key up to the last
+// Ensure Compact retains only the latest message for each key up to the last
 // segment.
 func TestCompactCleaner(t *testing.T) {
 	opts := Options{
@@ -91,7 +91,7 @@ func TestCompactCleaner(t *testing.T) {
 	}
 }
 
-// Ensure Clean retains only the latest message for each key up to the HW.
+// Ensure Compact retains only the latest message for each key up to the HW.
 func TestCompactCleanerHW(t *testing.T) {
 	opts := Options{
 		Path:            tempDir(t),
@@ -144,7 +144,7 @@ func TestCompactCleanerHW(t *testing.T) {
 	}
 }
 
-// Ensure Clean retains all messages that do not have keys.
+// Ensure Compact retains all messages that do not have keys.
 func TestCompactCleanerNoKeys(t *testing.T) {
 	opts := Options{
 		Path:            tempDir(t),
diff --git a/server/server_test.go b/server/server_test.go
index d9c6d40..46fd45b 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -116,6 +116,16 @@ func getStreamLeader(t *testing.T, timeout time.Duration, subject, name string,
 	return leader
 }
 
+func forceLogClean(t *testing.T, subject, name string, s *Server) {
+	stream := s.metadata.GetStream(subject, name)
+	if stream == nil {
+		stackFatalf(t, "Stream not found")
+	}
+	if err := stream.log.Clean(); err != nil {
+		stackFatalf(t, "Log clean failed: %s", err)
+	}
+}
+
 // Ensure starting a node fails when there is no seed node to join and no
 // cluster topology is provided.
 func TestNoSeed(t *testing.T) {
@@ -632,6 +642,9 @@ func TestSubscribeOffsetUnderflow(t *testing.T) {
 		t.Fatal("Did not receive expected acks")
 	}
 
+	// Force log clean.
+	forceLogClean(t, stream.Subject, stream.Name, s1)
+
 	// Subscribe with underflowed offset. This should set the offset to 1.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
@@ -717,7 +730,10 @@ func TestStreamRetentionBytes(t *testing.T) {
 		t.Fatal("Did not receive expected acks")
 	}
 
-	// The first message read back should have offset 85.
+	// Force log clean.
+	forceLogClean(t, stream.Subject, stream.Name, s1)
+
+	// The first message read back should have offset 86.
 	msgs := make(chan *proto.Message, 1)
 	ctx, cancel := context.WithCancel(context.Background())
 	err = client.Subscribe(ctx, stream.Subject, stream.Name, func(msg *proto.Message, err error) {
@@ -730,7 +746,7 @@ func TestStreamRetentionBytes(t *testing.T) {
 	// Wait to get the new message.
 	select {
 	case msg := <-msgs:
-		require.Equal(t, int64(85), msg.Offset)
+		require.Equal(t, int64(86), msg.Offset)
 	case <-time.After(5 * time.Second):
 		t.Fatal("Did not receive expected message")
 	}
@@ -748,7 +764,7 @@ func TestStreamRetentionMessages(t *testing.T) {
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.Log.SegmentMaxBytes = 1
-	s1Config.Log.RetentionMaxMessages = 50
+	s1Config.Log.RetentionMaxMessages = 5
 	s1Config.BatchMaxMessages = 1
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
@@ -774,7 +790,7 @@ func TestStreamRetentionMessages(t *testing.T) {
 
 	// Subscribe to acks.
 	acks := "acks"
-	num := 100
+	num := 10
 	acked := 0
 	gotAcks := make(chan struct{})
 	_, err = nc.Subscribe(acks, func(m *nats.Msg) {
@@ -802,7 +818,10 @@ func TestStreamRetentionMessages(t *testing.T) {
 		t.Fatal("Did not receive expected acks")
 	}
 
-	// The first message read back should have offset 49.
+	// Force log clean.
+	forceLogClean(t, stream.Subject, stream.Name, s1)
+
+	// The first message read back should have offset 5.
 	msgs := make(chan *proto.Message, 1)
 	ctx, cancel := context.WithCancel(context.Background())
 	err = client.Subscribe(ctx, stream.Subject, stream.Name, func(msg *proto.Message, err error) {
@@ -815,7 +834,7 @@ func TestStreamRetentionMessages(t *testing.T) {
 	// Wait to get the new message.
 	select {
 	case msg := <-msgs:
-		require.Equal(t, int64(49), msg.Offset)
+		require.Equal(t, int64(5), msg.Offset)
 	case <-time.After(5 * time.Second):
 		t.Fatal("Did not receive expected message")
 	}
@@ -887,6 +906,9 @@ func TestStreamRetentionAge(t *testing.T) {
 		t.Fatal("Did not receive expected acks")
 	}
 
+	// Force log clean.
+	forceLogClean(t, stream.Subject, stream.Name, s1)
+
 	// We expect all segments but the last to be truncated due to age, so the
 	// first message read back should have offset 99.
 	msgs := make(chan *proto.Message, 1)
@@ -1012,6 +1034,9 @@ func TestSubscribeEarliest(t *testing.T) {
 		t.Fatal("Did not receive expected acks")
 	}
 
+	// Force log clean.
+	forceLogClean(t, stream.Subject, stream.Name, s1)
+
 	// Subscribe with EARLIEST. This should start reading from offset 1.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())