Repository: cosmos/relayer
Pull request #882 (2022-08-01 22:45:51)

Initial prometheus
Adds prometheus listener, configurable either by config `global.prometheus-listen-addr` or flag `--prometheus-addr`.

Implements packet count metrics for both observed packets once the chains are in sync and relayed packets.

e.g. with prometheus listen addr of `:9090`, metrics:
```prometheus
go_goroutines 29
...
go_threads 39
...
observed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="acknowledge_packet"} 57
observed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="recv_packet"} 103
observed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="send_packet"} 58
observed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="acknowledge_packet"} 107
observed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="recv_packet"} 60
observed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="send_packet"} 102
...
relayed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="acknowledge_packet"} 31
relayed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="recv_packet"} 65
relayed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="acknowledge_packet"} 36
relayed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="recv_packet"} 35
```


Patch:
diff --git a/cmd/config.go b/cmd/config.go
index 4c5ca0d..37ba3ac 100644
--- a/cmd/config.go
+++ b/cmd/config.go
@@ -165,7 +165,8 @@ $ %s cfg i`, appName, defaultHome, appName)),
 			return fmt.Errorf("config already exists: %s", cfgPath)
 		},
 	}
-	return memoFlag(a.Viper, cmd)
+	cmd = memoFlag(a.Viper, cmd)
+	return cmd
 }
 
 // addChainsFromDirectory finds all JSON-encoded config files in dir,
@@ -535,7 +536,7 @@ func validateConfig(c *Config) error {
 	return nil
 }
 
-// initConfig reads in config file and ENV variables if set.
+// initConfig reads config file into a.Config if file is present.
 func initConfig(cmd *cobra.Command, a *appState) error {
 	home, err := cmd.PersistentFlags().GetString(flagHome)
 	if err != nil {
@@ -543,59 +544,63 @@ func initConfig(cmd *cobra.Command, a *appState) error {
 	}
 
 	cfgPath := path.Join(home, "config", "config.yaml")
-	if _, err := os.Stat(cfgPath); err == nil {
-		a.Viper.SetConfigFile(cfgPath)
-		if err := a.Viper.ReadInConfig(); err == nil {
-			// read the config file bytes
-			file, err := os.ReadFile(a.Viper.ConfigFileUsed())
-			if err != nil {
-				fmt.Fprintln(cmd.ErrOrStderr(), "Error reading file:", err)
-				return err
-			}
+	if _, err := os.Stat(cfgPath); err != nil {
+		// don't return error if file doesn't exist
+		return nil
+	}
+	a.Viper.SetConfigFile(cfgPath)
+	if err := a.Viper.ReadInConfig(); err != nil {
+		return err
+	}
+	// read the config file bytes
+	file, err := os.ReadFile(a.Viper.ConfigFileUsed())
+	if err != nil {
+		fmt.Fprintln(cmd.ErrOrStderr(), "Error reading file:", err)
+		return err
+	}
 
-			// unmarshall them into the wrapper struct
-			cfgWrapper := &ConfigInputWrapper{}
-			err = yaml.Unmarshal(file, cfgWrapper)
-			if err != nil {
-				fmt.Fprintln(cmd.ErrOrStderr(), "Error unmarshalling config:", err)
-				return err
-			}
+	// unmarshall them into the wrapper struct
+	cfgWrapper := &ConfigInputWrapper{}
+	err = yaml.Unmarshal(file, cfgWrapper)
+	if err != nil {
+		fmt.Fprintln(cmd.ErrOrStderr(), "Error unmarshalling config:", err)
+		return err
+	}
 
-			// verify that the channel filter rule is valid for every path in the config
-			for _, p := range cfgWrapper.Paths {
-				if err := p.ValidateChannelFilterRule(); err != nil {
-					return fmt.Errorf("error initializing the relayer config for path %s: %w", p.String(), err)
-				}
-			}
+	// verify that the channel filter rule is valid for every path in the config
+	for _, p := range cfgWrapper.Paths {
+		if err := p.ValidateChannelFilterRule(); err != nil {
+			return fmt.Errorf("error initializing the relayer config for path %s: %w", p.String(), err)
+		}
+	}
 
-			// build the config struct
-			chains := make(relayer.Chains)
-			for chainName, pcfg := range cfgWrapper.ProviderConfigs {
-				prov, err := pcfg.Value.(provider.ProviderConfig).NewProvider(
-					a.Log.With(zap.String("provider_type", pcfg.Type)),
-					a.HomePath, a.Debug, chainName,
-				)
-				if err != nil {
-					return fmt.Errorf("failed to build ChainProviders: %w", err)
-				}
+	// build the config struct
+	chains := make(relayer.Chains)
+	for chainName, pcfg := range cfgWrapper.ProviderConfigs {
+		prov, err := pcfg.Value.(provider.ProviderConfig).NewProvider(
+			a.Log.With(zap.String("provider_type", pcfg.Type)),
+			a.HomePath, a.Debug, chainName,
+		)
+		if err != nil {
+			return fmt.Errorf("failed to build ChainProviders: %w", err)
+		}
 
-				chain := relayer.NewChain(a.Log, prov, a.Debug)
-				chains[chainName] = chain
-			}
+		chain := relayer.NewChain(a.Log, prov, a.Debug)
+		chains[chainName] = chain
+	}
 
-			a.Config = &Config{
-				Global: cfgWrapper.Global,
-				Chains: chains,
-				Paths:  cfgWrapper.Paths,
-			}
+	a.Config = &Config{
+		Global: cfgWrapper.Global,
+		Chains: chains,
+		Paths:  cfgWrapper.Paths,
+	}
 
-			// ensure config has []*relayer.Chain used for all chain operations
-			if err := validateConfig(a.Config); err != nil {
-				fmt.Fprintln(cmd.ErrOrStderr(), "Error parsing chain config:", err)
-				return err
-			}
-		}
+	// ensure config has []*relayer.Chain used for all chain operations
+	if err := validateConfig(a.Config); err != nil {
+		fmt.Fprintln(cmd.ErrOrStderr(), "Error parsing chain config:", err)
+		return err
 	}
+
 	return nil
 }
 
diff --git a/cmd/start.go b/cmd/start.go
index 2a5ae98..b2f7330 100644
--- a/cmd/start.go
+++ b/cmd/start.go
@@ -29,6 +29,7 @@ import (
 	"github.com/avast/retry-go/v4"
 	"github.com/cosmos/relayer/v2/internal/relaydebug"
 	"github.com/cosmos/relayer/v2/relayer"
+	"github.com/cosmos/relayer/v2/relayer/processor"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -47,7 +48,8 @@ $ %s start demo-path -p events # to use event processor
 $ %s start demo-path --max-msgs 3
 $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			c, src, dst, err := a.Config.ChainsFromPath(args[0])
+			pathName := args[0]
+			c, src, dst, err := a.Config.ChainsFromPath(pathName)
 			if err != nil {
 				return err
 			}
@@ -56,7 +58,7 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 				return err
 			}
 
-			path := a.Config.Paths.MustGet(args[0])
+			path := a.Config.Paths.MustGet(pathName)
 
 			maxTxSize, maxMsgLength, err := GetStartOptions(cmd)
 			if err != nil {
@@ -65,6 +67,8 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 
 			filter := path.Filter
 
+			var prometheusMetrics *processor.PrometheusMetrics
+
 			debugAddr, err := cmd.Flags().GetString(flagDebugAddr)
 			if err != nil {
 				return err
@@ -80,6 +84,7 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 				log := a.Log.With(zap.String("sys", "debughttp"))
 				log.Info("Debug server listening", zap.String("addr", debugAddr))
 				relaydebug.StartDebugServer(cmd.Context(), log, ln)
+				prometheusMetrics = processor.NewPrometheusMetrics()
 			}
 
 			processorType, err := cmd.Flags().GetString(flagProcessor)
@@ -91,7 +96,17 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 				return err
 			}
 
-			rlyErrCh := relayer.StartRelayer(cmd.Context(), a.Log, c[src], c[dst], filter, maxTxSize, maxMsgLength, a.Config.memo(cmd), processorType, initialBlockHistory)
+			rlyErrCh := relayer.StartRelayer(
+				cmd.Context(),
+				a.Log,
+				c[src], c[dst],
+				filter,
+				maxTxSize, maxMsgLength,
+				a.Config.memo(cmd),
+				processorType, initialBlockHistory,
+				pathName,
+				prometheusMetrics,
+			)
 
 			// NOTE: This block of code is useful for ensuring that the clients tracking each chain do not expire
 			// when there are no packets flowing across the channels. It is currently a source of errors that have been
diff --git a/cmd/tx.go b/cmd/tx.go
index e70d1cc..b8b3de9 100644
--- a/cmd/tx.go
+++ b/cmd/tx.go
@@ -376,7 +376,9 @@ $ %s tx conn demo-path --timeout 5s`,
 				return err
 			}
 
