fix: gsoc via pushsync

pkg/check/gsoc/gsoc.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ethersphere/beekeeper/pkg/beekeeper"
 	"github.com/ethersphere/beekeeper/pkg/logging"
 	"github.com/ethersphere/beekeeper/pkg/orchestration"
+	"github.com/ethersphere/beekeeper/pkg/random"
 	"github.com/ethersphere/beekeeper/pkg/wslistener"
 	"golang.org/x/sync/errgroup"
 )
@@ -65,18 +66,21 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, opts int
 		return fmt.Errorf("invalid options type")
 	}
 
-	fullNodeNames := cluster.FullNodeNames()
-	clients, err := cluster.NodesClients(ctx)
+	rnd := random.PseudoGenerator(time.Now().UnixNano())
+	fullNodeClients, err := cluster.ShuffledFullNodeClients(ctx, rnd)
 	if err != nil {
 		return err
 	}
 
-	if len(fullNodeNames) < 2 {
+	if len(fullNodeClients) < 2 {
 		return fmt.Errorf("gsoc test require at least 2 full nodes")
 	}
 
-	uploadClient := clients[fullNodeNames[0]]
-	listenClient := clients[fullNodeNames[1]]
+	uploadClient := fullNodeClients[0]
+	listenClient, err := cluster.ClosetFullNodeClient(ctx, uploadClient, rnd)
+	if err != nil {
+		return err
+	}
 
 	batches := make([]string, 2)
 	for i := 0; i < 2; i++ {
@@ -92,14 +96,14 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, opts int
 	c.logger.Infof("send messages with different postage batches sequentially...")
 	err = run(ctx, uploadClient, listenClient, batches, c.logger, false)
 	if err != nil {
-		return err
+		return fmt.Errorf("sequential: %w", err)
 	}
 	c.logger.Infof("done")
 
 	c.logger.Infof("send messages with different postage batches parallel...")
 	err = run(ctx, uploadClient, listenClient, batches, c.logger, true)
 	if err != nil {
-		return err
+		return fmt.Errorf("parallel: %w", err)
 	}
 	c.logger.Infof("done")
 

pkg/orchestration/cluster.go

@@ -36,6 +36,7 @@ type Cluster interface {
 	ShuffledFullNodeClients(ctx context.Context, r *rand.Rand) ([]*bee.Client, error)
 	Size() (size int)
 	Topologies(ctx context.Context) (topologies ClusterTopologies, err error)
+	ClosetFullNodeClient(ctx context.Context, s *bee.Client, r *rand.Rand) (*bee.Client, error)
 }
 
 // ClusterOptions represents Bee cluster options

pkg/orchestration/k8s/cluster.go

@@ -443,3 +443,43 @@ func (c *Cluster) FlattenTopologies(ctx context.Context) (topologies map[string]
 
 	return
 }
+
+// ClosetFullNodeClient returns the closest full node client to the supplied node.
+func (c *Cluster) ClosetFullNodeClient(ctx context.Context, s *bee.Client, r *rand.Rand) (*bee.Client, error) {
+	addrToNode := make(map[string]orchestration.Node)
+	for _, n := range c.Nodes() {
+		res, err := n.Client().Addresses(ctx)
+		if err != nil {
+			return nil, err
+		}
+		addrToNode[res.Overlay.String()] = n
+	}
+
+	t, err := s.Topology(ctx)
+	if err != nil {
+		return nil, err
+	}
+	const maxBin = 32
+	for b := range maxBin {
+		bin := t.Bins[fmt.Sprintf("bin_%d", b)]
+		var fullNodes []orchestration.Node
+		for _, peer := range bin.ConnectedPeers {
+			node, ok := addrToNode[peer.Address.String()]
+			if !ok {
+				return nil, fmt.Errorf("peer overlay %s not found in address map", peer.Address.String())
+			}
+			cfg := node.Config()
+			if cfg.FullNode && !cfg.BootnodeMode {
+				fullNodes = append(fullNodes, node)
+			}
+		}
+		if len(fullNodes) == 0 {
+			continue
+		}
+		r.Shuffle(len(fullNodes), func(i, j int) {
+			fullNodes[i], fullNodes[j] = fullNodes[j], fullNodes[i]
+		})
+		return fullNodes[0].Client(), nil
+	}
+	return nil, fmt.Errorf("cannot find closest fullnode")
+}