feat: add metrics to status protocol

go.mod

@@ -32,32 +32,30 @@ require (
 	github.com/multiformats/go-multihash v0.2.3
 	github.com/multiformats/go-multistream v0.5.0
 	github.com/opentracing/opentracing-go v1.2.0
-	github.com/prometheus/client_golang v1.18.0
+	github.com/prometheus/client_golang v1.21.1
 	github.com/spf13/afero v1.6.0
 	github.com/spf13/cobra v1.5.0
 	github.com/spf13/viper v1.7.0
-	github.com/stretchr/testify v1.8.4
+	github.com/stretchr/testify v1.10.0
 	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
 	github.com/uber/jaeger-client-go v2.24.0+incompatible
 	github.com/vmihailenco/msgpack/v5 v5.3.4
 	github.com/wealdtech/go-ens/v3 v3.5.1
 	gitlab.com/nolash/go-mockbytes v0.0.7
 	go.uber.org/atomic v1.11.0
 	go.uber.org/goleak v1.3.0
-	golang.org/x/crypto v0.23.0
-	golang.org/x/net v0.25.0
-	golang.org/x/sync v0.7.0
-	golang.org/x/sys v0.20.0
-	golang.org/x/term v0.20.0
+	golang.org/x/crypto v0.31.0
+	golang.org/x/net v0.33.0
+	golang.org/x/sync v0.10.0
+	golang.org/x/sys v0.28.0
+	golang.org/x/term v0.27.0
 	golang.org/x/time v0.5.0
 	gopkg.in/yaml.v2 v2.4.0
 	resenje.org/multex v0.1.0
 	resenje.org/singleflight v0.4.0
 	resenje.org/web v0.4.3
 )
 
-require golang.org/x/exp v0.0.0-20240213143201-ec583247a57a // indirect
-
 require (
 	github.com/BurntSushi/toml v1.1.0 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
@@ -66,7 +64,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.10.0 // indirect
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/codahale/hdrhistogram v0.0.0-00010101000000-000000000000 // indirect
 	github.com/consensys/bavard v0.1.13 // indirect
 	github.com/consensys/gnark-crypto v0.12.1 // indirect
@@ -104,7 +102,7 @@ require (
 	github.com/ipfs/go-log/v2 v2.5.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
-	github.com/klauspost/compress v1.17.6 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
@@ -132,16 +130,17 @@ require (
 	github.com/multiformats/go-multibase v0.2.0 // indirect
 	github.com/multiformats/go-multicodec v0.9.0 // indirect
 	github.com/multiformats/go-varint v0.0.7 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/onsi/ginkgo/v2 v2.15.0 // indirect
 	github.com/opencontainers/runtime-spec v1.2.0 // indirect
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
 	github.com/pelletier/go-toml v1.8.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_model v0.6.0 // indirect
-	github.com/prometheus/common v0.47.0 // indirect
-	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.62.0
+	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/prometheus/statsd_exporter v0.22.7 // indirect
 	github.com/quic-go/qpack v0.4.0 // indirect
 	github.com/quic-go/quic-go v0.42.0 // indirect
@@ -166,10 +165,11 @@ require (
 	go.uber.org/mock v0.4.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/exp v0.0.0-20240213143201-ec583247a57a // indirect
 	golang.org/x/mod v0.17.0 // indirect
-	golang.org/x/text v0.15.0 // indirect
-	golang.org/x/tools v0.20.0 // indirect
-	google.golang.org/protobuf v1.33.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
+	google.golang.org/protobuf v1.36.1 // indirect
 	gopkg.in/ini.v1 v1.57.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect

pkg/api/api.go

@@ -506,6 +506,32 @@ func (s *Service) contentLengthMetricMiddleware() func(h http.Handler) http.Hand
 	}
 }
 
+func (s *Service) downloadSpeedMetricMiddleware() func(h http.Handler) http.Handler {
+	return func(h http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			now := time.Now()
+			h.ServeHTTP(w, r)
+
+			if rw, ok := w.(*responseWriter); ok {
+				speed := float64(rw.size) / time.Since(now).Seconds()
+				s.metrics.DownloadSpeed.Observe(speed)
+			}
+		})
+	}
+}
+
+func (s *Service) uploadSpeedMetricMiddleware() func(h http.Handler) http.Handler {
+	return func(h http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			now := time.Now()
+			h.ServeHTTP(w, r)
+
+			speed := float64(r.ContentLength) / time.Since(now).Seconds()
+			s.metrics.UploadSpeed.Observe(speed)
+		})
+	}
+}
+
 // gasConfigMiddleware can be used by the APIs that allow block chain transactions to set
 // gas price and gas limit through the HTTP API headers.
 func (s *Service) gasConfigMiddleware(handlerName string) func(h http.Handler) http.Handler {

pkg/api/metrics.go

@@ -29,6 +29,8 @@
 	ResponseCodeCounts *prometheus.CounterVec
 
 	ContentApiDuration prometheus.HistogramVec
+	UploadSpeed        prometheus.Histogram
+	DownloadSpeed      prometheus.Histogram
 }
 
 func newMetrics() metrics {
@@ -64,6 +66,20 @@
 			Help:      "Histogram of file upload API response durations.",
 			Buckets:   []float64{0.5, 1, 2.5, 5, 10, 30, 60},
 		}, []string{"filesize", "method"}),
