master_grpc_server_volume.go 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/raft"
  6. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  7. "github.com/chrislusf/seaweedfs/weed/security"
  8. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  9. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  10. "github.com/chrislusf/seaweedfs/weed/topology"
  11. )
  12. func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
  13. if !ms.Topo.IsLeader() {
  14. return nil, raft.NotLeaderError
  15. }
  16. resp := &master_pb.LookupVolumeResponse{}
  17. volumeLocations := ms.lookupVolumeId(req.VolumeIds, req.Collection)
  18. for _, result := range volumeLocations {
  19. var locations []*master_pb.Location
  20. for _, loc := range result.Locations {
  21. locations = append(locations, &master_pb.Location{
  22. Url: loc.Url,
  23. PublicUrl: loc.PublicUrl,
  24. })
  25. }
  26. resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{
  27. VolumeId: result.VolumeId,
  28. Locations: locations,
  29. Error: result.Error,
  30. })
  31. }
  32. return resp, nil
  33. }
  34. func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest) (*master_pb.AssignResponse, error) {
  35. if !ms.Topo.IsLeader() {
  36. return nil, raft.NotLeaderError
  37. }
  38. if req.Count == 0 {
  39. req.Count = 1
  40. }
  41. if req.Replication == "" {
  42. req.Replication = ms.option.DefaultReplicaPlacement
  43. }
  44. replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
  45. if err != nil {
  46. return nil, err
  47. }
  48. ttl, err := needle.ReadTTL(req.Ttl)
  49. if err != nil {
  50. return nil, err
  51. }
  52. option := &topology.VolumeGrowOption{
  53. Collection: req.Collection,
  54. ReplicaPlacement: replicaPlacement,
  55. Ttl: ttl,
  56. Prealloacte: ms.preallocateSize,
  57. DataCenter: req.DataCenter,
  58. Rack: req.Rack,
  59. DataNode: req.DataNode,
  60. MemoryMapMaxSizeMb: req.MemoryMapMaxSizeMb,
  61. }
  62. if !ms.Topo.HasWritableVolume(option) {
  63. if ms.Topo.FreeSpace() <= 0 {
  64. return nil, fmt.Errorf("No free volumes left!")
  65. }
  66. ms.vgLock.Lock()
  67. if !ms.Topo.HasWritableVolume(option) {
  68. if _, err = ms.vg.AutomaticGrowByType(option, ms.grpcDialOption, ms.Topo, int(req.WritableVolumeCount)); err != nil {
  69. ms.vgLock.Unlock()
  70. return nil, fmt.Errorf("Cannot grow volume group! %v", err)
  71. }
  72. }
  73. ms.vgLock.Unlock()
  74. }
  75. fid, count, dn, err := ms.Topo.PickForWrite(req.Count, option)
  76. if err != nil {
  77. return nil, fmt.Errorf("%v", err)
  78. }
  79. return &master_pb.AssignResponse{
  80. Fid: fid,
  81. Url: dn.Url(),
  82. PublicUrl: dn.PublicUrl,
  83. Count: count,
  84. Auth: string(security.GenJwt(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fid)),
  85. }, nil
  86. }
  87. func (ms *MasterServer) Statistics(ctx context.Context, req *master_pb.StatisticsRequest) (*master_pb.StatisticsResponse, error) {
  88. if !ms.Topo.IsLeader() {
  89. return nil, raft.NotLeaderError
  90. }
  91. if req.Replication == "" {
  92. req.Replication = ms.option.DefaultReplicaPlacement
  93. }
  94. replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
  95. if err != nil {
  96. return nil, err
  97. }
  98. ttl, err := needle.ReadTTL(req.Ttl)
  99. if err != nil {
  100. return nil, err
  101. }
  102. volumeLayout := ms.Topo.GetVolumeLayout(req.Collection, replicaPlacement, ttl)
  103. stats := volumeLayout.Stats()
  104. totalSize := ms.Topo.GetMaxVolumeCount() * int64(ms.option.VolumeSizeLimitMB) * 1024 * 1024
  105. resp := &master_pb.StatisticsResponse{
  106. TotalSize: uint64(totalSize),
  107. UsedSize: stats.UsedSize,
  108. FileCount: stats.FileCount,
  109. }
  110. return resp, nil
  111. }
  112. func (ms *MasterServer) VolumeList(ctx context.Context, req *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) {
  113. if !ms.Topo.IsLeader() {
  114. return nil, raft.NotLeaderError
  115. }
  116. resp := &master_pb.VolumeListResponse{
  117. TopologyInfo: ms.Topo.ToTopologyInfo(),
  118. VolumeSizeLimitMb: uint64(ms.option.VolumeSizeLimitMB),
  119. }
  120. return resp, nil
  121. }
  122. func (ms *MasterServer) LookupEcVolume(ctx context.Context, req *master_pb.LookupEcVolumeRequest) (*master_pb.LookupEcVolumeResponse, error) {
  123. if !ms.Topo.IsLeader() {
  124. return nil, raft.NotLeaderError
  125. }
  126. resp := &master_pb.LookupEcVolumeResponse{}
  127. ecLocations, found := ms.Topo.LookupEcShards(needle.VolumeId(req.VolumeId))
  128. if !found {
  129. return resp, fmt.Errorf("ec volume %d not found", req.VolumeId)
  130. }
  131. resp.VolumeId = req.VolumeId
  132. for shardId, shardLocations := range ecLocations.Locations {
  133. var locations []*master_pb.Location
  134. for _, dn := range shardLocations {
  135. locations = append(locations, &master_pb.Location{
  136. Url: string(dn.Id()),
  137. PublicUrl: dn.PublicUrl,
  138. })
  139. }
  140. resp.ShardIdLocations = append(resp.ShardIdLocations, &master_pb.LookupEcVolumeResponse_EcShardIdLocation{
  141. ShardId: uint32(shardId),
  142. Locations: locations,
  143. })
  144. }
  145. return resp, nil
  146. }
  147. func (ms *MasterServer) GetMasterConfiguration(ctx context.Context, req *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
  148. resp := &master_pb.GetMasterConfigurationResponse{
  149. MetricsAddress: ms.option.MetricsAddress,
  150. MetricsIntervalSeconds: uint32(ms.option.MetricsIntervalSec),
  151. }
  152. return resp, nil
  153. }