volume_grpc_client.go 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. package weed_server
  2. import (
  3. "fmt"
  4. "time"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  7. "github.com/chrislusf/seaweedfs/weed/security"
  8. "github.com/chrislusf/seaweedfs/weed/storage"
  9. "golang.org/x/net/context"
  10. "google.golang.org/grpc"
  11. )
  12. func (vs *VolumeServer) heartbeat() {
  13. glog.V(0).Infof("Volume server bootstraps with master %s", vs.GetMasterNode())
  14. vs.masterNodes = storage.NewMasterNodes(vs.masterNode)
  15. vs.store.SetDataCenter(vs.dataCenter)
  16. vs.store.SetRack(vs.rack)
  17. for {
  18. err := vs.doHeartbeat(time.Duration(vs.pulseSeconds) * time.Second)
  19. if err != nil {
  20. glog.V(0).Infof("heartbeat error: %v", err)
  21. time.Sleep(time.Duration(vs.pulseSeconds) * time.Second)
  22. }
  23. }
  24. }
  25. func (vs *VolumeServer) doHeartbeat(sleepInterval time.Duration) error {
  26. vs.masterNodes.Reset()
  27. masterNode, err := vs.masterNodes.FindMaster()
  28. if err != nil {
  29. return fmt.Errorf("No master found: %v", err)
  30. }
  31. grpcConection, err := grpc.Dial(masterNode, grpc.WithInsecure())
  32. if err != nil {
  33. return fmt.Errorf("fail to dial: %v", err)
  34. }
  35. defer grpcConection.Close()
  36. client := master_pb.NewSeaweedClient(grpcConection)
  37. stream, err := client.SendHeartbeat(context.Background())
  38. if err != nil {
  39. glog.V(0).Infof("%v.SendHeartbeat(_) = _, %v", client, err)
  40. return err
  41. }
  42. vs.SetMasterNode(masterNode)
  43. glog.V(0).Infof("Heartbeat to %s", masterNode)
  44. vs.store.Client = stream
  45. defer func() { vs.store.Client = nil }()
  46. doneChan := make(chan error, 1)
  47. go func() {
  48. for {
  49. in, err := stream.Recv()
  50. if err != nil {
  51. doneChan <- err
  52. return
  53. }
  54. if in.GetVolumeSizeLimit() != 0 {
  55. vs.store.VolumeSizeLimit = in.GetVolumeSizeLimit()
  56. }
  57. if in.GetSecretKey() != "" {
  58. vs.guard.SecretKey = security.Secret(in.GetSecretKey())
  59. }
  60. if in.GetLeader() != "" && masterNode != in.GetLeader() {
  61. vs.masterNodes.SetPossibleLeader(in.GetLeader())
  62. doneChan <- nil
  63. return
  64. }
  65. }
  66. }()
  67. if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
  68. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
  69. return err
  70. }
  71. tickChan := time.Tick(sleepInterval)
  72. for {
  73. select {
  74. case <-tickChan:
  75. if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
  76. glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
  77. return err
  78. }
  79. case err := <-doneChan:
  80. glog.V(0).Infof("Volume Server heart beat stops with %v", err)
  81. return err
  82. }
  83. }
  84. }