Add per-beacon-node response attribution to the REST multi-handler

api/rest/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "get_options.go",
         "log.go",
+        "metrics.go",
         "mock_rest_provider.go",
         "multi_handler.go",
         "rest_connection_provider.go",
@@ -20,6 +21,8 @@ go_library(
         "//network/httputil:go_default_library",
         "//runtime/version:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@io_opentelemetry_go_contrib_instrumentation_net_http_otelhttp//:go_default_library",
     ],
@@ -42,6 +45,7 @@ go_test(
         "//runtime/version:go_default_library",
         "//testing/assert:go_default_library",
         "//testing/require:go_default_library",
+        "@com_github_prometheus_client_model//go:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
     ],

api/rest/metrics.go

@@ -0,0 +1,29 @@
+package rest
+
+import (
+	"github.com/OffchainLabs/prysm/v7/api"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// Outcomes recorded by nodeResponseTotal.
+const (
+	outcomeMatched  = "matched"
+	outcomeFallback = "fallback"
+	outcomeError    = "error"
+)
+
+// No endpoint label: endpoints embed the slot, so the cardinality is unbounded.
+var nodeResponseTotal = promauto.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: "validator",
+		Name:      "beacon_node_response_total",
+		Help:      "Beacon node responses by how the multi-handler used them within a single read round.",
+	},
+	[]string{"host", "outcome"},
+)
+
+// recordResponse credits a host with an outcome, redacting any credentials in the host.
+func recordResponse(host, outcome string) {
+	nodeResponseTotal.WithLabelValues(api.RedactEndpoint(host), outcome).Inc()
+}

api/rest/multi_handler.go

