Merge branch 'grpc:master' into FixIssue-11194

Author: SreeramdasLavanya

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -1388,24 +1388,18 @@ public void updateBalancingState(
       syncContext.throwIfNotInThisSynchronizationContext();
       checkNotNull(newState, "newState");
       checkNotNull(newPicker, "newPicker");
-      final class UpdateBalancingState implements Runnable {
-        @Override
-        public void run() {
-          if (LbHelperImpl.this != lbHelper || panicMode) {
-            return;
-          }
-          updateSubchannelPicker(newPicker);
-          // It's not appropriate to report SHUTDOWN state from lb.
-          // Ignore the case of newState == SHUTDOWN for now.
-          if (newState != SHUTDOWN) {
-            channelLogger.log(
-                ChannelLogLevel.INFO, "Entering {0} state with picker: {1}", newState, newPicker);
-            channelStateManager.gotoState(newState);
-          }
-        }
-      }
 
-      syncContext.execute(new UpdateBalancingState());
+      if (LbHelperImpl.this != lbHelper || panicMode) {
+        return;
+      }
+      updateSubchannelPicker(newPicker);
+      // It's not appropriate to report SHUTDOWN state from lb.
+      // Ignore the case of newState == SHUTDOWN for now.
+      if (newState != SHUTDOWN) {
+        channelLogger.log(
+            ChannelLogLevel.INFO, "Entering {0} state with picker: {1}", newState, newPicker);
+        channelStateManager.gotoState(newState);
+      }
     }
 
     @Override

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

@@ -27,7 +27,6 @@
 import io.grpc.xds.client.EnvoyProtoData;
 import io.grpc.xds.internal.security.SslContextProviderSupplier;
 import java.net.InetAddress;
-import java.net.UnknownHostException;
 import java.util.Objects;
 import javax.annotation.Nullable;
 
@@ -150,9 +149,9 @@ abstract static class CidrRange {
 
     abstract int prefixLen();
 
-    static CidrRange create(String addressPrefix, int prefixLen) throws UnknownHostException {
+    static CidrRange create(InetAddress addressPrefix, int prefixLen) {
       return new AutoValue_EnvoyServerProtoData_CidrRange(
-          InetAddress.getByName(addressPrefix), prefixLen);
+          addressPrefix, prefixLen);
     }
   }
 
