feat: add node, disk, server, remote-command related command

.gitignore

@@ -1,4 +1,5 @@
 bin/
 .admin-cli-history
 pegasus.log
+.idea
 shell.log

cmd/disk_balance.go

@@ -0,0 +1,43 @@
+package cmd
+
+import (
+	"admin-cli/executor"
+	"admin-cli/shell"
+
+	"github.com/desertbit/grumble"
+)
+
+func init() {
+	shell.AddCommand(&grumble.Command{
+		Name: "disk-migrate",
+		Help: "migrate replica between the two disks within a specified ReplicaServer node",
+		Flags: func(f *grumble.Flags) {
+			/*define the flags*/
+			f.String("g", "gpid", "", "gpid, for example, '2.1'")
+			f.String("n", "node", "", "target node, for example, 127.0.0.1:34801")
+			f.String("f", "from", "", "origin disk tag, for example, ssd1")
+			f.String("t", "to", "", "target disk tag, for example, ssd2")
+		},
+		Run: func(c *grumble.Context) error {
+			return executor.DiskMigrate(
+				pegasusClient,
+				c.Flags.String("node"),
+				c.Flags.String("gpid"),
+				c.Flags.String("from"),
+				c.Flags.String("to"))
+		},
+	})
+
+	// TODO(jiashuo1) need generate migrate strategy(step) depends the disk-info result to run
+	shell.AddCommand(&grumble.Command{
+		Name: "disk-balance",
+		Help: "migrate replica automatically to let the disk capacity balance within a specified ReplicaServer node",
+		Flags: func(f *grumble.Flags) {
+			/*define the flags*/
+		},
+		Run: func(c *grumble.Context) error {
+			return executor.DiskBalance()
+		},
+	})
+
+}

cmd/disk_info.go

@@ -0,0 +1,51 @@
+package cmd
+
+import (
+	"admin-cli/executor"
+	"admin-cli/shell"
+
+	"github.com/desertbit/grumble"
+)
+
+func init() {
+	shell.AddCommand(&grumble.Command{
+		Name: "disk-capacity",
+		Help: "query disk capacity info ",
+		Flags: func(f *grumble.Flags) {
+			/*define the flags*/
+			f.Bool("r", "resolve", false, "resolve input or output address, default false")
+			f.Bool("j", "json", false, "use JSON as the format of the output results. By default tabular format is used.")
+			f.String("n", "node", "", "node address(ip:port), for example, 127.0.0.1:34801")
+			f.String("d", "disk", "", "disk tag, for example, ssd1")
+			f.String("a", "app", "", "app name, for example, temp")
+		},
+		Run: func(c *grumble.Context) error {
+			return executor.QueryDiskInfo(
+				pegasusClient,
+				executor.CapacitySize,
+				c.Flags.String("node"),
+				c.Flags.String("app"),
+				c.Flags.String("disk"))
+		},
+	})
+
+	shell.AddCommand(&grumble.Command{
+		Name: "disk-replica",
+		Help: "query disk replica count info",
+		Flags: func(f *grumble.Flags) {
+			/*define the flags*/
+			f.Bool("r", "resolve", false, "resolve input or output address, default false")
+			f.Bool("j", "json", false, "use JSON as the format of the output results. By default tabular format is used.")
+			f.String("n", "node", "", "node address(ip:port), for example, 127.0.0.1:34801")
+			f.String("a", "app", "", "app name, for example, temp")
+		},
+		Run: func(c *grumble.Context) error {
+			return executor.QueryDiskInfo(
+				pegasusClient,
+				executor.ReplicaCount,
+				c.Flags.String("node"),
+				c.Flags.String("app"),
+				"")
+		},
+	})
+}

cmd/list_nodes.go

@@ -0,0 +1,24 @@
+package cmd
+
+import (
+	"admin-cli/executor"
+	"admin-cli/shell"
+	"github.com/desertbit/grumble"
+)
+
+func init() {
+	shell.AddCommand(&grumble.Command{
+		Name:    "list-nodes",
+		Aliases: []string{"nodes"},
+		Help:    "list all nodes in the cluster",
+		Flags: func(f *grumble.Flags) {
+			f.Bool("d", "detail", false, "show detail replica count in all node")
+		},
+		Run: func(c *grumble.Context) error {
+			return executor.ListNodes(
+				pegasusClient,
+				c.Flags.Bool("detail"),
+			)
+		},
+	})
+}

cmd/remote_command.go

@@ -0,0 +1,38 @@
+package cmd
+
+import (
+	"admin-cli/executor"
+	"github.com/desertbit/grumble"
+)
+
+func init() {
+	rootCmd := &grumble.Command{
+		Name: "remote-command",
+		Help: "send remote command, for example, remote-command meta or replica",
+	}
+
+	rootCmd.AddCommand(&grumble.Command{
+		Name:  "meta",
+		Help:  "send remote command to meta server",
+		Flags: initFlag,
+		Run: func(c *grumble.Context) error {
+			return executor.RemoteCommand(pegasusClient, "meta", c.Flags.String("node"), c.Flags.String("command"), c.Flags.String("arguments"))
+		},
+	})
+
+	rootCmd.AddCommand(&grumble.Command{
+		Name:  "replica",
+		Help:  "send remote command to replica server",
+		Flags: initFlag,
+		Run: func(c *grumble.Context) error {
+			return executor.RemoteCommand(pegasusClient, "replica", c.Flags.String("node"), c.Flags.String("command"), c.Flags.String("arguments"))
+		},
+	})
+}
+
+func initFlag(f *grumble.Flags) {
+	/*define the flags*/
+	f.String("n", "node", "", "specify server node address, such as 127.0.0.1:34801, empty mean all node")
+	f.String("c", "command", "help", "remote command name, you can -c help to see support command")
+	f.String("a", "arguments", "", "if empty means query the command argument value, otherwise mean set update value")
+}