-			c, src, dst, err := a.Config.ChainsFromPath(args[0])
+			pathName := args[0]
+
+			c, src, dst, err := a.Config.ChainsFromPath(pathName)
 			if err != nil {
 				return err
 			}
@@ -422,7 +424,7 @@ $ %s tx conn demo-path --timeout 5s`,
 				}
 			}
 
-			modified, err = c[src].CreateOpenConnections(cmd.Context(), c[dst], retries, to, memo, initialBlockHistory)
+			modified, err = c[src].CreateOpenConnections(cmd.Context(), c[dst], retries, to, memo, initialBlockHistory, pathName)
 			if err != nil {
 				return err
 			}
@@ -460,7 +462,10 @@ $ %s tx chan demo-path --timeout 5s --max-retries 10`,
 			appName, appName,
 		)),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			c, src, dst, err := a.Config.ChainsFromPath(args[0])
+
+			pathName := args[0]
+
+			c, src, dst, err := a.Config.ChainsFromPath(pathName)
 			if err != nil {
 				return err
 			}
@@ -509,7 +514,7 @@ $ %s tx chan demo-path --timeout 5s --max-retries 10`,
 			}
 
 			// create channel if it isn't already created
-			return c[src].CreateOpenChannels(cmd.Context(), c[dst], retries, to, srcPort, dstPort, order, version, override, a.Config.memo(cmd))
+			return c[src].CreateOpenChannels(cmd.Context(), c[dst], retries, to, srcPort, dstPort, order, version, override, a.Config.memo(cmd), pathName)
 		},
 	}
 
@@ -534,7 +539,9 @@ $ %s tx channel-close demo-path channel-0 transfer -o 3s`,
 			appName, appName, appName, appName,
 		)),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			c, src, dst, err := a.Config.ChainsFromPath(args[0])
+			pathName := args[0]
+
+			c, src, dst, err := a.Config.ChainsFromPath(pathName)
 			if err != nil {
 				return err
 			}
