CCIP Load Test connected to crib (not yet working)

deployment/ccip/changeset/state.go

@@ -380,6 +380,8 @@ func LoadChainState(chain deployment.Chain, addresses map[string]deployment.Type
 				return state, err
 			}
 			state.OffRamp = offRamp
+		case deployment.NewTypeAndVersion(ARMProxy, deployment.Version1_6_0_dev).String():
+			fallthrough
 		case deployment.NewTypeAndVersion(ARMProxy, deployment.Version1_0_0).String():
 			armProxy, err := rmn_proxy_contract.NewRMNProxyContract(common.HexToAddress(address), chain.Client)
 			if err != nil {

deployment/environment/crib/env.go

@@ -1,5 +1,7 @@
 package crib
 
+import "fmt"
+
 const (
 	AddressBookFileName   = "ccip-v2-scripts-address-book.json"
 	NodesDetailsFileName  = "ccip-v2-scripts-nodes-details.json"
@@ -16,15 +18,27 @@ func NewDevspaceEnvFromStateDir(envStateDir string) CRIBEnv {
 	}
 }
 
-func (c CRIBEnv) GetConfig() DeployOutput {
+func (c CRIBEnv) GetConfig(deployerKeys map[uint64]string) (DeployOutput, error) {
 	reader := NewOutputReader(c.envStateDir)
 	nodesDetails := reader.ReadNodesDetails()
 	chainConfigs := reader.ReadChainConfigs()
+	for i, chain := range chainConfigs {
+		key, ok := deployerKeys[chain.ChainID]
+		if !ok {
+			return DeployOutput{}, fmt.Errorf("could not find deployer key for %s", key)
+		}
+		err := chain.SetDeployerKey(&key)
+		if err != nil {
+			return DeployOutput{}, err
+		}
+		chainConfigs[i] = chain
+	}
+
 	return DeployOutput{
 		AddressBook: reader.ReadAddressBook(),
 		NodeIDs:     nodesDetails.NodeIDs,
 		Chains:      chainConfigs,
-	}
+	}, nil
 }
 
 type RPC struct {

deployment/environment/devenv/chain.go

@@ -67,7 +67,6 @@ func (c *ChainConfig) SetDeployerKey(pvtKeyStr *string) error {
 		if err != nil {
 			return fmt.Errorf("failed to create transactor: %w", err)
 		}
-		fmt.Printf("Deployer Address: %s for chain id %d\n", deployer.From.Hex(), c.ChainID)
 		c.DeployerKey = deployer
 		return nil
 	}

integration-tests/load/ccip/ccip_test.go

@@ -0,0 +1,170 @@
+package ccip
+
+import (
+	"fmt"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
+	"github.com/smartcontractkit/chainlink-testing-framework/wasp"
+	ccipchangeset "github.com/smartcontractkit/chainlink/deployment/ccip/changeset"
+	crib "github.com/smartcontractkit/chainlink/deployment/environment/crib"
+	tc "github.com/smartcontractkit/chainlink/integration-tests/testconfig"
+	"github.com/stretchr/testify/require"
+	"math/big"
+	"sync"
+	"testing"
+	"time"
+)
+
+var (
+	CommonTestLabels = map[string]string{
+		"branch": "ccip_load_crib",
+		"commit": "ccip_load_crib",
+	}
+	wg                     sync.WaitGroup
+	SIM_CHAIN_PRIVATE_KEYS = map[uint64]string{
+		1337: "ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80",
+		2337: "59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d",
+	}
+)
+
+const CRIB_DIRECTORY = "/Users/austin.wang/ccip-core/repos/crib/deployments/ccip-v2/.tmp"
+
+// step 1: setup
+// Parse the test config, initialize CRIB with configurations defined
+// step 2: load
+// Use wasp to initiate load
+// step 3: parse logs
+// Parse all events from the simulated chains, send to Loki
+// step 4: teardown
+// Stop the chains, cleanup the environment
+func TestCCIPLoad_RPS(t *testing.T) {
+	ctx := tests.Context(t)
+	lggr := logger.Test(t)
+
+	config, err := tc.GetConfig([]string{"Load"}, tc.CCIP)
+	require.NoError(t, err)
+	lggr.Infof("loaded ccip test config: %+v", config.CCIP.Load)
+
+	cribEnv := crib.NewDevspaceEnvFromStateDir(CRIB_DIRECTORY)
+
+	cribDeployOutput, err := cribEnv.GetConfig(SIM_CHAIN_PRIVATE_KEYS)
+	require.NoError(t, err)
+	env, err := crib.NewDeployEnvironmentFromCribOutput(lggr, cribDeployOutput)
+	require.NoError(t, err)
+	require.NotNil(t, env)
+
+	// Need to keep track of the block number for each chain so that event subscription can be done from that block.
+	startBlocks := make(map[uint64]*uint64)
+	state, err := ccipchangeset.LoadOnchainState(*env)
+	require.NoError(t, err)
+
+	// Parse all events from the simulated chains, send to Loki
+	loki, err := wasp.NewLokiClient(wasp.NewLokiConfig(config.CCIP.Load.LokiEndpoint, nil, nil, nil))
+	require.NoError(t, err)
+	defer loki.StopNow()
+
+	// Based on the config, initialize DestinationGun
+	p := wasp.NewProfile()
+	for selector, chain := range env.Chains {
+		latesthdr, err := chain.Client.HeaderByNumber(ctx, nil)
+		require.NoError(t, err)
+		block := latesthdr.Number.Uint64()
+		startBlocks[selector] = &block
+
+		p.Add(wasp.NewGenerator(&wasp.Config{
+			T:           t,
+			GenName:     "ccipLoad",
+			LoadType:    wasp.RPS,
+			CallTimeout: 5 * time.Second,
+			Schedule:    wasp.Plain(1, 5*time.Second),
+			// will need to be divided by number of chains
+			// this schedule is per generator
+			// in this example, it would be 1 request per 10seconds per generator (dest chain)
+			// so if there are 3 generators, it would be 3 requests per 10 seconds over the network
+			Gun:        NewDestinationGun(env.Logger, selector, *env, state.Chains[selector].Receiver.Address(), loki),
+			Labels:     CommonTestLabels,
+			LokiConfig: wasp.NewLokiConfig(config.CCIP.Load.LokiEndpoint, nil, nil, nil),
+			// use the same loki client using `NewLokiClient` with the same config for sending events
+		}))
+	}
+
+	_, err = p.Run(true)
+
+	lokiLabels := map[string]string{}
+	for chainSelector, startBlock := range startBlocks {
+		wg.Add(1)
+		go func(chainSelector uint64, startBlock *uint64) {
+			defer wg.Done()
+			lggr.Infow("Starting to query for events on ", "chainSelector", chainSelector, "startblock", startBlock)
+
+			filterOpts := &bind.FilterOpts{
+				Start:   *startBlock,
+				End:     nil, // To the latest block
+				Context: ctx,
+			}
+
+			offRamp := state.Chains[chainSelector].OffRamp
+			// Filter CommitReportAccepted events
+			commitIterator, err := offRamp.FilterCommitReportAccepted(filterOpts)
+			require.NoError(t, err)
+
+			fmt.Printf("Events on commitIterator %+v", commitIterator)
+
+			for commitIterator.Next() {
+				event := commitIterator.Event
+				fmt.Printf("CommitReportAccepted event: %+v\n", event)
+
+				blockNum := commitIterator.Event.Raw.BlockNumber
+				header, err := env.Chains[chainSelector].Client.HeaderByNumber(ctx, big.NewInt(int64(blockNum)))
+				require.NoError(t, err)
+				timestamp := time.Unix(int64(header.Time), 0)
+
+				for _, root := range event.MerkleRoots {
+					lokiLabels, err = setLokiLabels(root.SourceChainSelector, chainSelector)
+					require.NoError(t, err)
+
+					for i := root.MinSeqNr; i <= root.MaxSeqNr; i++ {
+						// todo: push loki calls to channel?
+						SendMetricsToLoki(lggr, loki, lokiLabels, &LokiMetric{
+							EventType:      committed,
+							Timestamp:      timestamp,
+							SequenceNumber: i,
+						})
+					}
+				}
+			}
+
+			// Filter ExecutionStateChanged events
+			execIterator, err := state.Chains[chainSelector].OffRamp.FilterExecutionStateChanged(filterOpts, []uint64{chainSelector}, nil, nil)
+			require.NoError(t, err)
+			for execIterator.Next() {
+				event := execIterator.Event
+				fmt.Printf("ExecutionStateChanged event: %+v\n", event)
+
+				blockNum := execIterator.Event.Raw.BlockNumber
+				header, err := env.Chains[chainSelector].Client.HeaderByNumber(ctx, big.NewInt(int64(blockNum)))
+				require.NoError(t, err)
+				timestamp := time.Unix(int64(header.Time), 0)
+
+				// todo: push loki calls to channel?
+				lokiLabels, err = setLokiLabels(execIterator.Event.SourceChainSelector, chainSelector)
+				require.NoError(t, err)
+				SendMetricsToLoki(lggr, loki, lokiLabels, &LokiMetric{
+					EventType:      executed,
+					Timestamp:      timestamp,
+					GasUsed:        execIterator.Event.GasUsed.Uint64(),
+					SequenceNumber: execIterator.Event.SequenceNumber,
+				})
+
+			}
+		}(chainSelector, startBlock)
+	}
+
+	wg.Wait()
+
+	// Stop the chains, cleanup the environment
+
+	// crib.StopChains(env)
+	// crib.StopNodes(env)
+}

integration-tests/load/ccip/destination_gun.go

@@ -0,0 +1,144 @@
+package ccip
+
+import (
+	"context"
+	"fmt"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-testing-framework/wasp"
+	"github.com/smartcontractkit/chainlink/deployment"
+	ccipchangeset "github.com/smartcontractkit/chainlink/deployment/ccip/changeset"
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils"
+	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/router"
+	"go.uber.org/atomic"
+	"time"
+)
+
+type ChainSelectorPair struct {
+	src uint64
+	dst uint64
+}
+
+type DestinationGun struct {
+	l             logger.Logger
+	env           deployment.Environment
+	seqNums       map[ChainSelectorPair]*atomic.Uint64
+	roundNum      *atomic.Int32
+	chainSelector uint64
+	receiver      common.Address
+	loki          *wasp.LokiClient
+}
+
+func NewDestinationGun(l logger.Logger, chainSelector uint64, env deployment.Environment, receiver common.Address, loki *wasp.LokiClient) *DestinationGun {
+	seqNums := make(map[ChainSelectorPair]*atomic.Uint64)
+	for _, cs := range env.AllChainSelectorsExcluding([]uint64{chainSelector}) {
+
+		seqNums[ChainSelectorPair{
+			src: cs,
+			dst: chainSelector,
+		}] = atomic.NewUint64(1)
+	}
+	return &DestinationGun{
+		l:             l,
+		env:           env,
+		seqNums:       seqNums,
+		roundNum:      &atomic.Int32{},
+		chainSelector: chainSelector,
+		receiver:      receiver,
+		loki:          loki,
+	}
+}
+
+func (m *DestinationGun) Call(_ *wasp.Generator) *wasp.Response {
+	m.roundNum.Add(1)
+	requestedRound := m.roundNum.Load()
+
+	waspGroup := fmt.Sprintf("%d-%s", m.chainSelector, "messageOnly")
+
+	state, err := ccipchangeset.LoadOnchainState(m.env)
+	if err != nil {
+		return &wasp.Response{Error: err.Error(), Group: waspGroup, Failed: true}
+	}
+
+	src, err := m.MustSourceChain()
+	if err != nil {
+		return &wasp.Response{Error: err.Error(), Group: waspGroup, Failed: true}
+	}
+
+	csPair := ChainSelectorPair{
+		src: src,
+		dst: m.chainSelector,
+	}
+	m.seqNums[csPair].Add(1)
+	m.l.Infow("Starting transmit with ", "RoundNum", requestedRound, "Destination ChainSelector", m.chainSelector, "Source ChainSelector", src, "SequenceNumber", m.seqNums[csPair].Load())
+
+	r := state.Chains[src].Router
+
+	msg, err := m.GetMessage()
+	if err != nil {
+		return &wasp.Response{Error: err.Error(), Group: waspGroup, Failed: true}
+	}
+
+	fee, err := r.GetFee(
+		&bind.CallOpts{Context: context.Background()}, m.chainSelector, msg)
+	if err != nil {
+		m.l.Errorw("could not get fee ", "dstChainSelector", m.chainSelector, "msg", msg, "fee", fee)
+		return &wasp.Response{Error: err.Error(), Group: waspGroup, Failed: true}
+	}
+	m.l.Debugw("setting fee for ", "srcChain", src, "dstChain", m.chainSelector, "fee", fee, "msg", msg)
+	if msg.FeeToken == common.HexToAddress("0x0") {
+		m.env.Chains[src].DeployerKey.Value = fee
+		defer func() { m.env.Chains[src].DeployerKey.Value = nil }()
+	}
+	tx, err := r.CcipSend(
+		m.env.Chains[src].DeployerKey,
+		m.chainSelector,
+		msg)
+	if err != nil {
+		m.l.Errorw("execution reverted from ", "sourceChain", src, "destchain", m.chainSelector, "err", err, "tx", tx)
+		return &wasp.Response{Error: err.Error(), Group: waspGroup, Failed: true}
+	}
+
+	lokiLabels, err := setLokiLabels(src, m.chainSelector)
+	if err != nil {
+		m.l.Errorw("Failed setting loki labels", "error", err)
+	}
+	SendMetricsToLoki(m.l, m.loki, lokiLabels, &LokiMetric{
+		EventType:      transmitted,
+		Timestamp:      time.Now(),
+		SequenceNumber: m.seqNums[csPair].Load(),
+	})
+
+	return &wasp.Response{Failed: false, Group: waspGroup}
+}
+
+// MustSourceChain will return a chain selector to send a message from
+func (m *DestinationGun) MustSourceChain() (uint64, error) {
+
+	// TODO: make this smarter by checking if this chain has sent a message recently, if so, switch to the next chain
+	otherCS := m.env.AllChainSelectorsExcluding([]uint64{m.chainSelector})
+	if len(otherCS) == 0 {
+		return 0, fmt.Errorf("no other chains to send from")
+	}
+	index := m.roundNum.Load() % int32(len(otherCS))
+	return otherCS[index], nil
+}
+
+// GetMessage will return the message to be sent while considering expected load of different messages
+// TODO: implement randomness and different types of messages
+func (m *DestinationGun) GetMessage() (router.ClientEVM2AnyMessage, error) {
+	rcv, err := utils.ABIEncode(`[{"type":"address"}]`, m.receiver)
+	if err != nil {
+		m.l.Error("Error encoding receiver address")
+		return router.ClientEVM2AnyMessage{}, err
+	}
+
+	return router.ClientEVM2AnyMessage{
+		Receiver:     rcv,
+		Data:         common.Hex2Bytes("hello world"),
+		TokenAmounts: nil,
+		FeeToken:     common.HexToAddress("0x0"),
+		ExtraArgs:    nil,
+	}, nil
+}