Merge pull request #83 from xmtp/e2e-convergence

e2e: convergence test

dev/e2e/run

@@ -3,7 +3,8 @@ set -eou pipefail
 
 GIT_COMMIT="$(git rev-parse HEAD)"
 
+# shellcheck disable=SC2068
 go run \
     -ldflags="-X 'main.GitCommit=${GIT_COMMIT}'" \
     cmd/xmtpd-e2e/main.go \
-    "$@"
+    $@

dev/e2e/run-spray-local

@@ -3,10 +3,11 @@ set -eou pipefail
 
 . dev/net/k8s-env
 
-nodes="$(kubectl -n xmtp-nodes get pods -l "app.kubernetes.io/part-of=xmtp-nodes" -o=json | jq -r '.items[].metadata.labels["app.kubernetes.io/name"]')"
-opts=""
+nodes="$(dev/terraform/tf output -json | jq -r '.nodes.value[].name')"
+opts=()
 while read -r node; do
-    opts="${opts} --api-url=${node}.localhost"
+    opts+=("--api-url=http://${node}.localhost")
 done <<< "$(echo -e "$nodes")"
+opts+=("$@")
 
-dev/e2e/run "${opts}" "$@"
+dev/e2e/run "${opts[*]}"

dev/terraform/plans/devnet-local/_variables.tf

@@ -14,6 +14,7 @@ variable "node_keys" {
   sensitive = true
 }
 variable "enable_chat_app" { default = true }
+variable "enable_e2e" { default = true }
 variable "enable_monitoring" { default = true }
 variable "e2e_delay" { default = "" }
 variable "node_container_cpu_limit" { default = "500m" }

dev/terraform/plans/devnet-local/main.tf

@@ -1,5 +1,5 @@
 module "cluster" {
-  source = "git@github.com:xmtp-labs/xmtpd-terraform.git//modules/xmtp-cluster-kind?ref=12d1e46"
+  source = "git@github.com:xmtp-labs/xmtpd-terraform.git//modules/xmtp-cluster-kind?ref=0d6dce9"
 
   # Uncomment this line and comment out the previous source line to use a
   # local instance of xmtpd-modules living in the parent directory of xmtpd.
@@ -12,6 +12,7 @@ module "cluster" {
   e2e_container_image         = var.e2e_container_image
   e2e_delay                   = var.e2e_delay
   enable_chat_app             = var.enable_chat_app
+  enable_e2e                  = var.enable_e2e
   enable_monitoring           = var.enable_monitoring
   node_container_cpu_limit    = var.node_container_cpu_limit
   node_container_memory_limit = var.node_container_memory_limit

pkg/e2e/e2e.go

@@ -27,18 +27,19 @@ type E2E struct {
 }
 
 type Options struct {
-	APIURLs          []string      `long:"api-url" env:"XMTP_API_URLS" description:"XMTP node API URLs" default:"http://localhost"`
-	ClientsPerURL    int           `long:"clients-per-url" description:"Number of clients for each API URL" default:"1"`
-	MessagePerClient int           `long:"messages-per-client" description:"Number of messages to publish for each client" default:"3"`
-	Continuous       bool          `long:"continuous" description:"Run continuously"`
-	ExitOnError      bool          `long:"exit-on-error" description:"Exit on error if running continuously"`
-	RunDelay         time.Duration `long:"delay" description:"Delay between runs (in seconds)" default:"5s"`
-	AdminPort        uint          `long:"admin-port" description:"Admin HTTP server listen port" default:"7777"`
+	APIURLs               []string      `long:"api-url" env:"XMTP_API_URLS" description:"XMTP node API URLs" default:"http://localhost"`
+	ClientsPerURL         int           `long:"clients-per-url" description:"Number of clients for each API URL" default:"1"`
+	MessagePerClient      int           `long:"messages-per-client" description:"Number of messages to publish for each client" default:"3"`
+	Continuous            bool          `long:"continuous" description:"Run continuously"`
+	ExitOnError           bool          `long:"exit-on-error" description:"Exit on error if running continuously"`
+	RunDelay              time.Duration `long:"delay" description:"Delay between runs" default:"5s"`
+	AdminPort             uint          `long:"admin-port" description:"Admin HTTP server listen port" default:"7777"`
+	QueryConvergenceDelay time.Duration `long:"query-convergence-delay" description:"Delay between query convergence checks" default:"10ms"`
 
 	GitCommit string
 }
 
-type testRunFunc func() error
+type testRunFunc func(name string) error
 
 type Test struct {
 	Name string
@@ -47,7 +48,7 @@ type Test struct {
 
 func (e *E2E) Tests() []*Test {
 	return []*Test{
-		e.newTest("messagev1 publish subscribe query", e.testMessageV1PublishSubscribeQuery),
+		e.newTest("convergence", e.testConvergence),
 	}
 }
 
@@ -59,7 +60,7 @@ func New(ctx context.Context, opts *Options) (*E2E, error) {
 		rand:    rand.New(rand.NewSource(time.Now().UTC().UnixNano())),
 		opts:    opts,
 	}
-	e.log.Info("running", zap.String("git-commit", opts.GitCommit))
+	e.log.Info("running", zap.String("git-commit", opts.GitCommit), zap.Strings("nodes", opts.APIURLs))
 
 	if e.opts.Continuous {
 		go func() {
@@ -102,7 +103,7 @@ func (e *E2E) runTest(test *Test) error {
 	started := time.Now().UTC()
 	log := e.log.With(zap.String("test", test.Name))
 
-	err := test.Run()
+	err := test.Run(test.Name)
 	duration := time.Since(started)
 	log = log.With(zap.Duration("duration", duration))
 	if err != nil {

pkg/e2e/e2e_test.go

@@ -42,7 +42,7 @@ func TestE2E(t *testing.T) {
 		t.Run(test.Name, func(t *testing.T) {
 			t.Parallel()
 
-			err := test.Run()
+			err := test.Run(test.Name)
 			require.NoError(t, err)
 		})
 	}

pkg/e2e/metrics.go

@@ -10,7 +10,11 @@ import (
 )
 
 type Metrics struct {
-	runDuration *prometheus.HistogramVec
+	runDuration                  *prometheus.HistogramVec
+	subscribeDuration            *prometheus.HistogramVec
+	publishDuration              *prometheus.HistogramVec
+	subscribeConvergenceDuration *prometheus.HistogramVec
+	queryConvergenceDuration     *prometheus.HistogramVec
 }
 
 func newMetrics() *Metrics {
@@ -25,6 +29,46 @@ func newMetrics() *Metrics {
 			},
 			[]string{"test", "status"},
 		),
+		subscribeDuration: promauto.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "xmtpd",
+				Subsystem: "e2e",
+				Name:      "subscribe_duration_us",
+				Help:      "duration of test case subscribe (microseconds)",
+				Buckets:   prometheus.ExponentialBuckets(10, 10, 10),
+			},
+			[]string{"test", "node", "status"},
+		),
+		publishDuration: promauto.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "xmtpd",
+				Subsystem: "e2e",
+				Name:      "publish_duration_us",
+				Help:      "duration of test case publish (microseconds)",
+				Buckets:   prometheus.ExponentialBuckets(10, 10, 10),
+			},
+			[]string{"test", "node", "status"},
+		),
+		subscribeConvergenceDuration: promauto.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "xmtpd",
+				Subsystem: "e2e",
+				Name:      "subscribe_convergence_duration_us",
+				Help:      "duration of test case subscribe convergence (microseconds)",
+				Buckets:   prometheus.ExponentialBuckets(10, 10, 10),
+			},
+			[]string{"test", "node", "status"},
+		),
+		queryConvergenceDuration: promauto.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "xmtpd",
+				Subsystem: "e2e",
+				Name:      "query_convergence_duration_us",
+				Help:      "duration of test case query convergence (microseconds)",
+				Buckets:   prometheus.ExponentialBuckets(10, 10, 10),
+			},
+			[]string{"test", "node", "status"},
+		),
 	}
 }
 
