miner: sealing jobs command

api/api_storage.go

@@ -61,6 +61,7 @@ type StorageMiner interface {
 	// WorkerConnect tells the node to connect to workers RPC
 	WorkerConnect(context.Context, string) error
 	WorkerStats(context.Context) (map[uint64]storiface.WorkerStats, error)
+	WorkerJobs(context.Context) (map[uint64][]storiface.WorkerJob, error)
 
 	stores.SectorIndex
 

api/apistruct/struct.go

@@ -228,6 +228,7 @@ type StorageMinerStruct struct {
 
 		WorkerConnect func(context.Context, string) error                             `perm:"admin"` // TODO: worker perm
 		WorkerStats   func(context.Context) (map[uint64]storiface.WorkerStats, error) `perm:"admin"`
+		WorkerJobs    func(context.Context) (map[uint64][]storiface.WorkerJob, error) `perm:"admin"`
 
 		StorageList          func(context.Context) (map[stores.ID][]stores.Decl, error)                                                                                    `perm:"admin"`
 		StorageLocal         func(context.Context) (map[stores.ID]string, error)                                                                                           `perm:"admin"`
@@ -892,6 +893,10 @@ func (c *StorageMinerStruct) WorkerStats(ctx context.Context) (map[uint64]storif
 	return c.Internal.WorkerStats(ctx)
 }
 
+func (c *StorageMinerStruct) WorkerJobs(ctx context.Context) (map[uint64][]storiface.WorkerJob, error) {
+	return c.Internal.WorkerJobs(ctx)
+}
+
 func (c *StorageMinerStruct) StorageAttach(ctx context.Context, si stores.StorageInfo, st fsutil.FsStat) error {
 	return c.Internal.StorageAttach(ctx, si, st)
 }

cmd/lotus-storage-miner/main.go

@@ -41,7 +41,7 @@ func main() {
 		lcli.WithCategory("storage", sectorsCmd),
 		lcli.WithCategory("storage", provingCmd),
 		lcli.WithCategory("storage", storageCmd),
-		lcli.WithCategory("storage", workersCmd),
+		lcli.WithCategory("storage", sealingCmd),
 	}
 	jaeger := tracing.SetupJaegerTracing("lotus")
 	defer func() {

cmd/lotus-storage-miner/sealing.go

@@ -2,8 +2,12 @@ package main
 
 import (
 	"fmt"
+	"golang.org/x/xerrors"
+	"os"
 	"sort"
 	"strings"
+	"text/tabwriter"
+	"time"
 
 	"github.com/fatih/color"
 	"github.com/urfave/cli/v2"
@@ -14,16 +18,17 @@ import (
 	lcli "github.com/filecoin-project/lotus/cli"
 )
 
-var workersCmd = &cli.Command{
+var sealingCmd = &cli.Command{
-	Name:  "workers",
+	Name:  "sealing",
-	Usage: "interact with workers",
+	Usage: "interact with sealing pipeline",
 	Subcommands: []*cli.Command{
-		workersListCmd,
+		sealingJobsCmd,
+		sealingWorkersCmd,
 	},
 }
 
-var workersListCmd = &cli.Command{
+var sealingWorkersCmd = &cli.Command{
-	Name:  "list",
+	Name:  "workers",
 	Usage: "list workers",
 	Flags: []cli.Flag{
 		&cli.BoolFlag{Name: "color"},
@@ -106,3 +111,68 @@ var workersListCmd = &cli.Command{
 		return nil
 	},
 }
+
+var sealingJobsCmd = &cli.Command{
+	Name:  "jobs",
+	Usage: "list workers",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{Name: "color"},
+	},
+	Action: func(cctx *cli.Context) error {
+		color.NoColor = !cctx.Bool("color")
+
+		nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		jobs, err := nodeApi.WorkerJobs(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting worker jobs: %w", err)
+		}
+
+		type line struct {
+			storiface.WorkerJob
+			wid uint64
+		}
+
+		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 {
+			return lines[i].Start.Before(lines[j].Start)
+		})
+
+		workerHostnames := map[uint64]string{}
+
+		wst, err := nodeApi.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\tTime\n")
+
+		for _, l := range lines {
+			_, _ = fmt.Fprintf(tw, "%d\t%d\t%d\t%s\t%s\t%s\n", l.ID, l.Sector.Number, l.wid, workerHostnames[l.wid], l.Task.Short(), time.Now().Sub(l.Start).Truncate(time.Millisecond*100))
+		}
+
+		return tw.Flush()
+	},
+}

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/filecoin-project/go-paramfetch v0.0.2-0.20200701152213-3e0f0afdc261
 	github.com/filecoin-project/go-statestore v0.1.0
 	github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b
-	github.com/filecoin-project/sector-storage v0.0.0-20200717213554-a109ef9cbeab
+	github.com/filecoin-project/sector-storage v0.0.0-20200721180125-c7da20e53cfa
 	github.com/filecoin-project/specs-actors v0.8.1-0.20200720115956-cd051eabf328
 	github.com/filecoin-project/specs-storage v0.1.1-0.20200622113353-88a9704877ea
 	github.com/filecoin-project/storage-fsm v0.0.0-20200720190000-2cfe2fe3c334

go.sum

@@ -260,8 +260,8 @@ github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b/
 github.com/filecoin-project/sector-storage v0.0.0-20200615154852-728a47ab99d6/go.mod h1:M59QnAeA/oV+Z8oHFLoNpGMv0LZ8Rll+vHVXX7GirPM=
 github.com/filecoin-project/sector-storage v0.0.0-20200712023225-1d67dcfa3c15 h1:miw6hiusb/MkV1ryoqUKKWnvHhPW00AYtyeCj0L8pqo=
 github.com/filecoin-project/sector-storage v0.0.0-20200712023225-1d67dcfa3c15/go.mod h1:salgVdX7qeXFo/xaiEQE29J4pPkjn71T0kt0n+VDBzo=
-github.com/filecoin-project/sector-storage v0.0.0-20200717213554-a109ef9cbeab h1:jEQtbWFyEKnCw3eAVCW3MSX/K7Nv03B3zzS/rfm2k+Q=
+github.com/filecoin-project/sector-storage v0.0.0-20200721180125-c7da20e53cfa h1:OZdl9TTLpNc+JpWc0yVUPJ5D7TwN70zHMuqLIzbMiVU=
-github.com/filecoin-project/sector-storage v0.0.0-20200717213554-a109ef9cbeab/go.mod h1:7EE+f7jM4kCy2MKHoiiwNDQGJSb+QQzZ+y+/17ugq4w=
+github.com/filecoin-project/sector-storage v0.0.0-20200721180125-c7da20e53cfa/go.mod h1:7EE+f7jM4kCy2MKHoiiwNDQGJSb+QQzZ+y+/17ugq4w=
 github.com/filecoin-project/specs-actors v0.0.0-20200210130641-2d1fbd8672cf/go.mod h1:xtDZUB6pe4Pksa/bAJbJ693OilaC5Wbot9jMhLm3cZA=
 github.com/filecoin-project/specs-actors v0.3.0/go.mod h1:nQYnFbQ7Y0bHZyq6HDEuVlCPR+U3z5Q3wMOQ+2aiV+Y=
 github.com/filecoin-project/specs-actors v0.6.0/go.mod h1:dRdy3cURykh2R8O/DKqy8olScl70rmIS7GrB4hB1IDY=

node/impl/storminer.go

@@ -74,6 +74,10 @@ func (sm *StorageMinerAPI) WorkerStats(context.Context) (map[uint64]storiface.Wo
 	return sm.StorageMgr.WorkerStats(), nil
 }
 
+func (sm *StorageMinerAPI) WorkerJobs(ctx context.Context) (map[uint64][]storiface.WorkerJob, error) {
+	return sm.StorageMgr.WorkerJobs(), nil
+}
+
 func (sm *StorageMinerAPI) ActorAddress(context.Context) (address.Address, error) {
 	return sm.Miner.Address(), nil
 }