Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[3.5] Adding optional revision bump and mark compacted to snapshot restore #16165

Merged
merged 3 commits into from
Jul 4, 2023
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion etcdctl/ctlv3/command/snapshot_command.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ var (
restorePeerURLs string
restoreName string
skipHashCheck bool
markCompacted bool
revisionBump uint64
)

// NewSnapshotCommand returns the cobra command for "snapshot".
Expand Down Expand Up @@ -89,6 +91,8 @@ func NewSnapshotRestoreCommand() *cobra.Command {
cmd.Flags().StringVar(&restorePeerURLs, "initial-advertise-peer-urls", defaultInitialAdvertisePeerURLs, "List of this member's peer URLs to advertise to the rest of the cluster")
cmd.Flags().StringVar(&restoreName, "name", defaultName, "Human-readable name for this member")
cmd.Flags().BoolVar(&skipHashCheck, "skip-hash-check", false, "Ignore snapshot integrity hash value (required if copied from data directory)")
cmd.Flags().Uint64Var(&revisionBump, "bump-revision", 0, "How much to increase the latest revision after restore (required if --mark-compacted)")
cmd.Flags().BoolVar(&markCompacted, "mark-compacted", false, "Mark the latest revision after restore as the point of scheduled compaction (required if --bump-revision > 0)")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please also backport #16166 to 3.5 in this PR or in a separate PR.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, I'll cherry pick it into the PR once the review is done

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done!


return cmd
}
Expand Down Expand Up @@ -127,7 +131,7 @@ func snapshotStatusCommandFunc(cmd *cobra.Command, args []string) {
func snapshotRestoreCommandFunc(cmd *cobra.Command, args []string) {
fmt.Fprintf(os.Stderr, "Deprecated: Use `etcdutl snapshot restore` instead.\n\n")
etcdutl.SnapshotRestoreCommandFunc(restoreCluster, restoreClusterToken, restoreDataDir, restoreWalDir,
restorePeerURLs, restoreName, skipHashCheck, args)
restorePeerURLs, restoreName, skipHashCheck, revisionBump, markCompacted, args)
}

