sealing.go

package main

import (
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"sort"
	"strings"
	"text/tabwriter"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/fatih/color"
	"github.com/google/uuid"
	"github.com/mitchellh/go-homedir"
	"github.com/urfave/cli/v2"
	"golang.org/x/xerrors"

	"github.com/filecoin-project/go-padreader"
	"github.com/filecoin-project/go-state-types/abi"

	"github.com/filecoin-project/lotus/chain/types"
	lcli "github.com/filecoin-project/lotus/cli"
	"github.com/filecoin-project/lotus/lib/httpreader"
	"github.com/filecoin-project/lotus/storage/sealer/sealtasks"
	"github.com/filecoin-project/lotus/storage/sealer/storiface"
)

var sealingCmd = &cli.Command{
	Name:  "sealing",
	Usage: "interact with sealing pipeline",
	Subcommands: []*cli.Command{
		sealingJobsCmd,
		workersCmd(true),
		sealingSchedDiagCmd,
		sealingAbortCmd,
		sealingDataCidCmd,
	},
}

func workersCmd(sealing bool) *cli.Command {
	return &cli.Command{
		Name:  "workers",
		Usage: "list workers",
		Action: func(cctx *cli.Context) error {
			minerApi, closer, err := lcli.GetStorageMinerAPI(cctx)
			if err != nil {
				return err
			}
			defer closer()

			ctx := lcli.ReqContext(cctx)

			stats, err := minerApi.WorkerStats(ctx)
			if err != nil {
				return err
			}

			type sortableStat struct {
				id uuid.UUID
				storiface.WorkerStats
			}

			st := make([]sortableStat, 0, len(stats))
			for id, stat := range stats {
				if len(stat.Tasks) > 0 {
					if (stat.Tasks[0].WorkerType() != sealtasks.WorkerSealing) == sealing {
						continue
					}
				}

				st = append(st, sortableStat{id, stat})
			}

			sort.Slice(st, func(i, j int) bool {
				return st[i].id.String() < st[j].id.String()
			})

			/*
				Example output:

				Worker c4d65451-07f8-4230-98ad-4f33dea2a8cc, host myhostname
				        TASK: PC1(1/4) AP(15/15) GET(3)
				        CPU:  [||||||||                                                        ] 16/128 core(s) in use
				        RAM:  [||||||||                                                        ] 12% 125.8 GiB/1008 GiB
				        VMEM: [||||||||                                                        ] 12% 125.8 GiB/1008 GiB
				        GPU:  [                                                                ] 0% 0.00/1 gpu(s) in use
				        GPU: NVIDIA GeForce RTX 3090, not used
			*/

			for _, stat := range st {
				// Worker uuid + name

				var disabled string
				if !stat.Enabled {
					disabled = color.RedString(" (disabled)")
				}

				fmt.Printf("Worker %s, host %s%s\n", stat.id, color.MagentaString(stat.Info.Hostname), disabled)

				// Task counts
				tc := make([][]string, 0, len(stat.TaskCounts))

				for st, c := range stat.TaskCounts {
					if c == 0 {
						continue
					}

					stt, err := sealtasks.SttFromString(st)
					if err != nil {
						return err
					}

					str := fmt.Sprint(c)
					if max := stat.Info.Resources.ResourceSpec(stt.RegisteredSealProof, stt.TaskType).MaxConcurrent; max > 0 {
						switch {
						case c < max:
							str = color.GreenString(str)
						case c >= max:
							str = color.YellowString(str)
						}
						str = fmt.Sprintf("%s/%d", str, max)
					} else {
						str = color.CyanString(str)
					}
					str = fmt.Sprintf("%s(%s)", color.BlueString(stt.Short()), str)

					tc = append(tc, []string{string(stt.TaskType), str})
				}
				sort.Slice(tc, func(i, j int) bool {
					return sealtasks.TaskType(tc[i][0]).Less(sealtasks.TaskType(tc[j][0]))
				})
				var taskStr string
				for _, t := range tc {
					taskStr += t[1] + " "
				}
				if taskStr != "" {
					fmt.Printf("\tTASK: %s\n", taskStr)
				}

				// CPU use

				fmt.Printf("\tCPU:  [%s] %d/%d core(s) in use\n",
					lcli.BarString(float64(stat.Info.Resources.CPUs), 0, float64(stat.CpuUse)), stat.CpuUse, stat.Info.Resources.CPUs)

				// RAM use

				ramTotal := stat.Info.Resources.MemPhysical
				ramTasks := stat.MemUsedMin
				ramUsed := stat.Info.Resources.MemUsed
				var ramReserved uint64 = 0
				if ramUsed > ramTasks {
					ramReserved = ramUsed - ramTasks
				}
				ramBar := lcli.BarString(float64(ramTotal), float64(ramReserved), float64(ramTasks))

				fmt.Printf("\tRAM:  [%s] %d%% %s/%s\n", ramBar,
					(ramTasks+ramReserved)*100/stat.Info.Resources.MemPhysical,
					types.SizeStr(types.NewInt(ramTasks+ramUsed)),
					types.SizeStr(types.NewInt(stat.Info.Resources.MemPhysical)))

				// VMEM use (ram+swap)

				vmemTotal := stat.Info.Resources.MemPhysical + stat.Info.Resources.MemSwap
				vmemTasks := stat.MemUsedMax
				vmemUsed := stat.Info.Resources.MemUsed + stat.Info.Resources.MemSwapUsed
				var vmemReserved uint64 = 0
				if vmemUsed > vmemTasks {
					vmemReserved = vmemUsed - vmemTasks
				}
				vmemBar := lcli.BarString(float64(vmemTotal), float64(vmemReserved), float64(vmemTasks))

				fmt.Printf("\tVMEM: [%s] %d%% %s/%s\n", vmemBar,
					(vmemTasks+vmemReserved)*100/vmemTotal,
					types.SizeStr(types.NewInt(vmemTasks+vmemReserved)),
					types.SizeStr(types.NewInt(vmemTotal)))

				// GPU use

				if len(stat.Info.Resources.GPUs) > 0 {
					gpuBar := lcli.BarString(float64(len(stat.Info.Resources.GPUs)), 0, stat.GpuUsed)
					fmt.Printf("\tGPU:  [%s] %.f%% %.2f/%d gpu(s) in use\n", color.GreenString(gpuBar),
						stat.GpuUsed*100/float64(len(stat.Info.Resources.GPUs)),
						stat.GpuUsed, len(stat.Info.Resources.GPUs))
				}

				gpuUse := "not "
				gpuCol := color.FgBlue
				if stat.GpuUsed > 0 {
					gpuCol = color.FgGreen
					gpuUse = ""
				}
				for _, gpu := range stat.Info.Resources.GPUs {
					fmt.Printf("\tGPU: %s\n", color.New(gpuCol).Sprintf("%s, %sused", gpu, gpuUse))
				}
			}

			return nil
		},
	}
}

