seaweedfs/weed/server/volume_grpc_client.go

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
		}
	}
}