diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact.go index f279c6cabba..e431bd16e04 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact.go @@ -23,7 +23,9 @@ import ( "time" clientv3 "go.etcd.io/etcd/client/v3" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/klog/v2" + "k8s.io/utils/clock" ) const ( @@ -32,36 +34,95 @@ const ( var ( endpointsMapMu sync.Mutex - endpointsMap map[string]struct{} + endpointsMap map[string]*compactor ) func init() { - endpointsMap = make(map[string]struct{}) + endpointsMap = make(map[string]*compactor) } // StartCompactor starts a compactor in the background to compact old version of keys that's not needed. // By default, we save the most recent 5 minutes data and compact versions > 5minutes ago. // It should be enough for slow watchers and to tolerate burst. // TODO: We might keep a longer history (12h) in the future once storage API can take advantage of past version of keys. -func StartCompactor(ctx context.Context, client *clientv3.Client, compactInterval time.Duration) { +func StartCompactor(client *clientv3.Client, compactInterval time.Duration) *compactor { endpointsMapMu.Lock() defer endpointsMapMu.Unlock() // In one process, we can have only one compactor for one cluster. // Currently we rely on endpoints to differentiate clusters. for _, ep := range client.Endpoints() { - if _, ok := endpointsMap[ep]; ok { + if c, ok := endpointsMap[ep]; ok { klog.V(4).Infof("compactor already exists for endpoints %v", client.Endpoints()) - return + return c } } + c := newCompactor(client, compactInterval, clock.RealClock{}, func() { + endpointsMapMu.Lock() + defer endpointsMapMu.Unlock() + for _, ep := range client.Endpoints() { + delete(endpointsMap, ep) + } + }) for _, ep := range client.Endpoints() { - endpointsMap[ep] = struct{}{} + endpointsMap[ep] = c } + return c +} - if compactInterval != 0 { - go compactor(ctx, client, compactInterval) +func newCompactor(client *clientv3.Client, compactInterval time.Duration, clock clock.Clock, onClose func()) *compactor { + c := &compactor{ + client: client, + interval: compactInterval, + stop: make(chan struct{}), + clock: clock, + onClose: onClose, } + if compactInterval != 0 { + c.wg.Add(1) + go func() { + defer c.wg.Done() + c.runCompactLoop(c.stop) + }() + } + return c +} + +type compactor struct { + client *clientv3.Client + wg sync.WaitGroup + clock clock.Clock + onClose func() + + stopOnce sync.Once + stop chan struct{} + + mux sync.Mutex + compactRevision int64 + interval time.Duration +} + +func (c *compactor) Stop() { + // Prevent integration tests stopping compactor twice. + c.stopOnce.Do(func() { + if c.onClose != nil { + c.onClose() + } + close(c.stop) + c.wg.Wait() + }) +} + +func (c *compactor) CompactRevision() int64 { + c.mux.Lock() + defer c.mux.Unlock() + return c.compactRevision +} + +func (c *compactor) SetCompactRevision(rev int64) { + c.mux.Lock() + defer c.mux.Unlock() + c.compactRevision = rev } // compactor periodically compacts historical versions of keys in etcd. @@ -69,7 +130,7 @@ func StartCompactor(ctx context.Context, client *clientv3.Client, compactInterva // In other words, after compaction, it will only contain keys set during last interval. // Any API call for the older versions of keys will return error. // Interval is the time interval between each compaction. The first compaction happens after "interval". -func compactor(ctx context.Context, client *clientv3.Client, interval time.Duration) { +func (c *compactor) runCompactLoop(stopCh chan struct{}) { // Technical definitions: // We have a special key in etcd defined as *compactRevKey*. // compactRevKey's value will be set to the string of last compacted revision. @@ -109,54 +170,63 @@ func compactor(ctx context.Context, client *clientv3.Client, interval time.Durat // - What happened under heavy load scenarios? Initially, each apiserver will do only one compaction // every 5 minutes. This is very unlikely affecting or affected w.r.t. server load. + ctx := wait.ContextForChannel(stopCh) var compactTime int64 var rev int64 + var compactRev int64 var err error for { select { - case <-time.After(interval): + case <-c.clock.After(c.interval): case <-ctx.Done(): return } - compactTime, rev, err = compact(ctx, client, compactTime, rev) + compactTime, rev, compactRev, err = compact(ctx, c.client, compactTime, rev) if err != nil { - klog.Errorf("etcd: endpoint (%v) compact failed: %v", client.Endpoints(), err) + klog.Errorf("etcd: endpoint (%v) compact failed: %v", c.client.Endpoints(), err) continue } + if compactRev != 0 { + c.SetCompactRevision(compactRev) + } } } // compact compacts etcd store and returns current rev. // It will return the current compact time and global revision if no error occurred. // Note that CAS fail will not incur any error. -func compact(ctx context.Context, client *clientv3.Client, t, rev int64) (int64, int64, error) { +func compact(ctx context.Context, client *clientv3.Client, expectVersion, rev int64) (currentVersion, currentRev, compactRev int64, err error) { resp, err := client.KV.Txn(ctx).If( - clientv3.Compare(clientv3.Version(compactRevKey), "=", t), + clientv3.Compare(clientv3.Version(compactRevKey), "=", expectVersion), ).Then( clientv3.OpPut(compactRevKey, strconv.FormatInt(rev, 10)), // Expect side effect: increment Version ).Else( clientv3.OpGet(compactRevKey), ).Commit() if err != nil { - return t, rev, err + return expectVersion, rev, 0, err } - curRev := resp.Header.Revision + currentRev = resp.Header.Revision if !resp.Succeeded { - curTime := resp.Responses[0].GetResponseRange().Kvs[0].Version - return curTime, curRev, nil + currentVersion = resp.Responses[0].GetResponseRange().Kvs[0].Version + compactRev, err = strconv.ParseInt(string(resp.Responses[0].GetResponseRange().Kvs[0].Value), 10, 64) + if err != nil { + return currentVersion, currentRev, 0, nil + } + return currentVersion, currentRev, compactRev, nil } - curTime := t + 1 + currentVersion = expectVersion + 1 if rev == 0 { // We don't compact on bootstrap. - return curTime, curRev, nil + return currentVersion, currentRev, 0, nil } if _, err = client.Compact(ctx, rev); err != nil { - return curTime, curRev, err + return currentVersion, currentRev, 0, err } klog.V(4).Infof("etcd: compacted rev (%d), endpoints (%v)", rev, client.Endpoints()) - return curTime, curRev, nil + return currentVersion, currentRev, rev, nil } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact_test.go index 97c61e6ee17..aec7a9c2510 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact_test.go @@ -19,42 +19,138 @@ package etcd3 import ( "context" "testing" + "time" etcdrpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" clientv3 "go.etcd.io/etcd/client/v3" "k8s.io/apiserver/pkg/storage/etcd3/testserver" + testingclock "k8s.io/utils/clock/testing" +) + +const ( + waitDelay = time.Millisecond + waitTimeout = 100 * waitDelay ) func TestCompact(t *testing.T) { client := testserver.RunEtcd(t, nil).Client ctx := context.Background() + clock := testingclock.NewFakeClock(time.Now()) + c := newCompactor(client, time.Minute, clock, nil) + t.Cleanup(c.Stop) + waitForClockWaiters(t, clock) + t.Log("First compaction cycle saves revision before first write") + clock.Step(time.Minute) + waitForClockWaiters(t, clock) + compactRev := c.CompactRevision() + if compactRev != 0 { + t.Errorf("CompactRevision()=%d, expected %d", compactRev, 0) + } + + t.Log("First write") putResp, err := client.Put(ctx, "/somekey", "data") if err != nil { t.Fatalf("Put failed: %v", err) } + assertNotCompacted(t, ctx, client, putResp.Header.Revision) + t.Log("Second compaction cycle compacts before first write") + clock.Step(time.Minute) + waitForClockWaiters(t, clock) + assertNotCompacted(t, ctx, client, putResp.Header.Revision) + compactRev = c.CompactRevision() + if compactRev != putResp.Header.Revision-1 { + t.Errorf("CompactRevision()=%d, expected %d", compactRev, 0) + } + + t.Log("Create second revision") putResp1, err := client.Put(ctx, "/somekey", "data2") if err != nil { t.Fatalf("Put failed: %v", err) } + assertNotCompacted(t, ctx, client, putResp1.Header.Revision) - _, _, err = compact(ctx, client, 0, putResp1.Header.Revision) - if err != nil { - t.Fatalf("compact failed: %v", err) + t.Log("Third compaction cycle compacts revision after first write") + clock.Step(time.Minute) + waitForClockWaiters(t, clock) + assertCompacted(t, ctx, client, putResp.Header.Revision) + compactRev = c.CompactRevision() + if compactRev != putResp.Header.Revision+1 { + t.Errorf("CompactRevision()=%d, expected %d", compactRev, putResp.Header.Revision) } - obj, err := client.Get(ctx, "/somekey", clientv3.WithRev(putResp.Header.Revision)) - if err != etcdrpc.ErrCompacted { - t.Errorf("Expecting ErrCompacted, but get=%v err=%v", obj, err) + assertNotCompacted(t, ctx, client, putResp1.Header.Revision) + + t.Log("Fourth compaction cycle compacts second write") + clock.Step(time.Minute) + waitForClockWaiters(t, clock) + assertCompacted(t, ctx, client, putResp.Header.Revision) + assertCompacted(t, ctx, client, putResp1.Header.Revision) + compactRev = c.CompactRevision() + if compactRev != putResp1.Header.Revision+1 { + t.Errorf("CompactRevision()=%d, expected %d", compactRev, putResp1.Header.Revision) } } -// TestCompactConflict tests that two compactors (Let's use C1, C2) are trying to compact etcd cluster with the same +func assertCompacted(t *testing.T, ctx context.Context, client *clientv3.Client, rev int64) { + t.Helper() + _, err := client.Get(ctx, "/somekey", clientv3.WithRev(rev)) + if err != etcdrpc.ErrCompacted { + t.Errorf("Expecting rev %d compacted, but err=%v", rev, err) + } +} + +func assertNotCompacted(t *testing.T, ctx context.Context, client *clientv3.Client, rev int64) { + t.Helper() + _, err := client.Get(ctx, "/somekey", clientv3.WithRev(rev)) + if err != nil { + t.Errorf("Get on rev %d failed: %v", rev, err) + } +} + +func TestCompactIntervalZero(t *testing.T) { + client := testserver.RunEtcd(t, nil).Client + clock := testingclock.NewFakeClock(time.Now()) + c := newCompactor(client, 0, clock, nil) + t.Cleanup(c.Stop) + + t.Log("Compact loop is disabled, no goroutine is waiting on clock") + clockNoWaiters(t, clock) + clock.Step(time.Minute) + clockNoWaiters(t, clock) +} + +func waitForClockWaiters(t *testing.T, clock *testingclock.FakeClock) { + t.Helper() + for start := time.Now(); time.Since(start) < waitTimeout; { + if clock.HasWaiters() { + return + } + time.Sleep(waitDelay) + } + t.Fatal("No waiters") +} + +func clockNoWaiters(t *testing.T, clock *testingclock.FakeClock) { + t.Helper() + for start := time.Now(); time.Since(start) < waitTimeout; { + if clock.Waiters() != 0 { + t.Fatal("waiter") + } + time.Sleep(waitDelay) + } + if clock.Waiters() != 0 { + t.Fatal("waiter") + } +} + +// TestCompactConflict tests that multiple compactors are trying to compact etcd cluster with the same // logical time. -// - C1 compacts first. It will succeed. -// - C2 compacts after. It will fail. But it will get latest logical time, which should be larger by one. +// - C1 compacts on time 0. It will succeed. +// - C2 compacts on time 0. It will fail as this time was compacted. But it will get latest logical time, which should be larger by one. +// - C3 compacts on time 1. It will succeed. func TestCompactConflict(t *testing.T) { client := testserver.RunEtcd(t, nil).Client ctx := context.Background() @@ -64,21 +160,61 @@ func TestCompactConflict(t *testing.T) { t.Fatalf("Put failed: %v", err) } - // Compact first. It would do the compaction and return compact time which is incremented by 1. - curTime, _, err := compact(ctx, client, 0, putResp.Header.Revision) + t.Log("First compact on time 0") + wantCompactRev := putResp.Header.Revision + curTime, curRev, compactRev, err := compact(ctx, client, 0, wantCompactRev) if err != nil { t.Fatalf("compact failed: %v", err) } + t.Log("Compaction should succeed") + if compactRev != wantCompactRev { + t.Errorf("Expect compact revision = %d, get = %d", wantCompactRev, compactRev) + } + t.Log("Current time should increase by 1") if curTime != 1 { t.Errorf("Expect current logical time = 1, get = %v", curTime) } + t.Log("Current revision should increase by 1") + wantCurrentRev := putResp.Header.Revision + 1 + if curRev != wantCurrentRev { + t.Errorf("Expect current revision = %d, get = %d", wantCurrentRev, curRev) + } - // Compact again with the same parameters. It won't do compaction but return the latest compact time. - curTime2, _, err := compact(ctx, client, 0, putResp.Header.Revision) + t.Log("Second compact on time 0") + curTime2, curRev2, compactRev2, err := compact(ctx, client, 0, wantCompactRev+1) if err != nil { t.Fatalf("compact failed: %v", err) } + t.Log("Compaction should fail") + if compactRev2 != wantCompactRev { + t.Errorf("Expect compact revision = %d, get = %d", wantCompactRev, compactRev2) + } + t.Log("Should return same time as from the first compacty") if curTime != curTime2 { t.Errorf("Unexpected curTime (%v) != curTime2 (%v)", curTime, curTime2) } + t.Log("Current revision should stay the same") + if curRev2 != wantCurrentRev { + t.Errorf("Expect current revision = %d, get = %d", wantCurrentRev, curRev2) + } + + t.Log("Third compact on time 1") + curTime3, curRev3, compactRev3, err := compact(ctx, client, 1, wantCompactRev+1) + if err != nil { + t.Fatalf("compact failed: %v", err) + } + t.Log("Compaction should succeed") + wantCompactRev += 1 + if compactRev3 != wantCompactRev { + t.Errorf("Expect compact revision = %d, get = %d", wantCompactRev, compactRev3) + } + t.Log("Current time should increase by 1") + if curTime3 != 2 { + t.Errorf("Expect current logical time = 2, get = %v", curTime3) + } + t.Log("Current revision should increase by 1") + wantCurrentRev += 1 + if curRev3 != wantCurrentRev { + t.Errorf("Expect current revision = %d, get = %d", wantCurrentRev, curRev3) + } } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd3.go b/staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd3.go index 0006afe0e3e..00f7bf98db2 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd3.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd3.go @@ -357,8 +357,8 @@ var newETCD3Client = func(c storagebackend.TransportConfig) (*kubernetes.Client, type runningCompactor struct { interval time.Duration - cancel context.CancelFunc client *clientv3.Client + cancel DestroyFunc refs int } @@ -393,20 +393,16 @@ func startCompactorOnce(c storagebackend.TransportConfig, interval time.Duration if foundBefore { // replace compactor compactor.cancel() - compactor.client.Close() } else { // start new compactor compactor = &runningCompactor{} compactors[key] = compactor } - ctx, cancel := context.WithCancel(context.Background()) - compactor.interval = interval - compactor.cancel = cancel compactor.client = compactorClient - - etcd3.StartCompactor(ctx, compactorClient, interval) + c := etcd3.StartCompactor(compactorClient, interval) + compactor.cancel = c.Stop } compactors[key].refs++