[Merged by Bors] - Add peer info api

api/grpcserver/admin_service.go

@@ -31,14 +31,16 @@
 	logger  log.Logger
 	db      *sql.Database
 	dataDir string
+	p       peers
 }
 
 // NewAdminService creates a new admin grpc service.
-func NewAdminService(db *sql.Database, dataDir string, lg log.Logger) *AdminService {
+func NewAdminService(db *sql.Database, dataDir string, lg log.Logger, p peers) *AdminService {
 	return &AdminService{
 		logger:  lg,
 		db:      db,
 		dataDir: dataDir,
+		p:       p,
 	}
 }
 
@@ -128,3 +130,37 @@
 		}
 	}
 }
+
+func (a AdminService) PeerInfoStream(_ *empty.Empty, stream pb.AdminService_PeerInfoStreamServer) error {
+	for _, p := range a.p.GetPeers() {
+		select {
+		case <-stream.Context().Done():
+			return nil
+		default:
+			info := a.p.ConnectedPeerInfo(p)
+			// There is no guarantee that the peers originally returned will still
+			// be connected by the time we call ConnectedPeerInfo.
+			if info == nil {
+				continue
+			}
+			connections := make([]*pb.ConnectionInfo, len(info.Connections))
+			for j, c := range info.Connections {
+				connections[j] = &pb.ConnectionInfo{
+					Address:  c.Address.String(),
+					Uptime:   c.Uptime.String(),
+					Outbound: c.Outbound,
+				}
+			}
+			err := stream.Send(&pb.PeerInfo{
+				Id:          info.ID.String(),
+				Connections: connections,
+				Tags:        info.Tags,
+			})
+			if err != nil {
+				return fmt.Errorf("send to stream: %w", err)
+			}
+		}
+	}
+
+	return nil
+}

api/grpcserver/admin_service_test.go

