test: testground sanity check with knuu

Makefile

@@ -212,3 +212,4 @@ prebuilt-binary:
 govulncheck:
 	@go run golang.org/x/vuln/cmd/[email protected] ./...
 .PHONY: govulncheck
+

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

@@ -24,6 +24,7 @@ const (
 	secp256k1Type = "secp256k1"
 	ed25519Type   = "ed25519"
 	remoteRootDir = "/home/celestia/.celestia-app"
+	txsimRootDir  = "/home/celestia"
 )
 
 type Node struct {
@@ -41,12 +42,20 @@ type Node struct {
 	grpcProxyPort int
 }
 
+type Resources struct {
+	memoryRequest string
+	memoryLimit   string
+	cpu           string
+	volume        string
+}
+
 func NewNode(
 	name, version string,
 	startHeight, selfDelegation int64,
 	peers []string,
 	signerKey, networkKey, accountKey crypto.PrivKey,
 	upgradeHeight int64,
+	resources Resources,
 ) (*Node, error) {
 	instance, err := knuu.NewInstance(name)
 	if err != nil {
@@ -68,11 +77,11 @@ func NewNode(
 	if err := instance.AddPortUDP(metricsPort); err != nil {
 		return nil, 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
 	}

test/e2e/setup.go

@@ -138,6 +138,8 @@ 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.P2P.SendRate = 5 * 1024 * 1024 // 5MiB/s
+	cfg.P2P.RecvRate = 5 * 1024 * 1024 // 5MiB/s
 	cfg.Consensus.TimeoutPropose = time.Second
 	cfg.Consensus.TimeoutCommit = time.Second
 	cfg.Instrumentation.Prometheus = true

test/e2e/simple_test.go

@@ -46,7 +46,8 @@ func TestE2ESimple(t *testing.T) {
 	testnet, err := New(t.Name(), seed)
 	require.NoError(t, err)
 	t.Cleanup(testnet.Cleanup)
-	require.NoError(t, testnet.CreateGenesisNodes(4, latestVersion, 10000000, 0))
+	require.NoError(t, testnet.CreateGenesisNodes(4, latestVersion, 10000000,
+		0, defaultResources))
 
 	kr, err := testnet.CreateAccount("alice", 1e12)
 	require.NoError(t, err)

test/e2e/testnet.go

@@ -18,6 +18,7 @@ type Testnet struct {
 	nodes           []*Node
 	genesisAccounts []*Account
 	keygen          *keyGenerator
+	txSimNode       *TxSim
 }
 
 func New(name string, seed int64) (*Testnet, error) {
@@ -34,32 +35,92 @@ func New(name string, seed int64) (*Testnet, error) {
 	}, nil
 }
 
-func (t *Testnet) CreateGenesisNode(version string, selfDelegation, upgradeHeight int64) error {
+func (t *Testnet) CreateGenesisNode(version string, selfDelegation, upgradeHeight int64, resources Resources) error {
 	signerKey := t.keygen.Generate(ed25519Type)
 	networkKey := t.keygen.Generate(ed25519Type)
 	accountKey := t.keygen.Generate(secp256k1Type)
-	node, err := NewNode(fmt.Sprintf("val%d", len(t.nodes)), version, 0, selfDelegation, nil, signerKey, networkKey, accountKey, upgradeHeight)
+	node, err := NewNode(fmt.Sprintf("val%d", len(t.nodes)), version, 0, selfDelegation, nil, signerKey, networkKey, accountKey, upgradeHeight, resources)
 	if err != nil {
 		return err
 	}
 	t.nodes = append(t.nodes, node)
 	return nil
 }
 
-func (t *Testnet) CreateGenesisNodes(num int, version string, selfDelegation, upgradeHeight int64) error {
+func (t *Testnet) CreateGenesisNodes(num int, version string, selfDelegation, upgradeHeight int64, resources Resources) error {
 	for i := 0; i < num; i++ {
-		if err := t.CreateGenesisNode(version, selfDelegation, upgradeHeight); err != nil {
+		if err := t.CreateGenesisNode(version, selfDelegation, upgradeHeight, resources); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func (t *Testnet) CreateNode(version string, startHeight, upgradeHeight int64) error {
+func (t *Testnet) CreateTxSimAccount(name string, tokens int64, rootDir string) (keyring.Keyring, error) {
+	cdc := encoding.MakeConfig(app.ModuleEncodingRegisters...).Codec
+	kr, err := keyring.New(app.Name, keyring.BackendTest, rootDir, nil, cdc)
+	if err != nil {
+		return nil, err
+	}
+	key, _, err := kr.NewMnemonic(name, keyring.English, "", "", hd.Secp256k1)
+	if err != nil {
+		return nil, err
+	}
+
+	pk, err := key.GetPubKey()
+	if err != nil {
+		return nil, err
+	}
+	t.genesisAccounts = append(t.genesisAccounts, &Account{
+		PubKey:        pk,
+		InitialTokens: tokens,
+	})
+	return kr, nil
+}
+
+func (t *Testnet) SetupTxsimNode(
+	name, version string,
+	endpoint string,
+	seed int,
+	sequences int,
+	blobRange int,
+	pollTime int,
+	resources Resources,
+	volumePath string,
+	keyringPath string,
+) error {
+	txsim, err := CreateTxSimNode(name, version, endpoint, seed, sequences,
+		blobRange, pollTime, resources, volumePath)
+	if err != nil {
+		return err
+	}
+	err = txsim.Instance.AddFolder(keyringPath, volumePath, "10001:10001")
+	if err != nil {
+		return err
+	}
+
+	err = txsim.Instance.Commit()
+	if err != nil {
+		return err
+	}
+
+	t.txSimNode = txsim
+	return nil
+}
+
+func (t *Testnet) StartTxSimNode() error {
+	err := t.txSimNode.Instance.Start()
+	if err != nil {
+		return err
+	}
+	return t.txSimNode.Instance.WaitInstanceIsRunning()
+}
+
+func (t *Testnet) CreateNode(version string, startHeight, upgradeHeight int64, resources Resources) error {
 	signerKey := t.keygen.Generate(ed25519Type)
 	networkKey := t.keygen.Generate(ed25519Type)
 	accountKey := t.keygen.Generate(secp256k1Type)
-	node, err := NewNode(fmt.Sprintf("val%d", len(t.nodes)), version, startHeight, 0, nil, signerKey, networkKey, accountKey, upgradeHeight)
+	node, err := NewNode(fmt.Sprintf("val%d", len(t.nodes)), version, startHeight, 0, nil, signerKey, networkKey, accountKey, upgradeHeight, resources)
 	if err != nil {
 		return err
 	}
@@ -74,6 +135,7 @@ func (t *Testnet) CreateAccount(name string, tokens int64) (keyring.Keyring, err
 	if err != nil {
 		return nil, err
 	}
+
 	pk, err := key.GetPubKey()
 	if err != nil {
 		return nil, err
@@ -178,6 +240,12 @@ func (t *Testnet) Cleanup() {
 			log.Err(err).Msg(fmt.Sprintf("node %s failed to cleanup", node.Name))
 		}
 	}
+	if t.txSimNode.Instance != nil {
+		err := t.txSimNode.Instance.Destroy()
+		if err != nil {
+			log.Err(err).Msg(fmt.Sprintf("txsim %s failed to cleanup", t.txSimNode.Name))
+		}
+	}
 }
 
 func (t *Testnet) Node(i int) *Node {

test/e2e/throughput_test.go

@@ -0,0 +1,145 @@
+package e2e
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	v1 "github.com/celestiaorg/celestia-app/pkg/appconsts/v1"
+	"github.com/celestiaorg/celestia-app/test/util/testnode"
+	"github.com/stretchr/testify/require"
+	"github.com/tendermint/tendermint/types"
+)
+
+//const seed = 42
+//
+//var latestVersion = "latest"
+
+func TestE2EThroughput(t *testing.T) {
+	if os.Getenv("KNUU_NAMESPACE") != "test" {
+		t.Skip("skipping e2e throughput test")
+	}
+
+	if os.Getenv("E2E_LATEST_VERSION") != "" {
+		latestVersion = os.Getenv("E2E_LATEST_VERSION")
+		_, isSemVer := ParseVersion(latestVersion)
+		switch {
+		case isSemVer:
+		case latestVersion == "latest":
+		case len(latestVersion) == 7:
+		case len(latestVersion) >= 8:
+			// assume this is a git commit hash (we need to trim the last digit to match the docker image tag)
+			latestVersion = latestVersion[:7]
+		default:
+			t.Fatalf("unrecognised version: %s", latestVersion)
+		}
+	}
+	t.Log("Running throughput test", "version", latestVersion)
+
+	testnet, err := New(t.Name(), seed)
+	require.NoError(t, err)
+	t.Cleanup(testnet.Cleanup)
+	require.NoError(t, testnet.CreateGenesisNodes(4, latestVersion, 10000000,
+		0, Resources{"200Mi", "200Mi", "300m", "200Mi"}))
+
+	// create an account, store it in a temp directory and add the account as genesis account
+	txsimKeyringDir := filepath.Join(os.TempDir(), "txsim")
+	_, err = testnet.CreateTxSimAccount("alice", 1e12, txsimKeyringDir)
+	require.NoError(t, err)
+
+	// start the testnet
+	require.NoError(t, testnet.Setup()) // configs, genesis files, etc.
+	require.NoError(t, testnet.Start())
+
+	t.Log("Starting txsim")
+	// TODO pull the latest version if possible
+	// TODO increase blob size range
+	err = testnet.SetupTxsimNode("txsim", "65c1a8e",
+		testnet.GRPCEndpoints()[0], seed, 5, 50*1024, 3,
+		Resources{"200Mi", "200Mi", "300m", "1Gi"},
+		txsimRootDir, txsimKeyringDir)
+	require.NoError(t, err)
+
+	err = testnet.StartTxSimNode()
+	require.NoError(t, err)
+
+	time.Sleep(30 * time.Second)
+
+	blockchain, err := testnode.ReadBlockchain(context.Background(), testnet.Node(0).AddressRPC())
+	require.NoError(t, err)
+
+	blockTimes, blockSizes, thputs := throughput(blockchain)
+	t.Log("blockTimes", blockTimes)
+	t.Log("blockSizes", blockSizes)
+	t.Log("thputs", thputs)
+	//plotData(blockSizes, "blocksizes.png", "Block Size in bytes", "Height",
+	//	"Block Size")
+	//plotData(blockTimes, "blockTimes.png", "Block Time in seconds", "Height",
+	//	"Block Time in seconds")
+	//plotData(thputs, "thputs.png", "Throughput in bytes/second",
+	//	"Height", "Throughput in bytes/second")
+
+	totalTxs := 0
+	for _, block := range blockchain {
+		require.Equal(t, v1.Version, block.Version.App)
+		totalTxs += len(block.Data.Txs)
+	}
+	require.Greater(t, totalTxs, 10)
+}
+
+func throughput(blockchain []*types.Block) ([]float64, []float64, []float64) {
+	blockTimes := make([]float64, 0, len(blockchain)-1)
+	blockSizes := make([]float64, 0, len(blockchain)-1)
+	throughputs := make([]float64, 0, len(blockchain)-1)
+	// timestamp of the last processed block
+	lastBlockTs := blockchain[0].Header.Time
+
+	for _, block := range blockchain[1:] {
+		blockTime := float64(block.Header.Time.Sub(lastBlockTs) / 1e9) // Convert time from nanoseconds to seconds
+		blockSize := float64(block.Size() / (1024 * 1024))             // Convert size from bytes to MiB
+		thput := blockSize / blockTime
+
+		blockTimes = append(blockTimes, blockTime)
+		blockSizes = append(blockSizes, blockSize)
+		throughputs = append(throughputs, thput)
+
+		lastBlockTs = block.Header.Time // update lastBlockTs for the next block
+	}
+	return blockTimes, blockSizes, throughputs
+}
+
+//func plotData(data []float64, fileName string, title, xLabel,
+//	yLabel string,
+//) {
+//	pts := make(plotter.XYs, len(data))
+//	for i := range data {
+//		pts[i].X = float64(i)
+//		pts[i].Y = data[i]
+//	}
+//
+//	p, err := plot.New()
+//	if err != nil {
+//		panic(err)
+//	}
+//	p.Title.Text = title
+//	p.X.Label.Text = xLabel
+//	p.Y.Label.Text = yLabel
+//
+//	err = plotutil.AddLinePoints(p, yLabel, pts)
+//	if err != nil {
+//		panic(err)
+//	}
+//
+//	// save the plot
+//	if err := p.Save(10*vg.Inch, 5*vg.Inch, fileName); err != nil {
+//		panic(err)
+//	}
+//}
+//
+//func TestPlotBlockSize(_ *testing.T) {
+//	blockSizes := []float64{100, 200, 300, 400, 500, 600, 700, 800, 900, 1000}
+//	plotData(blockSizes, "blocksize.png", "Block Size in bytes",
+//		"Block Height", "Size in bytes")
+//}