var sealingJobsCmd = &cli.Command{
	Name:  "jobs",
	Usage: "list running jobs",
	Flags: []cli.Flag{
		&cli.BoolFlag{
			Name:  "show-ret-done",
			Usage: "show returned but not consumed calls",
		},
	},
	Action: func(cctx *cli.Context) error {
		minerApi, closer, err := lcli.GetStorageMinerAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		ctx := lcli.ReqContext(cctx)

		jobs, err := minerApi.WorkerJobs(ctx)
		if err != nil {
			return xerrors.Errorf("getting worker jobs: %w", err)
		}

		type line struct {
			storiface.WorkerJob
			wid uuid.UUID
		}

		lines := make([]line, 0)

		for wid, jobs := range jobs {
			for _, job := range jobs {
				lines = append(lines, line{
					WorkerJob: job,
					wid:       wid,
				})
			}
		}

		// oldest first
		sort.Slice(lines, func(i, j int) bool {
			if lines[i].RunWait != lines[j].RunWait {
				return lines[i].RunWait < lines[j].RunWait
			}
			if lines[i].Start.Equal(lines[j].Start) {
				return lines[i].ID.ID.String() < lines[j].ID.ID.String()
			}
			return lines[i].Start.Before(lines[j].Start)
		})

		workerHostnames := map[uuid.UUID]string{}

		wst, err := minerApi.WorkerStats(ctx)
		if err != nil {
			return xerrors.Errorf("getting worker stats: %w", err)
		}

		for wid, st := range wst {
			workerHostnames[wid] = st.Info.Hostname
		}

		tw := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
		_, _ = fmt.Fprintf(tw, "ID\tSector\tWorker\tHostname\tTask\tState\tTime\n")

		for _, l := range lines {
			state := "running"
			switch {
			case l.RunWait > 1:
				state = fmt.Sprintf("assigned(%d)", l.RunWait-1)
			case l.RunWait == storiface.RWPrepared:
				state = "prepared"
			case l.RunWait == storiface.RWRetDone:
				if !cctx.Bool("show-ret-done") {
					continue
				}
				state = "ret-done"
			case l.RunWait == storiface.RWReturned:
				state = "returned"
			case l.RunWait == storiface.RWRetWait:
				state = "ret-wait"
			}
			dur := "n/a"
			if !l.Start.IsZero() {
				dur = time.Now().Sub(l.Start).Truncate(time.Millisecond * 100).String()
			}

			hostname, ok := workerHostnames[l.wid]
			if !ok {
				hostname = l.Hostname
			}

			_, _ = fmt.Fprintf(tw, "%s\t%d\t%s\t%s\t%s\t%s\t%s\n",
				hex.EncodeToString(l.ID.ID[:4]),
				l.Sector.Number,
				hex.EncodeToString(l.wid[:4]),
				hostname,
				l.Task.Short(),
				state,
				dur)
		}

		return tw.Flush()
	},
}