@@ -56,7 +56,7 @@ func createMesh(tb testing.TB, db *sql.Database) {
 func TestAdminService_Checkpoint(t *testing.T) {
 	db := sql.InMemory()
 	createMesh(t, db)
-	svc := NewAdminService(db, t.TempDir(), logtest.New(t))
+	svc := NewAdminService(db, t.TempDir(), logtest.New(t), nil)
 	t.Cleanup(launchServer(t, cfg, svc))
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@@ -91,7 +91,7 @@ func TestAdminService_Checkpoint(t *testing.T) {
 
 func TestAdminService_CheckpointError(t *testing.T) {
 	db := sql.InMemory()
-	svc := NewAdminService(db, t.TempDir(), logtest.New(t))
+	svc := NewAdminService(db, t.TempDir(), logtest.New(t), nil)
 	t.Cleanup(launchServer(t, cfg, svc))
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

api/grpcserver/grpc.go

@@ -18,9 +18,14 @@ type ServiceAPI interface {
 
 // Server is a very basic grpc server.
 type Server struct {
-	Listener   string
-	logger     log.Logger
-	GrpcServer *grpc.Server
+	Listener string
+	logger   log.Logger
+	// BoundAddress contains the address that the server bound to, useful if
+	// the server uses a dynamic port. It is set during startup and can be
+	// safely accessed after Start has completed (I.E. the returned channel has
+	// been waited on)
+	BoundAddress string
+	GrpcServer   *grpc.Server
 }
 
 // New creates and returns a new Server with port and interface.
@@ -58,6 +63,7 @@ func (s *Server) startInternal(started chan<- struct{}) {
 		s.logger.Error("error listening: %v", err)
 		return
 	}
+	s.BoundAddress = lis.Addr().String()
 	reflection.Register(s.GrpcServer)
 	s.logger.Info("starting new grpc server on %s", s.Listener)
 	close(started)

api/grpcserver/interface.go

@@ -60,6 +60,12 @@ type peerCounter interface {
 	PeerCount() uint64
 }
 
+// Peers is an api to get peer related info.
+type peers interface {
+	ConnectedPeerInfo(p2p.Peer) *p2p.PeerInfo
+	GetPeers() []p2p.Peer
+}
+
 // genesisTimeAPI is an API to get genesis time and current layer of the system.
 type genesisTimeAPI interface {
 	GenesisTime() time.Time

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/pyroscope-io/pyroscope v0.37.2
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/seehuhn/mt19937 v1.0.0
-	github.com/spacemeshos/api/release/go v1.19.0
+	github.com/spacemeshos/api/release/go v1.19.1-0.20230818155305-f006d101d548
 	github.com/spacemeshos/economics v0.1.0
 	github.com/spacemeshos/fixed v0.1.0
 	github.com/spacemeshos/go-scale v1.1.10

go.sum

@@ -606,8 +606,8 @@ github.com/smartystreets/goconvey v1.7.2 h1:9RBaZCeXEQ3UselpuwUQHltGVXvdwm6cv1hg
 github.com/smartystreets/goconvey v1.7.2/go.mod h1:Vw0tHAZW6lzCRk3xgdin6fKYcG+G3Pg9vgXWeJpQFMM=
 github.com/sourcegraph/annotate v0.0.0-20160123013949-f4cad6c6324d/go.mod h1:UdhH50NIW0fCiwBSr0co2m7BnFLdv4fQTgdqdJTHFeE=
 github.com/sourcegraph/syntaxhighlight v0.0.0-20170531221838-bd320f5d308e/go.mod h1:HuIsMU8RRBOtsCgI77wP899iHVBQpCmg4ErYMZB+2IA=
-github.com/spacemeshos/api/release/go v1.19.0 h1:QPt1nUuSVQ4DfZPNsSAuJpjjYlbS9HJhDunE7K8rn08=
-github.com/spacemeshos/api/release/go v1.19.0/go.mod h1:nC5g7IVRNxF8Kl/Tzvr7cQeYBwHN4sMTVsIEa6Ebtl4=
+github.com/spacemeshos/api/release/go v1.19.1-0.20230818155305-f006d101d548 h1:ilyQXo1rPTLjR0X3hRtP8BsIq8bGPUT7iSOz7wEEcIM=
+github.com/spacemeshos/api/release/go v1.19.1-0.20230818155305-f006d101d548/go.mod h1:nC5g7IVRNxF8Kl/Tzvr7cQeYBwHN4sMTVsIEa6Ebtl4=
 github.com/spacemeshos/economics v0.1.0 h1:PJAKbhBKqbbdCYTB29pkmc8sYqK3pKUAiuAvQxuSJEg=
 github.com/spacemeshos/economics v0.1.0/go.mod h1:Bz0wRDwCOUP1A6w3cPW6iuUBGME8Tz48sIriYiohsBg=
 github.com/spacemeshos/fixed v0.1.0 h1:20KIGvxLlAsuidQrvuwwHe6PrvqeTKzbJIsScbmnUPQ=

node/adminservice_api_test.go

@@ -0,0 +1,78 @@
+package node
+
+import (
+	"context"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/libp2p/go-libp2p/core/peer"
+	pb "github.com/spacemeshos/api/release/go/spacemesh/v1"
+	"github.com/stretchr/testify/require"
+
+	"github.com/spacemeshos/go-spacemesh/api/grpcserver"
+	"github.com/spacemeshos/go-spacemesh/config"
+	"github.com/spacemeshos/go-spacemesh/log/logtest"
+)
+
+func TestPeerInfoApi(t *testing.T) {
+	cfg := config.DefaultTestConfig()
+	cfg.P2P.DisableNatPort = true
+	cfg.P2P.Listen = "/ip4/127.0.0.1/tcp/0"
+
+	cfg.API.PublicListener = "0.0.0.0:0"
+	cfg.API.PrivateServices = nil
+	cfg.API.PublicServices = []string{grpcserver.Admin}
+	l := logtest.New(t)
+	networkSize := 3
+	network := NewTestNetwork(t, cfg, l, networkSize)
+	infos := make([][]*pb.PeerInfo, networkSize)
+	for i, app := range network {
+		adminapi := pb.NewAdminServiceClient(app.Conn)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+		defer cancel()
+
+		streamClient, err := adminapi.PeerInfoStream(ctx, &empty.Empty{})
+		require.NoError(t, err)
+		for {
+			info, err := streamClient.Recv()
+			if err == io.EOF {
+				break
+			}
+			require.NoError(t, err)
+			infos[i] = append(infos[i], info)
+		}
+	}
+
+	for i, app := range network {
+		for j, innerApp := range network {
+			if j == i {
+				continue
+			}
+			peers := infos[i]
+			require.Len(t, peers, networkSize-1, "expecting each node to have connections to all other nodes")
+			peer := getPeerInfo(peers, innerApp.host.ID())
+			require.NotNil(t, peer, "info is missing connection to %v")
+			require.Len(t, peer.Connections, 1, "expecting only 1 connection to each peer")
+			require.Equal(t, innerApp.host.Addrs()[0].String(), peer.Connections[0].Address, "connection address should match address of peer")
+			outbound := peer.Connections[0].Outbound
+
+			// Check that outbound matches with the other side of the connection
+			otherSide := getPeerInfo(infos[j], app.host.ID())
+			require.NotNil(t, peer, "one side missing peer connection")
+			require.Len(t, otherSide.Connections, 1, "expecting only 1 connection to each peer")
+			require.Equal(t, outbound, !otherSide.Connections[0].Outbound, "expecting pairwise connections to agree on outbound direction")
+		}
+	}
+}
+
+func getPeerInfo(peers []*pb.PeerInfo, id peer.ID) *pb.PeerInfo {
+	str := id.String()
+	for _, p := range peers {
+		if str == p.Id {
+			return p
+		}
+	}
+	return nil
+}

node/bad_peer_test.go

@@ -14,30 +14,31 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 	varint "github.com/multiformats/go-varint"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/spacemeshos/go-spacemesh/codec"
 	"github.com/spacemeshos/go-spacemesh/common/types"
 	"github.com/spacemeshos/go-spacemesh/config"
-	"github.com/spacemeshos/go-spacemesh/events"
-	"github.com/spacemeshos/go-spacemesh/log"
+	"github.com/spacemeshos/go-spacemesh/log/logtest"
 	ps "github.com/spacemeshos/go-spacemesh/p2p/pubsub"
-	"github.com/spacemeshos/go-spacemesh/signing"
 )
 
 func TestPeerDisconnectForMessageResultValidationReject(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	l := logtest.New(t)
+
 	// Make 2 node instances
 	conf1 := config.DefaultTestConfig()
 	conf1.DataDirParent = t.TempDir()
 	conf1.FileLock = filepath.Join(conf1.DataDirParent, "LOCK")
 	conf1.P2P.Listen = "/ip4/127.0.0.1/tcp/0"
-	app1, err := NewApp(&conf1)
-	require.NoError(t, err)
+	// We setup the api to listen on an OS assigned port, which avoids the second instance getting stuck when
+	conf1.API.PublicListener = "0.0.0.0:0"
+	conf1.API.PrivateListener = "0.0.0.0:0"
+	app1 := NewApp(t, &conf1, l)
+
 	conf2 := config.DefaultTestConfig()
 	// We need to copy the genesis config to ensure that both nodes share the
 	// same gnenesis ID, otherwise they will not be able to connect to each
@@ -46,8 +47,9 @@ func TestPeerDisconnectForMessageResultValidationReject(t *testing.T) {
 	conf2.DataDirParent = t.TempDir()
 	conf2.FileLock = filepath.Join(conf2.DataDirParent, "LOCK")
 	conf2.P2P.Listen = "/ip4/127.0.0.1/tcp/0"
-	app2, err := NewApp(&conf2)
-	require.NoError(t, err)
+	conf2.API.PublicListener = "0.0.0.0:0"
+	conf2.API.PrivateListener = "0.0.0.0:0"
+	app2 := NewApp(t, &conf2, l)
 
 	types.SetLayersPerEpoch(conf1.LayersPerEpoch)
 	t.Cleanup(func() {
@@ -65,7 +67,7 @@ func TestPeerDisconnectForMessageResultValidationReject(t *testing.T) {
 	<-app2.Started()
 
 	// Connect app2 to app1
-	err = app2.Host().Connect(context.Background(), peer.AddrInfo{
+	err := app2.Host().Connect(context.Background(), peer.AddrInfo{
 		ID:    app1.Host().ID(),
 		Addrs: app1.Host().Addrs(),
 	})
@@ -122,26 +124,6 @@ func TestPeerDisconnectForMessageResultValidationReject(t *testing.T) {
 	require.NoError(t, g.Wait())
 }
 
-func NewApp(conf *config.Config) (*App, error) {
-	app := New(
-		WithConfig(conf),
-		WithLog(log.RegisterHooks(
-			log.NewWithLevel("", zap.NewAtomicLevelAt(zapcore.DebugLevel)),
-			events.EventHook()),
-		),
-	)
-
-	var err error
-	if err = app.Initialize(); err != nil {
-		return nil, err
-	}
-	app.edSgn, err = signing.NewEdSigner()
-	if err != nil {
-		return nil, err
-	}
-	return app, err
-}
-
 func getStream(c network.Conn, p protocol.ID, dir network.Direction) network.Stream {
 	for _, s := range c.GetStreams() {
 		if s.Protocol() == p && s.Stat().Direction == dir {

node/node.go

@@ -1045,7 +1045,7 @@ func (app *App) initService(ctx context.Context, svc grpcserver.Service) (grpcse
 	case grpcserver.Node:
 		return grpcserver.NewNodeService(app.host, app.mesh, app.clock, app.syncer, cmd.Version, cmd.Commit, logger.WithName("Node")), nil
 	case grpcserver.Admin:
-		return grpcserver.NewAdminService(app.db, app.Config.DataDir(), logger.WithName("Admin")), nil
+		return grpcserver.NewAdminService(app.db, app.Config.DataDir(), logger.WithName("Admin"), app.host), nil
 	case grpcserver.Smesher:
 		return grpcserver.NewSmesherService(app.postSetupMgr, app.atxBuilder, app.Config.API.SmesherStreamInterval, app.Config.SMESHING.Opts, logger.WithName("Smesher")), nil
 	case grpcserver.Transaction:
@@ -1109,10 +1109,10 @@ func (app *App) startAPIServices(ctx context.Context) error {
 		app.jsonAPIService.StartService(ctx, public...)
 	}
 	if app.grpcPublicService != nil {
-		app.grpcPublicService.Start()
+		<-app.grpcPublicService.Start()
 	}
 	if app.grpcPrivateService != nil {
-		app.grpcPrivateService.Start()
+		<-app.grpcPrivateService.Start()
 	}
 	return nil
 }