master: wire the etcd volume id allocator behind master.volume_id_allocator=etcd (#11763)

2026-10-22 06:22:38 +00:00 · 2026-10-20 07:35:43 -07:00
parent 16f4ea5822
commit 165d967dc8
3 changed files with 262 additions and 0 deletions
@@ -127,6 +127,10 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 	option.VolumeGrowthDisabled = v.GetBool("master.volume_growth.disable")
 	v.SetDefault("master.volume_growth.freeze_on_topology_mismatch", true)
 	v.SetDefault("master.volume_growth.state_loss_recovery_wait", "0s")
+	v.SetDefault("master.volume_id_allocator", "")
+	v.SetDefault("master.volume_id_etcd.servers", "localhost:2379")
+	v.SetDefault("master.volume_id_etcd.key_prefix", "/seaweedfs")
+	v.SetDefault("master.volume_id_etcd.timeout", "3s")

 	topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1")
 	topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2")
@@ -181,6 +185,17 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 		}
 		return err
 	}
+	switch v.GetString("master.volume_id_allocator") {
+	case "":
+	case "etcd":
+		allocator, err := newVolumeIdAllocator(v, ms.Topo)
+		if err != nil {
+			glog.Fatalf("volume id allocator: %v", err)
+		}
+		ms.Topo.SetVolumeIdAllocator(allocator)
+	default:
+		glog.Fatalf("unknown master.volume_id_allocator %q", v.GetString("master.volume_id_allocator"))
+	}
 	ms.vg = topology.NewDefaultVolumeGrowth()
 	glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")

