Merge pull request ipfs-force-community#131 from ipfs-force-community/feat/dtynn/worker_management

Feat/dtynn/worker management

Author: diwufeiwen

venus-sector-manager/cmd/venus-sector-manager/internal/global.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"os/signal"
+	"reflect"
 	"strconv"
 	"syscall"
 	"time"
@@ -230,3 +231,18 @@ func OuputJSON(w io.Writer, v interface{}) error {
 	enc.SetIndent("", "\t")
 	return enc.Encode(v)
 }
+
+func FormatOrNull(arg interface{}, f func() string) string {
+	if arg == nil {
+		return "NULL"
+	}
+
+	rv := reflect.ValueOf(arg)
+	rt := rv.Type()
+	if rt.Kind() == reflect.Ptr && rv.IsNil() {
+		return "NULL"
+
+	}
+
+	return f()
+}

venus-sector-manager/cmd/venus-sector-manager/internal/util.go

@@ -14,6 +14,10 @@ var UtilCmd = &cli.Command{
 		utilSealerCmd,
 		utilMarketCmd,
 		utilStorageCmd,
+		utilWorkerCmd,
+	},
+	Flags: []cli.Flag{
+		SealerListenFlag,
 	},
 	Before: func(cctx *cli.Context) error {
 		logging.SetupForSub(logSubSystem)

venus-sector-manager/cmd/venus-sector-manager/internal/util_sealer.go

@@ -12,10 +12,8 @@ import (
 )
 
 var utilSealerCmd = &cli.Command{
-	Name: "sealer",
-	Flags: []cli.Flag{
-		SealerListenFlag,
-	},
+	Name:  "sealer",
+	Flags: []cli.Flag{},
 	Subcommands: []*cli.Command{
 		utilSealerSectorsCmd,
 		utilSealerProvingCmd,

venus-sector-manager/cmd/venus-sector-manager/internal/util_worker.go

@@ -0,0 +1,267 @@
+package internal
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+	"text/tabwriter"
+	"time"
+
+	"github.com/urfave/cli/v2"
+
+	"github.com/ipfs-force-community/venus-cluster/venus-sector-manager/core"
+	"github.com/ipfs-force-community/venus-cluster/venus-sector-manager/modules/util"
+	"github.com/ipfs-force-community/venus-cluster/venus-sector-manager/pkg/workercli"
+)
+
+var utilWorkerCmd = &cli.Command{
+	Name:  "worker",
+	Flags: []cli.Flag{},
+	Usage: "utils for worker management",
+	Subcommands: []*cli.Command{
+		utilWorkerListCmd,
+		utilWorkerInfoCmd,
+		utilWorkerPauseCmd,
+		utilWorkerResumeCmd,
+	},
+}
+
+var utilWorkerListCmd = &cli.Command{
+	Name: "list",
+	Flags: []cli.Flag{
+		&cli.DurationFlag{
+			Name:  "expiration",
+			Value: 10 * time.Minute,
+			Usage: "timeout for regarding a woker as missing",
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		a, actx, stopper, err := extractAPI(cctx)
+		if err != nil {
+			return fmt.Errorf("get api: %w", err)
+		}
+		defer stopper()
+
+		pinfos, err := a.Sealer.WorkerPingInfoList(actx)
+		if err != nil {
+			return RPCCallError("WorkerPingInfoList", err)
+		}
+
+		expiration := cctx.Duration("expiration")
+
+		tw := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
+		defer tw.Flush()
+		_, _ = fmt.Fprintln(tw, "Name\tDest\tThreads\tEmpty\tPaused\tErrors\tLastPing(with ! if expired)")
+		for _, pinfo := range pinfos {
+			lastPing := time.Since(time.Unix(pinfo.LastPing, 0))
+			lastPingWarn := ""
+			if lastPing > expiration {
+				lastPingWarn = " (!)"
+			}
+
+			_, _ = fmt.Fprintf(
+				tw, "%s\t%s\t%d\t%d\t%d\t%d\t%s%s\n",
+				pinfo.Info.Name,
+				pinfo.Info.Dest,
+				pinfo.Info.Summary.Threads,
+				pinfo.Info.Summary.Empty,
+				pinfo.Info.Summary.Paused,
+				pinfo.Info.Summary.Errors,
+				lastPing,
+				lastPingWarn,
+			)
+		}
+
+		return nil
+	},
+}
+
+var utilWorkerInfoCmd = &cli.Command{
+	Name:      "info",
+	Usage:     "show details about the specific worker",
+	ArgsUsage: "<worker instance name or address>",
+	Action: func(cctx *cli.Context) error {
+		args := cctx.Args()
+		if args.Len() < 1 {
+			return cli.ShowSubcommandHelp(cctx)
+		}
+
+		name := args.First()
+
+		a, actx, stopper, err := extractAPI(cctx)
+		if err != nil {
+			return fmt.Errorf("get api: %w", err)
+		}
+		defer stopper()
+
+		dest, err := resolveWorkerDest(actx, a, name)
+		if err != nil {
+			return fmt.Errorf("resolve worker dest: %w", err)
+		}
+
+		wcli, wstop, err := workercli.Connect(context.Background(), fmt.Sprintf("http://%s/", dest))
+		if err != nil {
+			return fmt.Errorf("connect to %s: %w", dest, err)
+		}
+
+		defer wstop()
+
+		// use context.Background to avoid meta info
+		details, err := wcli.WorkerList()
+		if err != nil {
+			return RPCCallError("WorkerList", err)
+		}
+
+		if len(details) == 0 {
+			return nil
+		}
+
+		tw := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
+		defer tw.Flush()
+		_, _ = fmt.Fprintln(tw, "Index\tLoc\tSectorID\tPaused\tPausedElapsed\tState\tLastErr")
+
+		for _, detail := range details {
+			_, _ = fmt.Fprintf(
+				tw, "%d\t%s\t%s\t%v\t%s\t%s\t%s\n",
+				detail.Index,
+				detail.Location,
+				FormatOrNull(detail.SectorID, func() string { return util.FormatSectorID(*detail.SectorID) }),
+				detail.Paused,
+				FormatOrNull(detail.PausedElapsed, func() string { return (time.Duration(*detail.PausedElapsed) * time.Second).String() }),
+				detail.State,
+				FormatOrNull(detail.LastError, func() string { return *detail.LastError }),
+			)
+		}
+
+		return nil
+	},
+}
+
+var utilWorkerPauseCmd = &cli.Command{
+	Name:      "pause",
+	Usage:     "pause the specified sealing thread inside target worker",
+	ArgsUsage: "<worker instance name or address> <thread index>",
+	Action: func(cctx *cli.Context) error {
+		args := cctx.Args()
+		if args.Len() < 2 {
+			return cli.ShowSubcommandHelp(cctx)
+		}
+
+		name := args.First()
+		index, err := strconv.ParseUint(args.Get(1), 10, 64)
+		if err != nil {
+			return fmt.Errorf("parse thread index: %w", err)
+		}
+
+		a, actx, stopper, err := extractAPI(cctx)
+		if err != nil {
+			return fmt.Errorf("get api: %w", err)
+		}
+		defer stopper()
+
+		dest, err := resolveWorkerDest(actx, a, name)
+		if err != nil {
+			return fmt.Errorf("resolve worker dest: %w", err)
+		}
+
+		wcli, wstop, err := workercli.Connect(context.Background(), fmt.Sprintf("http://%s/", dest))
+		if err != nil {
+			return fmt.Errorf("connect to %s: %w", dest, err)
+		}
+
+		defer wstop()
+
+		ok, err := wcli.WorkerPause(index)
+		if err != nil {
+			return RPCCallError("WorkerPause", err)
+		}
+
+		Log.With("name", name, "dest", dest, "index", index).Infof("pause call done, ok = %v", ok)
+		return nil
+	},
+}
+
+var utilWorkerResumeCmd = &cli.Command{
+	Name:      "resume",
+	Usage:     "resume the specified sealing thread inside target worker, with the given state if any",
+	ArgsUsage: "<worker instance name or address> <thread index> [<next state>]",
+	Action: func(cctx *cli.Context) error {
+		args := cctx.Args()
+		if args.Len() < 2 {
+			return cli.ShowSubcommandHelp(cctx)
+		}
+
+		name := args.First()
+		index, err := strconv.ParseUint(args.Get(1), 10, 64)
+		if err != nil {
+			return fmt.Errorf("parse thread index: %w", err)
+		}
+
+		var state *string
+		if st := args.Get(2); st != "" {
+			state = &st
+		}
+
+		a, actx, stopper, err := extractAPI(cctx)
+		if err != nil {
+			return fmt.Errorf("get api: %w", err)
+		}
+		defer stopper()
+
+		dest, err := resolveWorkerDest(actx, a, name)
+		if err != nil {
+			return fmt.Errorf("resolve worker dest: %w", err)
+		}
+
+		wcli, wstop, err := workercli.Connect(context.Background(), fmt.Sprintf("http://%s/", dest))
+		if err != nil {
+			return fmt.Errorf("connect to %s: %w", dest, err)
+		}
+
+		defer wstop()
+
+		ok, err := wcli.WorkerResume(index, state)
+		if err != nil {
+			return RPCCallError("WorkerPause", err)
+		}
+
+		Log.With("name", name, "dest", dest, "index", index, "state", state).Infof("resume call done, ok = %v", ok)
+		return nil
+	},
+}
+
+func resolveWorkerDest(ctx context.Context, a *API, name string) (string, error) {
+	var info *core.WorkerPingInfo
+	var err error
+	if a != nil {
+		info, err = a.Sealer.WorkerGetPingInfo(ctx, name)
+		if err != nil {
+			return "", RPCCallError("WorkerGetPingInfo", err)
+		}
+	}
+
+	if info != nil {
+		return info.Info.Dest, nil
+	}
+
+	addr, err := net.ResolveTCPAddr("tcp", name)
+	if err != nil {
+		ip, err := net.ResolveIPAddr("", name)
+		if err != nil {
+			return "", fmt.Errorf("no instance found, and unable to parse %q as address", name)
+		}
+
+		addr = &net.TCPAddr{
+			IP:   ip.IP,
+			Zone: ip.Zone,
+		}
+	}
+
+	if addr.Port == 0 {
+		addr.Port = core.DefaultWorkerListenPort
+	}
+
+	return addr.String(), nil
+}