mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-22 06:22:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

allocator: never go below local bound, replicate external allocations via raft

Browse Source 
A stale etcd counter (restored backup or allocator re-enabled after being
off) could hand out ids this master already knows are allocated; clamp the
CAS base to max(stored, local seed) sampled once per call. Also apply the
allocated id through raft so followers boundaries and mirrors track
allocations made against the external allocator, and dedupe the raft
apply path into raftApply.
This commit is contained in:
Chris Lu
2026-10-20 07:35:43 -07:00
parent 165d967dc8
commit c2b7e50b2b
3 changed files with 74 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/topology/topology.go
+28 -27
View File
@@ -480,6 +480,11 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
if err := t.PersistMaxVolumeId(vid); err != nil {
return 0, fmt.Errorf("can not assign next volume id %d: %w", vid, err)
}
// Still record through raft so followers' boundaries and mirrors track
// allocations made against the external allocator.
if err := t.raftApply(NewMaxVolumeIdCommand(vid, t.GetTopologyId())); err != nil {
return 0, err
}
t.UpAdjustMaxVolumeId(vid)
return vid, nil
}
@@ -490,25 +495,29 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
return 0, fmt.Errorf("persist max volume id %d: %w", next, err)
}
t.RaftServerAccessLock.RLock()
defer t.RaftServerAccessLock.RUnlock()
if t.RaftServer != nil {
if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next, t.GetTopologyId())); err != nil {
return 0, err
}
} else if t.HashicorpRaft != nil {
b, err := json.Marshal(NewMaxVolumeIdCommand(next, t.GetTopologyId()))
if err != nil {
return 0, fmt.Errorf("failed marshal NewMaxVolumeIdCommand: %+v", err)
}
if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
return 0, future.Error()
}
if err := t.raftApply(NewMaxVolumeIdCommand(next, t.GetTopologyId())); err != nil {
return 0, err
}
return next, nil
}
func (t *Topology) raftApply(cmd *MaxVolumeIdCommand) error {
t.RaftServerAccessLock.RLock()
defer t.RaftServerAccessLock.RUnlock()
if t.RaftServer != nil {
_, err := t.RaftServer.Do(cmd)
return err
}
if t.HashicorpRaft != nil {
b, err := json.Marshal(cmd)
if err != nil {
return fmt.Errorf("failed marshal %s: %+v", cmd.CommandName(), err)
}
return t.HashicorpRaft.Apply(b, time.Second).Error()
}
return nil
}
// DefaultNeedleSizeEstimate is the fallback per-file-ID size estimate when
// the client does not provide an expected data size.
const DefaultNeedleSizeEstimate uint64 = 1024 * 1024 // 1 MB
@@ -1297,21 +1306,13 @@ func (t *Topology) SetGrowthFrozen(frozen bool) {
func (t *Topology) applyGrowthFreeze(frozen bool) {
cmd := NewMaxVolumeIdCommand(t.GetMaxVolumeId(), t.GetTopologyId())
cmd.GrowthFrozen = &frozen
t.RaftServerAccessLock.RLock()
defer t.RaftServerAccessLock.RUnlock()
if t.RaftServer != nil {
if _, err := t.RaftServer.Do(cmd); err != nil {
if t.RaftServer != nil || t.HashicorpRaft != nil {
if err := t.raftApply(cmd); err != nil {
glog.Warningf("apply growth freeze %v: %v", frozen, err)
}
} else if t.HashicorpRaft != nil {
if b, err := json.Marshal(cmd); err != nil {
glog.Warningf("marshal growth freeze %v: %v", frozen, err)
} else if future := t.HashicorpRaft.Apply(b, time.Second); future.Error() != nil {
glog.Warningf("apply growth freeze %v: %v", frozen, future.Error())
}
} else {
t.SetGrowthFrozen(frozen)
return
}
t.SetGrowthFrozen(frozen)
}
// SetTopologyInitPending gates allocation while a topology id is being
weed/topology/volume_id_allocator.go
+25 -3
View File
@@ -25,12 +25,19 @@ type VolumeIdAllocator interface {
// residual window: an id cannot be issued without the etcd write committing.
type EtcdVolumeIdAllocator struct {
kv clientv3.KV
client *clientv3.Client
key string
timeout time.Duration
seed func() needle.VolumeId
}
func NewEtcdVolumeIdAllocator(kv clientv3.KV, keyPrefix string, timeout time.Duration, seed func() needle.VolumeId) *EtcdVolumeIdAllocator {
func NewEtcdVolumeIdAllocator(client *clientv3.Client, keyPrefix string, timeout time.Duration, seed func() needle.VolumeId) *EtcdVolumeIdAllocator {
a := newEtcdVolumeIdAllocator(client.KV, keyPrefix, timeout, seed)
a.client = client
return a
}
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",
@@ -39,14 +46,28 @@ func NewEtcdVolumeIdAllocator(kv clientv3.KV, keyPrefix string, timeout time.Dur
}
}
func (a *EtcdVolumeIdAllocator) Close() error {
if a.client == nil {
return nil
}
return a.client.Close()
}
func (a *EtcdVolumeIdAllocator) NextVolumeId() (needle.VolumeId, error) {
ctx, cancel := context.WithTimeout(context.Background(), a.timeout)
defer cancel()
var seedBound uint64
seeded := false
for {
resp, err := a.kv.Get(ctx, a.key)
if err != nil {
return 0, fmt.Errorf("volume id allocator read: %w", err)
}
if !seeded {
// A stale etcd counter (restored backup, re-enabled allocator) must
// never go below the bound this master already knows.
seedBound, seeded = uint64(a.seed()), true
}
var cur uint64
cmp := clientv3.Compare(clientv3.CreateRevision(a.key), "=", 0)
if len(resp.Kvs) > 0 {
@@ -54,8 +75,9 @@ func (a *EtcdVolumeIdAllocator) NextVolumeId() (needle.VolumeId, error) {
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 seedBound > cur {
cur = seedBound
}
if cur >= math.MaxUint32 {
return 0, fmt.Errorf("volume id counter exhausted at %d", cur)
weed/topology/volume_id_allocator_test.go
+21 -5
View File
@@ -31,7 +31,8 @@ func (kv *fakeEtcdKV) Get(ctx context.Context, key string, opts ...clientv3.OpOp
kv.mu.Lock()
defer kv.mu.Unlock()
if v, ok := kv.data[key]; ok {
return &clientv3.GetResponse{Kvs: []*mvccpb.KeyValue{v}}, nil
cp := *v
return &clientv3.GetResponse{Kvs: []*mvccpb.KeyValue{&cp}}, nil
}
return &clientv3.GetResponse{}, nil
}
@@ -136,7 +137,7 @@ func (kv *fakeEtcdKV) evalCmp(cmp clientv3.Cmp) bool {
}
func TestEtcdAllocatorSeedsFromLocalBound(t *testing.T) {
alloc := NewEtcdVolumeIdAllocator(newFakeEtcdKV(), "/sw/", time.Second, func() needle.VolumeId {
alloc := newEtcdVolumeIdAllocator(newFakeEtcdKV(), "/sw/", time.Second, func() needle.VolumeId {
return 41
})
for i, want := range []needle.VolumeId{42, 43, 44} {
@@ -150,10 +151,25 @@ func TestEtcdAllocatorSeedsFromLocalBound(t *testing.T) {
}
}
func TestEtcdAllocatorNeverBelowLocalBound(t *testing.T) {
kv := newFakeEtcdKV()
kv.Put(context.Background(), "seaweedfs/max_volume_id", "10")
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 != 42 {
t.Fatalf("a stale stored counter must not reissue ids: got %d", got)
}
}
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 {
alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
return 41
})
got, err := alloc.NextVolumeId()
@@ -169,7 +185,7 @@ 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 {
alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
kv.Put(context.Background(), "seaweedfs/max_volume_id", "55")
return 5
})
@@ -185,7 +201,7 @@ func TestEtcdAllocatorRetriesOnCasConflict(t *testing.T) {
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 {
alloc := newEtcdVolumeIdAllocator(kv, "seaweedfs", time.Second, func() needle.VolumeId {
return 5
})
if _, err := alloc.NextVolumeId(); err == nil {