PSM e2e: works!

Author: sergiitk

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

@@ -28,6 +28,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 +60,9 @@ private static JsonFormat.Printer newPrinter() {
               .add(RBAC.getDescriptor())
               .add(RBACPerRoute.getDescriptor())
               .add(Router.getDescriptor())
+              // RLQS
+              .add(RateLimitQuotaFilterConfig.getDescriptor())
+              .add(RateLimitQuotaOverride.getDescriptor())
               // UpstreamTlsContext and DownstreamTlsContext in v3 are not transitively imported
               // by top-level resource types.
               .add(UpstreamTlsContext.getDescriptor())

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

@@ -54,6 +54,8 @@
 // TODO(sergiitk): introduce a layer between the filter and interceptor.
 // lds has filter names and the names are unique - even for server instances.
 final class RlqsFilter implements Filter, ServerInterceptorBuilder {
+  private final XdsLogger logger;
+
   static final boolean enabled = GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_ENABLE_RLQS", false);
 
   // TODO(sergiitk): [IMPL] remove
@@ -69,12 +71,11 @@ final class RlqsFilter implements Filter, ServerInterceptorBuilder {
 
   private final AtomicReference<RlqsCache> rlqsCache = new AtomicReference<>();
 
-  private final XdsLogger logger;
-
   public RlqsFilter() {
-    InternalLogId logId = InternalLogId.allocate("rlqs-filter", null);
-    logger = XdsLogger.withLogId(logId);
-    logger.log(XdsLogLevel.INFO, "Created RLQS Filter with logId=" + logId);
+    // TODO(sergiitk): one per new instance when filters are refactored.
+    logger = XdsLogger.withLogId(InternalLogId.allocate(this.getClass(), null));
+    logger.log(XdsLogLevel.DEBUG,
+        "Created RLQS Filter with enabled=" + enabled + ", dryRun=" + dryRun);
   }
 
   @Override
@@ -177,7 +178,7 @@ public <ReqT, RespT> Listener<ReqT> interceptCall(
 
         // TODO(sergiitk): [IMPL] Remove
         if (dryRun) {
-          logger.log(XdsLogLevel.INFO, "RLQS DRY RUN: request <<" + httpMatchInput + ">>");
+          // logger.log(XdsLogLevel.INFO, "RLQS DRY RUN: request <<" + httpMatchInput + ">>");
           return next.startCall(call, headers);
         }
 
@@ -204,7 +205,7 @@ RlqsFilterConfig parseRlqsFilter(RateLimitQuotaFilterConfig rlqsFilterProto)
 
     // TODO(sergiitk): [IMPL] Remove
     if (dryRun) {
-      logger.log(XdsLogLevel.INFO, "RLQS DRY RUN: skipping matchers");
+      logger.log(XdsLogLevel.DEBUG, "Dry run: not parsing  matchers in the filter filter");
       return builder.build();
     }
 

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

@@ -59,6 +59,7 @@
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -82,7 +83,9 @@ final class XdsServerWrapper extends Server {
       new Thread.UncaughtExceptionHandler() {
         @Override
         public void uncaughtException(Thread t, Throwable e) {
-          logger.log(Level.SEVERE, "Exception!" + e);
+          logger.log(Level.SEVERE, "Exception! "
+              + e + "\nTrace:\n"
+              + Arrays.toString(e.getStackTrace()).replace(',', '\n'));
           // TODO(chengyuanzhang): implement cleanup.
         }
       });

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -44,6 +44,7 @@
 import io.grpc.xds.client.XdsClient.XdsResponseHandler;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
 import java.net.URI;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -72,8 +73,9 @@ public final class XdsClientImpl extends XdsClient implements XdsResponseHandler
         public void uncaughtException(Thread t, Throwable e) {
           logger.log(
               XdsLogLevel.ERROR,
-              "Uncaught exception in XdsClient SynchronizationContext. Panic!",
-              e);
+              "Uncaught exception in XdsClient SynchronizationContext. Panic!  "
+              + e + "\nTrace:\n"
+              + Arrays.toString(e.getStackTrace()).replace(',', '\n'));
           // TODO(chengyuanzhang): better error handling.
           throw new AssertionError(e);
         }

xds/src/main/java/io/grpc/xds/internal/rlqs/RlqsCache.java

@@ -99,7 +99,12 @@ private long hashRlqsFilterConfig(RlqsFilterConfig config) {
     // TODO(sergiitk): [DESIGN] the key should be hashed (domain + buckets) merged config?
     // TODO(sergiitk): [IMPL] Hash buckets
     int k1 = Objects.hash(config.rlqsService().targetUri(), config.domain());
-    int k2 = config.bucketMatchers().hashCode();
+    int k2;
+    if (config.bucketMatchers() == null) {
+      k2 = 0x42c0ffee;
+    } else {
+      k2 = config.bucketMatchers().hashCode();
+    }
     return Long.rotateLeft(Integer.toUnsignedLong(k1), 32) + Integer.toUnsignedLong(k2);
   }
 

xds/src/main/java/io/grpc/xds/internal/rlqs/RlqsClient.java

@@ -25,31 +25,41 @@
 import io.envoyproxy.envoy.service.rate_limit_quota.v3.RateLimitQuotaUsageReports;
 import io.envoyproxy.envoy.service.rate_limit_quota.v3.RateLimitQuotaUsageReports.BucketQuotaUsage;
 import io.grpc.Grpc;
+import io.grpc.InternalLogId;
 import io.grpc.ManagedChannel;
+import io.grpc.internal.GrpcUtil;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import io.grpc.xds.client.Bootstrapper.RemoteServerInfo;
+import io.grpc.xds.client.XdsLogger;
+import io.grpc.xds.client.XdsLogger.XdsLogLevel;
 import io.grpc.xds.internal.rlqs.RlqsBucket.RlqsBucketUsage;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
-import java.util.logging.Level;
-import java.util.logging.Logger;
+import javax.annotation.Nullable;
 
 public final class RlqsClient {
-  private static final Logger logger = Logger.getLogger(RlqsClient.class.getName());
+  // TODO(sergiitk): [IMPL] remove
+  // Do do not fail on parsing errors, only log requests.
+  static final boolean dryRun = GrpcUtil.getFlag("GRPC_EXPERIMENTAL_RLQS_DRY_RUN", false);
+
+  private final XdsLogger logger;
 
   private final RemoteServerInfo serverInfo;
   private final Consumer<List<RlqsUpdateBucketAction>> bucketsUpdateCallback;
   private final RlqsStream rlqsStream;
 
   RlqsClient(
       RemoteServerInfo serverInfo, String domain,
-      Consumer<List<RlqsUpdateBucketAction>> bucketsUpdateCallback) {
+      Consumer<List<RlqsUpdateBucketAction>> bucketsUpdateCallback, String prettyHash) {
     // TODO(sergiitk): [post] check not null.
     this.serverInfo = serverInfo;
     this.bucketsUpdateCallback = bucketsUpdateCallback;
+
+    logger = XdsLogger.withLogId(
+        InternalLogId.allocate(this.getClass(), "<" + prettyHash + "> " + serverInfo.target()));
+
     this.rlqsStream = new RlqsStream(serverInfo, domain);
   }
 
@@ -62,7 +72,7 @@ public void sendUsageReports(List<RlqsBucketUsage> bucketUsages) {
   }
 
   public void shutdown() {
-    logger.log(Level.FINER, "Shutting down RlqsClient to {0}", serverInfo.target());
+    logger.log(XdsLogLevel.DEBUG, "Shutting down RlqsClient to {0}", serverInfo.target());
     // TODO(sergiitk): [IMPL] RlqsClient shutdown
   }
 
@@ -72,18 +82,26 @@ public void handleStreamClosed() {
 
   private class RlqsStream {
     private final AtomicBoolean isFirstReport = new AtomicBoolean(true);
-    private final ManagedChannel channel;
     private final String domain;
+    @Nullable
     private final ClientCallStreamObserver<RateLimitQuotaUsageReports> clientCallStream;
 
     RlqsStream(RemoteServerInfo serverInfo, String domain) {
       this.domain = domain;
-      channel = Grpc.newChannelBuilder(serverInfo.target(), serverInfo.channelCredentials())
-          .keepAliveTime(10, TimeUnit.SECONDS)
-          .keepAliveWithoutCalls(true)
-          .build();
-      // keepalive?
+
+      if (dryRun) {
+        clientCallStream = null;
+        logger.log(XdsLogLevel.DEBUG, "Dry run, not connecting to " + serverInfo.target());
+        return;
+      }
+
       // TODO(sergiitk): [IMPL] Manage State changes?
+      ManagedChannel channel =
+          Grpc.newChannelBuilder(serverInfo.target(), serverInfo.channelCredentials()).build();
+      // keepalive?
+      //   .keepAliveTime(10, TimeUnit.SECONDS)
+      //   .keepAliveWithoutCalls(true)
+
       RateLimitQuotaServiceStub stub = RateLimitQuotaServiceGrpc.newStub(channel);
       clientCallStream = (ClientCallStreamObserver<RateLimitQuotaUsageReports>)
           stub.streamRateLimitQuotas(new RlqsStreamObserver());
@@ -107,6 +125,10 @@ void reportUsage(List<RlqsBucket.RlqsBucketUsage> usageReports) {
       for (RlqsBucket.RlqsBucketUsage bucketUsage : usageReports) {
         report.addBucketQuotaUsages(toUsageReport(bucketUsage));
       }
+      if (clientCallStream == null) {
+        logger.log(XdsLogLevel.DEBUG, "Dry run, skipping bucket usage report: " + report.build());
+        return;
+      }
       clientCallStream.onNext(report.build());
     }
 
@@ -128,12 +150,12 @@ public void onNext(RateLimitQuotaResponse response) {
 
       @Override
       public void onError(Throwable t) {
-
+        logger.log(XdsLogLevel.DEBUG, "Got error in RlqsStreamObserver: " + t.toString());
       }
 
       @Override
       public void onCompleted() {
-
+        logger.log(XdsLogLevel.DEBUG, "RlqsStreamObserver completed");
       }
     }
   }

xds/src/main/java/io/grpc/xds/internal/rlqs/RlqsEngine.java

@@ -17,7 +17,10 @@
 package io.grpc.xds.internal.rlqs;
 
 import com.google.common.collect.ImmutableList;
+import io.grpc.InternalLogId;
 import io.grpc.xds.client.Bootstrapper.RemoteServerInfo;
+import io.grpc.xds.client.XdsLogger;
+import io.grpc.xds.client.XdsLogger.XdsLogLevel;
 import io.grpc.xds.internal.datatype.RateLimitStrategy;
 import io.grpc.xds.internal.matchers.HttpMatchInput;
 import io.grpc.xds.internal.matchers.Matcher;
@@ -29,11 +32,9 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
 public class RlqsEngine {
-  private static final Logger logger = Logger.getLogger(RlqsEngine.class.getName());
+  private final XdsLogger logger;
 
   private final RlqsClient rlqsClient;
   private final Matcher<HttpMatchInput, RlqsBucketSettings> bucketMatchers;
@@ -49,8 +50,14 @@ public RlqsEngine(
     this.bucketMatchers = bucketMatchers;
     this.configHash = configHash;
     this.scheduler = scheduler;
+
+    String prettyHash = "0x" + Long.toHexString(configHash);
+    logger = XdsLogger.withLogId(InternalLogId.allocate(this.getClass(), prettyHash));
+    logger.log(XdsLogLevel.DEBUG,
+        "Initialized RlqsEngine for hash={0}, domain={1}", prettyHash, domain);
+
     bucketCache = new RlqsBucketCache();
-    rlqsClient = new RlqsClient(rlqsServer, domain, this::onBucketsUpdate);
+    rlqsClient = new RlqsClient(rlqsServer, domain, this::onBucketsUpdate, prettyHash);
   }
 
   public RlqsRateLimitResult rateLimit(HttpMatchInput input) {
@@ -95,7 +102,8 @@ private void scheduleImmediateReport(RlqsBucket newBucket) {
           1, TimeUnit.MICROSECONDS);
     } catch (RejectedExecutionException e) {
       // Shouldn't happen.
-      logger.finer("Couldn't schedule immediate report for bucket " + newBucket.getBucketId());
+      logger.log(XdsLogLevel.WARNING,
+          "Couldn't schedule immediate report for bucket " + newBucket.getBucketId());
     }
   }
 
@@ -123,7 +131,7 @@ private void reportBucketsWithInterval(long intervalMillis) {
   public void shutdown() {
     // TODO(sergiitk): [IMPL] Timers shutdown
     // TODO(sergiitk): [IMPL] RlqsEngine shutdown
-    logger.log(Level.FINER, "Shutting down RlqsEngine with hash {0}", configHash);
+    logger.log(XdsLogLevel.DEBUG, "Shutting down RlqsEngine with hash {0}", configHash);
     rlqsClient.shutdown();
   }
 }