Run tests in multiple simulated chains (#197)

integration-tests/ccip-tests/Makefile

@@ -1,3 +1,11 @@
+# example usage: make test_load_ccip_simulated_k8 image=chainlink-ccip tag=latest testimage=chainlink-ccip-tests:latest
+.PHONY: test_load_ccip_simulated_k8
+test_load_ccip_simulated_k8:
+	source ./load-test.env && \
+    CHAINLINK_IMAGE=$(image)  \
+    CHAINLINK_VERSION=$(tag)  \
+    ENV_JOB_IMAGE=$(testimage)  \
+	go test -timeout 24h -count=1 -v -run ^TestLoadCCIPStableRequestTriggeringWithNetworkChaos$$ ./load
 
 # example usage: make test_smoke_ccip_simulated_local image=chainlink-ccip tag=latest testname=TestSmokeCCIPForBidirectionalLane
 .PHONY: test_smoke_ccip_simulated_local

integration-tests/ccip-tests/contracts/contract_deployer.go

@@ -505,7 +505,31 @@ func (e *CCIPContractsDeployer) DeployWrappedNative() (*common.Address, error) {
 	return address, err
 }
 
-func DefaultOffChainAggregatorV2Config(numberNodes int) contracts.OffChainAggregatorV2Config {
+var OCR2ParamsForCommit = contracts.OffChainAggregatorV2Config{
+	DeltaProgress:                           2 * time.Minute,
+	DeltaResend:                             5 * time.Second,
+	DeltaRound:                              75 * time.Second,
+	DeltaGrace:                              5 * time.Second,
+	MaxDurationQuery:                        100 * time.Millisecond,
+	MaxDurationObservation:                  35 * time.Second,
+	MaxDurationReport:                       10 * time.Second,
+	MaxDurationShouldAcceptFinalizedReport:  5 * time.Second,
+	MaxDurationShouldTransmitAcceptedReport: 10 * time.Second,
+}
+
+var OCR2ParamsForExec = contracts.OffChainAggregatorV2Config{
+	DeltaProgress:                           100 * time.Second,
+	DeltaResend:                             5 * time.Second,
+	DeltaRound:                              40 * time.Second,
+	DeltaGrace:                              5 * time.Second,
+	MaxDurationQuery:                        100 * time.Millisecond,
+	MaxDurationObservation:                  20 * time.Second,
+	MaxDurationReport:                       8 * time.Second,
+	MaxDurationShouldAcceptFinalizedReport:  5 * time.Second,
+	MaxDurationShouldTransmitAcceptedReport: 8 * time.Second,
+}
+
+func OffChainAggregatorV2ConfigWithNodes(numberNodes int, inflightExpiry time.Duration, cfg contracts.OffChainAggregatorV2Config) contracts.OffChainAggregatorV2Config {
 	if numberNodes <= 4 {
 		log.Err(fmt.Errorf("insufficient number of nodes (%d) supplied for OCR, need at least 5", numberNodes)).
 			Int("Number Chainlink Nodes", numberNodes).
@@ -523,20 +547,20 @@ func DefaultOffChainAggregatorV2Config(numberNodes int) contracts.OffChainAggreg
 		faultyNodes = 1
 	}
 	return contracts.OffChainAggregatorV2Config{
-		DeltaProgress:                           70 * time.Second,
-		DeltaResend:                             5 * time.Second,
-		DeltaRound:                              30 * time.Second,
-		DeltaGrace:                              2 * time.Second,
-		DeltaStage:                              40 * time.Second,
+		DeltaProgress:                           cfg.DeltaProgress,
+		DeltaResend:                             cfg.DeltaResend,
+		DeltaRound:                              cfg.DeltaRound,
+		DeltaGrace:                              cfg.DeltaGrace,
+		DeltaStage:                              inflightExpiry,
 		RMax:                                    3,
 		S:                                       s,
 		F:                                       faultyNodes,
 		Oracles:                                 []ocrConfigHelper2.OracleIdentityExtra{},
-		MaxDurationQuery:                        5 * time.Second,
-		MaxDurationObservation:                  32 * time.Second,
-		MaxDurationReport:                       20 * time.Second,
-		MaxDurationShouldAcceptFinalizedReport:  10 * time.Second,
-		MaxDurationShouldTransmitAcceptedReport: 10 * time.Second,
+		MaxDurationQuery:                        cfg.MaxDurationQuery,
+		MaxDurationObservation:                  cfg.MaxDurationObservation,
+		MaxDurationReport:                       cfg.MaxDurationReport,
+		MaxDurationShouldAcceptFinalizedReport:  cfg.MaxDurationShouldAcceptFinalizedReport,
+		MaxDurationShouldTransmitAcceptedReport: cfg.MaxDurationShouldTransmitAcceptedReport,
 		OnchainConfig:                           []byte{},
 	}
 }
@@ -549,10 +573,12 @@ func stripKeyPrefix(key string) string {
 	return key
 }
 
-func NewOffChainAggregatorV2Config[T ccipconfig.OffchainConfig](
+func NewOffChainAggregatorV2ConfigForCCIPPlugin[T ccipconfig.OffchainConfig](
 	nodes []*client.CLNodesWithKeys,
 	offchainCfg T,
 	onchainCfg abihelpers.AbiDefined,
+	ocr2Params contracts.OffChainAggregatorV2Config,
+	inflightExpiry time.Duration,
 ) (
 	signers []common.Address,
 	transmitters []common.Address,
@@ -563,7 +589,7 @@ func NewOffChainAggregatorV2Config[T ccipconfig.OffchainConfig](
 	err error,
 ) {
 	oracleIdentities := make([]ocrConfigHelper2.OracleIdentityExtra, 0)
-	ocrConfig := DefaultOffChainAggregatorV2Config(len(nodes))
+	ocrConfig := OffChainAggregatorV2ConfigWithNodes(len(nodes), inflightExpiry, ocr2Params)
 	var onChainKeys []ocrtypes2.OnchainPublicKey
 	for i, nodeWithKeys := range nodes {
 		ocr2Key := nodeWithKeys.KeysBundle.OCR2Key.Data

integration-tests/ccip-tests/contracts/contract_models.go

@@ -520,7 +520,11 @@ func (r *Router) CCIPSend(destChainSelector uint64, msg router.ClientEVM2AnyMess
 		opts.Value = valueForNative
 	}
 
-	tx, err := r.Instance.CcipSend(opts, destChainSelector, msg)
+	return r.Instance.CcipSend(opts, destChainSelector, msg)
+}
+
+func (r *Router) CCIPSendAndProcessTx(destChainSelector uint64, msg router.ClientEVM2AnyMessage, valueForNative *big.Int) (*types.Transaction, error) {
+	tx, err := r.CCIPSend(destChainSelector, msg, valueForNative)
 	if err != nil {
 		return nil, err
 	}

integration-tests/ccip-tests/load-test.env

@@ -0,0 +1,54 @@
+# This file is used to set the environment variables for the ccip load test.
+
+export DATABASE_URL="postgresql://postgres:node@localhost:5432/chainlink_test?sslmode=disable"
+
+# if CCIP_DEPLOY_ON_LOCAL is set to false, the env will be deployed on k8s cluster, otherwise it will be deployed on local docker.
+export CCIP_DEPLOY_ON_LOCAL=False
+# the test will create new environment with new contracts, chainlink nodes and jobs if CCIP_TESTS_ON_EXISTING_DEPLOYMENT is set to false.
+# otherwise, it will use the existing environment. It will assume that deployment has already been completed and
+# will ensure the ccip-send and receive is working with the provided contracts under `./integration-tests/ccip-tests/contracts/laneconfig/contracts.json`
+export CCIP_TESTS_ON_EXISTING_DEPLOYMENT=False
+
+# the test will use simulated networks
+export SELECTED_NETWORKS="SIMULATED,SIMULATED_1,SIMULATED_2"
+export CCIP_NETWORK_PAIRS=""
+# th
+export CCIP_NO_OF_NETWORKS=10
+export CCIP_NO_OF_LANES_PER_PAIR=2
+
+# The load will be triggered as <CCIP_LOAD_TEST_RATE> per <CCIP_LOAD_TEST_RATEUNIT>
+# for <CCIP_TEST_DURATION>. Example for following: 1 request per 10s for 1h
+export CCIP_LOAD_TEST_RATEUNIT=10s
+export CCIP_LOAD_TEST_RATE=1
+export CCIP_TEST_DURATION=1h
+
+# if CCIP_KEEP_ENV_ALIVE is set to true, the env will not be destroyed after the test.
+export CCIP_KEEP_ENV_ALIVE=True
+
+# if CCIP_CHAINLINK_NODE_FUNDING is set, chainlink nodes will be funded with the mentioned amount in native.
+export CCIP_CHAINLINK_NODE_FUNDING=1000
+
+# if CCIP_KEEP_ENV_TTL is set, the env will be destroyed after the mentioned duration.
+export CCIP_KEEP_ENV_TTL=24h
+
+# Msg type to use for the load test. Default value is WithToken unless specified.
+# Values to choose from WithToken,WithoutToken
+export CCIP_MSG_TYPE=WithoutToken
+
+# remote runner resource requirements
+export RR_MEM=16Gi
+export RR_CPU=4
+
+# pg resource requirements
+export CCIP_DB_MEM=6Gi
+export CCIP_DB_CPU=2
+
+# node resource requirements
+export CCIP_NODE_MEM=4Gi
+export CCIP_NODE_CPU=2
+
+export DETACH_RUNNER=true
+export TEST_SUITE=load
+export TEST_ARGS="-test.timeout 900h"
+# creates chainlink node with this toml config
+#export CCIP_TOML_PATH="the toml config path"

integration-tests/ccip-tests/load/ccip_loadgen.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	chain_selectors "github.com/smartcontractkit/chain-selectors"
+	"github.com/smartcontractkit/chainlink-testing-framework/blockchain"
 	"github.com/smartcontractkit/wasp"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
@@ -126,6 +127,19 @@ func (c *CCIPE2ELoad) BeforeAllCall(msgType string) {
 
 	sourceCCIP.Common.ChainClient.ParallelTransactions(false)
 	destCCIP.Common.ChainClient.ParallelTransactions(false)
+	// close all header subscriptions for dest chains
+	queuedEvents := destCCIP.Common.ChainClient.GetHeaderSubscriptions()
+	for subName := range queuedEvents {
+		destCCIP.Common.ChainClient.DeleteHeaderEventSubscription(subName)
+	}
+	// close all header subscriptions for source chains except for finalized header
+	queuedEvents = sourceCCIP.Common.ChainClient.GetHeaderSubscriptions()
+	for subName := range queuedEvents {
+		if subName == blockchain.FinalizedHeaderKey {
+			continue
+		}
+		sourceCCIP.Common.ChainClient.DeleteHeaderEventSubscription(subName)
+	}
 }
 
 func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
@@ -135,7 +149,8 @@ func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
 	c.CurrentMsgSerialNo.Inc()
 
 	lggr := c.Lane.Logger.With().Int("msg Number", int(msgSerialNo)).Logger()
-
+	stats := testreporters.NewCCIPRequestStats(msgSerialNo)
+	defer c.reports.UpdatePhaseStatsForReq(stats)
 	// form the message for transfer
 	msgStr := fmt.Sprintf("new message with Id %d", msgSerialNo)
 
@@ -188,36 +203,39 @@ func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
 	}
 
 	if err != nil {
-		c.reports.UpdatePhaseStats(msgSerialNo, 0, testreporters.TX, time.Since(startTime), testreporters.Failure)
+		stats.UpdateState(lggr, 0, testreporters.TX, time.Since(startTime), testreporters.Failure)
 		res.Error = fmt.Sprintf("ccip-send tx error %+v for msg ID %d", err, msgSerialNo)
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
 	lggr = lggr.With().Str("Msg Tx", sendTx.Hash().String()).Logger()
 	txConfirmationTime := time.Now().UTC()
-	rcpt, err1 := c.Lane.Source.Common.ChainClient.GetTxReceipt(sendTx.Hash())
+	rcpt, err1 := bind.WaitMined(context.Background(), sourceCCIP.Common.ChainClient.DeployBackend(), sendTx)
 	if err1 == nil {
 		hdr, err1 := c.Lane.Source.Common.ChainClient.HeaderByNumber(context.Background(), rcpt.BlockNumber)
 		if err1 == nil {
 			txConfirmationTime = hdr.Timestamp
 		}
 	}
-	c.reports.UpdatePhaseStats(msgSerialNo, 0, testreporters.TX, startTime.Sub(txConfirmationTime), testreporters.Success,
+	var gasUsed uint64
+	if rcpt != nil {
+		gasUsed = rcpt.GasUsed
+	}
+	stats.UpdateState(lggr, 0, testreporters.TX, startTime.Sub(txConfirmationTime), testreporters.Success,
 		testreporters.TransactionStats{
 			Fee:                fee.String(),
-			GasUsed:            rcpt.GasUsed,
+			GasUsed:            gasUsed,
 			TxHash:             sendTx.Hash().Hex(),
 			NoOfTokensSent:     len(msg.TokenAmounts),
 			MessageBytesLength: len(msg.Data),
 		})
 	// wait for
 	// - CCIPSendRequested Event log to be generated,
-	msgLog, sourceLogTime, err := c.Lane.Source.AssertEventCCIPSendRequested(
-		lggr, msgSerialNo, sendTx.Hash().Hex(), c.CallTimeOut, txConfirmationTime, c.reports)
+	msgLog, sourceLogTime, err := c.Lane.Source.AssertEventCCIPSendRequested(lggr, sendTx.Hash().Hex(), c.CallTimeOut, txConfirmationTime, stats)
 	if err != nil || msgLog == nil {
 		res.Error = err.Error()
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
@@ -227,7 +245,7 @@ func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
 
 	if bytes.Compare(sentMsg.Data, []byte(msgStr)) != 0 {
 		res.Error = fmt.Sprintf("the message byte didnot match expected %s received %s msg ID %d", msgStr, string(sentMsg.Data), msgSerialNo)
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
@@ -239,7 +257,7 @@ func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
 	if c.Lane.Source.Common.ChainClient.GetNetworkConfig().FinalityDepth == 0 &&
 		lstFinalizedBlock != 0 && lstFinalizedBlock > msgLog.Raw.BlockNumber {
 		sourceLogFinalizedAt = c.LastFinalizedTimestamp.Load()
-		c.reports.UpdatePhaseStats(msgSerialNo, seqNum, testreporters.SourceLogFinalized,
+		stats.UpdateState(lggr, seqNum, testreporters.SourceLogFinalized,
 			sourceLogFinalizedAt.Sub(sourceLogTime), testreporters.Success,
 			testreporters.TransactionStats{
 				TxHash:           msgLog.Raw.TxHash.String(),
@@ -249,10 +267,10 @@ func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
 	} else {
 		var finalizingBlock uint64
 		sourceLogFinalizedAt, finalizingBlock, err = c.Lane.Source.AssertSendRequestedLogFinalized(
-			lggr, msgSerialNo, seqNum, msgLog, sourceLogTime, c.reports)
+			lggr, seqNum, msgLog, sourceLogTime, stats)
 		if err != nil {
 			res.Error = err.Error()
-			res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+			res.Data = stats.StatusByPhase
 			res.Failed = true
 			return res
 		}
@@ -262,37 +280,37 @@ func (c *CCIPE2ELoad) Call(_ *wasp.Generator) *wasp.CallResult {
 
 	// wait for
 	// - CommitStore to increase the seq number,
-	err = c.Lane.Dest.AssertSeqNumberExecuted(lggr, msgSerialNo, seqNum, c.CallTimeOut, sourceLogFinalizedAt, c.reports)
+	err = c.Lane.Dest.AssertSeqNumberExecuted(lggr, seqNum, c.CallTimeOut, sourceLogFinalizedAt, stats)
 	if err != nil {
 		res.Error = err.Error()
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
 	// wait for ReportAccepted event
-	commitReport, reportAcceptedAt, err := c.Lane.Dest.AssertEventReportAccepted(lggr, msgSerialNo, seqNum, c.CallTimeOut, sourceLogFinalizedAt, c.reports)
+	commitReport, reportAcceptedAt, err := c.Lane.Dest.AssertEventReportAccepted(lggr, seqNum, c.CallTimeOut, sourceLogFinalizedAt, stats)
 	if err != nil || commitReport == nil {
 		res.Error = err.Error()
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
-	blessedAt, err := c.Lane.Dest.AssertReportBlessed(lggr, msgSerialNo, seqNum, c.CallTimeOut, *commitReport, reportAcceptedAt, c.reports)
+	blessedAt, err := c.Lane.Dest.AssertReportBlessed(lggr, seqNum, c.CallTimeOut, *commitReport, reportAcceptedAt, stats)
 	if err != nil {
 		res.Error = err.Error()
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
-	err = c.Lane.Dest.AssertEventExecutionStateChanged(lggr, msgSerialNo, seqNum, c.CallTimeOut, blessedAt, c.reports)
+	err = c.Lane.Dest.AssertEventExecutionStateChanged(lggr, seqNum, c.CallTimeOut, blessedAt, stats)
 	if err != nil {
 		res.Error = err.Error()
-		res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+		res.Data = stats.StatusByPhase
 		res.Failed = true
 		return res
 	}
 
-	res.Data = c.reports.GetPhaseStatsForRequest(msgSerialNo)
+	res.Data = stats.StatusByPhase
 	return res
 }