Configure relayers to watch only channels associated with an individu…

…al test (#6685)

* wip: relayer configuration experimentation

* chore: test passing using relayer packet filter

* chore: configure the hermes config to watch a specific port and channel ID

* chore: small refactor for SetupPath

* chore: adding link to interchaintest issue and doing some cleanup

* chore: adding interchaintest issue link in comment

* chore: merge main

* chore: fix linter

* chore: sync main

* chore: fix go mod

* chore: only store last channel for each test

* chore: fetch channels directly in tests

* chore: PR feedback

* Co-authored-by: anhductn2001

e2e/go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/cosmos/ibc-go/modules/light-clients/08-wasm v0.0.0-00010101000000-000000000000
 	github.com/cosmos/ibc-go/v8 v8.1.0
 	github.com/docker/docker v24.0.7+incompatible
+	github.com/pelletier/go-toml v1.9.5
 	github.com/strangelove-ventures/interchaintest/v8 v8.2.1-0.20240419152858-c8b741617cd8
 	github.com/stretchr/testify v1.9.0
 	go.uber.org/zap v1.27.0
@@ -187,7 +188,6 @@ require (
 	github.com/oklog/run v1.1.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.0-rc5 // indirect
-	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
 	github.com/petermattis/goid v0.0.0-20231207134359-e60b3f734c67 // indirect
 	github.com/pierrec/xxHash v0.1.5 // indirect

e2e/relayer/relayer.go

@@ -1,13 +1,16 @@
 package relayer
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
 	dockerclient "github.com/docker/docker/client"
+	"github.com/pelletier/go-toml"
 	"github.com/strangelove-ventures/interchaintest/v8"
 	"github.com/strangelove-ventures/interchaintest/v8/ibc"
 	"github.com/strangelove-ventures/interchaintest/v8/relayer"
+	"github.com/strangelove-ventures/interchaintest/v8/relayer/hermes"
 	"go.uber.org/zap"
 )
 
@@ -24,6 +27,9 @@ const (
 	// TODO: https://github.com/cosmos/ibc-go/issues/4965
 	HyperspaceRelayerRepository = "ghcr.io/misko9/hyperspace"
 	hyperspaceRelayerUser       = "1000:1000"
+
+	// relativeHermesConfigFilePath is the path to the hermes config file relative to the home directory within the container.
+	relativeHermesConfigFilePath = ".hermes/config.toml"
 )
 
 // Config holds configuration values for the relayer used in the tests.
@@ -51,6 +57,87 @@ func New(t *testing.T, cfg Config, logger *zap.Logger, dockerClient *dockerclien
 	}
 }
 
