xds: RLQS Prototype

gradle/libs.versions.toml

@@ -27,6 +27,8 @@ commons-math3 = "org.apache.commons:commons-math3:3.6.1"
 conscrypt = "org.conscrypt:conscrypt-openjdk-uber:2.5.2"
 cronet-api = "org.chromium.net:cronet-api:119.6045.31"
 cronet-embedded = "org.chromium.net:cronet-embedded:119.6045.31"
+dev-cel-compiler = "dev.cel:compiler:0.6.0"
+dev-cel-runtime = "dev.cel:runtime:0.6.0"
 # error-prone 2.31.0+ blocked on https://github.com/grpc/grpc-java/issues/10152
 # It breaks Bazel (ArrayIndexOutOfBoundsException in turbine) and Dexing ("D8:
 # java.lang.NullPointerException"). We can trivially upgrade the Bazel CI to

interop-testing/src/main/java/io/grpc/testing/integration/XdsTestServer.java

@@ -32,6 +32,7 @@
 import io.grpc.ServerCredentials;
 import io.grpc.ServerInterceptor;
 import io.grpc.ServerInterceptors;
+import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
 import io.grpc.gcp.csm.observability.CsmObservability;
 import io.grpc.health.v1.HealthCheckResponse.ServingStatus;
@@ -84,6 +85,7 @@ public final class XdsTestServer {
   private int port = 8080;
   private int maintenancePort = 8080;
   private boolean secureMode = false;
+  private boolean xdsServerMode = false;
   private boolean enableCsmObservability;
   private String serverId = "java_server";
   private HealthStatusManager health;
@@ -144,7 +146,10 @@ void parseArgs(String[] args) {
         maintenancePort = Integer.valueOf(value);
       } else if ("secure_mode".equals(key)) {
         secureMode = Boolean.parseBoolean(value);
-      } else if ("enable_csm_observability".equals(key)) {
+      } else if ("xds_server_mode".equals(key)) {
+        xdsServerMode = Boolean.parseBoolean(value);
+      }
+      else if ("enable_csm_observability".equals(key)) {
         enableCsmObservability = Boolean.valueOf(value);
       } else if ("server_id".equals(key)) {
         serverId = value;
@@ -165,6 +170,9 @@ void parseArgs(String[] args) {
               + maintenancePort);
       usage = true;
     }
+    if (secureMode) {
+      xdsServerMode = true;
+    }
 
     if (usage) {
       XdsTestServer s = new XdsTestServer();
@@ -181,6 +189,9 @@ void parseArgs(String[] args) {
               + " port and maintenance_port should be different for secure mode."
               + "\n                      Default: "
               + s.secureMode
+              + "\n  --xds_server_mode=BOOLEAN   Start in xDS Server mode."
+              + "\n                      Default: "
+              + s.xdsServerMode
               + "\n  --enable_csm_observability=BOOL  Enable CSM observability reporting. Default: "
               + s.enableCsmObservability
               + "\n  --server_id=STRING  server ID for response."
@@ -215,66 +226,72 @@ void start() throws Exception {
       throw new RuntimeException(e);
     }
     health = new HealthStatusManager();
+    ServerServiceDefinition testServiceInterceptor = ServerInterceptors.intercept(
+        new TestServiceImpl(serverId, host),
+        new TestInfoInterceptor(host));
+    ServerCredentials insecureServerCreds = InsecureServerCredentials.create();
+
     @SuppressWarnings("deprecation")
     BindableService oldReflectionService = ProtoReflectionService.newInstance();
     if (secureMode) {
       if (addressType != Util.AddressType.IPV4_IPV6) {
         throw new IllegalArgumentException("Secure mode only supports IPV4_IPV6 address type");
       }
       maintenanceServer =
-          Grpc.newServerBuilderForPort(maintenancePort, InsecureServerCredentials.create())
+          Grpc.newServerBuilderForPort(maintenancePort, insecureServerCreds)
               .addService(new XdsUpdateHealthServiceImpl(health))
               .addService(health.getHealthService())
               .addService(oldReflectionService)
               .addService(ProtoReflectionServiceV1.newInstance())
               .addServices(AdminInterface.getStandardServices())
               .build();
       maintenanceServer.start();
-      server =
-          XdsServerBuilder.forPort(
-                  port, XdsServerCredentials.create(InsecureServerCredentials.create()))
-              .addService(
-                  ServerInterceptors.intercept(
-                      new TestServiceImpl(serverId, host), new TestInfoInterceptor(host)))
+      server = XdsServerBuilder.forPort(port, XdsServerCredentials.create(insecureServerCreds))
+              .addService(testServiceInterceptor)
               .build();
       server.start();
-    } else {
-      ServerBuilder<?> serverBuilder;
-      ServerCredentials insecureServerCreds = InsecureServerCredentials.create();
-      switch (addressType) {
-        case IPV4_IPV6:
-          serverBuilder = Grpc.newServerBuilderForPort(port, insecureServerCreds);
-          break;
-        case IPV4:
-          SocketAddress v4Address = Util.getV4Address(port);
-          InetSocketAddress localV4Address = new InetSocketAddress("127.0.0.1", port);
-          serverBuilder = NettyServerBuilder.forAddress(
-                  localV4Address, insecureServerCreds);
-          if (v4Address != null && !v4Address.equals(localV4Address) ) {
-            ((NettyServerBuilder) serverBuilder).addListenAddress(v4Address);
-          }
-          break;
-        case IPV6:
-          List<SocketAddress> v6Addresses = Util.getV6Addresses(port);
-          InetSocketAddress localV6Address = new InetSocketAddress("::1", port);
-          serverBuilder = NettyServerBuilder.forAddress(localV6Address, insecureServerCreds);
-          for (SocketAddress address : v6Addresses) {
-            if (!address.equals(localV6Address)) {
-              ((NettyServerBuilder) serverBuilder).addListenAddress(address);
-            }
+      health.setStatus("", ServingStatus.SERVING);
+      return;
+    }
+
+    ServerBuilder<?> serverBuilder;
+    switch (addressType) {
+      case IPV4_IPV6:
+        serverBuilder = Grpc.newServerBuilderForPort(port, insecureServerCreds);
+        break;
+      case IPV4:
+        SocketAddress v4Address = Util.getV4Address(port);
+        InetSocketAddress localV4Address = new InetSocketAddress("127.0.0.1", port);
+        serverBuilder = NettyServerBuilder.forAddress(
+                localV4Address, insecureServerCreds);
+        if (v4Address != null && !v4Address.equals(localV4Address) ) {
+          ((NettyServerBuilder) serverBuilder).addListenAddress(v4Address);
+        }
+        break;
+      case IPV6:
+        List<SocketAddress> v6Addresses = Util.getV6Addresses(port);
+        InetSocketAddress localV6Address = new InetSocketAddress("::1", port);
+        serverBuilder = NettyServerBuilder.forAddress(localV6Address, insecureServerCreds);
+        for (SocketAddress address : v6Addresses) {
+          if (!address.equals(localV6Address)) {
+            ((NettyServerBuilder) serverBuilder).addListenAddress(address);
           }
-          break;
-        default:
-          throw new AssertionError("Unknown address type: " + addressType);
+        }
+        break;
+      default:
+        throw new AssertionError("Unknown address type: " + addressType);
+    }
+
+    if (xdsServerMode) {
+      if (addressType != Util.AddressType.IPV4_IPV6) {
+        throw new IllegalArgumentException("xDS Server mode only supports IPV4_IPV6 address type");
       }
 
       logger.info("Starting server on port " + port + " with address type " + addressType);
 
       server =
           serverBuilder
-              .addService(
-                  ServerInterceptors.intercept(
-                      new TestServiceImpl(serverId, host), new TestInfoInterceptor(host)))
+              .addService(testServiceInterceptor)
               .addService(new XdsUpdateHealthServiceImpl(health))
               .addService(health.getHealthService())
               .addService(oldReflectionService)
@@ -283,7 +300,23 @@ void start() throws Exception {
               .build();
       server.start();
       maintenanceServer = null;
+      health.setStatus("", ServingStatus.SERVING);
+      return;
     }
+
+    logger.info("Starting server on port " + port + " with address type " + addressType);
+
+    server =
+        serverBuilder
+            .addService(testServiceInterceptor)
+            .addService(new XdsUpdateHealthServiceImpl(health))
+            .addService(health.getHealthService())
+            .addService(oldReflectionService)
+            .addService(ProtoReflectionServiceV1.newInstance())
+            .addServices(AdminInterface.getStandardServices())
+            .build();
+    server.start();
+    maintenanceServer = null;
     health.setStatus("", ServingStatus.SERVING);
   }
 

xds/build.gradle

@@ -52,6 +52,7 @@ dependencies {
             project(':grpc-services'),
             project(':grpc-auth'),
             project(path: ':grpc-alts', configuration: 'shadow'),
+            libraries.dev.cel.runtime,
             libraries.guava,
             libraries.gson,
             libraries.re2j,
@@ -70,7 +71,8 @@ dependencies {
     compileOnly libraries.netty.transport.epoll
 
     testImplementation project(':grpc-testing'),
-            project(':grpc-testing-proto')
+            project(':grpc-testing-proto'),
+            libraries.dev.cel.compiler
     testImplementation (libraries.netty.transport.epoll) {
         artifact {
             classifier = "linux-x86_64"

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

@@ -38,6 +38,10 @@
    */
   String[] typeUrls();
 
+  default boolean isEnabled() {
+    return true;
+  }
+
   /**
    * Parses the top-level filter config from raw proto message. The message may be either a {@link
    * com.google.protobuf.Any} or a {@link com.google.protobuf.Struct}.
@@ -50,6 +54,12 @@
    */
   ConfigOrError<? extends FilterConfig> parseFilterConfigOverride(Message rawProtoMessage);
 
+  default void shutdown() {
+    // Implement as needed.
+    // TODO(sergiitk): [DESIGN] important to cover and discuss in the design.
+    // TODO(sergiitk): [QUESTION] should it be in ServerInterceptorBuilder?
+  }
+
   /** Represents an opaque data structure holding configuration for a filter. */
   interface FilterConfig {
     String typeUrl();
@@ -68,6 +78,15 @@
     @Nullable
     ServerInterceptor buildServerInterceptor(
         FilterConfig config, @Nullable FilterConfig overrideConfig);
+
+    @Nullable
+    default ServerInterceptor buildServerInterceptor(
+        FilterConfig config,
+        @Nullable FilterConfig overrideConfig,
+        ScheduledExecutorService scheduler) {
+      return buildServerInterceptor(config, overrideConfig);
+    }
+
   }
 
   /** Filter config with instance name. */

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

@@ -108,6 +108,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
         drainGraceTime = drainGraceNanosObj;
         drainGraceTimeUnit = TimeUnit.NANOSECONDS;
       }
+      // TODO(sergiitk): [design] drains connections on LDS update.
       FilterChainSelectorManager.Closer closer = new FilterChainSelectorManager.Closer(
           new GracefullyShutdownChannelRunnable(ctx.channel(), drainGraceTime, drainGraceTimeUnit));
       FilterChainSelector selector = filterChainSelectorManager.register(closer);

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

@@ -65,6 +65,7 @@ public void updateSelector(FilterChainSelector newSelector) {
       closers = new TreeSet<Closer>(closers.comparator());
       selector = newSelector;
     }
+    // TODO(sergiitk): [design] calls the closer of FilterChainMatchingNegotiatorServerFactory
     for (Closer closer : oldClosers) {
       closer.closer.run();
     }

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

@@ -37,7 +37,8 @@ static synchronized FilterRegistry getDefaultRegistry() {
       instance = newRegistry().register(
               FaultFilter.INSTANCE,
               RouterFilter.INSTANCE,
-              RbacFilter.INSTANCE);
+              RbacFilter.INSTANCE,
+              RlqsFilter.INSTANCE);
     }
     return instance;
   }
@@ -50,6 +51,9 @@ static FilterRegistry newRegistry() {
   @VisibleForTesting
   FilterRegistry register(Filter... filters) {
     for (Filter filter : filters) {
+      if (!filter.isEnabled()) {
+        continue;
+      }
       for (String typeUrl : filter.typeUrls()) {
         supportedFilters.put(typeUrl, filter);
       }

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

@@ -16,6 +16,8 @@
 
 package io.grpc.xds;
 
+import com.github.xds.type.matcher.v3.CelMatcher;
+import com.github.xds.type.matcher.v3.HttpAttributesCelMatchInput;
 import com.google.protobuf.Descriptors.Descriptor;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
@@ -28,6 +30,8 @@
 import io.envoyproxy.envoy.config.route.v3.RouteConfiguration;
 import io.envoyproxy.envoy.extensions.clusters.aggregate.v3.ClusterConfig;
 import io.envoyproxy.envoy.extensions.filters.http.fault.v3.HTTPFault;
+import io.envoyproxy.envoy.extensions.filters.http.rate_limit_quota.v3.RateLimitQuotaFilterConfig;
+import io.envoyproxy.envoy.extensions.filters.http.rate_limit_quota.v3.RateLimitQuotaOverride;
 import io.envoyproxy.envoy.extensions.filters.http.rbac.v3.RBAC;
 import io.envoyproxy.envoy.extensions.filters.http.rbac.v3.RBACPerRoute;
 import io.envoyproxy.envoy.extensions.filters.http.router.v3.Router;
@@ -58,6 +62,11 @@ private static JsonFormat.Printer newPrinter() {
               .add(RBAC.getDescriptor())
               .add(RBACPerRoute.getDescriptor())
               .add(Router.getDescriptor())
+              // RLQS
+              .add(RateLimitQuotaFilterConfig.getDescriptor())
+              .add(RateLimitQuotaOverride.getDescriptor())
+              .add(HttpAttributesCelMatchInput.getDescriptor())
+              .add(CelMatcher.getDescriptor())
               // UpstreamTlsContext and DownstreamTlsContext in v3 are not transitively imported
               // by top-level resource types.
               .add(UpstreamTlsContext.getDescriptor())