mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-24 02:59:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1351 from ekozlov-search/master

Browse Source 
Set volumes ReadOnly if low free disk space
This commit is contained in:
Chris Lu 2020-06-05 09:56:46 -07:00 committed by GitHub
commit 03c2f03509
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 73 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
weed/command/server.go
View File

@ -55,6 +55,8 @@ var (
serverDisableHttp = cmdServer.Flag.Bool("disableHttp", false, "disable http requests, only gRPC operations are allowed.")
volumeDataFolders = cmdServer.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
volumeMaxDataVolumeCounts = cmdServer.Flag.String("volume.max", "7", "maximum numbers of volumes, count[,count]... If set to zero on non-windows OS, the limit will be auto configured.")
volumeMinFreeSpacePercent = cmdServer.Flag.String("volume.minFreeSpacePercent", "0", "minimum free disk space(in percents). If free disk space lower this value - all volumes marks as ReadOnly")
// pulseSeconds = cmdServer.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
isStartingFiler = cmdServer.Flag.Bool("filer", false, "whether to start filer")
isStartingS3 = cmdServer.Flag.Bool("s3", false, "whether to start S3 gateway")
@ -206,7 +208,8 @@ func runServer(cmd *Command, args []string) bool {
// start volume server
{
go serverOptions.v.startVolumeServer(*volumeDataFolders, *volumeMaxDataVolumeCounts, *serverWhiteListOption)
go serverOptions.v.startVolumeServer(*volumeDataFolders, *volumeMaxDataVolumeCounts, *serverWhiteListOption, *volumeMinFreeSpacePercent )
}
startMaster(masterOptions, serverWhiteList)

18
weed/command/volume.go
View File

@ -52,6 +52,7 @@ type VolumeServerOptions struct {
memProfile *string
compactionMBPerSecond *int
fileSizeLimitMB *int
minFreeSpacePercent []float32
}
func init() {
@ -87,6 +88,7 @@ var (
volumeFolders = cmdVolume.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
maxVolumeCounts = cmdVolume.Flag.String("max", "7", "maximum numbers of volumes, count[,count]... If set to zero on non-windows OS, the limit will be auto configured.")
volumeWhiteListOption = cmdVolume.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
minFreeSpacePercent = cmdVolume.Flag.String("minFreeSpacePercent ", "0", "minimum free disk space(in percents). If free disk space lower this value - all volumes marks as ReadOnly")
)
func runVolume(cmd *Command, args []string) bool {
@ -96,12 +98,12 @@ func runVolume(cmd *Command, args []string) bool {
runtime.GOMAXPROCS(runtime.NumCPU())
grace.SetupProfiling(*v.cpuProfile, *v.memProfile)
v.startVolumeServer(*volumeFolders, *maxVolumeCounts, *volumeWhiteListOption)
v.startVolumeServer(*volumeFolders, *maxVolumeCounts, *volumeWhiteListOption, *minFreeSpacePercent)
return true
}
func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, volumeWhiteListOption string) {
func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, volumeWhiteListOption, minFreeSpacePercent string) {
// Set multiple folders and each folder's max volume count limit'
v.folders = strings.Split(volumeFolders, ",")
@ -116,6 +118,16 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
if len(v.folders) != len(v.folderMaxLimits) {
glog.Fatalf("%d directories by -dir, but only %d max is set by -max", len(v.folders), len(v.folderMaxLimits))
}
minFreeSpacePercentStrings := strings.Split(minFreeSpacePercent, ",")
for _, freeString := range minFreeSpacePercentStrings {
if value, e := strconv.ParseFloat(freeString, 32); e == nil {
v.minFreeSpacePercent = append(v.minFreeSpacePercent, float32(value))
} else {
glog.Fatalf("The value specified in -minFreeSpacePercent not a valid value %s", freeString)
}
}
for _, folder := range v.folders {
if err := util.TestFolderWritable(folder); err != nil {
glog.Fatalf("Check Data Folder(-dir) Writable %s : %s", folder, err)
@ -159,7 +171,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
*v.ip, *v.port, *v.publicUrl,
v.folders, v.folderMaxLimits,
v.folders, v.folderMaxLimits, v.minFreeSpacePercent,
volumeNeedleMapKind,
strings.Split(masters, ","), 5, *v.dataCenter, *v.rack,
v.whiteList,

5
weed/server/volume_server.go
View File

@ -35,7 +35,7 @@ type VolumeServer struct {
func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
port int, publicUrl string,
folders []string, maxCounts []int,
folders []string, maxCounts []int, minFreeSpacePercent []float32,
needleMapKind storage.NeedleMapType,
masterNodes []string, pulseSeconds int,
dataCenter string, rack string,
@ -68,8 +68,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
fileSizeLimitBytes: int64(fileSizeLimitMB) * 1024 * 1024,
}
vs.SeedMasterNodes = masterNodes
vs.store = storage.NewStore(vs.grpcDialOption, port, ip, publicUrl, folders, maxCounts, vs.needleMapKind)
vs.store = storage.NewStore(vs.grpcDialOption, port, ip, publicUrl, folders, maxCounts, minFreeSpacePercent, vs.needleMapKind)
vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
handleStaticResources(adminMux)

37
weed/storage/disk_location.go
View File

@ -1,11 +1,14 @@
package storage
import (
"fmt"
"io/ioutil"
"os"
"strings"
"sync"
"fmt"
"github.com/chrislusf/seaweedfs/weed/stats"
"io/ioutil"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
@ -15,6 +18,7 @@ import (
type DiskLocation struct {
Directory string
MaxVolumeCount int
MinFreeSpacePercent float32
volumes map[needle.VolumeId]*Volume
volumesLock sync.RWMutex
@ -23,10 +27,11 @@ type DiskLocation struct {
ecVolumesLock sync.RWMutex
}
func NewDiskLocation(dir string, maxVolumeCount int) *DiskLocation {
location := &DiskLocation{Directory: dir, MaxVolumeCount: maxVolumeCount}
func NewDiskLocation(dir string, maxVolumeCount int, minFreeSpacePercent float32) *DiskLocation {
location := &DiskLocation{Directory: dir, MaxVolumeCount: maxVolumeCount, MinFreeSpacePercent: minFreeSpacePercent}
location.volumes = make(map[needle.VolumeId]*Volume)
location.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume)
go location.CheckDiskSpace()
return location
}
@ -293,3 +298,21 @@ func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64)
return
}
func (l *DiskLocation) CheckDiskSpace() {
lastStat := false
t := time.NewTicker(time.Minute)
for _ = range t.C {
if dir, e := filepath.Abs(l.Directory); e == nil {
s := stats.NewDiskStatus(dir)
if (s.PercentFree < l.MinFreeSpacePercent) != lastStat {
lastStat = !lastStat
for _, v := range l.volumes {
v.SetLowDiskSpace(lastStat)
}
}
}
}
}

4
weed/storage/store.go
View File

@ -48,11 +48,11 @@ func (s *Store) String() (str string) {
return
}
func NewStore(grpcDialOption grpc.DialOption, port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, needleMapKind NeedleMapType) (s *Store) {
func NewStore(grpcDialOption grpc.DialOption, port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, freeDiskSpaceWatermark []float32, needleMapKind NeedleMapType) (s *Store) {
s = &Store{grpcDialOption: grpcDialOption, Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind}
s.Locations = make([]*DiskLocation, 0)
for i := 0; i < len(dirnames); i++ {
location := NewDiskLocation(dirnames[i], maxVolumeCounts[i])
location := NewDiskLocation(dirnames[i], maxVolumeCounts[i], freeDiskSpaceWatermark[i])
location.loadExistingVolumes(needleMapKind)
s.Locations = append(s.Locations, location)
stats.VolumeServerMaxVolumeCounter.Add(float64(maxVolumeCounts[i]))

8
weed/storage/volume.go
View File

@ -27,6 +27,7 @@ type Volume struct {
needleMapKind NeedleMapType
noWriteOrDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
noWriteCanDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
lowDiskSpace bool
hasRemoteFile bool // if the volume has a remote file
MemoryMapMaxSizeMb uint32
@ -45,6 +46,11 @@ type Volume struct {
volumeInfo *volume_server_pb.VolumeInfo
}
func (v *Volume) SetLowDiskSpace(lowDiskSpace bool) {
glog.V(0).Infof("SetLowDiskSpace id %d value %t", v.Id, lowDiskSpace)
v.lowDiskSpace = lowDiskSpace
}
func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) {
// if replicaPlacement is nil, the superblock will be loaded from disk
v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb,
@ -244,5 +250,5 @@ func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
}
func (v *Volume) IsReadOnly() bool {
return v.noWriteOrDelete || v.noWriteCanDelete
return v.noWriteOrDelete || v.noWriteCanDelete || v.lowDiskSpace
}

10
weed/topology/data_node.go
View File

@ -41,7 +41,7 @@ func (dn *DataNode) String() string {
return fmt.Sprintf("Node:%s, volumes:%v, Ip:%s, Port:%d, PublicUrl:%s", dn.NodeImpl.String(), dn.volumes, dn.Ip, dn.Port, dn.PublicUrl)
}
func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew bool) {
func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO bool) {
dn.Lock()
defer dn.Unlock()
if oldV, ok := dn.volumes[v.Id]; !ok {
@ -64,12 +64,13 @@ func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew bool) {
dn.UpAdjustRemoteVolumeCountDelta(-1)
}
}
isChangedRO = dn.volumes[v.Id].ReadOnly != v.ReadOnly
dn.volumes[v.Id] = v
}
return
}
func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolumes, deletedVolumes []storage.VolumeInfo) {
func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolumes, deletedVolumes, changeRO []storage.VolumeInfo) {
actualVolumeMap := make(map[needle.VolumeId]storage.VolumeInfo)
for _, v := range actualVolumes {
actualVolumeMap[v.Id] = v
@ -91,10 +92,13 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume
}
dn.Unlock()
for _, v := range actualVolumes {
isNew := dn.AddOrUpdateVolume(v)
isNew, isChangedRO := dn.AddOrUpdateVolume(v)
if isNew {
newVolumes = append(newVolumes, v)
}
if isChangedRO {
changeRO = append(changeRO, v)
}
}
return
}

7
weed/topology/topology.go
View File

@ -212,13 +212,18 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
}
}
// find out the delta volumes
newVolumes, deletedVolumes = dn.UpdateVolumes(volumeInfos)
var changedVolumes []storage.VolumeInfo
newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
for _, v := range newVolumes {
t.RegisterVolumeLayout(v, dn)
}
for _, v := range deletedVolumes {
t.UnRegisterVolumeLayout(v, dn)
}
for _, v := range changedVolumes {
vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl)
vl.ensureCorrectWritables(&v)
}
return
}