feat: add peers proxy

Author: hunjixin

.gitignore

@@ -8,6 +8,7 @@ badgerStAskDb/
 *.puml
 market-client
 venus-market
+market-proxy
 sequence_chart.md
 .idea
 .vscode

cmd/market-proxy/main.go

@@ -4,10 +4,17 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/hex"
+	"fmt"
 	"log"
 	"os"
+	"time"
+
+	datatransfer "github.com/filecoin-project/go-data-transfer"
+
+	network2 "github.com/filecoin-project/venus-market/v2/network"
 
 	"github.com/filecoin-project/venus-market/v2/config"
+	"github.com/filecoin-project/venus-market/v2/utils"
 
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/urfave/cli/v2"
@@ -44,57 +51,104 @@ var protocols = []protocol.ID{
 
 	network.ProtocolGraphsync_1_0_0,
 	network.ProtocolGraphsync_2_0_0,
+	datatransfer.ProtocolDataTransfer1_2,
 }
 
 var mainLog = logging.Logger("market-proxy")
 
 func main() {
 	app := &cli.App{
-		Name:                 "venus-market",
-		Usage:                "venus-market",
+		Name:                 "venus-market-proxy",
+		Usage:                "proxy multiple venus-market backends like nginx",
 		Version:              version.UserVersion(),
 		EnableBashCompletion: true,
+		Commands: []*cli.Command{
+			{
+				Name:  "run",
+				Usage: "start a libp2p proxy",
+				Flags: []cli.Flag{
+					&cli.StringFlag{
+						Name:  "listen",
+						Usage: "specify listen address ",
+						Value: "/ip4/0.0.0.0/tcp/11023",
+					},
+					&cli.StringFlag{
+						Name:  "peer-key",
+						Usage: "peer key for p2p identify, if not specify, will generate new one",
+						Value: "",
+					},
+					&cli.StringSliceFlag{
+						Name:     "backends",
+						Usage:    "a group of backends libp2p backends server",
+						Required: true,
+					},
+				},
+				Action: run,
+			},
+			{
+				Name:   "new-peer-key",
+				Usage:  "generate random peer key and corresponding private key ",
+				Flags:  []cli.Flag{},
+				Action: genKey,
+			},
+		},
 	}
 
 	if err := app.Run(os.Args); err != nil {
 		log.Fatal(err)
 	}
 }
 
