Add error counter for SSE endpoint (#14681)

* add error counter for SSE endpoint

* add changelog entry

CHANGELOG.md

@@ -30,6 +30,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Better attestation packing for Electra. [PR](https://github.com/prysmaticlabs/prysm/pull/14534)
 - P2P: Add logs when a peer is (dis)connected. Add the reason of the disconnection when we initiate it.
 - Added a Prometheus error counter metric for HTTP requests to track beacon node requests.
+- Added a Prometheus error counter metric for SSE requests.
 
 ### Changed
 

beacon-chain/rpc/endpoints.go

@@ -62,6 +62,13 @@ func (e *endpoint) handlerWithMiddleware() http.HandlerFunc {
 	)
 
 	return func(w http.ResponseWriter, r *http.Request) {
+		// SSE errors are handled separately to avoid interference with the streaming
+		// mechanism and ensure accurate error tracking.
+		if e.template == "/eth/v1/events" {
+			handler.ServeHTTP(w, r)
+			return
+		}
+
 		rw := &responseWriter{ResponseWriter: w, statusCode: http.StatusOK}
 		handler.ServeHTTP(rw, r)
 

beacon-chain/rpc/eth/events/BUILD.bazel

@@ -32,6 +32,8 @@ go_library(
         "//time/slots:go_default_library",
         "@com_github_ethereum_go_ethereum//common/hexutil: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",
     ],
 )

beacon-chain/rpc/eth/events/events.go

@@ -12,6 +12,8 @@ import (
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prysmaticlabs/prysm/v5/api"
 	"github.com/prysmaticlabs/prysm/v5/api/server/structs"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed"
@@ -76,6 +78,14 @@ var (
 	errWriterUnusable     = errors.New("http response writer is unusable")
 )
 
+var httpSSEErrorCount = promauto.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "http_sse_error_count",
+		Help: "Total HTTP errors for server sent events endpoint",
+	},
+	[]string{"endpoint", "error"},
+)
+
 // The eventStreamer uses lazyReaders to defer serialization until the moment the value is ready to be written to the client.
 type lazyReader func() io.Reader
 
@@ -145,6 +155,13 @@ func newTopicRequest(topics []string) (*topicRequest, error) {
 // Servers may send SSE comments beginning with ':' for any purpose,
 // including to keep the event stream connection alive in the presence of proxy servers.
 func (s *Server) StreamEvents(w http.ResponseWriter, r *http.Request) {
+	var err error
+	defer func() {
+		if err != nil {
+			httpSSEErrorCount.WithLabelValues(r.URL.Path, err.Error()).Inc()
+		}
+	}()
+
 	log.Debug("Starting StreamEvents handler")
 	ctx, span := trace.StartSpan(r.Context(), "events.StreamEvents")
 	defer span.End()
@@ -174,7 +191,7 @@ func (s *Server) StreamEvents(w http.ResponseWriter, r *http.Request) {
 	defer cancel()
 	es := newEventStreamer(buffSize, ka)
 
-	go es.outboxWriteLoop(ctx, cancel, sw)
+	go es.outboxWriteLoop(ctx, cancel, sw, r.URL.Path)
 	if err := es.recvEventLoop(ctx, cancel, topics, s); err != nil {
 		log.WithError(err).Debug("Shutting down StreamEvents handler.")
 	}
@@ -264,11 +281,12 @@ func newlineReader() io.Reader {
 
 // outboxWriteLoop runs in a separate goroutine. Its job is to write the values in the outbox to
 // the client as fast as the client can read them.
-func (es *eventStreamer) outboxWriteLoop(ctx context.Context, cancel context.CancelFunc, w *streamingResponseWriterController) {
+func (es *eventStreamer) outboxWriteLoop(ctx context.Context, cancel context.CancelFunc, w *streamingResponseWriterController, endpoint string) {
 	var err error
 	defer func() {
 		if err != nil {
 			log.WithError(err).Debug("Event streamer shutting down due to error.")
+			httpSSEErrorCount.WithLabelValues(endpoint, err.Error()).Inc()
 		}
 		es.exit()
 	}()

beacon-chain/rpc/eth/events/server.go

@@ -12,7 +12,7 @@ import (
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
 )
 
-// Server defines a server implementation of the gRPC events service,
+// Server defines a server implementation of the http events service,
 // providing RPC endpoints to subscribe to events from the beacon node.
 type Server struct {
 	StateNotifier          statefeed.Notifier