@@ -207,7 +206,7 @@ public static FilterChainMatch create(int destinationPort,
   @AutoValue
   abstract static class FilterChain {
 
-    // possibly empty
+    // Must be unique per server instance (except the default chain).
     abstract String name();
 
     // TODO(sanjaypujare): flatten structure by moving FilterChainMatch class members here.

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

@@ -20,6 +20,7 @@
 import com.google.protobuf.Message;
 import io.grpc.ClientInterceptor;
 import io.grpc.ServerInterceptor;
+import java.io.Closeable;
 import java.util.Objects;
 import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nullable;
@@ -32,7 +33,7 @@
  * {@link Provider#isClientFilter()}, {@link Provider#isServerFilter()} to indicate that the filter
  * is capable of working on the client side or server side or both, respectively.
  */
-interface Filter {
+interface Filter extends Closeable {
 
   /** Represents an opaque data structure holding configuration for a filter. */
   interface FilterConfig {
@@ -72,6 +73,19 @@ default boolean isServerFilter() {
      *
      * <p>Returns a filter instance registered with the same typeUrls as the provider,
      * capable of working with the same FilterConfig type returned by provider's parse functions.
+     *
+     * <p>For xDS gRPC clients, new filter instances are created per combination of:
+     * <ol>
+     *   <li><code>XdsNameResolver</code> instance,</li>
+     *   <li>Filter name+typeUrl in HttpConnectionManager (HCM) http_filters.</li>
+     * </ol>
+     *
+     * <p>For xDS-enabled gRPC servers, new filter instances are created per combination of:
+     * <ol>
+     *   <li>Server instance,</li>
+     *   <li>FilterChain name,</li>
+     *   <li>Filter name+typeUrl in FilterChain's HCM.http_filters.</li>
+     * </ol>
      */
     Filter newInstance();
 
@@ -103,6 +117,14 @@ default ServerInterceptor buildServerInterceptor(
     return null;
   }
 
+  /**
+   * Releases filter resources like shared resources and remote connections.
+   *
+   * <p>See {@link Provider#newInstance()} for details on filter instance creation.
+   */
+  @Override
+  default void close() {}
+
   /** Filter config with instance name. */
   final class NamedFilterConfig {
     // filter instance name
@@ -114,6 +136,10 @@ final class NamedFilterConfig {
       this.filterConfig = filterConfig;
     }
 
+    String filterStateKey() {
+      return name + "_" + filterConfig.typeUrl();
+    }
+
     @Override
     public boolean equals(Object o) {
       if (this == o) {

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

@@ -151,6 +151,10 @@ static final class FilterChainSelector {
         this.defaultRoutingConfig = checkNotNull(defaultRoutingConfig, "defaultRoutingConfig");
       }
 
+      FilterChainSelector(Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs) {
+        this(routingConfigs, null, new AtomicReference<>());
+      }
+
       @VisibleForTesting
       Map<FilterChain, AtomicReference<ServerRoutingConfig>> getRoutingConfigs() {
         return routingConfigs;

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

@@ -25,6 +25,7 @@
 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import com.google.common.net.InetAddresses;
 import com.google.protobuf.Any;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
@@ -43,7 +44,6 @@
 import io.grpc.xds.Filter.FilterConfig;
 import io.grpc.xds.XdsListenerResource.LdsUpdate;
 import io.grpc.xds.client.XdsResourceType;
-import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
@@ -178,6 +178,7 @@ static EnvoyServerProtoData.Listener parseServerSideListener(
     }
 
     ImmutableList.Builder<FilterChain> filterChains = ImmutableList.builder();
+    Set<String> filterChainNames = new HashSet<>();
     Set<FilterChainMatch> filterChainMatchSet = new HashSet<>();
     int i = 0;
     for (io.envoyproxy.envoy.config.listener.v3.FilterChain fc : proto.getFilterChainsList()) {
@@ -187,6 +188,10 @@ static EnvoyServerProtoData.Listener parseServerSideListener(
         // Generate a name, so we can identify it in the logs.
         filterChainName = "chain_" + i;
       }
+      if (!filterChainNames.add(filterChainName)) {
+        throw new ResourceInvalidException("Filter chain names must be unique. "
+            + "Found duplicate: " + filterChainName);
+      }
       filterChains.add(
           parseFilterChain(fc, filterChainName, tlsContextManager, filterRegistry,
               filterChainMatchSet, certProviderInstances, args));
@@ -446,16 +451,18 @@ private static FilterChainMatch parseFilterChainMatch(
     try {
       for (io.envoyproxy.envoy.config.core.v3.CidrRange range : proto.getPrefixRangesList()) {
         prefixRanges.add(
-            CidrRange.create(range.getAddressPrefix(), range.getPrefixLen().getValue()));
+            CidrRange.create(InetAddresses.forString(range.getAddressPrefix()),
+                range.getPrefixLen().getValue()));
       }
       for (io.envoyproxy.envoy.config.core.v3.CidrRange range
           : proto.getSourcePrefixRangesList()) {
-        sourcePrefixRanges.add(
-            CidrRange.create(range.getAddressPrefix(), range.getPrefixLen().getValue()));
+        sourcePrefixRanges.add(CidrRange.create(
+            InetAddresses.forString(range.getAddressPrefix()), range.getPrefixLen().getValue()));
       }
-    } catch (UnknownHostException e) {
-      throw new ResourceInvalidException("Failed to create CidrRange", e);
+    } catch (IllegalArgumentException ex) {
+      throw new ResourceInvalidException("Failed to create CidrRange", ex);
     }
+
     ConnectionSourceType sourceType;
     switch (proto.getSourceType()) {
       case ANY:

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

@@ -127,6 +127,10 @@ final class XdsNameResolver extends NameResolver {
   private final ConfigSelector configSelector = new ConfigSelector();
   private final long randomChannelId;
   private final MetricRecorder metricRecorder;
+  // Must be accessed in syncContext.
+  // Filter instances are unique per channel, and per filter (name+typeUrl).
+  // NamedFilterConfig.filterStateKey -> filter_instance.
+  private final HashMap<String, Filter> activeFilters = new HashMap<>();
 
   private volatile RoutingConfig routingConfig = RoutingConfig.EMPTY;
   private Listener2 listener;
@@ -658,18 +662,23 @@ public void onChanged(final XdsListenerResource.LdsUpdate update) {
       HttpConnectionManager httpConnectionManager = update.httpConnectionManager();
       List<VirtualHost> virtualHosts = httpConnectionManager.virtualHosts();
       String rdsName = httpConnectionManager.rdsName();
+      ImmutableList<NamedFilterConfig> filterConfigs = httpConnectionManager.httpFilterConfigs();
+      long streamDurationNano = httpConnectionManager.httpMaxStreamDurationNano();
+
+      // Create/update HCM-bound state.
       cleanUpRouteDiscoveryState();
+      updateActiveFilters(filterConfigs);
+
+      // Routes specified directly in LDS.
       if (virtualHosts != null) {
-        updateRoutes(virtualHosts, httpConnectionManager.httpMaxStreamDurationNano(),
-            httpConnectionManager.httpFilterConfigs());
-      } else {
-        routeDiscoveryState = new RouteDiscoveryState(
-            rdsName, httpConnectionManager.httpMaxStreamDurationNano(),
-            httpConnectionManager.httpFilterConfigs());
-        logger.log(XdsLogLevel.INFO, "Start watching RDS resource {0}", rdsName);
-        xdsClient.watchXdsResource(XdsRouteConfigureResource.getInstance(),
-            rdsName, routeDiscoveryState, syncContext);
+        updateRoutes(virtualHosts, streamDurationNano, filterConfigs);
+        return;
       }
+      // Routes provided by RDS.
+      routeDiscoveryState = new RouteDiscoveryState(rdsName, streamDurationNano, filterConfigs);
+      logger.log(XdsLogLevel.INFO, "Start watching RDS resource {0}", rdsName);
+      xdsClient.watchXdsResource(XdsRouteConfigureResource.getInstance(),
+          rdsName, routeDiscoveryState, syncContext);
     }
 
     @Override
@@ -690,6 +699,7 @@ public void onResourceDoesNotExist(final String resourceName) {
       String error = "LDS resource does not exist: " + resourceName;
       logger.log(XdsLogLevel.INFO, error);
       cleanUpRouteDiscoveryState();
+      updateActiveFilters(null);
       cleanUpRoutes(error);
     }
 
@@ -703,9 +713,35 @@ private void stop() {
       logger.log(XdsLogLevel.INFO, "Stop watching LDS resource {0}", ldsResourceName);
       stopped = true;
       cleanUpRouteDiscoveryState();
+      updateActiveFilters(null);
       xdsClient.cancelXdsResourceWatch(XdsListenerResource.getInstance(), ldsResourceName, this);
     }
 
+    // called in syncContext
+    private void updateActiveFilters(@Nullable List<NamedFilterConfig> filterConfigs) {
+      if (filterConfigs == null) {
+        filterConfigs = ImmutableList.of();
+      }
+      Set<String> filtersToShutdown = new HashSet<>(activeFilters.keySet());
+      for (NamedFilterConfig namedFilter : filterConfigs) {
+        String typeUrl = namedFilter.filterConfig.typeUrl();
+        String filterKey = namedFilter.filterStateKey();
+
+        Filter.Provider provider = filterRegistry.get(typeUrl);
+        checkNotNull(provider, "provider %s", typeUrl);
+        Filter filter = activeFilters.computeIfAbsent(filterKey, k -> provider.newInstance());
+        checkNotNull(filter, "filter %s", filterKey);
+        filtersToShutdown.remove(filterKey);
+      }
+
+      // Shutdown filters not present in current HCM.
+      for (String filterKey : filtersToShutdown) {
+        Filter filterToShutdown = activeFilters.remove(filterKey);
+        checkNotNull(filterToShutdown, "filterToShutdown %s", filterKey);
+        filterToShutdown.close();
+      }
+    }
+
     // called in syncContext
     private void updateRoutes(List<VirtualHost> virtualHosts, long httpMaxStreamDurationNano,
         @Nullable List<NamedFilterConfig> filterConfigs) {
@@ -836,19 +872,16 @@ private ClientInterceptor createFilters(
 
       ImmutableList.Builder<ClientInterceptor> filterInterceptors = ImmutableList.builder();
       for (NamedFilterConfig namedFilter : filterConfigs) {
-        FilterConfig config = namedFilter.filterConfig;
         String name = namedFilter.name;
-        String typeUrl = config.typeUrl();
-
-        Filter.Provider provider = filterRegistry.get(typeUrl);
-        if (provider == null || !provider.isClientFilter()) {
-          continue;
-        }
-
-        Filter filter = provider.newInstance();
+        FilterConfig config = namedFilter.filterConfig;
+        FilterConfig overrideConfig = selectedOverrideConfigs.get(name);
+        String filterKey = namedFilter.filterStateKey();
 
+        Filter filter = activeFilters.get(filterKey);
+        checkNotNull(filter, "activeFilters.get(%s)", filterKey);
         ClientInterceptor interceptor =
-            filter.buildClientInterceptor(config, selectedOverrideConfigs.get(name), scheduler);
+            filter.buildClientInterceptor(config, overrideConfig, scheduler);
+
         if (interceptor != null) {
           filterInterceptors.add(interceptor);
         }