+// ApplyPacketFilter applies a packet filter to the hermes config file, which specifies a complete set of channels
+// to watch for packets.
+func ApplyPacketFilter(ctx context.Context, t *testing.T, r ibc.Relayer, chainID string, channels []ibc.ChannelOutput) error {
+	t.Helper()
+
+	h, ok := r.(*hermes.Relayer)
+	if !ok {
+		t.Logf("relayer %T does not support packet filtering, or it has not been implemented yet.", r)
+		return nil
+	}
+
+	return modifyHermesConfigFile(ctx, h, func(config map[string]interface{}) error {
+		chains, ok := config["chains"].([]map[string]interface{})
+		if !ok {
+			return fmt.Errorf("failed to get chains from hermes config")
+		}
+		var chain map[string]interface{}
+		for _, c := range chains {
+			if c["id"] == chainID {
+				chain = c
+				break
+			}
+		}
+
+		if chain == nil {
+			return fmt.Errorf("failed to find chain with id %s", chainID)
+		}
+
+		var chanelEndpoints [][]string
+		for _, c := range channels {
+			chanelEndpoints = append(chanelEndpoints, []string{c.PortID, c.ChannelID})
+		}
+
+		// [chains.packet_filter]
+		//	# policy = 'allow'
+		//	# list = [
+		//	#   ['ica*', '*'],
+		//	#   ['transfer', 'channel-0'],
+		//	# ]
+
+		// TODO(chatton): explicitly enable watching of ICA channels
+		// this will ensure the ICA tests pass, but this will need to be modified to make sure
+		// ICA tests will succeed in parallel.
+		chanelEndpoints = append(chanelEndpoints, []string{"ica*", "*"})
+
+		// we explicitly override the full list, this allows this function to provide a complete set of channels to watch.
+		chain["packet_filter"] = map[string]interface{}{
+			"policy": "allow",
+			"list":   chanelEndpoints,
+		}
+
+		return nil
+	})
+}
+
+// modifyHermesConfigFile reads the hermes config file, applies a modification function and returns an error if any.
+func modifyHermesConfigFile(ctx context.Context, h *hermes.Relayer, modificationFn func(map[string]interface{}) error) error {
+	bz, err := h.ReadFileFromHomeDir(ctx, relativeHermesConfigFilePath)
+	if err != nil {
+		return fmt.Errorf("failed to read hermes config file: %w", err)
+	}
+
+	var config map[string]interface{}
+	if err := toml.Unmarshal(bz, &config); err != nil {
+		return fmt.Errorf("failed to unmarshal hermes config bytes")
+	}
+
+	if modificationFn != nil {
+		if err := modificationFn(config); err != nil {
+			return fmt.Errorf("failed to modify hermes config: %w", err)
+		}
+	}
+
+	bz, err = toml.Marshal(config)
+	if err != nil {
+		return fmt.Errorf("failed to marshal hermes config bytes")
+	}
+
+	return h.WriteFileToHomeDir(ctx, relativeHermesConfigFilePath, bz)
+}
+
 // newCosmosRelayer returns an instance of the go relayer.
 // Options are used to allow for relayer version selection and specifying the default processing option.
 func newCosmosRelayer(t *testing.T, tag string, logger *zap.Logger, dockerClient *dockerclient.Client, network, relayerImage string) ibc.Relayer {

e2e/tests/transfer/forwarding_test.go

@@ -37,9 +37,12 @@ func (s *TransferForwardingTestSuite) TestThreeChainSetup() {
 
 	chainA, chainB, chainC := chains[0], chains[1], chains[2]
 
-	chainAChannels := s.GetChannels(chainA)
-	chainBChannels := s.GetChannels(chainB)
-	chainCChannels := s.GetChannels(chainC)
+	chainAChannels, err := relayer.GetChannels(ctx, s.GetRelayerExecReporter(), chainA.Config().ChainID)
+	s.Require().NoError(err)
+	chainBChannels, err := relayer.GetChannels(ctx, s.GetRelayerExecReporter(), chainB.Config().ChainID)
+	s.Require().NoError(err)
+	chainCChannels, err := relayer.GetChannels(ctx, s.GetRelayerExecReporter(), chainC.Config().ChainID)
+	s.Require().NoError(err)
 
 	s.Require().Len(chainAChannels, 1, "expected 1 channels on chain A")
 	s.Require().Len(chainBChannels, 2, "expected 2 channels on chain B")

e2e/testsuite/testconfig.go

@@ -527,6 +527,7 @@ func IsFork() bool {
 type ChainOptions struct {
 	ChainSpecs       []*interchaintest.ChainSpec
 	SkipPathCreation bool
+	RelayerCount     int
 }
 
 // ChainOptionConfiguration enables arbitrary configuration of ChainOptions.
@@ -557,6 +558,9 @@ func DefaultChainOptions() ChainOptions {
 
 	return ChainOptions{
 		ChainSpecs: []*interchaintest.ChainSpec{chainASpec, chainBSpec},
+		// arbitrary number that will not be required if https://github.com/strangelove-ventures/interchaintest/issues/1153 is resolved.
+		// It can be overridden in individual test suites in SetupSuite if required.
+		RelayerCount: 10,
 	}
 }
 

e2e/testsuite/testsuite.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 	"strings"
+	"sync"
 
 	dockerclient "github.com/docker/docker/client"
 	interchaintest "github.com/strangelove-ventures/interchaintest/v8"
@@ -63,15 +64,18 @@ type E2ETestSuite struct {
 	// pathNameIndex is the latest index to be used for generating chains
 	pathNameIndex int64
 
-	// TODO: refactor this to use a relayer per test
-	// relayer is a single relayer which only works when running tests one per host.
-	// this needs to be refactored to use a different relayer per test.
-	relayer ibc.Relayer
-
 	// testSuiteName is the name of the test suite, used to store chains under the test suite name.
 	testSuiteName string
 	testPaths     map[string][]string
 	channels      map[string]map[ibc.Chain][]ibc.ChannelOutput
+
+	// relayerLock ensures concurrent tests are not accessing the pool of relayers as the same time.
+	relayerLock sync.Mutex
+	// relayerPool is a pool of relayers that can be used in tests.
+	relayerPool []ibc.Relayer
+	// testRelayerMap is a map of test suite names to relayers that are used in the test suite.
+	// this is used as a cache after a relayer has been assigned to a test suite.
+	testRelayerMap map[string]ibc.Relayer
 }
 
 // initState populates variables that are used across the test suite.
@@ -81,6 +85,8 @@ func (s *E2ETestSuite) initState() {
 	s.proposalIDs = map[string]uint64{}
 	s.testPaths = make(map[string][]string)
 	s.channels = make(map[string]map[ibc.Chain][]ibc.ChannelOutput)
+	s.relayerPool = []ibc.Relayer{}
+	s.testRelayerMap = make(map[string]ibc.Relayer)
 
 	// testSuiteName gets populated in the context of SetupSuite and stored as s.T().Name()
 	// will return the name of the suite and test when called from SetupTest or within the body of tests.
@@ -141,6 +147,18 @@ func (s *E2ETestSuite) configureGenesisDebugExport() {
 	t.Setenv("EXPORT_GENESIS_CHAIN", genesisChainName)
 }
 
+// initalizeRelayerPool pre-loads the relayer pool with n relayers.
+// this is a workaround due to the restriction on relayer creation during the test
+// ref: https://github.com/strangelove-ventures/interchaintest/issues/1153
+// if the above issue is resolved, it should be possible to lazily create relayers in each test.
+func (s *E2ETestSuite) initalizeRelayerPool(n int) []ibc.Relayer {
+	var relayers []ibc.Relayer
+	for i := 0; i < n; i++ {
+		relayers = append(relayers, relayer.New(s.T(), *LoadConfig().GetActiveRelayerConfig(), s.logger, s.DockerClient, s.network))
+	}
+	return relayers
+}
+
 // SetupChains creates the chains for the test suite, and also a relayer that is wired up to establish
 // connections and channels between the chains.
 func (s *E2ETestSuite) SetupChains(ctx context.Context, channelOptionsModifier ChainOptionModifier, chainSpecOpts ...ChainOptionConfiguration) {
@@ -155,10 +173,9 @@ func (s *E2ETestSuite) SetupChains(ctx context.Context, channelOptionsModifier C
 
 	s.chains = s.createChains(chainOptions)
 
-	// TODO: we need to create a relayer for each test that will run
-	// having a single relayer for all tests will cause issues when running tests in parallel.
-	s.relayer = relayer.New(s.T(), *LoadConfig().GetActiveRelayerConfig(), s.logger, s.DockerClient, s.network)
-	ic := s.newInterchain(ctx, s.relayer, s.chains, channelOptionsModifier)
+	s.relayerPool = s.initalizeRelayerPool(chainOptions.RelayerCount)
+
+	ic := s.newInterchain(ctx, s.relayerPool, s.chains, channelOptionsModifier)
 
 	buildOpts := interchaintest.InterchainBuildOptions{
 		TestName:  s.T().Name(),
@@ -180,7 +197,11 @@ func (s *E2ETestSuite) SetupTest() {
 // SetupPath creates a path between the chains using the provided client and channel options.
 func (s *E2ETestSuite) SetupPath(clientOpts ibc.CreateClientOptions, channelOpts ibc.CreateChannelOptions) {
 	s.T().Logf("Setting up path for: %s", s.T().Name())
-	r := s.relayer
+	r := s.GetRelayer()
+
+	if s.channels[s.T().Name()] == nil {
+		s.channels[s.T().Name()] = make(map[ibc.Chain][]ibc.ChannelOutput)
+	}
 
 	ctx := context.TODO()
 	allChains := s.GetAllChains()
@@ -208,20 +229,18 @@ func (s *E2ETestSuite) SetupPath(clientOpts ibc.CreateClientOptions, channelOpts
 		err = test.WaitForBlocks(ctx, 1, chainA, chainB)
 		s.Require().NoError(err)
 
-		channelsA, err := r.GetChannels(ctx, s.GetRelayerExecReporter(), chainA.Config().ChainID)
-		s.Require().NoError(err)
+		s.testPaths[s.T().Name()] = append(s.testPaths[s.T().Name()], pathName)
 
-		channelsB, err := r.GetChannels(ctx, s.GetRelayerExecReporter(), chainB.Config().ChainID)
-		s.Require().NoError(err)
+		for _, c := range []ibc.Chain{chainA, chainB} {
+			channels, err := r.GetChannels(ctx, s.GetRelayerExecReporter(), c.Config().ChainID)
+			s.Require().NoError(err)
 
-		if s.channels[s.T().Name()] == nil {
-			s.channels[s.T().Name()] = make(map[ibc.Chain][]ibc.ChannelOutput)
-		}
+			// only the most recent channel is relevant.
+			s.channels[s.T().Name()][c] = []ibc.ChannelOutput{channels[len(channels)-1]}
 
-		// keep track of channels associated with a given chain for access within the tests.
-		s.channels[s.T().Name()][chainA] = channelsA
-		s.channels[s.T().Name()][chainB] = channelsB
-		s.testPaths[s.T().Name()] = append(s.testPaths[s.T().Name()], pathName)
+			err = relayer.ApplyPacketFilter(ctx, s.T(), r, c.Config().ChainID, channels)
+			s.Require().NoError(err, "failed to watch port and channel on chain: %s", c.Config().ChainID)
+		}
 	}
 }
 
@@ -240,10 +259,28 @@ func (s *E2ETestSuite) GetChannels(chain ibc.Chain) []ibc.ChannelOutput {
 	return channels
 }
 
-// GetRelayer returns the relayer to be used in the specific test.
-// TODO: for now a single instance is still used, preventing parallel test runs.
+// GetRelayer returns the relayer for the current test from the available pool of relayers.
+// once a relayer has been returned to a test, it is cached and will be reused for the duration of the test.
 func (s *E2ETestSuite) GetRelayer() ibc.Relayer {
-	return s.relayer
+	s.relayerLock.Lock()
+	defer s.relayerLock.Unlock()
+
+	if r, ok := s.testRelayerMap[s.T().Name()]; ok {
+		return r
+	}
+
+	if len(s.relayerPool) == 0 {
+		panic(errors.New("relayer pool is empty"))
+	}
+
+	r := s.relayerPool[0]
+
+	// remove the relayer from the pool
+	s.relayerPool = s.relayerPool[1:]
+
+	s.testRelayerMap[s.T().Name()] = r
+
+	return r
 }
 
 // GetRelayerUsers returns two ibc.Wallet instances which can be used for the relayer users
@@ -274,12 +311,15 @@ func (s *E2ETestSuite) GetRelayerUsers(ctx context.Context, chainOpts ...ChainOp
 type ChainOptionModifier func(chainA, chainB ibc.Chain) func(options *ibc.CreateChannelOptions)
 
 // newInterchain constructs a new interchain instance that creates channels between the chains.
-func (s *E2ETestSuite) newInterchain(ctx context.Context, r ibc.Relayer, chains []ibc.Chain, modificationProvider ChainOptionModifier) *interchaintest.Interchain {
+func (s *E2ETestSuite) newInterchain(ctx context.Context, relayers []ibc.Relayer, chains []ibc.Chain, modificationProvider ChainOptionModifier) *interchaintest.Interchain {
 	ic := interchaintest.NewInterchain()
 	for _, chain := range chains {
 		ic.AddChain(chain)
 	}
-	ic.AddRelayer(r, "r")
+
+	for i, r := range relayers {
+		ic.AddRelayer(r, fmt.Sprintf("r-%d", i))
+	}
 
 	// iterate through all chains, and create links such that there is a channel between
 	// - chainA and chainB
@@ -296,13 +336,15 @@ func (s *E2ETestSuite) newInterchain(ctx context.Context, r ibc.Relayer, chains
 			modificationFn(&channelOpts)
 		}
 
-		ic.AddLink(interchaintest.InterchainLink{
-			Chain1:            chains[i],
-			Chain2:            chains[i+1],
-			Relayer:           r,
-			Path:              pathName,
-			CreateChannelOpts: channelOpts,
-		})
+		for _, r := range relayers {
+			ic.AddLink(interchaintest.InterchainLink{
+				Chain1:            chains[i],
+				Chain2:            chains[i+1],
+				Relayer:           r,
+				Path:              pathName,
+				CreateChannelOpts: channelOpts,
+			})
+		}
 	}
 
 	s.startRelayerFn = func(relayer ibc.Relayer) {