xds: unify client and server handling HttpConnectionManager (#8228)

Enables parsing HttpConnectionManager filter for the server side TCP listener, with the same codepath for handling it on the client side. Major changes include:

- Remodeled LdsUpdate with HttpConnectionManager. Now LdsUpdate is an oneof of HttpConnectionManager (for client side) or Listener (for server side). Each of Listener's FiliterChain contains an HttpConnectionManager (required).
Refactored code for validating and parsing the TCP Listener (for server side), put it into ClientXdsClient. The common part of validating/parsing HttpConnectionManager is reused/shared for client side.
- Included the name of FilterChain in the parsed form. As specified by the API, each FilterChain has a unique name. If the name is not provided by the control plane, a UUID is used. FilterChain names can be used for bookkeeping a set of FilterChain easily (e.g., used as map key).
- Added methods isSupportedOnClients() and isSupportedOnServers() to the Filter interface. Parsing the top-level HttpFilter requires knowing if the HttpFilter implementation is supported for the target usage (client-side or server-side). Note, parsing override HttpFilter configs does not need to know whether the config is used for an HttpFilter that is only supported for the client-side or server side.
- Added a new kind of Route: Route with non-forwarding action. Updated the XdsNameResolver being able to handle Route with non-forwarding action: if such a Route is matched to an RPC, that RPC is failed. Note, it is possible that XdsNameResolver receives xDS updates with all Routes with non-forwarding action. That is, the service config will not reference any cluster. Such case can be handled by cluster_manager LB policy's LB config parser: the parser returns the error to Channel and the Channel will handle it as error service config.

Author: voidzcy

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

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+import io.grpc.xds.Filter.NamedFilterConfig;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * HttpConnectionManager is a network filter for proxying HTTP requests.
+ */
+@AutoValue
+abstract class HttpConnectionManager {
+  // Total number of nanoseconds to keep alive an HTTP request/response stream.
+  abstract long httpMaxStreamDurationNano();
+
+  // Name of the route configuration to be used for RDS resource discovery.
+  @Nullable
+  abstract String rdsName();
+
+  // List of virtual hosts that make up the route table.
+  @Nullable
+  abstract ImmutableList<VirtualHost> virtualHosts();
+
+  // List of http filter configs. Null if HttpFilter support is not enabled.
+  @Nullable
+  abstract ImmutableList<NamedFilterConfig> httpFilterConfigs();
+
+  static HttpConnectionManager forRdsName(long httpMaxStreamDurationNano, String rdsName,
+      @Nullable List<NamedFilterConfig> httpFilterConfigs) {
+    checkNotNull(rdsName, "rdsName");
+    return create(httpMaxStreamDurationNano, rdsName, null, httpFilterConfigs);
+  }
+
+  static HttpConnectionManager forVirtualHosts(long httpMaxStreamDurationNano,
+      List<VirtualHost> virtualHosts, @Nullable List<NamedFilterConfig> httpFilterConfigs) {
+    checkNotNull(virtualHosts, "virtualHosts");
+    return create(httpMaxStreamDurationNano, null, virtualHosts,
+        httpFilterConfigs);
+  }
+
+  private static HttpConnectionManager create(long httpMaxStreamDurationNano,
+      @Nullable String rdsName, @Nullable List<VirtualHost> virtualHosts,
+      @Nullable List<NamedFilterConfig> httpFilterConfigs) {
+    return new AutoValue_HttpConnectionManager(
+        httpMaxStreamDurationNano, rdsName,
+        virtualHosts == null ? null : ImmutableList.copyOf(virtualHosts),
+        httpFilterConfigs == null ? null : ImmutableList.copyOf(httpFilterConfigs));
+  }
+}

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

@@ -19,14 +19,16 @@
 import com.google.protobuf.Message;
 import io.grpc.ClientInterceptor;
 import io.grpc.LoadBalancer.PickSubchannelArgs;
+import io.grpc.ServerInterceptor;
 import io.grpc.xds.Filter.ClientInterceptorBuilder;
+import io.grpc.xds.Filter.ServerInterceptorBuilder;
 import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nullable;
 
 /**
  * Router filter implementation. Currently this filter does not parse any field in the config.
  */
-enum RouterFilter implements Filter, ClientInterceptorBuilder {
+enum RouterFilter implements Filter, ClientInterceptorBuilder, ServerInterceptorBuilder {
   INSTANCE;
 
   static final String TYPE_URL =
@@ -66,4 +68,11 @@ public ClientInterceptor buildClientInterceptor(
       ScheduledExecutorService scheduler) {
     return null;
   }
+
+  @Nullable
+  @Override
+  public ServerInterceptor buildServerInterceptor(
+      FilterConfig config, @Nullable Filter.FilterConfig overrideConfig) {
+    return null;
+  }
 }

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

@@ -57,12 +57,23 @@ public static VirtualHost create(
   abstract static class Route {
     abstract RouteMatch routeMatch();
 
+    @Nullable
     abstract RouteAction routeAction();
 
     abstract ImmutableMap<String, FilterConfig> filterConfigOverrides();
 
-    static Route create(
-        RouteMatch routeMatch, RouteAction routeAction,
+    static Route forAction(RouteMatch routeMatch, RouteAction routeAction,
+        Map<String, FilterConfig> filterConfigOverrides) {
+      return create(routeMatch, routeAction, filterConfigOverrides);
+    }
+
+    static Route forNonForwardingAction(RouteMatch routeMatch,
+        Map<String, FilterConfig> filterConfigOverrides) {
+      return create(routeMatch, null, filterConfigOverrides);
+    }
+
+    private static Route create(
+        RouteMatch routeMatch, @Nullable RouteAction routeAction,
         Map<String, FilterConfig> filterConfigOverrides) {
       return new AutoValue_VirtualHost_Route(
           routeMatch, routeAction, ImmutableMap.copyOf(filterConfigOverrides));

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

@@ -20,7 +20,6 @@
 
 import com.google.auto.value.AutoValue;
 import com.google.common.base.MoreObjects;
-import com.google.common.base.MoreObjects.ToStringHelper;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Any;
 import io.grpc.Status;
@@ -29,7 +28,6 @@
 import io.grpc.xds.Endpoints.LocalityLbEndpoints;
 import io.grpc.xds.EnvoyServerProtoData.Listener;
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
-import io.grpc.xds.Filter.NamedFilterConfig;
 import io.grpc.xds.LoadStatsManager2.ClusterDropStats;
 import io.grpc.xds.LoadStatsManager2.ClusterLocalityStats;
 import java.util.ArrayList;
@@ -48,92 +46,25 @@
  */
 abstract class XdsClient {
 
-  static final class LdsUpdate implements ResourceUpdate {
-    // Total number of nanoseconds to keep alive an HTTP request/response stream.
-    final long httpMaxStreamDurationNano;
-    // The name of the route configuration to be used for RDS resource discovery.
-    @Nullable
-    final String rdsName;
-    // The list virtual hosts that make up the route table.
-    @Nullable
-    final List<VirtualHost> virtualHosts;
-    // Filter instance names. Null if HttpFilter support is not enabled.
-    @Nullable final List<NamedFilterConfig> filterChain;
-    // Server side Listener.
+  @AutoValue
+  abstract static class LdsUpdate implements ResourceUpdate {
+    // Http level api listener configuration.
     @Nullable
-    final Listener listener;
-
-    LdsUpdate(
-        long httpMaxStreamDurationNano, String rdsName,
-        @Nullable List<NamedFilterConfig> filterChain) {
-      this(httpMaxStreamDurationNano, rdsName, null, filterChain);
-    }
-
-    LdsUpdate(
-        long httpMaxStreamDurationNano, List<VirtualHost> virtualHosts,
-        @Nullable List<NamedFilterConfig> filterChain) {
-      this(httpMaxStreamDurationNano, null, virtualHosts, filterChain);
-    }
-
-    private LdsUpdate(
-        long httpMaxStreamDurationNano, @Nullable String rdsName,
-        @Nullable List<VirtualHost> virtualHosts, @Nullable List<NamedFilterConfig> filterChain) {
-      this.httpMaxStreamDurationNano = httpMaxStreamDurationNano;
-      this.rdsName = rdsName;
-      this.virtualHosts = virtualHosts == null
-          ? null : Collections.unmodifiableList(new ArrayList<>(virtualHosts));
-      this.filterChain = filterChain == null ? null : Collections.unmodifiableList(filterChain);
-      this.listener = null;
-    }
-
-    LdsUpdate(Listener listener) {
-      this.listener = listener;
-      this.httpMaxStreamDurationNano = 0L;
-      this.rdsName = null;
-      this.filterChain = null;
-      this.virtualHosts = null;
-    }
+    abstract HttpConnectionManager httpConnectionManager();
 
-    @Override
-    public int hashCode() {
-      return Objects.hash(
-          httpMaxStreamDurationNano, rdsName, virtualHosts, filterChain, listener);
-    }
+    // Tcp level listener configuration.
+    @Nullable
+    abstract Listener listener();
 
-    @Override
-    public boolean equals(Object o) {
-      if (this == o) {
-        return true;
-      }
-      if (o == null || getClass() != o.getClass()) {
-        return false;
-      }
-      LdsUpdate that = (LdsUpdate) o;
-      return httpMaxStreamDurationNano == that.httpMaxStreamDurationNano
-          && Objects.equals(rdsName, that.rdsName)
-          && Objects.equals(virtualHosts, that.virtualHosts)
-          && Objects.equals(filterChain, that.filterChain)
-          && Objects.equals(listener, that.listener);
+    static LdsUpdate forApiListener(HttpConnectionManager httpConnectionManager) {
+      checkNotNull(httpConnectionManager, "httpConnectionManager");
+      return new AutoValue_XdsClient_LdsUpdate(httpConnectionManager, null);
     }
 
-    @Override
-    public String toString() {
-      ToStringHelper toStringHelper = MoreObjects.toStringHelper(this);
-      toStringHelper.add("httpMaxStreamDurationNano", httpMaxStreamDurationNano);
-      if (rdsName != null) {
-        toStringHelper.add("rdsName", rdsName);
-      } else {
-        toStringHelper.add("virtualHosts", virtualHosts);
-      }
-      if (filterChain != null) {
-        toStringHelper.add("filterChain", filterChain);
-      }
-      if (listener != null) {
-        toStringHelper.add("listener", listener);
-      }
-      return toStringHelper.toString();
+    static LdsUpdate forTcpListener(Listener listener) {
+      checkNotNull(listener, "listener");
+      return new AutoValue_XdsClient_LdsUpdate(null, listener);
     }
-
   }
 
   static final class RdsUpdate implements ResourceUpdate {

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

@@ -114,7 +114,7 @@ public void start() {
         new XdsClient.LdsResourceWatcher() {
           @Override
           public void onChanged(XdsClient.LdsUpdate update) {
-            releaseOldSuppliers(curListener.getAndSet(update.listener));
+            releaseOldSuppliers(curListener.getAndSet(update.listener()));
             reportSuccess();
           }
 

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

@@ -359,6 +359,10 @@ public Result selectConfig(PickSubchannelArgs args) {
           return Result.forError(
               Status.UNAVAILABLE.withDescription("Could not find xDS route matching RPC"));
         }
+        if (selectedRoute.routeAction() == null) {
+          return Result.forError(Status.UNAVAILABLE.withDescription(
+              "Could not route RPC to Route with non-forwarding action"));
+        }
         RouteAction action = selectedRoute.routeAction();
         if (action.cluster() != null) {
           cluster = action.cluster();
@@ -648,14 +652,17 @@ public void run() {
             return;
           }
           logger.log(XdsLogLevel.INFO, "Receive LDS resource update: {0}", update);
-          List<VirtualHost> virtualHosts = update.virtualHosts;
-          String rdsName = update.rdsName;
+          HttpConnectionManager httpConnectionManager = update.httpConnectionManager();
+          List<VirtualHost> virtualHosts = httpConnectionManager.virtualHosts();
+          String rdsName = httpConnectionManager.rdsName();
           cleanUpRouteDiscoveryState();
           if (virtualHosts != null) {
-            updateRoutes(virtualHosts, update.httpMaxStreamDurationNano, update.filterChain);
+            updateRoutes(virtualHosts, httpConnectionManager.httpMaxStreamDurationNano(),
+                httpConnectionManager.httpFilterConfigs());
           } else {
             routeDiscoveryState = new RouteDiscoveryState(
-                rdsName, update.httpMaxStreamDurationNano, update.filterChain);
+                rdsName, httpConnectionManager.httpMaxStreamDurationNano(),
+                httpConnectionManager.httpFilterConfigs());
             logger.log(XdsLogLevel.INFO, "Start watching RDS resource {0}", rdsName);
             xdsClient.watchRdsResource(rdsName, routeDiscoveryState);
           }
@@ -739,11 +746,13 @@ private void updateRoutes(List<VirtualHost> virtualHosts, long httpMaxStreamDura
       Set<String> clusters = new HashSet<>();
       for (Route route : routes) {
         RouteAction action = route.routeAction();
-        if (action.cluster() != null) {
-          clusters.add(action.cluster());
-        } else if (action.weightedClusters() != null) {
-          for (ClusterWeight weighedCluster : action.weightedClusters()) {
-            clusters.add(weighedCluster.name());
+        if (action != null) {
+          if (action.cluster() != null) {
+            clusters.add(action.cluster());
+          } else if (action.weightedClusters() != null) {
+            for (ClusterWeight weighedCluster : action.weightedClusters()) {
+              clusters.add(weighedCluster.name());
+            }
           }
         }
       }