-type ProxyConfig struct {
-	ProxyServer string
-}
+func run(c *cli.Context) error {
+	ctx := c.Context
+	utils.SetupLogLevels()
+	cfg := config.ProxyServer{}
+	cfg.Libp2p.ListenAddresses = []string{c.String("listen")}
+	cfg.Libp2p.PrivateKey = c.String("peer-key")
+	cfg.Backends = c.StringSlice("backends")
 
-func run(cfg *config.ProxyServer) error {
 	var pkey crypto.PrivKey
 	var err error
-	if len(cfg.PrivateKey) == 0 {
+	if len(cfg.PrivateKey) > 0 {
 		privateKeyBytes, err := hex.DecodeString(cfg.PrivateKey)
 		if err != nil {
 			return err
 		}
-		crypto.UnmarshalPrivateKey(privateKeyBytes)
+		pkey, err = crypto.UnmarshalPrivateKey(privateKeyBytes)
+		if err != nil {
+			return err
+		}
 	} else {
 		pkey, _, err = crypto.GenerateEd25519Key(rand.Reader)
 		if err != nil {
 			return err
 		}
+
+		privateKeyBytes, err := crypto.MarshalPrivateKey(pkey)
+		if err != nil {
+			return err
+		}
+
+		fmt.Println(hex.EncodeToString(privateKeyBytes))
 	}
 
 	opts := []libp2p.Option{
+		network2.MakeSmuxTransportOption(),
+		libp2p.DefaultTransports,
 		libp2p.ListenAddrStrings(cfg.ListenAddresses...),
 		libp2p.Identity(pkey),
+		libp2p.WithDialTimeout(time.Second * 5),
 		libp2p.DefaultPeerstore,
-		libp2p.NoListenAddrs,
+		libp2p.DisableRelay(),
 		libp2p.Ping(true),
+
 		libp2p.UserAgent("venus-market-proxy" + version.UserVersion()),
 	}
 
 	h, err := libp2p.New(opts...)
 	if err != nil {
 		return err
 	}
-
 	addrInfo, err := peer.AddrInfoFromString(cfg.Backends[0])
 	if err != nil {
 		return err
@@ -105,8 +159,51 @@ func run(cfg *config.ProxyServer) error {
 	if err != nil {
 		return err
 	}
+
 	defer proxyHost.Close()
 
 	proxyHost.Start(context.Background())
+
+	//try to connect backends
+	go func() {
+		timer := time.NewTicker(time.Minute)
+		defer timer.Stop()
+		for {
+			err = h.Connect(ctx, *addrInfo)
+			if err != nil {
+				mainLog.Errorf("connect to %s %v", addrInfo, err)
+			}
+			<-timer.C
+		}
+	}()
+
+	var libp2pAddrs []string
+	for _, peer := range h.Addrs() {
+		libp2pAddrs = append(libp2pAddrs, fmt.Sprintf("%s/p2p/%s\n", peer, h.ID()))
+	}
+
+	mainLog.Infof("start listen at %v", libp2pAddrs)
+	shutdownChan := make(chan struct{})
+	finishCh := utils.MonitorShutdown(shutdownChan)
+	<-finishCh
+	return nil
+}
+
+func genKey(c *cli.Context) error {
+	pkey, _, err := crypto.GenerateEd25519Key(rand.Reader)
+	if err != nil {
+		return err
+	}
+
+	privateKeyBytes, err := crypto.MarshalPrivateKey(pkey)
+	if err != nil {
+		return err
+	}
+	peerId, err := peer.IDFromPrivateKey(pkey)
+	if err != nil {
+		return err
+	}
+	fmt.Println("PeerId:", peerId)
+	fmt.Println("Pkey:", hex.EncodeToString(privateKeyBytes))
 	return nil
 }

network/host.go

@@ -51,7 +51,7 @@ func Host(mctx metrics.MetricsCtx, lc fx.Lifecycle, params P2PHostIn) (host.Host
 		return nil, err
 	}
 
-	if len(params.Cfg.Proxy) == 0 {
+	if len(params.Cfg.Proxy) > 0 {
 		addrInfo, err := peer.AddrInfoFromString(params.Cfg.Proxy)
 		if err != nil {
 			return nil, err

network/smux.go

@@ -7,7 +7,7 @@ import (
 	yamux "github.com/libp2p/go-libp2p/p2p/muxer/yamux"
 )
 
-func makeSmuxTransportOption() libp2p.Option {
+func MakeSmuxTransportOption() libp2p.Option {
 	const yamuxID = "/yamux/1.0.0"
 
 	ymxtpt := *yamux.DefaultTransport
@@ -22,7 +22,7 @@ func makeSmuxTransportOption() libp2p.Option {
 
 func SmuxTransport() func() (opts Libp2pOpts, err error) {
 	return func() (opts Libp2pOpts, err error) {
-		opts.Opts = append(opts.Opts, makeSmuxTransportOption())
+		opts.Opts = append(opts.Opts, MakeSmuxTransportOption())
 		return
 	}
 }

protocolproxy/README.md

@@ -0,0 +1 @@
+Code inspired by boost project

protocolproxy/forwardinghost.go

@@ -47,6 +47,7 @@ func (fh *ForwardingHost) Close() error {
 func (fh *ForwardingHost) SetStreamHandler(pid protocol.ID, handler network.StreamHandler) {
 	fh.Host.SetStreamHandler(pid, handler)
 
+	fmt.Println("protocol", pid)
 	// Save the handler so it can be invoked from the forwarding protocol's handler
 	// only set the handler if we are successful in registering the route
 	fh.handlersLk.Lock()
@@ -141,6 +142,7 @@ func (fh *ForwardingHost) validateForwardingRequest(request *messages.Forwarding
 // Calls to "NewStream" open an outbound forwarding request to the proxy, that is then sent on
 // the the specified peer
 func (fh *ForwardingHost) NewStream(ctx context.Context, p peer.ID, protocols ...protocol.ID) (network.Stream, error) {
+	log.Infow("ForwardingHost receive stream")
 	// If there is a direct connection to the peer (or there was a connection
 	// recently) open the stream over the direct connection
 	if p != fh.proxy {

protocolproxy/protocolproxy.go

@@ -192,39 +192,6 @@ func (pp *ProtocolProxy) processForwardingRequest(p peer.ID, remote peer.ID, pro
 	return s, nil
 }
 
-// pipe a stream through the PP
-func (pp *ProtocolProxy) bridgeStreams(s1, s2 network.Stream) {
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		// pipe reads on s1 to writes on s2
-		defer wg.Done()
-		_, err := io.Copy(s2, s1)
-		if err != nil {
-			_ = s1.Reset()
-			return
-		}
-		err = s2.CloseWrite()
-		if err != nil {
-			_ = s1.Reset()
-		}
-	}()
-	go func() {
-		// pipe reads on s2 to writes on s1
-		defer wg.Done()
-		_, err := io.Copy(s1, s2)
-		if err != nil {
-			_ = s2.Reset()
-			return
-		}
-		err = s1.CloseWrite()
-		if err != nil {
-			_ = s2.Reset()
-		}
-	}()
-	wg.Wait()
-}
-
 func (pp *ProtocolProxy) handleIncoming(s network.Stream) {
 	defer s.Close()
 
@@ -262,3 +229,39 @@ func (pp *ProtocolProxy) handleIncoming(s network.Stream) {
 
 	pp.bridgeStreams(s, routedStream)
 }
+
+// pipe a stream through the PP
+func (pp *ProtocolProxy) bridgeStreams(s1, s2 network.Stream) {
+	defer func() {
+		fmt.Println("exit stream")
+	}()
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		// pipe reads on s1 to writes on s2
+		defer wg.Done()
+		_, err := io.Copy(s2, s1)
+		if err != nil {
+			_ = s1.Reset()
+			return
+		}
+		err = s2.CloseWrite()
+		if err != nil {
+			_ = s1.Reset()
+		}
+	}()
+	go func() {
+		// pipe reads on s2 to writes on s1
+		defer wg.Done()
+		_, err := io.Copy(s1, s2)
+		if err != nil {
+			_ = s2.Reset()
+			return
+		}
+		err = s1.CloseWrite()
+		if err != nil {
+			_ = s2.Reset()
+		}
+	}()
+	wg.Wait()
+}

storageprovider/stream.go

@@ -78,6 +78,8 @@ func (storageDealStream *StorageDealStream) HandleAskStream(s network.StorageAsk
 			log.Errorf("unable to close err %v", err)
 		}
 	}()
+
+	log.Infof("receive ask stream")
 	ar, err := s.ReadAskRequest()
 	if err != nil {
 		log.Errorf("failed to read AskRequest from incoming stream: %s", err)