func initialClusterFromName(name string) string {
Expand Down
15 changes: 14 additions & 1 deletion etcdutl/etcdutl/snapshot_command.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ var (
restorePeerURLs string
restoreName string
skipHashCheck bool
markCompacted bool
revisionBump uint64
)

// NewSnapshotCommand returns the cobra command for "snapshot".
Expand Down Expand Up @@ -91,6 +93,8 @@ func NewSnapshotRestoreCommand() *cobra.Command {
cmd.Flags().StringVar(&restorePeerURLs, "initial-advertise-peer-urls", defaultInitialAdvertisePeerURLs, "List of this member's peer URLs to advertise to the rest of the cluster")
cmd.Flags().StringVar(&restoreName, "name", defaultName, "Human-readable name for this member")
cmd.Flags().BoolVar(&skipHashCheck, "skip-hash-check", false, "Ignore snapshot integrity hash value (required if copied from data directory)")
cmd.Flags().Uint64Var(&revisionBump, "bump-revision", 0, "How much to increase the latest revision after restore")
cmd.Flags().BoolVar(&markCompacted, "mark-compacted", false, "Mark the latest revision after restore as the point of scheduled compaction (required if --bump-revision > 0, disallowed otherwise)")

cmd.MarkFlagRequired("data-dir")

Expand All @@ -115,7 +119,7 @@ func SnapshotStatusCommandFunc(cmd *cobra.Command, args []string) {

func snapshotRestoreCommandFunc(_ *cobra.Command, args []string) {
SnapshotRestoreCommandFunc(restoreCluster, restoreClusterToken, restoreDataDir, restoreWalDir,
restorePeerURLs, restoreName, skipHashCheck, args)
restorePeerURLs, restoreName, skipHashCheck, revisionBump, markCompacted, args)
}

func SnapshotRestoreCommandFunc(restoreCluster string,
Expand All @@ -125,12 +129,19 @@ func SnapshotRestoreCommandFunc(restoreCluster string,
restorePeerURLs string,
restoreName string,
skipHashCheck bool,
revisionBump uint64,
markCompacted bool,
args []string) {
if len(args) != 1 {
err := fmt.Errorf("snapshot restore requires exactly one argument")
cobrautl.ExitWithError(cobrautl.ExitBadArgs, err)
}

if (revisionBump == 0 && markCompacted) || (revisionBump > 0 && !markCompacted) {
err := fmt.Errorf("--mark-compacted required if --revision-bump > 0")
cobrautl.ExitWithError(cobrautl.ExitBadArgs, err)
}

dataDir := restoreDataDir
if dataDir == "" {
dataDir = restoreName + ".etcd"
Expand All @@ -153,6 +164,8 @@ func SnapshotRestoreCommandFunc(restoreCluster string,
InitialCluster: restoreCluster,
InitialClusterToken: restoreClusterToken,
SkipHashCheck: skipHashCheck,
RevisionBump: revisionBump,
MarkCompacted: markCompacted,
}); err != nil {
cobrautl.ExitWithError(cobrautl.ExitError, err)
}
Expand Down
22 changes: 22 additions & 0 deletions etcdutl/snapshot/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,31 @@ type revision struct {
sub int64
}

// GreaterThan should be synced with function in server
// https://github.com/etcd-io/etcd/blob/main/server/storage/mvcc/revision.go
func (a revision) GreaterThan(b revision) bool {
if a.main > b.main {
return true
}
if a.main < b.main {
return false
}
return a.sub > b.sub
}

// bytesToRev should be synced with function in server
// https://github.com/etcd-io/etcd/blob/main/server/storage/mvcc/revision.go
func bytesToRev(bytes []byte) revision {
return revision{
main: int64(binary.BigEndian.Uint64(bytes[0:8])),
sub: int64(binary.BigEndian.Uint64(bytes[9:])),
}
}

// revToBytes should be synced with function in server
// https://github.com/etcd-io/etcd/blob/main/server/storage/mvcc/revision.go
func revToBytes(bytes []byte, rev revision) {
binary.BigEndian.PutUint64(bytes[0:8], uint64(rev.main))
bytes[8] = '_'
binary.BigEndian.PutUint64(bytes[9:], uint64(rev.sub))
}
83 changes: 83 additions & 0 deletions etcdutl/snapshot/v3_snapshot.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,9 @@ import (
"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
"go.etcd.io/etcd/server/v3/etcdserver/api/v2store"
"go.etcd.io/etcd/server/v3/etcdserver/cindex"
"go.etcd.io/etcd/server/v3/mvcc"
"go.etcd.io/etcd/server/v3/mvcc/backend"
"go.etcd.io/etcd/server/v3/mvcc/buckets"
"go.etcd.io/etcd/server/v3/verify"
"go.etcd.io/etcd/server/v3/wal"
"go.etcd.io/etcd/server/v3/wal/walpb"
Expand Down Expand Up @@ -194,6 +196,16 @@ type RestoreConfig struct {
// SkipHashCheck is "true" to ignore snapshot integrity hash value
// (required if copied from data directory).
SkipHashCheck bool

// RevisionBump is the amount to increase the latest revision after restore,
// to allow administrators to trick clients into thinking that revision never decreased.
// If 0, revision bumping is skipped.
// (required if MarkCompacted == true)
RevisionBump uint64

// MarkCompacted is "true" to mark the latest revision as compacted.
// (required if RevisionBump > 0)
MarkCompacted bool
}

// Restore restores a new etcd data directory from given snapshot file.
Expand Down Expand Up @@ -257,6 +269,13 @@ func (s *v3Manager) Restore(cfg RestoreConfig) error {
if err = s.saveDB(); err != nil {
return err
}

if cfg.MarkCompacted && cfg.RevisionBump > 0 {
if err = s.modifyLatestRevision(cfg.RevisionBump); err != nil {
return err
}
}

hardstate, err := s.saveWALAndSnap()
if err != nil {
return err
Expand Down Expand Up @@ -303,6 +322,70 @@ func (s *v3Manager) saveDB() error {
return nil
}

// modifyLatestRevision can increase the latest revision by the given amount and sets the scheduled compaction
// to that revision so that the server will consider this revision compacted.
func (s *v3Manager) modifyLatestRevision(bumpAmount uint64) error {
be := backend.NewDefaultBackend(s.outDbPath())
defer func() {
be.ForceCommit()
be.Close()
}()

tx := be.BatchTx()
tx.LockOutsideApply()
defer tx.Unlock()

latest, err := s.unsafeGetLatestRevision(tx)
if err != nil {
return err
}

latest = s.unsafeBumpRevision(tx, latest, int64(bumpAmount))
s.unsafeMarkRevisionCompacted(tx, latest)

return nil
}

func (s *v3Manager) unsafeBumpRevision(tx backend.BatchTx, latest revision, amount int64) revision {
s.lg.Info(
"bumping latest revision",
zap.Int64("latest-revision", latest.main),
zap.Int64("bump-amount", amount),
zap.Int64("new-latest-revision", latest.main+amount),
)

latest.main += amount
latest.sub = 0
k := make([]byte, 17)
revToBytes(k, latest)
tx.UnsafePut(buckets.Key, k, []byte{})

return latest
}

func (s *v3Manager) unsafeMarkRevisionCompacted(tx backend.BatchTx, latest revision) {
s.lg.Info(
"marking revision compacted",
zap.Int64("revision", latest.main),
)

mvcc.UnsafeSetScheduledCompact(tx, latest.main)
}

func (s *v3Manager) unsafeGetLatestRevision(tx backend.BatchTx) (revision, error) {
var latest revision
err := tx.UnsafeForEach(buckets.Key, func(k, _ []byte) (err error) {
rev := bytesToRev(k)

if rev.GreaterThan(latest) {
latest = rev
}

return nil
})
return latest, err
}

func (s *v3Manager) copyAndVerifyDB() error {
srcf, ferr := os.Open(s.srcDbPath)
if ferr != nil {
Expand Down
6 changes: 6 additions & 0 deletions server/mvcc/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,3 +35,9 @@ func WriteKV(be backend.Backend, kv mvccpb.KeyValue) {
be.BatchTx().UnsafePut(buckets.Key, ibytes, d)
be.BatchTx().Unlock()
}

func UnsafeSetScheduledCompact(tx backend.BatchTx, value int64) {
rbytes := newRevBytes()
revToBytes(revision{main: value}, rbytes)
tx.UnsafePut(buckets.Meta, scheduledCompactKeyName, rbytes)
}
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if you have a better place, please let me know

Loading