internal/metrics/metrics.go

@@ -60,18 +60,23 @@ func New() *Tracker {
 	}
 }
 
-// SetConnected updates the connection status
+// SetConnected updates the connection status. Every connection after the
+// first counts one reconnect.
 func (t *Tracker) SetConnected(connected bool) {
-	wasConnected := t.isConnected.Swap(connected)
-	if connected {
+	t.isConnected.Store(connected)
+	if !connected {
+		return
+	}
+
 	t.mu.Lock()
+	// connectedSince is zero until the first connection.
+	connectedBefore := !t.connectedSince.IsZero()
 	t.connectedSince = time.Now()
 	t.mu.Unlock()
-		// Increment reconnect count (but not for the initial connection)
-		if wasConnected || t.reconnectCount.Load() > 0 {
+
+	if connectedBefore {
 		t.reconnectCount.Add(1)
 	}
-	}
 }
 
 // GetReconnectCount returns the number of reconnections since startup

internal/metrics/metrics_test.go

@@ -0,0 +1,21 @@
+package metrics
+
+import "testing"
+
+// TestReconnectCount connects three times with a disconnect between each, the
+// way the streamer does, and expects every connection after the first to count
+// one reconnect.
+func TestReconnectCount(t *testing.T) {
+	tracker := New()
+
+	tracker.SetConnected(true)
+	tracker.SetConnected(false)
+	tracker.SetConnected(true)
+	tracker.SetConnected(false)
+	tracker.SetConnected(true)
+
+	const wantReconnects = 2
+	if got := tracker.GetReconnectCount(); got != wantReconnects {
+		t.Fatalf("reconnect count = %d, want %d", got, wantReconnects)
+	}
+}