xds: unify client and server handling HttpConnectionManager

xds/src/main/java/io/grpc/xds/HttpConnectionManager.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/RouterFilter.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/VirtualHost.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/XdsClient.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/XdsClientWrapperForServerSds.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/XdsNameResolver.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());
+            }
           }
         }
       }