volume_grpc_client_to_master.go 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266
  1. package weed_server
  2. import (
  3. "fmt"
  4. "os"
  5. "time"
  6. "github.com/seaweedfs/seaweedfs/weed/operation"
  7. "google.golang.org/grpc"
  8. "github.com/seaweedfs/seaweedfs/weed/pb"
  9. "github.com/seaweedfs/seaweedfs/weed/security"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/backend"
  11. "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
  12. "golang.org/x/net/context"
  13. "github.com/seaweedfs/seaweedfs/weed/glog"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/util"
  16. )
  17. func (vs *VolumeServer) GetMaster() pb.ServerAddress {
  18. return vs.currentMaster
  19. }
  20. func (vs *VolumeServer) checkWithMaster() (err error) {
  21. for {
  22. for _, master := range vs.SeedMasterNodes {
  23. err = operation.WithMasterServerClient(false, master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
  24. resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
  25. if err != nil {
  26. return fmt.Errorf("get master %s configuration: %v", master, err)
  27. }
  28. vs.metricsAddress, vs.metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSeconds)
  29. backend.LoadFromPbStorageBackends(resp.StorageBackends)
  30. return nil
  31. })
  32. if err == nil {
  33. return
  34. } else {
  35. glog.V(0).Infof("checkWithMaster %s: %v", master, err)
  36. }
  37. }
  38. time.Sleep(1790 * time.Millisecond)
  39. }
  40. }
  41. func (vs *VolumeServer) heartbeat() {
  42. glog.V(0).Infof("Volume server start with seed master nodes: %v", vs.SeedMasterNodes)
  43. vs.store.SetDataCenter(vs.dataCenter)
  44. vs.store.SetRack(vs.rack)
  45. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume")
  46. var err error
  47. var newLeader pb.ServerAddress
  48. for vs.isHeartbeating {
  49. for _, master := range vs.SeedMasterNodes {
  50. if newLeader != "" {
  51. // the new leader may actually is the same master
  52. // need to wait a bit before adding itself
  53. time.Sleep(3 * time.Second)
  54. master = newLeader
  55. }
  56. vs.store.MasterAddress = master
  57. newLeader, err = vs.doHeartbeat(master, grpcDialOption, time.Duration(vs.pulseSeconds)*time.Second)
  58. if err != nil {
  59. glog.V(0).Infof("heartbeat to %s error: %v", master, err)
  60. time.Sleep(time.Duration(vs.pulseSeconds) * time.Second)
  61. newLeader = ""
  62. vs.store.MasterAddress = ""
  63. }
  64. if !vs.isHeartbeating {
  65. break
  66. }
  67. }
  68. }
  69. }
  70. func (vs *VolumeServer) StopHeartbeat() (isAlreadyStopping bool) {
  71. if !vs.isHeartbeating {
  72. return true
  73. }
  74. vs.isHeartbeating = false
  75. close(vs.stopChan)
  76. return false
  77. }
  78. func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOption grpc.DialOption, sleepInterval time.Duration) (newLeader pb.ServerAddress, err error) {
  79. ctx, cancel := context.WithCancel(context.Background())
  80. defer cancel()
  81. grpcConnection, err := pb.GrpcDial(ctx, masterAddress.ToGrpcAddress(), false, grpcDialOption)
  82. if err != nil {
  83. return "", fmt.Errorf("fail to dial %s : %v", masterAddress, err)
  84. }
  85. defer grpcConnection.Close()
  86. client := master_pb.NewSeaweedClient(grpcConnection)
  87. stream, err := client.SendHeartbeat(ctx)
  88. if err != nil {
  89. glog.V(0).Infof("SendHeartbeat to %s: %v", masterAddress, err)
  90. return "", err
  91. }
  92. glog.V(0).Infof("Heartbeat to: %v", masterAddress)
  93. vs.currentMaster = masterAddress
  94. doneChan := make(chan error, 1)
  95. go func() {
  96. for {
  97. in, err := stream.Recv()
  98. if err != nil {
  99. doneChan <- err
  100. return
  101. }
  102. if len(in.DuplicatedUuids) > 0 {
  103. var duplicateDir []string
  104. for _, loc := range vs.store.Locations {
  105. for _, uuid := range in.DuplicatedUuids {
  106. if uuid == loc.DirectoryUuid {
  107. duplicateDir = append(duplicateDir, loc.Directory)
  108. }
  109. }
  110. }
  111. glog.Errorf("Shut down Volume Server due to duplicate volume directories: %v", duplicateDir)
  112. os.Exit(1)
  113. }
  114. if in.GetVolumeSizeLimit() != 0 && vs.store.GetVolumeSizeLimit() != in.GetVolumeSizeLimit() {
  115. vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit())
  116. if vs.store.MaybeAdjustVolumeMax() {
  117. if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
  118. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", vs.currentMaster, err)
  119. return
  120. }
  121. }
  122. }
  123. if in.GetLeader() != "" && string(vs.currentMaster) != in.GetLeader() {
  124. glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), vs.currentMaster)
  125. newLeader = pb.ServerAddress(in.GetLeader())
  126. doneChan <- nil
  127. return
  128. }
  129. }
  130. }()
  131. if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
  132. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
  133. return "", err
  134. }
  135. if err = stream.Send(vs.store.CollectErasureCodingHeartbeat()); err != nil {
  136. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
  137. return "", err
  138. }
  139. volumeTickChan := time.Tick(sleepInterval)
  140. ecShardTickChan := time.Tick(17 * sleepInterval)
  141. dataCenter := vs.store.GetDataCenter()
  142. rack := vs.store.GetRack()
  143. ip := vs.store.Ip
  144. port := uint32(vs.store.Port)
  145. for {
  146. select {
  147. case volumeMessage := <-vs.store.NewVolumesChan:
  148. deltaBeat := &master_pb.Heartbeat{
  149. Ip: ip,
  150. Port: port,
  151. DataCenter: dataCenter,
  152. Rack: rack,
  153. NewVolumes: []*master_pb.VolumeShortInformationMessage{
  154. &volumeMessage,
  155. },
  156. }
  157. glog.V(0).Infof("volume server %s:%d adds volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
  158. if err = stream.Send(deltaBeat); err != nil {
  159. glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
  160. return "", err
  161. }
  162. case ecShardMessage := <-vs.store.NewEcShardsChan:
  163. deltaBeat := &master_pb.Heartbeat{
  164. Ip: ip,
  165. Port: port,
  166. DataCenter: dataCenter,
  167. Rack: rack,
  168. NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
  169. &ecShardMessage,
  170. },
  171. }
  172. glog.V(0).Infof("volume server %s:%d adds ec shard %d:%d", vs.store.Ip, vs.store.Port, ecShardMessage.Id,
  173. erasure_coding.ShardBits(ecShardMessage.EcIndexBits).ShardIds())
  174. if err = stream.Send(deltaBeat); err != nil {
  175. glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
  176. return "", err
  177. }
  178. case volumeMessage := <-vs.store.DeletedVolumesChan:
  179. deltaBeat := &master_pb.Heartbeat{
  180. Ip: ip,
  181. Port: port,
  182. DataCenter: dataCenter,
  183. Rack: rack,
  184. DeletedVolumes: []*master_pb.VolumeShortInformationMessage{
  185. &volumeMessage,
  186. },
  187. }
  188. glog.V(0).Infof("volume server %s:%d deletes volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
  189. if err = stream.Send(deltaBeat); err != nil {
  190. glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
  191. return "", err
  192. }
  193. case ecShardMessage := <-vs.store.DeletedEcShardsChan:
  194. deltaBeat := &master_pb.Heartbeat{
  195. Ip: ip,
  196. Port: port,
  197. DataCenter: dataCenter,
  198. Rack: rack,
  199. DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{
  200. &ecShardMessage,
  201. },
  202. }
  203. glog.V(0).Infof("volume server %s:%d deletes ec shard %d:%d", vs.store.Ip, vs.store.Port, ecShardMessage.Id,
  204. erasure_coding.ShardBits(ecShardMessage.EcIndexBits).ShardIds())
  205. if err = stream.Send(deltaBeat); err != nil {
  206. glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
  207. return "", err
  208. }
  209. case <-volumeTickChan:
  210. glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
  211. vs.store.MaybeAdjustVolumeMax()
  212. if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
  213. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
  214. return "", err
  215. }
  216. case <-ecShardTickChan:
  217. glog.V(4).Infof("volume server %s:%d ec heartbeat", vs.store.Ip, vs.store.Port)
  218. if err = stream.Send(vs.store.CollectErasureCodingHeartbeat()); err != nil {
  219. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
  220. return "", err
  221. }
  222. case err = <-doneChan:
  223. return
  224. case <-vs.stopChan:
  225. var volumeMessages []*master_pb.VolumeInformationMessage
  226. emptyBeat := &master_pb.Heartbeat{
  227. Ip: ip,
  228. Port: port,
  229. PublicUrl: vs.store.PublicUrl,
  230. MaxFileKey: uint64(0),
  231. DataCenter: dataCenter,
  232. Rack: rack,
  233. Volumes: volumeMessages,
  234. HasNoVolumes: len(volumeMessages) == 0,
  235. }
  236. glog.V(1).Infof("volume server %s:%d stops and deletes all volumes", vs.store.Ip, vs.store.Port)
  237. if err = stream.Send(emptyBeat); err != nil {
  238. glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
  239. return "", err
  240. }
  241. return
  242. }
  243. }
  244. }