From dcda47dc29a43560b593cdbadc7bd7122dd3eb1b Mon Sep 17 00:00:00 2001 From: Wei Fu Date: Wed, 24 Apr 2024 10:30:36 +0800 Subject: [PATCH 1/5] mvcc: introduce compactBeforeSetFinishedCompact failpoint Signed-off-by: Wei Fu --- mvcc/kvstore_compaction.go | 1 + 1 file changed, 1 insertion(+) diff --git a/mvcc/kvstore_compaction.go b/mvcc/kvstore_compaction.go index 963ebe950d3..10035e897a8 100644 --- a/mvcc/kvstore_compaction.go +++ b/mvcc/kvstore_compaction.go @@ -49,6 +49,7 @@ func (s *store) scheduleCompaction(compactMainRev int64, keep map[revision]struc } if len(keys) < s.cfg.CompactionBatchLimit { + // gofail: var compactBeforeSetFinishedCompact struct{} rbytes := make([]byte, 8+1+8) revToBytes(revision{main: compactMainRev}, rbytes) tx.UnsafePut(metaBucketName, finishedCompactKeyName, rbytes) From 844c4b02f73518e4aaa267bb3e497fc2cc273506 Mon Sep 17 00:00:00 2001 From: Wei Fu Date: Wed, 24 Apr 2024 11:00:07 +0800 Subject: [PATCH 2/5] tests/e2e: support CompactionBatchLimit flag Signed-off-by: Wei Fu --- tests/e2e/cluster_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/e2e/cluster_test.go b/tests/e2e/cluster_test.go index 982eb3a820b..97cae3898c6 100644 --- a/tests/e2e/cluster_test.go +++ b/tests/e2e/cluster_test.go @@ -140,6 +140,7 @@ type etcdProcessClusterConfig struct { MaxConcurrentStreams uint32 // default is math.MaxUint32 WatchProcessNotifyInterval time.Duration + CompactionBatchLimit int debug bool @@ -333,6 +334,9 @@ func (cfg *etcdProcessClusterConfig) etcdServerProcessConfigs() []*etcdServerPro if cfg.WatchProcessNotifyInterval != 0 { args = append(args, "--experimental-watch-progress-notify-interval", cfg.WatchProcessNotifyInterval.String()) } + if cfg.CompactionBatchLimit != 0 { + args = append(args, "--experimental-compaction-batch-limit", fmt.Sprintf("%d", cfg.CompactionBatchLimit)) + } if cfg.debug { args = append(args, "--debug") From a9727e60164fe9a82996b979d0ea08d62e3fd695 Mon Sep 17 00:00:00 2001 From: Wei Fu Date: Wed, 17 Apr 2024 20:31:48 +0800 Subject: [PATCH 3/5] tests/e2e: reproduce #17780 Signed-off-by: Wei Fu (cherry picked from commit 71733911544f8fce6d06d2a8e9cca0944b3659be) Signed-off-by: Wei Fu --- tests/e2e/reproduce_17780_test.go | 114 ++++++++++++++++++++++++++++++ 1 file changed, 114 insertions(+) create mode 100644 tests/e2e/reproduce_17780_test.go diff --git a/tests/e2e/reproduce_17780_test.go b/tests/e2e/reproduce_17780_test.go new file mode 100644 index 00000000000..c31ded33be4 --- /dev/null +++ b/tests/e2e/reproduce_17780_test.go @@ -0,0 +1,114 @@ +// Copyright 2024 The etcd Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.etcd.io/etcd/clientv3" + "go.etcd.io/etcd/pkg/stringutil" + "go.etcd.io/etcd/pkg/testutil" +) + +// TestReproduce17780 reproduces the issue: https://github.com/etcd-io/etcd/issues/17780. +func TestReproduce17780(t *testing.T) { + defer testutil.AfterTest(t) + + compactionBatchLimit := 10 + + ctx := context.TODO() + clus, cerr := newEtcdProcessCluster(t, &etcdProcessClusterConfig{ + clusterSize: 3, + goFailEnabled: true, + goFailClientTimeout: 40 * time.Second, + snapshotCount: 1000, + CompactionBatchLimit: compactionBatchLimit, + WatchProcessNotifyInterval: 100 * time.Millisecond, + }) + require.NoError(t, cerr) + + t.Cleanup(func() { require.NoError(t, clus.Stop()) }) + + leaderIdx := clus.WaitLeader(t) + targetIdx := (leaderIdx + 1) % clus.cfg.clusterSize + + cli := newClient(t, clus.procs[targetIdx].EndpointsGRPC(), clientNonTLS, false) + + // Revision: 2 -> 8 for new keys + n := compactionBatchLimit - 2 + valueSize := 16 + for i := 2; i <= n; i++ { + _, err := cli.Put(ctx, fmt.Sprintf("%d", i), stringutil.RandString(uint(valueSize))) + require.NoError(t, err) + } + + // Revision: 9 -> 11 for delete keys with compared revision + // + // We need last compaction batch is no-op and all the tombstones should + // be deleted in previous compaction batch. So that we just lost the + // finishedCompactRev after panic. + for i := 9; i <= compactionBatchLimit+1; i++ { + rev := i - 5 + key := fmt.Sprintf("%d", rev) + + _, err := cli.Delete(ctx, key) + require.NoError(t, err) + } + + require.NoError(t, clus.procs[targetIdx].Failpoints().SetupHTTP(ctx, "compactBeforeSetFinishedCompact", `panic`)) + + _, err := cli.Compact(ctx, 11, clientv3.WithCompactPhysical()) + require.Error(t, err) + + require.Error(t, clus.procs[targetIdx].Stop()) + // NOTE: The proc panics and exit code is 2. It's impossible to restart + // that etcd proc because last exit code is 2 and Restart() refuses to + // start new one. Using IsRunning() function is to cleanup status. + require.False(t, clus.procs[targetIdx].IsRunning()) + require.NoError(t, clus.procs[targetIdx].Restart()) + + // NOTE: We should not decrease the revision if there is no record + // about finished compact operation. + resp, err := cli.Get(ctx, fmt.Sprintf("%d", n)) + require.NoError(t, err) + assert.GreaterOrEqual(t, resp.Header.Revision, int64(11)) + + // Revision 4 should be deleted by compaction. + resp, err = cli.Get(ctx, fmt.Sprintf("%d", 4)) + require.NoError(t, err) + require.True(t, resp.Count == 0) + + next := 20 + for i := 12; i <= next; i++ { + _, err := cli.Put(ctx, fmt.Sprintf("%d", i), stringutil.RandString(uint(valueSize))) + require.NoError(t, err) + } + + expectedRevision := next + for procIdx, proc := range clus.procs { + cli = newClient(t, proc.EndpointsGRPC(), clientNonTLS, false) + resp, err := cli.Get(ctx, fmt.Sprintf("%d", next)) + require.NoError(t, err) + + assert.GreaterOrEqual(t, resp.Header.Revision, int64(expectedRevision), + fmt.Sprintf("LeaderIdx: %d, Current: %d", leaderIdx, procIdx)) + } +} From 41eb03a55e3e313b89b02ef51eed4b39dc35301b Mon Sep 17 00:00:00 2001 From: Wei Fu Date: Wed, 24 Apr 2024 11:13:34 +0800 Subject: [PATCH 4/5] mvcc: update currentRev if scheduledCompact > currentRev Signed-off-by: Wei Fu (cherry picked from commit 9ea234913a99670d18b66aa23915781f89713177) Signed-off-by: Wei Fu --- mvcc/kvstore.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/mvcc/kvstore.go b/mvcc/kvstore.go index 57bb823b84e..ba114273d73 100644 --- a/mvcc/kvstore.go +++ b/mvcc/kvstore.go @@ -474,6 +474,17 @@ func (s *store) restore() error { } } + // If the latest revision was a tombstone revision and etcd just compacted + // it, but crashed right before persisting the FinishedCompactRevision, + // then it would lead to revision decreasing in bbolt db file. In such + // a scenario, we should adjust the current revision using the scheduled + // compact revision on bootstrap when etcd gets started again. + // + // See https://github.com/etcd-io/etcd/issues/17780#issuecomment-2061900231 + if s.currentRev < scheduledCompact { + s.currentRev = scheduledCompact + } + tx.Unlock() if scheduledCompact != 0 { From 4cb197e0faaed1551601cec9f225a44cf54358f5 Mon Sep 17 00:00:00 2001 From: Wei Fu Date: Wed, 24 Apr 2024 11:15:26 +0800 Subject: [PATCH 5/5] mvcc: should update currentRev in revMu Signed-off-by: Wei Fu --- mvcc/kvstore.go | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/mvcc/kvstore.go b/mvcc/kvstore.go index ba114273d73..098e684f6e8 100644 --- a/mvcc/kvstore.go +++ b/mvcc/kvstore.go @@ -450,6 +450,17 @@ func (s *store) restore() error { s.currentRev = s.compactMainRev } + // If the latest revision was a tombstone revision and etcd just compacted + // it, but crashed right before persisting the FinishedCompactRevision, + // then it would lead to revision decreasing in bbolt db file. In such + // a scenario, we should adjust the current revision using the scheduled + // compact revision on bootstrap when etcd gets started again. + // + // See https://github.com/etcd-io/etcd/issues/17780#issuecomment-2061900231 + if s.currentRev < scheduledCompact { + s.currentRev = scheduledCompact + } + if scheduledCompact <= s.compactMainRev { scheduledCompact = 0 } @@ -474,17 +485,6 @@ func (s *store) restore() error { } } - // If the latest revision was a tombstone revision and etcd just compacted - // it, but crashed right before persisting the FinishedCompactRevision, - // then it would lead to revision decreasing in bbolt db file. In such - // a scenario, we should adjust the current revision using the scheduled - // compact revision on bootstrap when etcd gets started again. - // - // See https://github.com/etcd-io/etcd/issues/17780#issuecomment-2061900231 - if s.currentRev < scheduledCompact { - s.currentRev = scheduledCompact - } - tx.Unlock() if scheduledCompact != 0 {