Merge pull request #132848 from serathius/etcd3-compactor

Expose compaction revision from compactor

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
 }

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)
+	}
 }

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++