Merge pull request #9101 from filecoin-project/sbansal/lotus-worker-stop-cmd

feat: worker: Add stop cmd for lotus worker

api/api_worker.go

@@ -76,6 +76,10 @@ type Worker interface {
 
 	// Like ProcessSession, but returns an error when worker is disabled
 	Session(context.Context) (uuid.UUID, error) //perm:admin
+
+	// Trigger shutdown
+	Shutdown(context.Context) error //perm:admin
+
 }
 
 var _ storiface.WorkerCalls = *new(Worker)

api/proxy_gen.go

@@ -967,6 +967,8 @@ type WorkerStruct struct {
 
 		SetEnabled func(p0 context.Context, p1 bool) error `perm:"admin"`
 
+		Shutdown func(p0 context.Context) error `perm:"admin"`
+
 		StorageAddLocal func(p0 context.Context, p1 string) error `perm:"admin"`
 
 		TaskDisable func(p0 context.Context, p1 sealtasks.TaskType) error `perm:"admin"`
@@ -5573,6 +5575,17 @@ func (s *WorkerStub) SetEnabled(p0 context.Context, p1 bool) error {
 	return ErrNotSupported
 }
 
+func (s *WorkerStruct) Shutdown(p0 context.Context) error {
+	if s.Internal.Shutdown == nil {
+		return ErrNotSupported
+	}
+	return s.Internal.Shutdown(p0)
+}
+
+func (s *WorkerStub) Shutdown(p0 context.Context) error {
+	return ErrNotSupported
+}
+
 func (s *WorkerStruct) StorageAddLocal(p0 context.Context, p1 string) error {
 	if s.Internal.StorageAddLocal == nil {
 		return ErrNotSupported

cmd/lotus-worker/main.go

@@ -55,6 +55,7 @@ func main() {
 
 	local := []*cli.Command{
 		runCmd,
+		stopCmd,
 		infoCmd,
 		storageCmd,
 		setCmd,
@@ -115,6 +116,27 @@ func main() {
 	}
 }
 
+var stopCmd = &cli.Command{
+	Name:  "stop",
+	Usage: "Stop a running lotus worker",
+	Flags: []cli.Flag{},
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := lcli.GetWorkerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+		err = api.Shutdown(ctx)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	},
+}
+
 var runCmd = &cli.Command{
 	Name:  "run",
 	Usage: "Start lotus worker",
@@ -623,6 +645,17 @@ var runCmd = &cli.Command{
 			}
 		}()
 
+		go func() {
+			<-workerApi.Done()
+			// Wait 20s to allow the miner to unregister the worker on next heartbeat
+			time.Sleep(20 * time.Second)
+			log.Warn("Shutting down...")
+			if err := srv.Shutdown(context.TODO()); err != nil {
+				log.Errorf("shutting down RPC server failed: %s", err)
+			}
+			log.Warn("Graceful shutdown successful")
+		}()
+
 		return srv.Serve(nl)
 	},
 }

cmd/lotus-worker/sealworker/rpc.go

@@ -118,4 +118,8 @@ func (w *Worker) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error)
 	return build.OpenRPCDiscoverJSON_Worker(), nil
 }
 
+func (w *Worker) Shutdown(ctx context.Context) error {
+	return w.LocalWorker.Close()
+}
+
 var _ storiface.WorkerCalls = &Worker{}

documentation/en/api-v0-methods-worker.md

@@ -6,6 +6,7 @@
   * [Paths](#Paths)
   * [Remove](#Remove)
   * [Session](#Session)
+  * [Shutdown](#Shutdown)
   * [Version](#Version)
 * [Add](#Add)
   * [AddPiece](#AddPiece)
@@ -1453,6 +1454,16 @@ Inputs: `null`
 
 Response: `"07070707-0707-0707-0707-070707070707"`
 
+### Shutdown
+Trigger shutdown
+
+
+Perms: admin
+
+Inputs: `null`
+
+Response: `{}`
+
 ### Version
 
 

documentation/en/cli-lotus-worker.md

@@ -11,6 +11,7 @@ VERSION:
 
 COMMANDS:
    run         Start lotus worker
+   stop        Stop a running lotus worker
    info        Print worker info
    storage     manage sector storage
    set         Manage worker settings
@@ -58,6 +59,19 @@ OPTIONS:
    
 ```
 
+## lotus-worker stop
+```
+NAME:
+   lotus-worker stop - Stop a running lotus worker
+
+USAGE:
+   lotus-worker stop [command options] [arguments...]
+
+OPTIONS:
+   --help, -h  show help (default: false)
+   
+```
+
 ## lotus-worker info
 ```
 NAME:

storage/sealer/worker_local.go

@@ -829,6 +829,10 @@ func (l *LocalWorker) Close() error {
 	return nil
 }
 
+func (l *LocalWorker) Done() <-chan struct{} {
+	return l.closing
+}
+
 // WaitQuiet blocks as long as there are tasks running
 func (l *LocalWorker) WaitQuiet() {
 	l.running.Wait()