Improve flaky connect/proxy Listener tests (#4498)

Improve flaky connect/proxy Listener tests

- Add sleep to TestEchoConn to allow for Read/Write to finish before fetching data in reportStats

- Account for flakiness around interval for Gauge

- Improve debug output when dumping metrics

connect/proxy/listener_test.go

@@ -35,13 +35,31 @@ func assertCurrentGaugeValue(t *testing.T, sink *metrics.InmemSink,
 
 	data := sink.Data()
 
-	// Current interval is the last one
+	// Loop backward through intervals until there is a non-empty one
-	currentInterval := data[len(data)-1]
+	// Addresses flakiness around recording to one interval but accessing during the next
-	currentInterval.RLock()
+	var got float32
-	defer currentInterval.RUnlock()
+	for i := len(data) - 1; i >= 0; i-- {
+		currentInterval := data[i]
 
-	assert.Equalf(t, value, currentInterval.Gauges[name].Value,
+		currentInterval.RLock()
-		"gauge value mismatch. Current Interval:\n%v", currentInterval)
+		if len(currentInterval.Gauges) > 0 {
+			got = currentInterval.Gauges[name].Value
+			break
+		}
+		currentInterval.RUnlock()
+	}
+
+	if !assert.Equal(t, value, got) {
+		buf := bytes.NewBuffer(nil)
+		for _, intv := range data {
+			intv.RLock()
+			for name, val := range intv.Gauges {
+				fmt.Fprintf(buf, "[%v][G] '%s': %0.3f\n", intv.Interval, name, val.Value)
+			}
+			intv.RUnlock()
+		}
+		t.Log(buf.String())
+	}
 }
 
 func assertAllTimeCounterValue(t *testing.T, sink *metrics.InmemSink,
@@ -67,7 +85,7 @@ func assertAllTimeCounterValue(t *testing.T, sink *metrics.InmemSink,
 		buf := bytes.NewBuffer(nil)
 		for _, intv := range data {
 			intv.RLock()
-			for _, val := range intv.Gauges {
+			for name, val := range intv.Gauges {
 				fmt.Fprintf(buf, "[%v][G] '%s': %0.3f\n", intv.Interval, name, val.Value)
 			}
 			for name, vals := range intv.Points {
@@ -75,10 +93,10 @@ func assertAllTimeCounterValue(t *testing.T, sink *metrics.InmemSink,
 					fmt.Fprintf(buf, "[%v][P] '%s': %0.3f\n", intv.Interval, name, val)
 				}
 			}
-			for _, agg := range intv.Counters {
+			for name, agg := range intv.Counters {
 				fmt.Fprintf(buf, "[%v][C] '%s': %s\n", intv.Interval, name, agg.AggregateSample)
 			}
-			for _, agg := range intv.Samples {
+			for name, agg := range intv.Samples {
 				fmt.Fprintf(buf, "[%v][S] '%s': %s\n", intv.Interval, name, agg.AggregateSample)
 			}
 			intv.RUnlock()
@@ -131,8 +149,7 @@ func TestPublicListener(t *testing.T) {
 	// Check active conn is tracked in gauges
 	assertCurrentGaugeValue(t, sink, "consul.proxy.test.inbound.conns;dst=db", 1)
 
-	// Close listener to ensure all conns are closed and have reported their
+	// Close listener to ensure all conns are closed and have reported their metrics
-	// metrics
 	l.Close()
 
 	// Check all the tx/rx counters got added
@@ -194,8 +211,7 @@ func TestUpstreamListener(t *testing.T) {
 	// Check active conn is tracked in gauges
 	assertCurrentGaugeValue(t, sink, "consul.proxy.test.upstream.conns;src=web;dst_type=service;dst=db", 1)
 
-	// Close listener to ensure all conns are closed and have reported their
+	// Close listener to ensure all conns are closed and have reported their metrics
-	// metrics
 	l.Close()
 
 	// Check all the tx/rx counters got added

connect/proxy/testing.go

@@ -6,6 +6,7 @@ import (
 	"log"
 	"net"
 	"sync/atomic"
+	"time"
 
 	"github.com/hashicorp/consul/lib/freeport"
 	"github.com/mitchellh/go-testing-interface"
@@ -104,4 +105,8 @@ func TestEchoConn(t testing.T, conn net.Conn, prefix string) {
 	}
 	require.Equal(t, expectLen, got)
 	require.Equal(t, prefix+"Hello World", string(buf[:]))
+
+	// Addresses test flakiness around returning before Write or Read finish
+	// see PR #4498
+	time.Sleep(time.Millisecond)
 }