seaweedfs/weed/server/volume_grpc_client_to_master.go

133 lines
3.9 KiB
Go
Raw Normal View History

2017-01-10 09:01:12 +00:00
package weed_server
import (
"fmt"
2019-04-15 06:00:37 +00:00
"time"
2019-02-18 20:11:52 +00:00
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/spf13/viper"
"google.golang.org/grpc"
2017-01-10 09:01:12 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
2018-05-10 06:11:54 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
2018-07-04 02:07:55 +00:00
"github.com/chrislusf/seaweedfs/weed/util"
2018-07-22 00:39:10 +00:00
"golang.org/x/net/context"
2017-01-10 09:01:12 +00:00
)
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)
2019-02-18 20:11:52 +00:00
grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "volume")
var err error
var newLeader string
2017-01-10 09:01:12 +00:00
for {
for _, master := range vs.MasterNodes {
if newLeader != "" {
master = newLeader
}
masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master)
if parseErr != nil {
2019-04-16 08:15:30 +00:00
glog.V(0).Infof("failed to parse master grpc %v: %v", masterGrpcAddress, parseErr)
continue
}
2019-03-16 00:20:24 +00:00
newLeader, err = vs.doHeartbeat(context.Background(), master, masterGrpcAddress, grpcDialOption, 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)
2019-04-15 16:09:46 +00:00
newLeader = ""
}
2017-01-10 09:01:12 +00:00
}
}
}
2019-03-16 00:20:24 +00:00
func (vs *VolumeServer) doHeartbeat(ctx context.Context, masterNode, masterGrpcAddress string, grpcDialOption grpc.DialOption, sleepInterval time.Duration) (newLeader string, err error) {
2017-01-10 09:01:12 +00:00
2019-03-16 00:20:24 +00:00
grpcConection, err := util.GrpcDial(ctx, masterGrpcAddress, grpcDialOption)
2017-01-10 09:01:12 +00:00
if err != nil {
2019-01-11 10:23:31 +00:00
return "", fmt.Errorf("fail to dial %s : %v", masterNode, 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)
2019-03-16 00:20:24 +00:00
stream, err := client.SendHeartbeat(ctx)
2017-01-10 09:01:12 +00:00
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 {
2019-04-15 06:00:37 +00:00
vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit())
}
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 {
2019-04-20 18:35:20 +00:00
case volumeMessage := <-vs.store.NewVolumesChan:
deltaBeat := &master_pb.Heartbeat{
2019-04-20 18:35:20 +00:00
NewVolumes:[]*master_pb.VolumeShortInformationMessage{
&volumeMessage,
},
}
2019-04-20 18:35:20 +00:00
glog.V(1).Infof("volume server %s:%d adds volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
if err = stream.Send(deltaBeat); err != nil {
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
return "", err
}
2019-04-20 18:35:20 +00:00
case volumeMessage := <-vs.store.DeletedVolumesChan:
deltaBeat := &master_pb.Heartbeat{
2019-04-20 18:35:20 +00:00
DeletedVolumes:[]*master_pb.VolumeShortInformationMessage{
&volumeMessage,
},
}
2019-04-20 18:35:20 +00:00
glog.V(1).Infof("volume server %s:%d deletes volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
if err = stream.Send(deltaBeat); err != nil {
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
return "", err
}
case <-tickChan:
2019-04-20 18:35:20 +00:00
glog.V(1).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
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:
return
2017-01-10 09:01:12 +00:00
}
}
}