topology: pluggable VolumeIdAllocator with an etcd CAS backend (#11763)

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

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