master_grpc_server_cluster.go 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. package weed_server
  2. import (
  3. "context"
  4. "github.com/seaweedfs/seaweedfs/weed/cluster"
  5. "github.com/seaweedfs/seaweedfs/weed/pb"
  6. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  7. "math/rand/v2"
  8. )
  9. func (ms *MasterServer) ListClusterNodes(ctx context.Context, req *master_pb.ListClusterNodesRequest) (*master_pb.ListClusterNodesResponse, error) {
  10. resp := &master_pb.ListClusterNodesResponse{}
  11. filerGroup := cluster.FilerGroupName(req.FilerGroup)
  12. clusterNodes := ms.Cluster.ListClusterNode(filerGroup, req.ClientType)
  13. clusterNodes = limitTo(clusterNodes, req.Limit)
  14. for _, node := range clusterNodes {
  15. resp.ClusterNodes = append(resp.ClusterNodes, &master_pb.ListClusterNodesResponse_ClusterNode{
  16. Address: string(node.Address),
  17. Version: node.Version,
  18. CreatedAtNs: node.CreatedTs.UnixNano(),
  19. DataCenter: string(node.DataCenter),
  20. Rack: string(node.Rack),
  21. })
  22. }
  23. return resp, nil
  24. }
  25. func (ms *MasterServer) GetOneFiler(filerGroup cluster.FilerGroupName) pb.ServerAddress {
  26. filers := ms.Cluster.ListClusterNode(filerGroup, cluster.FilerType)
  27. if len(filers) > 0 {
  28. return filers[rand.IntN(len(filers))].Address
  29. }
  30. return "localhost:8888"
  31. }
  32. func limitTo(nodes []*cluster.ClusterNode, limit int32) (selected []*cluster.ClusterNode) {
  33. if limit <= 0 || len(nodes) < int(limit) {
  34. return nodes
  35. }
  36. selectedSet := make(map[pb.ServerAddress]*cluster.ClusterNode)
  37. for i := 0; i < int(limit)*3; i++ {
  38. x := rand.IntN(len(nodes))
  39. if _, found := selectedSet[nodes[x].Address]; found {
  40. continue
  41. }
  42. selectedSet[nodes[x].Address] = nodes[x]
  43. }
  44. for _, node := range selectedSet {
  45. selected = append(selected, node)
  46. }
  47. return
  48. }