test: adds the ability to run txsim as knuu instance for testground s…

…anity test (#3244)

Closes #3189

Summary of changes:
- Introduces the capability to customize the resource allocation for
each Node and TxSimNode individually
- Enables running txsim as a knuu instance. 
- Allows spinning up multiple instances of txsim (e.g., one or more per
validator)
- Enhanced logs

I have also tested all the e2e tests, and they pass successfully.

---------

Co-authored-by: Callum Waters <[email protected]>

test/e2e/defaults.go

@@ -0,0 +1,8 @@
+package e2e
+
+var defaultResources = Resources{
+	MemoryRequest: "200Mi",
+	MemoryLimit:   "200Mi",
+	CPU:           "300m",
+	Volume:        "1Gi",
+}

test/e2e/node.go

@@ -8,6 +8,7 @@ import (
 	"github.com/celestiaorg/celestia-app/v2/test/util/genesis"
 	"github.com/celestiaorg/knuu/pkg/knuu"
 	serverconfig "github.com/cosmos/cosmos-sdk/server/config"
+	"github.com/rs/zerolog/log"
 	"github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/crypto"
 	"github.com/tendermint/tendermint/p2p"
@@ -25,6 +26,7 @@ const (
 	secp256k1Type  = "secp256k1"
 	ed25519Type    = "ed25519"
 	remoteRootDir  = "/home/celestia/.celestia-app"
+	txsimRootDir   = "/home/celestia"
 )
 
 type Node struct {
@@ -41,12 +43,25 @@ type Node struct {
 	grpcProxyPort int
 }
 
+// Resources defines the resource requirements for a Node.
+type Resources struct {
+	// MemoryRequest specifies the initial memory allocation for the Node.
+	MemoryRequest string
+	// MemoryLimit specifies the maximum memory allocation for the Node.
+	MemoryLimit string
+	// CPU specifies the CPU allocation for the Node.
+	CPU string
+	// Volume specifies the storage volume allocation for the Node.
+	Volume string
+}
+
 func NewNode(
 	name, version string,
 	startHeight, selfDelegation int64,
 	peers []string,
 	signerKey, networkKey crypto.PrivKey,
 	upgradeHeight int64,
+	resources Resources,
 	grafana *GrafanaInfo,
 ) (*Node, error) {
 	instance, err := knuu.NewInstance(name)
@@ -82,11 +97,11 @@ func NewNode(
 			return nil, fmt.Errorf("error setting jaeger exporter: %v", err)
 		}
 	}
-	err = instance.SetMemory("200Mi", "200Mi")
+	err = instance.SetMemory(resources.MemoryRequest, resources.MemoryLimit)
 	if err != nil {
 		return nil, err
 	}
-	err = instance.SetCPU("300m")
+	err = instance.SetCPU(resources.CPU)
 	if err != nil {
 		return nil, err
 	}
@@ -124,6 +139,9 @@ func (n *Node) Init(genesis *types.GenesisDoc, peers []string) error {
 	// Initialize file directories
 	rootDir := os.TempDir()
 	nodeDir := filepath.Join(rootDir, n.Name)
+	log.Info().Str("name", n.Name).
+		Str("directory", nodeDir).
+		Msg("Creating validator's config and data directories")
 	for _, dir := range []string{
 		filepath.Join(nodeDir, "config"),
 		filepath.Join(nodeDir, "data"),
@@ -213,6 +231,24 @@ func (n Node) AddressGRPC() string {
 	return fmt.Sprintf("127.0.0.1:%d", n.grpcProxyPort)
 }
 
+// RemoteAddressGRPC retrieves the gRPC endpoint address of a node within the cluster.
+func (n Node) RemoteAddressGRPC() (string, error) {
+	ip, err := n.Instance.GetIP()
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s:%d", ip, grpcPort), nil
+}
+
+// RemoteAddressRPC retrieves the RPC endpoint address of a node within the cluster.
+func (n Node) RemoteAddressRPC() (string, error) {
+	ip, err := n.Instance.GetIP()
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s:%d", ip, rpcPort), nil
+}
+
 func (n Node) IsValidator() bool {
 	return n.SelfDelegation != 0
 }

test/e2e/setup.go

@@ -18,8 +18,10 @@ func MakeConfig(node *Node) (*config.Config, error) {
 	cfg.RPC.ListenAddress = "tcp://0.0.0.0:26657"
 	cfg.P2P.ExternalAddress = fmt.Sprintf("tcp://%v", node.AddressP2P(false))
 	cfg.P2P.PersistentPeers = strings.Join(node.InitialPeers, ",")
-	cfg.Consensus.TimeoutPropose = time.Second
-	cfg.Consensus.TimeoutCommit = time.Second
+	cfg.P2P.SendRate = 5 * 1024 * 1024 // 5MiB/s
+	cfg.P2P.RecvRate = 5 * 1024 * 1024 // 5MiB/s
+	cfg.Consensus.TimeoutPropose = 1 * time.Second
+	cfg.Consensus.TimeoutCommit = 1 * time.Second
 	cfg.Instrumentation.Prometheus = true
 	return cfg, nil
 }

test/e2e/simple_test.go

@@ -46,14 +46,21 @@ func TestE2ESimple(t *testing.T) {
 	testnet, err := New(t.Name(), seed, GetGrafanaInfoFromEnvVar())
 	require.NoError(t, err)
 	t.Cleanup(testnet.Cleanup)
-	require.NoError(t, testnet.CreateGenesisNodes(4, latestVersion, 10000000, 0))
 
-	kr, err := testnet.CreateAccount("alice", 1e12)
+	t.Log("Creating testnet validators")
+	require.NoError(t, testnet.CreateGenesisNodes(4, latestVersion, 10000000,
+		0, defaultResources))
+
+	t.Log("Creating account")
+	kr, err := testnet.CreateAccount("alice", 1e12, "")
 	require.NoError(t, err)
 
+	t.Log("Setting up testnet")
 	require.NoError(t, testnet.Setup())
+	t.Log("Starting testnet")
 	require.NoError(t, testnet.Start())
 
+	t.Log("Running txsim")
 	sequences := txsim.NewBlobSequence(txsim.NewRange(200, 4000), txsim.NewRange(1, 3)).Clone(5)
 	sequences = append(sequences, txsim.NewSendSequence(4, 1000, 100).Clone(5)...)
 
@@ -64,6 +71,7 @@ func TestE2ESimple(t *testing.T) {
 	err = txsim.Run(ctx, testnet.GRPCEndpoints()[0], kr, encCfg, opts, sequences...)
 	require.True(t, errors.Is(err, context.DeadlineExceeded), err.Error())
 
+	t.Log("Reading blockchain")
 	blockchain, err := testnode.ReadBlockchain(context.Background(), testnet.Node(0).AddressRPC())
 	require.NoError(t, err)