@@ -350,22 +350,24 @@ func raceRound[T any](ctx context.Context, handlers []*handler, fallbackDeadline
 	defer cancel()
 
 	type result struct {
-		val T
-		err error
+		val  T
+		err  error
+		host string
 	}
 
 	// Call fn concurrently and asynchronously on every handler, sending the result to results.
 	results := make(chan result, len(handlers))
 	for _, h := range handlers {
 		go func(h *handler) {
 			val, err := fn(ctx, h)
-			results <- result{val: val, err: err}
+			results <- result{val: val, err: err, host: h.Host()}
 		}(h)
 	}
 
 	var (
-		fallback *T
-		errs     []error
+		fallback     *T
+		fallbackHost string
+		errs         []error
 	)
 
 	var fallbackExpiry <-chan time.Time
@@ -376,10 +378,12 @@ func raceRound[T any](ctx context.Context, handlers []*handler, fallbackDeadline
 		select {
 		case r = <-results:
 		case <-fallbackExpiry:
+			recordResponse(fallbackHost, outcomeFallback)
 			return *fallback, false, true, errs
 		case <-ctx.Done():
 			errs = append(errs, ctx.Err())
 			if fallback != nil {
+				recordResponse(fallbackHost, outcomeFallback)
 				return *fallback, false, true, errs
 			}
 
@@ -388,17 +392,20 @@ func raceRound[T any](ctx context.Context, handlers []*handler, fallbackDeadline
 		}
 
 		if r.err != nil {
+			recordResponse(r.host, outcomeError)
 			errs = append(errs, r.err)
 			continue
 		}
 
 		// If r.val satisfies accept, return it immediately.
 		if accept(r.val) {
+			recordResponse(r.host, outcomeMatched)
 			return r.val, true, true, errs
 		}
 
 		if fallback == nil {
 			fallback = &r.val
+			fallbackHost = r.host
 
 			if !fallbackDeadline.IsZero() {
 				fallbackExpiry = time.After(time.Until(fallbackDeadline))
@@ -407,6 +414,7 @@ func raceRound[T any](ctx context.Context, handlers []*handler, fallbackDeadline
 	}
 
 	if fallback != nil {
+		recordResponse(fallbackHost, outcomeFallback)
 		return *fallback, false, true, errs
 	}
 
@@ -425,8 +433,9 @@ func raceRound[T any](ctx context.Context, handlers []*handler, fallbackDeadline
 //     the context was cancelled mid-run).
 func inOrderRound[T any](ctx context.Context, handlers []*handler, fallbackDeadline time.Time, accept func(T) bool, fn queryFunc[T]) (T, bool, bool, []error) {
 	var (
-		fallback *T
-		errs     []error
+		fallback     *T
+		fallbackHost string
+		errs         []error
 	)
 
 	for _, handler := range handlers {
@@ -452,22 +461,26 @@ func inOrderRound[T any](ctx context.Context, handlers []*handler, fallbackDeadl
 		val, err := fn(callCtx, handler)
 		cancel()
 		if err != nil {
+			recordResponse(handler.Host(), outcomeError)
 			errs = append(errs, err)
 			continue
 		}
 
 		// If val satisfies accept, return it immediately.
 		if accept(val) {
+			recordResponse(handler.Host(), outcomeMatched)
 			return val, true, true, errs
 		}
 
 		// If no fallback has been recorded yet, record this val as a best-effort fallback.
 		if fallback == nil {
 			fallback = &val
+			fallbackHost = handler.Host()
 		}
 	}
 
 	if fallback != nil {
+		recordResponse(fallbackHost, outcomeFallback)
 		return *fallback, false, true, errs
 	}
 

api/rest/multi_handler_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/network/httputil"
 	"github.com/OffchainLabs/prysm/v7/testing/assert"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
+	dto "github.com/prometheus/client_model/go"
 )
 
 type testResponse struct {
@@ -443,6 +444,7 @@ func TestRaceRound(t *testing.T) {
 	handlers := []*handler{newTestHandler("http://a"), newTestHandler("http://b")}
 
 	t.Run("returns the first response satisfying accept", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		accept := func(s string) bool { return s == "fresh" }
 		fn := func(_ context.Context, h *handler) (string, error) {
 			if h == handlers[1] {
@@ -456,19 +458,26 @@ func TestRaceRound(t *testing.T) {
 		assert.Equal(t, true, ok)
 		assert.Equal(t, "fresh", val)
 		assert.Equal(t, 0, len(errs))
+		assert.Equal(t, float64(1), counterValue(t, "http://b", outcomeMatched), "the matching host should be credited")
+		assert.Equal(t, float64(0), counterValue(t, "http://a", outcomeMatched))
 	})
 
 	t.Run("falls back to a usable response when none match", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		accept := func(string) bool { return false } // nothing matches
 		fn := func(context.Context, *handler) (string, error) { return "stale", nil }
 
 		val, matched, ok, _ := raceRound(context.Background(), handlers, time.Time{}, accept, fn)
 		assert.Equal(t, false, matched)
 		assert.Equal(t, true, ok, "a non-matching 2XX is still a usable response")
 		assert.Equal(t, "stale", val)
+		// Either host can win the race, but exactly one fallback is recorded.
+		fallbacks := counterValue(t, "http://a", outcomeFallback) + counterValue(t, "http://b", outcomeFallback)
+		assert.Equal(t, float64(1), fallbacks)
 	})
 
 	t.Run("reports failure when every handler errors", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		sentinel := errors.New("boom")
 		accept := func(string) bool { return true }
 		fn := func(context.Context, *handler) (string, error) { return "", sentinel }
@@ -478,9 +487,12 @@ func TestRaceRound(t *testing.T) {
 		assert.Equal(t, false, ok)
 		assert.Equal(t, len(handlers), len(errs))
 		assert.Equal(t, true, errors.Is(errors.Join(errs...), sentinel))
+		assert.Equal(t, float64(1), counterValue(t, "http://a", outcomeError))
+		assert.Equal(t, float64(1), counterValue(t, "http://b", outcomeError))
 	})
 
 	t.Run("stops waiting for a hung handler at the fallback deadline", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		accept := func(string) bool { return false } // the responding node is lagging
 		fn := func(ctx context.Context, h *handler) (string, error) {
 			if h == handlers[1] {
@@ -501,9 +513,11 @@ func TestRaceRound(t *testing.T) {
 		assert.Equal(t, true, ok)
 		assert.Equal(t, "stale", val, "the usable response must be returned rather than waiting on the hung node")
 		assert.Equal(t, true, time.Since(start) < 10*time.Second, "the round must not wait for the read deadline")
+		assert.Equal(t, float64(1), counterValue(t, "http://a", outcomeFallback), "the expiry path still credits the fallback host")
 	})
 
 	t.Run("returns the fallback when the context is canceled with a usable response in hand", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		release := make(chan struct{})
 		t.Cleanup(func() { close(release) })
 
@@ -532,6 +546,7 @@ func TestRaceRound(t *testing.T) {
 		assert.Equal(t, true, ok, "a response collected before the cancellation is still usable")
 		assert.Equal(t, "stale", val)
 		assert.Equal(t, true, errors.Is(errors.Join(errs...), context.Canceled))
+		assert.Equal(t, float64(1), counterValue(t, "http://a", outcomeFallback), "the cancellation path still credits the fallback host")
 	})
 
 	t.Run("returns promptly when the context is canceled with nothing in hand", func(t *testing.T) {
@@ -555,6 +570,7 @@ func TestInOrderRound(t *testing.T) {
 	handlers := []*handler{newTestHandler("http://a"), newTestHandler("http://b")}
 
 	t.Run("returns the first response satisfying accept", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		accept := func(s string) bool { return s == "fresh" }
 		fn := func(_ context.Context, h *handler) (string, error) {
 			if h == handlers[0] {
@@ -568,19 +584,25 @@ func TestInOrderRound(t *testing.T) {
 		assert.Equal(t, true, ok)
 		assert.Equal(t, "fresh", val)
 		assert.Equal(t, 0, len(errs))
+		assert.Equal(t, float64(1), counterValue(t, "http://a", outcomeMatched))
+		assert.Equal(t, float64(0), counterValue(t, "http://b", outcomeMatched))
 	})
 
 	t.Run("falls back to a usable response when none match", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		accept := func(string) bool { return false } // nothing matches
 		fn := func(context.Context, *handler) (string, error) { return "stale", nil }
 
 		val, matched, ok, _ := inOrderRound(context.Background(), handlers, time.Time{}, accept, fn)
 		assert.Equal(t, false, matched)
 		assert.Equal(t, true, ok, "a non-matching 2XX is still a usable response")
 		assert.Equal(t, "stale", val)
+		assert.Equal(t, float64(1), counterValue(t, "http://a", outcomeFallback), "the first usable host is the fallback")
+		assert.Equal(t, float64(0), counterValue(t, "http://b", outcomeFallback))
 	})
 
 	t.Run("reports failure when every handler errors", func(t *testing.T) {
+		nodeResponseTotal.Reset()
 		sentinel := errors.New("boom")
 		accept := func(string) bool { return true }
 		fn := func(context.Context, *handler) (string, error) { return "", sentinel }
@@ -590,6 +612,20 @@ func TestInOrderRound(t *testing.T) {
 		assert.Equal(t, false, ok)
 		assert.Equal(t, len(handlers), len(errs))
 		assert.Equal(t, true, errors.Is(errors.Join(errs...), sentinel))
+		assert.Equal(t, float64(1), counterValue(t, "http://a", outcomeError))
+		assert.Equal(t, float64(1), counterValue(t, "http://b", outcomeError))
+	})
+
+	t.Run("redacts credentials from the host label", func(t *testing.T) {
+		nodeResponseTotal.Reset()
+		withCreds := []*handler{newTestHandler("http://user:password@first:3500")}
+		accept := func(string) bool { return true }
+		fn := func(context.Context, *handler) (string, error) { return "ok", nil }
+
+		_, matched, _, _ := inOrderRound(context.Background(), withCreds, time.Time{}, accept, fn)
+		assert.Equal(t, true, matched)
+		// The single match landed on the redacted label, so the raw one was never used.
+		assert.Equal(t, float64(1), counterValue(t, "http://user:xxxxx@first:3500", outcomeMatched))
 	})
 
 	t.Run("stops when the context is already canceled", func(t *testing.T) {
@@ -700,6 +736,13 @@ func waitFor(cond func() bool) error {
 	return context.DeadlineExceeded
 }
 
+// counterValue reads the current nodeResponseTotal count for a host/outcome pair.
+func counterValue(t *testing.T, host, outcome string) float64 {
+	var m dto.Metric
+	require.NoError(t, nodeResponseTotal.WithLabelValues(host, outcome).Write(&m))
+	return m.GetCounter().GetValue()
+}
+
 // newTestHandler builds a *handler pointing at the given base URL.
 func newTestHandler(host string) *handler {
 	return newHandler(http.Client{Timeout: 5 * time.Second}, host)

changelog/syjn99_bn-response-attribution-metrics.md

@@ -0,0 +1,3 @@
+### Added
+
+- Added the `validator_beacon_node_response_total{host,outcome}` metric, attributing each multi-beacon-node read to the node that served it.