From 165d967dc82c7e0d65c5259c2d86253e6f785b85 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 18 Oct 2026 20:34:47 -0700 Subject: [PATCH] master: wire the etcd volume id allocator behind master.volume_id_allocator=etcd (#11763) --- weed/server/master_server.go | 15 ++ weed/server/master_volume_id_allocator.go | 53 ++++++ weed/topology/volume_id_allocator_test.go | 194 ++++++++++++++++++++++ 3 files changed, 262 insertions(+) create mode 100644 weed/server/master_volume_id_allocator.go create mode 100644 weed/topology/volume_id_allocator_test.go diff --git a/weed/server/master_server.go b/weed/server/master_server.go index b36679ebcd..10f0c19a5f 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -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") diff --git a/weed/server/master_volume_id_allocator.go b/weed/server/master_volume_id_allocator.go new file mode 100644 index 0000000000..8a4987d677 --- /dev/null +++ b/weed/server/master_volume_id_allocator.go @@ -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 +} diff --git a/weed/topology/volume_id_allocator_test.go b/weed/topology/volume_id_allocator_test.go new file mode 100644 index 0000000000..39cbfbbb79 --- /dev/null +++ b/weed/topology/volume_id_allocator_test.go @@ -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") + } +}