Introduce randomized timings and reproducible randomization into controller integration tests.

internal/catalog/catalogtest/run_test.go

@@ -11,10 +11,15 @@ import (
 	"github.com/hashicorp/consul/internal/catalog/internal/controllers"
 	"github.com/hashicorp/consul/internal/controller"
 	"github.com/hashicorp/consul/internal/resource/reaper"
+	rtest "github.com/hashicorp/consul/internal/resource/resourcetest"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
+var (
+	clientOpts = rtest.ConfigureTestCLIFlags()
+)
+
 func runInMemResourceServiceAndControllers(t *testing.T, deps controllers.Dependencies) pbresource.ResourceServiceClient {
 	t.Helper()
 
@@ -38,10 +43,10 @@ func runInMemResourceServiceAndControllers(t *testing.T, deps controllers.Depend
 
 func TestControllers_Integration(t *testing.T) {
 	client := runInMemResourceServiceAndControllers(t, catalog.DefaultControllerDependencies())
-	RunCatalogV2Beta1IntegrationTest(t, client)
+	RunCatalogV2Beta1IntegrationTest(t, client, clientOpts.ClientOptions(t)...)
 }
 
 func TestControllers_Lifecycle(t *testing.T) {
 	client := runInMemResourceServiceAndControllers(t, catalog.DefaultControllerDependencies())
-	RunCatalogV2Beta1LifecycleIntegrationTest(t, client)
+	RunCatalogV2Beta1LifecycleIntegrationTest(t, client, clientOpts.ClientOptions(t)...)
 }

internal/catalog/catalogtest/test_integration_v2beta1.go

@@ -38,19 +38,19 @@ var (
 // is another RunCatalogIntegrationTestLifeCycle function that can be used for those
 // purposes. The two are distinct so that the data being published and the assertions
 // made against the system can be reused in upgrade tests.
-func RunCatalogV2Beta1IntegrationTest(t *testing.T, client pbresource.ResourceServiceClient) {
+func RunCatalogV2Beta1IntegrationTest(t *testing.T, client pbresource.ResourceServiceClient, opts ...rtest.ClientOption) {
 	t.Helper()
 
-	PublishCatalogV2Beta1IntegrationTestData(t, client)
+	PublishCatalogV2Beta1IntegrationTestData(t, client, opts...)
 	VerifyCatalogV2Beta1IntegrationTestResults(t, client)
 }
 
 // PublishCatalogV2Beta1IntegrationTestData will perform a whole bunch of resource writes
 // for Service, ServiceEndpoints, Workload, Node and HealthStatus objects
-func PublishCatalogV2Beta1IntegrationTestData(t *testing.T, client pbresource.ResourceServiceClient) {
+func PublishCatalogV2Beta1IntegrationTestData(t *testing.T, client pbresource.ResourceServiceClient, opts ...rtest.ClientOption) {
 	t.Helper()
 
-	c := rtest.NewClient(client)
+	c := rtest.NewClient(client, opts...)
 
 	resources := rtest.ParseResourcesFromFilesystem(t, testData, "integration_test_data/v2beta1")
 	c.PublishResources(t, resources)

internal/catalog/catalogtest/test_lifecycle_v2beta1.go

@@ -16,19 +16,19 @@ import (
 // RunCatalogV2Beta1LifecycleIntegrationTest intends to excercise functionality of
 // managing catalog resources over their normal lifecycle where they will be modified
 // several times, change state etc.
-func RunCatalogV2Beta1LifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient) {
+func RunCatalogV2Beta1LifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient, opts ...rtest.ClientOption) {
 	t.Helper()
 
 	testutil.RunStep(t, "node-lifecycle", func(t *testing.T) {
-		RunCatalogV2Beta1NodeLifecycleIntegrationTest(t, client)
+		RunCatalogV2Beta1NodeLifecycleIntegrationTest(t, client, opts...)
 	})
 
 	testutil.RunStep(t, "workload-lifecycle", func(t *testing.T) {
-		RunCatalogV2Beta1WorkloadLifecycleIntegrationTest(t, client)
+		RunCatalogV2Beta1WorkloadLifecycleIntegrationTest(t, client, opts...)
 	})
 
 	testutil.RunStep(t, "endpoints-lifecycle", func(t *testing.T) {
-		RunCatalogV2Beta1EndpointsLifecycleIntegrationTest(t, client)
+		RunCatalogV2Beta1EndpointsLifecycleIntegrationTest(t, client, opts...)
 	})
 }
 
@@ -41,8 +41,8 @@ func RunCatalogV2Beta1LifecycleIntegrationTest(t *testing.T, client pbresource.R
 // * Changing HealthStatus to a better health will cause recomputation of the Health
 // * Deletion of associated HealthStatuses will recompute the Health (back to passing)
 // * Deletion of the node will cause deletion of associated health statuses
-func RunCatalogV2Beta1NodeLifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient) {
-	c := rtest.NewClient(client)
+func RunCatalogV2Beta1NodeLifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient, opts ...rtest.ClientOption) {
+	c := rtest.NewClient(client, opts...)
 
 	nodeName := "test-lifecycle"
 	nodeHealthName := "test-lifecycle-node-status"
@@ -131,8 +131,8 @@ func RunCatalogV2Beta1NodeLifecycleIntegrationTest(t *testing.T, client pbresour
 //   - Overall health is computed as the worst health amongst the nodes health and all
 //     of the workloads associated HealthStatuses
 //   - Deletion of the workload will cause deletion of all associated health statuses.
-func RunCatalogV2Beta1WorkloadLifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient) {
-	c := rtest.NewClient(client)
+func RunCatalogV2Beta1WorkloadLifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient, opts ...rtest.ClientOption) {
+	c := rtest.NewClient(client, opts...)
 	testutil.RunStep(t, "nodeless-workload", func(t *testing.T) {
 		runV2Beta1NodelessWorkloadLifecycleIntegrationTest(t, c)
 	})
@@ -379,8 +379,8 @@ func runV2Beta1NodeAssociatedWorkloadLifecycleIntegrationTest(t *testing.T, c *r
 // * Adding ports to a service will recalculate the endpoints
 // * Removing ports from a service will recalculate the endpoints
 // * Changing the workload will recalculate the endpoints (ports, addresses, or health)
-func RunCatalogV2Beta1EndpointsLifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient) {
-	c := rtest.NewClient(client)
+func RunCatalogV2Beta1EndpointsLifecycleIntegrationTest(t *testing.T, client pbresource.ResourceServiceClient, opts ...rtest.ClientOption) {
+	c := rtest.NewClient(client, opts...)
 	serviceName := "test-lifecycle"
 
 	// Create the service without a selector. We should not see endpoints generated but we should see the

internal/resource/resourcetest/builder.go

@@ -178,7 +178,7 @@ func (b *resourceBuilder) Write(t T, client pbresource.ResourceServiceClient) *p
 		id := proto.Clone(rsp.Resource.Id).(*pbresource.ID)
 		id.Uid = ""
 		t.Cleanup(func() {
-			rtestClient.MustDelete(t, id)
+			rtestClient.CleanupDelete(t, id)
 		})
 	}
 

internal/resource/resourcetest/client.go

@@ -5,12 +5,14 @@ package resourcetest
 
 import (
 	"context"
+	"flag"
 	"fmt"
 	"math/rand"
 	"time"
 
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/slices"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -21,25 +23,67 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil/retry"
 )
 
+type ClientOption func(*Client)
+
+func WithRNGSeed(seed int64) ClientOption {
+	return func(c *Client) {
+		c.rng = rand.New(rand.NewSource(seed))
+	}
+}
+
+func WithRequestDelay(minMilliseconds int, maxMilliseconds int) ClientOption {
+	return func(c *Client) {
+
+		min := minMilliseconds
+		max := maxMilliseconds
+		if max < min {
+			min = maxMilliseconds
+			max = minMilliseconds
+		}
+		c.requestDelayMin = min
+		c.requestDelayMax = max
+	}
+}
+
+func WithACLToken(token string) ClientOption {
+	return func(c *Client) {
+		c.token = token
+	}
+}
+
 type Client struct {
 	pbresource.ResourceServiceClient
 
 	timeout time.Duration
 	wait    time.Duration
 	token   string
-}
 
-func NewClient(client pbresource.ResourceServiceClient) *Client {
-	return NewClientWithACLToken(client, "")
+	rng *rand.Rand
+
+	requestDelayMin int
+	requestDelayMax int
 }
 
-func NewClientWithACLToken(client pbresource.ResourceServiceClient, token string) *Client {
-	return &Client{
+func NewClient(client pbresource.ResourceServiceClient, opts ...ClientOption) *Client {
+	c := &Client{
 		ResourceServiceClient: client,
 		timeout:               7 * time.Second,
 		wait:                  25 * time.Millisecond,
-		token:                 token,
+		rng:                   rand.New(rand.NewSource(time.Now().UnixNano())),
+		// arbitrary write delays are opt-in only
+		requestDelayMin: 0,
+		requestDelayMax: 0,
+	}
+
+	for _, opt := range opts {
+		opt(c)
 	}
+
+	return c
+}
+
+func NewClientWithACLToken(client pbresource.ResourceServiceClient, token string) *Client {
+	return NewClient(client, WithACLToken(token))
 }
 
 func (client *Client) SetRetryerConfig(timeout time.Duration, wait time.Duration) {
@@ -60,7 +104,7 @@ func (client *Client) PublishResources(t T, resources []*pbresource.Resource) {
 	// controllers should eventually converge on the desired state. The exception to this
 	// is that you cannot insert resources with owner refs before the resource they are
 	// owned by or insert a resource into a non-default tenant before that tenant exists.
-	rand.Shuffle(len(resources), func(i, j int) {
+	client.rng.Shuffle(len(resources), func(i, j int) {
 		temp := resources[i]
 		resources[i] = resources[j]
 		resources[j] = temp
@@ -97,7 +141,7 @@ func (client *Client) PublishResources(t T, resources []*pbresource.Resource) {
 
 			id := rsp.Resource.Id
 			t.Cleanup(func() {
-				client.MustDelete(t, id)
+				client.CleanupDelete(t, id)
 			})
 
 			// track the number of resources published
@@ -119,6 +163,11 @@ func (client *Client) PublishResources(t T, resources []*pbresource.Resource) {
 	require.Empty(t, resources, "Could not publish all resources - some resources have invalid owner references")
 }
 
+func (client *Client) Write(ctx context.Context, in *pbresource.WriteRequest, opts ...grpc.CallOption) (*pbresource.WriteResponse, error) {
+	client.delayRequest()
+	return client.ResourceServiceClient.Write(ctx, in, opts...)
+}
+
 func (client *Client) Context(t T) context.Context {
 	ctx := testutil.TestContext(t)
 
@@ -280,11 +329,32 @@ func (client *Client) ResolveResourceID(t T, id *pbresource.ID) *pbresource.ID {
 	return client.RequireResourceExists(t, id).Id
 }
 
+// MustDelete will delete a resource by its id, retrying if necessary and fail the test
+// if it cannot delete it within the timeout. The clients request delay settings are
+// taken into account with this operation.
 func (client *Client) MustDelete(t T, id *pbresource.ID) {
+	t.Helper()
+	client.retryDelete(t, id, true)
+}
+
+// CleanupDelete will perform the same operations as MustDelete to ensure the resource is
+// deleted. The clients request delay settings are ignored for this operation and it is
+// assumed this will only be called in the context of test Cleanup routines where we
+// are no longer testing that a controller eventually converges on some values in response
+// to the delete.
+func (client *Client) CleanupDelete(t T, id *pbresource.ID) {
+	t.Helper()
+	client.retryDelete(t, id, true)
+}
+
+func (client *Client) retryDelete(t T, id *pbresource.ID, shouldDelay bool) {
 	t.Helper()
 	ctx := client.Context(t)
 
 	client.retry(t, func(r *retry.R) {
+		if shouldDelay {
+			client.delayRequest()
+		}
 		_, err := client.Delete(ctx, &pbresource.DeleteRequest{Id: id})
 		if status.Code(err) == codes.NotFound {
 			return
@@ -300,3 +370,53 @@ func (client *Client) MustDelete(t T, id *pbresource.ID) {
 		require.NoError(r, err)
 	})
 }
+
+func (client *Client) delayRequest() {
+	if client.requestDelayMin == 0 && client.requestDelayMax == 0 {
+		return
+	}
+
+	var delay time.Duration
+	if client.requestDelayMin == client.requestDelayMax {
+		delay = time.Duration(client.requestDelayMin) * time.Millisecond
+	} else {
+		delay = time.Duration(client.rng.Intn(client.requestDelayMax-client.requestDelayMin)+client.requestDelayMin) * time.Millisecond
+	}
+	time.Sleep(delay)
+}
+
+type CLIOptions struct {
+	minRequestDelay int
+	maxRequestDelay int
+	seed            int64
+}
+
+type CLIOptionT interface {
+	Helper()
+	Logf(string, ...any)
+}
+
+func (o *CLIOptions) ClientOptions(t CLIOptionT) []ClientOption {
+	t.Helper()
+	t.Logf("Using %d for the random number generator seed. Pass -rng-seed=<value> to overwrite the time based seed", o.seed)
+	t.Logf("Using random request delays between %dms and %dms. Use -min-request-delay=<value> or -max-request-delay=<value> to override the defaults", o.minRequestDelay, o.maxRequestDelay)
+
+	return []ClientOption{
+		WithRNGSeed(o.seed),
+		WithRequestDelay(o.minRequestDelay, o.maxRequestDelay),
+	}
+}
+
+func ConfigureTestCLIFlags() *CLIOptions {
+	opts := &CLIOptions{
+		minRequestDelay: 0,
+		maxRequestDelay: 0,
+		seed:            time.Now().UnixNano(),
+	}
+
+	flag.Int64Var(&opts.seed, "rng-seed", opts.seed, "Seed to use for pseudo-random-number-generators")
+	flag.IntVar(&opts.minRequestDelay, "min-request-delay", 10, "Minimum delay before performing a resource write (milliseconds: default=10)")
+	flag.IntVar(&opts.maxRequestDelay, "max-request-delay", 50, "Maximum delay before performing a resource write (milliseconds: default=50)")
+
+	return opts
+}

test/integration/consul-container/test/catalog/catalog_test.go

@@ -12,9 +12,14 @@ import (
 	libtopology "github.com/hashicorp/consul/test/integration/consul-container/libs/topology"
 
 	"github.com/hashicorp/consul/internal/catalog/catalogtest"
+	rtest "github.com/hashicorp/consul/internal/resource/resourcetest"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
+var (
+	cli = rtest.ConfigureTestCLIFlags()
+)
+
 func TestCatalog(t *testing.T) {
 	t.Parallel()
 
@@ -29,10 +34,10 @@ func TestCatalog(t *testing.T) {
 	client := pbresource.NewResourceServiceClient(followers[0].GetGRPCConn())
 
 	t.Run("one-shot", func(t *testing.T) {
-		catalogtest.RunCatalogV2Beta1IntegrationTest(t, client)
+		catalogtest.RunCatalogV2Beta1IntegrationTest(t, client, cli.ClientOptions(t)...)
 	})
 
 	t.Run("lifecycle", func(t *testing.T) {
-		catalogtest.RunCatalogV2Beta1LifecycleIntegrationTest(t, client)
+		catalogtest.RunCatalogV2Beta1LifecycleIntegrationTest(t, client, cli.ClientOptions(t)...)
 	})
 }