Merge branch 'develop' into BCF-2612-ChainReader-Next

.github/workflows/integration-tests.yml

@@ -280,6 +280,13 @@ jobs:
           PYROSCOPE_SERVER: ${{ matrix.product.pyroscope_env == '' && '' || !startsWith(github.ref, 'refs/tags/') && '' || secrets.QA_PYROSCOPE_INSTANCE }} # Avoid sending blank envs https://github.com/orgs/community/discussions/25725
           PYROSCOPE_ENVIRONMENT: ${{ matrix.product.pyroscope_env }}
           PYROSCOPE_KEY: ${{ secrets.QA_PYROSCOPE_KEY }}
+          LOKI_TENANT_ID: ${{ vars.LOKI_TENANT_ID }}
+          LOKI_URL: ${{ secrets.LOKI_URL }}
+          LOKI_BASIC_AUTH: ${{ secrets.LOKI_BASIC_AUTH }}
+          LOGSTREAM_LOG_TARGETS: ${{ vars.LOGSTREAM_LOG_TARGETS }}
+          GRAFANA_URL: ${{ vars.GRAFANA_URL }}
+          GRAFANA_DATASOURCE: ${{ vars.GRAFANA_DATASOURCE }}
+          RUN_ID: ${{ github.run_id }}          
         with:
           test_command_to_run: cd ./integration-tests && go test -timeout 30m -count=1 -json -test.parallel=${{ matrix.product.nodes }} ${{ steps.build-go-test-command.outputs.run_command }} 2>&1 | tee /tmp/gotest.log | gotestfmt
           test_download_vendor_packages_command: cd ./integration-tests && go mod download
@@ -295,6 +302,11 @@ jobs:
           QA_AWS_REGION: ${{ secrets.QA_AWS_REGION }}
           QA_AWS_ROLE_TO_ASSUME: ${{ secrets.QA_AWS_ROLE_TO_ASSUME }}
           QA_KUBECONFIG: ""
+      - name: Print failed test summary
+        if: always()
+        uses: smartcontractkit/chainlink-github-actions/chainlink-testing-framework/show-test-summary@ea889b3133bd7f16ab19ba4ba130de5d9162c669 # v2.3.4
+        with:
+          test_directory: ./integration-tests/smoke/          
 
   eth-smoke-tests-matrix:
     if: ${{ !contains(join(github.event.pull_request.labels.*.name, ' '), 'skip-smoke-tests') }}
@@ -495,28 +507,12 @@ jobs:
         uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
         with:
           name: trace-data
-          path: ./integration-tests/smoke/traces/trace-data.json
+          path: ./integration-tests/smoke/traces/trace-data.json          
       - name: Print failed test summary
         if: always()
