aboutsummaryrefslogtreecommitdiff
path: root/weed/cluster
diff options
context:
space:
mode:
Diffstat (limited to 'weed/cluster')
-rw-r--r--weed/cluster/cluster.go286
-rw-r--r--weed/cluster/cluster_test.go22
-rw-r--r--weed/cluster/master_client.go34
3 files changed, 207 insertions, 135 deletions
diff --git a/weed/cluster/cluster.go b/weed/cluster/cluster.go
index ad6e6b879..c320065d5 100644
--- a/weed/cluster/cluster.go
+++ b/weed/cluster/cluster.go
@@ -15,88 +15,156 @@ const (
BrokerType = "broker"
)
-type FilerGroup string
-type Filers struct {
- filers map[pb.ServerAddress]*ClusterNode
- leaders *Leaders
-}
+type FilerGroupName string
+type DataCenter string
+type Rack string
+
type Leaders struct {
leaders [3]pb.ServerAddress
}
-
type ClusterNode struct {
- Address pb.ServerAddress
- Version string
- counter int
- CreatedTs time.Time
+ Address pb.ServerAddress
+ Version string
+ counter int
+ CreatedTs time.Time
+ DataCenter DataCenter
+ Rack Rack
+}
+type GroupMembers struct {
+ members map[pb.ServerAddress]*ClusterNode
+ leaders *Leaders
+}
+type ClusterNodeGroups struct {
+ groupMembers map[FilerGroupName]*GroupMembers
+ sync.RWMutex
}
-
type Cluster struct {
- filerGroup2filers map[FilerGroup]*Filers
- filersLock sync.RWMutex
- brokers map[pb.ServerAddress]*ClusterNode
- brokersLock sync.RWMutex
+ filerGroups *ClusterNodeGroups
+ brokerGroups *ClusterNodeGroups
}
-func NewCluster() *Cluster {
- return &Cluster{
- filerGroup2filers: make(map[FilerGroup]*Filers),
- brokers: make(map[pb.ServerAddress]*ClusterNode),
+func newClusterNodeGroups() *ClusterNodeGroups {
+ return &ClusterNodeGroups{
+ groupMembers: map[FilerGroupName]*GroupMembers{},
}
}
-
-func (cluster *Cluster) getFilers(filerGroup FilerGroup, createIfNotFound bool) *Filers {
- filers, found := cluster.filerGroup2filers[filerGroup]
+func (g *ClusterNodeGroups) getGroupMembers(filerGroup FilerGroupName, createIfNotFound bool) *GroupMembers {
+ members, found := g.groupMembers[filerGroup]
if !found && createIfNotFound {
- filers = &Filers{
- filers: make(map[pb.ServerAddress]*ClusterNode),
+ members = &GroupMembers{
+ members: make(map[pb.ServerAddress]*ClusterNode),
leaders: &Leaders{},
}
- cluster.filerGroup2filers[filerGroup] = filers
+ g.groupMembers[filerGroup] = members
+ }
+ return members
+}
+
+func (m *GroupMembers) addMember(dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string) *ClusterNode {
+ if existingNode, found := m.members[address]; found {
+ existingNode.counter++
+ return nil
+ }
+ t := &ClusterNode{
+ Address: address,
+ Version: version,
+ counter: 1,
+ CreatedTs: time.Now(),
+ DataCenter: dataCenter,
+ Rack: rack,
+ }
+ m.members[address] = t
+ return t
+}
+func (m *GroupMembers) removeMember(address pb.ServerAddress) bool {
+ if existingNode, found := m.members[address]; !found {
+ return false
+ } else {
+ existingNode.counter--
+ if existingNode.counter <= 0 {
+ delete(m.members, address)
+ return true
+ }
+ }
+ return false
+}
+
+func (g *ClusterNodeGroups) AddClusterNode(filerGroup FilerGroupName, nodeType string, dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string) []*master_pb.KeepConnectedResponse {
+ g.Lock()
+ defer g.Unlock()
+ m := g.getGroupMembers(filerGroup, true)
+ if t := m.addMember(dataCenter, rack, address, version); t != nil {
+ return ensureGroupLeaders(m, true, filerGroup, nodeType, address)
+ }
+ return nil
+}
+func (g *ClusterNodeGroups) RemoveClusterNode(filerGroup FilerGroupName, nodeType string, address pb.ServerAddress) []*master_pb.KeepConnectedResponse {
+ g.Lock()
+ defer g.Unlock()
+ m := g.getGroupMembers(filerGroup, false)
+ if m == nil {
+ return nil
+ }
+ if m.removeMember(address) {
+ return ensureGroupLeaders(m, false, filerGroup, nodeType, address)
}
- return filers
+ return nil
+}
+func (g *ClusterNodeGroups) ListClusterNode(filerGroup FilerGroupName) (nodes []*ClusterNode) {
+ g.Lock()
+ defer g.Unlock()
+ m := g.getGroupMembers(filerGroup, false)
+ if m == nil {
+ return nil
+ }
+ for _, node := range m.members {
+ nodes = append(nodes, node)
+ }
+ return
+}
+func (g *ClusterNodeGroups) IsOneLeader(filerGroup FilerGroupName, address pb.ServerAddress) bool {
+ g.Lock()
+ defer g.Unlock()
+ m := g.getGroupMembers(filerGroup, false)
+ if m == nil {
+ return false
+ }
+ return m.leaders.isOneLeader(address)
+}
+func (g *ClusterNodeGroups) ListClusterNodeLeaders(filerGroup FilerGroupName) (nodes []pb.ServerAddress) {
+ g.Lock()
+ defer g.Unlock()
+ m := g.getGroupMembers(filerGroup, false)
+ if m == nil {
+ return nil
+ }
+ return m.leaders.GetLeaders()
}
-func (cluster *Cluster) AddClusterNode(ns, nodeType string, address pb.ServerAddress, version string) []*master_pb.KeepConnectedResponse {
- filerGroup := FilerGroup(ns)
+func NewCluster() *Cluster {
+ return &Cluster{
+ filerGroups: newClusterNodeGroups(),
+ brokerGroups: newClusterNodeGroups(),
+ }
+}
+
+func (cluster *Cluster) getGroupMembers(filerGroup FilerGroupName, nodeType string, createIfNotFound bool) *GroupMembers {
switch nodeType {
case FilerType:
- cluster.filersLock.Lock()
- defer cluster.filersLock.Unlock()
- filers := cluster.getFilers(filerGroup, true)
- if existingNode, found := filers.filers[address]; found {
- existingNode.counter++
- return nil
- }
- filers.filers[address] = &ClusterNode{
- Address: address,
- Version: version,
- counter: 1,
- CreatedTs: time.Now(),
- }
- return cluster.ensureFilerLeaders(filers, true, filerGroup, nodeType, address)
+ return cluster.filerGroups.getGroupMembers(filerGroup, createIfNotFound)
case BrokerType:
- cluster.brokersLock.Lock()
- defer cluster.brokersLock.Unlock()
- if existingNode, found := cluster.brokers[address]; found {
- existingNode.counter++
- return nil
- }
- cluster.brokers[address] = &ClusterNode{
- Address: address,
- Version: version,
- counter: 1,
- CreatedTs: time.Now(),
- }
- return []*master_pb.KeepConnectedResponse{
- {
- ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
- NodeType: nodeType,
- Address: string(address),
- IsAdd: true,
- },
- },
- }
+ return cluster.brokerGroups.getGroupMembers(filerGroup, createIfNotFound)
+ }
+ return nil
+}
+
+func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string) []*master_pb.KeepConnectedResponse {
+ filerGroup := FilerGroupName(ns)
+ switch nodeType {
+ case FilerType:
+ return cluster.filerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version)
+ case BrokerType:
+ return cluster.brokerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version)
case MasterType:
return []*master_pb.KeepConnectedResponse{
{
@@ -112,44 +180,12 @@ func (cluster *Cluster) AddClusterNode(ns, nodeType string, address pb.ServerAdd
}
func (cluster *Cluster) RemoveClusterNode(ns string, nodeType string, address pb.ServerAddress) []*master_pb.KeepConnectedResponse {
- filerGroup := FilerGroup(ns)
+ filerGroup := FilerGroupName(ns)
switch nodeType {
case FilerType:
- cluster.filersLock.Lock()
- defer cluster.filersLock.Unlock()
- filers := cluster.getFilers(filerGroup, false)
- if filers == nil {
- return nil
- }
- if existingNode, found := filers.filers[address]; !found {
- return nil
- } else {
- existingNode.counter--
- if existingNode.counter <= 0 {
- delete(filers.filers, address)
- return cluster.ensureFilerLeaders(filers, false, filerGroup, nodeType, address)
- }
- }
+ return cluster.filerGroups.RemoveClusterNode(filerGroup, nodeType, address)
case BrokerType:
- cluster.brokersLock.Lock()
- defer cluster.brokersLock.Unlock()
- if existingNode, found := cluster.brokers[address]; !found {
- return nil
- } else {
- existingNode.counter--
- if existingNode.counter <= 0 {
- delete(cluster.brokers, address)
- return []*master_pb.KeepConnectedResponse{
- {
- ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
- NodeType: nodeType,
- Address: string(address),
- IsAdd: false,
- },
- },
- }
- }
- }
+ return cluster.brokerGroups.RemoveClusterNode(filerGroup, nodeType, address)
case MasterType:
return []*master_pb.KeepConnectedResponse{
{
@@ -164,40 +200,42 @@ func (cluster *Cluster) RemoveClusterNode(ns string, nodeType string, address pb
return nil
}
-func (cluster *Cluster) ListClusterNode(filerGroup FilerGroup, nodeType string) (nodes []*ClusterNode) {
+func (cluster *Cluster) ListClusterNode(filerGroup FilerGroupName, nodeType string) (nodes []*ClusterNode) {
switch nodeType {
case FilerType:
- cluster.filersLock.RLock()
- defer cluster.filersLock.RUnlock()
- filers := cluster.getFilers(filerGroup, false)
- if filers == nil {
- return
- }
- for _, node := range filers.filers {
- nodes = append(nodes, node)
- }
+ return cluster.filerGroups.ListClusterNode(filerGroup)
case BrokerType:
- cluster.brokersLock.RLock()
- defer cluster.brokersLock.RUnlock()
- for _, node := range cluster.brokers {
- nodes = append(nodes, node)
- }
+ return cluster.brokerGroups.ListClusterNode(filerGroup)
case MasterType:
}
return
}
-func (cluster *Cluster) IsOneLeader(filerGroup FilerGroup, address pb.ServerAddress) bool {
- filers := cluster.getFilers(filerGroup, false)
- if filers == nil {
- return false
+func (cluster *Cluster) ListClusterNodeLeaders(filerGroup FilerGroupName, nodeType string) (nodes []pb.ServerAddress) {
+ switch nodeType {
+ case FilerType:
+ return cluster.filerGroups.ListClusterNodeLeaders(filerGroup)
+ case BrokerType:
+ return cluster.brokerGroups.ListClusterNodeLeaders(filerGroup)
+ case MasterType:
}
- return filers.leaders.isOneLeader(address)
+ return
+}
+
+func (cluster *Cluster) IsOneLeader(filerGroup FilerGroupName, nodeType string, address pb.ServerAddress) bool {
+ switch nodeType {
+ case FilerType:
+ return cluster.filerGroups.IsOneLeader(filerGroup, address)
+ case BrokerType:
+ return cluster.brokerGroups.IsOneLeader(filerGroup, address)
+ case MasterType:
+ }
+ return false
}
-func (cluster *Cluster) ensureFilerLeaders(filers *Filers, isAdd bool, filerGroup FilerGroup, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) {
+func ensureGroupLeaders(m *GroupMembers, isAdd bool, filerGroup FilerGroupName, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) {
if isAdd {
- if filers.leaders.addLeaderIfVacant(address) {
+ if m.leaders.addLeaderIfVacant(address) {
// has added the address as one leader
result = append(result, &master_pb.KeepConnectedResponse{
ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
@@ -220,7 +258,7 @@ func (cluster *Cluster) ensureFilerLeaders(filers *Filers, isAdd bool, filerGrou
})
}
} else {
- if filers.leaders.removeLeaderIfExists(address) {
+ if m.leaders.removeLeaderIfExists(address) {
result = append(result, &master_pb.KeepConnectedResponse{
ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
@@ -236,8 +274,8 @@ func (cluster *Cluster) ensureFilerLeaders(filers *Filers, isAdd bool, filerGrou
var shortestDuration int64 = math.MaxInt64
now := time.Now()
var candidateAddress pb.ServerAddress
- for _, node := range filers.filers {
- if filers.leaders.isOneLeader(node.Address) {
+ for _, node := range m.members {
+ if m.leaders.isOneLeader(node.Address) {
continue
}
duration := now.Sub(node.CreatedTs).Nanoseconds()
@@ -247,7 +285,7 @@ func (cluster *Cluster) ensureFilerLeaders(filers *Filers, isAdd bool, filerGrou
}
}
if candidateAddress != "" {
- filers.leaders.addLeaderIfVacant(candidateAddress)
+ m.leaders.addLeaderIfVacant(candidateAddress)
// added a new leader
result = append(result, &master_pb.KeepConnectedResponse{
ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
diff --git a/weed/cluster/cluster_test.go b/weed/cluster/cluster_test.go
index 1187642de..57ee4a023 100644
--- a/weed/cluster/cluster_test.go
+++ b/weed/cluster/cluster_test.go
@@ -11,29 +11,29 @@ import (
func TestClusterAddRemoveNodes(t *testing.T) {
c := NewCluster()
- c.AddClusterNode("", "filer", pb.ServerAddress("111:1"), "23.45")
- c.AddClusterNode("", "filer", pb.ServerAddress("111:2"), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress("111:1"), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress("111:2"), "23.45")
assert.Equal(t, []pb.ServerAddress{
pb.ServerAddress("111:1"),
pb.ServerAddress("111:2"),
- }, c.getFilers("", false).leaders.GetLeaders())
+ }, c.getGroupMembers("", "filer", true).leaders.GetLeaders())
- c.AddClusterNode("", "filer", pb.ServerAddress("111:3"), "23.45")
- c.AddClusterNode("", "filer", pb.ServerAddress("111:4"), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress("111:3"), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress("111:4"), "23.45")
assert.Equal(t, []pb.ServerAddress{
pb.ServerAddress("111:1"),
pb.ServerAddress("111:2"),
pb.ServerAddress("111:3"),
- }, c.getFilers("", false).leaders.GetLeaders())
+ }, c.getGroupMembers("", "filer", true).leaders.GetLeaders())
- c.AddClusterNode("", "filer", pb.ServerAddress("111:5"), "23.45")
- c.AddClusterNode("", "filer", pb.ServerAddress("111:6"), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress("111:5"), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress("111:6"), "23.45")
c.RemoveClusterNode("", "filer", pb.ServerAddress("111:4"))
assert.Equal(t, []pb.ServerAddress{
pb.ServerAddress("111:1"),
pb.ServerAddress("111:2"),
pb.ServerAddress("111:3"),
- }, c.getFilers("", false).leaders.GetLeaders())
+ }, c.getGroupMembers("", "filer", true).leaders.GetLeaders())
// remove oldest
c.RemoveClusterNode("", "filer", pb.ServerAddress("111:1"))
@@ -41,7 +41,7 @@ func TestClusterAddRemoveNodes(t *testing.T) {
pb.ServerAddress("111:6"),
pb.ServerAddress("111:2"),
pb.ServerAddress("111:3"),
- }, c.getFilers("", false).leaders.GetLeaders())
+ }, c.getGroupMembers("", "filer", true).leaders.GetLeaders())
// remove oldest
c.RemoveClusterNode("", "filer", pb.ServerAddress("111:1"))
@@ -56,7 +56,7 @@ func TestConcurrentAddRemoveNodes(t *testing.T) {
go func(i int) {
defer wg.Done()
address := strconv.Itoa(i)
- c.AddClusterNode("", "filer", pb.ServerAddress(address), "23.45")
+ c.AddClusterNode("", "filer", "", "", pb.ServerAddress(address), "23.45")
}(i)
}
wg.Wait()
diff --git a/weed/cluster/master_client.go b/weed/cluster/master_client.go
new file mode 100644
index 000000000..b9b1673f3
--- /dev/null
+++ b/weed/cluster/master_client.go
@@ -0,0 +1,34 @@
+package cluster
+
+import (
+ "context"
+ "github.com/chrislusf/seaweedfs/weed/glog"
+ "github.com/chrislusf/seaweedfs/weed/pb"
+ "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+ "google.golang.org/grpc"
+)
+
+func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate) {
+
+ if grpcErr := pb.WithMasterClient(false, master, grpcDialOption, func(client master_pb.SeaweedClient) error {
+ resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+ ClientType: clientType,
+ FilerGroup: filerGroup,
+ })
+
+ glog.V(0).Infof("the cluster has %d %s\n", len(resp.ClusterNodes), clientType)
+ for _, node := range resp.ClusterNodes {
+ existingNodes = append(existingNodes, &master_pb.ClusterNodeUpdate{
+ NodeType: FilerType,
+ Address: node.Address,
+ IsLeader: node.IsLeader,
+ IsAdd: true,
+ CreatedAtNs: node.CreatedAtNs,
+ })
+ }
+ return err
+ }); grpcErr != nil {
+ glog.V(0).Infof("connect to %s: %v", master, grpcErr)
+ }
+ return
+}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 21:04:32 +0000