+		UploadSpeed: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "upload_speed",
+			Help:      "Histogram of upload speed in B/s.",
+			Buckets:   []float64{0.25, 0.5, 0.75, 1, 1.25, 1.5, 1.75, 2, 2.5, 3, 4, 5},
+		}),
+		DownloadSpeed: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "download_speed",
+			Help:      "Histogram of download speed in B/s.",
+			Buckets:   []float64{0.5, 1, 1.5, 2, 2.5, 3, 4, 5, 6, 7, 8, 9},
+		}),
 	}
 }
 
@@ -82,6 +98,14 @@
 	return m.PrometheusCollectorsFromFields(s.metrics)
 }
 
+// StatusMetrics exposes metrics that are exposed on the status protocol.
+func (s *Service) StatusMetrics() []prometheus.Collector {
+	return []prometheus.Collector{
+		s.metrics.UploadSpeed,
+		s.metrics.DownloadSpeed,
+	}
+}
+
 func (s *Service) pageviewMetricsHandler(h http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		start := time.Now()
@@ -114,18 +138,25 @@
 	UpgradedResponseWriter
 	statusCode  int
 	wroteHeader bool
+	size        int
 }
 
 func newResponseWriter(w http.ResponseWriter) *responseWriter {
 	// StatusOK is called by default if nothing else is called
 	uw := w.(UpgradedResponseWriter)
-	return &responseWriter{uw, http.StatusOK, false}
+	return &responseWriter{uw, http.StatusOK, false, 0}
 }
 
 func (rw *responseWriter) Status() int {
 	return rw.statusCode
 }
 
+func (rr *responseWriter) Write(b []byte) (int, error) {
+	size, err := rr.UpgradedResponseWriter.Write(b)
+	rr.size += size
+	return size, err
+}
+
 func (rw *responseWriter) WriteHeader(code int) {
 	if rw.wroteHeader {
 		return

pkg/api/router.go

@@ -301,6 +301,7 @@ func (s *Service) mountAPI() {
 		"POST": web.ChainHandlers(
 			s.contentLengthMetricMiddleware(),
 			s.newTracingHandler("bzz-upload"),
+			s.uploadSpeedMetricMiddleware(),
 			web.FinalHandlerFunc(s.bzzUploadHandler),
 		),
 	})
@@ -325,6 +326,7 @@ func (s *Service) mountAPI() {
 			s.contentLengthMetricMiddleware(),
 			s.newTracingHandler("bzz-download"),
 			s.actDecryptionHandler(),
+			s.downloadSpeedMetricMiddleware(),
 			web.FinalHandlerFunc(s.bzzDownloadHandler),
 		),
 		"HEAD": web.ChainHandlers(

pkg/api/status.go

@@ -32,6 +32,7 @@ type statusSnapshotResponse struct {
 	LastSyncedBlock         uint64  `json:"lastSyncedBlock"`
 	CommittedDepth          uint8   `json:"committedDepth"`
 	IsWarmingUp             bool    `json:"isWarmingUp"`
+	Metrics                 string  `json:"metrics,omitempty"`
 }
 
 type statusResponse struct {
@@ -98,6 +99,7 @@ func (s *Service) statusGetHandler(w http.ResponseWriter, _ *http.Request) {
 		LastSyncedBlock:         ss.LastSyncedBlock,
 		CommittedDepth:          uint8(ss.CommittedDepth),
 		IsWarmingUp:             s.isWarmingUp,
+		Metrics:                 ss.Metrics,
 	})
 }
 
@@ -146,6 +148,7 @@ func (s *Service) statusGetPeersHandler(w http.ResponseWriter, r *http.Request)
 				snapshot.IsReachable = ss.IsReachable
 				snapshot.LastSyncedBlock = ss.LastSyncedBlock
 				snapshot.CommittedDepth = uint8(ss.CommittedDepth)
+				snapshot.Metrics = ss.Metrics
 			}
 
 			mu.Lock()

