diff --git a/internal/datamanager/datamanager.go b/internal/datamanager/datamanager.go index e4de93c..d0ccc4f 100644 --- a/internal/datamanager/datamanager.go +++ b/internal/datamanager/datamanager.go @@ -57,10 +57,9 @@ var ( etcdWalsDataKey = path.Join(etcdWalBaseDir, "walsdata") etcdWalSeqKey = path.Join(etcdWalBaseDir, "walseq") etcdLastCommittedStorageWalSeqKey = path.Join(etcdWalBaseDir, "lastcommittedstoragewalseq") + etcdCheckpointSeqKey = path.Join(etcdWalBaseDir, "checkpointseq") - etcdCheckpointSeqKey = path.Join(etcdWalBaseDir, "checkpointseq") - - etcdInitEtcdLockKey = path.Join(etcdWalBaseDir, "initetcd") + etcdInitEtcdLockKey = path.Join(etcdWalBaseDir, "initetcd") etcdSyncLockKey = path.Join(etcdWalBaseDir, "synclock") etcdCompactChangeGroupsLockKey = path.Join(etcdWalBaseDir, "compactchangegroupslock") etcdCheckpointLockKey = path.Join(etcdWalBaseDir, "checkpointlock") @@ -172,6 +171,26 @@ func etcdWalKey(walSeq string) string { return path.Join(etcdWalsDir, walSeq) } +// deleteEtcd deletes all etcd data excluding keys used for locking +func (d *DataManager) deleteEtcd(ctx context.Context) error { + prefixes := []string{ + etcdWalsDir + "/", + etcdWalsDataKey, + etcdWalSeqKey, + etcdLastCommittedStorageWalSeqKey, + etcdCheckpointSeqKey, + etcdChangeGroupsDir + "/", + etcdChangeGroupMinRevisionKey, + } + for _, prefix := range prefixes { + if err := d.e.DeletePrefix(ctx, prefix); err != nil { + return err + } + } + + return nil +} + func (d *DataManager) Run(ctx context.Context, readyCh chan struct{}) error { for { err := d.InitEtcd(ctx)