Method TestControlChannelConnectivityStateMonitoring

balancer/rls/balancer_test.go:1023–1163

TestControlChannelConnectivityStateMonitoring tests the scenario where the control channel goes down and comes back up again and verifies that backoff state is reset for cache entries in this scenario. It also verifies that: - Backoff is NOT reset when the control channel first becomes READY (i.e.,

(t *testing.T)

//     the initial CONNECTING → READY transition should not trigger a backoff reset)
//   - Backoff is reset for READY → TRANSIENT_FAILURE → READY transitions
func (s) TestControlChannelConnectivityStateMonitoring(t *testing.T) {
	// Create a restartable listener which can close existing connections.
	l, err := testutils.LocalTCPListener()
	if err != nil {
		t.Fatalf("net.Listen() failed: %v", err)
	}
	lis := testutils.NewRestartableListener(l)

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

	// Override the reset backoff hook to get notified.
	resetBackoffDone := make(chan struct{}, 1)
	origResetBackoffHook := resetBackoffHook
	resetBackoffHook = func() { resetBackoffDone <- struct{}{} }
	defer func() { resetBackoffHook = origResetBackoffHook }()

	// Override the backoff strategy to return a large backoff which
	// will make sure the date cache entry remains in backoff for the
	// duration of the test.
	origBackoffStrategy := defaultBackoffStrategy
	defaultBackoffStrategy = &fakeBackoffStrategy{backoff: defaultTestTimeout}
	defer func() { defaultBackoffStrategy = origBackoffStrategy }()

	// Override the connectivity state subscriber to wrap the original and
	// make connectivity state changes visible to the test.
	wrappedSubscriber := &wrappingConnectivityStateSubscriber{connStateCh: buffer.NewUnbounded()}
	origConnectivityStateSubscriber := newConnectivityStateSubscriber
	newConnectivityStateSubscriber = func(delegate grpcsync.Subscriber) grpcsync.Subscriber {
		return &wrappingConnectivityStateSubscriber{
			delegate:    delegate,
			connStateCh: wrappedSubscriber.connStateCh,
		}
	}
	defer func() { newConnectivityStateSubscriber = origConnectivityStateSubscriber }()

	// 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 a very low value for
	// maxAge to ensure that cache entries become invalid very soon.
	rlsConfig := buildBasicRLSConfig(childPolicyName, rlsServer.Address)
	rlsConfig.RouteLookupConfig.MaxAge = durationpb.New(defaultTestShortTimeout)

	// 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)