pkg/api/status_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/status"
 	"github.com/ethersphere/bee/v2/pkg/storer"
 	"github.com/ethersphere/bee/v2/pkg/topology"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 func TestGetStatus(t *testing.T) {
@@ -41,6 +42,20 @@ func TestGetStatus(t *testing.T) {
 			IsReachable:             true,
 			LastSyncedBlock:         6092500,
 			CommittedDepth:          1,
+			Metrics: `# HELP test_response_duration_seconds Histogram of API response durations.
+# TYPE test_response_duration_seconds histogram
+test_response_duration_seconds_bucket{test="label",le="0.01"} 1
+test_response_duration_seconds_bucket{test="label",le="0.1"} 1
+test_response_duration_seconds_bucket{test="label",le="0.25"} 2
+test_response_duration_seconds_bucket{test="label",le="0.5"} 2
+test_response_duration_seconds_bucket{test="label",le="1"} 3
+test_response_duration_seconds_bucket{test="label",le="2.5"} 4
+test_response_duration_seconds_bucket{test="label",le="5"} 4
+test_response_duration_seconds_bucket{test="label",le="10"} 6
+test_response_duration_seconds_bucket{test="label",le="+Inf"} 7
+test_response_duration_seconds_sum{test="label"} 78.15
+test_response_duration_seconds_count{test="label"} 7
+`,
 		}
 
 		ssMock := &statusSnapshotMock{
@@ -53,13 +68,35 @@ func TestGetStatus(t *testing.T) {
 			committedDepth:          ssr.CommittedDepth,
 		}
 
+		metricsRegistry := prometheus.NewRegistry()
+
+		h := prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: "test",
+			Name:      "response_duration_seconds",
+			Help:      "Histogram of API response durations.",
+			Buckets:   []float64{0.01, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+			ConstLabels: prometheus.Labels{
+				"test": "label",
+			},
+		})
+
+		metricsRegistry.MustRegister(h)
+
+		points := []float64{0.25, 5.2, 1.5, 1, 5.2, 0, 65}
+		var sum float64
+		for _, p := range points {
+			h.Observe(p)
+			sum += p
+		}
+
 		statusSvc := status.NewService(
 			log.Noop,
 			nil,
 			new(topologyPeersIterNoopMock),
 			mode.String(),
 			ssMock,
 			ssMock,
+			metricsRegistry,
 		)
 
 		statusSvc.SetSync(ssMock)
@@ -86,6 +123,7 @@ func TestGetStatus(t *testing.T) {
 				"",
 				nil,
 				nil,
+				nil,
 			),
 		})
 

pkg/node/node.go

@@ -77,6 +77,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/util/syncutil"
 	"github.com/hashicorp/go-multierror"
 	ma "github.com/multiformats/go-multiaddr"
+	"github.com/prometheus/client_golang/prometheus"
 	promc "github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/crypto/sha3"
 	"golang.org/x/sync/errgroup"
