aboutsummaryrefslogtreecommitdiff
path: root/weed/server
diff options
context:
space:
mode:
Diffstat (limited to 'weed/server')
-rw-r--r--weed/server/master_grpc_server_assign.go3
-rw-r--r--weed/server/master_grpc_server_volume.go51
-rw-r--r--weed/server/master_server.go18
-rw-r--r--weed/server/master_server_handlers.go4
-rw-r--r--weed/server/master_server_handlers_admin.go13
5 files changed, 56 insertions, 33 deletions
diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go
index 523db6038..efde9eacb 100644
--- a/weed/server/master_grpc_server_assign.go
+++ b/weed/server/master_grpc_server_assign.go
@@ -74,6 +74,7 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
}
vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
+ vl.SetLastGrowCount(req.WritableVolumeCount)
var (
lastErr error
@@ -91,7 +92,7 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
vl.AddGrowRequest()
ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
Option: option,
- Count: int(req.WritableVolumeCount),
+ Count: req.WritableVolumeCount,
}
}
if err != nil {
diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go
index afe062a2d..f48326459 100644
--- a/weed/server/master_grpc_server_volume.go
+++ b/weed/server/master_grpc_server_volume.go
@@ -3,6 +3,8 @@ package weed_server
import (
"context"
"fmt"
+ "github.com/seaweedfs/seaweedfs/weed/topology"
+ "math/rand"
"reflect"
"strings"
"sync"
@@ -18,8 +20,39 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage/types"
)
+func (ms *MasterServer) DoAutomaticVolumeGrow(req *topology.VolumeGrowRequest) {
+ glog.V(1).Infoln("starting automatic volume grow")
+ start := time.Now()
+ newVidLocations, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
+ glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
+ if err != nil {
+ glog.V(1).Infof("automatic volume grow failed: %+v", err)
+ return
+ }
+ for _, newVidLocation := range newVidLocations {
+ ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: newVidLocation})
+ }
+}
+
func (ms *MasterServer) ProcessGrowRequest() {
go func() {
+ for {
+ time.Sleep(14*time.Minute + time.Duration(120*rand.Float32())*time.Second)
+ if !ms.Topo.IsLeader() {
+ continue
+ }
+ for _, vl := range ms.Topo.ListVolumeLyauts() {
+ if !vl.HasGrowRequest() && vl.ShouldGrowVolumes(&topology.VolumeGrowOption{}) {
+ vl.AddGrowRequest()
+ ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
+ Option: vl.ToGrowOption(),
+ Count: vl.GetLastGrowCount(),
+ }
+ }
+ }
+ }
+ }()
+ go func() {
filter := sync.Map{}
for {
req, ok := <-ms.volumeGrowthRequestChan
@@ -50,23 +83,11 @@ func (ms *MasterServer) ProcessGrowRequest() {
if !found && vl.ShouldGrowVolumes(option) {
filter.Store(req, nil)
// we have lock called inside vg
- go func() {
- glog.V(1).Infoln("starting automatic volume grow")
- start := time.Now()
- newVidLocations, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
- glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
- if err == nil {
- for _, newVidLocation := range newVidLocations {
- ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: newVidLocation})
- }
- } else {
- glog.V(1).Infof("automatic volume grow failed: %+v", err)
- }
+ go func(req *topology.VolumeGrowRequest, vl *topology.VolumeLayout) {
+ ms.DoAutomaticVolumeGrow(req)
vl.DoneGrowRequest()
-
filter.Delete(req)
- }()
-
+ }(req, vl)
} else {
glog.V(4).Infoln("discard volume grow request")
time.Sleep(time.Millisecond * 211)
diff --git a/weed/server/master_server.go b/weed/server/master_server.go
index 3499a2e13..014bdb7f8 100644
--- a/weed/server/master_server.go
+++ b/weed/server/master_server.go
@@ -92,15 +92,15 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
v.SetDefault("master.replication.treat_replication_as_minimums", false)
replicationAsMin := v.GetBool("master.replication.treat_replication_as_minimums")
- v.SetDefault("master.volume_growth.copy_1", 7)
- v.SetDefault("master.volume_growth.copy_2", 6)
- v.SetDefault("master.volume_growth.copy_3", 3)
- v.SetDefault("master.volume_growth.copy_other", 1)
- v.SetDefault("master.volume_growth.threshold", 0.9)
- topology.VolumeGrowStrategy.Copy1Count = v.GetInt("master.volume_growth.copy_1")
- topology.VolumeGrowStrategy.Copy2Count = v.GetInt("master.volume_growth.copy_2")
- topology.VolumeGrowStrategy.Copy3Count = v.GetInt("master.volume_growth.copy_3")
- topology.VolumeGrowStrategy.CopyOtherCount = v.GetInt("master.volume_growth.copy_other")
+ v.SetDefault("master.volume_growth.copy_1", topology.VolumeGrowStrategy.Copy1Count)
+ v.SetDefault("master.volume_growth.copy_2", topology.VolumeGrowStrategy.Copy2Count)
+ v.SetDefault("master.volume_growth.copy_3", topology.VolumeGrowStrategy.Copy3Count)
+ v.SetDefault("master.volume_growth.copy_other", topology.VolumeGrowStrategy.CopyOtherCount)
+ v.SetDefault("master.volume_growth.threshold", topology.VolumeGrowStrategy.Threshold)
+ topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1")
+ topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2")
+ topology.VolumeGrowStrategy.Copy3Count = v.GetUint32("master.volume_growth.copy_3")
+ topology.VolumeGrowStrategy.CopyOtherCount = v.GetUint32("master.volume_growth.copy_other")
topology.VolumeGrowStrategy.Threshold = v.GetFloat64("master.volume_growth.threshold")
var preallocateSize int64
diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go
index 65c4589d8..d2f982226 100644
--- a/weed/server/master_server_handlers.go
+++ b/weed/server/master_server_handlers.go
@@ -107,7 +107,7 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
requestedCount = 1
}
- writableVolumeCount, e := strconv.Atoi(r.FormValue("writableVolumeCount"))
+ writableVolumeCount, e := strconv.ParseUint(r.FormValue("writableVolumeCount"), 10, 32)
if e != nil {
writableVolumeCount = 0
}
@@ -145,7 +145,7 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
vl.AddGrowRequest()
ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
Option: option,
- Count: writableVolumeCount,
+ Count: uint32(writableVolumeCount),
}
}
if err != nil {
diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go
index 07c0a94d2..5e3e42dea 100644
--- a/weed/server/master_server_handlers_admin.go
+++ b/weed/server/master_server_handlers_admin.go
@@ -70,7 +70,7 @@ func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Reque
}
func (ms *MasterServer) volumeGrowHandler(w http.ResponseWriter, r *http.Request) {
- count := 0
+ count := uint64(0)
option, err := ms.getVolumeGrowOption(r)
if err != nil {
writeJsonError(w, r, http.StatusNotAcceptable, err)
@@ -78,15 +78,16 @@ func (ms *MasterServer) volumeGrowHandler(w http.ResponseWriter, r *http.Request
}
glog.V(0).Infof("volumeGrowHandler received %v from %v", option.String(), r.RemoteAddr)
- if count, err = strconv.Atoi(r.FormValue("count")); err == nil {
- if ms.Topo.AvailableSpaceFor(option) < int64(count*option.ReplicaPlacement.GetCopyCount()) {
- err = fmt.Errorf("only %d volumes left, not enough for %d", ms.Topo.AvailableSpaceFor(option), count*option.ReplicaPlacement.GetCopyCount())
+ if count, err = strconv.ParseUint(r.FormValue("count"), 10, 32); err == nil {
+ replicaCount := int64(count * uint64(option.ReplicaPlacement.GetCopyCount()))
+ if ms.Topo.AvailableSpaceFor(option) < replicaCount {
+ err = fmt.Errorf("only %d volumes left, not enough for %d", ms.Topo.AvailableSpaceFor(option), replicaCount)
} else if !ms.Topo.DataCenterExists(option.DataCenter) {
err = fmt.Errorf("data center %v not found in topology", option.DataCenter)
} else {
var newVidLocations []*master_pb.VolumeLocation
- newVidLocations, err = ms.vg.GrowByCountAndType(ms.grpcDialOption, count, option, ms.Topo)
- count = len(newVidLocations)
+ newVidLocations, err = ms.vg.GrowByCountAndType(ms.grpcDialOption, uint32(count), option, ms.Topo)
+ count = uint64(len(newVidLocations))
}
} else {
err = fmt.Errorf("can not parse parameter count %s", r.FormValue("count"))
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 11:57:48 +0000