Restore experimental property in @after of test

larry-safran · larry-safran · commit 003348bd8f48 · 2024-10-31T15:42:09.000-07:00
Fix fallback logic for new watch triggering fallback to use the right CPC.
Some non-functional clarity improvement
Change hasSubscribers detection to be across all types and not trying to involve any logic in ControlPlaneClient.
diff --git a/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java b/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java
@@ -280,16 +280,6 @@ void sendDiscoveryRequests() {
     }
   }
 
-  @SuppressWarnings("rawtypes")
-  boolean hasSubscribedResources(String authority) {
-    for (XdsResourceType<?> type : resourceStore.getSubscribedResourceTypesWithTypeUrl().values()) {
-      if (resourceStore.hasSubscribers(type, authority)) {
-        return true;
-      }
-    }
-    return false;
-  }
-
   public boolean isStartingUp() {
     return startingUp;
   }
diff --git a/xds/src/main/java/io/grpc/xds/client/XdsClient.java b/xds/src/main/java/io/grpc/xds/client/XdsClient.java
@@ -428,7 +428,6 @@ Collection<String> getSubscribedResources(
 
     Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();
 
-    boolean hasSubscribers(XdsResourceType<? extends ResourceUpdate> type, String authority);
-
+    boolean hasSubscribers(String authority);
   }
 }
diff --git a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java
@@ -199,11 +199,17 @@ public Collection<String> getSubscribedResources(
   }
 
   @Override
-  public boolean hasSubscribers(XdsResourceType<? extends ResourceUpdate> type, String authority) {
-    Map<String, ResourceSubscriber<? extends ResourceUpdate>> resources =
-        resourceSubscribers.getOrDefault(type, Collections.emptyMap());
-    return resources.values().stream()
-        .anyMatch(res -> Objects.equals(res.authority, authority));
+  public boolean hasSubscribers(String authority) {
+    for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> map
+        : resourceSubscribers.values()) {
+      for (ResourceSubscriber<? extends ResourceUpdate> subscriber : map.values()) {
+        if (subscriber != null && Objects.equals(subscriber.authority, authority)) {
+          return true;
+        }
+      }
+    }
+
+    return false;
   }
 
   // As XdsClient APIs becomes resource agnostic, subscribed resource types are dynamic.
