xds: gRFC A88 - Changes to XdsClient Watcher APIs

xds/src/main/java/io/grpc/xds/XdsDependencyManager.java

@@ -632,6 +632,9 @@ private abstract class XdsWatcherBase<T extends ResourceUpdate>
 
     @Nullable
     private StatusOr<T> data;
+    @Nullable
+    @SuppressWarnings("unused")
+    private Status ambientError;
 
 
     private XdsWatcherBase(XdsResourceType<T> type, String resourceName) {
@@ -640,42 +643,39 @@ private XdsWatcherBase(XdsResourceType<T> type, String resourceName) {
     }
 
     @Override
-    public void onError(Status error) {
-      checkNotNull(error, "error");
+    public void onResourceChanged(StatusOr<T> update) {
       if (cancelled) {
         return;
       }
-      // Don't update configuration on error, if we've already received configuration
-      if (!hasDataValue()) {
-        this.data = StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
-            String.format("Error retrieving %s: %s: %s",
-              toContextString(), error.getCode(), error.getDescription())));
-        maybePublishConfig();
-      }
-    }
+      ambientError = null;
+      if (update.hasValue()) {
+        data = update;
+        subscribeToChildren(update.getValue());
+      } else {
+        Status status = update.getStatus();
+        Status translatedStatus = Status.UNAVAILABLE.withDescription(
+            String.format("Error retrieving %s: %s. Details: %s%s",
+                toContextString(),
+                status.getCode(),
+                status.getDescription() != null ? status.getDescription() : "",
+                nodeInfo()));
 
-    @Override
-    public void onResourceDoesNotExist(String resourceName) {
-      if (cancelled) {
-        return;
+        data = StatusOr.fromStatus(translatedStatus);
       }
-
-      checkArgument(this.resourceName.equals(resourceName), "Resource name does not match");
-      this.data = StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
-          toContextString() + " does not exist" + nodeInfo()));
       maybePublishConfig();
     }
 
     @Override
