master_grpc_server.go 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427
  1. package weed_server
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "github.com/google/uuid"
  7. "github.com/seaweedfs/seaweedfs/weed/cluster"
  8. "net"
  9. "sort"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/pb"
  12. "github.com/seaweedfs/seaweedfs/weed/stats"
  13. "github.com/seaweedfs/seaweedfs/weed/storage/backend"
  14. "github.com/seaweedfs/seaweedfs/weed/util"
  15. "github.com/seaweedfs/raft"
  16. "google.golang.org/grpc/peer"
  17. "github.com/seaweedfs/seaweedfs/weed/glog"
  18. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  19. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  20. "github.com/seaweedfs/seaweedfs/weed/topology"
  21. )
  22. func (ms *MasterServer) RegisterUuids(heartbeat *master_pb.Heartbeat) (duplicated_uuids []string, err error) {
  23. ms.Topo.UuidAccessLock.Lock()
  24. defer ms.Topo.UuidAccessLock.Unlock()
  25. key := fmt.Sprintf("%s:%d", heartbeat.Ip, heartbeat.Port)
  26. if ms.Topo.UuidMap == nil {
  27. ms.Topo.UuidMap = make(map[string][]string)
  28. }
  29. // find whether new uuid exists
  30. for k, v := range ms.Topo.UuidMap {
  31. sort.Strings(v)
  32. for _, id := range heartbeat.LocationUuids {
  33. index := sort.SearchStrings(v, id)
  34. if index < len(v) && v[index] == id {
  35. duplicated_uuids = append(duplicated_uuids, id)
  36. glog.Errorf("directory of %s on %s has been loaded", id, k)
  37. }
  38. }
  39. }
  40. if len(duplicated_uuids) > 0 {
  41. return duplicated_uuids, errors.New("volume: Duplicated volume directories were loaded")
  42. }
  43. ms.Topo.UuidMap[key] = heartbeat.LocationUuids
  44. glog.V(0).Infof("found new uuid:%v %v , %v", key, heartbeat.LocationUuids, ms.Topo.UuidMap)
  45. return nil, nil
  46. }
  47. func (ms *MasterServer) UnRegisterUuids(ip string, port int) {
  48. ms.Topo.UuidAccessLock.Lock()
  49. defer ms.Topo.UuidAccessLock.Unlock()
  50. key := fmt.Sprintf("%s:%d", ip, port)
  51. delete(ms.Topo.UuidMap, key)
  52. glog.V(0).Infof("remove volume server %v, online volume server: %v", key, ms.Topo.UuidMap)
  53. }
  54. func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
  55. var dn *topology.DataNode
  56. defer func() {
  57. if dn != nil {
  58. dn.Counter--
  59. if dn.Counter > 0 {
  60. glog.V(0).Infof("disconnect phantom volume server %s:%d remaining %d", dn.Ip, dn.Port, dn.Counter)
  61. return
  62. }
  63. message := &master_pb.VolumeLocation{
  64. DataCenter: dn.GetDataCenterId(),
  65. Url: dn.Url(),
  66. PublicUrl: dn.PublicUrl,
  67. GrpcPort: uint32(dn.GrpcPort),
  68. }
  69. for _, v := range dn.GetVolumes() {
  70. message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
  71. }
  72. for _, s := range dn.GetEcShards() {
  73. message.DeletedVids = append(message.DeletedVids, uint32(s.VolumeId))
  74. }
  75. // if the volume server disconnects and reconnects quickly
  76. // the unregister and register can race with each other
  77. ms.Topo.UnRegisterDataNode(dn)
  78. glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
  79. ms.UnRegisterUuids(dn.Ip, dn.Port)
  80. if len(message.DeletedVids) > 0 {
  81. ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
  82. }
  83. }
  84. }()
  85. for {
  86. heartbeat, err := stream.Recv()
  87. if err != nil {
  88. if dn != nil {
  89. glog.Warningf("SendHeartbeat.Recv server %s:%d : %v", dn.Ip, dn.Port, err)
  90. } else {
  91. glog.Warningf("SendHeartbeat.Recv: %v", err)
  92. }
  93. stats.MasterReceivedHeartbeatCounter.WithLabelValues("error").Inc()
  94. return err
  95. }
  96. if !ms.Topo.IsLeader() {
  97. // tell the volume servers about the leader
  98. newLeader, err := ms.Topo.Leader()
  99. if err != nil {
  100. glog.Warningf("SendHeartbeat find leader: %v", err)
  101. return err
  102. }
  103. if err := stream.Send(&master_pb.HeartbeatResponse{
  104. Leader: string(newLeader),
  105. }); err != nil {
  106. if dn != nil {
  107. glog.Warningf("SendHeartbeat.Send response to %s:%d %v", dn.Ip, dn.Port, err)
  108. } else {
  109. glog.Warningf("SendHeartbeat.Send response %v", err)
  110. }
  111. return err
  112. }
  113. continue
  114. }
  115. ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
  116. if dn == nil {
  117. // Skip delta heartbeat for volume server versions better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630
  118. if heartbeat.Ip == "" {
  119. continue
  120. } // ToDo must be removed after update major version
  121. dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
  122. dc := ms.Topo.GetOrCreateDataCenter(dcName)
  123. rack := dc.GetOrCreateRack(rackName)
  124. dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.MaxVolumeCounts)
  125. glog.V(0).Infof("added volume server %d: %v:%d %v", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids)
  126. uuidlist, err := ms.RegisterUuids(heartbeat)
  127. if err != nil {
  128. if stream_err := stream.Send(&master_pb.HeartbeatResponse{
  129. DuplicatedUuids: uuidlist,
  130. }); stream_err != nil {
  131. glog.Warningf("SendHeartbeat.Send DuplicatedDirectory response to %s:%d %v", dn.Ip, dn.Port, stream_err)
  132. return stream_err
  133. }
  134. return err
  135. }
  136. if err := stream.Send(&master_pb.HeartbeatResponse{
  137. VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
  138. }); err != nil {
  139. glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
  140. return err
  141. }
  142. stats.MasterReceivedHeartbeatCounter.WithLabelValues("dataNode").Inc()
  143. dn.Counter++
  144. }
  145. dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)
  146. glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
  147. stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()
  148. message := &master_pb.VolumeLocation{
  149. Url: dn.Url(),
  150. PublicUrl: dn.PublicUrl,
  151. DataCenter: dn.GetDataCenterId(),
  152. GrpcPort: uint32(dn.GrpcPort),
  153. }
  154. if len(heartbeat.NewVolumes) > 0 {
  155. stats.MasterReceivedHeartbeatCounter.WithLabelValues("newVolumes").Inc()
  156. }
  157. if len(heartbeat.DeletedVolumes) > 0 {
  158. stats.MasterReceivedHeartbeatCounter.WithLabelValues("deletedVolumes").Inc()
  159. }
  160. if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
  161. // process delta volume ids if exists for fast volume id updates
  162. for _, volInfo := range heartbeat.NewVolumes {
  163. message.NewVids = append(message.NewVids, volInfo.Id)
  164. }
  165. for _, volInfo := range heartbeat.DeletedVolumes {
  166. message.DeletedVids = append(message.DeletedVids, volInfo.Id)
  167. }
  168. // update master internal volume layouts
  169. ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
  170. }
  171. if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
  172. if heartbeat.Ip != "" {
  173. dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
  174. ms.Topo.DataNodeRegistration(dcName, rackName, dn)
  175. }
  176. // process heartbeat.Volumes
  177. stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc()
  178. newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)
  179. for _, v := range newVolumes {
  180. glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
  181. message.NewVids = append(message.NewVids, uint32(v.Id))
  182. }
  183. for _, v := range deletedVolumes {
  184. glog.V(0).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
  185. message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
  186. }
  187. }
  188. if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
  189. stats.MasterReceivedHeartbeatCounter.WithLabelValues("newEcShards").Inc()
  190. // update master internal volume layouts
  191. ms.Topo.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)
  192. for _, s := range heartbeat.NewEcShards {
  193. message.NewEcVids = append(message.NewEcVids, s.Id)
  194. }
  195. for _, s := range heartbeat.DeletedEcShards {
  196. if dn.HasEcShards(needle.VolumeId(s.Id)) {
  197. continue
  198. }
  199. message.DeletedEcVids = append(message.DeletedEcVids, s.Id)
  200. }
  201. }
  202. if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
  203. stats.MasterReceivedHeartbeatCounter.WithLabelValues("ecShards").Inc()
  204. glog.V(4).Infof("master received ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
  205. newShards, deletedShards := ms.Topo.SyncDataNodeEcShards(heartbeat.EcShards, dn)
  206. // broadcast the ec vid changes to master clients
  207. for _, s := range newShards {
  208. message.NewEcVids = append(message.NewEcVids, uint32(s.VolumeId))
  209. }
  210. for _, s := range deletedShards {
  211. if dn.HasVolumesById(s.VolumeId) {
  212. continue
  213. }
  214. message.DeletedEcVids = append(message.DeletedEcVids, uint32(s.VolumeId))
  215. }
  216. }
  217. if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 {
  218. ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
  219. }
  220. }
  221. }
  222. // KeepConnected keep a stream gRPC call to the master. Used by clients to know the master is up.
  223. // And clients gets the up-to-date list of volume locations
  224. func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServer) error {
  225. req, recvErr := stream.Recv()
  226. if recvErr != nil {
  227. return recvErr
  228. }
  229. if !ms.Topo.IsLeader() {
  230. return ms.informNewLeader(stream)
  231. }
  232. clientAddress := req.ClientAddress
  233. // Ensure that the clientAddress is unique.
  234. if clientAddress == "" {
  235. clientAddress = uuid.New().String()
  236. }
  237. peerAddress := pb.ServerAddress(clientAddress)
  238. // buffer by 1 so we don't end up getting stuck writing to stopChan forever
  239. stopChan := make(chan bool, 1)
  240. clientName, messageChan := ms.addClient(req.FilerGroup, req.ClientType, peerAddress)
  241. for _, update := range ms.Cluster.AddClusterNode(req.FilerGroup, req.ClientType, cluster.DataCenter(req.DataCenter), cluster.Rack(req.Rack), peerAddress, req.Version) {
  242. ms.broadcastToClients(update)
  243. }
  244. defer func() {
  245. for _, update := range ms.Cluster.RemoveClusterNode(req.FilerGroup, req.ClientType, peerAddress) {
  246. ms.broadcastToClients(update)
  247. }
  248. ms.deleteClient(clientName)
  249. }()
  250. for i, message := range ms.Topo.ToVolumeLocations() {
  251. if i == 0 {
  252. if leader, err := ms.Topo.Leader(); err == nil {
  253. message.Leader = string(leader)
  254. }
  255. }
  256. if sendErr := stream.Send(&master_pb.KeepConnectedResponse{VolumeLocation: message}); sendErr != nil {
  257. return sendErr
  258. }
  259. }
  260. go func() {
  261. for {
  262. _, err := stream.Recv()
  263. if err != nil {
  264. glog.V(2).Infof("- client %v: %v", clientName, err)
  265. go func() {
  266. // consume message chan to avoid deadlock, go routine exit when message chan is closed
  267. for range messageChan {
  268. // no op
  269. }
  270. }()
  271. close(stopChan)
  272. return
  273. }
  274. }
  275. }()
  276. ticker := time.NewTicker(5 * time.Second)
  277. defer ticker.Stop()
  278. for {
  279. select {
  280. case message := <-messageChan:
  281. if err := stream.Send(message); err != nil {
  282. glog.V(0).Infof("=> client %v: %+v", clientName, message)
  283. return err
  284. }
  285. case <-ticker.C:
  286. if !ms.Topo.IsLeader() {
  287. stats.MasterRaftIsleader.Set(0)
  288. stats.MasterAdminLock.Reset()
  289. stats.MasterReplicaPlacementMismatch.Reset()
  290. return ms.informNewLeader(stream)
  291. } else {
  292. stats.MasterRaftIsleader.Set(1)
  293. }
  294. case <-stopChan:
  295. return nil
  296. }
  297. }
  298. }
  299. func (ms *MasterServer) broadcastToClients(message *master_pb.KeepConnectedResponse) {
  300. ms.clientChansLock.RLock()
  301. for _, ch := range ms.clientChans {
  302. ch <- message
  303. }
  304. ms.clientChansLock.RUnlock()
  305. }
  306. func (ms *MasterServer) informNewLeader(stream master_pb.Seaweed_KeepConnectedServer) error {
  307. leader, err := ms.Topo.Leader()
  308. if err != nil {
  309. glog.Errorf("topo leader: %v", err)
  310. return raft.NotLeaderError
  311. }
  312. if err := stream.Send(&master_pb.KeepConnectedResponse{
  313. VolumeLocation: &master_pb.VolumeLocation{
  314. Leader: string(leader),
  315. },
  316. }); err != nil {
  317. return err
  318. }
  319. return nil
  320. }
  321. func (ms *MasterServer) addClient(filerGroup, clientType string, clientAddress pb.ServerAddress) (clientName string, messageChan chan *master_pb.KeepConnectedResponse) {
  322. clientName = filerGroup + "." + clientType + "@" + string(clientAddress)
  323. glog.V(0).Infof("+ client %v", clientName)
  324. // we buffer this because otherwise we end up in a potential deadlock where
  325. // the KeepConnected loop is no longer listening on this channel but we're
  326. // trying to send to it in SendHeartbeat and so we can't lock the
  327. // clientChansLock to remove the channel and we're stuck writing to it
  328. messageChan = make(chan *master_pb.KeepConnectedResponse, 10000)
  329. ms.clientChansLock.Lock()
  330. ms.clientChans[clientName] = messageChan
  331. ms.clientChansLock.Unlock()
  332. return
  333. }
  334. func (ms *MasterServer) deleteClient(clientName string) {
  335. glog.V(0).Infof("- client %v", clientName)
  336. ms.clientChansLock.Lock()
  337. // close message chan, so that the KeepConnected go routine can exit
  338. if clientChan, ok := ms.clientChans[clientName]; ok {
  339. close(clientChan)
  340. delete(ms.clientChans, clientName)
  341. }
  342. ms.clientChansLock.Unlock()
  343. }
  344. func findClientAddress(ctx context.Context, grpcPort uint32) string {
  345. // fmt.Printf("FromContext %+v\n", ctx)
  346. pr, ok := peer.FromContext(ctx)
  347. if !ok {
  348. glog.Error("failed to get peer from ctx")
  349. return ""
  350. }
  351. if pr.Addr == net.Addr(nil) {
  352. glog.Error("failed to get peer address")
  353. return ""
  354. }
  355. if grpcPort == 0 {
  356. return pr.Addr.String()
  357. }
  358. if tcpAddr, ok := pr.Addr.(*net.TCPAddr); ok {
  359. externalIP := tcpAddr.IP
  360. return util.JoinHostPort(externalIP.String(), int(grpcPort))
  361. }
  362. return pr.Addr.String()
  363. }
  364. func (ms *MasterServer) GetMasterConfiguration(ctx context.Context, req *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
  365. // tell the volume servers about the leader
  366. leader, _ := ms.Topo.Leader()
  367. resp := &master_pb.GetMasterConfigurationResponse{
  368. MetricsAddress: ms.option.MetricsAddress,
  369. MetricsIntervalSeconds: uint32(ms.option.MetricsIntervalSec),
  370. StorageBackends: backend.ToPbStorageBackends(),
  371. DefaultReplication: ms.option.DefaultReplicaPlacement,
  372. VolumeSizeLimitMB: uint32(ms.option.VolumeSizeLimitMB),
  373. VolumePreallocate: ms.option.VolumePreallocate,
  374. Leader: string(leader),
  375. }
  376. return resp, nil
  377. }