Refactor ExternalProcessorFilter to use new FilterContext API.

Merged changes from sauravzg-feat-bootstrap-filter-context branch and updated ExternalProcessorFilter and its test to work with the new Filter.Provider.parseFilterConfig API which now takes a FilterContext containing BootstrapInfo and ServerInfo.

Author: kannanjgithub

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

@@ -20,7 +20,6 @@
 import io.grpc.ClientInterceptor;
 import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
 import io.grpc.ForwardingClientCallListener;
-import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
@@ -30,12 +29,10 @@
 import io.grpc.xds.internal.grpcservice.GrpcServiceConfig;
 import io.grpc.xds.internal.grpcservice.GrpcServiceConfigParser;
 import io.grpc.xds.internal.grpcservice.GrpcServiceParseException;
-import io.grpc.xds.internal.grpcservice.GrpcServiceXdsContextProvider;
 import io.grpc.xds.internal.grpcservice.HeaderValue;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
@@ -53,7 +50,6 @@ public ExternalProcessorFilter(String name) {
   }
 
   static final class Provider implements Filter.Provider {
-    private GrpcServiceXdsContextProvider grpcServiceXdsContextProvider;
     @Override
     public String[] typeUrls() {
       return new String[]{TYPE_URL};
@@ -65,13 +61,13 @@ public boolean isClientFilter() {
     }
 
     @Override
-    public ExternalProcessorFilter newInstance(String name, GrpcServiceXdsContextProvider grpcServiceXdsContextProvider) {
-      this.grpcServiceXdsContextProvider = grpcServiceXdsContextProvider;
+    public ExternalProcessorFilter newInstance(String name) {
       return new ExternalProcessorFilter(name);
     }
 
     @Override
-    public ConfigOrError<ExternalProcessorFilterConfig> parseFilterConfig(Message rawProtoMessage) {
+    public ConfigOrError<ExternalProcessorFilterConfig> parseFilterConfig(
+        Message rawProtoMessage, FilterContext context) {
       if (!(rawProtoMessage instanceof Any)) {
         return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
       }
@@ -91,16 +87,18 @@ public ConfigOrError<ExternalProcessorFilterConfig> parseFilterConfig(Message ra
       }
 
       try {
-        GrpcServiceConfig grpcServiceConfig = GrpcServiceConfigParser.parse(externalProcessor.getGrpcService(), grpcServiceXdsContextProvider);
+        GrpcServiceConfig grpcServiceConfig = GrpcServiceConfigParser.parse(
+            externalProcessor.getGrpcService(), context.bootstrapInfo(), context.serverInfo());
         return ConfigOrError.fromConfig(new ExternalProcessorFilterConfig(externalProcessor, grpcServiceConfig));
       } catch (GrpcServiceParseException e) {
         return ConfigOrError.fromError("Error parsing GrpcService config: " + e.getMessage());
       }
     }
 
     @Override
-    public ConfigOrError<? extends FilterConfig> parseFilterConfigOverride(Message rawProtoMessage) {
-      return parseFilterConfig(rawProtoMessage);
+    public ConfigOrError<? extends FilterConfig> parseFilterConfigOverride(
+        Message rawProtoMessage, FilterContext context) {
+      return parseFilterConfig(rawProtoMessage, context);
     }
   }
 
@@ -487,11 +485,7 @@ else if (response.hasResponseBody()) {
                   extProcClientCallRequestObserver.onCompleted();
                 }
               }
-            }
-            // For robustness. For any internal processing failure make sure the internal state
-            // machine is notified and the dataplane call is properly cancelled (or failed-open if
-            // configured)
-            catch (Throwable t) {
+            } catch (Throwable t) {
               onError(t);
             }
           }

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

@@ -46,7 +46,6 @@
 import io.grpc.xds.FaultConfig.FaultAbort;
 import io.grpc.xds.FaultConfig.FaultDelay;
 import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
-import io.grpc.xds.internal.grpcservice.GrpcServiceXdsContextProvider;
 import java.util.Locale;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
@@ -100,7 +99,7 @@ public boolean isClientFilter() {
     }
 
     @Override
-    public FaultFilter newInstance(String name, GrpcServiceXdsContextProvider grpcServiceXdsContextProvider) {
+    public FaultFilter newInstance(String name) {
       return INSTANCE;
     }
 

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

@@ -45,7 +45,6 @@
 import io.grpc.xds.MetadataRegistry.MetadataValueParser;
 import io.grpc.xds.XdsConfig.XdsClusterConfig;
 import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
-import io.grpc.xds.internal.grpcservice.GrpcServiceXdsContextProvider;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
@@ -82,7 +81,7 @@ public boolean isClientFilter() {
     }
 
     @Override
-    public GcpAuthenticationFilter newInstance(String name, GrpcServiceXdsContextProvider grpcServiceXdsContextProvider) {
+    public GcpAuthenticationFilter newInstance(String name) {
       return new GcpAuthenticationFilter(name, cacheSize);
     }
 

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

@@ -33,7 +33,7 @@ public static ServerInterceptor createInterceptor(RBAC rbac) {
       throw new IllegalArgumentException(
         String.format("Failed to parse Rbac policy: %s", filterConfig.errorDetail));
     }
-    return new RbacFilter.Provider().newInstance("internalRbacFilter", null)
+    return new RbacFilter.Provider().newInstance("internalRbacFilter")
         .buildServerInterceptor(filterConfig.config, null);
   }
 }

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

@@ -35,7 +35,6 @@
 import io.grpc.Status;
 import io.grpc.xds.internal.MatcherParser;
 import io.grpc.xds.internal.Matchers;
-import io.grpc.xds.internal.grpcservice.GrpcServiceXdsContextProvider;
 import io.grpc.xds.internal.rbac.engine.GrpcAuthorizationEngine;
 import io.grpc.xds.internal.rbac.engine.GrpcAuthorizationEngine.AlwaysTrueMatcher;
 import io.grpc.xds.internal.rbac.engine.GrpcAuthorizationEngine.AndMatcher;
@@ -90,7 +89,7 @@ public boolean isServerFilter() {
     }
 
     @Override
-    public RbacFilter newInstance(String name, GrpcServiceXdsContextProvider grpcServiceXdsContextProvider) {
+    public RbacFilter newInstance(String name) {
       return INSTANCE;
     }
 

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

@@ -17,7 +17,6 @@
 package io.grpc.xds;
 
 import com.google.protobuf.Message;
-import io.grpc.xds.internal.grpcservice.GrpcServiceXdsContextProvider;
 
 /**
  * Router filter implementation. Currently this filter does not parse any field in the config.
@@ -57,7 +56,7 @@ public boolean isServerFilter() {
     }
 
     @Override
-    public RouterFilter newInstance(String name, GrpcServiceXdsContextProvider grpcServiceXdsContextProvider) {
+    public RouterFilter newInstance(String name) {
       return INSTANCE;
     }
 

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

@@ -67,6 +67,7 @@
 import io.grpc.xds.client.XdsInitializationException;
 import io.grpc.xds.client.XdsLogger;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -109,6 +110,7 @@ final class XdsNameResolver extends NameResolver {
   private final XdsLogger logger;
   @Nullable
   private final String targetAuthority;
+  private final String target;
   private final String serviceAuthority;
   // Encoded version of the service authority as per 
   // https://datatracker.ietf.org/doc/html/rfc3986#section-3.2.
@@ -139,12 +141,12 @@ final class XdsNameResolver extends NameResolver {
   private ResolveState resolveState;
 
   XdsNameResolver(
-      String target, @Nullable String targetAuthority, String name,
-      @Nullable String overrideAuthority, ServiceConfigParser serviceConfigParser,
+      URI targetUri, String name, @Nullable String overrideAuthority,
+      ServiceConfigParser serviceConfigParser,
       SynchronizationContext syncContext, ScheduledExecutorService scheduler,
       @Nullable Map<String, ?> bootstrapOverride,
       MetricRecorder metricRecorder, Args nameResolverArgs) {
-    this(target, targetAuthority, name, overrideAuthority, serviceConfigParser,
+    this(targetUri, targetUri.getAuthority(), name, overrideAuthority, serviceConfigParser,
         syncContext, scheduler,
         bootstrapOverride == null
           ? SharedXdsClientPoolProvider.getDefaultProvider()
@@ -155,13 +157,14 @@ final class XdsNameResolver extends NameResolver {
 
   @VisibleForTesting
   XdsNameResolver(
-      String target, @Nullable String targetAuthority, String name,
+      URI targetUri, @Nullable String targetAuthority, String name,
       @Nullable String overrideAuthority, ServiceConfigParser serviceConfigParser,
       SynchronizationContext syncContext, ScheduledExecutorService scheduler,
       XdsClientPoolFactory xdsClientPoolFactory, ThreadSafeRandom random,
       FilterRegistry filterRegistry, @Nullable Map<String, ?> bootstrapOverride,
       MetricRecorder metricRecorder, Args nameResolverArgs) {
     this.targetAuthority = targetAuthority;
+    target = targetUri.toString();
 
     // The name might have multiple slashes so encode it before verifying.
     serviceAuthority = checkNotNull(name, "name");
@@ -733,7 +736,7 @@ private void updateActiveFilters(@Nullable List<NamedFilterConfig> filterConfigs
         Filter.Provider provider = filterRegistry.get(typeUrl);
         checkNotNull(provider, "provider %s", typeUrl);
         Filter filter = activeFilters.computeIfAbsent(
-            filterKey, k -> provider.newInstance(namedFilter.name, null));
+            filterKey, k -> provider.newInstance(namedFilter.name));
         checkNotNull(filter, "filter %s", filterKey);
         filtersToShutdown.remove(filterKey);
       }
@@ -875,7 +878,6 @@ private ClientInterceptor createFilters(
       }
 
       ImmutableList.Builder<ClientInterceptor> filterInterceptors = ImmutableList.builder();
-      ClientInterceptor extProcInterceptor = null;
       for (NamedFilterConfig namedFilter : filterConfigs) {
         String name = namedFilter.name;
         FilterConfig config = namedFilter.filterConfig;
@@ -888,18 +890,10 @@ private ClientInterceptor createFilters(
             filter.buildClientInterceptor(config, overrideConfig, scheduler);
 
         if (interceptor != null) {
-          if (config.typeUrl().equals(ExternalProcessorFilter.TYPE_URL)) {
-            extProcInterceptor = interceptor;
-          } else {
-            filterInterceptors.add(interceptor);
-          }
+          filterInterceptors.add(interceptor);
         }
       }
 
-      if (extProcInterceptor != null) {
-        filterInterceptors.add(extProcInterceptor);
-      }
-
       // Combine interceptors produced by different filters into a single one that executes
       // them sequentially. The order is preserved.
       return combineInterceptors(filterInterceptors.build());

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

@@ -22,7 +22,6 @@
 import io.grpc.Internal;
 import io.grpc.NameResolver.Args;
 import io.grpc.NameResolverProvider;
-import io.grpc.Uri;
 import io.grpc.xds.client.XdsClient;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -87,39 +86,16 @@ public XdsNameResolver newNameResolver(URI targetUri, Args args) {
           targetPath,
           targetUri);
       String name = targetPath.substring(1);
-      return newNameResolver(targetUri.toString(), targetUri.getAuthority(), name, args);
+      return new XdsNameResolver(
+          targetUri, name, args.getOverrideAuthority(),
+          args.getServiceConfigParser(), args.getSynchronizationContext(),
+          args.getScheduledExecutorService(),
+          bootstrapOverride,
+          args.getMetricRecorder(), args);
     }
     return null;
   }
 
-  @Override
-  public XdsNameResolver newNameResolver(Uri targetUri, Args args) {
-    if (scheme.equals(targetUri.getScheme())) {
-      Preconditions.checkArgument(
-          targetUri.isPathAbsolute(),
-          "the path component of the target (%s) must start with '/'",
-          targetUri);
-      return newNameResolver(
-          targetUri.toString(), targetUri.getAuthority(), targetUri.getPath().substring(1), args);
-    }
-    return null;
-  }
-
-  private XdsNameResolver newNameResolver(
-      String targetUri, String targetAuthority, String name, Args args) {
-    return new XdsNameResolver(
-        targetUri.toString(),
-        targetAuthority,
-        name,
-        args.getOverrideAuthority(),
-        args.getServiceConfigParser(),
-        args.getSynchronizationContext(),
-        args.getScheduledExecutorService(),
-        bootstrapOverride,
-        args.getMetricRecorder(),
-        args);
-  }
-
   @Override
   public String getDefaultScheme() {
     return scheme;

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

@@ -612,7 +612,7 @@ private void updateActiveFiltersForChain(
         Filter.Provider provider = filterRegistry.get(typeUrl);
         checkNotNull(provider, "provider %s", typeUrl);
         Filter filter = chainFilters.computeIfAbsent(
-            filterKey, k -> provider.newInstance(namedFilter.name, null));
+            filterKey, k -> provider.newInstance(namedFilter.name));
         checkNotNull(filter, "filter %s", filterKey);
         filtersToShutdown.remove(filterKey);
       }