555 lines
14 KiB
Go
555 lines
14 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io/ioutil"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/gorilla/mux"
|
|
"github.com/ipfs/go-datastore/namespace"
|
|
logging "github.com/ipfs/go-log/v2"
|
|
manet "github.com/multiformats/go-multiaddr/net"
|
|
"github.com/urfave/cli/v2"
|
|
"go.opencensus.io/stats/view"
|
|
"go.opencensus.io/tag"
|
|
"golang.org/x/xerrors"
|
|
|
|
"github.com/filecoin-project/go-jsonrpc"
|
|
"github.com/filecoin-project/go-jsonrpc/auth"
|
|
paramfetch "github.com/filecoin-project/go-paramfetch"
|
|
"github.com/filecoin-project/go-statestore"
|
|
|
|
"github.com/filecoin-project/lotus/api"
|
|
"github.com/filecoin-project/lotus/api/apistruct"
|
|
"github.com/filecoin-project/lotus/build"
|
|
lcli "github.com/filecoin-project/lotus/cli"
|
|
sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
|
|
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
|
|
"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
|
|
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
|
|
"github.com/filecoin-project/lotus/lib/lotuslog"
|
|
"github.com/filecoin-project/lotus/lib/rpcenc"
|
|
"github.com/filecoin-project/lotus/metrics"
|
|
"github.com/filecoin-project/lotus/node/modules"
|
|
"github.com/filecoin-project/lotus/node/repo"
|
|
)
|
|
|
|
var log = logging.Logger("main")
|
|
|
|
const FlagWorkerRepo = "worker-repo"
|
|
|
|
// TODO remove after deprecation period
|
|
const FlagWorkerRepoDeprecation = "workerrepo"
|
|
|
|
func main() {
|
|
build.RunningNodeType = build.NodeWorker
|
|
|
|
lotuslog.SetupLogLevels()
|
|
|
|
local := []*cli.Command{
|
|
runCmd,
|
|
infoCmd,
|
|
storageCmd,
|
|
setCmd,
|
|
waitQuietCmd,
|
|
}
|
|
|
|
app := &cli.App{
|
|
Name: "lotus-worker",
|
|
Usage: "Remote miner worker",
|
|
Version: build.UserVersion(),
|
|
Flags: []cli.Flag{
|
|
&cli.StringFlag{
|
|
Name: FlagWorkerRepo,
|
|
Aliases: []string{FlagWorkerRepoDeprecation},
|
|
EnvVars: []string{"LOTUS_WORKER_PATH", "WORKER_PATH"},
|
|
Value: "~/.lotusworker", // TODO: Consider XDG_DATA_HOME
|
|
Usage: fmt.Sprintf("Specify worker repo path. flag %s and env WORKER_PATH are DEPRECATION, will REMOVE SOON", FlagWorkerRepoDeprecation),
|
|
},
|
|
&cli.StringFlag{
|
|
Name: "miner-repo",
|
|
Aliases: []string{"storagerepo"},
|
|
EnvVars: []string{"LOTUS_MINER_PATH", "LOTUS_STORAGE_PATH"},
|
|
Value: "~/.lotusminer", // TODO: Consider XDG_DATA_HOME
|
|
Usage: fmt.Sprintf("Specify miner repo path. flag storagerepo and env LOTUS_STORAGE_PATH are DEPRECATION, will REMOVE SOON"),
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "enable-gpu-proving",
|
|
Usage: "enable use of GPU for mining operations",
|
|
Value: true,
|
|
},
|
|
},
|
|
|
|
Commands: local,
|
|
}
|
|
app.Setup()
|
|
app.Metadata["repoType"] = repo.Worker
|
|
|
|
if err := app.Run(os.Args); err != nil {
|
|
log.Warnf("%+v", err)
|
|
return
|
|
}
|
|
}
|
|
|
|
var runCmd = &cli.Command{
|
|
Name: "run",
|
|
Usage: "Start lotus worker",
|
|
Flags: []cli.Flag{
|
|
&cli.StringFlag{
|
|
Name: "listen",
|
|
Usage: "host address and port the worker api will listen on",
|
|
Value: "0.0.0.0:3456",
|
|
},
|
|
&cli.StringFlag{
|
|
Name: "address",
|
|
Hidden: true,
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "no-local-storage",
|
|
Usage: "don't use storageminer repo for sector storage",
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "no-swap",
|
|
Usage: "don't use swap",
|
|
Value: false,
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "addpiece",
|
|
Usage: "enable addpiece",
|
|
Value: true,
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "precommit1",
|
|
Usage: "enable precommit1 (32G sectors: 1 core, 128GiB Memory)",
|
|
Value: true,
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "unseal",
|
|
Usage: "enable unsealing (32G sectors: 1 core, 128GiB Memory)",
|
|
Value: true,
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "precommit2",
|
|
Usage: "enable precommit2 (32G sectors: all cores, 96GiB Memory)",
|
|
Value: true,
|
|
},
|
|
&cli.BoolFlag{
|
|
Name: "commit",
|
|
Usage: "enable commit (32G sectors: all cores or GPUs, 128GiB Memory + 64GiB swap)",
|
|
Value: true,
|
|
},
|
|
&cli.IntFlag{
|
|
Name: "parallel-fetch-limit",
|
|
Usage: "maximum fetch operations to run in parallel",
|
|
Value: 5,
|
|
},
|
|
&cli.StringFlag{
|
|
Name: "timeout",
|
|
Usage: "used when 'listen' is unspecified. must be a valid duration recognized by golang's time.ParseDuration function",
|
|
Value: "30m",
|
|
},
|
|
},
|
|
Before: func(cctx *cli.Context) error {
|
|
if cctx.IsSet("address") {
|
|
log.Warnf("The '--address' flag is deprecated, it has been replaced by '--listen'")
|
|
if err := cctx.Set("listen", cctx.String("address")); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
},
|
|
Action: func(cctx *cli.Context) error {
|
|
log.Info("Starting lotus worker")
|
|
|
|
if !cctx.Bool("enable-gpu-proving") {
|
|
if err := os.Setenv("BELLMAN_NO_GPU", "true"); err != nil {
|
|
return xerrors.Errorf("could not set no-gpu env: %+v", err)
|
|
}
|
|
}
|
|
|
|
// Connect to storage-miner
|
|
ctx := lcli.ReqContext(cctx)
|
|
|
|
var nodeApi api.StorageMiner
|
|
var closer func()
|
|
var err error
|
|
for {
|
|
nodeApi, closer, err = lcli.GetStorageMinerAPI(cctx, lcli.StorageMinerUseHttp)
|
|
if err == nil {
|
|
_, err = nodeApi.Version(ctx)
|
|
if err == nil {
|
|
break
|
|
}
|
|
}
|
|
fmt.Printf("\r\x1b[0KConnecting to miner API... (%s)", err)
|
|
time.Sleep(time.Second)
|
|
continue
|
|
}
|
|
|
|
defer closer()
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
|
|
// Register all metric views
|
|
if err := view.Register(
|
|
metrics.DefaultViews...,
|
|
); err != nil {
|
|
log.Fatalf("Cannot register the view: %v", err)
|
|
}
|
|
|
|
v, err := nodeApi.Version(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if v.APIVersion != build.MinerAPIVersion {
|
|
return xerrors.Errorf("lotus-miner API version doesn't match: expected: %s", api.Version{APIVersion: build.MinerAPIVersion})
|
|
}
|
|
log.Infof("Remote version %s", v)
|
|
|
|
// Check params
|
|
|
|
act, err := nodeApi.ActorAddress(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ssize, err := nodeApi.ActorSectorSize(ctx, act)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if cctx.Bool("commit") {
|
|
if err := paramfetch.GetParams(ctx, build.ParametersJSON(), uint64(ssize)); err != nil {
|
|
return xerrors.Errorf("get params: %w", err)
|
|
}
|
|
}
|
|
|
|
var taskTypes []sealtasks.TaskType
|
|
|
|
taskTypes = append(taskTypes, sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize)
|
|
|
|
if cctx.Bool("addpiece") {
|
|
taskTypes = append(taskTypes, sealtasks.TTAddPiece)
|
|
}
|
|
if cctx.Bool("precommit1") {
|
|
taskTypes = append(taskTypes, sealtasks.TTPreCommit1)
|
|
}
|
|
if cctx.Bool("unseal") {
|
|
taskTypes = append(taskTypes, sealtasks.TTUnseal)
|
|
}
|
|
if cctx.Bool("precommit2") {
|
|
taskTypes = append(taskTypes, sealtasks.TTPreCommit2)
|
|
}
|
|
if cctx.Bool("commit") {
|
|
taskTypes = append(taskTypes, sealtasks.TTCommit2)
|
|
}
|
|
|
|
if len(taskTypes) == 0 {
|
|
return xerrors.Errorf("no task types specified")
|
|
}
|
|
|
|
// Open repo
|
|
|
|
repoPath := cctx.String(FlagWorkerRepo)
|
|
r, err := repo.NewFS(repoPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ok, err := r.Exists()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !ok {
|
|
if err := r.Init(repo.Worker); err != nil {
|
|
return err
|
|
}
|
|
|
|
lr, err := r.Lock(repo.Worker)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var localPaths []stores.LocalPath
|
|
|
|
if !cctx.Bool("no-local-storage") {
|
|
b, err := json.MarshalIndent(&stores.LocalStorageMeta{
|
|
ID: stores.ID(uuid.New().String()),
|
|
Weight: 10,
|
|
CanSeal: true,
|
|
CanStore: false,
|
|
}, "", " ")
|
|
if err != nil {
|
|
return xerrors.Errorf("marshaling storage config: %w", err)
|
|
}
|
|
|
|
if err := ioutil.WriteFile(filepath.Join(lr.Path(), "sectorstore.json"), b, 0644); err != nil {
|
|
return xerrors.Errorf("persisting storage metadata (%s): %w", filepath.Join(lr.Path(), "sectorstore.json"), err)
|
|
}
|
|
|
|
localPaths = append(localPaths, stores.LocalPath{
|
|
Path: lr.Path(),
|
|
})
|
|
}
|
|
|
|
if err := lr.SetStorage(func(sc *stores.StorageConfig) {
|
|
sc.StoragePaths = append(sc.StoragePaths, localPaths...)
|
|
}); err != nil {
|
|
return xerrors.Errorf("set storage config: %w", err)
|
|
}
|
|
|
|
{
|
|
// init datastore for r.Exists
|
|
_, err := lr.Datastore("/metadata")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := lr.Close(); err != nil {
|
|
return xerrors.Errorf("close repo: %w", err)
|
|
}
|
|
}
|
|
|
|
lr, err := r.Lock(repo.Worker)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
if err := lr.Close(); err != nil {
|
|
log.Error("closing repo", err)
|
|
}
|
|
}()
|
|
ds, err := lr.Datastore("/metadata")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Info("Opening local storage; connecting to master")
|
|
const unspecifiedAddress = "0.0.0.0"
|
|
address := cctx.String("listen")
|
|
addressSlice := strings.Split(address, ":")
|
|
if ip := net.ParseIP(addressSlice[0]); ip != nil {
|
|
if ip.String() == unspecifiedAddress {
|
|
timeout, err := time.ParseDuration(cctx.String("timeout"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rip, err := extractRoutableIP(timeout)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
address = rip + ":" + addressSlice[1]
|
|
}
|
|
}
|
|
|
|
localStore, err := stores.NewLocal(ctx, lr, nodeApi, []string{"http://" + address + "/remote"})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Setup remote sector store
|
|
spt, err := ffiwrapper.SealProofTypeFromSectorSize(ssize)
|
|
if err != nil {
|
|
return xerrors.Errorf("getting proof type: %w", err)
|
|
}
|
|
|
|
sminfo, err := lcli.GetAPIInfo(cctx, repo.StorageMiner)
|
|
if err != nil {
|
|
return xerrors.Errorf("could not get api info: %w", err)
|
|
}
|
|
|
|
remote := stores.NewRemote(localStore, nodeApi, sminfo.AuthHeader(), cctx.Int("parallel-fetch-limit"))
|
|
|
|
// Create / expose the worker
|
|
|
|
wsts := statestore.New(namespace.Wrap(ds, modules.WorkerCallsPrefix))
|
|
|
|
workerApi := &worker{
|
|
LocalWorker: sectorstorage.NewLocalWorker(sectorstorage.WorkerConfig{
|
|
SealProof: spt,
|
|
TaskTypes: taskTypes,
|
|
NoSwap: cctx.Bool("no-swap"),
|
|
}, remote, localStore, nodeApi, nodeApi, wsts),
|
|
localStore: localStore,
|
|
ls: lr,
|
|
}
|
|
|
|
mux := mux.NewRouter()
|
|
|
|
log.Info("Setting up control endpoint at " + address)
|
|
|
|
readerHandler, readerServerOpt := rpcenc.ReaderParamDecoder()
|
|
rpcServer := jsonrpc.NewServer(readerServerOpt)
|
|
rpcServer.Register("Filecoin", apistruct.PermissionedWorkerAPI(metrics.MetricedWorkerAPI(workerApi)))
|
|
|
|
mux.Handle("/rpc/v0", rpcServer)
|
|
mux.Handle("/rpc/streams/v0/push/{uuid}", readerHandler)
|
|
mux.PathPrefix("/remote").HandlerFunc((&stores.FetchHandler{Local: localStore}).ServeHTTP)
|
|
mux.PathPrefix("/").Handler(http.DefaultServeMux) // pprof
|
|
|
|
ah := &auth.Handler{
|
|
Verify: nodeApi.AuthVerify,
|
|
Next: mux.ServeHTTP,
|
|
}
|
|
|
|
srv := &http.Server{
|
|
Handler: ah,
|
|
BaseContext: func(listener net.Listener) context.Context {
|
|
ctx, _ := tag.New(context.Background(), tag.Upsert(metrics.APIInterface, "lotus-worker"))
|
|
return ctx
|
|
},
|
|
}
|
|
|
|
go func() {
|
|
<-ctx.Done()
|
|
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")
|
|
}()
|
|
|
|
nl, err := net.Listen("tcp", address)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
{
|
|
a, err := net.ResolveTCPAddr("tcp", address)
|
|
if err != nil {
|
|
return xerrors.Errorf("parsing address: %w", err)
|
|
}
|
|
|
|
ma, err := manet.FromNetAddr(a)
|
|
if err != nil {
|
|
return xerrors.Errorf("creating api multiaddress: %w", err)
|
|
}
|
|
|
|
if err := lr.SetAPIEndpoint(ma); err != nil {
|
|
return xerrors.Errorf("setting api endpoint: %w", err)
|
|
}
|
|
|
|
ainfo, err := lcli.GetAPIInfo(cctx, repo.StorageMiner)
|
|
if err != nil {
|
|
return xerrors.Errorf("could not get miner API info: %w", err)
|
|
}
|
|
|
|
// TODO: ideally this would be a token with some permissions dropped
|
|
if err := lr.SetAPIToken(ainfo.Token); err != nil {
|
|
return xerrors.Errorf("setting api token: %w", err)
|
|
}
|
|
}
|
|
|
|
minerSession, err := nodeApi.Session(ctx)
|
|
if err != nil {
|
|
return xerrors.Errorf("getting miner session: %w", err)
|
|
}
|
|
|
|
waitQuietCh := func() chan struct{} {
|
|
out := make(chan struct{})
|
|
go func() {
|
|
workerApi.LocalWorker.WaitQuiet()
|
|
close(out)
|
|
}()
|
|
return out
|
|
}
|
|
|
|
go func() {
|
|
heartbeats := time.NewTicker(stores.HeartbeatInterval)
|
|
defer heartbeats.Stop()
|
|
|
|
var redeclareStorage bool
|
|
var readyCh chan struct{}
|
|
for {
|
|
// If we're reconnecting, redeclare storage first
|
|
if redeclareStorage {
|
|
log.Info("Redeclaring local storage")
|
|
|
|
if err := localStore.Redeclare(ctx); err != nil {
|
|
log.Errorf("Redeclaring local storage failed: %+v", err)
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return // graceful shutdown
|
|
case <-heartbeats.C:
|
|
}
|
|
continue
|
|
}
|
|
}
|
|
|
|
// TODO: we could get rid of this, but that requires tracking resources for restarted tasks correctly
|
|
if readyCh == nil {
|
|
log.Info("Making sure no local tasks are running")
|
|
readyCh = waitQuietCh()
|
|
}
|
|
|
|
for {
|
|
curSession, err := nodeApi.Session(ctx)
|
|
if err != nil {
|
|
log.Errorf("heartbeat: checking remote session failed: %+v", err)
|
|
} else {
|
|
if curSession != minerSession {
|
|
minerSession = curSession
|
|
break
|
|
}
|
|
}
|
|
|
|
select {
|
|
case <-readyCh:
|
|
if err := nodeApi.WorkerConnect(ctx, "http://"+address+"/rpc/v0"); err != nil {
|
|
log.Errorf("Registering worker failed: %+v", err)
|
|
cancel()
|
|
return
|
|
}
|
|
|
|
log.Info("Worker registered successfully, waiting for tasks")
|
|
|
|
readyCh = nil
|
|
case <-heartbeats.C:
|
|
case <-ctx.Done():
|
|
return // graceful shutdown
|
|
}
|
|
}
|
|
|
|
log.Errorf("LOTUS-MINER CONNECTION LOST")
|
|
|
|
redeclareStorage = true
|
|
}
|
|
}()
|
|
|
|
return srv.Serve(nl)
|
|
},
|
|
}
|
|
|
|
func extractRoutableIP(timeout time.Duration) (string, error) {
|
|
minerMultiAddrKey := "MINER_API_INFO"
|
|
deprecatedMinerMultiAddrKey := "STORAGE_API_INFO"
|
|
env, ok := os.LookupEnv(minerMultiAddrKey)
|
|
if !ok {
|
|
// TODO remove after deprecation period
|
|
_, ok = os.LookupEnv(deprecatedMinerMultiAddrKey)
|
|
if ok {
|
|
log.Warnf("Using a deprecated env(%s) value, please use env(%s) instead.", deprecatedMinerMultiAddrKey, minerMultiAddrKey)
|
|
}
|
|
return "", xerrors.New("MINER_API_INFO environment variable required to extract IP")
|
|
}
|
|
minerAddr := strings.Split(env, "/")
|
|
conn, err := net.DialTimeout("tcp", minerAddr[2]+":"+minerAddr[4], timeout)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer conn.Close() //nolint:errcheck
|
|
|
|
localAddr := conn.LocalAddr().(*net.TCPAddr)
|
|
|
|
return strings.Split(localAddr.IP.String(), ":")[0], nil
|
|
}
|