Fix various flaky tests (#16396)

agent/grpc-external/services/peerstream/stream_test.go

@@ -1252,8 +1252,8 @@ func TestStreamResources_Server_DisconnectsOnHeartbeatTimeout(t *testing.T) {
 	})
 
 	testutil.RunStep(t, "stream is disconnected due to heartbeat timeout", func(t *testing.T) {
-		disconnectTime := ptr(it.FutureNow(1))
 		retry.Run(t, func(r *retry.R) {
+			disconnectTime := ptr(it.StaticNow())
 			status, ok := srv.StreamStatus(testPeerID)
 			require.True(r, ok)
 			require.False(r, status.Connected)
@@ -1423,7 +1423,7 @@ func makeClient(t *testing.T, srv *testServer, peerID string) *MockClient {
 		},
 	}))
 
-	// Receive a services and roots subscription request pair from server
+	// Receive ExportedService, ExportedServiceList, and PeeringTrustBundle subscription requests from server
 	receivedSub1, err := client.Recv()
 	require.NoError(t, err)
 	receivedSub2, err := client.Recv()

agent/grpc-external/services/peerstream/testing.go

@@ -150,6 +150,16 @@ func (t *incrementalTime) Now() time.Time {
 	return t.base.Add(dur)
 }
 
+// StaticNow returns the current internal clock without advancing it.
+func (t *incrementalTime) StaticNow() time.Time {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	dur := time.Duration(t.next) * time.Second
+
+	return t.base.Add(dur)
+}
+
 // FutureNow will return a given future value of the Now() function.
 // The numerical argument indicates which future Now value you wanted. The
 // value must be > 0.

agent/xds/delta_test.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/armon/go-metrics"
 	envoy_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
-	"github.com/hashicorp/consul/api"
 	"github.com/stretchr/testify/require"
 	rpcstatus "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc/codes"
@@ -21,8 +20,10 @@ import (
 	"github.com/hashicorp/consul/agent/grpc-external/limiter"
 	"github.com/hashicorp/consul/agent/proxycfg"
 	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/envoyextensions/xdscommon"
 	"github.com/hashicorp/consul/sdk/testutil"
+	"github.com/hashicorp/consul/sdk/testutil/retry"
 	"github.com/hashicorp/consul/version"
 )
 
@@ -1057,19 +1058,23 @@ func TestServer_DeltaAggregatedResources_v3_ACLEnforcement(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			// aclResolve may be called in a goroutine even after a
+			// testcase tt returns. Capture the variable as tc so the
+			// values don't swap in the next iteration.
+			tc := tt
 			aclResolve := func(id string) (acl.Authorizer, error) {
-				if !tt.defaultDeny {
+				if !tc.defaultDeny {
 					// Allow all
 					return acl.RootAuthorizer("allow"), nil
 				}
-				if tt.acl == "" {
+				if tc.acl == "" {
 					// No token and defaultDeny is denied
 					return acl.RootAuthorizer("deny"), nil
 				}
 				// Ensure the correct token was passed
-				require.Equal(t, tt.token, id)
+				require.Equal(t, tc.token, id)
 				// Parse the ACL and enforce it
-				policy, err := acl.NewPolicyFromSource(tt.acl, nil, nil)
+				policy, err := acl.NewPolicyFromSource(tc.acl, nil, nil)
 				require.NoError(t, err)
 				return acl.NewPolicyAuthorizerWithDefaults(acl.RootAuthorizer("deny"), []*acl.Policy{policy}, nil)
 			}
@@ -1095,13 +1100,15 @@ func TestServer_DeltaAggregatedResources_v3_ACLEnforcement(t *testing.T) {
 
 			// If there is no token, check that we increment the gauge
 			if tt.token == "" {
-				data := scenario.sink.Data()
-				require.Len(t, data, 1)
+				retry.Run(t, func(r *retry.R) {
+					data := scenario.sink.Data()
+					require.Len(r, data, 1)
 
-				item := data[0]
-				val, ok := item.Gauges["consul.xds.test.xds.server.streamsUnauthenticated"]
-				require.True(t, ok)
-				require.Equal(t, float32(1), val.Value)
+					item := data[0]
+					val, ok := item.Gauges["consul.xds.test.xds.server.streamsUnauthenticated"]
+					require.True(r, ok)
+					require.Equal(r, float32(1), val.Value)
+				})
 			}
 
 			if !tt.wantDenied {
@@ -1138,13 +1145,15 @@ func TestServer_DeltaAggregatedResources_v3_ACLEnforcement(t *testing.T) {
 
 			// If there is no token, check that we decrement the gauge
 			if tt.token == "" {
-				data := scenario.sink.Data()
-				require.Len(t, data, 1)
+				retry.Run(t, func(r *retry.R) {
+					data := scenario.sink.Data()
+					require.Len(r, data, 1)
 
-				item := data[0]
-				val, ok := item.Gauges["consul.xds.test.xds.server.streamsUnauthenticated"]
-				require.True(t, ok)
-				require.Equal(t, float32(0), val.Value)
+					item := data[0]
+					val, ok := item.Gauges["consul.xds.test.xds.server.streamsUnauthenticated"]
+					require.True(r, ok)
+					require.Equal(r, float32(0), val.Value)
+				})
 			}
 		})
 	}

agent/xds/xds_protocol_helpers_test.go

@@ -166,9 +166,6 @@ func newTestServerDeltaScenario(
 ) *testServerScenario {
 	mgr := newTestManager(t)
 	envoy := NewTestEnvoy(t, proxyID, token)
-	t.Cleanup(func() {
-		envoy.Close()
-	})
 
 	sink := metrics.NewInmemSink(1*time.Minute, 1*time.Minute)
 	cfg := metrics.DefaultConfig("consul.xds.test")
@@ -177,6 +174,7 @@ func newTestServerDeltaScenario(
 	metrics.NewGlobal(cfg, sink)
 
 	t.Cleanup(func() {
+		envoy.Close()
 		sink := &metrics.BlackholeSink{}
 		metrics.NewGlobal(cfg, sink)
 	})

command/debug/debug.go

@@ -270,7 +270,8 @@ func (c *cmd) prepare() (version string, err error) {
 	// If none are specified we will collect information from
 	// all by default
 	if len(c.capture) == 0 {
-		c.capture = defaultTargets
+		c.capture = make([]string, len(defaultTargets))
+		copy(c.capture, defaultTargets)
 	}
 
 	// If EnableDebug is not true, skip collecting pprof