From 6c8810e4d2d3ba9da5155a60a47e20375346959f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 17 Dec 2012 16:48:54 -0800 Subject: [PATCH] ensure only compatible volume versions are writable --- weed-fs/src/pkg/storage/volume_info.go | 1 + weed-fs/src/pkg/topology/topology_event_handling.go | 4 ++-- weed-fs/src/pkg/topology/volume_layout.go | 6 +++++- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/weed-fs/src/pkg/storage/volume_info.go b/weed-fs/src/pkg/storage/volume_info.go index 060277e0a..e4c5f6ec4 100644 --- a/weed-fs/src/pkg/storage/volume_info.go +++ b/weed-fs/src/pkg/storage/volume_info.go @@ -6,6 +6,7 @@ type VolumeInfo struct { Id VolumeId Size uint64 RepType ReplicationType + Version Version FileCount int DeleteCount int DeletedByteCount uint64 diff --git a/weed-fs/src/pkg/topology/topology_event_handling.go b/weed-fs/src/pkg/topology/topology_event_handling.go index 08c08fcde..b33b4d768 100644 --- a/weed-fs/src/pkg/topology/topology_event_handling.go +++ b/weed-fs/src/pkg/topology/topology_event_handling.go @@ -59,8 +59,8 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) { } func (t *Topology) RegisterRecoveredDataNode(dn *DataNode) { for _, v := range dn.volumes { - if uint64(v.Size) < t.volumeSizeLimit { - vl := t.GetVolumeLayout(v.RepType) + vl := t.GetVolumeLayout(v.RepType) + if vl.isWritable(&v) { vl.SetVolumeAvailable(dn, v.Id) } } diff --git a/weed-fs/src/pkg/topology/volume_layout.go b/weed-fs/src/pkg/topology/volume_layout.go index 535effde6..80eff5061 100644 --- a/weed-fs/src/pkg/topology/volume_layout.go +++ b/weed-fs/src/pkg/topology/volume_layout.go @@ -31,13 +31,17 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) { } if vl.vid2location[v.Id].Add(dn) { if len(vl.vid2location[v.Id].list) == v.RepType.GetCopyCount() { - if uint64(v.Size) < vl.volumeSizeLimit { + if vl.isWritable(v) { vl.writables = append(vl.writables, v.Id) } } } } +func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool{ + return uint64(v.Size) < vl.volumeSizeLimit && v.Version == storage.CurrentVersion +} + func (vl *VolumeLayout) Lookup(vid storage.VolumeId) *[]*DataNode { return &vl.vid2location[vid].list }