xds: Add composite filter

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

@@ -106,15 +106,24 @@ public FaultFilter newInstance(String name) {
     @Override
     public ConfigOrError<FaultConfig> parseFilterConfig(Message rawProtoMessage) {
       HTTPFault httpFaultProto;
-      if (!(rawProtoMessage instanceof Any)) {
+      if (rawProtoMessage instanceof Any) {
+        try {
+          httpFaultProto = ((Any) rawProtoMessage).unpack(HTTPFault.class);
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Invalid proto: " + e);
+        }
+      } else if (rawProtoMessage instanceof com.google.protobuf.Struct) {
+        try {
+          HTTPFault.Builder builder = HTTPFault.newBuilder();
+          com.google.protobuf.util.JsonFormat.parser().merge(
+              com.google.protobuf.util.JsonFormat.printer().print(rawProtoMessage), builder);
+          httpFaultProto = builder.build();
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Failed to parse Struct to HTTPFault: " + e);
+        }
+      } else {
         return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
       }
-      Any anyMessage = (Any) rawProtoMessage;
-      try {
-        httpFaultProto = anyMessage.unpack(HTTPFault.class);
-      } catch (InvalidProtocolBufferException e) {
-        return ConfigOrError.fromError("Invalid proto: " + e);
-      }
       return parseHttpFault(httpFaultProto);
     }
 

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

@@ -35,10 +35,11 @@ private FilterRegistry() {}
   static synchronized FilterRegistry getDefaultRegistry() {
     if (instance == null) {
       instance = newRegistry().register(
-              new FaultFilter.Provider(),
-              new RouterFilter.Provider(),
-              new RbacFilter.Provider(),
-              new GcpAuthenticationFilter.Provider());
+          new FaultFilter.Provider(),
+          new RouterFilter.Provider(),
+          new RbacFilter.Provider(),
+          new GcpAuthenticationFilter.Provider(),
+          new CompositeFilter.Provider());
     }
     return instance;
   }
@@ -58,6 +59,13 @@ FilterRegistry register(Filter.Provider... filters) {
     return this;
   }
 
+  @VisibleForTesting
+  void deregister(Filter.Provider provider) {
+    for (String typeUrl : provider.typeUrls()) {
+      supportedFilters.remove(typeUrl);
+    }
+  }
+
   @Nullable
   Filter.Provider get(String typeUrl) {
     return supportedFilters.get(typeUrl);

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

@@ -88,16 +88,24 @@ public GcpAuthenticationFilter newInstance(String name) {
     @Override
     public ConfigOrError<GcpAuthenticationConfig> parseFilterConfig(Message rawProtoMessage) {
       GcpAuthnFilterConfig gcpAuthnProto;
-      if (!(rawProtoMessage instanceof Any)) {
+      if (rawProtoMessage instanceof Any) {
+        try {
+          gcpAuthnProto = ((Any) rawProtoMessage).unpack(GcpAuthnFilterConfig.class);
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Invalid proto: " + e);
+        }
+      } else if (rawProtoMessage instanceof com.google.protobuf.Struct) {
+        try {
+          GcpAuthnFilterConfig.Builder builder = GcpAuthnFilterConfig.newBuilder();
+          com.google.protobuf.util.JsonFormat.parser().merge(
+              com.google.protobuf.util.JsonFormat.printer().print(rawProtoMessage), builder);
+          gcpAuthnProto = builder.build();
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Failed to parse Struct to GcpAuthnFilterConfig: " + e);
+        }
+      } else {
         return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
       }
-      Any anyMessage = (Any) rawProtoMessage;
-
-      try {
-        gcpAuthnProto = anyMessage.unpack(GcpAuthnFilterConfig.class);
-      } catch (InvalidProtocolBufferException e) {
-        return ConfigOrError.fromError("Invalid proto: " + e);
-      }
 
       long cacheSize = 10;
       // Validate cache_config

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

@@ -96,30 +96,48 @@ public RbacFilter newInstance(String name) {
     @Override
     public ConfigOrError<RbacConfig> parseFilterConfig(Message rawProtoMessage) {
       RBAC rbacProto;
-      if (!(rawProtoMessage instanceof Any)) {
+      if (rawProtoMessage instanceof Any) {
+        try {
+          rbacProto = ((Any) rawProtoMessage).unpack(RBAC.class);
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Invalid proto: " + e);
+        }
+      } else if (rawProtoMessage instanceof com.google.protobuf.Struct) {
+        try {
+          RBAC.Builder builder = RBAC.newBuilder();
+          com.google.protobuf.util.JsonFormat.parser().merge(
+              com.google.protobuf.util.JsonFormat.printer().print(rawProtoMessage), builder);
+          rbacProto = builder.build();
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Failed to parse Struct to RBAC: " + e);
+        }
+      } else {
         return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
       }
-      Any anyMessage = (Any) rawProtoMessage;
-      try {
-        rbacProto = anyMessage.unpack(RBAC.class);
-      } catch (InvalidProtocolBufferException e) {
-        return ConfigOrError.fromError("Invalid proto: " + e);
-      }
       return parseRbacConfig(rbacProto);
     }
 
     @Override
     public ConfigOrError<RbacConfig> parseFilterConfigOverride(Message rawProtoMessage) {
       RBACPerRoute rbacPerRoute;
-      if (!(rawProtoMessage instanceof Any)) {
+      if (rawProtoMessage instanceof Any) {
+        try {
+          rbacPerRoute = ((Any) rawProtoMessage).unpack(RBACPerRoute.class);
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Invalid proto: " + e);
+        }
+      } else if (rawProtoMessage instanceof com.google.protobuf.Struct) {
+        try {
+          RBACPerRoute.Builder builder = RBACPerRoute.newBuilder();
+          com.google.protobuf.util.JsonFormat.parser().merge(
+              com.google.protobuf.util.JsonFormat.printer().print(rawProtoMessage), builder);
+          rbacPerRoute = builder.build();
+        } catch (InvalidProtocolBufferException e) {
+          return ConfigOrError.fromError("Failed to parse Struct to RBACPerRoute: " + e);
+        }
+      } else {
         return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
       }
-      Any anyMessage = (Any) rawProtoMessage;
-      try {
-        rbacPerRoute = anyMessage.unpack(RBACPerRoute.class);
-      } catch (InvalidProtocolBufferException e) {
-        return ConfigOrError.fromError("Invalid proto: " + e);
-      }
       if (rbacPerRoute.hasRbac()) {
         return parseRbacConfig(rbacPerRoute.getRbac());
       } else {

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

@@ -65,21 +65,38 @@ abstract static class Route {
 
     abstract ImmutableMap<String, FilterConfig> filterConfigOverrides();
 
+    @Nullable
+    abstract ImmutableMap<String, com.google.protobuf.Struct> filterMetadata();
+
     static Route forAction(RouteMatch routeMatch, RouteAction routeAction,
         Map<String, FilterConfig> filterConfigOverrides) {
-      return create(routeMatch, routeAction, filterConfigOverrides);
+      return create(routeMatch, routeAction, filterConfigOverrides, null);
+    }
+
+    static Route forAction(RouteMatch routeMatch, RouteAction routeAction,
+        Map<String, FilterConfig> filterConfigOverrides,
+        @Nullable Map<String, com.google.protobuf.Struct> filterMetadata) {
+      return create(routeMatch, routeAction, filterConfigOverrides, filterMetadata);
     }
 
     static Route forNonForwardingAction(RouteMatch routeMatch,
         Map<String, FilterConfig> filterConfigOverrides) {
-      return create(routeMatch, null, filterConfigOverrides);
+      return create(routeMatch, null, filterConfigOverrides, null);
+    }
+
+    static Route forNonForwardingAction(RouteMatch routeMatch,
+        Map<String, FilterConfig> filterConfigOverrides,
+        @Nullable Map<String, com.google.protobuf.Struct> filterMetadata) {
+      return create(routeMatch, null, filterConfigOverrides, filterMetadata);
     }
 
     private static Route create(
         RouteMatch routeMatch, @Nullable RouteAction routeAction,
-        Map<String, FilterConfig> filterConfigOverrides) {
+        Map<String, FilterConfig> filterConfigOverrides,
+        @Nullable Map<String, com.google.protobuf.Struct> filterMetadata) {
       return new AutoValue_VirtualHost_Route(
-          routeMatch, routeAction, ImmutableMap.copyOf(filterConfigOverrides));
+          routeMatch, routeAction, ImmutableMap.copyOf(filterConfigOverrides),
+          filterMetadata == null ? null : ImmutableMap.copyOf(filterMetadata));
     }
 
     @AutoValue

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

@@ -16,13 +16,15 @@
 
 package io.grpc.xds;
 
+import com.google.protobuf.Struct;
 import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.Grpc;
 import io.grpc.NameResolver;
 import io.grpc.xds.XdsNameResolverProvider.CallCounterProvider;
 import io.grpc.xds.client.Locality;
 import io.grpc.xds.client.XdsClient;
+import java.util.Map;
 
 /**
  * Attributes used for xDS implementation.
@@ -100,5 +102,18 @@ final class XdsAttributes {
   static final Attributes.Key<Long> ATTR_DRAIN_GRACE_NANOS =
       Attributes.Key.create("io.grpc.xds.XdsAttributes.drainGraceTime");
 
+  /**
+   * Attribute key for xDS route metadata used in filter matching.
+   */
+  @NameResolver.ResolutionResultAttr
+  public static final Attributes.Key<Map<String, Struct>> ATTR_FILTER_METADATA = Attributes.Key
+          .create("io.grpc.xds.XdsAttributes.filterMetadata");
+
+  /**
+   * CallOptions key for xDS route metadata used in filter matching.
+   */
+  public static final io.grpc.CallOptions.Key<Map<String, Struct>> CALL_OPTIONS_FILTER_METADATA =
+      io.grpc.CallOptions.Key.create("io.grpc.xds.XdsAttributes.filterMetadata");
+
   private XdsAttributes() {}
 }

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

@@ -230,7 +230,7 @@ static FilterChain parseFilterChain(
     // FilterChain contains L4 filters, so we ensure it contains only HCM.
     if (proto.getFiltersCount() != 1) {
       throw new ResourceInvalidException("FilterChain " + filterChainName
-          + " should contain exact one HttpConnectionManager filter");
+          + " should contain exactly one HttpConnectionManager filter");
     }
     io.envoyproxy.envoy.config.listener.v3.Filter l4Filter = proto.getFiltersList().get(0);
     if (!l4Filter.hasTypedConfig()) {
@@ -513,6 +513,7 @@ static io.grpc.xds.HttpConnectionManager parseHttpConnectionManager(
     }
 
     // Parse http filters.
+    // todo: AgraVator are we changing the assumption here ??
     if (proto.getHttpFiltersList().isEmpty()) {
       throw new ResourceInvalidException("Missing HttpFilter in HttpConnectionManager.");
     }
@@ -582,6 +583,7 @@ private static boolean isTerminalFilter(Filter.FilterConfig filterConfig) {
   static StructOrError<Filter.FilterConfig> parseHttpFilter(
       io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter
           httpFilter, FilterRegistry filterRegistry, boolean isForClient) {
+    // todo: AgraVator do we need to change anything here for composite filter ??
     String filterName = httpFilter.getName();
     boolean isOptional = httpFilter.getIsOptional();
     if (!httpFilter.hasTypedConfig()) {

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

@@ -336,6 +336,7 @@ private void updateResolutionResult(XdsConfig xdsConfig) {
             .setAttributes(attrs)
             .setServiceConfig(parsedServiceConfig)
             .build();
+    // todo: abhishek probably the filters are getting applied here ??
     if (!listener.onResult2(result).isOk()) {
       resolveState.xdsDependencyManager.requestReresolution();
     }
@@ -404,6 +405,7 @@ static boolean matchHostName(String hostName, String pattern) {
   private final class ConfigSelector extends InternalConfigSelector {
     @Override
     public Result selectConfig(PickSubchannelArgs args) {
+      // todo: AgraVator probably the filters are getting populated here
       RoutingConfig routingCfg;
       RouteData selectedRoute;
       String cluster;
@@ -716,6 +718,7 @@ public void onUpdate(StatusOr<XdsConfig> updateOrStatus) {
       }
 
       VirtualHost virtualHost = update.getVirtualHost();
+      // filters and there configurations
       ImmutableList<NamedFilterConfig> filterConfigs = httpConnectionManager.httpFilterConfigs();
       long streamDurationNano = httpConnectionManager.httpMaxStreamDurationNano();
 

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

@@ -29,6 +29,7 @@
 import com.google.protobuf.Duration;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
+import com.google.protobuf.Struct;
 import com.google.protobuf.util.Durations;
 import com.google.re2j.Pattern;
 import com.google.re2j.PatternSyntaxException;
@@ -289,6 +290,11 @@ static StructOrError<Route> parseRoute(
     }
     Map<String, FilterConfig> overrideConfigs = overrideConfigsOrError.getStruct();
 
+    Map<String, Struct> filterMetadata = null;
+    if (proto.hasMetadata()) {
+      filterMetadata = proto.getMetadata().getFilterMetadataMap();
+    }
+
     switch (proto.getActionCase()) {
       case ROUTE:
         StructOrError<RouteAction> routeAction =
@@ -303,10 +309,11 @@ static StructOrError<Route> parseRoute(
                   + routeAction.getErrorDetail());
         }
         return StructOrError.fromStruct(
-            Route.forAction(routeMatch.getStruct(), routeAction.getStruct(), overrideConfigs));
+            Route.forAction(routeMatch.getStruct(), routeAction.getStruct(), overrideConfigs,
+                filterMetadata));
       case NON_FORWARDING_ACTION:
         return StructOrError.fromStruct(
-            Route.forNonForwardingAction(routeMatch.getStruct(), overrideConfigs));
+            Route.forNonForwardingAction(routeMatch.getStruct(), overrideConfigs, filterMetadata));
       case REDIRECT:
       case DIRECT_RESPONSE:
       case FILTER_ACTION:

xds/src/main/java/io/grpc/xds/internal/MatcherParser.java

@@ -97,4 +97,26 @@ public static Matchers.StringMatcher parseStringMatcher(
                 "Unknown StringMatcher match pattern: " + proto.getMatchPatternCase());
     }
   }
+
+  /** Translate StringMatcher xDS proto to internal StringMatcher. */
+  public static Matchers.StringMatcher parseStringMatcher(
+      com.github.xds.type.matcher.v3.StringMatcher proto) {
+    switch (proto.getMatchPatternCase()) {
+      case EXACT:
+        return Matchers.StringMatcher.forExact(proto.getExact(), proto.getIgnoreCase());
+      case PREFIX:
+        return Matchers.StringMatcher.forPrefix(proto.getPrefix(), proto.getIgnoreCase());
+      case SUFFIX:
+        return Matchers.StringMatcher.forSuffix(proto.getSuffix(), proto.getIgnoreCase());
+      case SAFE_REGEX:
+        return Matchers.StringMatcher.forSafeRegEx(
+            Pattern.compile(proto.getSafeRegex().getRegex()));
+      case CONTAINS:
+        return Matchers.StringMatcher.forContains(proto.getContains());
+      case MATCHPATTERN_NOT_SET:
+      default:
+        throw new IllegalArgumentException(
+            "Unknown StringMatcher match pattern: " + proto.getMatchPatternCase());
+    }
+  }
 }