Method TestDataCachePurging

balancer/rls/balancer_test.go:899–978

TestDataCachePurging verifies that the LB policy periodically evicts expired entries from the data cache.

(t *testing.T)

// TestDataCachePurging verifies that the LB policy periodically evicts expired
// entries from the data cache.
func (s) TestDataCachePurging(t *testing.T) {
	// Override the frequency of the data cache purger to a small one.
	origDataCachePurgeTicker := dataCachePurgeTicker
	ticker := time.NewTicker(defaultTestShortTimeout)
	defer ticker.Stop()
	dataCachePurgeTicker = func() *time.Ticker { return ticker }
	defer func() { dataCachePurgeTicker = origDataCachePurgeTicker }()

	// Override the data cache purge hook to get notified.
	dataCachePurgeDone := make(chan struct{}, 1)
	origDataCachePurgeHook := dataCachePurgeHook
	dataCachePurgeHook = func() { dataCachePurgeDone <- struct{}{} }
	defer func() { dataCachePurgeHook = origDataCachePurgeHook }()

	// Start an RLS server and set the throttler to never throttle requests.
	rlsServer, rlsReqCh := rlstest.SetupFakeRLSServer(t, nil)
	overrideAdaptiveThrottler(t, neverThrottlingThrottler())

	// Register an LB policy to act as the child policy for RLS LB policy.
	childPolicyName := "test-child-policy" + t.Name()
	e2e.RegisterRLSChildPolicy(childPolicyName, nil)
	t.Logf("Registered child policy with name %q", childPolicyName)

	// Build RLS service config with header matchers and lookupService pointing to
	// the fake RLS server created above. Set a very low value for maxAge to
	// ensure that the entry expires soon.
	rlsConfig := buildBasicRLSConfig(childPolicyName, rlsServer.Address)
	rlsConfig.RouteLookupConfig.MaxAge = durationpb.New(time.Millisecond)

	// Start a test backend, and set up the fake RLS server to return this as a
	// target in the RLS response.
	backendCh, backendAddress := startBackend(t)
	rlsServer.SetResponseCallback(func(_ context.Context, _ *rlspb.RouteLookupRequest) *rlstest.RouteLookupResponse {
		return &rlstest.RouteLookupResponse{Resp: &rlspb.RouteLookupResponse{Targets: []string{backendAddress}}}
	})

	// Register a manual resolver and push the RLS service config through it.
	r := startManualResolverWithConfig(t, rlsConfig)

	cc, err := grpc.NewClient(r.Scheme()+":///", grpc.WithResolvers(r), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatalf("Failed to create gRPC client: %v", err)
	}
	defer cc.Close()

	// Make an RPC and ensure it gets routed to the test backend.
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	ctxOutgoing := metadata.AppendToOutgoingContext(ctx, "n1", "v1")
	makeTestRPCAndExpectItToReachBackend(ctxOutgoing, t, cc, backendCh)

	// Make sure an RLS request is sent out.
	verifyRLSRequest(t, rlsReqCh, true)

	// Make another RPC with different headers. This will force the LB policy to
	// send out a new RLS request, resulting in a new data cache entry.
	ctxOutgoing = metadata.AppendToOutgoingContext(ctx, "n2", "v2")
	makeTestRPCAndExpectItToReachBackend(ctxOutgoing, t, cc, backendCh)