GODRIVER-3095 Add moving STD to RTT Stats

x/mongo/driver/topology/rtt_monitor.go

@@ -10,6 +10,7 @@ import (
 	"container/list"
 	"context"
 	"fmt"
+	"math"
 	"sync"
 	"time"
 
@@ -41,11 +42,14 @@ type rttMonitor struct {
 	// connMu guards connecting and disconnecting. This is necessary since
 	// disconnecting will await the cancellation of a started connection. The
 	// use case for rttMonitor.connect needs to be goroutine safe.
-	connMu        sync.Mutex
-	averageRTT    time.Duration
-	averageRTTSet bool
-	movingMin     *list.List
-	minRTT        time.Duration
+	connMu                 sync.Mutex
+	averageRTT             time.Duration
+	averageRTTSet          bool
+	movingMin              *list.List
+	minRTT                 time.Duration
+	stddevRTT              time.Duration
+	stddevSum              float64
+	callsToAppendMovingMin int
 
 	closeWg  sync.WaitGroup
 	cfg      *rttConfig
@@ -179,20 +183,41 @@ func (r *rttMonitor) runHellos(conn *connection) {
 	}
 }
 
-// reset sets the average and min RTT to 0. This should only be called from the server monitor when an error
-// occurs during a server check. Errors in the RTT monitor should not reset the RTTs.
+// reset sets the average, min, and stddev RTT to 0. This should only be called from the server monitor
+// when an error occurs during a server check. Errors in the RTT monitor should not reset the RTTs.
 func (r *rttMonitor) reset() {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
 	r.movingMin = list.New()
 	r.averageRTT = 0
 	r.averageRTTSet = false
+	r.stddevSum = 0
+	r.callsToAppendMovingMin = 0
+}
+
+func (r *rttMonitor) calcStddev() float64 {
+	var mean, stddev float64
+
+	for element := r.movingMin.Front(); element != nil; element = element.Next() {
+		mean += float64(element.Value.(time.Duration))
+	}
+	mean /= float64(r.movingMin.Len())
+
+	for element := r.movingMin.Front(); element != nil; element = element.Next() {
+		sample := float64(element.Value.(time.Duration))
+		stddev += math.Pow(sample-mean, 2)
+	}
+	stddev = math.Sqrt(stddev / float64(r.movingMin.Len()))
+
+	return stddev
 }
 
 // appendMovingMin will append the RTT to the movingMin list which tracks a
 // minimum RTT within the last "minRTTSamplesForMovingMin" RTT samples.
 func (r *rttMonitor) appendMovingMin(rtt time.Duration) {
+	r.callsToAppendMovingMin++
+
 	if r.movingMin == nil || rtt < 0 {
 		return
 	}
@@ -202,6 +227,12 @@ func (r *rttMonitor) appendMovingMin(rtt time.Duration) {
 	}
 
 	r.movingMin.PushBack(rtt)
+
+	// Collect a sum of stddevs over maxRTTSamplesForMovingMin calls, ignore if calls are less than max
+	if r.callsToAppendMovingMin >= maxRTTSamplesForMovingMin {
+		stddev := r.calcStddev()
+		r.stddevSum += stddev
+	}
 }
 
 // min will return the minimum value in the movingMin list.
@@ -222,6 +253,21 @@ func (r *rttMonitor) min() time.Duration {
 	return min
 }
 
+// stddev will return the current moving stddev.
+func (r *rttMonitor) stddev() time.Duration {
+	var stddev time.Duration
+
+	if r.callsToAppendMovingMin < maxRTTSamplesForMovingMin {
+		return 0
+	}
+
+	// Get the number of times stddev was updated and calculate the average stddev
+	frequency := (r.callsToAppendMovingMin + 1) - maxRTTSamplesForMovingMin
+	stddev = time.Duration(r.stddevSum / float64(frequency))
+
+	return stddev
+}
+
 func (r *rttMonitor) addSample(rtt time.Duration) {
 	// Lock for the duration of this method. We're doing compuationally inexpensive work very infrequently, so lock
 	// contention isn't expected.
@@ -230,6 +276,7 @@ func (r *rttMonitor) addSample(rtt time.Duration) {
 
 	r.appendMovingMin(rtt)
 	r.minRTT = r.min()
+	r.stddevRTT = r.stddev()
 
 	if !r.averageRTTSet {
 		r.averageRTT = rtt
@@ -262,7 +309,8 @@ func (r *rttMonitor) Stats() string {
 	defer r.mu.RUnlock()
 
 	return fmt.Sprintf(
-		"network round-trip time stats: moving avg: %v, min: %v",
+		"network round-trip time stats: moving avg: %v, min: %v, moving stddev: %v",
 		r.averageRTT,
-		r.minRTT)
+		r.minRTT,
+		r.stddevRTT)
 }

x/mongo/driver/topology/rtt_monitor_test.go

@@ -416,3 +416,71 @@ func TestRTTMonitor_min(t *testing.T) {
 		})
 	}
 }
+
+func TestRTTMonitor_stddev(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		samples []time.Duration
+		want    float64
+	}{
+		{
+			name:    "empty",
+			samples: []time.Duration{},
+			want:    0,
+		},
+		{
+			name:    "one",
+			samples: makeArithmeticSamples(1, 1),
+			want:    0,
+		},
+		{
+			name:    "below maxRTTSamples",
+			samples: makeArithmeticSamples(1, 5),
+			want:    0,
+		},
+		{
+			name:    "equal maxRTTSamples",
+			samples: makeArithmeticSamples(1, 10),
+			want:    2.872281e+06,
+		},
+		{
+			name:    "exceed maxRTTSamples",
+			samples: makeArithmeticSamples(1, 15),
+			want:    2.872281e+06,
+		},
+		{
+			name: "non-sequential",
+			samples: []time.Duration{
+				2 * time.Millisecond,
+				1 * time.Millisecond,
+				4 * time.Millisecond,
+				3 * time.Millisecond,
+				7 * time.Millisecond,
+				12 * time.Millisecond,
+				6 * time.Millisecond,
+				8 * time.Millisecond,
+				5 * time.Millisecond,
+				13 * time.Millisecond,
+			},
+			want: 3.806573e+06,
+		},
+	}
+
+	for _, test := range tests {
+		test := test // capture the range variable
+
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			rtt := &rttMonitor{
+				movingMin: list.New(),
+			}
+			for _, sample := range test.samples {
+				rtt.appendMovingMin(sample)
+			}
+			assert.Equal(t, test.want, float64(rtt.stddev()))
+		})
+	}
+}