Merge pull request #9 from filecoin-project/feat/netcli

Network CLI
2019-07-09 12:05:56 -04:00 · 2019-07-09 12:05:56 -04:00 · b843dcb590
commit b843dcb590
parent 83d1df15b4 271c268e28
15 changed files with 383 additions and 39 deletions
--- a/api/api.go
+++ b/api/api.go
@ -34,9 +34,10 @@ type API interface {

 	// network

-	// // peers
+	NetPeers(context.Context) ([]peer.AddrInfo, error) // TODO: check serialization
+	NetConnect(context.Context, peer.AddrInfo) error
+	NetAddrsListen(context.Context) (MultiaddrSlice, error)
 	// // ping
-	// // connect

 	// Struct

--- a/api/struct.go
+++ b/api/struct.go
@ -11,9 +11,25 @@ type Struct struct {
 	Internal struct {
 		ID      func(context.Context) (peer.ID, error)
 		Version func(context.Context) (Version, error)
+
+		NetPeers       func(context.Context) ([]peer.AddrInfo, error)
+		NetConnect     func(context.Context, peer.AddrInfo) error
+		NetAddrsListen func(context.Context) (MultiaddrSlice, error)
 	}
 }

+func (c *Struct) NetPeers(ctx context.Context) ([]peer.AddrInfo, error) {
+	return c.Internal.NetPeers(ctx)
+}
+
+func (c *Struct) NetConnect(ctx context.Context, p peer.AddrInfo) error {
+	return c.Internal.NetConnect(ctx, p)
+}
+
+func (c *Struct) NetAddrsListen(ctx context.Context) (MultiaddrSlice, error) {
+	return c.Internal.NetAddrsListen(ctx)
+}
+
 // ID implements API.ID
 func (c *Struct) ID(ctx context.Context) (peer.ID, error) {
 	return c.Internal.ID(ctx)
--- a/api/types.go
+++ b/api/types.go
@ -0,0 +1,28 @@
+package api
+
+import (
+	"encoding/json"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+// TODO: check if this exists anywhere else
+type MultiaddrSlice []ma.Multiaddr
+
+func (m *MultiaddrSlice) UnmarshalJSON(raw []byte) (err error) {
+	var temp []string
+	if err := json.Unmarshal(raw, &temp); err != nil {
+		return err
+	}
+
+	res := make([]ma.Multiaddr, len(temp))
+	for i, str := range temp {
+		res[i], err = ma.NewMultiaddr(str)
+		if err != nil {
+			return err
+		}
+	}
+	*m = res
+	return nil
+}
+
+var _ json.Unmarshaler = new(MultiaddrSlice)
--- a/cli/cmd.go
+++ b/cli/cmd.go
@ -1,10 +1,48 @@
 package cli

 import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/filecoin-project/go-lotus/api"
 	"gopkg.in/urfave/cli.v2"
 )

-// Commands is the root group of CLI commands
+const (
+	metadataContext = "context"
+	metadataAPI     = "api"
+)
+
+// ApiConnector returns API instance
+type ApiConnector func() api.API
+
+func getApi(ctx *cli.Context) api.API {
+	return ctx.App.Metadata[metadataAPI].(ApiConnector)()
+}
+
+// reqContext returns context for cli execution. Calling it for the first time
+// installs SIGTERM handler that will close returned context.
+// Not safe for concurrent execution.
+func reqContext(cctx *cli.Context) context.Context {
+	if uctx, ok := cctx.App.Metadata[metadataContext]; ok {
+		// unchecked cast as if something else is in there
+		// it is crash worthy either way
+		return uctx.(context.Context)
+	}
+	ctx, done := context.WithCancel(context.Background())
+	sigChan := make(chan os.Signal, 2)
+	go func() {
+		<-sigChan
+		done()
+	}()
+	signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)
+
+	return ctx
+}
+
 var Commands = []*cli.Command{
+	netCmd,
 	versionCmd,
 }
--- a/cli/net.go
+++ b/cli/net.go
@ -0,0 +1,164 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+	ma "github.com/multiformats/go-multiaddr"
+	madns "github.com/multiformats/go-multiaddr-dns"
+	"gopkg.in/urfave/cli.v2"
+)
+
+var netCmd = &cli.Command{
+	Name:  "net",
+	Usage: "Manage P2P Network",
+	Subcommands: []*cli.Command{
+		netPeers,
+		netConnect,
+		netListen,
+	},
+}
+
+var netPeers = &cli.Command{
+	Name:  "peers",
+	Usage: "Print peers",
+	Action: func(cctx *cli.Context) error {
+		api := getApi(cctx)
+		ctx := reqContext(cctx)
+		peers, err := api.NetPeers(ctx)
+		if err != nil {
+			return err
+		}
+
+		for _, peer := range peers {
+			fmt.Println(peer)
+		}
+
+		return nil
+	},
+}
+
+var netListen = &cli.Command{
+	Name:  "listen",
+	Usage: "List listen addresses",
+	Action: func(cctx *cli.Context) error {
+		api := getApi(cctx)
+		ctx := reqContext(cctx)
+
+		addrs, err := api.NetAddrsListen(ctx)
+		if err != nil {
+			return err
+		}
+
+		for _, peer := range addrs {
+			fmt.Println(peer)
+		}
+		return nil
+	},
+}
+
+var netConnect = &cli.Command{
+	Name:  "connect",
+	Usage: "Connect to a peer",
+	Action: func(cctx *cli.Context) error {
+		api := getApi(cctx)
+		ctx := reqContext(cctx)
+
+		pis, err := parseAddresses(ctx, cctx.Args().Slice())
+		if err != nil {
+			return err
+		}
+
+		for _, pi := range pis {
+			fmt.Printf("connect %s: ", pi.ID.Pretty())
+			err := api.NetConnect(ctx, pi)
+			if err != nil {
+				fmt.Println("failure")
+				return err
+			}
+			fmt.Println("success")
+		}
+
+		return nil
+	},
+}
+
+// parseAddresses is a function that takes in a slice of string peer addresses
+// (multiaddr + peerid) and returns a slice of properly constructed peers
+func parseAddresses(ctx context.Context, addrs []string) ([]peer.AddrInfo, error) {
+	// resolve addresses
+	maddrs, err := resolveAddresses(ctx, addrs)
+	if err != nil {
+		return nil, err
+	}
+
+	return peer.AddrInfosFromP2pAddrs(maddrs...)
+}
+
+const (
+	dnsResolveTimeout = 10 * time.Second
+)
+
+// resolveAddresses resolves addresses parallelly
+func resolveAddresses(ctx context.Context, addrs []string) ([]ma.Multiaddr, error) {
+	ctx, cancel := context.WithTimeout(ctx, dnsResolveTimeout)
+	defer cancel()
+
+	var maddrs []ma.Multiaddr
+	var wg sync.WaitGroup
+	resolveErrC := make(chan error, len(addrs))
+
+	maddrC := make(chan ma.Multiaddr)
+
+	for _, addr := range addrs {
+		maddr, err := ma.NewMultiaddr(addr)
+		if err != nil {
+			return nil, err
+		}
+
+		// check whether address ends in `ipfs/Qm...`
+		if _, last := ma.SplitLast(maddr); last.Protocol().Code == ma.P_IPFS {
+			maddrs = append(maddrs, maddr)
+			continue
+		}
+		wg.Add(1)
+		go func(maddr ma.Multiaddr) {
+			defer wg.Done()
+			raddrs, err := madns.Resolve(ctx, maddr)
+			if err != nil {
+				resolveErrC <- err
+				return
+			}
+			// filter out addresses that still doesn't end in `ipfs/Qm...`
+			found := 0
+			for _, raddr := range raddrs {
+				if _, last := ma.SplitLast(raddr); last != nil && last.Protocol().Code == ma.P_IPFS {
+					maddrC <- raddr
+					found++
+				}
+			}
+			if found == 0 {
+				resolveErrC <- fmt.Errorf("found no ipfs peers at %s", maddr)
+			}
+		}(maddr)
+	}
+	go func() {
+		wg.Wait()
+		close(maddrC)
+	}()
+
+	for maddr := range maddrC {
+		maddrs = append(maddrs, maddr)
+	}
+
+	select {
+	case err := <-resolveErrC:
+		return nil, err
+	default:
+	}
+
+	return maddrs, nil
+}
--- a/cmd/lotus/main.go
+++ b/cmd/lotus/main.go
@ -6,6 +6,8 @@ import (

 	"gopkg.in/urfave/cli.v2"

+	"github.com/filecoin-project/go-lotus/api"
+	"github.com/filecoin-project/go-lotus/api/client"
 	"github.com/filecoin-project/go-lotus/build"
 	lcli "github.com/filecoin-project/go-lotus/cli"
 	"github.com/filecoin-project/go-lotus/daemon"
@ -20,6 +22,12 @@ func main() {
 		Name:    "lotus",
 		Usage:   "Filecoin decentralized storage network client",
 		Version: build.Version,
+		Metadata: map[string]interface{}{
+			"api": lcli.ApiConnector(func() api.API {
+				// TODO: get this from repo
+				return client.NewRPC("http://127.0.0.1:1234/rpc/v0")
+			}),
+		},

 		Commands: append(local, lcli.Commands...),
 	}
--- a/daemon/cmd.go
+++ b/daemon/cmd.go
@ -1,3 +1,5 @@
+// +build !nodaemon
+
 package daemon

 import (
@ -13,6 +15,12 @@ import (
 var Cmd = &cli.Command{
 	Name:  "daemon",
 	Usage: "Start a lotus daemon process",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:  "api",
+			Value: ":1234",
+		},
+	},
 	Action: func(cctx *cli.Context) error {
 		ctx := context.Background()

@ -26,6 +34,6 @@ var Cmd = &cli.Command{
 			return err
 		}

-		return serveRPC(api)
+		return serveRPC(api, cctx.String("api"))
 	},
 }
--- a/daemon/cmd_nodaemon.go
+++ b/daemon/cmd_nodaemon.go
@ -0,0 +1,24 @@
+// +build nodaemon
+
+package daemon
+
+import (
+	"errors"
+
+	"gopkg.in/urfave/cli.v2"
+)
+
+// Cmd is the `go-lotus daemon` command
+var Cmd = &cli.Command{
+	Name:  "daemon",
+	Usage: "Start a lotus daemon process",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:  "api",
+			Value: ":1234",
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		return errors.New("daemon support not included in this binary")
+	},
+}
--- a/daemon/rpc.go
+++ b/daemon/rpc.go
@ -7,9 +7,9 @@ import (
 	"github.com/filecoin-project/go-lotus/lib/jsonrpc"
 )