@@ -910,7 +911,19 @@ func NewBee(
 
 	validStamp := postage.ValidStamp(batchStore)
 
-	nodeStatus := status.NewService(logger, p2ps, kad, beeNodeMode.String(), batchStore, localStore)
+	// metrics exposed on the status protocol
+	statusMetricsRegistry := prometheus.NewRegistry()
+	if apiService != nil {
+		statusMetricsRegistry.MustRegister(apiService.StatusMetrics()...)
+	}
+	if localStore != nil {
+		statusMetricsRegistry.MustRegister(localStore.StatusMetrics()...)
+	}
+	if p2ps != nil {
+		statusMetricsRegistry.MustRegister(p2ps.StatusMetrics()...)
+	}
+
+	nodeStatus := status.NewService(logger, p2ps, kad, beeNodeMode.String(), batchStore, localStore, statusMetricsRegistry)
 	if err = p2ps.AddProtocol(nodeStatus.Protocol()); err != nil {
 		return nil, fmt.Errorf("status service: %w", err)
 	}
@@ -968,6 +981,8 @@ func NewBee(
 	retrieval := retrieval.New(swarmAddress, waitNetworkRFunc, localStore, p2ps, kad, logger, acc, pricer, tracer, o.RetrievalCaching)
 	localStore.SetRetrievalService(retrieval)
 
+	statusMetricsRegistry.MustRegister(retrieval.StatusMetrics()...)
+
 	pusherService := pusher.New(networkID, localStore, pushSyncProtocol, batchStore, logger, warmupTime, pusher.DefaultRetryCount)
 	b.pusherCloser = pusherService
 

pkg/p2p/libp2p/libp2p.go

@@ -563,7 +563,6 @@ func (s *Service) AddProtocol(p p2p.ProtocolSpec) (err error) {
 		}
 
 		s.host.SetStreamHandlerMatch(id, matcher, func(streamlibp2p network.Stream) {
-			start := time.Now()
 			peerID := streamlibp2p.Conn().RemotePeer()
 			overlay, found := s.peers.overlay(peerID)
 			if !found {
@@ -581,14 +580,15 @@ func (s *Service) AddProtocol(p p2p.ProtocolSpec) (err error) {
 			stream := newStream(streamlibp2p, s.metrics)
 
 			// exchange headers
+			headersStartTime := time.Now()
 			ctx, cancel := context.WithTimeout(s.ctx, s.HeadersRWTimeout)
 			defer cancel()
 			if err := handleHeaders(ctx, ss.Headler, stream, overlay); err != nil {
 				s.logger.Debug("handle protocol: handle headers failed", "protocol", p.Name, "version", p.Version, "stream", ss.Name, "peer", overlay, "error", err)
 				_ = stream.Reset()
 				return
 			}
-			s.metrics.HeadersExchangeDuration.Observe(time.Since(start).Seconds())
+			s.metrics.HeadersExchangeDuration.Observe(time.Since(headersStartTime).Seconds())
 
 			ctx, cancel = context.WithCancel(s.ctx)
 

pkg/p2p/libp2p/metrics.go

@@ -123,3 +123,10 @@ func newMetrics() metrics {
 func (s *Service) Metrics() []prometheus.Collector {
 	return append(m.PrometheusCollectorsFromFields(s.metrics), s.handshakeService.Metrics()...)
 }
+
+// StatusMetrics exposes metrics that are exposed on the status protocol.
+func (s *Service) StatusMetrics() []prometheus.Collector {
+	return []prometheus.Collector{
+		s.metrics.HeadersExchangeDuration,
+	}
+}

pkg/retrieval/metrics.go

@@ -105,3 +105,12 @@ func newMetrics() metrics {
 func (s *Service) Metrics() []prometheus.Collector {
 	return m.PrometheusCollectorsFromFields(s.metrics)
 }
+
+// StatusMetrics exposes metrics that are exposed on the status protocol.
+func (s *Service) StatusMetrics() []prometheus.Collector {
+	return []prometheus.Collector{
+		s.metrics.RequestAttempts,
+		s.metrics.ChunkRetrieveTime,
+		s.metrics.RequestDurationTime,
+	}
+}