cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

enable libp2p options

Browse Source
This commit is contained in:
Anton Evangelatov
2021-06-30 12:55:43 +02:00
parent a8d5ae5213
commit 9ec7cd7239
23 changed files with 760 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/impl/common/common.go
+16 -186
View File
@@ -2,32 +2,18 @@ package common
import (
"context"
"sort"
"strings"
"github.com/gbrlsnchs/jwt/v3"
"github.com/google/uuid"
logging "github.com/ipfs/go-log/v2"
"go.uber.org/fx"
"golang.org/x/xerrors"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/host"
metrics "github.com/libp2p/go-libp2p-core/metrics"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
protocol "github.com/libp2p/go-libp2p-core/protocol"
swarm "github.com/libp2p/go-libp2p-swarm"
basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
"github.com/libp2p/go-libp2p/p2p/net/conngater"
ma "github.com/multiformats/go-multiaddr"
"github.com/filecoin-project/go-jsonrpc/auth"
"github.com/filecoin-project/lotus/api"
apitypes "github.com/filecoin-project/lotus/api/types"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/lp2p"
)
var session = uuid.New()
@@ -35,13 +21,9 @@ var session = uuid.New()
type CommonAPI struct {
fx.In
NetAPI
APISecret *dtypes.APIAlg
RawHost lp2p.RawHost
Host host.Host
Router lp2p.BaseIpfsRouting
ConnGater *conngater.BasicConnectionGater
Reporter metrics.Reporter
Sk *dtypes.ScoreKeeper
ShutdownChan dtypes.ShutdownChan
}
@@ -66,170 +48,6 @@ func (a *CommonAPI) AuthNew(ctx context.Context, perms []auth.Permission) ([]byt
return jwt.Sign(&p, (*jwt.HMACSHA)(a.APISecret))
}
func (a *CommonAPI) NetConnectedness(ctx context.Context, pid peer.ID) (network.Connectedness, error) {
return a.Host.Network().Connectedness(pid), nil
}
func (a *CommonAPI) NetPubsubScores(context.Context) ([]api.PubsubScore, error) {
scores := a.Sk.Get()
out := make([]api.PubsubScore, len(scores))
i := 0
for k, v := range scores {
out[i] = api.PubsubScore{ID: k, Score: v}
i++
}
sort.Slice(out, func(i, j int) bool {
return strings.Compare(string(out[i].ID), string(out[j].ID)) > 0
})
return out, nil
}
func (a *CommonAPI) 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 *CommonAPI) NetPeerInfo(_ context.Context, p peer.ID) (*api.ExtendedPeerInfo, error) {
info := &api.ExtendedPeerInfo{ID: p}
agent, err := a.Host.Peerstore().Get(p, "AgentVersion")
if err == nil {
info.Agent = agent.(string)
}
for _, a := range a.Host.Peerstore().Addrs(p) {
info.Addrs = append(info.Addrs, a.String())
}
sort.Strings(info.Addrs)
protocols, err := a.Host.Peerstore().GetProtocols(p)
if err == nil {
sort.Strings(protocols)
info.Protocols = protocols
}
if cm := a.Host.ConnManager().GetTagInfo(p); cm != nil {
info.ConnMgrMeta = &api.ConnMgrInfo{
FirstSeen: cm.FirstSeen,
Value: cm.Value,
Tags: cm.Tags,
Conns: cm.Conns,
}
}
return info, nil
}
func (a *CommonAPI) NetConnect(ctx context.Context, p peer.AddrInfo) error {
if swrm, ok := a.Host.Network().(*swarm.Swarm); ok {
swrm.Backoff().Clear(p.ID)
}
return a.Host.Connect(ctx, p)
}
func (a *CommonAPI) NetAddrsListen(context.Context) (peer.AddrInfo, error) {
return peer.AddrInfo{
ID: a.Host.ID(),
Addrs: a.Host.Addrs(),
}, nil
}
func (a *CommonAPI) NetDisconnect(ctx context.Context, p peer.ID) error {
return a.Host.Network().ClosePeer(p)
}
func (a *CommonAPI) NetFindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo, error) {
return a.Router.FindPeer(ctx, p)
}
func (a *CommonAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error) {
autonat := a.RawHost.(*basichost.BasicHost).GetAutoNat()
if autonat == nil {
return api.NatInfo{
Reachability: network.ReachabilityUnknown,
}, nil
}
var maddr string
if autonat.Status() == network.ReachabilityPublic {
pa, err := autonat.PublicAddr()
if err != nil {
return api.NatInfo{}, err
}
maddr = pa.String()
}
return api.NatInfo{
Reachability: autonat.Status(),
PublicAddr: maddr,
}, nil
}
func (a *CommonAPI) NetAgentVersion(ctx context.Context, p peer.ID) (string, error) {
ag, err := a.Host.Peerstore().Get(p, "AgentVersion")
if err != nil {
return "", err
}
if ag == nil {
return "unknown", nil
}
return ag.(string), nil
}
func (a *CommonAPI) NetBandwidthStats(ctx context.Context) (metrics.Stats, error) {
return a.Reporter.GetBandwidthTotals(), nil
}
func (a *CommonAPI) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
out := make(map[string]metrics.Stats)
for p, s := range a.Reporter.GetBandwidthByPeer() {
out[p.String()] = s
}
return out, nil
}
func (a *CommonAPI) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
return a.Reporter.GetBandwidthByProtocol(), nil
}
func (a *CommonAPI) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error) {
return build.OpenRPCDiscoverJSON_Full(), nil
}
func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
return a.Host.ID(), nil
}
func (a *CommonAPI) Version(context.Context) (api.APIVersion, error) {
v, err := api.VersionForType(api.RunningNodeType)
if err != nil {
return api.APIVersion{}, err
}
return api.APIVersion{
Version: build.UserVersion(),
APIVersion: v,
BlockDelay: build.BlockDelaySecs,
}, nil
}
func (a *CommonAPI) LogList(context.Context) ([]string, error) {
return logging.GetSubsystems(), nil
}
@@ -251,4 +69,16 @@ func (a *CommonAPI) Closing(ctx context.Context) (<-chan struct{}, error) {
return make(chan struct{}), nil // relies on jsonrpc closing
}
var _ api.Common = &CommonAPI{}
func (a *CommonAPI) Version(context.Context) (api.APIVersion, error) {
v, err := api.VersionForType(api.RunningNodeType)
if err != nil {
return api.APIVersion{}, err
}
return api.APIVersion{
Version: build.UserVersion(),
APIVersion: v,
BlockDelay: build.BlockDelaySecs,
}, nil
}
node/impl/common/mock/mock.go
+101
View File
@@ -0,0 +1,101 @@
package mock
import (
"context"
"errors"
"github.com/filecoin-project/lotus/api"
apitypes "github.com/filecoin-project/lotus/api/types"
"github.com/libp2p/go-libp2p-core/metrics"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/protocol"
"go.uber.org/fx"
)
var (
errNotImplemented = errors.New("not implemented")
)
type MockNetAPI struct {
fx.In
}
func (a *MockNetAPI) NetAgentVersion(ctx context.Context, p peer.ID) (string, error) {
return "", errNotImplemented
}
func (a *MockNetAPI) NetConnectedness(ctx context.Context, pid peer.ID) (conn network.Connectedness, err error) {
err = errNotImplemented
return
}
func (a *MockNetAPI) NetPubsubScores(context.Context) ([]api.PubsubScore, error) {
return nil, errNotImplemented
}
func (a *MockNetAPI) NetPeers(context.Context) ([]peer.AddrInfo, error) {
return nil, errNotImplemented
}
func (a *MockNetAPI) NetPeerInfo(_ context.Context, p peer.ID) (*api.ExtendedPeerInfo, error) {
return nil, errNotImplemented
}
func (a *MockNetAPI) NetConnect(ctx context.Context, p peer.AddrInfo) error {
return errNotImplemented
}
func (a *MockNetAPI) NetAddrsListen(context.Context) (ai peer.AddrInfo, err error) {
err = errNotImplemented
return
}
func (a *MockNetAPI) NetDisconnect(ctx context.Context, p peer.ID) error {
return errNotImplemented
}
func (a *MockNetAPI) NetFindPeer(ctx context.Context, p peer.ID) (ai peer.AddrInfo, err error) {
err = errNotImplemented
return
}
func (a *MockNetAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error) {
err = errNotImplemented
return
}
func (a *MockNetAPI) NetBandwidthStats(ctx context.Context) (s metrics.Stats, err error) {
err = errNotImplemented
return
}
func (a *MockNetAPI) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
return nil, errNotImplemented
}
func (a *MockNetAPI) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
return nil, errNotImplemented
}
func (a *MockNetAPI) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error) {
return nil, errNotImplemented
}
func (a *MockNetAPI) ID(context.Context) (p peer.ID, err error) {
err = errNotImplemented
return
}
func (a *MockNetAPI) NetBlockAdd(ctx context.Context, acl api.NetBlockList) error {
return errNotImplemented
}
func (a *MockNetAPI) NetBlockRemove(ctx context.Context, acl api.NetBlockList) error {
return errNotImplemented
}
func (a *MockNetAPI) NetBlockList(ctx context.Context) (result api.NetBlockList, err error) {
err = errNotImplemented
return
}
node/impl/common/net.go
+34
View File
@@ -0,0 +1,34 @@
package common
import (
"context"
metrics "github.com/libp2p/go-libp2p-core/metrics"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
protocol "github.com/libp2p/go-libp2p-core/protocol"
"github.com/filecoin-project/lotus/api"
apitypes "github.com/filecoin-project/lotus/api/types"
)
type NetAPI interface {
NetConnectedness(ctx context.Context, pid peer.ID) (network.Connectedness, error)
NetPubsubScores(context.Context) ([]api.PubsubScore, error)
NetPeers(context.Context) ([]peer.AddrInfo, error)
NetPeerInfo(_ context.Context, p peer.ID) (*api.ExtendedPeerInfo, error)
NetConnect(ctx context.Context, p peer.AddrInfo) error
NetAddrsListen(context.Context) (peer.AddrInfo, error)
NetDisconnect(ctx context.Context, p peer.ID) error
NetFindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo, error)
NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error)
NetAgentVersion(ctx context.Context, p peer.ID) (string, error)
NetBandwidthStats(ctx context.Context) (metrics.Stats, error)
NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error)
NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error)
Discover(ctx context.Context) (apitypes.OpenRPCDocument, error)
ID(context.Context) (peer.ID, error)
NetBlockAdd(ctx context.Context, acl api.NetBlockList) error
NetBlockRemove(ctx context.Context, acl api.NetBlockList) error
NetBlockList(ctx context.Context) (api.NetBlockList, error)
}
node/impl/common/conngater.go → node/impl/common/net_conngater.go
+3 -3
View File
@@ -14,7 +14,7 @@ import (
var cLog = logging.Logger("conngater")
func (a *CommonAPI) NetBlockAdd(ctx context.Context, acl api.NetBlockList) error {
func (a *Libp2pNetAPI) NetBlockAdd(ctx context.Context, acl api.NetBlockList) error {
for _, p := range acl.Peers {
err := a.ConnGater.BlockPeer(p)
if err != nil {
@@ -89,7 +89,7 @@ func (a *CommonAPI) NetBlockAdd(ctx context.Context, acl api.NetBlockList) error
return nil
}
func (a *CommonAPI) NetBlockRemove(ctx context.Context, acl api.NetBlockList) error {
func (a *Libp2pNetAPI) NetBlockRemove(ctx context.Context, acl api.NetBlockList) error {
for _, p := range acl.Peers {
err := a.ConnGater.UnblockPeer(p)
if err != nil {
@@ -124,7 +124,7 @@ func (a *CommonAPI) NetBlockRemove(ctx context.Context, acl api.NetBlockList) er
return nil
}
func (a *CommonAPI) NetBlockList(ctx context.Context) (result api.NetBlockList, err error) {
func (a *Libp2pNetAPI) NetBlockList(ctx context.Context) (result api.NetBlockList, err error) {
result.Peers = a.ConnGater.ListBlockedPeers()
for _, ip := range a.ConnGater.ListBlockedAddrs() {
result.IPAddrs = append(result.IPAddrs, ip.String())
node/impl/common/net_libp2p.go
+186
View File
@@ -0,0 +1,186 @@
package common
import (
"context"
"sort"
"strings"
"github.com/libp2p/go-libp2p-core/host"
metrics "github.com/libp2p/go-libp2p-core/metrics"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
protocol "github.com/libp2p/go-libp2p-core/protocol"
swarm "github.com/libp2p/go-libp2p-swarm"
basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
"github.com/libp2p/go-libp2p/p2p/net/conngater"
ma "github.com/multiformats/go-multiaddr"
"go.uber.org/fx"
"github.com/filecoin-project/lotus/api"
apitypes "github.com/filecoin-project/lotus/api/types"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/lp2p"
)
type Libp2pNetAPI struct {
fx.In
RawHost lp2p.RawHost
Host host.Host
Router lp2p.BaseIpfsRouting
ConnGater *conngater.BasicConnectionGater
Reporter metrics.Reporter
Sk *dtypes.ScoreKeeper
}
func (a *Libp2pNetAPI) NetConnectedness(ctx context.Context, pid peer.ID) (network.Connectedness, error) {
return a.Host.Network().Connectedness(pid), nil
}
func (a *Libp2pNetAPI) NetPubsubScores(context.Context) ([]api.PubsubScore, error) {
scores := a.Sk.Get()
out := make([]api.PubsubScore, len(scores))
i := 0
for k, v := range scores {
out[i] = api.PubsubScore{ID: k, Score: v}
i++
}
sort.Slice(out, func(i, j int) bool {
return strings.Compare(string(out[i].ID), string(out[j].ID)) > 0
})
return out, nil
}
func (a *Libp2pNetAPI) 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 *Libp2pNetAPI) NetPeerInfo(_ context.Context, p peer.ID) (*api.ExtendedPeerInfo, error) {
info := &api.ExtendedPeerInfo{ID: p}
agent, err := a.Host.Peerstore().Get(p, "AgentVersion")
if err == nil {
info.Agent = agent.(string)
}
for _, a := range a.Host.Peerstore().Addrs(p) {
info.Addrs = append(info.Addrs, a.String())
}
sort.Strings(info.Addrs)
protocols, err := a.Host.Peerstore().GetProtocols(p)
if err == nil {
sort.Strings(protocols)
info.Protocols = protocols
}
if cm := a.Host.ConnManager().GetTagInfo(p); cm != nil {
info.ConnMgrMeta = &api.ConnMgrInfo{
FirstSeen: cm.FirstSeen,
Value: cm.Value,
Tags: cm.Tags,
Conns: cm.Conns,
}
}
return info, nil
}
func (a *Libp2pNetAPI) NetConnect(ctx context.Context, p peer.AddrInfo) error {
if swrm, ok := a.Host.Network().(*swarm.Swarm); ok {
swrm.Backoff().Clear(p.ID)
}
return a.Host.Connect(ctx, p)
}
func (a *Libp2pNetAPI) NetAddrsListen(context.Context) (peer.AddrInfo, error) {
return peer.AddrInfo{
ID: a.Host.ID(),
Addrs: a.Host.Addrs(),
}, nil
}
func (a *Libp2pNetAPI) NetDisconnect(ctx context.Context, p peer.ID) error {
return a.Host.Network().ClosePeer(p)
}
func (a *Libp2pNetAPI) NetFindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo, error) {
return a.Router.FindPeer(ctx, p)
}
func (a *Libp2pNetAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error) {
autonat := a.RawHost.(*basichost.BasicHost).GetAutoNat()
if autonat == nil {
return api.NatInfo{
Reachability: network.ReachabilityUnknown,
}, nil
}
var maddr string
if autonat.Status() == network.ReachabilityPublic {
pa, err := autonat.PublicAddr()
if err != nil {
return api.NatInfo{}, err
}
maddr = pa.String()
}
return api.NatInfo{
Reachability: autonat.Status(),
PublicAddr: maddr,
}, nil
}
func (a *Libp2pNetAPI) NetAgentVersion(ctx context.Context, p peer.ID) (string, error) {
ag, err := a.Host.Peerstore().Get(p, "AgentVersion")
if err != nil {
return "", err
}
if ag == nil {
return "unknown", nil
}
return ag.(string), nil
}
func (a *Libp2pNetAPI) NetBandwidthStats(ctx context.Context) (metrics.Stats, error) {
return a.Reporter.GetBandwidthTotals(), nil
}
func (a *Libp2pNetAPI) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
out := make(map[string]metrics.Stats)
for p, s := range a.Reporter.GetBandwidthByPeer() {
out[p.String()] = s
}
return out, nil
}
func (a *Libp2pNetAPI) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
return a.Reporter.GetBandwidthByProtocol(), nil
}
func (a *Libp2pNetAPI) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error) {
return build.OpenRPCDiscoverJSON_Full(), nil
}
func (a *Libp2pNetAPI) ID(context.Context) (peer.ID, error) {
return a.Host.ID(), nil
}
node/impl/storminer.go
+1 -2
View File
@@ -15,7 +15,6 @@ import (
"github.com/filecoin-project/lotus/build"
"github.com/google/uuid"
"github.com/ipfs/go-cid"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/peer"
"golang.org/x/xerrors"
@@ -59,6 +58,7 @@ type StorageMinerAPI struct {
DataTransfer dtypes.ProviderDataTransfer `optional:"true"`
DealPublisher *storageadapter.DealPublisher `optional:"true"`
SectorBlocks *sectorblocks.SectorBlocks `optional:"true"`
Host dtypes.Libp2pHost `optional:"true"`
// Miner / storage
Miner *storage.Miner `optional:"true"`
@@ -67,7 +67,6 @@ type StorageMinerAPI struct {
IStorageMgr sectorstorage.SectorManager `optional:"true"`
stores.SectorIndex
storiface.WorkerReturn `optional:"true"`
Host host.Host
AddrSel *storage.AddressSelector
Epp gen.WinningPoStProver `optional:"true"`