xds: Support filter state retention

This PR adds support filter state retention in Java. The mechanism
will be similar to the one described in [A83]
(https://github.com/grpc/proposal/blob/master/A83-xds-gcp-authn-filter.md#filter-call-credentials-cache)
for C-core, and will serve the same purpose. However, the
implementation details are very different due to the different nature
of xDS HTTP filter support in C-core and Java.

### Filter instance lifecycle
#### xDS gRPC clients
New filter instances are created per combination of:
1. `XdsNameResolver` instance,
2. Filter name+typeUrl as configured in 
   HttpConnectionManager (HCM) http_filters.

Existing client-side filter instances are shutdown:
- A single a filter instance is shutdown when an LDS update contains
  HCM that is missing filter configuration for name+typeUrl
  combination of this instance.
- All filter instances when watched LDS resource is missing from an
  LDS update.
- All filter instances name resolver shutdown.

#### xDS-enabled gRPC servers
New filter instances are created per combination of:
1. Server instance,
2. FilterChain name,
3. Filter name+typeUrl as configured in FilterChain's HCM.http_filters

Filter instances of Default Filter Chain is tracked separately per:
1. Server instance,
2. Filter name+typeUrl in default_filter_chain's HCM.http_filters.

Existing server-side filter instances are shutdown:
- A single a filter instance is shutdown when an LDS update contains
  FilterChain with HCM.http_filters that is missing configuration for
  filter name+typeUrl.
- All filter instances associated with the FilterChain when an LDS
  update no longer contains FilterChain's name.
- All filter instances when watched LDS resource is missing from an
  LDS update.
- All filter instances on server shutdown.

### Related
- Part 1: #11883

Author: sergiitk

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

@@ -206,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

@@ -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));

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);
         }