Skip to content

Commit

Permalink
compactor: Make periodic compactor runs every hour
Browse files Browse the repository at this point in the history
Closes #7868.
  • Loading branch information
Iwasaki Yudai committed May 4, 2017
1 parent 3b251b0 commit a7b2a17
Show file tree
Hide file tree
Showing 2 changed files with 32 additions and 30 deletions.
18 changes: 11 additions & 7 deletions compactor/compactor.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,8 @@ var (
)

const (
checkCompactionInterval = 5 * time.Minute
checkCompactionInterval = 5 * time.Minute
executeCompactionInterval = time.Hour
)

type Compactable interface {
Expand All @@ -41,6 +42,8 @@ type RevGetter interface {
Rev() int64
}

// Periodic compacts the log by purging revisions older than
// the configured retention time. Compaction happenes hourly.
type Periodic struct {
clock clockwork.Clock
periodInHour int
Expand Down Expand Up @@ -85,19 +88,20 @@ func (t *Periodic) Run() {
continue
}
}
if clock.Now().Sub(last) < time.Duration(t.periodInHour)*time.Hour {

if clock.Now().Sub(last) < executeCompactionInterval {
continue
}

rev := t.getRev(t.periodInHour)
rev, remaining := t.getRev(t.periodInHour)
if rev < 0 {
continue
}

plog.Noticef("Starting auto-compaction at revision %d", rev)
_, err := t.c.Compact(t.ctx, &pb.CompactionRequest{Revision: rev})
if err == nil || err == mvcc.ErrCompacted {
t.revs = make([]int64, 0)
t.revs = remaining
last = clock.Now()
plog.Noticef("Finished auto-compaction at revision %d", rev)
} else {
Expand All @@ -124,10 +128,10 @@ func (t *Periodic) Resume() {
t.paused = false
}

func (t *Periodic) getRev(h int) int64 {
func (t *Periodic) getRev(h int) (int64, []int64) {
i := len(t.revs) - int(time.Duration(h)*time.Hour/checkCompactionInterval)
if i < 0 {
return -1
return -1, t.revs
}
return t.revs[i]
return t.revs[i], t.revs[i+1:]
}
44 changes: 21 additions & 23 deletions compactor/compactor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,14 @@ import (
)

func TestPeriodic(t *testing.T) {
retentionHours := 2

fc := clockwork.NewFakeClock()
rg := &fakeRevGetter{testutil.NewRecorderStream(), 0}
compactable := &fakeCompactable{testutil.NewRecorderStream()}
tb := &Periodic{
clock: fc,
periodInHour: 1,
periodInHour: retentionHours,
rg: rg,
c: compactable,
}
Expand All @@ -40,31 +42,26 @@ func TestPeriodic(t *testing.T) {
defer tb.Stop()

n := int(time.Hour / checkCompactionInterval)
// collect 3 hours of revisions
for i := 0; i < 3; i++ {
// advance one (hour - checkCompactionInterval), one revision for each interval
for j := 0; j < n-1; j++ {
_, err := rg.Wait(1)
if err != nil {
t.Fatal(err)
}
// collect 5 hours of revisions
for i := 0; i < 5; i++ {
// advance one hour, one revision for each interval
for j := 0; j < n; j++ {
rg.Wait(1)
fc.Advance(checkCompactionInterval)
}
_, err := rg.Wait(1)
if err != nil {
t.Fatal(err)

// compaction doesn't happen til 2 hours elapses
if i+1 < retentionHours {
continue
}
// ready to acknowledge hour "i"
// block until compactor calls clock.After()
fc.BlockUntil(1)
// unblock the After()
fc.Advance(checkCompactionInterval)

a, err := compactable.Wait(1)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(a[0].Params[0], &pb.CompactionRequest{Revision: int64(i*n) + 1}) {
t.Errorf("compact request = %v, want %v", a[0].Params[0], &pb.CompactionRequest{Revision: int64(i*n) + 1})
expectedRevision := int64(1 + (i+1)*n - retentionHours*n)
if !reflect.DeepEqual(a[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision}) {
t.Errorf("compact request = %v, want %v", a[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision})
}
}

Expand Down Expand Up @@ -92,8 +89,8 @@ func TestPeriodicPause(t *testing.T) {
// tb will collect 3 hours of revisions but not compact since paused
n := int(time.Hour / checkCompactionInterval)
for i := 0; i < 3*n; i++ {
fc.Advance(checkCompactionInterval)
rg.Wait(1)
fc.Advance(checkCompactionInterval)
}
// tb ends up waiting for the clock

Expand All @@ -106,14 +103,15 @@ func TestPeriodicPause(t *testing.T) {
// tb resumes to being blocked on the clock
tb.Resume()

// unblock clock, will kick off a compaction at hour 3
// unblock clock, will kick off a compaction at hour 3:05
rg.Wait(1)
fc.Advance(checkCompactionInterval)
a, err := compactable.Wait(1)
if err != nil {
t.Fatal(err)
}
// compact the revision from hour 2
wreq := &pb.CompactionRequest{Revision: int64(2*n + 1)}
// compact the revision from hour 2:05
wreq := &pb.CompactionRequest{Revision: int64(1 + 2*n + 1)}
if !reflect.DeepEqual(a[0].Params[0], wreq) {
t.Errorf("compact request = %v, want %v", a[0].Params[0], wreq.Revision)
}
Expand Down

0 comments on commit a7b2a17

Please sign in to comment.