Gracefully handle tcp listeners. Fix error processing in XdsNR

Author: ejona86

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

@@ -243,6 +243,7 @@ XdsConfigBuilder setVirtualHost(VirtualHost virtualHost) {
     XdsConfig build() {
       checkNotNull(listener, "listener");
       checkNotNull(route, "route");
+      checkNotNull(virtualHost, "virtualHost");
       return new XdsConfig(listener, route, clusters, virtualHost);
     }
   }

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

@@ -87,7 +87,7 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
     syncContext.execute(() -> addWatcher(new LdsWatcher(listenerName)));
   }
 
-  public static String  toContextStr(String typeName, String resourceName) {
+  public static String toContextStr(String typeName, String resourceName) {
     return typeName + " resource: " + resourceName;
   }
 
@@ -280,8 +280,7 @@ XdsConfig buildConfig() {
     // Iterate watchers and build the XdsConfig
 
     // Will only be 1 listener and 1 route resource
-    RdsUpdate rdsUpdate = null;
-    XdsWatcherBase<?> routeSource = null;
+    RdsUpdateSupplier routeSource = null;
     for (XdsWatcherBase<XdsListenerResource.LdsUpdate> ldsWatcher :
         getWatchers(XdsListenerResource.getInstance()).values()) {
       if (!ldsWatcher.getData().hasValue()) {
@@ -290,23 +289,15 @@ XdsConfig buildConfig() {
       }
       XdsListenerResource.LdsUpdate ldsUpdate = ldsWatcher.getData().getValue();
       builder.setListener(ldsUpdate);
-      if (ldsUpdate.httpConnectionManager() != null
-          && ldsUpdate.httpConnectionManager().virtualHosts() != null) {
-        rdsUpdate = new RdsUpdate(ldsUpdate.httpConnectionManager().virtualHosts());
-        routeSource = ldsWatcher;
-      }
+      routeSource = ((LdsWatcher) ldsWatcher).getRouteSource();
     }
 
-    XdsWatcherBase<RdsUpdate> rdsWatcher = getWatchers(XdsRouteConfigureResource.getInstance())
-        .values().stream().findFirst().orElse(null);
-    if (rdsWatcher != null) {
-      if (!rdsWatcher.getData().hasValue()) {
-        xdsConfigWatcher.onError(rdsWatcher.toContextString(), rdsWatcher.getData().getStatus());
-        return null;
-      }
-      rdsUpdate = rdsWatcher.getData().getValue();
-      routeSource = rdsWatcher;
+    StatusOr<RdsUpdate> statusOrRdsUpdate = routeSource.getRdsUpdate();
+    if (!statusOrRdsUpdate.hasValue()) {
+      xdsConfigWatcher.onError(routeSource.toContextString(), statusOrRdsUpdate.getStatus());
+      return null;
     }
+    RdsUpdate rdsUpdate = statusOrRdsUpdate.getValue();
     builder.setRoute(rdsUpdate);
 
     VirtualHost activeVirtualHost =
@@ -492,13 +483,14 @@ private void addClusterWatcher(String clusterName, Object parentContext, int dep
   }
 
   private void updateRoutes(List<VirtualHost> virtualHosts, Object newParentContext,
-                            VirtualHost oldVirtualHost, boolean sameParentContext) {
-    Set<String> oldClusters = getClusterNamesFromVirtualHost(oldVirtualHost);
-
+                            List<VirtualHost> oldVirtualHosts, boolean sameParentContext) {
+    VirtualHost oldVirtualHost =
+        RoutingUtils.findVirtualHostForHostName(oldVirtualHosts, dataPlaneAuthority);
     VirtualHost virtualHost =
         RoutingUtils.findVirtualHostForHostName(virtualHosts, dataPlaneAuthority);
-    Set<String> newClusters =
-        virtualHost != null ? getClusterNamesFromVirtualHost(virtualHost) : Collections.emptySet();
+
+    Set<String> newClusters = getClusterNamesFromVirtualHost(virtualHost);
+    Set<String> oldClusters = getClusterNamesFromVirtualHost(oldVirtualHost);
 
     if (sameParentContext) {
       // Calculate diffs.
@@ -537,24 +529,6 @@ private static Set<String> getClusterNamesFromVirtualHost(VirtualHost virtualHos
     return clusters;
   }
 
-  @Nullable
-  private VirtualHost getActiveVirtualHost() {
-    TypeWatchers<?> rdsWatchers = resourceWatchers.get(XdsRouteConfigureResource.getInstance());
-    if (rdsWatchers == null) {
-      return null;
-    }
-
-    RdsWatcher activeRdsWatcher =
-        (RdsWatcher) rdsWatchers.watchers.values().stream().findFirst().orElse(null);
-    if (activeRdsWatcher == null || activeRdsWatcher.missingResult()
-        || !activeRdsWatcher.getData().hasValue()) {
-      return null;
-    }
-
-    return RoutingUtils.findVirtualHostForHostName(
-        activeRdsWatcher.getData().getValue().virtualHosts, dataPlaneAuthority);
-  }
-
   private CdsWatcher getCluster(String clusterName) {
     return (CdsWatcher) resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
   }
@@ -660,12 +634,19 @@ protected void setDataAsStatus(Status status) {
       this.data = StatusOr.fromStatus(status);
     }
 
-    String toContextString() {
+    public String toContextString() {
       return toContextStr(type.typeName(), resourceName);
     }
   }
 
-  private class LdsWatcher extends XdsWatcherBase<XdsListenerResource.LdsUpdate> {
+  private interface RdsUpdateSupplier {
+    StatusOr<RdsUpdate> getRdsUpdate();
+
+    String toContextString();
+  }
+
+  private class LdsWatcher extends XdsWatcherBase<XdsListenerResource.LdsUpdate>
+      implements RdsUpdateSupplier {
     String rdsName;
 
     private LdsWatcher(String resourceName) {
@@ -677,9 +658,20 @@ public void onChanged(XdsListenerResource.LdsUpdate update) {
       checkNotNull(update, "update");
 
       HttpConnectionManager httpConnectionManager = update.httpConnectionManager();
-      List<VirtualHost> virtualHosts = httpConnectionManager.virtualHosts();
-      String rdsName = httpConnectionManager.rdsName();
-      VirtualHost activeVirtualHost = getActiveVirtualHost();
+      List<VirtualHost> virtualHosts;
+      String rdsName;
+      if (httpConnectionManager == null) {
+        // TCP listener. Unsupported config
+        virtualHosts = Collections.emptyList(); // Not null, to not delegate to RDS
+        rdsName = null;
+      } else {
+        virtualHosts = httpConnectionManager.virtualHosts();
+        rdsName = httpConnectionManager.rdsName();
+      }
+      StatusOr<RdsUpdate> activeRdsUpdate = getRouteSource().getRdsUpdate();
+      List<VirtualHost> activeVirtualHosts = activeRdsUpdate.hasValue()
+          ? activeRdsUpdate.getValue().virtualHosts
+          : Collections.emptyList();
 
       boolean changedRdsName = !Objects.equals(rdsName, this.rdsName);
       if (changedRdsName) {
@@ -688,7 +680,7 @@ public void onChanged(XdsListenerResource.LdsUpdate update) {
 
       if (virtualHosts != null) {
         // No RDS watcher since we are getting RDS updates via LDS
-        updateRoutes(virtualHosts, this, activeVirtualHost, this.rdsName == null);
+        updateRoutes(virtualHosts, this, activeVirtualHosts, this.rdsName == null);
         this.rdsName = null;
       } else if (changedRdsName) {
         this.rdsName = rdsName;
@@ -720,8 +712,7 @@ private void cleanUpRdsWatcher() {
         logger.log(XdsLogger.XdsLogLevel.DEBUG, "Stop watching RDS resource {0}", rdsName);
 
         // Cleanup clusters (as appropriate) that had the old rds watcher as a parent
-        if (!oldRdsWatcher.hasDataValue() || !oldRdsWatcher.getData().hasValue()
-            || resourceWatchers.get(CLUSTER_RESOURCE) == null) {
+        if (!oldRdsWatcher.hasDataValue() || resourceWatchers.get(CLUSTER_RESOURCE) == null) {
           return;
         }
         for (XdsWatcherBase<?> watcher :
@@ -732,16 +723,58 @@ private void cleanUpRdsWatcher() {
     }
 
     private RdsWatcher getRdsWatcher() {
+      if (rdsName == null) {
+        return null;
+      }
       TypeWatchers<?> watchers = resourceWatchers.get(XdsRouteConfigureResource.getInstance());
-      if (watchers == null || rdsName == null || watchers.watchers.isEmpty()) {
+      if (watchers == null) {
         return null;
       }
-
       return (RdsWatcher) watchers.watchers.get(rdsName);
     }
+
+    public RdsUpdateSupplier getRouteSource() {
+      if (!hasDataValue()) {
+        return this;
+      }
+      HttpConnectionManager hcm = getData().getValue().httpConnectionManager();
+      if (hcm == null) {
+        return this;
+      }
+      List<VirtualHost> virtualHosts = hcm.virtualHosts();
+      if (virtualHosts != null) {
+        return this;
+      }
+      RdsWatcher rdsWatcher = getRdsWatcher();
+      assert rdsWatcher != null;
+      return rdsWatcher;
+    }
+
+    @Override
+    public StatusOr<RdsUpdate> getRdsUpdate() {
+      if (missingResult()) {
+        return StatusOr.fromStatus(Status.UNAVAILABLE.withDescription("Not yet loaded"));
+      }
+      if (!getData().hasValue()) {
+        return StatusOr.fromStatus(getData().getStatus());
+      }
+      HttpConnectionManager hcm = getData().getValue().httpConnectionManager();
+      if (hcm == null) {
+        return StatusOr.fromStatus(
+            Status.UNAVAILABLE.withDescription("Not an API listener"));
+      }
+      List<VirtualHost> virtualHosts = hcm.virtualHosts();
+      if (virtualHosts == null) {
+        // Code shouldn't trigger this case, as it should be calling RdsWatcher instead. This would
+        // be easily implemented with getRdsWatcher().getRdsUpdate(), but getting here is likely a
+        // bug
+        return StatusOr.fromStatus(Status.INTERNAL.withDescription("Routes are in RDS, not LDS"));
+      }
+      return StatusOr.fromValue(new RdsUpdate(virtualHosts));
+    }
   }
 
-  private class RdsWatcher extends XdsWatcherBase<RdsUpdate> {
+  private class RdsWatcher extends XdsWatcherBase<RdsUpdate> implements RdsUpdateSupplier {
 
     public RdsWatcher(String resourceName) {
       super(XdsRouteConfigureResource.getInstance(), checkNotNull(resourceName, "resourceName"));
@@ -750,13 +783,11 @@ public RdsWatcher(String resourceName) {
     @Override
     public void onChanged(RdsUpdate update) {
       checkNotNull(update, "update");
-      RdsUpdate oldData = hasDataValue() ? getData().getValue() : null;
-      VirtualHost oldVirtualHost =
-          (oldData != null)
-          ? RoutingUtils.findVirtualHostForHostName(oldData.virtualHosts, dataPlaneAuthority)
-          : null;
+      List<VirtualHost> oldVirtualHosts = hasDataValue()
+          ? getData().getValue().virtualHosts
+          : Collections.emptyList();
       setData(update);
-      updateRoutes(update.virtualHosts, this, oldVirtualHost, true);
+      updateRoutes(update.virtualHosts, this, oldVirtualHosts, true);
       maybePublishConfig();
     }
 
@@ -770,6 +801,14 @@ public void onResourceDoesNotExist(String resourceName) {
       xdsConfigWatcher.onResourceDoesNotExist(toContextString());
       lastXdsConfig = null; // Published an empty result
     }
+
+    @Override
+    public StatusOr<RdsUpdate> getRdsUpdate() {
+      if (missingResult()) {
+        return StatusOr.fromStatus(Status.UNAVAILABLE.withDescription("Not yet loaded"));
+      }
+      return getData();
+    }
   }
 
   private class CdsWatcher extends XdsWatcherBase<XdsClusterResource.CdsUpdate> {

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

@@ -133,9 +133,6 @@ final class XdsNameResolver extends NameResolver {
   private XdsClient xdsClient;
   private CallCounterProvider callCounterProvider;
   private ResolveState resolveState;
-  // Workaround for https://github.com/grpc/grpc-java/issues/8886 . This should be handled in
-  // XdsClient instead of here.
-  private boolean receivedConfig;
 
   XdsNameResolver(
       URI targetUri, String name, @Nullable String overrideAuthority,
@@ -319,7 +316,6 @@ private void updateResolutionResult() {
             .setServiceConfig(parsedServiceConfig)
             .build();
     listener.onResult(result);
-    receivedConfig = true;
   }
 
   /**
@@ -673,22 +669,21 @@ public void onUpdate(XdsConfig update) {
 
       // Process Route
       HttpConnectionManager httpConnectionManager = update.getListener().httpConnectionManager();
-      VirtualHost virtualHost = update.getVirtualHost(); // httpConnectionManager.virtualHosts();
+      VirtualHost virtualHost = update.getVirtualHost();
 
       updateRoutes(virtualHost, httpConnectionManager.httpMaxStreamDurationNano(),
           httpConnectionManager.httpFilterConfigs());
     }
 
     @Override
     public void onError(String resourceContext, Status error) {
-      if (stopped || receivedConfig) {
+      if (stopped) {
         return;
       }
-      String errorMsg = String.format("Unable to load %s. xDS server returned: %s: %s",
+      String errorMsg = String.format("Error for %s: %s: %s",
           resourceContext, error.getCode(), error.getDescription());
-      listener.onError(Status.UNAVAILABLE.withCause(error.getCause()).withDescription(errorMsg));
-
-      sendEmptyResult(errorMsg);
+      logger.log(XdsLogLevel.INFO, errorMsg);
+      cleanUpRoutes(errorMsg);
     }
 
     @Override
@@ -703,13 +698,6 @@ public void onResourceDoesNotExist(String resourceContext) {
 
     private void updateRoutes(@Nullable VirtualHost virtualHost, long httpMaxStreamDurationNano,
                               @Nullable List<NamedFilterConfig> filterConfigs) {
-      if (virtualHost == null) {
-        String error = "Failed to find virtual host matching hostname: " + authority;
-        logger.log(XdsLogLevel.WARNING, error);
-        cleanUpRoutes(error);
-        return;
-      }
-
       List<Route> routes = virtualHost.routes();
       ImmutableList.Builder<RouteData> routesData = ImmutableList.builder();
 
@@ -867,7 +855,6 @@ private void cleanUpRoutes(String error) {
       // failure information for addresses yet still providing a service config, the config seector
       // could be avoided.
       sendEmptyResult(error);
-      receivedConfig = true;
     }
 
     private void sendEmptyResult(String error) {

xds/src/test/java/io/grpc/xds/XdsDependencyManagerTest.java

@@ -399,6 +399,23 @@ public void testMissingLds() {
     testWatcher.verifyStats(0, 0, 1);
   }
 
+  @Test
+  public void testTcpListenerErrors() {
+    Listener serverListener =
+        ControlPlaneRule.buildServerListener().toBuilder().setName(serverName).build();
+    controlPlaneService.setXdsConfig(ADS_TYPE_URL_LDS, ImmutableMap.of(serverName, serverListener));
+    xdsDependencyManager = new XdsDependencyManager(xdsClient, xdsConfigWatcher, syncContext,
+        serverName, serverName, nameResolverArgs, scheduler);
+
+    fakeClock.forwardTime(16, TimeUnit.SECONDS);
+    Status expectedStatus = Status.UNAVAILABLE.withDescription("Not an API listener");
+    String context = toContextStr(XdsListenerResource.getInstance().typeName(), serverName);
+    verify(xdsConfigWatcher, timeout(1000))
+        .onError(eq(context), argThat(new XdsTestUtils.StatusMatcher(expectedStatus)));
+
+    testWatcher.verifyStats(0, 1, 0);
+  }
+
   @Test
   public void testMissingRds() {
     Listener clientListener =