2017-01-10 09:01:12 +00:00
|
|
|
package weed_server
|
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
2022-05-16 02:41:18 +00:00
|
|
|
"os"
|
2019-04-15 06:00:37 +00:00
|
|
|
"time"
|
|
|
|
|
2022-07-29 07:17:28 +00:00
|
|
|
"github.com/seaweedfs/seaweedfs/weed/operation"
|
2022-05-16 02:41:18 +00:00
|
|
|
|
2020-01-29 17:09:55 +00:00
|
|
|
"google.golang.org/grpc"
|
|
|
|
|
2022-07-29 07:17:28 +00:00
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/security"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/backend"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
2020-01-29 17:09:55 +00:00
|
|
|
|
|
|
|
"golang.org/x/net/context"
|
2017-01-10 09:01:12 +00:00
|
|
|
|
2022-07-29 07:17:28 +00:00
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
2017-01-10 09:01:12 +00:00
|
|
|
)
|
|
|
|
|
2021-09-13 05:47:52 +00:00
|
|
|
func (vs *VolumeServer) GetMaster() pb.ServerAddress {
|
2018-06-01 07:39:39 +00:00
|
|
|
return vs.currentMaster
|
|
|
|
}
|
2020-09-16 08:27:05 +00:00
|
|
|
|
|
|
|
func (vs *VolumeServer) checkWithMaster() (err error) {
|
2021-09-11 09:06:30 +00:00
|
|
|
for {
|
2020-09-19 07:03:00 +00:00
|
|
|
for _, master := range vs.SeedMasterNodes {
|
2021-12-26 08:15:03 +00:00
|
|
|
err = operation.WithMasterServerClient(false, master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
|
2020-09-19 07:03:00 +00:00
|
|
|
resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("get master %s configuration: %v", master, err)
|
|
|
|
}
|
|
|
|
vs.metricsAddress, vs.metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSeconds)
|
|
|
|
backend.LoadFromPbStorageBackends(resp.StorageBackends)
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err == nil {
|
|
|
|
return
|
|
|
|
} else {
|
|
|
|
glog.V(0).Infof("checkWithMaster %s: %v", master, err)
|
2020-09-16 08:27:05 +00:00
|
|
|
}
|
|
|
|
}
|
2020-09-19 07:03:00 +00:00
|
|
|
time.Sleep(1790 * time.Millisecond)
|
2020-09-16 08:27:05 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-01-10 09:01:12 +00:00
|
|
|
func (vs *VolumeServer) heartbeat() {
|
|
|
|
|
2019-05-28 04:22:23 +00:00
|
|
|
glog.V(0).Infof("Volume server start with seed master nodes: %v", vs.SeedMasterNodes)
|
2017-01-10 09:01:12 +00:00
|
|
|
vs.store.SetDataCenter(vs.dataCenter)
|
|
|
|
vs.store.SetRack(vs.rack)
|
|
|
|
|
2020-01-29 17:09:55 +00:00
|
|
|
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume")
|
2019-02-18 20:11:52 +00:00
|
|
|
|
2018-06-01 07:39:39 +00:00
|
|
|
var err error
|
2021-09-13 05:47:52 +00:00
|
|
|
var newLeader pb.ServerAddress
|
2020-09-14 04:25:51 +00:00
|
|
|
for vs.isHeartbeating {
|
2019-05-28 04:22:23 +00:00
|
|
|
for _, master := range vs.SeedMasterNodes {
|
2018-06-01 07:39:39 +00:00
|
|
|
if newLeader != "" {
|
2020-04-29 00:29:10 +00:00
|
|
|
// the new leader may actually is the same master
|
|
|
|
// need to wait a bit before adding itself
|
|
|
|
time.Sleep(3 * time.Second)
|
2018-06-01 07:39:39 +00:00
|
|
|
master = newLeader
|
|
|
|
}
|
2019-05-28 05:54:58 +00:00
|
|
|
vs.store.MasterAddress = master
|
2021-09-13 05:47:52 +00:00
|
|
|
newLeader, err = vs.doHeartbeat(master, grpcDialOption, time.Duration(vs.pulseSeconds)*time.Second)
|
2018-06-01 07:39:39 +00:00
|
|
|
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 = ""
|
2019-05-28 05:54:58 +00:00
|
|
|
vs.store.MasterAddress = ""
|
2018-06-01 07:39:39 +00:00
|
|
|
}
|
2020-09-14 04:25:51 +00:00
|
|
|
if !vs.isHeartbeating {
|
|
|
|
break
|
|
|
|
}
|
2017-01-10 09:01:12 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-09-14 04:25:51 +00:00
|
|
|
func (vs *VolumeServer) StopHeartbeat() (isAlreadyStopping bool) {
|
|
|
|
if !vs.isHeartbeating {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
vs.isHeartbeating = false
|
2020-10-08 06:48:24 +00:00
|
|
|
close(vs.stopChan)
|
2020-09-14 04:25:51 +00:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2021-09-13 05:47:52 +00:00
|
|
|
func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOption grpc.DialOption, sleepInterval time.Duration) (newLeader pb.ServerAddress, err error) {
|
2017-01-10 09:01:12 +00:00
|
|
|
|
2020-09-09 19:07:15 +00:00
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
|
|
defer cancel()
|
|
|
|
|
2022-09-15 09:04:29 +00:00
|
|
|
grpcConnection, err := pb.GrpcDial(ctx, masterAddress.ToGrpcAddress(), false, grpcDialOption)
|
2017-01-10 09:01:12 +00:00
|
|
|
if err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
return "", fmt.Errorf("fail to dial %s : %v", masterAddress, err)
|
2017-01-10 09:01:12 +00:00
|
|
|
}
|
2022-09-15 09:04:29 +00:00
|
|
|
defer grpcConnection.Close()
|
2017-01-10 09:01:12 +00:00
|
|
|
|
2022-09-15 09:04:29 +00:00
|
|
|
client := master_pb.NewSeaweedClient(grpcConnection)
|
2020-09-09 19:07:15 +00:00
|
|
|
stream, err := client.SendHeartbeat(ctx)
|
2017-01-10 09:01:12 +00:00
|
|
|
if err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("SendHeartbeat to %s: %v", masterAddress, err)
|
2018-06-01 07:39:39 +00:00
|
|
|
return "", err
|
2017-01-10 09:01:12 +00:00
|
|
|
}
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Heartbeat to: %v", masterAddress)
|
|
|
|
vs.currentMaster = masterAddress
|
2017-01-10 09:01:12 +00:00
|
|
|
|
2017-01-10 09:30:00 +00:00
|
|
|
doneChan := make(chan error, 1)
|
|
|
|
|
2017-01-10 09:01:12 +00:00
|
|
|
go func() {
|
|
|
|
for {
|
|
|
|
in, err := stream.Recv()
|
|
|
|
if err != nil {
|
2017-01-10 09:30:00 +00:00
|
|
|
doneChan <- err
|
2017-01-10 09:01:12 +00:00
|
|
|
return
|
|
|
|
}
|
2022-05-17 06:51:01 +00:00
|
|
|
if len(in.DuplicatedUuids) > 0 {
|
2022-06-10 14:03:11 +00:00
|
|
|
var duplicateDir []string
|
2022-05-17 06:51:01 +00:00
|
|
|
for _, loc := range vs.store.Locations {
|
|
|
|
for _, uuid := range in.DuplicatedUuids {
|
2022-06-10 13:38:17 +00:00
|
|
|
if uuid == loc.DirectoryUuid {
|
2022-06-10 14:03:11 +00:00
|
|
|
duplicateDir = append(duplicateDir, loc.Directory)
|
2022-05-17 06:51:01 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2022-06-10 14:03:11 +00:00
|
|
|
glog.Errorf("Shut down Volume Server due to duplicate volume directories: %v", duplicateDir)
|
2022-06-10 13:38:17 +00:00
|
|
|
os.Exit(1)
|
2022-05-16 02:41:18 +00:00
|
|
|
}
|
2020-03-22 23:21:42 +00:00
|
|
|
if in.GetVolumeSizeLimit() != 0 && vs.store.GetVolumeSizeLimit() != in.GetVolumeSizeLimit() {
|
2019-04-15 06:00:37 +00:00
|
|
|
vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit())
|
2020-03-22 23:21:42 +00:00
|
|
|
if vs.store.MaybeAdjustVolumeMax() {
|
|
|
|
if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
|
2020-08-06 16:48:54 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", vs.currentMaster, err)
|
2021-09-11 09:06:43 +00:00
|
|
|
return
|
2020-03-22 23:21:42 +00:00
|
|
|
}
|
|
|
|
}
|
2017-01-23 05:36:36 +00:00
|
|
|
}
|
2021-09-13 05:47:52 +00:00
|
|
|
if in.GetLeader() != "" && string(vs.currentMaster) != in.GetLeader() {
|
2020-08-06 16:48:54 +00:00
|
|
|
glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), vs.currentMaster)
|
2021-09-13 05:47:52 +00:00
|
|
|
newLeader = pb.ServerAddress(in.GetLeader())
|
2017-01-18 17:34:27 +00:00
|
|
|
doneChan <- nil
|
|
|
|
return
|
|
|
|
}
|
2017-01-10 09:01:12 +00:00
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2017-01-10 09:30:00 +00:00
|
|
|
if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
|
2018-06-01 07:39:39 +00:00
|
|
|
return "", err
|
2017-01-10 09:30:00 +00:00
|
|
|
}
|
|
|
|
|
2019-05-22 05:41:20 +00:00
|
|
|
if err = stream.Send(vs.store.CollectErasureCodingHeartbeat()); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
|
2019-05-22 05:41:20 +00:00
|
|
|
return "", err
|
|
|
|
}
|
|
|
|
|
|
|
|
volumeTickChan := time.Tick(sleepInterval)
|
|
|
|
ecShardTickChan := time.Tick(17 * sleepInterval)
|
2022-09-12 05:31:53 +00:00
|
|
|
dataCenter := vs.store.GetDataCenter()
|
|
|
|
rack := vs.store.GetRack()
|
|
|
|
ip := vs.store.Ip
|
|
|
|
port := uint32(vs.store.Port)
|
2017-01-10 09:01:12 +00:00
|
|
|
for {
|
2017-01-10 09:30:00 +00:00
|
|
|
select {
|
2019-04-20 18:35:20 +00:00
|
|
|
case volumeMessage := <-vs.store.NewVolumesChan:
|
2018-08-24 08:26:56 +00:00
|
|
|
deltaBeat := &master_pb.Heartbeat{
|
2022-09-12 05:31:53 +00:00
|
|
|
Ip: ip,
|
|
|
|
Port: port,
|
|
|
|
DataCenter: dataCenter,
|
|
|
|
Rack: rack,
|
2019-04-30 03:22:19 +00:00
|
|
|
NewVolumes: []*master_pb.VolumeShortInformationMessage{
|
2019-04-20 18:35:20 +00:00
|
|
|
&volumeMessage,
|
|
|
|
},
|
2018-08-24 08:26:56 +00:00
|
|
|
}
|
2021-09-11 19:59:15 +00:00
|
|
|
glog.V(0).Infof("volume server %s:%d adds volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
|
2018-08-24 08:26:56 +00:00
|
|
|
if err = stream.Send(deltaBeat); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
|
2018-08-24 08:26:56 +00:00
|
|
|
return "", err
|
|
|
|
}
|
2019-05-22 05:41:20 +00:00
|
|
|
case ecShardMessage := <-vs.store.NewEcShardsChan:
|
|
|
|
deltaBeat := &master_pb.Heartbeat{
|
2022-09-12 05:31:53 +00:00
|
|
|
Ip: ip,
|
|
|
|
Port: port,
|
|
|
|
DataCenter: dataCenter,
|
|
|
|
Rack: rack,
|
2019-05-22 05:41:20 +00:00
|
|
|
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
|
|
|
&ecShardMessage,
|
|
|
|
},
|
|
|
|
}
|
2021-09-11 19:59:15 +00:00
|
|
|
glog.V(0).Infof("volume server %s:%d adds ec shard %d:%d", vs.store.Ip, vs.store.Port, ecShardMessage.Id,
|
2019-05-24 20:28:44 +00:00
|
|
|
erasure_coding.ShardBits(ecShardMessage.EcIndexBits).ShardIds())
|
2019-05-22 05:41:20 +00:00
|
|
|
if err = stream.Send(deltaBeat); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
|
2019-05-22 05:41:20 +00:00
|
|
|
return "", err
|
|
|
|
}
|
2019-04-20 18:35:20 +00:00
|
|
|
case volumeMessage := <-vs.store.DeletedVolumesChan:
|
2018-08-24 08:26:56 +00:00
|
|
|
deltaBeat := &master_pb.Heartbeat{
|
2022-09-12 05:31:53 +00:00
|
|
|
Ip: ip,
|
|
|
|
Port: port,
|
|
|
|
DataCenter: dataCenter,
|
|
|
|
Rack: rack,
|
2019-04-30 03:22:19 +00:00
|
|
|
DeletedVolumes: []*master_pb.VolumeShortInformationMessage{
|
2019-04-20 18:35:20 +00:00
|
|
|
&volumeMessage,
|
|
|
|
},
|
2018-08-24 08:26:56 +00:00
|
|
|
}
|
2021-09-11 19:59:15 +00:00
|
|
|
glog.V(0).Infof("volume server %s:%d deletes volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
|
2018-08-24 08:26:56 +00:00
|
|
|
if err = stream.Send(deltaBeat); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
|
2018-08-24 08:26:56 +00:00
|
|
|
return "", err
|
|
|
|
}
|
2019-05-22 05:41:20 +00:00
|
|
|
case ecShardMessage := <-vs.store.DeletedEcShardsChan:
|
|
|
|
deltaBeat := &master_pb.Heartbeat{
|
2022-09-12 05:31:53 +00:00
|
|
|
Ip: ip,
|
|
|
|
Port: port,
|
|
|
|
DataCenter: dataCenter,
|
|
|
|
Rack: rack,
|
2019-05-22 05:41:20 +00:00
|
|
|
DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
|
|
|
&ecShardMessage,
|
|
|
|
},
|
|
|
|
}
|
2021-09-11 19:59:15 +00:00
|
|
|
glog.V(0).Infof("volume server %s:%d deletes ec shard %d:%d", vs.store.Ip, vs.store.Port, ecShardMessage.Id,
|
2019-05-24 20:28:44 +00:00
|
|
|
erasure_coding.ShardBits(ecShardMessage.EcIndexBits).ShardIds())
|
2019-05-22 05:41:20 +00:00
|
|
|
if err = stream.Send(deltaBeat); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
|
2019-05-22 05:41:20 +00:00
|
|
|
return "", err
|
|
|
|
}
|
|
|
|
case <-volumeTickChan:
|
2020-09-14 04:25:51 +00:00
|
|
|
glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
|
2020-11-03 22:43:17 +00:00
|
|
|
vs.store.MaybeAdjustVolumeMax()
|
2020-09-14 04:25:51 +00:00
|
|
|
if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
|
2020-09-14 04:25:51 +00:00
|
|
|
return "", err
|
2017-01-10 09:30:00 +00:00
|
|
|
}
|
2019-05-22 05:41:20 +00:00
|
|
|
case <-ecShardTickChan:
|
2020-08-31 03:12:04 +00:00
|
|
|
glog.V(4).Infof("volume server %s:%d ec heartbeat", vs.store.Ip, vs.store.Port)
|
2019-05-22 05:41:20 +00:00
|
|
|
if err = stream.Send(vs.store.CollectErasureCodingHeartbeat()); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
|
2019-05-22 05:41:20 +00:00
|
|
|
return "", err
|
|
|
|
}
|
2019-01-07 08:09:15 +00:00
|
|
|
case err = <-doneChan:
|
2018-06-01 07:39:39 +00:00
|
|
|
return
|
2020-09-14 04:25:51 +00:00
|
|
|
case <-vs.stopChan:
|
2020-10-30 21:48:25 +00:00
|
|
|
var volumeMessages []*master_pb.VolumeInformationMessage
|
|
|
|
emptyBeat := &master_pb.Heartbeat{
|
2022-09-12 05:31:53 +00:00
|
|
|
Ip: ip,
|
|
|
|
Port: port,
|
2021-02-18 04:57:08 +00:00
|
|
|
PublicUrl: vs.store.PublicUrl,
|
|
|
|
MaxFileKey: uint64(0),
|
2022-09-12 05:31:53 +00:00
|
|
|
DataCenter: dataCenter,
|
|
|
|
Rack: rack,
|
2021-02-18 04:57:08 +00:00
|
|
|
Volumes: volumeMessages,
|
|
|
|
HasNoVolumes: len(volumeMessages) == 0,
|
2020-10-30 21:48:25 +00:00
|
|
|
}
|
2020-10-30 22:21:54 +00:00
|
|
|
glog.V(1).Infof("volume server %s:%d stops and deletes all volumes", vs.store.Ip, vs.store.Port)
|
2020-10-30 21:48:25 +00:00
|
|
|
if err = stream.Send(emptyBeat); err != nil {
|
2021-09-13 05:47:52 +00:00
|
|
|
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
|
2020-10-30 21:48:25 +00:00
|
|
|
return "", err
|
|
|
|
}
|
2020-09-14 04:25:51 +00:00
|
|
|
return
|
2017-01-10 09:01:12 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|