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.

297 lines
8.2 KiB

6 years ago
6 years ago
6 years ago
5 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
5 years ago
5 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "net"
  6. "strings"
  7. "time"
  8. "github.com/chrislusf/raft"
  9. "google.golang.org/grpc/peer"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  12. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  13. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  14. "github.com/chrislusf/seaweedfs/weed/topology"
  15. )
  16. func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
  17. var dn *topology.DataNode
  18. t := ms.Topo
  19. defer func() {
  20. if dn != nil {
  21. glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
  22. t.UnRegisterDataNode(dn)
  23. message := &master_pb.VolumeLocation{
  24. Url: dn.Url(),
  25. PublicUrl: dn.PublicUrl,
  26. }
  27. for _, v := range dn.GetVolumes() {
  28. message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
  29. }
  30. for _, s := range dn.GetEcShards() {
  31. message.DeletedVids = append(message.DeletedVids, uint32(s.VolumeId))
  32. }
  33. if len(message.DeletedVids) > 0 {
  34. ms.clientChansLock.RLock()
  35. for _, ch := range ms.clientChans {
  36. ch <- message
  37. }
  38. ms.clientChansLock.RUnlock()
  39. }
  40. }
  41. }()
  42. for {
  43. heartbeat, err := stream.Recv()
  44. if err != nil {
  45. if dn != nil {
  46. glog.Warningf("SendHeartbeat.Recv server %s:%d : %v", dn.Ip, dn.Port, err)
  47. } else {
  48. glog.Warningf("SendHeartbeat.Recv: %v", err)
  49. }
  50. return err
  51. }
  52. t.Sequence.SetMax(heartbeat.MaxFileKey)
  53. if dn == nil {
  54. dcName, rackName := t.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
  55. dc := t.GetOrCreateDataCenter(dcName)
  56. rack := dc.GetOrCreateRack(rackName)
  57. dn = rack.GetOrCreateDataNode(heartbeat.Ip,
  58. int(heartbeat.Port), heartbeat.PublicUrl,
  59. int64(heartbeat.MaxVolumeCount))
  60. glog.V(0).Infof("added volume server %v:%d", heartbeat.GetIp(), heartbeat.GetPort())
  61. if err := stream.Send(&master_pb.HeartbeatResponse{
  62. VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
  63. MetricsAddress: ms.option.MetricsAddress,
  64. MetricsIntervalSeconds: uint32(ms.option.MetricsIntervalSec),
  65. StorageBackends: backend.ToPbStorageBackends(),
  66. }); err != nil {
  67. glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
  68. return err
  69. }
  70. }
  71. glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
  72. message := &master_pb.VolumeLocation{
  73. Url: dn.Url(),
  74. PublicUrl: dn.PublicUrl,
  75. }
  76. if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
  77. // process delta volume ids if exists for fast volume id updates
  78. for _, volInfo := range heartbeat.NewVolumes {
  79. message.NewVids = append(message.NewVids, volInfo.Id)
  80. }
  81. for _, volInfo := range heartbeat.DeletedVolumes {
  82. message.DeletedVids = append(message.DeletedVids, volInfo.Id)
  83. }
  84. // update master internal volume layouts
  85. t.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
  86. }
  87. if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
  88. // process heartbeat.Volumes
  89. newVolumes, deletedVolumes := t.SyncDataNodeRegistration(heartbeat.Volumes, dn)
  90. for _, v := range newVolumes {
  91. glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
  92. message.NewVids = append(message.NewVids, uint32(v.Id))
  93. }
  94. for _, v := range deletedVolumes {
  95. glog.V(0).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
  96. message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
  97. }
  98. }
  99. if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
  100. // update master internal volume layouts
  101. t.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)
  102. for _, s := range heartbeat.NewEcShards {
  103. message.NewVids = append(message.NewVids, s.Id)
  104. }
  105. for _, s := range heartbeat.DeletedEcShards {
  106. if dn.HasVolumesById(needle.VolumeId(s.Id)) {
  107. continue
  108. }
  109. message.DeletedVids = append(message.DeletedVids, s.Id)
  110. }
  111. }
  112. if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
  113. glog.V(1).Infof("master recieved ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
  114. newShards, deletedShards := t.SyncDataNodeEcShards(heartbeat.EcShards, dn)
  115. // broadcast the ec vid changes to master clients
  116. for _, s := range newShards {
  117. message.NewVids = append(message.NewVids, uint32(s.VolumeId))
  118. }
  119. for _, s := range deletedShards {
  120. if dn.HasVolumesById(s.VolumeId) {
  121. continue
  122. }
  123. message.DeletedVids = append(message.DeletedVids, uint32(s.VolumeId))
  124. }
  125. }
  126. if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 {
  127. ms.clientChansLock.RLock()
  128. for host, ch := range ms.clientChans {
  129. glog.V(0).Infof("master send to %s: %s", host, message.String())
  130. ch <- message
  131. }
  132. ms.clientChansLock.RUnlock()
  133. }
  134. // tell the volume servers about the leader
  135. newLeader, err := t.Leader()
  136. if err != nil {
  137. glog.Warningf("SendHeartbeat find leader: %v", err)
  138. return err
  139. }
  140. if err := stream.Send(&master_pb.HeartbeatResponse{
  141. Leader: newLeader,
  142. }); err != nil {
  143. glog.Warningf("SendHeartbeat.Send response to to %s:%d %v", dn.Ip, dn.Port, err)
  144. return err
  145. }
  146. }
  147. }
  148. // KeepConnected keep a stream gRPC call to the master. Used by clients to know the master is up.
  149. // And clients gets the up-to-date list of volume locations
  150. func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServer) error {
  151. req, err := stream.Recv()
  152. if err != nil {
  153. return err
  154. }
  155. if !ms.Topo.IsLeader() {
  156. return ms.informNewLeader(stream)
  157. }
  158. peerAddress := findClientAddress(stream.Context(), req.GrpcPort)
  159. stopChan := make(chan bool)
  160. clientName, messageChan := ms.addClient(req.Name, peerAddress)
  161. defer ms.deleteClient(clientName)
  162. for _, message := range ms.Topo.ToVolumeLocations() {
  163. if err := stream.Send(message); err != nil {
  164. return err
  165. }
  166. }
  167. go func() {
  168. for {
  169. _, err := stream.Recv()
  170. if err != nil {
  171. glog.V(2).Infof("- client %v: %v", clientName, err)
  172. stopChan <- true
  173. break
  174. }
  175. }
  176. }()
  177. ticker := time.NewTicker(5 * time.Second)
  178. for {
  179. select {
  180. case message := <-messageChan:
  181. if err := stream.Send(message); err != nil {
  182. glog.V(0).Infof("=> client %v: %+v", clientName, message)
  183. return err
  184. }
  185. case <-ticker.C:
  186. if !ms.Topo.IsLeader() {
  187. return ms.informNewLeader(stream)
  188. }
  189. case <-stopChan:
  190. return nil
  191. }
  192. }
  193. return nil
  194. }
  195. func (ms *MasterServer) informNewLeader(stream master_pb.Seaweed_KeepConnectedServer) error {
  196. leader, err := ms.Topo.Leader()
  197. if err != nil {
  198. glog.Errorf("topo leader: %v", err)
  199. return raft.NotLeaderError
  200. }
  201. if err := stream.Send(&master_pb.VolumeLocation{
  202. Leader: leader,
  203. }); err != nil {
  204. return err
  205. }
  206. return nil
  207. }
  208. func (ms *MasterServer) addClient(clientType string, clientAddress string) (clientName string, messageChan chan *master_pb.VolumeLocation) {
  209. clientName = clientType + "@" + clientAddress
  210. glog.V(0).Infof("+ client %v", clientName)
  211. messageChan = make(chan *master_pb.VolumeLocation)
  212. ms.clientChansLock.Lock()
  213. ms.clientChans[clientName] = messageChan
  214. ms.clientChansLock.Unlock()
  215. return
  216. }
  217. func (ms *MasterServer) deleteClient(clientName string) {
  218. glog.V(0).Infof("- client %v", clientName)
  219. ms.clientChansLock.Lock()
  220. delete(ms.clientChans, clientName)
  221. ms.clientChansLock.Unlock()
  222. }
  223. func findClientAddress(ctx context.Context, grpcPort uint32) string {
  224. // fmt.Printf("FromContext %+v\n", ctx)
  225. pr, ok := peer.FromContext(ctx)
  226. if !ok {
  227. glog.Error("failed to get peer from ctx")
  228. return ""
  229. }
  230. if pr.Addr == net.Addr(nil) {
  231. glog.Error("failed to get peer address")
  232. return ""
  233. }
  234. if grpcPort == 0 {
  235. return pr.Addr.String()
  236. }
  237. if tcpAddr, ok := pr.Addr.(*net.TCPAddr); ok {
  238. externalIP := tcpAddr.IP
  239. return fmt.Sprintf("%s:%d", externalIP, grpcPort)
  240. }
  241. return pr.Addr.String()
  242. }
  243. func (ms *MasterServer) ListMasterClients(ctx context.Context, req *master_pb.ListMasterClientsRequest) (*master_pb.ListMasterClientsResponse, error) {
  244. resp := &master_pb.ListMasterClientsResponse{}
  245. ms.clientChansLock.RLock()
  246. defer ms.clientChansLock.RUnlock()
  247. for k := range ms.clientChans {
  248. if strings.HasPrefix(k, req.ClientType+"@") {
  249. resp.GrpcAddresses = append(resp.GrpcAddresses, k[len(req.ClientType)+1:])
  250. }
  251. }
  252. return resp, nil
  253. }