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 34241ef commit bbf1555
Show file tree
Hide file tree
Showing 2 changed files with 86 additions and 65 deletions.
37 changes: 19 additions & 18 deletions compactor/periodic.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,12 +61,9 @@ func newPeriodic(clock clockwork.Clock, h time.Duration, rg RevGetter, c Compact
return t
}

// periodDivisor divides Periodic.period in into checkCompactInterval duration
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 @@ -75,6 +72,8 @@ func (t *Periodic) Run() {
case <-t.ctx.Done():
return
case <-t.clock.After(interval):
// reset interval
interval = t.getInterval()
t.mu.Lock()
p := t.paused
t.mu.Unlock()
Expand All @@ -87,26 +86,36 @@ 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 {
// retry fast
interval = t.getInterval() / 10
plog.Noticef("Failed auto-compaction at revision %d (%v)", rev, err)
plog.Noticef("Retry after %v", interval)
}
}
}()
}

// If given compaction period x is <1-hour, compact every x duration, with x retention window
// (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-hour, with x retention window
// (e.g. --auto-compaction-mode 'periodic' --auto-compaction-retention='72h', then compact every 1-hour).
func (t *Periodic) getInterval() time.Duration {
itv := t.period
if itv > time.Hour {
itv = time.Hour
}
return itv
}

// Stop stops periodic compactor.
func (t *Periodic) Stop() {
t.cancel()
Expand All @@ -125,11 +134,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, 12*time.Hour, rg, compactable)

tb.Run()
defer tb.Stop()

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

// after 12-hour, periodic compact begins, every hour
// with 12-hour retention window
if i >= 11 {
ca, err := compactable.Wait(1)
if err != nil {
t.Fatal(err)
}
expectedRevision := int64(i + 2 - int(tb.period/time.Hour))
if !reflect.DeepEqual(ca[0].Params[0], &pb.CompactionRequest{Revision: expectedRevision}) {
t.Fatalf("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 bbf1555

Please sign in to comment.