cmd/server_info.go

@@ -0,0 +1,19 @@
+package cmd
+
+import (
+	"admin-cli/executor"
+	"admin-cli/shell"
+
+	"github.com/desertbit/grumble"
+)
+
+func init() {
+	shell.AddCommand(&grumble.Command{
+		Name: "server-info",
+		Help: "displays the overall server information",
+		Run: func(c *grumble.Context) error {
+			return executor.ServerInfo(pegasusClient)
+		},
+	})
+
+}

executor/client.go

@@ -1,9 +1,15 @@
 package executor
 
 import (
+	"context"
+	"fmt"
 	"io"
+	"time"
 
+	adminClient "github.com/XiaoMi/pegasus-go-client/admin"
+	"github.com/XiaoMi/pegasus-go-client/idl/admin"
 	"github.com/XiaoMi/pegasus-go-client/session"
+	"github.com/pegasus-kv/collector/aggregate"
 )
 
 // Client can access both Pegasus ReplicaServer and MetaServer.
@@ -15,14 +21,86 @@ type Client struct {
 	ReplicaPool *session.ReplicaManager
 
 	MetaAddresses []string
+
+	ReplicaAddresses []string
 }
 
 // NewClient creates a client for accessing Pegasus cluster for use of admin-cli.
 func NewClient(writer io.Writer, metaAddrs []string) *Client {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	meta := session.NewMetaManager(metaAddrs, session.NewNodeSession)
+	replicaPool := session.NewReplicaManager(session.NewNodeSession)
+
+	resp, err := meta.ListNodes(ctx, &admin.ListNodesRequest{
+		Status: admin.NodeStatus_NS_INVALID,
+	})
+	if err != nil {
+		return nil
+	}
+
+	var nodes []string
+	for _, node := range resp.Infos {
+		nodes = append(nodes, node.Address.GetAddress())
+	}
+
 	return &Client{
-		Writer:        writer,
-		Meta:          session.NewMetaManager(metaAddrs, session.NewNodeSession),
-		ReplicaPool:   session.NewReplicaManager(session.NewNodeSession),
-		MetaAddresses: metaAddrs,
+		Writer:           writer,
+		Meta:             meta,
+		ReplicaPool:      replicaPool,
+		MetaAddresses:    metaAddrs,
+		ReplicaAddresses: nodes,
+	}
+}
+
+func (client *Client) GetReplicaClient(addr string) (*session.ReplicaSession, error) {
+	err := client.validateReplicaAddress(addr)
+	if err != nil {
+		return nil, err
+	}
+	return client.ReplicaPool.GetReplica(addr), nil
+}
+
+func (client *Client) GetRemoteCommandClient(addr string, nodeType session.NodeType) (*adminClient.RemoteCmdClient, error) {
+	switch nodeType {
+	case session.NodeTypeMeta:
+		err := client.validateMetaAddress(addr)
+		if err != nil {
+			return nil, err
+		}
+	case session.NodeTypeReplica:
+		err := client.validateReplicaAddress(addr)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return adminClient.NewRemoteCmdClient(addr, nodeType), nil
+}
+
+func (client *Client) GetPerfCounterClient(addr string) (*aggregate.PerfSession, error) {
+	err := client.validateReplicaAddress(addr)
+	if err != nil {
+		return nil, err
+	}
+	return aggregate.NewPerfSession(addr), nil
+}
+
+func (client *Client) validateReplicaAddress(addr string) error {
+	for _, node := range client.ReplicaAddresses {
+		if node == addr {
+			return nil
+		}
+	}
+	return fmt.Errorf("The cluster doesn't exist the replica server node [%s]", addr)
+}
+
+// used for remote_command -t meta
+func (client *Client) validateMetaAddress(addr string) error {
+	for _, meta := range client.MetaAddresses {
+		if addr == meta {
+			return nil
+		}
 	}
+	return fmt.Errorf("The cluster doesn't exist the meta server node [%s]", addr)
 }

executor/disk_balance.go

@@ -0,0 +1,51 @@
+package executor
+
+import (
+	"admin-cli/helper"
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/XiaoMi/pegasus-go-client/idl/radmin"
+)
+
+func DiskMigrate(client *Client, replicaServer string, pidStr string, from string, to string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	var addr, err = helper.Resolve(replicaServer, helper.Host2Addr)
+	if err == nil {
+		replicaServer = addr
+	}
+
+	pid, err := helper.Str2Gpid(pidStr)
+	if err != nil {
+		return err
+	}
+
+	replicaClient, err := client.GetReplicaClient(replicaServer)
+	if err != nil {
+		return err
+	}
+
+	resp, err := replicaClient.DiskMigrate(ctx, &radmin.ReplicaDiskMigrateRequest{
+		Pid:        pid,
+		OriginDisk: from,
+		TargetDisk: to,
+	})
+
+	if err != nil {
+		if resp != nil && resp.Hint != nil {
+			return fmt.Errorf("Internal server error [%s:%s]", err, *resp.Hint)
+		}
+		return err
+	}
+
+	return nil
+}
+
+// TODO(jiashuo1) need generate migrate strategy(step) depends the disk-info result to run
+func DiskBalance() error {
+	fmt.Println("Wait support")
+	return nil
+}