Ext_proc filter and client interceptor (grpc#12792)

Implements ext_proc filter from [grfc A93](grpc/proposal#484)  (internal [design doc](http://go/ext-proc-design-java)).

Author: kannanjgithub

xds/src/generated/thirdparty/grpc/io/envoyproxy/envoy/service/ext_proc/v3/ExternalProcessorGrpc.java

@@ -45,6 +45,8 @@
 import io.grpc.internal.GrpcUtil;
 import io.grpc.xds.FaultConfig.FaultAbort;
 import io.grpc.xds.FaultConfig.FaultDelay;
+import io.grpc.xds.Filter.FilterConfigParseContext;
+import io.grpc.xds.Filter.FilterContext;
 import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
 import java.util.Locale;
 import java.util.concurrent.Executor;
@@ -99,7 +101,7 @@ public boolean isClientFilter() {
     }
 
     @Override
-    public FaultFilter newInstance(String name) {
+    public FaultFilter newInstance(FilterContext context) {
       return INSTANCE;
     }
 

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

@@ -21,6 +21,7 @@
 import com.google.common.base.MoreObjects;
 import com.google.protobuf.Message;
 import io.grpc.ClientInterceptor;
+import io.grpc.MetricRecorder;
 import io.grpc.ServerInterceptor;
 import io.grpc.xds.client.Bootstrapper.BootstrapInfo;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
@@ -91,7 +92,7 @@ default boolean isServerFilter() {
      *   <li>Filter name+typeUrl in FilterChain's HCM.http_filters.</li>
      * </ol>
      */
-    Filter newInstance(String name);
+    Filter newInstance(FilterContext context);
 
     /**
      * Parses the top-level filter config from raw proto message. The message may be either a {@link
@@ -152,6 +153,18 @@ abstract static class Builder {
     }
   }
 
+  /** Context containing naming and metrics reporting objects for a filter instance. */
+  @AutoValue
+  abstract static class FilterContext {
+    abstract String filterName();
+
+    abstract MetricRecorder metricsRecorder();
+
+    static FilterContext create(String filterName, MetricRecorder metricsRecorder) {
+      return new AutoValue_Filter_FilterContext(filterName, metricsRecorder);
+    }
+  }
+
   /** Filter config with instance name. */
   final class NamedFilterConfig {
     // filter instance name

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

@@ -38,11 +38,17 @@ static synchronized FilterRegistry getDefaultRegistry() {
               new FaultFilter.Provider(),
               new RouterFilter.Provider(),
               new RbacFilter.Provider(),
-              new GcpAuthenticationFilter.Provider());
+              new GcpAuthenticationFilter.Provider(),
+              new ExternalProcessorFilter.Provider());
     }
     return instance;
   }
 
+  @VisibleForTesting
+  static synchronized void reset() {
+    instance = null;
+  }
+
   @VisibleForTesting
   static FilterRegistry newRegistry() {
     return new FilterRegistry();

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

@@ -41,6 +41,8 @@
 import io.grpc.Status;
 import io.grpc.StatusOr;
 import io.grpc.auth.MoreCallCredentials;
+import io.grpc.xds.Filter.FilterConfigParseContext;
+import io.grpc.xds.Filter.FilterContext;
 import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser.AudienceWrapper;
 import io.grpc.xds.MetadataRegistry.MetadataValueParser;
 import io.grpc.xds.XdsConfig.XdsClusterConfig;
@@ -81,8 +83,8 @@ public boolean isClientFilter() {
     }
 
     @Override
-    public GcpAuthenticationFilter newInstance(String name) {
-      return new GcpAuthenticationFilter(name, cacheSize);
+    public GcpAuthenticationFilter newInstance(FilterContext context) {
+      return new GcpAuthenticationFilter(context.filterName(), cacheSize);
     }
 
     @Override

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

@@ -98,10 +98,14 @@ public static GrpcServiceConfig parse(GrpcService grpcServiceProto,
         .getInitialMetadataList()) {
       String key = header.getKey();
       HeaderValue headerValue;
-      if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
-        headerValue = HeaderValue.create(key, header.getRawValue());
-      } else {
-        headerValue = HeaderValue.create(key, header.getValue());
+      try {
+        if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+          headerValue = HeaderValue.create(key, header.getRawValue());
+        } else {
+          headerValue = HeaderValue.create(key, header.getValue());
+        }
+      } catch (IllegalArgumentException e) {
+        throw new GrpcServiceParseException("Invalid initial metadata header: " + key, e);
       }
       if (HeaderValueValidationUtils.isDisallowed(headerValue)) {
         throw new GrpcServiceParseException("Invalid initial metadata header: " + key);

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

@@ -19,6 +19,7 @@
 import io.envoyproxy.envoy.extensions.filters.http.rbac.v3.RBAC;
 import io.grpc.Internal;
 import io.grpc.ServerInterceptor;
+import io.grpc.xds.Filter.FilterContext;
 
 /** This class exposes some functionality in RbacFilter to other packages. */
 @Internal
@@ -33,7 +34,8 @@ 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")
+    return new RbacFilter.Provider().newInstance(
+        FilterContext.create("internalRbacFilter", new io.grpc.MetricRecorder() {}))
         .buildServerInterceptor(filterConfig.config, null);
   }
 }

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

@@ -33,6 +33,8 @@
 import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptor;
 import io.grpc.Status;
+import io.grpc.xds.Filter.FilterConfigParseContext;
+import io.grpc.xds.Filter.FilterContext;
 import io.grpc.xds.internal.MatcherParser;
 import io.grpc.xds.internal.Matchers;
 import io.grpc.xds.internal.rbac.engine.GrpcAuthorizationEngine;
@@ -89,7 +91,7 @@ public boolean isServerFilter() {
     }
 
     @Override
-    public RbacFilter newInstance(String name) {
+    public RbacFilter newInstance(FilterContext context) {
       return INSTANCE;
     }