xds: add terminal http filter verification, remove lame route filter, add hcm as terminal network filter verification

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

@@ -334,41 +334,32 @@ static FilterChain parseFilterChain(
       TlsContextManager tlsContextManager, FilterRegistry filterRegistry,
       Set<FilterChainMatch> uniqueSet, Set<String> certProviderInstances, boolean parseHttpFilters)
       throws ResourceInvalidException {
-    io.grpc.xds.HttpConnectionManager httpConnectionManager = null;
-    HashSet<String> uniqueNames = new HashSet<>();
-    for (io.envoyproxy.envoy.config.listener.v3.Filter filter : proto.getFiltersList()) {
-      if (!uniqueNames.add(filter.getName())) {
-        throw new ResourceInvalidException(
-            "FilterChain " + proto.getName() + " with duplicated filter: " + filter.getName());
-      }
-      if (!filter.hasTypedConfig()) {
-        throw new ResourceInvalidException(
-            "FilterChain " + proto.getName() + " contains filter " + filter.getName()
-                + " without typed_config");
-      }
-      Any any = filter.getTypedConfig();
-      // HttpConnectionManager is the only supported network filter at the moment.
-      if (!any.getTypeUrl().equals(TYPE_URL_HTTP_CONNECTION_MANAGER)) {
-        throw new ResourceInvalidException(
-            "FilterChain " + proto.getName() + " contains filter " + filter.getName()
-                + " with unsupported typed_config type " + any.getTypeUrl());
-      }
-      if (httpConnectionManager == null) {
-        HttpConnectionManager hcmProto;
-        try {
-          hcmProto = any.unpack(HttpConnectionManager.class);
-        } catch (InvalidProtocolBufferException e) {
-          throw new ResourceInvalidException("FilterChain " + proto.getName() + " with filter "
-              + filter.getName() + " failed to unpack message", e);
-        }
-        httpConnectionManager = parseHttpConnectionManager(
-            hcmProto, rdsResources, filterRegistry, parseHttpFilters, false /* isForClient */);
-      }
-    }
-    if (httpConnectionManager == null) {
+    if (proto.getFiltersCount() != 1) {
       throw new ResourceInvalidException("FilterChain " + proto.getName()
-          + " missing required HttpConnectionManager filter");
+              + " should contain exact one HttpConnectionManager filter");
     }
+    io.envoyproxy.envoy.config.listener.v3.Filter filter = proto.getFiltersList().get(0);
+    if (!filter.hasTypedConfig()) {
+      throw new ResourceInvalidException(
+          "FilterChain " + proto.getName() + " contains filter " + filter.getName()
+              + " without typed_config");
+    }
+    Any any = filter.getTypedConfig();
+    // HttpConnectionManager is the only supported network filter at the moment.
+    if (!any.getTypeUrl().equals(TYPE_URL_HTTP_CONNECTION_MANAGER)) {
+      throw new ResourceInvalidException(
+          "FilterChain " + proto.getName() + " contains filter " + filter.getName()
+              + " with unsupported typed_config type " + any.getTypeUrl());
+    }
+    HttpConnectionManager hcmProto;
+    try {
+      hcmProto = any.unpack(HttpConnectionManager.class);
+    } catch (InvalidProtocolBufferException e) {
+      throw new ResourceInvalidException("FilterChain " + proto.getName() + " with filter "
+          + filter.getName() + " failed to unpack message", e);
+    }
+    io.grpc.xds.HttpConnectionManager httpConnectionManager = parseHttpConnectionManager(
+            hcmProto, rdsResources, filterRegistry, parseHttpFilters, false /* isForClient */);
 
     EnvoyServerProtoData.DownstreamTlsContext downstreamTlsContext = null;
     if (proto.hasTransportSocket()) {
@@ -762,17 +753,26 @@ static io.grpc.xds.HttpConnectionManager parseHttpConnectionManager(
     // Parse http filters.
     List<NamedFilterConfig> filterConfigs = null;
     if (parseHttpFilter) {
+      if (proto.getHttpFiltersList().isEmpty()) {
+        throw new ResourceInvalidException("Missing HttpFilter in HttpConnectionManager.");
+      }
       filterConfigs = new ArrayList<>();
       Set<String> names = new HashSet<>();
-      for (io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter
-               httpFilter : proto.getHttpFiltersList()) {
+      for (int i = 0; i < proto.getHttpFiltersCount(); i++) {
+        io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter
+                httpFilter = proto.getHttpFiltersList().get(i);
         String filterName = httpFilter.getName();
         if (!names.add(filterName)) {
           throw new ResourceInvalidException(
               "HttpConnectionManager contains duplicate HttpFilter: " + filterName);
         }
         StructOrError<FilterConfig> filterConfig =
             parseHttpFilter(httpFilter, filterRegistry, isForClient);
+        if ((i == proto.getHttpFiltersCount() - 1)
+                && (filterConfig == null || !isTerminalFilter(filterConfig.struct))) {
+          throw new ResourceInvalidException("The last HttpFilter must be a terminal filter: "
+                  + filterName);
+        }
         if (filterConfig == null) {
           continue;
         }
@@ -781,6 +781,10 @@ static io.grpc.xds.HttpConnectionManager parseHttpConnectionManager(
               "HttpConnectionManager contains invalid HttpFilter: "
                   + filterConfig.getErrorDetail());
         }
+        if ((i < proto.getHttpFiltersCount() - 1) && isTerminalFilter(filterConfig.getStruct())) {
+          throw new ResourceInvalidException("A terminal HttpFilter must be the last filter: "
+                  + filterName);
+        }
         filterConfigs.add(new NamedFilterConfig(filterName, filterConfig.struct));
       }
     }
@@ -821,6 +825,11 @@ static io.grpc.xds.HttpConnectionManager parseHttpConnectionManager(
         "HttpConnectionManager neither has inlined route_config nor RDS");
   }
 
+  // hard-coded: currently router config is the only terminal filter.
+  private static boolean isTerminalFilter(FilterConfig filterConfig) {
+    return RouterFilter.ROUTER_CONFIG.equals(filterConfig);
+  }
+
   @VisibleForTesting
   @Nullable // Returns null if the filter is optional but not supported.
   static StructOrError<FilterConfig> parseHttpFilter(

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

@@ -23,7 +23,6 @@
 import com.google.common.base.Joiner;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 import com.google.gson.Gson;
 import com.google.protobuf.util.Durations;
@@ -95,8 +94,6 @@ final class XdsNameResolver extends NameResolver {
       CallOptions.Key.create("io.grpc.xds.CLUSTER_SELECTION_KEY");
   static final CallOptions.Key<Long> RPC_HASH_KEY =
       CallOptions.Key.create("io.grpc.xds.RPC_HASH_KEY");
-  private static final NamedFilterConfig LAME_FILTER =
-      new NamedFilterConfig(null, LameFilter.LAME_CONFIG);
   @VisibleForTesting
   static boolean enableTimeout =
       Strings.isNullOrEmpty(System.getenv("GRPC_XDS_EXPERIMENTAL_ENABLE_TIMEOUT"))
@@ -374,10 +371,6 @@ public Result selectConfig(PickSubchannelArgs args) {
       do {
         routingCfg = routingConfig;
         selectedOverrideConfigs = new HashMap<>(routingCfg.virtualHostOverrideConfig);
-        if (routingCfg.filterChain != null
-            && Iterables.getLast(routingCfg.filterChain).equals(LAME_FILTER)) {
-          break;
-        }
         for (Route route : routingCfg.routes) {
           if (matchRoute(route.routeMatch(), "/" + args.getMethodDescriptor().getFullMethodName(),
               headers, random)) {
@@ -442,12 +435,7 @@ public Result selectConfig(PickSubchannelArgs args) {
       if (routingCfg.filterChain != null) {
         for (NamedFilterConfig namedFilter : routingCfg.filterChain) {
           FilterConfig filterConfig = namedFilter.filterConfig;
-          Filter filter;
-          if (namedFilter.equals(LAME_FILTER)) {
-            filter = LameFilter.INSTANCE;
-          } else {
-            filter = filterRegistry.get(filterConfig.typeUrl());
-          }
+          Filter filter = filterRegistry.get(filterConfig.typeUrl());
           if (filter instanceof ClientInterceptorBuilder) {
             ClientInterceptor interceptor = ((ClientInterceptorBuilder) filter)
                 .buildClientInterceptor(
@@ -458,12 +446,6 @@ public Result selectConfig(PickSubchannelArgs args) {
             }
           }
         }
-        if (Iterables.getLast(routingCfg.filterChain).equals(LAME_FILTER)) {
-          return Result.newBuilder()
-              .setConfig(config)
-              .setInterceptor(combineInterceptors(filterInterceptors))
-              .build();
-        }
       }
       final String finalCluster = cluster;
       final long hash = generateHash(selectedRoute.routeAction().hashPolicies(), headers);
@@ -754,27 +736,7 @@ private void updateRoutes(List<VirtualHost> virtualHosts, long httpMaxStreamDura
         return;
       }
 
-      // A router filter is required for request routing. For backward compatibility, routing
-      // is always enabled for gRPC clients without HttpFilter support.
       List<Route> routes = virtualHost.routes();
-      List<NamedFilterConfig> filterChain = null;
-      if (filterConfigs != null) {
-        boolean hasRouter = false;
-        filterChain = new ArrayList<>(filterConfigs.size());
-        for (NamedFilterConfig namedFilter : filterConfigs) {
-          filterChain.add(namedFilter);
-          if (namedFilter.filterConfig.equals(RouterFilter.ROUTER_CONFIG)) {
-            hasRouter = true;
-            break;
-          }
-        }
-        if (!hasRouter) {
-          // Fail all RPCs if a router filter is not present. Reference counts for all currently
-          // selectable clusters should be reclaimed.
-          filterChain.add(LAME_FILTER);
-          routes = Collections.emptyList();
-        }
-      }
 
       // Populate all clusters to which requests can be routed to through the virtual host.
       Set<String> clusters = new HashSet<>();
@@ -815,7 +777,7 @@ private void updateRoutes(List<VirtualHost> virtualHosts, long httpMaxStreamDura
       // selectable.
       routingConfig =
           new RoutingConfig(
-              httpMaxStreamDurationNano, routes, filterChain,
+              httpMaxStreamDurationNano, routes, filterConfigs,
               virtualHost.filterConfigOverrides());
       shouldUpdateResult = false;
       for (String cluster : deletedClusters) {