You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

220 lines
6.5 KiB

  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/cluster"
  7. "github.com/chrislusf/seaweedfs/weed/pb"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  10. "io"
  11. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  12. )
  13. func init() {
  14. Commands = append(Commands, &commandClusterCheck{})
  15. }
  16. type commandClusterCheck struct {
  17. }
  18. func (c *commandClusterCheck) Name() string {
  19. return "cluster.check"
  20. }
  21. func (c *commandClusterCheck) Help() string {
  22. return `check current cluster network connectivity
  23. cluster.check
  24. `
  25. }
  26. func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  27. clusterPsCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  28. if err = clusterPsCommand.Parse(args); err != nil {
  29. return nil
  30. }
  31. // collect filers
  32. var filers []pb.ServerAddress
  33. err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
  34. resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
  35. ClientType: cluster.FilerType,
  36. })
  37. for _, node := range resp.ClusterNodes {
  38. filers = append(filers, pb.ServerAddress(node.Address))
  39. }
  40. return err
  41. })
  42. if err != nil {
  43. return
  44. }
  45. fmt.Fprintf(writer, "the cluster has %d filers: %+v\n", len(filers), filers)
  46. // collect volume servers
  47. var volumeServers []pb.ServerAddress
  48. t, _, err := collectTopologyInfo(commandEnv, 0)
  49. if err != nil {
  50. return err
  51. }
  52. for _, dc := range t.DataCenterInfos {
  53. for _, r := range dc.RackInfos {
  54. for _, dn := range r.DataNodeInfos {
  55. volumeServers = append(volumeServers, pb.NewServerAddressFromDataNode(dn))
  56. }
  57. }
  58. }
  59. fmt.Fprintf(writer, "the cluster has %d volume servers: %+v\n", len(volumeServers), volumeServers)
  60. // collect all masters
  61. var masters []pb.ServerAddress
  62. for _, master := range commandEnv.MasterClient.GetMasters() {
  63. masters = append(masters, master)
  64. }
  65. // check from master to volume servers
  66. for _, master := range masters {
  67. for _, volumeServer := range volumeServers {
  68. fmt.Fprintf(writer, "checking master %s to volume server %s ... ", string(master), string(volumeServer))
  69. err := pb.WithMasterClient(false, master, commandEnv.option.GrpcDialOption, func(client master_pb.SeaweedClient) error {
  70. _, err := client.Ping(context.Background(), &master_pb.PingRequest{
  71. Target: string(volumeServer),
  72. TargetType: cluster.VolumeServerType,
  73. })
  74. return err
  75. })
  76. if err == nil {
  77. fmt.Fprintf(writer, "ok\n")
  78. } else {
  79. fmt.Fprintf(writer, "%v\n", err)
  80. }
  81. }
  82. }
  83. // check between masters
  84. for _, sourceMaster := range masters {
  85. for _, targetMaster := range masters {
  86. if sourceMaster == targetMaster {
  87. continue
  88. }
  89. fmt.Fprintf(writer, "checking master %s to %s ... ", string(sourceMaster), string(targetMaster))
  90. err := pb.WithMasterClient(false, sourceMaster, commandEnv.option.GrpcDialOption, func(client master_pb.SeaweedClient) error {
  91. _, err := client.Ping(context.Background(), &master_pb.PingRequest{
  92. Target: string(targetMaster),
  93. TargetType: cluster.MasterType,
  94. })
  95. return err
  96. })
  97. if err == nil {
  98. fmt.Fprintf(writer, "ok\n")
  99. } else {
  100. fmt.Fprintf(writer, "%v\n", err)
  101. }
  102. }
  103. }
  104. // check from volume servers to masters
  105. for _, volumeServer := range volumeServers {
  106. for _, master := range masters {
  107. fmt.Fprintf(writer, "checking volume server %s to master %s ... ", string(volumeServer), string(master))
  108. err := pb.WithVolumeServerClient(false, volumeServer, commandEnv.option.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  109. _, err := client.Ping(context.Background(), &volume_server_pb.PingRequest{
  110. Target: string(master),
  111. TargetType: cluster.MasterType,
  112. })
  113. return err
  114. })
  115. if err == nil {
  116. fmt.Fprintf(writer, "ok\n")
  117. } else {
  118. fmt.Fprintf(writer, "%v\n", err)
  119. }
  120. }
  121. }
  122. // check from filers to masters
  123. for _, filer := range filers {
  124. for _, master := range masters {
  125. fmt.Fprintf(writer, "checking filer %s to master %s ... ", string(filer), string(master))
  126. err := pb.WithFilerClient(false, filer, commandEnv.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  127. _, err := client.Ping(context.Background(), &filer_pb.PingRequest{
  128. Target: string(master),
  129. TargetType: cluster.MasterType,
  130. })
  131. return err
  132. })
  133. if err == nil {
  134. fmt.Fprintf(writer, "ok\n")
  135. } else {
  136. fmt.Fprintf(writer, "%v\n", err)
  137. }
  138. }
  139. }
  140. // check from filers to volume servers
  141. for _, filer := range filers {
  142. for _, volumeServer := range volumeServers {
  143. fmt.Fprintf(writer, "checking filer %s to volume server %s ... ", string(filer), string(volumeServer))
  144. err := pb.WithFilerClient(false, filer, commandEnv.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  145. _, err := client.Ping(context.Background(), &filer_pb.PingRequest{
  146. Target: string(volumeServer),
  147. TargetType: cluster.VolumeServerType,
  148. })
  149. return err
  150. })
  151. if err == nil {
  152. fmt.Fprintf(writer, "ok\n")
  153. } else {
  154. fmt.Fprintf(writer, "%v\n", err)
  155. }
  156. }
  157. }
  158. // check between volume servers
  159. for _, sourceVolumeServer := range volumeServers {
  160. for _, targetVolumeServer := range volumeServers {
  161. if sourceVolumeServer == targetVolumeServer {
  162. continue
  163. }
  164. fmt.Fprintf(writer, "checking volume server %s to %s ... ", string(sourceVolumeServer), string(targetVolumeServer))
  165. err := pb.WithVolumeServerClient(false, sourceVolumeServer, commandEnv.option.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  166. _, err := client.Ping(context.Background(), &volume_server_pb.PingRequest{
  167. Target: string(targetVolumeServer),
  168. TargetType: cluster.VolumeServerType,
  169. })
  170. return err
  171. })
  172. if err == nil {
  173. fmt.Fprintf(writer, "ok\n")
  174. } else {
  175. fmt.Fprintf(writer, "%v\n", err)
  176. }
  177. }
  178. }
  179. // check between filers, and need to connect to itself
  180. for _, sourceFiler := range filers {
  181. for _, targetFiler := range filers {
  182. fmt.Fprintf(writer, "checking filer %s to %s ... ", string(sourceFiler), string(targetFiler))
  183. err := pb.WithFilerClient(false, sourceFiler, commandEnv.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  184. _, err := client.Ping(context.Background(), &filer_pb.PingRequest{
  185. Target: string(targetFiler),
  186. TargetType: cluster.FilerType,
  187. })
  188. return err
  189. })
  190. if err == nil {
  191. fmt.Fprintf(writer, "ok\n")
  192. } else {
  193. fmt.Fprintf(writer, "%v\n", err)
  194. }
  195. }
  196. }
  197. return nil
  198. }