You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

103 lines
2.7 KiB

1 year ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "time"
  6. "github.com/seaweedfs/raft"
  7. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  8. "github.com/seaweedfs/seaweedfs/weed/security"
  9. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/super_block"
  11. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  12. "github.com/seaweedfs/seaweedfs/weed/topology"
  13. )
  14. func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest) (*master_pb.AssignResponse, error) {
  15. if !ms.Topo.IsLeader() {
  16. return nil, raft.NotLeaderError
  17. }
  18. if req.Count == 0 {
  19. req.Count = 1
  20. }
  21. if req.Replication == "" {
  22. req.Replication = ms.option.DefaultReplicaPlacement
  23. }
  24. replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
  25. if err != nil {
  26. return nil, err
  27. }
  28. ttl, err := needle.ReadTTL(req.Ttl)
  29. if err != nil {
  30. return nil, err
  31. }
  32. diskType := types.ToDiskType(req.DiskType)
  33. option := &topology.VolumeGrowOption{
  34. Collection: req.Collection,
  35. ReplicaPlacement: replicaPlacement,
  36. Ttl: ttl,
  37. DiskType: diskType,
  38. Preallocate: ms.preallocateSize,
  39. DataCenter: req.DataCenter,
  40. Rack: req.Rack,
  41. DataNode: req.DataNode,
  42. MemoryMapMaxSizeMb: req.MemoryMapMaxSizeMb,
  43. }
  44. vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
  45. if !vl.HasGrowRequest() && vl.ShouldGrowVolumes(option) {
  46. if ms.Topo.AvailableSpaceFor(option) <= 0 {
  47. return nil, fmt.Errorf("no free volumes left for " + option.String())
  48. }
  49. vl.AddGrowRequest()
  50. ms.vgCh <- &topology.VolumeGrowRequest{
  51. Option: option,
  52. Count: int(req.WritableVolumeCount),
  53. }
  54. }
  55. var (
  56. lastErr error
  57. maxTimeout = time.Second * 10
  58. startTime = time.Now()
  59. )
  60. for time.Now().Sub(startTime) < maxTimeout {
  61. fid, count, dnList, err := ms.Topo.PickForWrite(req.Count, option)
  62. if err == nil {
  63. dn := dnList.Head()
  64. var replicas []*master_pb.Location
  65. for _, r := range dnList.Rest() {
  66. replicas = append(replicas, &master_pb.Location{
  67. Url: r.Url(),
  68. PublicUrl: r.PublicUrl,
  69. GrpcPort: uint32(r.GrpcPort),
  70. DataCenter: r.GetDataCenterId(),
  71. })
  72. }
  73. return &master_pb.AssignResponse{
  74. Fid: fid,
  75. Location: &master_pb.Location{
  76. Url: dn.Url(),
  77. PublicUrl: dn.PublicUrl,
  78. GrpcPort: uint32(dn.GrpcPort),
  79. DataCenter: dn.GetDataCenterId(),
  80. },
  81. Count: count,
  82. Auth: string(security.GenJwtForVolumeServer(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fid)),
  83. Replicas: replicas,
  84. }, nil
  85. }
  86. //glog.V(4).Infoln("waiting for volume growing...")
  87. lastErr = err
  88. time.Sleep(200 * time.Millisecond)
  89. }
  90. return nil, lastErr
  91. }