123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899 |
- package weed_server
- import (
- "fmt"
- "time"
- "github.com/chrislusf/seaweedfs/weed/glog"
- "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
- "github.com/chrislusf/seaweedfs/weed/security"
- "github.com/chrislusf/seaweedfs/weed/storage"
- "golang.org/x/net/context"
- "google.golang.org/grpc"
- )
- func (vs *VolumeServer) heartbeat() {
- glog.V(0).Infof("Volume server bootstraps with master %s", vs.GetMasterNode())
- vs.masterNodes = storage.NewMasterNodes(vs.masterNode)
- vs.store.SetDataCenter(vs.dataCenter)
- vs.store.SetRack(vs.rack)
- for {
- err := vs.doHeartbeat(time.Duration(vs.pulseSeconds) * time.Second)
- if err != nil {
- glog.V(0).Infof("heartbeat error: %v", err)
- time.Sleep(time.Duration(vs.pulseSeconds) * time.Second)
- }
- }
- }
- func (vs *VolumeServer) doHeartbeat(sleepInterval time.Duration) error {
- vs.masterNodes.Reset()
- masterNode, err := vs.masterNodes.FindMaster()
- if err != nil {
- return fmt.Errorf("No master found: %v", err)
- }
- grpcConection, err := grpc.Dial(masterNode, grpc.WithInsecure())
- if err != nil {
- return fmt.Errorf("fail to dial: %v", err)
- }
- defer grpcConection.Close()
- client := master_pb.NewSeaweedClient(grpcConection)
- stream, err := client.SendHeartbeat(context.Background())
- if err != nil {
- glog.V(0).Infof("%v.SendHeartbeat(_) = _, %v", client, err)
- return err
- }
- vs.SetMasterNode(masterNode)
- glog.V(0).Infof("Heartbeat to %s", masterNode)
- vs.store.Client = stream
- defer func() { vs.store.Client = nil }()
- doneChan := make(chan error, 1)
- go func() {
- for {
- in, err := stream.Recv()
- if err != nil {
- doneChan <- err
- return
- }
- if in.GetVolumeSizeLimit() != 0 {
- vs.store.VolumeSizeLimit = in.GetVolumeSizeLimit()
- }
- if in.GetSecretKey() != "" {
- vs.guard.SecretKey = security.Secret(in.GetSecretKey())
- }
- if in.GetLeader() != "" && masterNode != in.GetLeader() {
- vs.masterNodes.SetPossibleLeader(in.GetLeader())
- doneChan <- nil
- return
- }
- }
- }()
- if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
- glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
- return err
- }
- tickChan := time.Tick(sleepInterval)
- for {
- select {
- case <-tickChan:
- if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
- glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
- return err
- }
- case err := <-doneChan:
- glog.V(0).Infof("Volume Server heart beat stops with %v", err)
- return err
- }
- }
- }
|