@@ -0,0 +1,53 @@
+package weed_server
+
+import (
+	"crypto/tls"
+	"fmt"
+	"strings"
+	"time"
+
+	"go.etcd.io/etcd/client/pkg/v3/transport"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func newVolumeIdAllocator(v *util.ViperProxy, topo *topology.Topology) (topology.VolumeIdAllocator, error) {
+	servers := v.GetString("master.volume_id_etcd.servers")
+	timeout, err := time.ParseDuration(v.GetString("master.volume_id_etcd.timeout"))
+	if err != nil {
+		return nil, fmt.Errorf("parse master.volume_id_etcd.timeout: %w", err)
+	}
+
+	var tlsConfig *tls.Config
+	if caFile := v.GetString("master.volume_id_etcd.tls_ca_file"); caFile != "" {
+		tlsInfo := transport.TLSInfo{
+			CertFile:      v.GetString("master.volume_id_etcd.tls_client_crt_file"),
+			KeyFile:       v.GetString("master.volume_id_etcd.tls_client_key_file"),
+			TrustedCAFile: caFile,
+		}
+		if tlsConfig, err = tlsInfo.ClientConfig(); err != nil {
+			return nil, fmt.Errorf("etcd TLS client configuration error: %w", err)
+		}
+	}
+
+	client, err := clientv3.New(clientv3.Config{
+		Endpoints:   strings.Split(servers, ","),
+		Username:    v.GetString("master.volume_id_etcd.username"),
+		Password:    v.GetString("master.volume_id_etcd.password"),
+		DialTimeout: timeout,
+		TLS:         tlsConfig,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("connect volume id etcd: %w", err)
+	}
+
+	glog.Infof("volume id allocator: etcd %s", servers)
+	return topology.NewEtcdVolumeIdAllocator(
+		client.KV,
+		v.GetString("master.volume_id_etcd.key_prefix"),
+		timeout,
+		topo.GetMaxVolumeId), nil
+}
@@ -0,0 +1,194 @@
+package topology
+
+import (
+	"context"
+	"math"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// fakeEtcdKV is an in-memory clientv3.KV honoring CreateRevision/ModRevision
+// compare-and-swap, enough to exercise the allocator's CAS loop.
+type fakeEtcdKV struct {
+	mu   sync.Mutex
+	rev  int64
+	data map[string]*mvccpb.KeyValue
+}
+
+func newFakeEtcdKV() *fakeEtcdKV {
+	return &fakeEtcdKV{data: make(map[string]*mvccpb.KeyValue)}
+}
+
+func (kv *fakeEtcdKV) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
+	kv.mu.Lock()
+	defer kv.mu.Unlock()
+	if v, ok := kv.data[key]; ok {
+		return &clientv3.GetResponse{Kvs: []*mvccpb.KeyValue{v}}, nil
+	}
+	return &clientv3.GetResponse{}, nil
+}
+
+func (kv *fakeEtcdKV) Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
+	kv.mu.Lock()
+	defer kv.mu.Unlock()
+	kv.applyPut(key, val)
+	return &clientv3.PutResponse{}, nil
+}
+
+func (kv *fakeEtcdKV) applyPut(key, val string) {
+	kv.rev++
+	if v, ok := kv.data[key]; ok {
+		v.Value = []byte(val)
+		v.ModRevision = kv.rev
+	} else {
+		kv.data[key] = &mvccpb.KeyValue{Key: []byte(key), Value: []byte(val), CreateRevision: kv.rev, ModRevision: kv.rev}
+	}
+}
+
+func (kv *fakeEtcdKV) GetStream(ctx context.Context, key string, opts ...clientv3.OpOption) (clientv3.GetStreamChan, error) {
+	return nil, nil
+}
+
+func (kv *fakeEtcdKV) Delete(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.DeleteResponse, error) {
+	kv.mu.Lock()
+	defer kv.mu.Unlock()
+	delete(kv.data, key)
+	return &clientv3.DeleteResponse{}, nil
+}
+
+func (kv *fakeEtcdKV) Compact(ctx context.Context, rev int64, opts ...clientv3.CompactOption) (*clientv3.CompactResponse, error) {
+	return &clientv3.CompactResponse{}, nil
+}
+
+func (kv *fakeEtcdKV) Do(ctx context.Context, op clientv3.Op) (clientv3.OpResponse, error) {
+	return clientv3.OpResponse{}, nil
+}
+
+func (kv *fakeEtcdKV) Txn(ctx context.Context) clientv3.Txn {
+	return &fakeEtcdTxn{kv: kv}
+}
+
+type fakeEtcdTxn struct {
+	kv   *fakeEtcdKV
+	cmps []clientv3.Cmp
+	ops  []clientv3.Op
+}
+
+func (t *fakeEtcdTxn) If(cs ...clientv3.Cmp) clientv3.Txn {
+	t.cmps = append(t.cmps, cs...)
+	return t
+}
+
+func (t *fakeEtcdTxn) Then(ops ...clientv3.Op) clientv3.Txn {
+	t.ops = append(t.ops, ops...)
+	return t
+}
+
+func (t *fakeEtcdTxn) Else(ops ...clientv3.Op) clientv3.Txn {
+	return t
+}
+
+func (t *fakeEtcdTxn) Commit() (*clientv3.TxnResponse, error) {
+	t.kv.mu.Lock()
+	defer t.kv.mu.Unlock()
+	for _, cmp := range t.cmps {
+		if !t.kv.evalCmp(cmp) {
+			return &clientv3.TxnResponse{Succeeded: false}, nil
+		}
+	}
+	for _, op := range t.ops {
+		if op.IsPut() {
+			t.kv.applyPut(string(op.KeyBytes()), string(op.ValueBytes()))
+		}
+	}
+	return &clientv3.TxnResponse{Succeeded: true}, nil
+}
+
+func (kv *fakeEtcdKV) evalCmp(cmp clientv3.Cmp) bool {
+	c := cmp.GetCompare()
+	if c.Result != etcdserverpb.Compare_EQUAL {
+		return false
+	}
+	existing := kv.data[string(c.Key)]
+	switch u := c.TargetUnion.(type) {
+	case *etcdserverpb.Compare_CreateRevision:
+		var rev int64
+		if existing != nil {
+			rev = existing.CreateRevision
+		}
+		return rev == u.CreateRevision
+	case *etcdserverpb.Compare_ModRevision:
+		var rev int64
+		if existing != nil {
+			rev = existing.ModRevision
+		}
+		return rev == u.ModRevision
+	}
+	return false
+}
+
+func TestEtcdAllocatorSeedsFromLocalBound(t *testing.T) {
+	alloc := NewEtcdVolumeIdAllocator(newFakeEtcdKV(), "/sw/", time.Second, func() needle.VolumeId {
+		return 41
+	})
+	for i, want := range []needle.VolumeId{42, 43, 44} {
+		got, err := alloc.NextVolumeId()
+		if err != nil {
+			t.Fatalf("alloc %d: %v", i, err)
+		}
+		if got != want {
+			t.Fatalf("alloc %d = %d, want %d", i, got, want)
+		}
+	}
+}
+
+func TestEtcdAllocatorResumesFromStoredKey(t *testing.T) {
+	kv := newFakeEtcdKV()
+	kv.Put(context.Background(), "seaweedfs/max_volume_id", "100")
+	alloc := NewEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
+		return 41
+	})
+	got, err := alloc.NextVolumeId()
+	if err != nil {
+		t.Fatalf("alloc: %v", err)
+	}
+	if got != 101 {
+		t.Fatalf("stored counter wins over the seed: got %d", got)
+	}
+}
+
+func TestEtcdAllocatorRetriesOnCasConflict(t *testing.T) {
+	kv := newFakeEtcdKV()
+	// the seed runs after the empty read and a competing writer lands the key
+	// before the allocator's create-CAS, forcing a retry
+	alloc := NewEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
+		kv.Put(context.Background(), "seaweedfs/max_volume_id", "55")
+		return 5
+	})
+	got, err := alloc.NextVolumeId()
+	if err != nil {
+		t.Fatalf("alloc: %v", err)
+	}
+	if got != 56 {
+		t.Fatalf("CAS retry should allocate 56, got %d", got)
+	}
+}
+
+func TestEtcdAllocatorExhaustion(t *testing.T) {
+	kv := newFakeEtcdKV()
+	kv.Put(context.Background(), "seaweedfs/max_volume_id", strconv.FormatUint(math.MaxUint32, 10))
+	alloc := NewEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
+		return 5
+	})
+	if _, err := alloc.NextVolumeId(); err == nil {
+		t.Fatal("expected exhaustion error at MaxUint32")
+	}
+}