@@ -44,3 +88,75 @@ func (m *Metrics) recordRun(ctx context.Context, test, status string, duration t
 	}
 	met.Observe(float64(duration.Microseconds()))
 }
+
+func (m *Metrics) recordSubscribe(ctx context.Context, test, node, status string, duration time.Duration) {
+	if m == nil || m.subscribeDuration == nil {
+		return
+	}
+	met, err := m.subscribeDuration.GetMetricWithLabelValues(test, node, status)
+	if err != nil {
+		ctx.Logger().Warn("error observing metric",
+			zap.Error(err),
+			zap.String("metric", "subscribe_duration_us"),
+			zap.String("test", test),
+			zap.String("node", node),
+			zap.String("status", status),
+		)
+		return
+	}
+	met.Observe(float64(duration.Microseconds()))
+}
+
+func (m *Metrics) recordPublish(ctx context.Context, test, node, status string, duration time.Duration) {
+	if m == nil || m.publishDuration == nil {
+		return
+	}
+	met, err := m.publishDuration.GetMetricWithLabelValues(test, node, status)
+	if err != nil {
+		ctx.Logger().Warn("error observing metric",
+			zap.Error(err),
+			zap.String("metric", "publish_duration_us"),
+			zap.String("test", test),
+			zap.String("node", node),
+			zap.String("status", status),
+		)
+		return
+	}
+	met.Observe(float64(duration.Microseconds()))
+}
+
+func (m *Metrics) recordSubscribeConvergence(ctx context.Context, test, node, status string, duration time.Duration) {
+	if m == nil || m.subscribeConvergenceDuration == nil {
+		return
+	}
+	met, err := m.subscribeConvergenceDuration.GetMetricWithLabelValues(test, node, status)
+	if err != nil {
+		ctx.Logger().Warn("error observing metric",
+			zap.Error(err),
+			zap.String("metric", "subscribe_convergence_duration_us"),
+			zap.String("test", test),
+			zap.String("node", node),
+			zap.String("status", status),
+		)
+		return
+	}
+	met.Observe(float64(duration.Microseconds()))
+}
+
+func (m *Metrics) recordQueryConvergence(ctx context.Context, test, node, status string, duration time.Duration) {
+	if m == nil || m.queryConvergenceDuration == nil {
+		return
+	}
+	met, err := m.queryConvergenceDuration.GetMetricWithLabelValues(test, node, status)
+	if err != nil {
+		ctx.Logger().Warn("error observing metric",
+			zap.Error(err),
+			zap.String("metric", "query_convergence_duration_us"),
+			zap.String("test", test),
+			zap.String("node", node),
+			zap.String("status", status),
+		)
+		return
+	}
+	met.Observe(float64(duration.Microseconds()))
+}