diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 143539648c..fbb992e8a0 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -476,6 +476,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 } @@ -486,25 +491,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 @@ -1280,19 +1289,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 { - t.HashicorpRaft.Apply(b, time.Second) - } - } else { - t.SetGrowthFrozen(frozen) + return } + t.SetGrowthFrozen(frozen) } // SetTopologyInitPending gates allocation while a topology id is being diff --git a/weed/topology/volume_id_allocator.go b/weed/topology/volume_id_allocator.go index 695f236785..1f87452f53 100644 --- a/weed/topology/volume_id_allocator.go +++ b/weed/topology/volume_id_allocator.go @@ -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) diff --git a/weed/topology/volume_id_allocator_test.go b/weed/topology/volume_id_allocator_test.go index 39cbfbbb79..a3189a46b7 100644 --- a/weed/topology/volume_id_allocator_test.go +++ b/weed/topology/volume_id_allocator_test.go @@ -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 {