In-progress changes for control plane connectivity state monitoring.

Author: kannanjgithub

rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java

@@ -261,8 +261,8 @@ public void accept(BatchRecorder recorder) {
   }
 
   private void rlsServerConnectionStateChanged() {
-    lastRlsServerConnectivityState = rlsChannel.getState(false);
-    if (wasInTransientFailure && lastRlsServerConnectivityState == ConnectivityState.READY) {
+    ConnectivityState currentState = rlsChannel.getState(false);
+    if (wasInTransientFailure && currentState == ConnectivityState.READY) {
       wasInTransientFailure = false;
       synchronized (lock) {
         for (CacheEntry value : linkedHashLruCache.values()) {
@@ -271,10 +271,11 @@ private void rlsServerConnectionStateChanged() {
           }
         }
       }
-    } else if (!wasInTransientFailure && lastRlsServerConnectivityState == ConnectivityState.TRANSIENT_FAILURE) {
+    } else if (lastRlsServerConnectivityState == ConnectivityState.READY && currentState == ConnectivityState.TRANSIENT_FAILURE) {
       wasInTransientFailure = true;
     }
-    rlsChannel.notifyWhenStateChanged(lastRlsServerConnectivityState, () -> rlsServerConnectionStateChanged());
+    rlsChannel.notifyWhenStateChanged(currentState, () -> rlsServerConnectionStateChanged());
+    lastRlsServerConnectivityState = currentState;
   }
 
   void init() {

rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java

@@ -59,6 +59,7 @@
 import io.grpc.MetricRecorder.BatchRecorder;
 import io.grpc.MetricRecorder.Registration;
 import io.grpc.NameResolver.ConfigOrError;
+import io.grpc.Server;
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.SynchronizationContext;
@@ -170,6 +171,7 @@ public void uncaughtException(Thread t, Throwable e) {
   private CachingRlsLbClient rlsLbClient;
   private Map<String, ?> rlsChannelServiceConfig;
   private String rlsChannelOverriddenAuthority;
+  private Server server;
 
   private void setUpRlsLbClient() {
     fakeThrottler.resetCounts();
@@ -336,6 +338,33 @@ public void get_throttledAndRecover() throws Exception {
     assertThat(fakeHelper.lastPicker.toString()).isEqualTo("RlsPicker{target=service1}");
   }
 
+  @Test
+  public void controlPlaneTransientToReady_backOffEntriesRemovedAndPickerUpdated() throws Exception {
+    setUpRlsLbClient();
+    RouteLookupRequest routeLookupRequest = RouteLookupRequest.create(ImmutableMap.of(
+        "server", "bigtable.googleapis.com", "service-key", "foo", "method-key", "bar"));
+    rlsServerImpl.setLookupTable(
+        ImmutableMap.of(
+            routeLookupRequest,
+            RouteLookupResponse.create(ImmutableList.of("target"), "header")));
+
+    fakeThrottler.nextResult = true;
+    fakeBackoffProvider.nextPolicy = createBackoffPolicy(10, TimeUnit.MILLISECONDS);
+
+    CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequest);
+
+    assertThat(resp.hasError()).isTrue();
+
+    // let it pass throttler
+    fakeThrottler.nextResult = false;
+    fakeClock.forwardTime(10, TimeUnit.MILLISECONDS);
+    // Backoff entry evicted from cache.
+    verify(evictionListener)
+        .onEviction(eq(routeLookupRequest), any(CacheEntry.class), eq(EvictionType.EXPLICIT));
+    // Assert that Rls LB policy picker was updated.
+    assertThat(fakeHelper.lastPicker.toString()).isEqualTo("RlsPicker{target=service1}");
+  }
+
   @Test
   public void get_updatesLbState() throws Exception {
     setUpRlsLbClient();
@@ -884,13 +913,13 @@ private final class FakeHelper extends Helper {
     @Override
     public ManagedChannelBuilder<?> createResolvingOobChannelBuilder(
         String target, ChannelCredentials creds) {
-      try {
-        grpcCleanupRule.register(
-            InProcessServerBuilder.forName(target)
-                .addService(rlsServerImpl)
-                .directExecutor()
-                .build()
-                .start());
+       try {
+         server = InProcessServerBuilder.forName(target)
+             .addService(rlsServerImpl)
+             .directExecutor()
+             .build()
+             .start();
+        grpcCleanupRule.register(server);
       } catch (IOException e) {
         throw new RuntimeException("cannot create server: " + target, e);
       }