-func serveRPC(api api.API) error {
+func serveRPC(api api.API, addr string) error {
 	rpcServer := jsonrpc.NewServer()
 	rpcServer.Register("Filecoin", api)
 	http.Handle("/rpc/v0", rpcServer)
-	return http.ListenAndServe(":1234", http.DefaultServeMux)
+	return http.ListenAndServe(addr, http.DefaultServeMux)
 }
--- a/go.mod
+++ b/go.mod
@ -16,7 +16,7 @@ require (
 	github.com/ipfs/go-ipfs-routing v0.1.0
 	github.com/ipfs/go-ipld-cbor v0.0.2
 	github.com/ipfs/go-ipld-format v0.0.2
-	github.com/ipfs/go-log v0.0.2-0.20190703113630-0c3cfb1eccc4
+	github.com/ipfs/go-log v0.0.2-0.20190708183747-9c9fd6111bea
 	github.com/ipfs/go-merkledag v0.0.2
 	github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52
 	github.com/libp2p/go-libp2p v0.2.0
@ -38,6 +38,7 @@ require (
 	github.com/libp2p/go-maddr-filter v0.0.4
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1
 	github.com/multiformats/go-multiaddr v0.0.4
+	github.com/multiformats/go-multiaddr-dns v0.0.2
 	github.com/multiformats/go-multihash v0.0.5
 	github.com/pkg/errors v0.8.1
 	github.com/polydawn/refmt v0.0.0-20190408063855-01bf1e26dd14
--- a/go.sum
+++ b/go.sum
@ -140,8 +140,8 @@ github.com/ipfs/go-ipld-format v0.0.2 h1:OVAGlyYT6JPZ0pEfGntFPS40lfrDmaDbQwNHEY2
 github.com/ipfs/go-ipld-format v0.0.2/go.mod h1:4B6+FM2u9OJ9zCV+kSbgFAZlOrv1Hqbf0INGQgiKf9k=
 github.com/ipfs/go-log v0.0.1 h1:9XTUN/rW64BCG1YhPK9Hoy3q8nr4gOmHHBpgFdfw6Lc=
 github.com/ipfs/go-log v0.0.1/go.mod h1:kL1d2/hzSpI0thNYjiKfjanbVNU+IIGA/WnNESY9leM=
-github.com/ipfs/go-log v0.0.2-0.20190703113630-0c3cfb1eccc4 h1:4GUopYwyu/8kX0UxYB7QDYOUbnt9HCg/j6J0sMqVvNQ=
-github.com/ipfs/go-log v0.0.2-0.20190703113630-0c3cfb1eccc4/go.mod h1:YTiqro5xwLoGra88hB8tMBlN+7ByaT3Kdaa0UqwCmI0=
+github.com/ipfs/go-log v0.0.2-0.20190708183747-9c9fd6111bea h1:T5Xz37N8fqK3u0aLbWXCoTOO+S62T+TvOLRZh8q8vX0=
+github.com/ipfs/go-log v0.0.2-0.20190708183747-9c9fd6111bea/go.mod h1:YTiqro5xwLoGra88hB8tMBlN+7ByaT3Kdaa0UqwCmI0=
 github.com/ipfs/go-merkledag v0.0.2 h1:U3Q74RLOwpbtERjCv/MODC99qSxHBw33ZeMfiGXl7ts=
 github.com/ipfs/go-merkledag v0.0.2/go.mod h1:CRdtHMROECqaehAGeJ0Wd9TtlmWv/ta5cUnvbTnniEI=
 github.com/ipfs/go-metrics-interface v0.0.1 h1:j+cpbjYvu4R8zbleSs36gvB7jR+wsL2fGD6n0jO4kdg=
@ -333,9 +333,11 @@ github.com/marten-seemann/qtls v0.2.3 h1:0yWJ43C62LsZt08vuQJDK1uC1czUc3FJeCLPoNA
 github.com/marten-seemann/qtls v0.2.3/go.mod h1:xzjG7avBwGGbdZ8dTGxlBnLArsVKLvwmjgmPuiQEcYk=
 github.com/mattn/go-colorable v0.1.1 h1:G1f5SKeVxmagw/IyvzvtZE4Gybcc4Tr1tf7I8z0XgOg=
 github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
+github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx8mU=
 github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
 github.com/mattn/go-isatty v0.0.5 h1:tHXDdz1cpzGaovsTB+TVB8q90WEokoVmfMqoVcrLUgw=
 github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.8 h1:HLtExJ+uU2HOZ+wI0Tt5DtUDrx8yhUqDcp7fYERX4CE=
 github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/miekg/dns v1.1.4/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/miekg/dns v1.1.12 h1:WMhc1ik4LNkTg8U9l3hI1LvxKmIL+f1+WV/SZtCbDDA=
@ -473,6 +475,7 @@ golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190501004415-9ce7a6920f09/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190611141213-3f473d35a33a h1:+KkCgOMgnKSgenxTBoiwkMqTiouMIy/3o8RLdmSbGoY=
 golang.org/x/net v0.0.0-20190611141213-3f473d35a33a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
--- a/lib/jsonrpc/rpc_client.go
+++ b/lib/jsonrpc/rpc_client.go
@ -6,11 +6,18 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"reflect"
 	"sync/atomic"
+
+	logging "github.com/ipfs/go-log"
 )

+var log = logging.Logger("rpc")
+
+const clientDebug = true
+
 var (
 	errorType   = reflect.TypeOf(new(error)).Elem()
 	contextType = reflect.TypeOf(new(context.Context)).Elem()
@ -33,7 +40,9 @@ func (e *ErrClient) Unwrap(err error) error {
 type result reflect.Value

 func (r *result) UnmarshalJSON(raw []byte) error {
-	return json.Unmarshal(raw, reflect.Value(*r).Interface())
+	err := json.Unmarshal(raw, reflect.Value(*r).Interface())
+	log.Debugw("rpc unmarshal response", "raw", string(raw), "err", err)
+	return err
 }

 type clientResponse struct {
@ -148,8 +157,23 @@ func NewClient(addr string, namespace string, handler interface{}) ClientCloser

 			// process response

+			if clientDebug {
+				rsp, err := ioutil.ReadAll(httpResp.Body)
+				if err != nil {
+					return processError(err)
+				}
+				if err := httpResp.Body.Close(); err != nil {
+					return processError(err)
+				}
+
+				log.Debugw("rpc response", "body", string(rsp))
+
+				httpResp.Body = ioutil.NopCloser(bytes.NewReader(rsp))
+			}
+
 			var resp clientResponse
 			if valOut != -1 {
+				log.Debugw("rpc result", "type", ftyp.Out(valOut))
 				resp.Result = result(reflect.New(ftyp.Out(valOut)))
 			}

--- a/node/api.go
+++ b/node/api.go
@ -0,0 +1,52 @@
+package node
+
+import (
+	"context"
+
+	"github.com/filecoin-project/go-lotus/api"
+	"github.com/filecoin-project/go-lotus/build"
+
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+type API struct {
+	Host host.Host
+}
+
+func (a *API) ID(context.Context) (peer.ID, error) {
+	return a.Host.ID(), nil
+}
+
+func (a *API) Version(context.Context) (api.Version, error) {
+	return api.Version{
+		Version: build.Version,
+	}, nil
+}
+
+func (a *API) NetPeers(context.Context) ([]peer.AddrInfo, error) {
+	conns := a.Host.Network().Conns()
+	out := make([]peer.AddrInfo, len(conns))
+
+	for i, conn := range conns {
+		out[i] = peer.AddrInfo{
+			ID: conn.RemotePeer(),
+			Addrs: []ma.Multiaddr{
+				conn.RemoteMultiaddr(),
+			},
+		}
+	}
+
+	return out, nil
+}
+
+func (a *API) NetConnect(ctx context.Context, p peer.AddrInfo) error {
+	return a.Host.Connect(ctx, p)
+}
+
+func (a *API) NetAddrsListen(context.Context) (api.MultiaddrSlice, error) {
+	return a.Host.Addrs(), nil
+}
+
+var _ api.API = &API{}
--- a/node/builder.go
+++ b/node/builder.go
@ -20,7 +20,6 @@ import (
 	"go.uber.org/fx"

 	"github.com/filecoin-project/go-lotus/api"
-	"github.com/filecoin-project/go-lotus/build"
 	"github.com/filecoin-project/go-lotus/chain"
 	"github.com/filecoin-project/go-lotus/node/config"
 	"github.com/filecoin-project/go-lotus/node/hello"
@ -149,7 +148,7 @@ func Online() Option {
 		Override(BaseRoutingKey, lp2p.BaseRouting),
 		Override(new(routing.Routing), lp2p.Routing),

-		Override(NatPortMapKey, lp2p.NatPortMap),
+		//Override(NatPortMapKey, lp2p.NatPortMap), //TODO: reenable when closing logic is actually there
 		Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second)),

 		Override(new(*pubsub.PubSub), lp2p.GossipSub()),
@ -194,7 +193,7 @@ func Config(cfg *config.Root) Option {

 // New builds and starts new Filecoin node
 func New(ctx context.Context, opts ...Option) (api.API, error) {
-	var resAPI api.Struct
+	resAPI := &API{}
 	settings := settings{
 		modules: map[interface{}]fx.Option{},
 		invokes: make([]fx.Option, _nInvokes),
@ -222,8 +221,7 @@ func New(ctx context.Context, opts ...Option) (api.API, error) {
 		fx.Options(ctors...),
 		fx.Options(settings.invokes...),

-		fx.Invoke(versionAPI(&resAPI.Internal.Version)),
-		fx.Invoke(idAPI(&resAPI.Internal.ID)),
+		fx.Extract(resAPI),
 	)

 	// TODO: we probably should have a 'firewall' for Closing signal
@ -233,7 +231,7 @@ func New(ctx context.Context, opts ...Option) (api.API, error) {
 		return nil, err
 	}

-	return &resAPI, nil
+	return resAPI, nil
 }

 // In-memory / testing
@ -250,24 +248,3 @@ func randomIdentity() Option {
 		Override(new(peer.ID), peer.IDFromPublicKey),
 	)
 }
-
-// API IMPL
-
-// TODO: figure out a better way, this isn't usable in long term
-func idAPI(set *func(ctx context.Context) (peer.ID, error)) func(id peer.ID) {
-	return func(id peer.ID) {
-		*set = func(ctx context.Context) (peer.ID, error) {
-			return id, nil
-		}
-	}
-}
-
-func versionAPI(set *func(context.Context) (api.Version, error)) func() {
-	return func() {
-		*set = func(context.Context) (api.Version, error) {
-			return api.Version{
-				Version: build.Version,
-			}, nil
-		}
-	}
-}
--- a/node/config/def.go
+++ b/node/config/def.go
@ -28,8 +28,8 @@ func Default() *Root {
 		},
 		Libp2p: Libp2p{
 			ListenAddresses: []string{
-				"/ip4/0.0.0.0/tcp/4001",
-				"/ip6/::/tcp/4001",
+				"/ip4/0.0.0.0/tcp/0",
+				"/ip6/::/tcp/0",
 			},
 		},
 	}