satellite/metabase/rangedloop: monkit durations (#5365)

Wire up duration measurement of observers with monkit.

Tested by attaching a SleepObserver, starting the rangedloop in storj-up
and navigating to http://<container>:11111/mon/stats. It reports the
following statistic:

completed-observer-duration,observer=*rangedlooptest.SleepObserver,scope=storj.io/storj/satellite/metabase/rangedloop duration=10.000117

Change-Id: Ief131d34001dd5d3ba1d7be6f161986e1f66440d

satellite/metabase/rangedloop/observerstats.go

@@ -0,0 +1,63 @@
+// Copyright (C) 2022 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package rangedloop
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/spacemonkeygo/monkit/v3"
+)
+
+func sendObserverDurations(observerDurations []ObserverDuration) {
+	initCompletedObserverStats()
+	completedObserverStatsInstance.setObserverDurations(observerDurations)
+}
+
+// completedObserverStatsInstance is initialized once
+// so that hopefully there is never more than once objec instance per satellite process
+// and statistics of different object instances don't clobber each other.
+var completedObserverStatsInstance *completedObserverStats
+
+// Implements monkit.StatSource.
+// Reports the duration per observer from the last completed run of the ranged segment loop.
+type completedObserverStats struct {
+	mu                sync.Mutex
+	observerDurations []ObserverDuration
+}
+
+func initCompletedObserverStats() {
+	if completedObserverStatsInstance != nil {
+		return
+	}
+
+	completedObserverStatsInstance = &completedObserverStats{
+		observerDurations: []ObserverDuration{},
+	}
+
+	// wire statistics up with monkit
+	mon.Chain(completedObserverStatsInstance)
+}
+
+// Stats implements monkit.StatSource to send the observer durations every time monkit is polled externally.
+func (o *completedObserverStats) Stats(cb func(key monkit.SeriesKey, field string, val float64)) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	// if there are no completed observers yet, no statistics will be sent
+	for _, observerDuration := range o.observerDurations {
+		key := monkit.NewSeriesKey("completed-observer-duration")
+		key = key.WithTag("observer", fmt.Sprintf("%T", observerDuration.Observer))
+
+		cb(key, "duration", observerDuration.Duration.Seconds())
+	}
+}
+
+// setObserverDurations sets the observer durations to report at ranged segment loop completion.
+func (o *completedObserverStats) setObserverDurations(observerDurations []ObserverDuration) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	o.observerDurations = observerDurations
+}

satellite/metabase/rangedloop/service.go

@@ -178,6 +178,8 @@ func finishObservers(ctx context.Context, observerStates []observerState) (obser
 		observerDurations = append(observerDurations, observerDuration)
 	}
 
+	sendObserverDurations(observerDurations)
+
 	return observerDurations, nil
 }