-    public void onChanged(T update) {
-      checkNotNull(update, "update");
+    public void onAmbientError(Status error) {
       if (cancelled) {
         return;
       }
-
-      this.data = StatusOr.fromValue(update);
-      subscribeToChildren(update);
-      maybePublishConfig();
+      ambientError = error.withDescription(
+          String.format("Ambient error for %s: %s. Details: %s%s",
+              toContextString(),
+              error.getCode(),
+              error.getDescription() != null ? error.getDescription() : "",
+              nodeInfo()));
     }
 
     protected abstract void subscribeToChildren(T update);

xds/src/main/java/io/grpc/xds/XdsServerWrapper.java

@@ -42,6 +42,7 @@
 import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
 import io.grpc.StatusException;
+import io.grpc.StatusOr;
 import io.grpc.SynchronizationContext;
 import io.grpc.SynchronizationContext.ScheduledHandle;
 import io.grpc.internal.GrpcUtil;
@@ -382,18 +383,30 @@ private DiscoveryState(String resourceName) {
     }
 
     @Override
-    public void onChanged(final LdsUpdate update) {
+    public void onResourceChanged(final StatusOr<LdsUpdate> update) {
       if (stopped) {
         return;
       }
-      logger.log(Level.FINEST, "Received Lds update {0}", update);
-      if (update.listener() == null) {
-        onResourceDoesNotExist("Non-API");
+
+      if (!update.hasValue()) {
+        Status status = update.getStatus();
+        StatusException statusException = Status.UNAVAILABLE.withDescription(
+                String.format("Listener %s unavailable: %s", resourceName, status.getDescription()))
+            .withCause(status.asException())
+            .asException();
+        handleConfigNotFoundOrMismatch(statusException);
         return;
       }
 
-      String ldsAddress = update.listener().address();
-      if (ldsAddress == null || update.listener().protocol() != Protocol.TCP
+      final LdsUpdate ldsUpdate = update.getValue();
+      logger.log(Level.FINEST, "Received Lds update {0}", ldsUpdate);
+      if (ldsUpdate.listener() == null) {
+        handleConfigNotFoundOrMismatch(
+            Status.NOT_FOUND.withDescription("Listener is null in LdsUpdate").asException());
+        return;
+      }
+      String ldsAddress = ldsUpdate.listener().address();
+      if (ldsAddress == null || ldsUpdate.listener().protocol() != Protocol.TCP
           || !ipAddressesMatch(ldsAddress)) {
         handleConfigNotFoundOrMismatch(
             Status.UNKNOWN.withDescription(
@@ -402,16 +415,15 @@ public void onChanged(final LdsUpdate update) {
                     listenerAddress, ldsAddress)).asException());
         return;
       }
+
       if (!pendingRds.isEmpty()) {
         // filter chain state has not yet been applied to filterChainSelectorManager and there
-        // are two sets of sslContextProviderSuppliers, so we release the old ones.
         releaseSuppliersInFlight();
         pendingRds.clear();
       }
 
-      filterChains = update.listener().filterChains();
-      defaultFilterChain = update.listener().defaultFilterChain();
-      // Filters are loaded even if the server isn't serving yet.
+      filterChains = ldsUpdate.listener().filterChains();
+      defaultFilterChain = ldsUpdate.listener().defaultFilterChain();
       updateActiveFilters();
 
       List<FilterChain> allFilterChains = filterChains;
@@ -450,43 +462,33 @@ public void onChanged(final LdsUpdate update) {
       }
     }
 
-    private boolean ipAddressesMatch(String ldsAddress) {
-      HostAndPort ldsAddressHnP = HostAndPort.fromString(ldsAddress);
-      HostAndPort listenerAddressHnP = HostAndPort.fromString(listenerAddress);
-      if (!ldsAddressHnP.hasPort() || !listenerAddressHnP.hasPort()
-          || ldsAddressHnP.getPort() != listenerAddressHnP.getPort()) {
-        return false;
-      }
-      InetAddress listenerIp = InetAddresses.forString(listenerAddressHnP.getHost());
-      InetAddress ldsIp = InetAddresses.forString(ldsAddressHnP.getHost());
-      return listenerIp.equals(ldsIp);
-    }
-
-    @Override
-    public void onResourceDoesNotExist(final String resourceName) {
-      if (stopped) {
-        return;
-      }
-      StatusException statusException = Status.UNAVAILABLE.withDescription(
-          String.format("Listener %s unavailable, xDS node ID: %s", resourceName,
-              xdsClient.getBootstrapInfo().node().getId())).asException();
-      handleConfigNotFoundOrMismatch(statusException);
-    }
-
     @Override
-    public void onError(final Status error) {
+    public void onAmbientError(final Status error) {
       if (stopped) {
         return;
       }
       String description = error.getDescription() == null ? "" : error.getDescription() + " ";
       Status errorWithNodeId = error.withDescription(
           description + "xDS node ID: " + xdsClient.getBootstrapInfo().node().getId());
       logger.log(Level.FINE, "Error from XdsClient", errorWithNodeId);
+
       if (!isServing) {
         listener.onNotServing(errorWithNodeId.asException());
       }
     }
 
+    private boolean ipAddressesMatch(String ldsAddress) {
+      HostAndPort ldsAddressHnP = HostAndPort.fromString(ldsAddress);
+      HostAndPort listenerAddressHnP = HostAndPort.fromString(listenerAddress);
+      if (!ldsAddressHnP.hasPort() || !listenerAddressHnP.hasPort()
+          || ldsAddressHnP.getPort() != listenerAddressHnP.getPort()) {
+        return false;
+      }
+      InetAddress listenerIp = InetAddresses.forString(listenerAddressHnP.getHost());
+      InetAddress ldsIp = InetAddresses.forString(ldsAddressHnP.getHost());
+      return listenerIp.equals(ldsIp);
+    }
+
     private void shutdown() {
       stopped = true;
       cleanUpRouteDiscoveryStates();
@@ -775,54 +777,42 @@ private RouteDiscoveryState(String resourceName) {
       }
 
       @Override
-      public void onChanged(final RdsUpdate update) {
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (!routeDiscoveryStates.containsKey(resourceName)) {
-              return;
-            }
-            if (savedVirtualHosts == null && !isPending) {
-              logger.log(Level.WARNING, "Received valid Rds {0} configuration.", resourceName);
-            }
-            savedVirtualHosts = ImmutableList.copyOf(update.virtualHosts);
-            updateRdsRoutingConfig();
-            maybeUpdateSelector();
+      public void onResourceChanged(final StatusOr<RdsUpdate> update) {
+        syncContext.execute(() -> {
+          if (!routeDiscoveryStates.containsKey(resourceName)) {
+            return; // Watcher has been cancelled.
           }
-        });
-      }
 
-      @Override
-      public void onResourceDoesNotExist(final String resourceName) {
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (!routeDiscoveryStates.containsKey(resourceName)) {
-              return;
+          if (update.hasValue()) {
+            if (savedVirtualHosts == null && !isPending) {
+              logger.log(Level.WARNING, "Received valid Rds {0} configuration.", resourceName);
             }
-            logger.log(Level.WARNING, "Rds {0} unavailable", resourceName);
+            savedVirtualHosts = ImmutableList.copyOf(update.getValue().virtualHosts);
+          } else {
+            logger.log(Level.WARNING, "Rds {0} unavailable: {1}",
+                new Object[]{resourceName, update.getStatus()});
             savedVirtualHosts = null;
-            updateRdsRoutingConfig();
-            maybeUpdateSelector();
           }
+          // In both cases, a change has occurred that requires a config update.
+          updateRdsRoutingConfig();
+          maybeUpdateSelector();
         });
       }
 
       @Override
-      public void onError(final Status error) {
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (!routeDiscoveryStates.containsKey(resourceName)) {
-              return;
-            }
-            String description = error.getDescription() == null ? "" : error.getDescription() + " ";
-            Status errorWithNodeId = error.withDescription(
-                    description + "xDS node ID: " + xdsClient.getBootstrapInfo().node().getId());
-            logger.log(Level.WARNING, "Error loading RDS resource {0} from XdsClient: {1}.",
-                    new Object[]{resourceName, errorWithNodeId});
-            maybeUpdateSelector();
+      public void onAmbientError(final Status error) {
+        syncContext.execute(() -> {
+          if (!routeDiscoveryStates.containsKey(resourceName)) {
+            return; // Watcher has been cancelled.
           }
+          String description = error.getDescription() == null ? "" : error.getDescription() + " ";
+          Status errorWithNodeId = error.withDescription(
+              description + "xDS node ID: " + xdsClient.getBootstrapInfo().node().getId());
+          logger.log(Level.WARNING, "Error loading RDS resource {0} from XdsClient: {1}.",
+              new Object[]{resourceName, errorWithNodeId});
+
+          // Per gRFC A88, ambient errors should not trigger a configuration change.
+          // Therefore, we do NOT call maybeUpdateSelector() here.
         });
       }
 

xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java

@@ -262,7 +262,9 @@ private List<ServerInfo> parseServerInfos(List<?> rawServerConfigs, XdsLogger lo
       List<?> serverFeatures = JsonUtil.getList(serverConfig, "server_features");
       if (serverFeatures != null) {
         logger.log(XdsLogLevel.INFO, "Server features: {0}", serverFeatures);
-        ignoreResourceDeletion = serverFeatures.contains(SERVER_FEATURE_IGNORE_RESOURCE_DELETION);
+        if (serverFeatures.contains(SERVER_FEATURE_IGNORE_RESOURCE_DELETION)) {
+          ignoreResourceDeletion = true;
+        }
         resourceTimerIsTransientError = xdsDataErrorHandlingEnabled
             && serverFeatures.contains(SERVER_FEATURE_RESOURCE_TIMER_IS_TRANSIENT_ERROR);
       }

xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java

@@ -457,10 +457,10 @@ private void handleRpcStreamClosed(Status status) {
       if (responseReceived) {
         // A closed ADS stream after a successful response is not considered an error. Servers may
         // close streams for various reasons during normal operation, such as load balancing or
-        // underlying connection hitting its max connection age limit  (see gRFC A9).
+        // underlying connection hitting its max connection age limit (see gRFC A9).
         if (!status.isOk()) {
           newStatus = Status.OK;
-          logger.log( XdsLogLevel.DEBUG, "ADS stream closed with error {0}: {1}. However, a "
+          logger.log(XdsLogLevel.DEBUG, "ADS stream closed with error {0}: {1}. However, a "
               + "response was received, so this will not be treated as an error. Cause: {2}",
               status.getCode(), status.getDescription(), status.getCause());
         } else {

xds/src/main/java/io/grpc/xds/client/XdsClient.java

@@ -27,6 +27,7 @@
 import com.google.protobuf.Any;
 import io.grpc.ExperimentalApi;
 import io.grpc.Status;
+import io.grpc.StatusOr;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -139,30 +140,29 @@ public interface ResourceUpdate {}
 
   /**
    * Watcher interface for a single requested xDS resource.
+   *
+   * <p>Note that we expect that the implementer to:
+   * - Comply with the guarantee to not generate certain statuses by the library:
+   *   https://grpc.github.io/grpc/core/md_doc_statuscodes.html. If the code needs to be
+   *   propagated to the channel, override it with {@link io.grpc.Status.Code#UNAVAILABLE}.
+   * - Keep {@link Status} description in one form or another, as it contains valuable debugging
+   *   information.
    */
   @ExperimentalApi("https://github.com/grpc/grpc-java/issues/10862")
   public interface ResourceWatcher<T extends ResourceUpdate> {
 
     /**
-     * Called when the resource discovery RPC encounters some transient error.
-     *
-     * <p>Note that we expect that the implementer to:
-     * - Comply with the guarantee to not generate certain statuses by the library:
-     *   https://grpc.github.io/grpc/core/md_doc_statuscodes.html. If the code needs to be
-     *   propagated to the channel, override it with {@link io.grpc.Status.Code#UNAVAILABLE}.
-     * - Keep {@link Status} description in one form or another, as it contains valuable debugging
-     *   information.
+     * Called to deliver a resource update or an error. If an error is passed after a valid
+     * resource has been delivered, the watcher should stop using the previously delivered
+     * resource.
      */
-    void onError(Status error);
+    void onResourceChanged(StatusOr<T> update);
 
     /**
-     * Called when the requested resource is not available.
-     *
-     * @param resourceName name of the resource requested in discovery request.
-     */
-    void onResourceDoesNotExist(String resourceName);
-
-    void onChanged(T update);
+     * Called to deliver a transient error that should not affect the watcher's use of any
+     * previously received resource.
+     *   */
+    void onAmbientError(Status error);
   }
 
   /**