lotus/node/rpc.go

197 lines
5.3 KiB
Go
Raw Normal View History

package node
2019-07-18 23:18:26 +00:00
import (
2019-09-17 14:23:08 +00:00
"context"
2019-10-23 09:18:22 +00:00
"encoding/json"
2020-10-21 08:37:50 +00:00
"net"
2019-07-18 23:18:26 +00:00
"net/http"
2019-09-30 23:25:45 +00:00
_ "net/http/pprof"
"runtime"
"strconv"
2019-07-18 23:18:26 +00:00
"github.com/gorilla/mux"
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
"github.com/multiformats/go-multiaddr"
manet "github.com/multiformats/go-multiaddr/net"
2020-10-21 08:37:50 +00:00
"go.opencensus.io/tag"
"golang.org/x/xerrors"
2020-05-20 17:43:22 +00:00
"github.com/filecoin-project/go-jsonrpc"
2020-05-20 18:23:51 +00:00
"github.com/filecoin-project/go-jsonrpc/auth"
2020-05-20 17:43:22 +00:00
2021-03-25 14:09:50 +00:00
"github.com/filecoin-project/lotus/api"
2021-03-23 15:40:22 +00:00
"github.com/filecoin-project/lotus/api/v0api"
"github.com/filecoin-project/lotus/api/v1api"
2021-06-14 09:27:54 +00:00
"github.com/filecoin-project/lotus/lib/rpcenc"
2020-10-21 08:10:27 +00:00
"github.com/filecoin-project/lotus/metrics"
2020-05-20 17:43:22 +00:00
"github.com/filecoin-project/lotus/node/impl"
2019-07-18 23:18:26 +00:00
)
var rpclog = logging.Logger("rpc")
2019-09-17 14:23:08 +00:00
// ServeRPC serves an HTTP handler over the supplied listen multiaddr.
2021-05-23 11:40:33 +00:00
//
// This function spawns a goroutine to run the server, and returns immediately.
2021-05-23 11:40:33 +00:00
// It returns the stop function to be called to terminate the endpoint.
//
// The supplied ID is used in tracing, by inserting a tag in the context.
func ServeRPC(h http.Handler, id string, addr multiaddr.Multiaddr) (StopFunc, error) {
// Start listening to the addr; if invalid or occupied, we will fail early.
lst, err := manet.Listen(addr)
if err != nil {
return nil, xerrors.Errorf("could not listen: %w", err)
}
// Instantiate the server and start listening.
srv := &http.Server{
Handler: h,
BaseContext: func(listener net.Listener) context.Context {
ctx, _ := tag.New(context.Background(), tag.Upsert(metrics.APIInterface, id))
return ctx
},
2020-11-17 13:39:53 +00:00
}
go func() {
err = srv.Serve(manet.NetListener(lst))
if err != http.ErrServerClosed {
rpclog.Warnf("rpc server failed: %s", err)
}
}()
return srv.Shutdown, err
}
// FullNodeHandler returns a full node handler, to be mounted as-is on the server.
func FullNodeHandler(a v1api.FullNode, permissioned bool, opts ...jsonrpc.ServerOption) (http.Handler, error) {
m := mux.NewRouter()
2021-03-23 15:40:22 +00:00
serveRpc := func(path string, hnd interface{}) {
rpcServer := jsonrpc.NewServer(opts...)
2021-03-23 15:40:22 +00:00
rpcServer.Register("Filecoin", hnd)
2019-07-24 13:40:42 +00:00
var handler http.Handler = rpcServer
if permissioned {
handler = &auth.Handler{Verify: a.AuthVerify, Next: rpcServer.ServeHTTP}
2021-03-23 15:40:22 +00:00
}
m.Handle(path, handler)
2019-07-24 13:40:42 +00:00
}
fnapi := metrics.MetricedFullAPI(a)
if permissioned {
fnapi = api.PermissionedFullAPI(fnapi)
}
2021-03-23 15:40:22 +00:00
serveRpc("/rpc/v1", fnapi)
serveRpc("/rpc/v0", &v0api.WrapperV1Full{FullNode: fnapi})
2019-09-17 14:23:08 +00:00
// Import handler
handleImportFunc := handleImport(a.(*impl.FullNodeAPI))
if permissioned {
importAH := &auth.Handler{
Verify: a.AuthVerify,
Next: handleImportFunc,
}
m.Handle("/rest/v0/import", importAH)
} else {
m.HandleFunc("/rest/v0/import", handleImportFunc)
2019-10-23 09:18:22 +00:00
}
// debugging
m.Handle("/debug/metrics", metrics.Exporter())
m.Handle("/debug/pprof-set/block", handleFractionOpt("BlockProfileRate", runtime.SetBlockProfileRate))
m.Handle("/debug/pprof-set/mutex", handleFractionOpt("MutexProfileFraction", func(x int) {
runtime.SetMutexProfileFraction(x)
}))
m.PathPrefix("/").Handler(http.DefaultServeMux) // pprof
2020-02-23 22:54:49 +00:00
return m, nil
}
// MinerHandler returns a miner handler, to be mounted as-is on the server.
func MinerHandler(a api.StorageMiner, permissioned bool) (http.Handler, error) {
m := mux.NewRouter()
2019-09-17 14:23:08 +00:00
mapi := metrics.MetricedStorMinerAPI(a)
if permissioned {
mapi = api.PermissionedStorMinerAPI(mapi)
}
2021-06-14 09:27:54 +00:00
readerHandler, readerServerOpt := rpcenc.ReaderParamDecoder()
rpcServer := jsonrpc.NewServer(readerServerOpt)
rpcServer.Register("Filecoin", mapi)
2019-09-17 14:23:08 +00:00
m.Handle("/rpc/v0", rpcServer)
2021-06-14 09:27:54 +00:00
m.Handle("/rpc/streams/v0/push/{uuid}", readerHandler)
m.PathPrefix("/remote").HandlerFunc(a.(*impl.StorageMinerAPI).ServeRemote)
// debugging
m.Handle("/debug/metrics", metrics.Exporter())
m.PathPrefix("/").Handler(http.DefaultServeMux) // pprof
if !permissioned {
2021-06-18 16:40:33 +00:00
return m, nil
}
ah := &auth.Handler{
Verify: a.AuthVerify,
Next: m.ServeHTTP,
}
return ah, nil
2019-07-18 23:18:26 +00:00
}
2019-10-23 09:18:22 +00:00
func handleImport(a *impl.FullNodeAPI) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
if r.Method != "PUT" {
w.WriteHeader(404)
return
}
2021-03-25 14:09:50 +00:00
if !auth.HasPerm(r.Context(), nil, api.PermWrite) {
2019-10-23 09:18:22 +00:00
w.WriteHeader(401)
_ = json.NewEncoder(w).Encode(struct{ Error string }{"unauthorized: missing write permission"})
2019-10-23 09:18:22 +00:00
return
}
c, err := a.ClientImportLocal(r.Context(), r.Body)
if err != nil {
w.WriteHeader(500)
_ = json.NewEncoder(w).Encode(struct{ Error string }{err.Error()})
2019-10-23 09:18:22 +00:00
return
}
w.WriteHeader(200)
err = json.NewEncoder(w).Encode(struct{ Cid cid.Cid }{c})
if err != nil {
rpclog.Errorf("/rest/v0/import: Writing response failed: %+v", err)
2019-10-23 09:18:22 +00:00
return
}
}
}
func handleFractionOpt(name string, setter func(int)) http.HandlerFunc {
return func(rw http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(rw, "only POST allowed", http.StatusMethodNotAllowed)
return
}
if err := r.ParseForm(); err != nil {
http.Error(rw, err.Error(), http.StatusBadRequest)
return
}
asfr := r.Form.Get("x")
if len(asfr) == 0 {
http.Error(rw, "parameter 'x' must be set", http.StatusBadRequest)
return
}
fr, err := strconv.Atoi(asfr)
if err != nil {
http.Error(rw, err.Error(), http.StatusBadRequest)
return
}
rpclog.Infof("setting %s to %d", name, fr)
setter(fr)
}
}