-        run: |
-          directory="./integration-tests/smoke/.test_summary"
-          files=("$directory"/*)
-          if [ -d "$directory" ]; then
-            echo "Test summary folder found"
-            if [ ${#files[@]} -gt 0 ]; then
-              first_file="${files[0]}"
-              echo "Name of the first test summary file: $(basename "$first_file")"
-              echo "### Failed Test Execution Logs Dashboard (over VPN):" >> $GITHUB_STEP_SUMMARY
-              cat "$first_file" | jq -r '.loki[] | "* [\(.test_name)](\(.value))"' >> $GITHUB_STEP_SUMMARY
-              if [ ${#files[@]} -gt 1 ]; then
-                echo "Found more than one test summary file. This is incorrect, there should be only one file"
-              fi
-            else
-                echo "Test summary directory is empty. This should not happen"
-            fi
-          else
-            echo "No test summary folder found. If no test failed or log collection wasn't explicitly requested this is correct. Exiting"
-          fi
+        uses: smartcontractkit/chainlink-github-actions/chainlink-testing-framework/show-test-summary@ea889b3133bd7f16ab19ba4ba130de5d9162c669 # v2.3.4
+        with:
+          test_directory: ./integration-tests/smoke/
 
   ### Used to check the required checks box when the matrix completes
   eth-smoke-tests:

core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/logprovider/buffer.go

@@ -15,10 +15,12 @@ import (
 )
 
 var (
-	// maxLogsPerUpkeepInBlock is the maximum number of logs allowed per upkeep in a block.
-	maxLogsPerUpkeepInBlock = 32
-	// maxLogsPerBlock is the maximum number of blocks in the buffer.
-	maxLogsPerBlock = 1024
+	// defaultFastExecLogsHigh is the default upper bound / maximum number of logs that Automation is committed to process for each upkeep,
+	// based on available capacity, i.e. if there are no logs from other upkeeps.
+	// Used by Log buffer to limit the number of logs we are saving in memory for each upkeep in a block
+	defaultFastExecLogsHigh = 32
+	// defaultNumOfLogUpkeeps is the default number of log upkeeps supported by the registry.
+	defaultNumOfLogUpkeeps = 50
 )
 
 // fetchedLog holds the log and the ID of the upkeep
@@ -143,20 +145,20 @@ type logEventBuffer struct {
 	// size is the number of blocks supported by the buffer
 	size int32
 
-	maxBlockLogs, maxUpkeepLogsPerBlock int
+	numOfLogUpkeeps, fastExecLogsHigh uint32
 	// blocks is the circular buffer of fetched blocks
 	blocks []fetchedBlock
 	// latestBlock is the latest block number seen
 	latestBlock int64
 }
 
-func newLogEventBuffer(lggr logger.Logger, size, maxBlockLogs, maxUpkeepLogsPerBlock int) *logEventBuffer {
+func newLogEventBuffer(lggr logger.Logger, size, numOfLogUpkeeps, fastExecLogsHigh int) *logEventBuffer {
 	return &logEventBuffer{
-		lggr:                  lggr.Named("KeepersRegistry.LogEventBuffer"),
-		size:                  int32(size),
-		blocks:                make([]fetchedBlock, size),
-		maxBlockLogs:          maxBlockLogs,
-		maxUpkeepLogsPerBlock: maxUpkeepLogsPerBlock,
+		lggr:             lggr.Named("KeepersRegistry.LogEventBuffer"),
+		size:             int32(size),
+		blocks:           make([]fetchedBlock, size),
+		numOfLogUpkeeps:  uint32(numOfLogUpkeeps),
+		fastExecLogsHigh: uint32(fastExecLogsHigh),
 	}
 }
 
@@ -168,6 +170,11 @@ func (b *logEventBuffer) bufferSize() int {
 	return int(atomic.LoadInt32(&b.size))
 }
 
+func (b *logEventBuffer) SetLimits(numOfLogUpkeeps, fastExecLogsHigh int) {
+	atomic.StoreUint32(&b.numOfLogUpkeeps, uint32(numOfLogUpkeeps))
+	atomic.StoreUint32(&b.fastExecLogsHigh, uint32(fastExecLogsHigh))
+}
+
 // enqueue adds logs (if not exist) to the buffer, returning the number of logs added
 // minus the number of logs dropped.
 func (b *logEventBuffer) enqueue(id *big.Int, logs ...logpoller.Log) int {
@@ -176,8 +183,8 @@ func (b *logEventBuffer) enqueue(id *big.Int, logs ...logpoller.Log) int {
 
 	lggr := b.lggr.With("id", id.String())
 
-	maxBlockLogs := b.maxBlockLogs
-	maxUpkeepLogs := b.maxUpkeepLogsPerBlock
+	maxBlockLogs := int(atomic.LoadUint32(&b.fastExecLogsHigh) * atomic.LoadUint32(&b.numOfLogUpkeeps))
+	maxUpkeepLogs := int(atomic.LoadUint32(&b.fastExecLogsHigh))
 
 	latestBlock := b.latestBlockSeen()
 	added, dropped := 0, 0

core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/logprovider/buffer_test.go

@@ -187,7 +187,7 @@ func TestLogEventBuffer_EnqueueDequeue(t *testing.T) {
 	})
 
 	t.Run("enqueue logs overflow", func(t *testing.T) {
-		buf := newLogEventBuffer(logger.TestLogger(t), 2, 2, 10)
+		buf := newLogEventBuffer(logger.TestLogger(t), 2, 2, 2)
 
 		require.Equal(t, 2, buf.enqueue(big.NewInt(1),
 			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 0},
@@ -199,6 +199,50 @@ func TestLogEventBuffer_EnqueueDequeue(t *testing.T) {
 		buf.lock.Unlock()
 	})
 
+	t.Run("enqueue logs overflow with dynamic limits", func(t *testing.T) {
+		buf := newLogEventBuffer(logger.TestLogger(t), 2, 10, 2)
+
+		require.Equal(t, 2, buf.enqueue(big.NewInt(1),
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 0},
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 1},
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 2},
+		))
+		buf.SetLimits(10, 3)
+		require.Equal(t, 3, buf.enqueue(big.NewInt(1),
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 0},
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 1},
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 2},
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 3},
+		))
+
+		buf.lock.Lock()
+		defer buf.lock.Unlock()
+		require.Equal(t, 2, len(buf.blocks[0].logs))
+		require.Equal(t, 3, len(buf.blocks[1].logs))
+	})
+
+	t.Run("enqueue logs overflow with dynamic limits", func(t *testing.T) {
+		buf := newLogEventBuffer(logger.TestLogger(t), 2, 10, 2)
+
+		require.Equal(t, 2, buf.enqueue(big.NewInt(1),
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 0},
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 1},
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 2},
+			logpoller.Log{BlockNumber: 1, TxHash: common.HexToHash("0x1"), LogIndex: 3},
+		))
+		buf.SetLimits(10, 3)
+		require.Equal(t, 3, buf.enqueue(big.NewInt(1),
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 0},
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 1},
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 2},
+			logpoller.Log{BlockNumber: 2, TxHash: common.HexToHash("0x21"), LogIndex: 3},
+		))
+
+		buf.lock.Lock()
+		defer buf.lock.Unlock()
+		require.Equal(t, 2, len(buf.blocks[0].logs))
+	})
+
 	t.Run("enqueue block overflow", func(t *testing.T) {
 		buf := newLogEventBuffer(logger.TestLogger(t), 3, 2, 10)
 

core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/logprovider/provider.go

@@ -109,7 +109,7 @@ func NewLogProvider(lggr logger.Logger, poller logpoller.LogPoller, packer LogDa
 		threadCtrl:  utils.NewThreadControl(),
 		lggr:        lggr.Named("KeepersRegistry.LogEventProvider"),
 		packer:      packer,
-		buffer:      newLogEventBuffer(lggr, int(opts.LookbackBlocks), maxLogsPerBlock, maxLogsPerUpkeepInBlock),
+		buffer:      newLogEventBuffer(lggr, int(opts.LookbackBlocks), defaultNumOfLogUpkeeps, defaultFastExecLogsHigh),
 		poller:      poller,
 		opts:        opts,
 		filterStore: filterStore,

integration-tests/docker/test_env/cl_node.go

@@ -52,12 +52,8 @@ type ClNode struct {
 	UserEmail             string                  `json:"userEmail"`
 	UserPassword          string                  `json:"userPassword"`
 	AlwaysPullImage       bool                    `json:"-"`
-	PostStartsHooks       []tc.ContainerHook      `json:"-"`
-	PostStopsHooks        []tc.ContainerHook      `json:"-"`
-	PreTerminatesHooks    []tc.ContainerHook      `json:"-"`
 	t                     *testing.T
 	l                     zerolog.Logger
-	ls                    *logstream.LogStream
 }
 
 type ClNodeOption = func(c *ClNode)
@@ -97,7 +93,7 @@ func WithDbContainerName(name string) ClNodeOption {
 
 func WithLogStream(ls *logstream.LogStream) ClNodeOption {
 	return func(c *ClNode) {
-		c.ls = ls
+		c.LogStream = ls
 	}
 }
 
@@ -150,25 +146,6 @@ func NewClNode(networks []string, imageName, imageVersion string, nodeConfig *ch
 	return n, nil
 }
 
-func (n *ClNode) SetDefaultHooks() {
-	n.PostStartsHooks = []tc.ContainerHook{
-		func(ctx context.Context, c tc.Container) error {
-			if n.ls != nil {
-				return n.ls.ConnectContainer(ctx, c, "cl-node")
-			}
-			return nil
-		},
-	}
-	n.PostStopsHooks = []tc.ContainerHook{
-		func(ctx context.Context, c tc.Container) error {
-			if n.ls != nil {
-				return n.ls.DisconnectContainer(c)
-			}
-			return nil
-		},
-	}
-}
-
 func (n *ClNode) SetTestLogger(t *testing.T) {
 	n.l = logging.GetTestLogger(t)
 	n.t = t

integration-tests/docker/test_env/test_env.go

@@ -1,7 +1,6 @@
 package test_env
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"math/big"
@@ -66,7 +65,7 @@ func (te *CLClusterTestEnv) WithTestEnvConfig(cfg *TestEnvConfig) *CLClusterTest
 	te.Cfg = cfg
 	if cfg.MockAdapter.ContainerName != "" {
 		n := []string{te.Network.Name}
-		te.MockAdapter = test_env.NewKillgrave(n, te.Cfg.MockAdapter.ImpostersPath, test_env.WithContainerName(te.Cfg.MockAdapter.ContainerName))
+		te.MockAdapter = test_env.NewKillgrave(n, te.Cfg.MockAdapter.ImpostersPath, test_env.WithContainerName(te.Cfg.MockAdapter.ContainerName), test_env.WithLogStream(te.LogStream))
 	}
 	return te
 }
@@ -154,7 +153,7 @@ func (te *CLClusterTestEnv) StartClCluster(nodeConfig *chainlink.Config, count i
 	if te.Cfg != nil && te.Cfg.ClCluster != nil {
 		te.ClCluster = te.Cfg.ClCluster
 	} else {
-		opts = append(opts, WithSecrets(secretsConfig))
+		opts = append(opts, WithSecrets(secretsConfig), WithLogStream(te.LogStream))
 		te.ClCluster = &ClCluster{}
 		for i := 0; i < count; i++ {
 			ocrNode, err := NewClNode([]string{te.Network.Name}, os.Getenv("CHAINLINK_IMAGE"), os.Getenv("CHAINLINK_VERSION"), nodeConfig, opts...)
@@ -172,18 +171,6 @@ func (te *CLClusterTestEnv) StartClCluster(nodeConfig *chainlink.Config, count i
 		}
 	}
 
-	if te.LogStream != nil {
-		for _, node := range te.ClCluster.Nodes {
-			node.ls = te.LogStream
-		}
-		if te.MockAdapter != nil {
-			err := te.LogStream.ConnectContainer(context.Background(), te.MockAdapter.Container, "mock-adapter")
-			if err != nil {
-				return err
-			}
-		}
-	}
-
 	// Start/attach node containers
 	return te.ClCluster.Start()
 }

integration-tests/docker/test_env/test_env_builder.go

@@ -238,7 +238,7 @@ func (b *CLTestEnvBuilder) Build() (*CLClusterTestEnv, error) {
 			return nil, fmt.Errorf("test environment builder failed: %w", fmt.Errorf("cannot start mock adapter without a network"))
 		}
 
-		b.te.MockAdapter = test_env.NewKillgrave([]string{b.te.Network.Name}, "")
+		b.te.MockAdapter = test_env.NewKillgrave([]string{b.te.Network.Name}, "", test_env.WithLogStream(b.te.LogStream))
 
 		err = b.te.StartMockAdapter()
 		if err != nil {

integration-tests/go.mod

@@ -26,7 +26,7 @@ require (
 	github.com/slack-go/slack v0.12.2
 	github.com/smartcontractkit/chainlink-automation v1.0.2-0.20240118014648-1ab6a88c9429
 	github.com/smartcontractkit/chainlink-common v0.1.7-0.20240119143538-04c7f63ad53a
-	github.com/smartcontractkit/chainlink-testing-framework v1.22.4
+	github.com/smartcontractkit/chainlink-testing-framework v1.22.6
 	github.com/smartcontractkit/chainlink-vrf v0.0.0-20231120191722-fef03814f868
 	github.com/smartcontractkit/chainlink/v2 v2.0.0-00010101000000-000000000000
 	github.com/smartcontractkit/libocr v0.0.0-20240112202000-6359502d2ff1

integration-tests/go.sum

@@ -1498,8 +1498,8 @@ github.com/smartcontractkit/chainlink-solana v1.0.3-0.20240119150605-bbac0e5e53a
 github.com/smartcontractkit/chainlink-solana v1.0.3-0.20240119150605-bbac0e5e53a5/go.mod h1:OZfzyayUdwsVBqxvbEMqwUntQT8HbFbgyqoudvwfVN0=
 github.com/smartcontractkit/chainlink-starknet/relayer v0.0.1-beta-test.0.20231221191127-1f32389044ea h1:WzMa0O6DEauMYMIjzS/T1JF8zvFDt4aG6EUTDlStaZo=
 github.com/smartcontractkit/chainlink-starknet/relayer v0.0.1-beta-test.0.20231221191127-1f32389044ea/go.mod h1:J4A5pQh3CiVs5S2eQMHezToXHoC+Wj0UHBtMXCiTIJA=
-github.com/smartcontractkit/chainlink-testing-framework v1.22.4 h1:617FyrN/wQNFLsGDofqKAQ5/P+swmIPo8sIJZqBPvsQ=
-github.com/smartcontractkit/chainlink-testing-framework v1.22.4/go.mod h1:FBRC6elqaqO8jiMZMfa3UKrvwzZhMGUtYqVTGYmfFrA=
+github.com/smartcontractkit/chainlink-testing-framework v1.22.6 h1:5kWMlo99RY/ys4EWGMPsEg1sfY67f5YQogI8PohdRvw=
+github.com/smartcontractkit/chainlink-testing-framework v1.22.6/go.mod h1:FBRC6elqaqO8jiMZMfa3UKrvwzZhMGUtYqVTGYmfFrA=
 github.com/smartcontractkit/chainlink-vrf v0.0.0-20231120191722-fef03814f868 h1:FFdvEzlYwcuVHkdZ8YnZR/XomeMGbz5E2F2HZI3I3w8=
 github.com/smartcontractkit/chainlink-vrf v0.0.0-20231120191722-fef03814f868/go.mod h1:Kn1Hape05UzFZ7bOUnm3GVsHzP0TNrVmpfXYNHdqGGs=
 github.com/smartcontractkit/go-plugin v0.0.0-20231003134350-e49dad63b306 h1:ko88+ZznniNJZbZPWAvHQU8SwKAdHngdDZ+pvVgB5ss=