Skip to content

Commit

Permalink
compactor: adjust interval for period <1-hour
Browse files Browse the repository at this point in the history 
Signed-off-by: Gyuho Lee <gyuhox@gmail.com>
  • Loading branch information
@gyuho
gyuho committed Mar 22, 2018
1 parent a3b7fbd commit e36779f
Show file tree
Hide file tree
Showing 2 changed files with 82 additions and 62 deletions.
30 changes: 15 additions & 15 deletions compactor/periodic.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ const periodDivisor = 10

// Run runs periodic compactor.
func (t *Periodic) Run() {
interval := t.period / time.Duration(periodDivisor)
interval := t.getInterval()
go func() {
initialWait := t.clock.Now()
for {
Expand All @@ -87,17 +87,13 @@ func (t *Periodic) Run() {
if t.clock.Now().Sub(initialWait) < t.period {
continue
}

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

plog.Noticef("Starting auto-compaction at revision %d (retention: %v)", rev, t.period)
_, err := t.c.Compact(t.ctx, &pb.CompactionRequest{Revision: rev})
if err == nil || err == mvcc.ErrCompacted {
// move to next sliding window
t.revs = remaining
t.revs = t.revs[1:]
plog.Noticef("Finished auto-compaction at revision %d", rev)
} else {
plog.Noticef("Failed auto-compaction at revision %d (%v)", rev, err)
Expand All @@ -107,6 +103,18 @@ func (t *Periodic) Run() {
}()
}

// if given compaction period x is <1-hour, compact every x duration, with x retention windown.
// (e.g. --auto-compaction-mode 'periodic' --auto-compaction-retention='10m', then compact every 10-minute)
// if given compaction period x is >1-hour, compact every 1/x duration, with x retention windown.
// (e.g. --auto-compaction-mode 'periodic' --auto-compaction-retention='72h', then compact every 7.2-hour)
func (t *Periodic) getInterval() time.Duration {
itv := t.period
if itv > time.Hour {
itv /= 10
}
return itv
}

// Stop stops periodic compactor.
func (t *Periodic) Stop() {
t.cancel()
Expand All @@ -125,11 +133,3 @@ func (t *Periodic) Resume() {
defer t.mu.Unlock()
t.paused = false
}

func (t *Periodic) getRev() (int64, []int64) {
i := len(t.revs) - periodDivisor
if i < 0 {
return -1, t.revs
}
return t.revs[i], t.revs[i+1:]
}
114 changes: 67 additions & 47 deletions compactor/periodic_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,83 +25,103 @@ import (
"github.com/jonboulle/clockwork"
)

func TestPeriodic(t *testing.T) {
retentionHours := 2
retentionDuration := time.Duration(retentionHours) * time.Hour
func TestPeriodicHourly(t *testing.T) {
fc := clockwork.NewFakeClock()
rg := &fakeRevGetter{testutil.NewRecorderStream(), 0}
compactable := &fakeCompactable{testutil.NewRecorderStream()}
tb := newPeriodic(fc, 72*time.Hour, rg, compactable)

tb.Run()
defer tb.Stop()

for i := 0; i < 20; i++ {
// first 72-hour only with rev gets
if _, err := rg.Wait(1); err != nil {
t.Fatal(err)
}
fc.Advance(tb.getInterval())

// after 72-hour, periodic compact begins, every hour
// with 72-hour retention window
if i >= 9 {
ca, err := compactable.Wait(1)
if err != nil {
t.Fatal(err)
}
expectedRevision := int64(i + 2 - 10)
if !reflect.DeepEqual(ca[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision}) {
t.Errorf("compact request = %v, want %v", ca[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision})
}
}
}
}

func TestPeriodicEveryMinute(t *testing.T) {
fc := clockwork.NewFakeClock()
rg := &fakeRevGetter{testutil.NewRecorderStream(), 0}
compactable := &fakeCompactable{testutil.NewRecorderStream()}
tb := newPeriodic(fc, retentionDuration, rg, compactable)
tb := newPeriodic(fc, time.Minute, rg, compactable)

tb.Run()
defer tb.Stop()
checkCompactInterval := retentionDuration / time.Duration(periodDivisor)
n := periodDivisor
// simulate 5 hours worth of intervals.
for i := 0; i < n/retentionHours*5; i++ {
rg.Wait(1)
fc.Advance(checkCompactInterval)
// compaction doesn't happen til 2 hours elapses.
if i < n {
continue

// expect compact every minute
for i := 0; i < 10; i++ {
if _, err := rg.Wait(1); err != nil {
t.Fatal(err)
}
// after 2 hours, compaction happens at every checkCompactInterval.
a, err := compactable.Wait(1)
fc.Advance(time.Minute)

ca, err := compactable.Wait(1)
if err != nil {
t.Fatal(err)
}
expectedRevision := int64(i + 1 - 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})
expectedRevision := int64(i + 1)
if !reflect.DeepEqual(ca[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision}) {
t.Errorf("compact request = %v, want %v", ca[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision})
}
}

// unblock the rev getter, so we can stop the compactor routine.
_, err := rg.Wait(1)
if err != nil {
t.Fatal(err)
}
}

func TestPeriodicPause(t *testing.T) {
func TestPeriodicPauseHourly(t *testing.T) {
fc := clockwork.NewFakeClock()
retentionDuration := time.Hour
rg := &fakeRevGetter{testutil.NewRecorderStream(), 0}
compactable := &fakeCompactable{testutil.NewRecorderStream()}
tb := newPeriodic(fc, retentionDuration, rg, compactable)
tb := newPeriodic(fc, 15*time.Hour, rg, compactable)

tb.Run()
defer tb.Stop()

tb.Pause()

// tb will collect 3 hours of revisions but not compact since paused
checkCompactInterval := retentionDuration / time.Duration(periodDivisor)
n := periodDivisor
for i := 0; i < 3*n; i++ {
rg.Wait(1)
fc.Advance(checkCompactInterval)
// collect 15*2 hours of revisions with no compaction
for i := 0; i < 15*2; i++ {
if _, err := rg.Wait(1); err != nil {
t.Fatal(err)
}
fc.Advance(tb.getInterval())
}
// tb ends up waiting for the clock

select {
case a := <-compactable.Chan():
t.Fatalf("unexpected action %v", a)
case <-time.After(10 * time.Millisecond):
}

// tb resumes to being blocked on the clock
tb.Resume()

// unblock clock, will kick off a compaction at hour 3:06
rg.Wait(1)
fc.Advance(checkCompactInterval)
a, err := compactable.Wait(1)
if err != nil {
t.Fatal(err)
}
// compact the revision from hour 2:06
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)
for i := 0; i < 20; i++ {
if _, err := rg.Wait(1); err != nil {
t.Fatal(err)
}
fc.Advance(tb.getInterval())

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

0 comments on commit e36779f

Please sign in to comment.