@@ -261,7 +267,6 @@ public void run() {
           subscriber.addWatcher(watcher, watcherExecutor);
           boolean hadErrorBeforeCpc = subscriber.errorDescription != null;
 
-          ControlPlaneClient originalCpc = getActiveCpc(subscriber.authority);
           ControlPlaneClient cpcToUse = subscriber.loadControlPlane();
 
           if (subscriber.errorDescription != null) {
@@ -273,9 +278,10 @@ public void run() {
             return;
           }
 
-          doFallbackIfNecessary(originalCpc, subscriber.authority);
+          doFallbackIfNecessary(cpcToUse, subscriber.authority);
 
           if (cpcToUse != null) {
+            addCpcToAuthority(subscriber.authority, cpcToUse);
             cpcToUse.adjustResourceSubscription(type);
             subscriber.restartTimer();
           } else {
@@ -291,16 +297,14 @@ public void run() {
   }
 
   private void addCpcToAuthority(String authority, ControlPlaneClient cpcToUse) {
-    List<ControlPlaneClient> controlPlaneClients = activeCpClients.get(authority);
-    if (controlPlaneClients == null) {
-      controlPlaneClients = new ArrayList<>();
-      activeCpClients.put(authority, controlPlaneClients);
-    }
+    List<ControlPlaneClient> controlPlaneClients =
+        activeCpClients.computeIfAbsent(authority, k -> new ArrayList<>());
+
     if (controlPlaneClients.contains(cpcToUse)) {
       return;
     }
 
-    // if there are any missing cpcs between the last one and cpcToUse, add them + cpcToUse
+    // if there are any missing CPCs between the last one and cpcToUse, add them + add cpcToUse
     ImmutableList<ServerInfo> serverInfos = getServerInfos(authority);
     for (int i = controlPlaneClients.size(); i < serverInfos.size(); i++) {
       ServerInfo serverInfo = serverInfos.get(i);
@@ -607,8 +611,8 @@ private void shutdownLowerPriorityCpcs(ControlPlaneClient activatedCpc) {
       int index = cpcsForAuth.indexOf(activatedCpc);
       if (index > -1 && index < cpcsForAuth.size() - 1) {
         cpcsToShutdown.addAll(cpcsForAuth.subList(index + 1, cpcsForAuth.size()));
-        for (int i = cpcsForAuth.size() - 1; i > index; i--) {
-          cpcsForAuth.remove(i);
+        if (cpcsForAuth.size() > index + 1) {
+          cpcsForAuth.subList(index + 1, cpcsForAuth.size()).clear();
         }
       }
     }
@@ -660,7 +664,7 @@ private boolean fallBackToCpc(ControlPlaneClient fallbackCpc, String authority,
           fallbackCpc.getServerInfo().target());
 
       // Get authorities that aren't falling back
-      // If don't already have a cached LDS resource, cache current data value
+      // If we don't already have a cached LDS resource, cache the current data value
 
       addCpcToAuthority(authority, fallbackCpc);
       restartMatchingSubscriberTimers(authority);
@@ -673,17 +677,15 @@ private boolean fallBackToCpc(ControlPlaneClient fallbackCpc, String authority,
     return didFallback;
   }
 
-  private void doFallbackIfNecessary(ControlPlaneClient cpc, String authority) {
-    if (cpc == null || !BootstrapperImpl.isEnabledXdsFallback()) {
+  private void doFallbackIfNecessary(ControlPlaneClient cpcToUse, String authority) {
+    if (cpcToUse == null) {
       return;
     }
 
     ControlPlaneClient activeCpClient = getActiveCpc(authority);
-    if (cpc == activeCpClient || activeCpClient == null) {
-      return;
+    if (cpcToUse != activeCpClient && activeCpClient != null) {
+      fallBackToCpc(cpcToUse, authority, activeCpClient);
     }
-
-    fallBackToCpc(cpc, authority, activeCpClient);
   }
 
   private void restartMatchingSubscriberTimers(String authority) {
@@ -744,7 +746,7 @@ private ControlPlaneClient loadControlPlane() {
         return null;
       }
 
-      ControlPlaneClient controlPlaneClient = null;
+      ControlPlaneClient controlPlaneClient;
       try {
         controlPlaneClient = getOrCreateControlPlaneClient(serverInfos);
         if (controlPlaneClient == null) {
@@ -1014,8 +1016,7 @@ public void handleStreamClosed(Status status, boolean inRetry,
       }
 
       Status error = status.isOk() ? Status.UNAVAILABLE.withDescription(CLOSED_BY_SERVER) : status;
-      boolean checkForFallback = !status.isOk() && BootstrapperImpl.isEnabledXdsFallback()
-          && !inRetry;
+      boolean checkForFallback = !status.isOk() && !inRetry;
 
       List<String> authoritiesForClosedCpc = getActiveAuthorities(cpcThatClosed);
       for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap :
@@ -1026,13 +1027,14 @@ public void handleStreamClosed(Status status, boolean inRetry,
           }
 
           // try to fallback to lower priority control plane client
-          if (checkForFallback && doFallbackForAuthority(
-              cpcThatClosed, serverInfo, subscriber.serverInfos, subscriber.authority)) {
+          if (checkForFallback
+              && doFallbackForAuthority(
+                  cpcThatClosed, serverInfo, subscriber.serverInfos, subscriber.authority)) {
             authoritiesForClosedCpc.remove(subscriber.authority);
             if (authoritiesForClosedCpc.isEmpty()) {
               return; // optimization: no need to continue once all authorities have done fallback
             }
-            continue;
+            continue; // since we did fallback, don't consider it an error
           }
 
           subscriber.onError(error, null);
@@ -1053,7 +1055,7 @@ public void handleStreamRestarted(ServerInfo serverInfo) {
       }
 
       for (String authority : getAuthoritiesForServerInfo(serverInfo)) {
-        if (controlPlaneClient.hasSubscribedResources(authority)) {
+        if (hasSubscribers(authority)) {
           internalHandleStreamReady(serverInfo, controlPlaneClient, authority);
         }
       }
diff --git a/xds/src/test/java/io/grpc/xds/CsdsServiceTest.java b/xds/src/test/java/io/grpc/xds/CsdsServiceTest.java
@@ -508,8 +508,7 @@ public Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl() {
     }
 
     @Override
-    public boolean hasSubscribers(XdsResourceType<? extends ResourceUpdate> type,
-                                  String authority) {
+    public boolean hasSubscribers(String authority) {
       return true;
     }
   }
diff --git a/xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java b/xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java
@@ -61,6 +61,7 @@ public class GrpcBootstrapperImplTest {
   private String originalBootstrapPathFromSysProp;
   private String originalBootstrapConfigFromEnvVar;
   private String originalBootstrapConfigFromSysProp;
+  private String originalExperimentalXdsFallbackFlag;
 
   @Before
   public void setUp() {
@@ -74,6 +75,8 @@ private void saveEnvironment() {
     originalBootstrapPathFromSysProp = bootstrapper.bootstrapPathFromSysProp;
     originalBootstrapConfigFromEnvVar = bootstrapper.bootstrapConfigFromEnvVar;
     originalBootstrapConfigFromSysProp = bootstrapper.bootstrapConfigFromSysProp;
+    originalExperimentalXdsFallbackFlag =
+        System.getProperty(BootstrapperImpl.GRPC_EXPERIMENTAL_XDS_FALLBACK);
   }
 
   @After
@@ -82,6 +85,12 @@ public void restoreEnvironment() {
     bootstrapper.bootstrapPathFromSysProp = originalBootstrapPathFromSysProp;
     bootstrapper.bootstrapConfigFromEnvVar = originalBootstrapConfigFromEnvVar;
     bootstrapper.bootstrapConfigFromSysProp = originalBootstrapConfigFromSysProp;
+    if (originalExperimentalXdsFallbackFlag != null) {
+      System.setProperty(
+          BootstrapperImpl.GRPC_EXPERIMENTAL_XDS_FALLBACK, originalExperimentalXdsFallbackFlag);
+    } else {
+      System.clearProperty(BootstrapperImpl.GRPC_EXPERIMENTAL_XDS_FALLBACK);
+    }
   }
 
   @Test

Original file line number	Diff line number	Diff line change
`@@ -280,16 +280,6 @@ void sendDiscoveryRequests() {`
`280`	`280`	`}`
`281`	`281`	`}`
`282`	`282`
`283`		`- @SuppressWarnings("rawtypes")`
`284`		`- boolean hasSubscribedResources(String authority) {`
`285`		`- for (XdsResourceType<?> type : resourceStore.getSubscribedResourceTypesWithTypeUrl().values()) {`
`286`		`- if (resourceStore.hasSubscribers(type, authority)) {`
`287`		`- return true;`
`288`		`- }`
`289`		`- }`
`290`		`- return false;`
`291`		`- }`
`292`		`-`
`293`	`283`	`public boolean isStartingUp() {`
`294`	`284`	`return startingUp;`
`295`	`285`	`}`
Original file line number	Diff line number	Diff line change
`@@ -428,7 +428,6 @@ Collection<String> getSubscribedResources(`
`428`	`428`
`429`	`429`	`Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();`
`430`	`430`
`431`		`- boolean hasSubscribers(XdsResourceType<? extends ResourceUpdate> type, String authority);`
`432`		`-`
	`431`	`+ boolean hasSubscribers(String authority);`
`433`	`432`	`}`
`434`	`433`	`}`
Original file line number	Diff line number	Diff line change
`@@ -508,8 +508,7 @@ public Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl() {`
`508`	`508`	`}`
`509`	`509`
`510`	`510`	`@Override`
`511`		`- public boolean hasSubscribers(XdsResourceType<? extends ResourceUpdate> type,`
`512`		`- String authority) {`
	`511`	`+ public boolean hasSubscribers(String authority) {`
`513`	`512`	`return true;`
`514`	`513`	`}`
`515`	`514`	`}`