diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 3c68d3586b..f5e5e6acec 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -90,6 +90,7 @@ type Topology struct { lastPersistedMaxVolumeId atomic.Uint32 lastPersistedTopologyId atomic.Value lastPersistedGrowthFrozen atomic.Bool + volumeIdAllocator VolumeIdAllocator growthFrozen atomic.Bool growthFrozenAt atomic.Int64 @@ -471,6 +472,17 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { return 0, fmt.Errorf("volume growth is frozen after master state loss; verify the cluster inventory, then run: volume.growth.resume") } } + if a := t.volumeIdAllocator; a != nil { + vid, err := a.NextVolumeId() + if err != nil { + return 0, err + } + if err := t.PersistMaxVolumeId(vid); err != nil { + return 0, fmt.Errorf("can not assign next volume id %d: %w", vid, err) + } + t.UpAdjustMaxVolumeId(vid) + return vid, nil + } vid := t.GetMaxVolumeId() next := vid.Next() @@ -797,7 +809,6 @@ func ttlEqual(a, b *needle.TTL) bool { return a.ToUint32() == b.ToUint32() } - // findDuplicateVolumeLayouts reports every layout that already holds v.Id on // data nodes other than dn. Volume ids are global, so the same id held by two // nodes under two layouts means the allocator reused an id. @@ -1202,6 +1213,10 @@ func (t *Topology) SetGrowthFreezeConfig(freezeOnMismatch bool, recoveryWait tim t.stateLossRecoveryWait.Store(int64(recoveryWait)) } +func (t *Topology) SetVolumeIdAllocator(a VolumeIdAllocator) { + t.volumeIdAllocator = a +} + // CheckReportedTopologyId freezes growth when a volume server reports a // topology id different from this cluster's: the topology id survives raft // state cleanup, so a mismatch means the master's state was rebuilt (or a diff --git a/weed/topology/volume_id_allocator.go b/weed/topology/volume_id_allocator.go new file mode 100644 index 0000000000..695f236785 --- /dev/null +++ b/weed/topology/volume_id_allocator.go @@ -0,0 +1,71 @@ +package topology + +import ( + "context" + "fmt" + "math" + "strconv" + "strings" + "time" + + clientv3 "go.etcd.io/etcd/client/v3" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// VolumeIdAllocator hands out the next volume id. The builtin path advances +// the raft-applied high-water mark; an external implementation makes the +// boundary survive total master state loss. +type VolumeIdAllocator interface { + NextVolumeId() (needle.VolumeId, error) +} + +// EtcdVolumeIdAllocator keeps the volume id counter in a single etcd key, +// advanced by compare-and-swap. Unlike the raft/mirror/echo layers it has no +// residual window: an id cannot be issued without the etcd write committing. +type EtcdVolumeIdAllocator struct { + kv clientv3.KV + key string + timeout time.Duration + seed func() needle.VolumeId +} + +func NewEtcdVolumeIdAllocator(kv clientv3.KV, keyPrefix string, timeout time.Duration, seed func() needle.VolumeId) *EtcdVolumeIdAllocator { + return &EtcdVolumeIdAllocator{ + kv: kv, + key: strings.TrimRight(keyPrefix, "/") + "/max_volume_id", + timeout: timeout, + seed: seed, + } +} + +func (a *EtcdVolumeIdAllocator) NextVolumeId() (needle.VolumeId, error) { + ctx, cancel := context.WithTimeout(context.Background(), a.timeout) + defer cancel() + for { + resp, err := a.kv.Get(ctx, a.key) + if err != nil { + return 0, fmt.Errorf("volume id allocator read: %w", err) + } + var cur uint64 + cmp := clientv3.Compare(clientv3.CreateRevision(a.key), "=", 0) + if len(resp.Kvs) > 0 { + if cur, err = strconv.ParseUint(string(resp.Kvs[0].Value), 10, 32); err != nil { + return 0, fmt.Errorf("volume id counter %q: %w", resp.Kvs[0].Value, err) + } + cmp = clientv3.Compare(clientv3.ModRevision(a.key), "=", resp.Kvs[0].ModRevision) + } else { + cur = uint64(a.seed()) + } + if cur >= math.MaxUint32 { + return 0, fmt.Errorf("volume id counter exhausted at %d", cur) + } + committed, err := a.kv.Txn(ctx).If(cmp).Then(clientv3.OpPut(a.key, strconv.FormatUint(cur+1, 10))).Commit() + if err != nil { + return 0, fmt.Errorf("volume id allocator commit: %w", err) + } + if committed.Succeeded { + return needle.VolumeId(cur + 1), nil + } + } +}