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.

257 lines
8.6 KiB

6 years ago
6 years ago
5 years ago
7 years ago
5 years ago
6 years ago
5 years ago
4 years ago
6 years ago
  1. package wdclient
  2. import (
  3. "context"
  4. "github.com/chrislusf/seaweedfs/weed/stats"
  5. "math/rand"
  6. "time"
  7. "github.com/chrislusf/seaweedfs/weed/util"
  8. "google.golang.org/grpc"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/pb"
  11. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  12. )
  13. type MasterClient struct {
  14. FilerGroup string
  15. clientType string
  16. clientHost pb.ServerAddress
  17. currentMaster pb.ServerAddress
  18. masters map[string]pb.ServerAddress
  19. grpcDialOption grpc.DialOption
  20. vidMap
  21. OnPeerUpdate func(update *master_pb.ClusterNodeUpdate, startFrom time.Time)
  22. }
  23. func NewMasterClient(grpcDialOption grpc.DialOption, filerGroup string, clientType string, clientHost pb.ServerAddress, clientDataCenter string, masters map[string]pb.ServerAddress) *MasterClient {
  24. return &MasterClient{
  25. FilerGroup: filerGroup,
  26. clientType: clientType,
  27. clientHost: clientHost,
  28. masters: masters,
  29. grpcDialOption: grpcDialOption,
  30. vidMap: newVidMap(clientDataCenter),
  31. }
  32. }
  33. func (mc *MasterClient) GetLookupFileIdFunction() LookupFileIdFunctionType {
  34. return mc.LookupFileIdWithFallback
  35. }
  36. func (mc *MasterClient) LookupFileIdWithFallback(fileId string) (fullUrls []string, err error) {
  37. fullUrls, err = mc.vidMap.LookupFileId(fileId)
  38. if err == nil {
  39. return
  40. }
  41. err = pb.WithMasterClient(false, mc.currentMaster, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
  42. resp, err := client.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{
  43. VolumeOrFileIds: []string{fileId},
  44. })
  45. if err != nil {
  46. return err
  47. }
  48. for vid, vidLocation := range resp.VolumeIdLocations {
  49. for _, vidLoc := range vidLocation.Locations {
  50. loc := Location{
  51. Url: vidLoc.Url,
  52. PublicUrl: vidLoc.PublicUrl,
  53. GrpcPort: int(vidLoc.GrpcPort),
  54. }
  55. mc.vidMap.addLocation(uint32(vid), loc)
  56. fullUrls = append(fullUrls, "http://"+loc.Url+"/"+fileId)
  57. }
  58. }
  59. return nil
  60. })
  61. return
  62. }
  63. func (mc *MasterClient) GetMaster() pb.ServerAddress {
  64. mc.WaitUntilConnected()
  65. return mc.currentMaster
  66. }
  67. func (mc *MasterClient) GetMasters() map[string]pb.ServerAddress {
  68. mc.WaitUntilConnected()
  69. return mc.masters
  70. }
  71. func (mc *MasterClient) WaitUntilConnected() {
  72. for mc.currentMaster == "" {
  73. time.Sleep(time.Duration(rand.Int31n(200)) * time.Millisecond)
  74. }
  75. }
  76. func (mc *MasterClient) KeepConnectedToMaster() {
  77. glog.V(1).Infof("%s.%s masterClient bootstraps with masters %v", mc.FilerGroup, mc.clientType, mc.masters)
  78. for {
  79. mc.tryAllMasters()
  80. time.Sleep(time.Second)
  81. }
  82. }
  83. func (mc *MasterClient) FindLeaderFromOtherPeers(myMasterAddress pb.ServerAddress) (leader string) {
  84. for _, master := range mc.masters {
  85. if master == myMasterAddress {
  86. continue
  87. }
  88. if grpcErr := pb.WithMasterClient(false, master, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
  89. ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
  90. defer cancel()
  91. resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})
  92. if err != nil {
  93. return err
  94. }
  95. leader = resp.Leader
  96. return nil
  97. }); grpcErr != nil {
  98. glog.V(0).Infof("connect to %s: %v", master, grpcErr)
  99. }
  100. if leader != "" {
  101. glog.V(0).Infof("existing leader is %s", leader)
  102. return
  103. }
  104. }
  105. glog.V(0).Infof("No existing leader found!")
  106. return
  107. }
  108. func (mc *MasterClient) tryAllMasters() {
  109. var nextHintedLeader pb.ServerAddress
  110. for _, master := range mc.masters {
  111. nextHintedLeader = mc.tryConnectToMaster(master)
  112. for nextHintedLeader != "" {
  113. nextHintedLeader = mc.tryConnectToMaster(nextHintedLeader)
  114. }
  115. mc.currentMaster = ""
  116. }
  117. }
  118. func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedLeader pb.ServerAddress) {
  119. glog.V(1).Infof("%s.%s masterClient Connecting to master %v", mc.FilerGroup, mc.clientType, master)
  120. stats.MasterClientConnectCounter.WithLabelValues("total").Inc()
  121. gprcErr := pb.WithMasterClient(true, master, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
  122. ctx, cancel := context.WithCancel(context.Background())
  123. defer cancel()
  124. stream, err := client.KeepConnected(ctx)
  125. if err != nil {
  126. glog.V(1).Infof("%s.%s masterClient failed to keep connected to %s: %v", mc.FilerGroup, mc.clientType, master, err)
  127. stats.MasterClientConnectCounter.WithLabelValues(stats.FailedToKeepConnected).Inc()
  128. return err
  129. }
  130. if err = stream.Send(&master_pb.KeepConnectedRequest{
  131. FilerGroup: mc.FilerGroup,
  132. ClientType: mc.clientType,
  133. ClientAddress: string(mc.clientHost),
  134. Version: util.Version(),
  135. }); err != nil {
  136. glog.V(0).Infof("%s.%s masterClient failed to send to %s: %v", mc.FilerGroup, mc.clientType, master, err)
  137. stats.MasterClientConnectCounter.WithLabelValues(stats.FailedToSend).Inc()
  138. return err
  139. }
  140. glog.V(1).Infof("%s.%s masterClient Connected to %v", mc.FilerGroup, mc.clientType, master)
  141. resp, err := stream.Recv()
  142. if err != nil {
  143. glog.V(0).Infof("%s.%s masterClient failed to receive from %s: %v", mc.FilerGroup, mc.clientType, master, err)
  144. stats.MasterClientConnectCounter.WithLabelValues(stats.FailedToReceive).Inc()
  145. return err
  146. }
  147. // check if it is the leader to determine whether to reset the vidMap
  148. if resp.VolumeLocation != nil && resp.VolumeLocation.Leader != "" && string(master) != resp.VolumeLocation.Leader {
  149. glog.V(0).Infof("master %v redirected to leader %v", master, resp.VolumeLocation.Leader)
  150. nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
  151. stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToleader).Inc()
  152. return nil
  153. }
  154. mc.currentMaster = master
  155. mc.vidMap = newVidMap("")
  156. for {
  157. resp, err := stream.Recv()
  158. if err != nil {
  159. glog.V(0).Infof("%s.%s masterClient failed to receive from %s: %v", mc.FilerGroup, mc.clientType, master, err)
  160. stats.MasterClientConnectCounter.WithLabelValues(stats.FailedToReceive).Inc()
  161. return err
  162. }
  163. if resp.VolumeLocation != nil {
  164. // maybe the leader is changed
  165. if resp.VolumeLocation.Leader != "" && string(mc.currentMaster) != resp.VolumeLocation.Leader {
  166. glog.V(0).Infof("currentMaster %v redirected to leader %v", mc.currentMaster, resp.VolumeLocation.Leader)
  167. nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
  168. stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToleader).Inc()
  169. return nil
  170. }
  171. // process new volume location
  172. loc := Location{
  173. Url: resp.VolumeLocation.Url,
  174. PublicUrl: resp.VolumeLocation.PublicUrl,
  175. DataCenter: resp.VolumeLocation.DataCenter,
  176. GrpcPort: int(resp.VolumeLocation.GrpcPort),
  177. }
  178. for _, newVid := range resp.VolumeLocation.NewVids {
  179. glog.V(1).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, loc.Url, newVid)
  180. mc.addLocation(newVid, loc)
  181. }
  182. for _, deletedVid := range resp.VolumeLocation.DeletedVids {
  183. glog.V(1).Infof("%s.%s: %s masterClient removes volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedVid)
  184. mc.deleteLocation(deletedVid, loc)
  185. }
  186. for _, newEcVid := range resp.VolumeLocation.NewEcVids {
  187. glog.V(1).Infof("%s.%s: %s masterClient adds ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, newEcVid)
  188. mc.addEcLocation(newEcVid, loc)
  189. }
  190. for _, deletedEcVid := range resp.VolumeLocation.DeletedEcVids {
  191. glog.V(1).Infof("%s.%s: %s masterClient removes ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedEcVid)
  192. mc.deleteEcLocation(deletedEcVid, loc)
  193. }
  194. }
  195. if resp.ClusterNodeUpdate != nil {
  196. update := resp.ClusterNodeUpdate
  197. if mc.OnPeerUpdate != nil {
  198. if update.FilerGroup == mc.FilerGroup {
  199. if update.IsAdd {
  200. glog.V(0).Infof("+ %s.%s %s leader:%v\n", update.FilerGroup, update.NodeType, update.Address, update.IsLeader)
  201. } else {
  202. glog.V(0).Infof("- %s.%s %s leader:%v\n", update.FilerGroup, update.NodeType, update.Address, update.IsLeader)
  203. }
  204. stats.MasterClientConnectCounter.WithLabelValues(stats.OnPeerUpdate).Inc()
  205. mc.OnPeerUpdate(update, time.Now())
  206. }
  207. }
  208. }
  209. }
  210. })
  211. if gprcErr != nil {
  212. stats.MasterClientConnectCounter.WithLabelValues(stats.Failed).Inc()
  213. glog.V(1).Infof("%s.%s masterClient failed to connect with master %v: %v", mc.FilerGroup, mc.clientType, master, gprcErr)
  214. }
  215. return
  216. }
  217. func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
  218. return util.Retry("master grpc", func() error {
  219. for mc.currentMaster == "" {
  220. time.Sleep(3 * time.Second)
  221. }
  222. return pb.WithMasterClient(streamingMode, mc.currentMaster, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
  223. return fn(client)
  224. })
  225. })
  226. }