compactor_test.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111
  1. // Copyright 2015 CoreOS, Inc.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package compactor
  15. import (
  16. "reflect"
  17. "testing"
  18. "time"
  19. "github.com/coreos/etcd/Godeps/_workspace/src/github.com/jonboulle/clockwork"
  20. "github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
  21. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  22. "github.com/coreos/etcd/pkg/testutil"
  23. )
  24. func TestPeriodic(t *testing.T) {
  25. fc := clockwork.NewFakeClock()
  26. compactable := &fakeCompactable{testutil.NewRecorderStream()}
  27. tb := &Periodic{
  28. clock: fc,
  29. periodInHour: 1,
  30. rg: &fakeRevGetter{},
  31. c: compactable,
  32. }
  33. tb.Run()
  34. defer tb.Stop()
  35. n := int(time.Hour / checkCompactionInterval)
  36. for i := 0; i < 3; i++ {
  37. for j := 0; j < n; j++ {
  38. time.Sleep(5 * time.Millisecond)
  39. fc.Advance(checkCompactionInterval)
  40. }
  41. a, err := compactable.Wait(1)
  42. if err != nil {
  43. t.Fatal(err)
  44. }
  45. if !reflect.DeepEqual(a[0].Params[0], &pb.CompactionRequest{Revision: int64(i*n) + 1}) {
  46. t.Errorf("compact request = %v, want %v", a[0].Params[0], &pb.CompactionRequest{Revision: int64(i*n) + 1})
  47. }
  48. }
  49. }
  50. func TestPeriodicPause(t *testing.T) {
  51. fc := clockwork.NewFakeClock()
  52. compactable := &fakeCompactable{testutil.NewRecorderStream()}
  53. tb := &Periodic{
  54. clock: fc,
  55. periodInHour: 1,
  56. rg: &fakeRevGetter{},
  57. c: compactable,
  58. }
  59. tb.Run()
  60. tb.Pause()
  61. n := int(time.Hour / checkCompactionInterval)
  62. for i := 0; i < 3*n; i++ {
  63. time.Sleep(5 * time.Millisecond)
  64. fc.Advance(checkCompactionInterval)
  65. }
  66. select {
  67. case a := <-compactable.Chan():
  68. t.Fatal("unexpected action %v", a)
  69. case <-time.After(10 * time.Millisecond):
  70. }
  71. tb.Resume()
  72. fc.Advance(checkCompactionInterval)
  73. a, err := compactable.Wait(1)
  74. if err != nil {
  75. t.Fatal(err)
  76. }
  77. if !reflect.DeepEqual(a[0].Params[0], &pb.CompactionRequest{Revision: int64(2*n) + 2}) {
  78. t.Errorf("compact request = %v, want %v", a[0].Params[0], &pb.CompactionRequest{Revision: int64(2*n) + 2})
  79. }
  80. }
  81. type fakeCompactable struct {
  82. testutil.Recorder
  83. }
  84. func (fc *fakeCompactable) Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error) {
  85. fc.Record(testutil.Action{Name: "c", Params: []interface{}{r}})
  86. return &pb.CompactionResponse{}, nil
  87. }
  88. type fakeRevGetter struct {
  89. rev int64
  90. }
  91. func (fr *fakeRevGetter) Rev() int64 {
  92. fr.rev++
  93. return fr.rev
  94. }