volume: automatically detect max volume count

This commit is contained in:
Chris Lu 2020-03-22 16:21:42 -07:00
parent 0bf148f49d
commit 3137777d83
6 changed files with 66 additions and 1 deletions

View file

@ -81,6 +81,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
} }
} }
if dn.GetMaxVolumeCount() != int64(heartbeat.MaxVolumeCount) {
delta := int64(heartbeat.MaxVolumeCount) - dn.GetMaxVolumeCount()
dn.UpAdjustMaxVolumeCountDelta(delta)
}
glog.V(4).Infof("master received heartbeat %s", heartbeat.String()) glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
message := &master_pb.VolumeLocation{ message := &master_pb.VolumeLocation{
Url: dn.Url(), Url: dn.Url(),

View file

@ -80,8 +80,13 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
doneChan <- err doneChan <- err
return return
} }
if in.GetVolumeSizeLimit() != 0 { if in.GetVolumeSizeLimit() != 0 && vs.store.GetVolumeSizeLimit() != in.GetVolumeSizeLimit() {
vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit()) vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit())
if vs.store.MaybeAdjustVolumeMax() {
if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
}
}
} }
if in.GetLeader() != "" && masterNode != in.GetLeader() && !isSameIP(in.GetLeader(), masterNode) { if in.GetLeader() != "" && masterNode != in.GetLeader() && !isSameIP(in.GetLeader(), masterNode) {
glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), masterNode) glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), masterNode)

View file

@ -275,3 +275,19 @@ func (l *DiskLocation) LocateVolume(vid needle.VolumeId) (os.FileInfo, bool) {
return nil, false return nil, false
} }
func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64) {
l.volumesLock.RLock()
defer l.volumesLock.RUnlock()
for _, vol := range l.volumes {
if vol.IsReadOnly() {
continue
}
datSize, idxSize, _ := vol.FileStat()
unUsedSpace += volumeSizeLimit - (datSize + idxSize)
}
return
}

View file

@ -183,3 +183,10 @@ func (l *DiskLocation) unmountEcVolumeByCollection(collectionName string) map[ne
} }
return deltaVols return deltaVols
} }
func (l *DiskLocation) EcVolumesLen() int {
l.ecVolumesLock.RLock()
defer l.ecVolumesLock.RUnlock()
return len(l.ecVolumes)
}

View file

@ -152,6 +152,13 @@ func (ev *EcVolume) ShardSize() int64 {
return 0 return 0
} }
func (ev *EcVolume) Size() (size int64) {
for _, shard := range ev.Shards {
size += shard.Size()
}
return
}
func (ev *EcVolume) CreatedAt() time.Time { func (ev *EcVolume) CreatedAt() time.Time {
return ev.ecxCreatedAt return ev.ecxCreatedAt
} }

View file

@ -12,6 +12,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb" "github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/stats" "github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
"github.com/chrislusf/seaweedfs/weed/storage/needle" "github.com/chrislusf/seaweedfs/weed/storage/needle"
"github.com/chrislusf/seaweedfs/weed/storage/super_block" "github.com/chrislusf/seaweedfs/weed/storage/super_block"
. "github.com/chrislusf/seaweedfs/weed/storage/types" . "github.com/chrislusf/seaweedfs/weed/storage/types"
@ -99,6 +100,9 @@ func (s *Store) FindFreeLocation() (ret *DiskLocation) {
max := 0 max := 0
for _, location := range s.Locations { for _, location := range s.Locations {
currentFreeCount := location.MaxVolumeCount - location.VolumesLen() currentFreeCount := location.MaxVolumeCount - location.VolumesLen()
currentFreeCount *= erasure_coding.DataShardsCount
currentFreeCount -= location.EcVolumesLen()
currentFreeCount /= erasure_coding.DataShardsCount
if currentFreeCount > max { if currentFreeCount > max {
max = currentFreeCount max = currentFreeCount
ret = location ret = location
@ -382,3 +386,24 @@ func (s *Store) SetVolumeSizeLimit(x uint64) {
func (s *Store) GetVolumeSizeLimit() uint64 { func (s *Store) GetVolumeSizeLimit() uint64 {
return atomic.LoadUint64(&s.volumeSizeLimit) return atomic.LoadUint64(&s.volumeSizeLimit)
} }
func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) {
volumeSizeLimit := s.GetVolumeSizeLimit()
for _, diskLocation := range s.Locations {
if diskLocation.MaxVolumeCount == 0 {
diskStatus := stats.NewDiskStatus(diskLocation.Directory)
unusedSpace := diskLocation.UnUsedSpace(volumeSizeLimit)
unclaimedSpaces := int64(diskStatus.Free) - int64(unusedSpace)
volCount := diskLocation.VolumesLen()
maxVolumeCount := volCount
if unclaimedSpaces > int64(volumeSizeLimit) {
maxVolumeCount += int(uint64(unclaimedSpaces)/volumeSizeLimit) - 1
}
diskLocation.MaxVolumeCount = maxVolumeCount
glog.V(0).Infof("disk %s max %d unclaimedSpace:%dMB, unused:%dMB volumeSizeLimit:%d/MB",
diskLocation.Directory, maxVolumeCount, unclaimedSpaces/1024/1024, unusedSpace/1024/1024, volumeSizeLimit/1024/1024)
hasChanges = true
}
}
return
}