feat: emit metrics when serving p2p registry API

pkg/discovery/content/registry/mirror.go

@@ -13,6 +13,7 @@
 
 	pcontext "github.com/azure/peerd/pkg/context"
 	"github.com/azure/peerd/pkg/discovery/routing"
+	"github.com/azure/peerd/pkg/metrics"
 	"github.com/azure/peerd/pkg/peernet"
 )
 
@@ -30,7 +31,8 @@
 	router         routing.Router
 	resolveRetries int
 
-	n peernet.Network
+	n               peernet.Network
+	metricsRecorder metrics.Metrics
 }
 
 // Handle handles a request to this registry mirror.
@@ -56,6 +58,8 @@
 		c.AbortWithError(http.StatusInternalServerError, errors.New("neither digest nor reference provided"))
 	}
 
+	startTime := time.Now()
+	peerCount := 0
 	peersChan, err := m.router.Resolve(resolveCtx, key, false, m.resolveRetries)
 	if err != nil {
 		//nolint
@@ -79,6 +83,12 @@
 				return
 			}
 
+			if peerCount == 0 {
+				// Only report the time it took to discover the first peer.
+				m.metricsRecorder.RecordPeerDiscovery(peer.HttpHost, time.Since(startTime).Seconds())
+				peerCount++
+			}
+
 			succeeded := false
 			u, err := url.Parse(peer.HttpHost)
 			if err != nil {
@@ -94,12 +104,16 @@
 				r.URL.RawQuery = c.Request.URL.RawQuery
 				pcontext.SetOutboundHeaders(r, c)
 			}
+
+			count := int64(0)
+
 			proxy.ModifyResponse = func(resp *http.Response) error {
 				if resp.StatusCode != http.StatusOK {
 					return fmt.Errorf("expected peer to respond with 200, got: %s", resp.Status)
 				}
 
 				succeeded = true
+				count = resp.ContentLength
 				return nil
 			}
 			proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
@@ -112,18 +126,20 @@
 				break
 			}
 
-			l.Info().Str("peer", u.Host).Msg("request served from peer")
+			m.metricsRecorder.RecordPeerResponse(peer.HttpHost, key, "pull", time.Since(startTime).Seconds(), count)
+			l.Info().Str("peer", u.Host).Int64("count", count).Msg("request served from peer")
 			return
 		}
 	}
 }
 
 // New creates a new mirror handler.
-func New(router routing.Router) *Mirror {
+func New(ctx context.Context, router routing.Router) *Mirror {
 	return &Mirror{
-		resolveTimeout: ResolveTimeout,
-		router:         router,
-		resolveRetries: ResolveRetries,
-		n:              router.Net(),
+		metricsRecorder: metrics.FromContext(ctx),
+		resolveTimeout:  ResolveTimeout,
+		router:          router,
+		resolveRetries:  ResolveRetries,
+		n:               router.Net(),
 	}
 }

pkg/discovery/content/registry/mirror_test.go

@@ -11,7 +11,9 @@ import (
 
 	pcontext "github.com/azure/peerd/pkg/context"
 	"github.com/azure/peerd/pkg/discovery/routing/mocks"
+	"github.com/azure/peerd/pkg/metrics"
 	"github.com/gin-gonic/gin"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 )
 
@@ -64,10 +66,11 @@ func TestMirrorHandler(t *testing.T) {
 	}
 	router := mocks.NewMockRouter(resolver)
 	m := &Mirror{
-		router:         router,
-		resolveRetries: ResolveRetries,
-		resolveTimeout: ResolveTimeout,
-		n:              router.Net(),
+		metricsRecorder: metrics.NewPromMetrics(prometheus.DefaultRegisterer, "test", "test"),
+		router:          router,
+		resolveRetries:  ResolveRetries,
+		resolveTimeout:  ResolveTimeout,
+		n:               router.Net(),
 	}
 
 	tests := []struct {

pkg/handlers/v2/handler.go

@@ -87,7 +87,7 @@ func (h *V2Handler) fill(c pcontext.Context) error {
 // New creates a new OCI content handler.
 func New(ctx context.Context, router routing.Router, containerdStore containerd.Store) (*V2Handler, error) {
 	return &V2Handler{
-		proxy:           registry.New(router),
+		proxy:           registry.New(ctx, router),
 		registry:        containerd.NewRegistry(containerdStore),
 		metricsRecorder: metrics.FromContext(ctx),
 	}, nil