Merge pull request #109 from smartcontractkit/runlog_soak

Runlog soak test

Author: skudasov

Makefile

@@ -68,6 +68,10 @@ test_ocr: ## run ocr tests
 test_ocr_soak: ## run OCR soak test
 	NETWORK="ethereum_geth_performance" ginkgo -r --focus="@soak-ocr"
 
+.PHONY: test_runlog_soak
+test_runlog_soak: ## run Runlog soak test
+	NETWORK="ethereum_geth_performance" ginkgo -r --focus="@soak-runlog"
+
 .PHONY: test_keeper_soak
 test_keeper_soak: ## run Keeper soak/performance test
 	NETWORK="ethereum_geth_performance" ginkgo -r --focus="@performance-keeper"

config.yml

@@ -240,4 +240,3 @@ contracts:
                     ocr:
                         path: ocr/artifacts/contract/src
                         commit: f27c14a905c5735abbb6e0c9699e9d0e3e9b7217
-

contracts/contract_models.go

@@ -131,6 +131,7 @@ type Oracle interface {
 
 type APIConsumer interface {
 	Address() string
+	RoundID(ctx context.Context) (*big.Int, error)
 	Fund(fromWallet client.BlockchainWallet, ethAmount, linkAmount *big.Float) error
 	Data(ctx context.Context) (*big.Int, error)
 	CreateRequestTo(

contracts/ethereum/APIConsumer.go

@@ -210,6 +210,19 @@
     "stateMutability": "view",
     "type": "function"
   },
+  {
+    "inputs": [],
+    "name": "roundID",
+    "outputs": [
+      {
+        "internalType": "uint256",
+        "name": "",
+        "type": "uint256"
+      }
+    ],
+    "stateMutability": "view",
+    "type": "function"
+  },
   {
     "inputs": [],
     "name": "selector",

contracts/ethereum/v0.6/abi/APIConsumer.abi

@@ -18,6 +18,7 @@ import "./Ownable.sol";
  * local test networks
  */
 contract APIConsumer is ChainlinkClient, Ownable {
+  uint256 public roundID = 0;
   uint256 public data;
   bytes4 public selector;
 
@@ -85,6 +86,7 @@ contract APIConsumer is ChainlinkClient, Ownable {
     public
   {
     data = _data;
+    roundID += 1;
   }
 
   /**

contracts/ethereum/v0.6/bin/APIConsumer.bin

@@ -60,6 +60,15 @@ func (e *EthereumAPIConsumer) Address() string {
 	return e.address.Hex()
 }
 
+func (e *EthereumAPIConsumer) RoundID(ctx context.Context) (*big.Int, error) {
+	opts := &bind.CallOpts{
+		From:    common.HexToAddress(e.callerWallet.Address()),
+		Pending: true,
+		Context: ctx,
+	}
+	return e.consumer.RoundID(opts)
+}
+
 func (e *EthereumAPIConsumer) Fund(fromWallet client.BlockchainWallet, ethAmount, linkAmount *big.Float) error {
 	return e.client.Fund(fromWallet, e.address.Hex(), ethAmount, linkAmount)
 }
@@ -731,6 +740,63 @@ func (o *EthereumOffchainAggregator) GetLatestRound(ctxt context.Context) (*Roun
 	}, err
 }
 
+// RunlogRoundConfirmer is a header subscription that awaits for a certain Runlog round to be completed
+type RunlogRoundConfirmer struct {
+	consumer APIConsumer
+	roundID  *big.Int
+	doneChan chan struct{}
+	context  context.Context
+	cancel   context.CancelFunc
+}
+
+// NewRunlogRoundConfirmer provides a new instance of a RunlogRoundConfirmer
+func NewRunlogRoundConfirmer(
+	contract APIConsumer,
+	roundID *big.Int,
+	timeout time.Duration,
+) *RunlogRoundConfirmer {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), timeout)
+	return &RunlogRoundConfirmer{
+		consumer: contract,
+		roundID:  roundID,
+		doneChan: make(chan struct{}),
+		context:  ctx,
+		cancel:   ctxCancel,
+	}
+}
+
+// ReceiveBlock will query the latest Runlog round and check to see whether the round has confirmed
+func (o *RunlogRoundConfirmer) ReceiveBlock(_ client.NodeBlock) error {
+	currentRoundID, err := o.consumer.RoundID(context.Background())
+	if err != nil {
+		return err
+	}
+	ocrLog := log.Info().
+		Str("Contract Address", o.consumer.Address()).
+		Int64("Current Round", currentRoundID.Int64()).
+		Int64("Waiting for Round", o.roundID.Int64())
+	if currentRoundID.Cmp(o.roundID) >= 0 {
+		ocrLog.Msg("Runlog round completed")
+		o.doneChan <- struct{}{}
+	} else {
+		ocrLog.Msg("Waiting for Runlog round")
+	}
+	return nil
+}
+
+// Wait is a blocking function that will wait until the round has confirmed, and timeout if the deadline has passed
+func (o *RunlogRoundConfirmer) Wait() error {
+	for {
+		select {
+		case <-o.doneChan:
+			o.cancel()
+			return nil
+		case <-o.context.Done():
+			return fmt.Errorf("timeout waiting for OCR round to confirm: %d", o.roundID)
+		}
+	}
+}
+
 // OffchainAggregatorRoundConfirmer is a header subscription that awaits for a certain OCR round to be completed
 type OffchainAggregatorRoundConfirmer struct {
 	ocrInstance OffchainAggregator

contracts/ethereum/v0.6/src/APIConsumer.sol

@@ -0,0 +1,256 @@
+package performance
+
+import (
+	"context"
+	"fmt"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/onsi/ginkgo"
+	"github.com/rs/zerolog/log"
+	uuid "github.com/satori/go.uuid"
+	"github.com/smartcontractkit/integrations-framework/actions"
+	"github.com/smartcontractkit/integrations-framework/client"
+	"github.com/smartcontractkit/integrations-framework/contracts"
+	"github.com/smartcontractkit/integrations-framework/environment"
+	"golang.org/x/sync/errgroup"
+	"math/big"
+	"strings"
+	"time"
+)
+
+// RunlogJobMap is a custom map type that holds the record of jobs by the contract instance and the chainlink node
+type RunlogJobMap map[ConsumerOraclePair]map[client.Chainlink]string
+
+// ConsumerOraclePair consumer and oracle pair
+type ConsumerOraclePair struct {
+	consumer contracts.APIConsumer
+	oracle   contracts.Oracle
+	jobUUID  string
+}
+
+// RunlogTestOptions contains the parameters for the Runlog soak test to be executed
+type RunlogTestOptions struct {
+	TestOptions
+	RoundTimeout time.Duration
+	AdapterValue int
+	TestDuration time.Duration
+}
+
+// RunlogTest is the implementation of Test that will configure and execute soak test
+// of Runlog contracts & jobs
+type RunlogTest struct {
+	TestOptions RunlogTestOptions
+	Environment environment.Environment
+	Blockchain  client.BlockchainClient
+	Wallets     client.BlockchainWallets
+	Deployer    contracts.ContractDeployer
+	Link        contracts.LinkToken
+
+	chainlinkClients  []client.Chainlink
+	nodeAddresses     []common.Address
+	contractInstances []*ConsumerOraclePair
+	adapter           environment.ExternalAdapter
+
+	jobMap RunlogJobMap
+}
+
+// NewRunlogTest creates new Runlog performance/soak test
+func NewRunlogTest(
+	testOptions RunlogTestOptions,
+	env environment.Environment,
+	link contracts.LinkToken,
+	blockchain client.BlockchainClient,
+	wallets client.BlockchainWallets,
+	deployer contracts.ContractDeployer,
+	adapter environment.ExternalAdapter,
+) Test {
+	return &RunlogTest{
+		TestOptions: testOptions,
+		Environment: env,
+		Link:        link,
+		Blockchain:  blockchain,
+		Wallets:     wallets,
+		Deployer:    deployer,
+		adapter:     adapter,
+		jobMap:      RunlogJobMap{},
+	}
+}
+
+// RecordValues records Runlog metrics
+func (f *RunlogTest) RecordValues(b ginkgo.Benchmarker) error {
+	// TODO: collect metrics
+	return nil
+}
+
+// Setup setups Runlog performance/soak test
+func (f *RunlogTest) Setup() error {
+	chainlinkClients, err := environment.GetChainlinkClients(f.Environment)
+	if err != nil {
+		return err
+	}
+	nodeAddresses, err := actions.ChainlinkNodeAddresses(chainlinkClients)
+	if err != nil {
+		return err
+	}
+	adapter, err := environment.GetExternalAdapter(f.Environment)
+	if err != nil {
+		return err
+	}
+	f.chainlinkClients = chainlinkClients
+	f.nodeAddresses = nodeAddresses
+	f.adapter = adapter
+	return f.deployContracts()
+}
+
+func (f *RunlogTest) deployContract(c chan<- *ConsumerOraclePair) error {
+	oracle, err := f.Deployer.DeployOracle(f.Wallets.Default(), f.Link.Address())
+	if err != nil {
+		return err
+	}
+	if err = oracle.SetFulfillmentPermission(f.Wallets.Default(), f.nodeAddresses[0].Hex(), true); err != nil {
+		return err
+	}
+	consumer, err := f.Deployer.DeployAPIConsumer(f.Wallets.Default(), f.Link.Address())
+	if err != nil {
+		return err
+	}
+	err = consumer.Fund(f.Wallets.Default(), nil, big.NewFloat(20000))
+	if err != nil {
+		return err
+	}
+	c <- &ConsumerOraclePair{consumer: consumer, oracle: oracle}
+	return nil
+}
+
+func (f *RunlogTest) deployContracts() error {
+	contractChan := make(chan *ConsumerOraclePair, f.TestOptions.NumberOfContracts)
+	g := errgroup.Group{}
+
+	for i := 0; i < f.TestOptions.NumberOfContracts; i++ {
+		g.Go(func() error {
+			return f.deployContract(contractChan)
+		})
+	}
+	if err := g.Wait(); err != nil {
+		return err
+	}
+	close(contractChan)
+	for contract := range contractChan {
+		f.contractInstances = append(f.contractInstances, contract)
+	}
+	log.Warn().Int("Pairs", len(f.contractInstances)).Msg("Pairs")
+	return f.Blockchain.WaitForEvents()
+}
+
+func (f *RunlogTest) requestData() error {
+	g := errgroup.Group{}
+	for _, p := range f.contractInstances {
+		p := p
+		g.Go(func() error {
+			jobUUIDReplaces := strings.Replace(p.jobUUID, "-", "", 4)
+			var jobID [32]byte
+			copy(jobID[:], jobUUIDReplaces)
+			if err := p.consumer.CreateRequestTo(
+				f.Wallets.Default(),
+				p.oracle.Address(),
+				jobID,
+				big.NewInt(1e18),
+				fmt.Sprintf("%s/five", f.adapter.ClusterURL()),
+				"data,result",
+				big.NewInt(100),
+			); err != nil {
+				return err
+			}
+			return nil
+		})
+	}
+	return g.Wait()
+}
+
+// Run runs Runlog performance/soak test
+func (f *RunlogTest) Run() error {
+	if err := f.createChainlinkJobs(); err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), f.TestOptions.TestDuration)
+	defer cancel()
+	i := 1
+	for {
+		select {
+		case <-ctx.Done():
+			log.Warn().Msg("Test finished")
+			return nil
+		default:
+			log.Warn().Int("RoundID", i).Msg("New round")
+			if err := f.requestData(); err != nil {
+				return err
+			}
+			if err := f.waitRoundEnd(i); err != nil {
+				return err
+			}
+			i++
+		}
+	}
+}
+
+func (f *RunlogTest) waitRoundEnd(roundID int) error {
+	for _, p := range f.contractInstances {
+		rc := contracts.NewRunlogRoundConfirmer(p.consumer, big.NewInt(int64(roundID)), f.TestOptions.RoundTimeout)
+		f.Blockchain.AddHeaderEventSubscription(p.consumer.Address(), rc)
+	}
+	return f.Blockchain.WaitForEvents()
+}
+
+func (f *RunlogTest) createChainlinkJobs() error {
+	jobsChan := make(chan RunlogJobMap, len(f.contractInstances))
+	g := errgroup.Group{}
+
+	bta := client.BridgeTypeAttributes{
+		Name: "five",
+		URL:  fmt.Sprintf("%s/five", f.adapter.ClusterURL()),
+	}
+	if err := f.chainlinkClients[0].CreateBridge(&bta); err != nil {
+		return err
+	}
+	os := &client.DirectRequestTxPipelineSpec{
+		BridgeTypeAttributes: bta,
+		DataPath:             "data,result",
+	}
+	ost, err := os.String()
+	if err != nil {
+		return err
+	}
+
+	for _, p := range f.contractInstances {
+		p := p
+		g.Go(func() error {
+			jobUUID := uuid.NewV4()
+			p.jobUUID = jobUUID.String()
+			_, err := f.chainlinkClients[0].CreateJob(&client.DirectRequestJobSpec{
+				Name:              "direct_request",
+				ContractAddress:   p.oracle.Address(),
+				ExternalJobID:     jobUUID.String(),
+				ObservationSource: ost,
+			})
+			if err != nil {
+				return err
+			}
+			return nil
+		})
+	}
+	if err := g.Wait(); err != nil {
+		return err
+	}
+	close(jobsChan)
+
+	for jobMap := range jobsChan {
+		for contractAddr, m := range jobMap {
+			if _, ok := f.jobMap[contractAddr]; !ok {
+				f.jobMap[contractAddr] = map[client.Chainlink]string{}
+			}
+			for k, v := range m {
+				f.jobMap[contractAddr][k] = v
+			}
+		}
+	}
+	return nil
+}