seaweedfs/weed/server/volume_grpc_client.go

102 lines
2.5 KiB
Go
Raw Normal View History

2017-01-10 09:01:12 +00:00
package weed_server
import (
"fmt"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
2018-05-10 06:11:54 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
2017-01-10 09:01:12 +00:00
"github.com/chrislusf/seaweedfs/weed/security"
"golang.org/x/net/context"
"google.golang.org/grpc"
)
func (vs *VolumeServer) GetMaster() string {
return vs.currentMaster
}
2017-01-10 09:01:12 +00:00
func (vs *VolumeServer) heartbeat() {
glog.V(0).Infof("Volume server start with masters: %v", vs.MasterNodes)
2017-01-10 09:01:12 +00:00
vs.store.SetDataCenter(vs.dataCenter)
vs.store.SetRack(vs.rack)
var err error
var newLeader string
2017-01-10 09:01:12 +00:00
for {
for _, master := range vs.MasterNodes {
if newLeader != "" {
master = newLeader
}
newLeader, err = vs.doHeartbeat(master, 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)
}
2017-01-10 09:01:12 +00:00
}
}
}
func (vs *VolumeServer) doHeartbeat(masterNode string, sleepInterval time.Duration) (newLeader string, err error) {
2017-01-10 09:01:12 +00:00
grpcConection, err := grpc.Dial(masterNode, grpc.WithInsecure())
if err != nil {
return "", fmt.Errorf("fail to dial: %v", err)
2017-01-10 09:01:12 +00:00
}
defer grpcConection.Close()
2018-05-10 06:11:54 +00:00
client := master_pb.NewSeaweedClient(grpcConection)
2017-01-10 09:01:12 +00:00
stream, err := client.SendHeartbeat(context.Background())
if err != nil {
glog.V(0).Infof("SendHeartbeat to %s: %v", masterNode, err)
return "", err
2017-01-10 09:01:12 +00:00
}
glog.V(0).Infof("Heartbeat to: %v", masterNode)
vs.currentMaster = masterNode
2017-01-10 09:01:12 +00:00
vs.store.Client = stream
defer func() { vs.store.Client = nil }()
doneChan := make(chan error, 1)
2017-01-10 09:01:12 +00:00
go func() {
for {
in, err := stream.Recv()
if err != nil {
doneChan <- err
2017-01-10 09:01:12 +00:00
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() {
glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), masterNode)
newLeader = in.GetLeader()
doneChan <- nil
return
}
2017-01-10 09:01:12 +00:00
}
}()
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)
2017-01-10 09:01:12 +00:00
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 <-doneChan:
return
2017-01-10 09:01:12 +00:00
}
}
}