aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-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
-rw-r--r--weed/topology/topology_info.go9
-rw-r--r--weed/topology/volume_growth.go18
-rw-r--r--weed/topology/volume_layout.go20
8 files changed, 93 insertions, 43 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"))
diff --git a/weed/topology/topology_info.go b/weed/topology/topology_info.go
index 120ae0d42..89f9097f6 100644
--- a/weed/topology/topology_info.go
+++ b/weed/topology/topology_info.go
@@ -42,6 +42,15 @@ func (t *Topology) ToInfo() (info TopologyInfo) {
return
}
+func (t *Topology) ListVolumeLyauts() (volumeLayouts []*VolumeLayout) {
+ for _, col := range t.collectionMap.Items() {
+ for _, volumeLayout := range col.(*Collection).storageType2VolumeLayout.Items() {
+ volumeLayouts = append(volumeLayouts, volumeLayout.(*VolumeLayout))
+ }
+ }
+ return volumeLayouts
+}
+
func (t *Topology) ToVolumeMap() interface{} {
m := make(map[string]interface{})
m["Max"] = t.diskUsages.GetMaxVolumeCount()
diff --git a/weed/topology/volume_growth.go b/weed/topology/volume_growth.go
index cfc31c8b1..ff516599d 100644
--- a/weed/topology/volume_growth.go
+++ b/weed/topology/volume_growth.go
@@ -27,14 +27,14 @@ This package is created to resolve these replica placement issues:
type VolumeGrowRequest struct {
Option *VolumeGrowOption
- Count int
+ Count uint32
}
type volumeGrowthStrategy struct {
- Copy1Count int
- Copy2Count int
- Copy3Count int
- CopyOtherCount int
+ Copy1Count uint32
+ Copy2Count uint32
+ Copy3Count uint32
+ CopyOtherCount uint32
Threshold float64
}
@@ -75,7 +75,7 @@ func NewDefaultVolumeGrowth() *VolumeGrowth {
// one replication type may need rp.GetCopyCount() actual volumes
// given copyCount, how many logical volumes to create
-func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count int) {
+func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count uint32) {
switch copyCount {
case 1:
count = VolumeGrowStrategy.Copy1Count
@@ -89,7 +89,7 @@ func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count int) {
return
}
-func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOption grpc.DialOption, topo *Topology, targetCount int) (result []*master_pb.VolumeLocation, err error) {
+func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOption grpc.DialOption, topo *Topology, targetCount uint32) (result []*master_pb.VolumeLocation, err error) {
if targetCount == 0 {
targetCount = vg.findVolumeCount(option.ReplicaPlacement.GetCopyCount())
}
@@ -99,11 +99,11 @@ func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOp
}
return result, err
}
-func (vg *VolumeGrowth) GrowByCountAndType(grpcDialOption grpc.DialOption, targetCount int, option *VolumeGrowOption, topo *Topology) (result []*master_pb.VolumeLocation, err error) {
+func (vg *VolumeGrowth) GrowByCountAndType(grpcDialOption grpc.DialOption, targetCount uint32, option *VolumeGrowOption, topo *Topology) (result []*master_pb.VolumeLocation, err error) {
vg.accessLock.Lock()
defer vg.accessLock.Unlock()
- for i := 0; i < targetCount; i++ {
+ for i := uint32(0); i < targetCount; i++ {
if res, e := vg.findAndGrow(grpcDialOption, topo, option); e == nil {
result = append(result, res...)
} else {
diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go
index 5711a6a9b..c33d0ea0c 100644
--- a/weed/topology/volume_layout.go
+++ b/weed/topology/volume_layout.go
@@ -107,7 +107,8 @@ func (v *volumesBinaryState) copyState(list *VolumeLocationList) copyState {
// mapping from volume to its locations, inverted from server to volume
type VolumeLayout struct {
growRequest atomic.Bool
- rp *super_block.ReplicaPlacement
+ lastGrowCount atomic.Uint32
+ rp *super_block.ReplicaPlacement
ttl *needle.TTL
diskType types.DiskType
vid2location map[needle.VolumeId]*VolumeLocationList
@@ -354,6 +355,16 @@ func (vl *VolumeLayout) DoneGrowRequest() {
vl.growRequest.Store(false)
}
+func (vl *VolumeLayout) SetLastGrowCount(count uint32) {
+ if vl.lastGrowCount.Load() != count {
+ vl.lastGrowCount.Store(count)
+ }
+}
+
+func (vl *VolumeLayout) GetLastGrowCount() uint32 {
+ return vl.lastGrowCount.Load()
+}
+
func (vl *VolumeLayout) ShouldGrowVolumes(option *VolumeGrowOption) bool {
total, active, crowded := vl.GetActiveVolumeCount(option)
stats.MasterVolumeLayout.WithLabelValues(option.Collection, option.DataCenter, "total").Set(float64(total))
@@ -539,6 +550,13 @@ func (vl *VolumeLayout) ToInfo() (info VolumeLayoutInfo) {
return
}
+func (vl *VolumeLayout) ToGrowOption() (option *VolumeGrowOption) {
+ option.ReplicaPlacement = vl.rp
+ option.Ttl = vl.ttl
+ option.DiskType = vl.diskType
+ return
+}
+
func (vl *VolumeLayout) Stats() *VolumeLayoutStats {
vl.accessLock.RLock()
defer vl.accessLock.RUnlock()
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-20 22:42:56 +0000