feat: extract server timings and measure individual retrievals

pkg/server/http/ipfs.go

@@ -5,9 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"strconv"
-	"time"
 
-	"github.com/filecoin-project/lassie/pkg/events"
 	lassie "github.com/filecoin-project/lassie/pkg/lassie"
 	"github.com/filecoin-project/lassie/pkg/retriever"
 	"github.com/filecoin-project/lassie/pkg/storage"
@@ -16,7 +14,6 @@ import (
 	"github.com/ipld/go-ipld-prime/datamodel"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	"github.com/libp2p/go-libp2p/core/peer"
-	servertiming "github.com/mitchellh/go-server-timing"
 	"github.com/multiformats/go-multicodec"
 )
 
@@ -187,41 +184,8 @@ func ipfsHandler(lassie *lassie.Lassie, cfg HttpServerConfig) func(http.Response
 		}
 
 		// servertiming metrics
-		startTimeMap := make(map[string]time.Time)
-
 		logger.Debugw("fetching CID", "retrievalId", retrievalId, "CID", rootCid.String(), "path", path.String(), "dagScope", dagScope)
-		stats, err := lassie.Fetch(req.Context(), request, func(re types.RetrievalEvent) {
-			header := servertiming.FromContext(req.Context())
-			if header == nil {
-				return
-			}
-
-			var metricName string
-			switch re.(type) {
-			case events.StartedFindingCandidatesEvent:
-				header.NewMetric("indexer")
-				startTimeMap["indexer"] = re.Time()
-
-			case events.StartedRetrievalEvent:
-				header.NewMetric("retrieval")
-				startTimeMap["retrieval"] = re.Time()
-				metricName = "indexer"
-
-			case events.FinishedEvent:
-				metricName = "retrieval"
-			}
-
-			// Set the metric duration
-			header.Lock()
-			if header.Metrics != nil {
-				for _, m := range header.Metrics {
-					if metricName != "" && m.Name == metricName {
-						m.Duration = re.Time().Sub(startTimeMap[metricName])
-					}
-				}
-			}
-			header.Unlock()
-		})
+		stats, err := lassie.Fetch(req.Context(), request, servertimingsSubscriber(req))
 
 		// force all blocks to flush
 		if cerr := carWriter.Close(); cerr != nil {

pkg/server/http/servertimingssubscriber.go

@@ -0,0 +1,86 @@
+package httpserver
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/filecoin-project/lassie/pkg/events"
+	"github.com/filecoin-project/lassie/pkg/types"
+	servertiming "github.com/mitchellh/go-server-timing"
+)
+
+func servertimingsSubscriber(req *http.Request) types.RetrievalEventSubscriber {
+	var fetchStartTime time.Time
+
+	var candidateFindingMetric *servertiming.Metric
+	var candidateFindingStartTime time.Time
+
+	retrievalMetricMap := make(map[string]*servertiming.Metric)
+	retrievalTimingMap := make(map[string]time.Time)
+
+	return func(re types.RetrievalEvent) {
+		timing := servertiming.FromContext(req.Context())
+		if timing == nil {
+			return
+		}
+
+		switch event := re.(type) {
+		case events.StartedFetchEvent:
+			fetchStartTime = re.Time()
+
+		// Candidate finding cases
+		case events.StartedFindingCandidatesEvent:
+			candidateFindingMetric = timing.NewMetric(string(re.Code()))
+			candidateFindingMetric.Extra = make(map[string]string)
+			candidateFindingMetric.Extra["dur"] = formatDuration(re.Time().Sub(fetchStartTime))
+			candidateFindingStartTime = re.Time()
+		case events.CandidatesFoundEvent:
+			candidateFindingMetric.Duration = re.Time().Sub(candidateFindingStartTime)
+			candidateFindingMetric.Extra[string(re.Code())] = fmt.Sprintf("%d", re.Time().Sub(candidateFindingStartTime))
+		case events.CandidatesFilteredEvent:
+			candidateFindingMetric.Duration = re.Time().Sub(candidateFindingStartTime)
+			candidateFindingMetric.Extra[string(re.Code())] = fmt.Sprintf("%d", re.Time().Sub(candidateFindingStartTime))
+
+		// Retrieval cases
+		case events.StartedRetrievalEvent:
+			name := fmt.Sprintf("retrieval-%s", events.Identifier(re))
+			retrievalMetric := timing.NewMetric(name)
+			retrievalMetric.Extra = make(map[string]string)
+			// We're using "dur" here to render the time since the "started-fetch" in the browser
+			retrievalMetric.Extra["dur"] = formatDuration(re.Time().Sub(fetchStartTime))
+			retrievalMetricMap[name] = retrievalMetric
+			retrievalTimingMap[name] = re.Time()
+		case events.FirstByteEvent:
+			name := fmt.Sprintf("retrieval-%s", events.Identifier(event))
+			if retrievalMetric, ok := retrievalMetricMap[name]; ok {
+				retrievalMetric.Extra[string(re.Code())] = fmt.Sprintf("%d", re.Time().Sub(retrievalTimingMap[name]))
+			}
+		case events.FailedRetrievalEvent:
+			name := fmt.Sprintf("retrieval-%s", events.Identifier(re))
+			if retrievalMetric, ok := retrievalMetricMap[name]; ok {
+				retrievalMetric.Extra[string(re.Code())] = fmt.Sprintf("%d", re.Time().Sub(retrievalTimingMap[name]))
+				retrievalMetric.Duration = re.Time().Sub(retrievalTimingMap[name])
+			}
+
+		// Due to the timing in which the Server-Timing header is written,
+		// the success and finished events are never emitted in time to make it into the header.
+		case events.SucceededEvent:
+		case events.FinishedEvent:
+
+		default:
+			if event, ok := re.(events.EventWithSPID); ok {
+				name := fmt.Sprintf("retrieval-%s", events.Identifier(event))
+				if retrievalMetric, ok := retrievalMetricMap[name]; ok {
+					retrievalMetric.Extra[string(re.Code())] = fmt.Sprintf("%d", re.Time().Sub(retrievalTimingMap[name]))
+				}
+			}
+		}
+	}
+}
+
+// formatDuration formats a duration in milliseconds in the same way the go-server-timing library does.
+func formatDuration(d time.Duration) string {
+	return strconv.FormatFloat(float64(d)/float64(time.Millisecond), 'f', -1, 64)
+}