var sealingSchedDiagCmd = &cli.Command{
	Name:  "sched-diag",
	Usage: "Dump internal scheduler state",
	Flags: []cli.Flag{
		&cli.BoolFlag{
			Name: "force-sched",
		},
	},
	Action: func(cctx *cli.Context) error {
		minerApi, closer, err := lcli.GetStorageMinerAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		ctx := lcli.ReqContext(cctx)

		st, err := minerApi.SealingSchedDiag(ctx, cctx.Bool("force-sched"))
		if err != nil {
			return err
		}

		j, err := json.MarshalIndent(&st, "", "  ")
		if err != nil {
			return err
		}

		fmt.Println(string(j))

		return nil
	},
}

var sealingAbortCmd = &cli.Command{
	Name:      "abort",
	Usage:     "Abort a running job",
	ArgsUsage: "[callid]",
	Flags: []cli.Flag{
		&cli.BoolFlag{
			Name:  "sched",
			Usage: "Specifies that the argument is UUID of the request to be removed from scheduler",
		},
	},
	Action: func(cctx *cli.Context) error {
		if cctx.NArg() != 1 {
			return lcli.IncorrectNumArgs(cctx)
		}

		minerApi, closer, err := lcli.GetStorageMinerAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		ctx := lcli.ReqContext(cctx)

		if cctx.Bool("sched") {
			err = minerApi.SealingRemoveRequest(ctx, uuid.Must(uuid.Parse(cctx.Args().First())))
			if err != nil {
				return xerrors.Errorf("Failed to removed the request with UUID %s: %w", cctx.Args().First(), err)
			}
			return nil
		}

		jobs, err := minerApi.WorkerJobs(ctx)
		if err != nil {
			return xerrors.Errorf("getting worker jobs: %w", err)
		}

		var job *storiface.WorkerJob
	outer:
		for _, workerJobs := range jobs {
			for _, j := range workerJobs {
				if strings.HasPrefix(j.ID.ID.String(), cctx.Args().First()) {
					j := j
					job = &j
					break outer
				}
			}
		}

		if job == nil {
			return xerrors.Errorf("job with specified id prefix not found")
		}

		fmt.Printf("aborting job %s, task %s, sector %d, running on host %s\n", job.ID.String(), job.Task.Short(), job.Sector.Number, job.Hostname)

		return minerApi.SealingAbort(ctx, job.ID)
	},
}

var sealingDataCidCmd = &cli.Command{
	Name:      "data-cid",
	Usage:     "Compute data CID using workers",
	ArgsUsage: "[file/url] <padded piece size>",
	Flags: []cli.Flag{
		&cli.Uint64Flag{
			Name:  "file-size",
			Usage: "real file size",
		},
	},
	Action: func(cctx *cli.Context) error {
		if cctx.NArg() < 1 || cctx.NArg() > 2 {
			return lcli.ShowHelp(cctx, xerrors.Errorf("expected 1 or 2 arguments"))
		}

		minerApi, closer, err := lcli.GetStorageMinerAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		ctx := lcli.ReqContext(cctx)

		var r io.Reader
		sz := cctx.Uint64("file-size")

		if strings.HasPrefix(cctx.Args().First(), "http://") || strings.HasPrefix(cctx.Args().First(), "https://") {
			r = &httpreader.HttpReader{
				URL: cctx.Args().First(),
			}

			if !cctx.IsSet("file-size") {
				resp, err := http.Head(cctx.Args().First())
				if err != nil {
					return xerrors.Errorf("http head: %w", err)
				}

				if resp.ContentLength < 0 {
					return xerrors.Errorf("head response didn't contain content length; specify --file-size")
				}
				sz = uint64(resp.ContentLength)
			}
		} else {
			p, err := homedir.Expand(cctx.Args().First())
			if err != nil {
				return xerrors.Errorf("expanding path: %w", err)
			}

			f, err := os.OpenFile(p, os.O_RDONLY, 0)
			if err != nil {
				return xerrors.Errorf("opening source file: %w", err)
			}

			if !cctx.IsSet("file-size") {
				st, err := f.Stat()
				if err != nil {
					return xerrors.Errorf("stat: %w", err)
				}
				sz = uint64(st.Size())
			}

			r = f
		}

		var psize abi.PaddedPieceSize
		if cctx.NArg() == 2 {
			rps, err := humanize.ParseBytes(cctx.Args().Get(1))
			if err != nil {
				return xerrors.Errorf("parsing piece size: %w", err)
			}
			psize = abi.PaddedPieceSize(rps)
			if err := psize.Validate(); err != nil {
				return xerrors.Errorf("checking piece size: %w", err)
			}
			if sz > uint64(psize.Unpadded()) {
				return xerrors.Errorf("file larger than the piece")
			}
		} else {
			psize = padreader.PaddedSize(sz).Padded()
		}

		pc, err := minerApi.ComputeDataCid(ctx, psize.Unpadded(), r)
		if err != nil {
			return xerrors.Errorf("computing data CID: %w", err)
		}

		fmt.Println(pc.PieceCID, " ", pc.Size)
		return nil
	},
}