lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-11-24 16:53:14 +08:00
Code Issues Actions 17 Packages Projects Releases Wiki Activity

auto updated filer peer list

Browse Source
This commit is contained in:
Chris Lu
2021-11-06 14:23:35 -07:00
parent 84bb8e7365
commit e0fc2898e9
8 changed files with 211 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
weed/filer/filer.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"os"
"strings"
"time"
@@ -66,22 +67,38 @@ func NewFiler(masters []pb.ServerAddress, grpcDialOption grpc.DialOption,
return f
}
func (f *Filer) AggregateFromPeers(self pb.ServerAddress, filers []pb.ServerAddress) {
func (f *Filer) AggregateFromPeers(self pb.ServerAddress) {
// set peers
found := false
for _, peer := range filers {
if peer == self {
found = true
f.MetaAggregator = NewMetaAggregator(f, self, f.GrpcDialOption)
f.MasterClient.OnPeerUpdate = f.MetaAggregator.OnPeerUpdate
for _, peerUpdate := range f.ListExistingPeerUpdates() {
f.MetaAggregator.OnPeerUpdate(peerUpdate)
}
}
func (f *Filer) ListExistingPeerUpdates() (existingNodes []*master_pb.ClusterNodeUpdate){
if grpcErr := pb.WithMasterClient(f.MasterClient.GetMaster(), f.GrpcDialOption, func(client master_pb.SeaweedClient) error {
resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
ClientType: "filer",
})
glog.V(0).Infof("the cluster has %d filers\n", len(resp.ClusterNodes))
for _, node := range resp.ClusterNodes {
existingNodes = append(existingNodes, &master_pb.ClusterNodeUpdate{
NodeType: "filer",
Address: node.Address,
IsLeader: node.IsLeader,
IsAdd: true,
})
}
return err
}); grpcErr != nil {
glog.V(0).Infof("connect to %s: %v", f.MasterClient.GetMaster(), grpcErr)
}
if !found {
filers = append(filers, self)
}
f.MetaAggregator = NewMetaAggregator(filers, f.GrpcDialOption)
f.MetaAggregator.StartLoopSubscribe(f, self)
return
}
func (f *Filer) SetStore(store FilerStore) {
@@ -117,7 +134,7 @@ func (fs *Filer) GetMaster() pb.ServerAddress {
return fs.MasterClient.GetMaster()
}
func (fs *Filer) KeepConnectedToMaster() {
func (fs *Filer) KeepMasterClientConnected() {
fs.MasterClient.KeepConnectedToMaster()
}