@@ -570,7 +577,7 @@ $ %s tx channel-close demo-path channel-0 transfer -o 3s`,
 				return err
 			}
 
-			return c[src].CloseChannel(cmd.Context(), c[dst], retries, to, channelID, portID, a.Config.memo(cmd))
+			return c[src].CloseChannel(cmd.Context(), c[dst], retries, to, channelID, portID, a.Config.memo(cmd), pathName)
 		},
 	}
 
@@ -611,7 +618,9 @@ $ %s tx connect demo-path --src-port transfer --dst-port transfer --order unorde
 				return err
 			}
 
-			pth, err := a.Config.Paths.Get(args[0])
+			pathName := args[0]
+
+			pth, err := a.Config.Paths.Get(pathName)
 			if err != nil {
 				return err
 			}
@@ -687,7 +696,7 @@ $ %s tx connect demo-path --src-port transfer --dst-port transfer --order unorde
 			}
 
 			// create connection if it isn't already created
-			modified, err = c[src].CreateOpenConnections(cmd.Context(), c[dst], retries, to, memo, initialBlockHistory)
+			modified, err = c[src].CreateOpenConnections(cmd.Context(), c[dst], retries, to, memo, initialBlockHistory, pathName)
 			if err != nil {
 				return fmt.Errorf("error creating connections: %w", err)
 			}
@@ -698,7 +707,7 @@ $ %s tx connect demo-path --src-port transfer --dst-port transfer --order unorde
 			}
 
 			// create channel if it isn't already created
-			return c[src].CreateOpenChannels(cmd.Context(), c[dst], retries, to, srcPort, dstPort, order, version, override, memo)
+			return c[src].CreateOpenChannels(cmd.Context(), c[dst], retries, to, srcPort, dstPort, order, version, override, memo, pathName)
 		},
 	}
 	cmd = timeoutFlag(a.Viper, cmd)
diff --git a/docs/monitoring.md b/docs/monitoring.md
new file mode 100644
index 0000000..de3aae2
--- /dev/null
+++ b/docs/monitoring.md
@@ -0,0 +1,26 @@
+# Monitoring
+
+**Prometheus exporter**
+
+If you started `rly` with the default `--debug-addr` argument,
+you can use `http://$IP:7597/metrics` as a target for your prometheus scraper.
+
+**Example metrics**
+
+```
+go_goroutines 29
+...
+go_threads 39
+...
+observed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="acknowledge_packet"} 57
+observed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="recv_packet"} 103
+observed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="send_packet"} 58
+observed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="acknowledge_packet"} 107
+observed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="recv_packet"} 60
+observed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="send_packet"} 102
+...
+relayed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="acknowledge_packet"} 31
+relayed_packets{chain="cosmoshub-4",channel="channel-141",path="hubosmo",port="transfer",type="recv_packet"} 65
+relayed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="acknowledge_packet"} 36
+relayed_packets{chain="osmosis-1",channel="channel-0",path="hubosmo",port="transfer",type="recv_packet"} 35
+```
diff --git a/go.mod b/go.mod
index 48d60f0..80ffb47 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,6 @@ go 1.18
 
 require (
 	github.com/avast/retry-go/v4 v4.1.0
-	github.com/cespare/permute/v2 v2.0.0-beta2
 	github.com/cosmos/cosmos-sdk v0.45.6
 	github.com/cosmos/ibc-go/v4 v4.0.0-rc0
 	github.com/gogo/protobuf v1.3.3
@@ -13,6 +12,7 @@ require (
 	github.com/gorilla/mux v1.8.0 // indirect
 	github.com/jsternberg/zap-logfmt v1.2.0
 	github.com/ory/dockertest/v3 v3.9.1
+	github.com/prometheus/client_golang v1.12.2
 	github.com/spf13/cobra v1.5.0
 	github.com/spf13/viper v1.12.0
 	github.com/strangelove-ventures/lens v0.5.2-0.20220725181212-615bf7a39fe6
@@ -68,7 +68,6 @@ require (
 	github.com/ethereum/go-ethereum v1.10.16 // indirect
 	github.com/felixge/httpsnoop v1.0.1 // indirect
 	github.com/fsnotify/fsnotify v1.5.4 // indirect
-	github.com/gin-gonic/gin v1.7.0 // indirect
 	github.com/go-kit/kit v0.12.0 // indirect
 	github.com/go-kit/log v0.2.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
@@ -116,7 +115,6 @@ require (
 	github.com/petermattis/goid v0.0.0-20180202154549-b0b1615b78e5 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.12.2 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
diff --git a/go.sum b/go.sum
index 8412c91..011f46a 100644
--- a/go.sum
+++ b/go.sum
@@ -211,8 +211,6 @@ github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8
 github.com/cenkalti/backoff/v4 v4.1.3/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/cp v0.1.0/go.mod h1:SOGHArjBr4JWaSDEVpWpo/hNg6RoKrls6Oh40hiwW+s=
-github.com/cespare/permute/v2 v2.0.0-beta2 h1:iiJWgDsInCbnwfDZiLJX7cVlJevgIM1pUiArcUylUMc=
-github.com/cespare/permute/v2 v2.0.0-beta2/go.mod h1:7e2Tqe0fjn1T4rTjDLJjQ+kNtt+UW4B9lg10asqWw0A=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -398,9 +396,8 @@ github.com/getkin/kin-openapi v0.61.0/go.mod h1:7Yn5whZr5kJi6t+kShccXS8ae1APpYTW
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
+github.com/gin-gonic/gin v1.6.3 h1:ahKqKTFpO5KTPHxWZjEdPScmYaGtLo8Y4DMHoEsnp14=
 github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M=
-github.com/gin-gonic/gin v1.7.0 h1:jGB9xAJQ12AIGNB4HguylppmDK1Am9ppF7XnGXXJuoU=
-github.com/gin-gonic/gin v1.7.0/go.mod h1:jD2toBW3GZUr5UMcdrwQA10I7RuaFOl/SGeDjXkfUtY=
 github.com/glycerine/go-unsnap-stream v0.0.0-20180323001048-9f0cb55181dd/go.mod h1:/20jfyN9Y5QPEAprSgKAUr+glWDY39ZiUEAYOEv5dsE=
 github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31/go.mod h1:Ogl1Tioa0aV7gstGFO7KhffUsb9M4ydbEbbxpcEDc24=
 github.com/go-chi/chi/v5 v5.0.0/go.mod h1:BBug9lr0cqtdAhsu6R4AAdvufI0/XBzAQSsUqJpoZOs=
@@ -432,7 +429,6 @@ github.com/go-playground/universal-translator v0.17.0 h1:icxd5fm+REJzpZx7ZfpaD87
 github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
 github.com/go-playground/validator/v10 v10.2.0/go.mod h1:uOYAAleCW8F/7oMFd6aG0GOhaH6EGOAJShg8Id5JGkI=
 github.com/go-playground/validator/v10 v10.4.1 h1:pH2c5ADXtd66mxoE0Zm9SUhxE20r7aM3F26W0hOn+GE=
-github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
 github.com/go-sourcemap/sourcemap v2.1.2+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
diff --git a/internal/relaydebug/debugserver.go b/internal/relaydebug/debugserver.go
index 2ae8aae..295beff 100644
--- a/internal/relaydebug/debugserver.go
+++ b/internal/relaydebug/debugserver.go
@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"net/http/pprof"
 
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
 )
 
@@ -31,6 +32,9 @@ func StartDebugServer(ctx context.Context, log *zap.Logger, ln net.Listener) {
 	// so operators don't see a mysterious 404 page.
 	mux.Handle("/", http.RedirectHandler("/debug/pprof", http.StatusSeeOther))
 
+	// Serve prometheus metrics
+	mux.Handle("/metrics", promhttp.Handler())
+
 	srv := &http.Server{
 		Handler:  mux,
 		ErrorLog: zap.NewStdLog(log),
diff --git a/relayer/chains/mock/message_handlers.go b/relayer/chains/mock/message_handlers.go
index d99b100..bd6ca91 100644
--- a/relayer/chains/mock/message_handlers.go
+++ b/relayer/chains/mock/message_handlers.go
@@ -30,7 +30,11 @@ func handleMsgTransfer(p msgHandlerParams) {
 		CounterpartyChannelID: p.packetInfo.DestinationChannel,
 		CounterpartyPortID:    p.packetInfo.DestinationPort,
 	}
-	p.ibcMessagesCache.PacketFlow.Retain(channelKey, chantypes.EventTypeSendPacket, provider.PacketInfo{Sequence: p.packetInfo.Sequence})
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, chantypes.EventTypeSendPacket, provider.PacketInfo{
+		Sequence:      p.packetInfo.Sequence,
+		Data:          p.packetInfo.Data,
+		TimeoutHeight: p.packetInfo.TimeoutHeight,
+	})
 	p.mcp.log.Debug("observed MsgTransfer",
 		zap.String("chain_id", p.mcp.chainID),
 		zap.Uint64("sequence", p.packetInfo.Sequence),
@@ -48,7 +52,10 @@ func handleMsgRecvPacket(p msgHandlerParams) {
 		CounterpartyChannelID: p.packetInfo.SourceChannel,
 		CounterpartyPortID:    p.packetInfo.SourcePort,
 	}
-	p.ibcMessagesCache.PacketFlow.Retain(channelKey, chantypes.EventTypeRecvPacket, provider.PacketInfo{Sequence: p.packetInfo.Sequence})
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, chantypes.EventTypeRecvPacket, provider.PacketInfo{
+		Sequence: p.packetInfo.Sequence,
+		Data:     p.packetInfo.Data,
+	})
 	p.mcp.log.Debug("observed MsgRecvPacket",
 		zap.String("chain_id", p.mcp.chainID),
 		zap.Uint64("sequence", p.packetInfo.Sequence),
@@ -66,7 +73,10 @@ func handleMsgAcknowledgement(p msgHandlerParams) {
 		CounterpartyChannelID: p.packetInfo.DestinationChannel,
 		CounterpartyPortID:    p.packetInfo.DestinationPort,
 	}
-	p.ibcMessagesCache.PacketFlow.Retain(channelKey, chantypes.EventTypeAcknowledgePacket, provider.PacketInfo{Sequence: p.packetInfo.Sequence})
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, chantypes.EventTypeAcknowledgePacket, provider.PacketInfo{
+		Sequence: p.packetInfo.Sequence,
+		Data:     p.packetInfo.Data,
+	})
 	p.mcp.log.Debug("observed MsgAcknowledgement",
 		zap.String("chain_id", p.mcp.chainID),
 		zap.Uint64("sequence", p.packetInfo.Sequence),
diff --git a/relayer/chains/mock/mock_chain_processor.go b/relayer/chains/mock/mock_chain_processor.go
index 04e8c09..3e33ed0 100644
--- a/relayer/chains/mock/mock_chain_processor.go
+++ b/relayer/chains/mock/mock_chain_processor.go
@@ -47,7 +47,7 @@ func NewMockChainProcessor(log *zap.Logger, chainID string, getMockMessages func
 		KeyringBackend: "test",
 		Timeout:        "10s",
 	}
-	chainProvider, _ := chainProviderCfg.NewProvider(zap.NewNop(), "/tmp", true, "mock-chain-name")
+	chainProvider, _ := chainProviderCfg.NewProvider(zap.NewNop(), "/tmp", true, "mock-chain-name-"+chainID)
 	_, _ = chainProvider.AddKey(chainProvider.Key(), 118)
 	return &MockChainProcessor{
 		log:             log,
diff --git a/relayer/channel.go b/relayer/channel.go
index cad7b8d..e5beca1 100644
--- a/relayer/channel.go
+++ b/relayer/channel.go
@@ -21,6 +21,7 @@ func (c *Chain) CreateOpenChannels(
 	srcPortID, dstPortID, order, version string,
 	override bool,
 	memo string,
+	pathName string,
 ) error {
 	// client and connection identifiers must be filled in
 	if err := ValidateConnectionPaths(c, dst); err != nil {
@@ -46,11 +47,11 @@ func (c *Chain) CreateOpenChannels(
 
 	srcPathChain := pathChain{
 		provider: c.ChainProvider,
-		pathEnd:  processor.NewPathEnd(c.PathEnd.ChainID, c.PathEnd.ClientID, "", []processor.ChannelKey{}),
+		pathEnd:  processor.NewPathEnd(pathName, c.PathEnd.ChainID, c.PathEnd.ClientID, "", []processor.ChannelKey{}),
 	}
 	dstPathChain := pathChain{
 		provider: dst.ChainProvider,
-		pathEnd:  processor.NewPathEnd(dst.PathEnd.ChainID, dst.PathEnd.ClientID, "", []processor.ChannelKey{}),
+		pathEnd:  processor.NewPathEnd(pathName, dst.PathEnd.ChainID, dst.PathEnd.ClientID, "", []processor.ChannelKey{}),
 	}
 
 	// Timeout is per message. Four channel handshake messages, allowing maxRetries for each.
@@ -63,6 +64,7 @@ func (c *Chain) CreateOpenChannels(
 		c.log,
 		srcPathChain.pathEnd,
 		dstPathChain.pathEnd,
+		nil,
 		memo,
 	)
 
@@ -114,14 +116,15 @@ func (c *Chain) CloseChannel(
 	srcChanID,
 	srcPortID string,
 	memo string,
+	pathName string,
 ) error {
 	srcPathChain := pathChain{
 		provider: c.ChainProvider,
-		pathEnd:  processor.NewPathEnd(c.PathEnd.ChainID, c.PathEnd.ClientID, "", []processor.ChannelKey{}),
+		pathEnd:  processor.NewPathEnd(pathName, c.PathEnd.ChainID, c.PathEnd.ClientID, "", []processor.ChannelKey{}),
 	}
 	dstPathChain := pathChain{
 		provider: dst.ChainProvider,
-		pathEnd:  processor.NewPathEnd(dst.PathEnd.ChainID, dst.PathEnd.ClientID, "", []processor.ChannelKey{}),
+		pathEnd:  processor.NewPathEnd(pathName, dst.PathEnd.ChainID, dst.PathEnd.ClientID, "", []processor.ChannelKey{}),
 	}
 
 	// Timeout is per message. Two close channel handshake messages, allowing maxRetries for each.
@@ -139,6 +142,7 @@ func (c *Chain) CloseChannel(
 			c.log,
 			srcPathChain.pathEnd,
 			dstPathChain.pathEnd,
+			nil,
 			memo,
 		)).
 		WithInitialBlockHistory(0).
diff --git a/relayer/connection.go b/relayer/connection.go
index ef39e7d..3b79fa6 100644
--- a/relayer/connection.go
+++ b/relayer/connection.go
@@ -19,6 +19,7 @@ func (c *Chain) CreateOpenConnections(
 	timeout time.Duration,
 	memo string,
 	initialBlockHistory uint64,
+	pathName string,
 ) (modified bool, err error) {
 	// client identifiers must be filled in
 	if err = ValidateClientPaths(c, dst); err != nil {
@@ -27,11 +28,11 @@ func (c *Chain) CreateOpenConnections(
 
 	srcpathChain := pathChain{
 		provider: c.ChainProvider,
-		pathEnd:  processor.NewPathEnd(c.PathEnd.ChainID, c.PathEnd.ClientID, "", []processor.ChannelKey{}),
+		pathEnd:  processor.NewPathEnd(pathName, c.PathEnd.ChainID, c.PathEnd.ClientID, "", []processor.ChannelKey{}),
 	}
 	dstpathChain := pathChain{
 		provider: dst.ChainProvider,
-		pathEnd:  processor.NewPathEnd(dst.PathEnd.ChainID, dst.PathEnd.ClientID, "", []processor.ChannelKey{}),
+		pathEnd:  processor.NewPathEnd(pathName, dst.PathEnd.ChainID, dst.PathEnd.ClientID, "", []processor.ChannelKey{}),
 	}
 
 	// Timeout is per message. Four connection handshake messages, allowing maxRetries for each.
@@ -44,6 +45,7 @@ func (c *Chain) CreateOpenConnections(
 		c.log,
 		srcpathChain.pathEnd,
 		dstpathChain.pathEnd,
+		nil,
 		memo,
 	)
 
diff --git a/relayer/processor/path_end.go b/relayer/processor/path_end.go
index c226231..73be970 100644
--- a/relayer/processor/path_end.go
+++ b/relayer/processor/path_end.go
@@ -3,6 +3,7 @@ package processor
 // PathEnd references one chain involved in a path.
 // A path is composed of two PathEnds.
 type PathEnd struct {
+	PathName string
 	ChainID  string
 	ClientID string
 	// ConnectionIDs are tracked by pathEndRuntime in PathProcessor for known connections on this client
@@ -13,8 +14,9 @@ type PathEnd struct {
 }
 
 // NewPathEnd constructs a PathEnd, validating initial parameters.
-func NewPathEnd(chainID string, clientID string, rule string, filterList []ChannelKey) PathEnd {
+func NewPathEnd(pathName string, chainID string, clientID string, rule string, filterList []ChannelKey) PathEnd {
 	return PathEnd{
+		PathName:   pathName,
 		ChainID:    chainID,
 		ClientID:   clientID,
 		Rule:       rule,
diff --git a/relayer/processor/path_end_runtime.go b/relayer/processor/path_end_runtime.go
index 78d2256..aa9a2da 100644
--- a/relayer/processor/path_end_runtime.go
+++ b/relayer/processor/path_end_runtime.go
@@ -43,11 +43,14 @@ type pathEndRuntime struct {
 
 	// inSync indicates whether queries are in sync with latest height of the chain.
 	inSync bool
+
+	metrics *PrometheusMetrics
 }
 
-func newPathEndRuntime(log *zap.Logger, pathEnd PathEnd) *pathEndRuntime {
+func newPathEndRuntime(log *zap.Logger, pathEnd PathEnd, metrics *PrometheusMetrics) *pathEndRuntime {
 	return &pathEndRuntime{
 		log: log.With(
+			zap.String("path_name", pathEnd.PathName),
 			zap.String("chain_id", pathEnd.ChainID),
 			zap.String("client_id", pathEnd.ClientID),
 		),
@@ -61,6 +64,7 @@ func newPathEndRuntime(log *zap.Logger, pathEnd PathEnd) *pathEndRuntime {
 		connProcessing:       make(connectionProcessingCache),
 		channelProcessing:    make(channelProcessingCache),
 		connSubscribers:      make(map[string][]func(provider.ConnectionInfo)),
+		metrics:              metrics,
 	}
 }
 
@@ -83,13 +87,20 @@ func (pathEnd *pathEndRuntime) isRelevantChannel(channelID string) bool {
 }
 
 // mergeMessageCache merges relevant IBC messages for packet flows, connection handshakes, and channel handshakes.
-func (pathEnd *pathEndRuntime) mergeMessageCache(messageCache IBCMessagesCache) {
+// inSync indicates whether both involved ChainProcessors are in sync or not. When true, the observed packets
+// metrics will be counted so that observed vs relayed packets can be compared.
+func (pathEnd *pathEndRuntime) mergeMessageCache(messageCache IBCMessagesCache, inSync bool) {
 	packetMessages := make(ChannelPacketMessagesCache)
 	connectionHandshakeMessages := make(ConnectionMessagesCache)
 	channelHandshakeMessages := make(ChannelMessagesCache)
 
 	for ch, pmc := range messageCache.PacketFlow {
 		if pathEnd.info.ShouldRelayChannel(ch) {
+			if inSync && pathEnd.metrics != nil {
+				for eventType, pCache := range pmc {
+					pathEnd.metrics.AddPacketsObserved(pathEnd.info.PathName, pathEnd.info.ChainID, ch.ChannelID, ch.PortID, eventType, len(pCache))
+				}
+			}
 			packetMessages[ch] = pmc
 		}
 	}
@@ -260,7 +271,7 @@ func (pathEnd *pathEndRuntime) shouldTerminate(ibcMessagesCache IBCMessagesCache
 	return false
 }
 
-func (pathEnd *pathEndRuntime) mergeCacheData(ctx context.Context, cancel func(), d ChainProcessorCacheData, messageLifecycle MessageLifecycle) {
+func (pathEnd *pathEndRuntime) mergeCacheData(ctx context.Context, cancel func(), d ChainProcessorCacheData, counterpartyInSync bool, messageLifecycle MessageLifecycle) {
 	pathEnd.inSync = d.InSync
 	pathEnd.latestBlock = d.LatestBlock
 	pathEnd.latestHeader = d.LatestHeader
@@ -276,7 +287,7 @@ func (pathEnd *pathEndRuntime) mergeCacheData(ctx context.Context, cancel func()
 	pathEnd.connectionStateCache.Merge(d.ConnectionStateCache) // Update latest connection open state for chain
 	pathEnd.channelStateCache.Merge(d.ChannelStateCache)       // Update latest channel open state for chain
 
-	pathEnd.mergeMessageCache(d.IBCMessagesCache) // Merge incoming packet IBC messages into the backlog
+	pathEnd.mergeMessageCache(d.IBCMessagesCache, pathEnd.inSync && counterpartyInSync) // Merge incoming packet IBC messages into the backlog
 
 	pathEnd.ibcHeaderCache.Merge(d.IBCHeaderCache)  // Update latest IBC header state
 	pathEnd.ibcHeaderCache.Prune(ibcHeadersToCache) // Only keep most recent IBC headers
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index ea7ca91..9a6100e 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -17,7 +17,7 @@ const (
 	// Amount of time to wait when sending transactions before giving up
 	// and continuing on. Messages will be retried later if they are still
 	// relevant.
-	messageSendTimeout = 10 * time.Second
+	messageSendTimeout = 60 * time.Second
 
 	// Amount of time to wait for a proof to be queried before giving up.
 	// The proof query will be retried later if the message still needs
@@ -31,7 +31,7 @@ const (
 
 	// If the message was assembled successfully, but sending the message failed,
 	// how many blocks should pass before retrying.
-	blocksToRetrySendAfter = 2
+	blocksToRetrySendAfter = 5
 
 	// How many times to retry sending a message before giving up on it.
 	maxMessageSendRetries = 5
@@ -59,6 +59,8 @@ type PathProcessor struct {
 	retryProcess chan struct{}
 
 	sentInitialMsg bool
+
+	metrics *PrometheusMetrics
 }
 
 // PathProcessors is a slice of PathProcessor instances
@@ -73,13 +75,20 @@ func (p PathProcessors) IsRelayedChannel(k ChannelKey, chainID string) bool {
 	return false
 }
 
-func NewPathProcessor(log *zap.Logger, pathEnd1 PathEnd, pathEnd2 PathEnd, memo string) *PathProcessor {
+func NewPathProcessor(
+	log *zap.Logger,
+	pathEnd1 PathEnd,
+	pathEnd2 PathEnd,
+	metrics *PrometheusMetrics,
+	memo string,
+) *PathProcessor {
 	return &PathProcessor{
 		log:          log,
-		pathEnd1:     newPathEndRuntime(log, pathEnd1),
-		pathEnd2:     newPathEndRuntime(log, pathEnd2),
+		pathEnd1:     newPathEndRuntime(log, pathEnd1, metrics),
+		pathEnd2:     newPathEndRuntime(log, pathEnd2, metrics),
 		retryProcess: make(chan struct{}, 2),
 		memo:         memo,
+		metrics:      metrics,
 	}
 }
 
@@ -231,11 +240,11 @@ func (pp *PathProcessor) processAvailableSignals(ctx context.Context, cancel fun
 		return true
 	case d := <-pp.pathEnd1.incomingCacheData:
 		// we have new data from ChainProcessor for pathEnd1
-		pp.pathEnd1.mergeCacheData(ctx, cancel, d, messageLifecycle)
+		pp.pathEnd1.mergeCacheData(ctx, cancel, d, pp.pathEnd2.inSync, messageLifecycle)
 
 	case d := <-pp.pathEnd2.incomingCacheData:
 		// we have new data from ChainProcessor for pathEnd2
-		pp.pathEnd2.mergeCacheData(ctx, cancel, d, messageLifecycle)
+		pp.pathEnd2.mergeCacheData(ctx, cancel, d, pp.pathEnd1.inSync, messageLifecycle)
 
 	case <-pp.retryProcess:
 		// No new data to merge in, just retry handling.
diff --git a/relayer/processor/path_processor_internal.go b/relayer/processor/path_processor_internal.go
index 7f56d37..18b5ebb 100644
--- a/relayer/processor/path_processor_internal.go
+++ b/relayer/processor/path_processor_internal.go
@@ -619,36 +619,14 @@ func (pp *PathProcessor) processLatestMessages(ctx context.Context, messageLifec
 	// if sending messages fails to one pathEnd, we don't need to halt sending to the other pathEnd.
 	var eg errgroup.Group
 	eg.Go(func() error {
-		if err := pp.assembleAndSendMessages(ctx, pp.pathEnd2, pp.pathEnd1, pathEnd1Messages); err != nil {
-			pp.logFailedTx(pp.pathEnd2.info, pp.pathEnd1.info, err)
-			return err
-		}
-		return nil
+		return pp.assembleAndSendMessages(ctx, pp.pathEnd2, pp.pathEnd1, pathEnd1Messages)
 	})
 	eg.Go(func() error {
-		if err := pp.assembleAndSendMessages(ctx, pp.pathEnd1, pp.pathEnd2, pathEnd2Messages); err != nil {
-			pp.logFailedTx(pp.pathEnd1.info, pp.pathEnd2.info, err)
-			return err
-		}
-		return nil
+		return pp.assembleAndSendMessages(ctx, pp.pathEnd1, pp.pathEnd2, pathEnd2Messages)
 	})
 	return eg.Wait()
 }
 
-func (pp *PathProcessor) logFailedTx(src, dst PathEnd, err error) {
-	if errors.Is(err, chantypes.ErrRedundantTx) {
-		pp.log.Debug("Packet(s) already handled by another relayer")
-		return
-	}
-	pp.log.Error("Error sending messages",
-		zap.String("src_chain_id", src.ChainID),
-		zap.String("dst_chain_id", dst.ChainID),
-		zap.String("src_client_id", src.ClientID),
-		zap.String("dst_client_id", dst.ClientID),
-		zap.Error(err),
-	)
-}
-
 func (pp *PathProcessor) assembleMessage(
 	ctx context.Context,
 	msg ibcMessage,
@@ -747,18 +725,55 @@ func (pp *PathProcessor) assembleAndSendMessages(
 		dst.trackProcessingChannelMessage(m)
 	}
 
+	go pp.sendMessages(ctx, src, dst, om, pp.memo)
+
+	return nil
+}
+
+func (pp *PathProcessor) sendMessages(ctx context.Context, src, dst *pathEndRuntime, om outgoingMessages, memo string) {
 	ctx, cancel := context.WithTimeout(ctx, messageSendTimeout)
 	defer cancel()
 
 	_, txSuccess, err := dst.chainProvider.SendMessages(ctx, om.msgs, pp.memo)
 	if err != nil {
-		return fmt.Errorf("error sending messages: %w", err)
+		if errors.Is(err, chantypes.ErrRedundantTx) {
+			pp.log.Debug("Packet(s) already handled by another relayer",
+				zap.String("src_chain_id", src.info.ChainID),
+				zap.String("dst_chain_id", dst.info.ChainID),
+				zap.String("src_client_id", src.info.ClientID),
+				zap.String("dst_client_id", dst.info.ClientID),
+				zap.Error(err),
+			)
+			return
+		}
+		pp.log.Error("Error sending messages",
+			zap.String("src_chain_id", src.info.ChainID),
+			zap.String("dst_chain_id", dst.info.ChainID),
+			zap.String("src_client_id", src.info.ClientID),
+			zap.String("dst_client_id", dst.info.ClientID),
+			zap.Error(err),
+		)
+		return
 	}
 	if !txSuccess {
-		return errors.New("error sending messages, transeventType was not successful")
+		dst.log.Error("Error sending messages, transaction was not successful")
+		return
 	}
 
-	return nil
+	if pp.metrics == nil {
+		return
+	}
+	for _, m := range om.pktMsgs {
+		var channel, port string
+		if m.msg.eventType == chantypes.EventTypeRecvPacket {
+			channel = m.msg.info.DestChannel
+			port = m.msg.info.DestPort
+		} else {
+			channel = m.msg.info.SourceChannel
+			port = m.msg.info.SourcePort
+		}
+		pp.metrics.IncPacketsRelayed(dst.info.PathName, dst.info.ChainID, channel, port, m.msg.eventType)
+	}
 }
 
 func (pp *PathProcessor) assemblePacketMessage(
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
index bad80b8..d447b32 100644
--- a/relayer/processor/types.go
+++ b/relayer/processor/types.go
@@ -6,6 +6,8 @@ import (
 
 	chantypes "github.com/cosmos/ibc-go/v4/modules/core/04-channel/types"
 	"github.com/cosmos/relayer/v2/relayer/provider"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"go.uber.org/zap/zapcore"
 )
 
@@ -465,3 +467,30 @@ func ConnectionInfoConnectionKey(info provider.ConnectionInfo) ConnectionKey {
 		CounterpartyConnID:   info.CounterpartyConnID,
 	}
 }
+
+type PrometheusMetrics struct {
+	PacketObservedCounter *prometheus.CounterVec
+	PacketRelayedCounter  *prometheus.CounterVec
+}
+
+func (m *PrometheusMetrics) AddPacketsObserved(path, chain, channel, port, eventType string, count int) {
+	m.PacketObservedCounter.WithLabelValues(path, chain, channel, port, eventType).Add(float64(count))
+}
+
+func (m *PrometheusMetrics) IncPacketsRelayed(path, chain, channel, port, eventType string) {
+	m.PacketRelayedCounter.WithLabelValues(path, chain, channel, port, eventType).Inc()
+}
+
+func NewPrometheusMetrics() *PrometheusMetrics {
+	packetLabels := []string{"path", "chain", "channel", "port", "type"}
+	return &PrometheusMetrics{
+		PacketObservedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "observed_packets",
+			Help: "The total number of observed packets",
+		}, packetLabels),
+		PacketRelayedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "relayed_packets",
+			Help: "The total number of relayed packets",
+		}, packetLabels),
+	}
+}
diff --git a/relayer/strategies.go b/relayer/strategies.go
index f418d89..12ac461 100644
--- a/relayer/strategies.go
+++ b/relayer/strategies.go
@@ -37,6 +37,8 @@ func StartRelayer(
 	memo string,
 	processorType string,
 	initialBlockHistory uint64,
+	pathName string,
+	metrics *processor.PrometheusMetrics,
 ) chan error {
 	errorChan := make(chan error, 1)
 
@@ -53,15 +55,15 @@ func StartRelayer(
 		paths := []path{{
 			src: pathChain{
 				provider: src.ChainProvider,
-				pathEnd:  processor.NewPathEnd(src.ChainProvider.ChainId(), src.ClientID(), filter.Rule, filterSrc),
+				pathEnd:  processor.NewPathEnd(pathName, src.ChainProvider.ChainId(), src.ClientID(), filter.Rule, filterSrc),
 			},
 			dst: pathChain{
 				provider: dst.ChainProvider,
-				pathEnd:  processor.NewPathEnd(dst.ChainProvider.ChainId(), dst.ClientID(), filter.Rule, filterDst),
+				pathEnd:  processor.NewPathEnd(pathName, dst.ChainProvider.ChainId(), dst.ClientID(), filter.Rule, filterDst),
 			},
 		}}
 
-		go relayerStartEventProcessor(ctx, log, paths, initialBlockHistory, maxTxSize, maxMsgLength, memo, errorChan)
+		go relayerStartEventProcessor(ctx, log, paths, initialBlockHistory, maxTxSize, maxMsgLength, memo, errorChan, metrics)
 		return errorChan
 	case ProcessorLegacy:
 		go relayerMainLoop(ctx, log, src, dst, filter, maxTxSize, maxMsgLength, memo, errorChan)
@@ -104,6 +106,7 @@ func relayerStartEventProcessor(
 	maxMsgLength uint64,
 	memo string,
 	errCh chan<- error,
+	metrics *processor.PrometheusMetrics,
 ) {
 	defer close(errCh)
 
@@ -119,6 +122,7 @@ func relayerStartEventProcessor(
 				log,
 				p.src.pathEnd,
 				p.dst.pathEnd,
+				metrics,
 				memo,
 			))
 	}


Test patch:
diff --git a/_test/relayer_chain_test.go b/_test/relayer_chain_test.go
index ea28273..9af5891 100644
--- a/_test/relayer_chain_test.go
+++ b/_test/relayer_chain_test.go
@@ -83,12 +83,12 @@ func chainTest(t *testing.T, tcs []testChain) {
 	require.NoError(t, err)
 
 	t.Log("Creating connections")
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
 	t.Log("Creating channels")
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 
 	t.Log("Querying open channels to ensure successful creation")
@@ -119,7 +119,7 @@ func chainTest(t *testing.T, tcs []testChain) {
 
 	t.Log("Starting relayer")
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20, "", nil)
 
 	t.Log("Waiting for relayer messages to reach both chains")
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 2))
@@ -171,11 +171,11 @@ func TestGaiaReuseIdentifiers(t *testing.T) {
 	timeout, err := src.GetTimeout()
 	require.NoError(t, err)
 
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 
 	// query open channels and ensure there is no error
@@ -198,11 +198,11 @@ func TestGaiaReuseIdentifiers(t *testing.T) {
 	require.NoError(t, err)
 	testClientPair(ctx, t, src, dst)
 
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 	testChannelPair(ctx, t, src, dst, channel.ChannelId, channel.PortId)
 
@@ -252,11 +252,11 @@ func TestGaiaMisbehaviourMonitoring(t *testing.T) {
 	timeout, err := src.GetTimeout()
 	require.NoError(t, err)
 
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 
 	// query open channels and ensure there is no error
@@ -269,7 +269,7 @@ func TestGaiaMisbehaviourMonitoring(t *testing.T) {
 	log := zaptest.NewLogger(t)
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20, "", nil)
 
 	// Wait for relay message inclusion in both chains
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 1))
@@ -393,15 +393,15 @@ func TestRelayAllChannelsOnConnection(t *testing.T) {
 	require.NoError(t, err)
 
 	t.Log("Creating connections")
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
 	t.Log("Creating channels")
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, true, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, true, "", "")
 	require.NoError(t, err)
 
 	t.Log("Ensuring two channels exist")
@@ -443,7 +443,7 @@ func TestRelayAllChannelsOnConnection(t *testing.T) {
 
 	t.Log("Starting relayer")
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20, "", nil)
 
 	t.Log("Waiting for relayer message inclusion in both chains")
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 1))
@@ -589,11 +589,11 @@ func TestUnorderedChannelBlockHeightTimeout(t *testing.T) {
 	timeout, err := src.GetTimeout()
 	require.NoError(t, err)
 
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 
 	// query open channels and ensure there is no error
@@ -619,7 +619,7 @@ func TestUnorderedChannelBlockHeightTimeout(t *testing.T) {
 
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20, "", nil)
 
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 5))
 
@@ -687,11 +687,11 @@ func TestUnorderedChannelTimestampTimeout(t *testing.T) {
 	timeout, err := src.GetTimeout()
 	require.NoError(t, err)
 
-	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0)
+	_, err = src.CreateOpenConnections(ctx, dst, 3, timeout, "", 0, "")
 	require.NoError(t, err)
 	testConnectionPair(ctx, t, src, dst)
 
-	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "")
+	err = src.CreateOpenChannels(ctx, dst, 3, timeout, DefaultSrcPortID, DefaultDstPortID, DefaultOrder, DefaultVersion, false, "", "")
 	require.NoError(t, err)
 
 	// query open channels and ensure there is no error
@@ -717,7 +717,7 @@ func TestUnorderedChannelTimestampTimeout(t *testing.T) {
 
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, "", relayer.ProcessorEvents, 20, "", nil)
 
 	time.Sleep(time.Second * 10)
 
diff --git a/relayer/chains/mock/mock_chain_processor_test.go b/relayer/chains/mock/mock_chain_processor_test.go
index 6770118..e9cc89b 100644
--- a/relayer/chains/mock/mock_chain_processor_test.go
+++ b/relayer/chains/mock/mock_chain_processor_test.go
@@ -11,6 +11,7 @@ import (
 	chantypes "github.com/cosmos/ibc-go/v4/modules/core/04-channel/types"
 	"github.com/cosmos/relayer/v2/relayer/chains/mock"
 	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
@@ -21,11 +22,12 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 		t.Skip()
 	}
 
+	mockPathName := "mockpath"
 	mockChainID1 := "mock-chain-1"
 	mockChainID2 := "mock-chain-2"
 
-	pathEnd1 := processor.PathEnd{ChainID: mockChainID1, ClientID: "mock-client-1"}
-	pathEnd2 := processor.PathEnd{ChainID: mockChainID2, ClientID: "mock-client-2"}
+	pathEnd1 := processor.PathEnd{PathName: mockPathName, ChainID: mockChainID1, ClientID: "mock-client-1"}
+	pathEnd2 := processor.PathEnd{PathName: mockPathName, ChainID: mockChainID2, ClientID: "mock-client-2"}
 
 	mockSequence1 := uint64(0)
 	mockSequence2 := uint64(0)
@@ -54,7 +56,9 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 	ctx, ctxCancel := context.WithTimeout(context.Background(), time.Second*20)
 	defer ctxCancel()
 
-	pathProcessor := processor.NewPathProcessor(log, pathEnd1, pathEnd2, "")
+	metrics := processor.NewPrometheusMetrics()
+
+	pathProcessor := processor.NewPathProcessor(log, pathEnd1, pathEnd2, metrics, "")
 
 	eventProcessor := processor.NewEventProcessor().
 		WithChainProcessors(
@@ -95,6 +99,32 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 	require.LessOrEqual(t, len(pathEnd2LeftoverMsgTransfer), 3)
 	require.LessOrEqual(t, len(pathEnd2LeftoverMsgRecvPacket), 2)
 	require.LessOrEqual(t, len(pathEnd2LeftoverMsgAcknowledgement), 1)
+
+	require.Equal(t, 6, testutil.CollectAndCount(metrics.PacketObservedCounter))
+
+	countChain1EventSend := testutil.ToFloat64(metrics.PacketObservedCounter.WithLabelValues(mockPathName, mockChainID1, mockChannelKey1.ChannelID, mockChannelKey1.PortID, chantypes.EventTypeSendPacket))
+	countChain1EventRecv := testutil.ToFloat64(metrics.PacketObservedCounter.WithLabelValues(mockPathName, mockChainID1, mockChannelKey1.ChannelID, mockChannelKey1.PortID, chantypes.EventTypeRecvPacket))
+	countChain1EventAck := testutil.ToFloat64(metrics.PacketObservedCounter.WithLabelValues(mockPathName, mockChainID1, mockChannelKey1.ChannelID, mockChannelKey1.PortID, chantypes.EventTypeAcknowledgePacket))
+
+	require.GreaterOrEqual(t, float64(mockSequence1), countChain1EventSend)
+	require.GreaterOrEqual(t, float64(mockSequence2-1), countChain1EventRecv)
+	require.GreaterOrEqual(t, float64(mockSequence1-2), countChain1EventAck)
+
+	require.Greater(t, countChain1EventSend, float64(0))
+	require.Greater(t, countChain1EventRecv, float64(0))
+	require.Greater(t, countChain1EventAck, float64(0))
+
+	countChain2EventSend := testutil.ToFloat64(metrics.PacketObservedCounter.WithLabelValues(mockPathName, mockChainID2, mockChannelKey2.ChannelID, mockChannelKey2.PortID, chantypes.EventTypeSendPacket))
+	countChain2EventRecv := testutil.ToFloat64(metrics.PacketObservedCounter.WithLabelValues(mockPathName, mockChainID2, mockChannelKey2.ChannelID, mockChannelKey2.PortID, chantypes.EventTypeRecvPacket))
+	countChain2EventAck := testutil.ToFloat64(metrics.PacketObservedCounter.WithLabelValues(mockPathName, mockChainID2, mockChannelKey2.ChannelID, mockChannelKey2.PortID, chantypes.EventTypeAcknowledgePacket))
+
+	require.GreaterOrEqual(t, float64(mockSequence2), countChain2EventSend)
+	require.GreaterOrEqual(t, float64(mockSequence1-1), countChain2EventRecv)
+	require.GreaterOrEqual(t, float64(mockSequence2-2), countChain2EventAck)
+
+	require.Greater(t, countChain2EventSend, float64(0))
+	require.Greater(t, countChain2EventRecv, float64(0))
+	require.Greater(t, countChain2EventAck, float64(0))
 }
 
 // will send cycles of:
@@ -104,7 +134,7 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 func getMockMessages(channelKey processor.ChannelKey, mockSequence, mockSequenceCounterparty, lastSentMockMsgRecvCounterparty *uint64, lock *sync.Mutex) []mock.TransactionMessage {
 	lock.Lock()
 	defer lock.Unlock()
-	if *mockSequence-*mockSequenceCounterparty > 0 {
+	if int64(*mockSequence)-int64(*mockSequenceCounterparty) > 0 {
 		return []mock.TransactionMessage{}
 	}
 	*mockSequence++