core: initial binary log class (#3699)

Each MethodDescriptor will have a binary log, though the log may
be a noop log. The binary log class is a skeleton class at the
moment, but does contain the max header and message length
info. The limits are determined by parsing the shell variable
GRPC_BINARY_LOG_CONFIG.

core/src/main/java/io/grpc/internal/BinaryLog.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2017, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
+import io.grpc.MethodDescriptor;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+
+/**
+ * A binary log class that is configured for a specific {@link MethodDescriptor}.
+ */
+final class BinaryLog {
+  private static final Logger logger = Logger.getLogger(BinaryLog.class.getName());
+  private static final BinaryLog NOOP_LOG =
+      new BinaryLog(/*maxHeaderBytes=*/ 0, /*maxMessageBytes=*/ 0);
+  private final int maxHeaderBytes;
+  private final int maxMessageBytes;
+
+  @VisibleForTesting
+  BinaryLog(int maxHeaderBytes, int maxMessageBytes) {
+    this.maxHeaderBytes = maxHeaderBytes;
+    this.maxMessageBytes = maxMessageBytes;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (!(o instanceof BinaryLog)) {
+      return false;
+    }
+    BinaryLog that = (BinaryLog) o;
+    return this.maxHeaderBytes == that.maxHeaderBytes
+        && this.maxMessageBytes == that.maxMessageBytes;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(maxHeaderBytes, maxMessageBytes);
+  }
+
+  @Override
+  public String toString() {
+    return getClass().getSimpleName() + '['
+        + "maxHeaderBytes=" + maxHeaderBytes + ", "
+        + "maxMessageBytes=" + maxMessageBytes + "]";
+  }
+
+  private static final Factory DEFAULT_FACTORY;
+  private static final Factory NOOP_FACTORY = new Factory() {
+    @Override
+    public BinaryLog getLog(String fullMethodName) {
+      return NOOP_LOG;
+    }
+  };
+
+  static {
+    Factory defaultFactory = NOOP_FACTORY;
+    try {
+      String configStr = System.getenv("GRPC_BINARY_LOG_CONFIG");
+      if (configStr != null && configStr.length() > 0) {
+        defaultFactory = new FactoryImpl(configStr);
+      }
+    } catch (Throwable t) {
+      logger.log(Level.SEVERE, "Failed to initialize binary log. Disabling binary log.", t);
+      defaultFactory = NOOP_FACTORY;
+    }
+    DEFAULT_FACTORY = defaultFactory;
+  }
+
+  /**
+   * Accepts the fullMethodName and returns the binary log that should be used. The log may be
+   * a log that does nothing.
+   */
+  static BinaryLog getLog(String fullMethodName) {
+    return DEFAULT_FACTORY.getLog(fullMethodName);
+  }
+
+  interface Factory {
+    BinaryLog getLog(String fullMethodName);
+  }
+
+  static final class FactoryImpl implements Factory {
+    // '*' for global, 'service/*' for service glob, or 'service/method' for fully qualified.
+    private static final Pattern logPatternRe = Pattern.compile("[^{]+");
+    // A curly brace wrapped expression. Will be further matched with the more specified REs below.
+    private static final Pattern logOptionsRe = Pattern.compile("\\{[^}]+}");
+    private static final Pattern configRe = Pattern.compile(
+        String.format("^(%s)(%s)?$", logPatternRe.pattern(), logOptionsRe.pattern()));
+    // Regexes to extract per-binlog options
+    // The form: {m:256}
+    private static final Pattern msgRe = Pattern.compile("\\{m(?::(\\d+))?}");
+    // The form: {h:256}
+    private static final Pattern headerRe = Pattern.compile("\\{h(?::(\\d+))?}");
+    // The form: {h:256,m:256}
+    private static final Pattern bothRe = Pattern.compile("\\{h(?::(\\d+))?;m(?::(\\d+))?}");
+
+    private final BinaryLog globalLog;
+    private final Map<String, BinaryLog> perServiceLogs;
+    private final Map<String, BinaryLog> perMethodLogs;
+
+    /**
+     * Accepts a string in the format specified by the binary log spec.
+     */
+    @VisibleForTesting
+    FactoryImpl(String configurationString) {
+      Preconditions.checkState(configurationString != null && configurationString.length() > 0);
+      BinaryLog globalLog = null;
+      Map<String, BinaryLog> perServiceLogs = new HashMap<String, BinaryLog>();
+      Map<String, BinaryLog> perMethodLogs = new HashMap<String, BinaryLog>();
+
+      String[] configurations = configurationString.split(",");
+      for (String configuration : configurations) {
+        Matcher configMatcher = configRe.matcher(configuration);
+        if (!configMatcher.matches()) {
+          throw new IllegalArgumentException("Bad input: " + configuration);
+        }
+        String methodOrSvc = configMatcher.group(1);
+        String binlogOptionStr = configMatcher.group(2);
+        BinaryLog binLog = createBinaryLog(binlogOptionStr);
+        if (binLog == null) {
+          continue;
+        }
+        if (methodOrSvc.equals("*")) {
+          if (globalLog != null) {
+            logger.log(Level.SEVERE, "Ignoring duplicate entry: " + configuration);
+            continue;
+          }
+          globalLog = binLog;
+          logger.info("Global binlog: " + globalLog);
+        } else if (isServiceGlob(methodOrSvc)) {
+          String service = MethodDescriptor.extractFullServiceName(methodOrSvc);
+          if (perServiceLogs.containsKey(service)) {
+            logger.log(Level.SEVERE, "Ignoring duplicate entry: " + configuration);
+            continue;
+          }
+          perServiceLogs.put(service, binLog);
+          logger.info(String.format("Service binlog: service=%s log=%s", service, binLog));
+        } else {
+          // assume fully qualified method name
+          if (perMethodLogs.containsKey(methodOrSvc)) {
+            logger.log(Level.SEVERE, "Ignoring duplicate entry: " + configuration);
+            continue;
+          }
+          perMethodLogs.put(methodOrSvc, binLog);
+          logger.info(String.format("Method binlog: method=%s log=%s", methodOrSvc, binLog));
+        }
+      }
+      this.globalLog = globalLog == null ? NOOP_LOG : globalLog;
+      this.perServiceLogs = Collections.unmodifiableMap(perServiceLogs);
+      this.perMethodLogs = Collections.unmodifiableMap(perMethodLogs);
+    }
+
+    /**
+     * Accepts a full method name and returns the log that should be used.
+     */
+    @Override
+    public BinaryLog getLog(String fullMethodName) {
+      BinaryLog methodLog = perMethodLogs.get(fullMethodName);
+      if (methodLog != null) {
+        return methodLog;
+      }
+      BinaryLog serviceLog = perServiceLogs.get(
+          MethodDescriptor.extractFullServiceName(fullMethodName));
+      if (serviceLog != null) {
+        return serviceLog;
+      }
+      return globalLog;
+    }
+
+    /**
+     * Returns a binlog with the correct header and message limits or {@code null} if the input
+     * is malformed. The input should be a string that is in one of these forms:
+     *
+     * <p>{@code {h(:\d+)?}, {m(:\d+)?}, {h(:\d+)?,m(:\d+)?}}
+     *
+     * <p>If the {@code logConfig} is null, the returned binlog will have a limit of
+     * Integer.MAX_VALUE.
+     */
+    @VisibleForTesting
+    @Nullable
+    static BinaryLog createBinaryLog(@Nullable String logConfig) {
+      if (logConfig == null) {
+        return new BinaryLog(Integer.MAX_VALUE, Integer.MAX_VALUE);
+      }
+      try {
+        Matcher headerMatcher;
+        Matcher msgMatcher;
+        Matcher bothMatcher;
+        final int maxHeaderBytes;
+        final int maxMsgBytes;
+        if ((headerMatcher = headerRe.matcher(logConfig)).matches()) {
+          String maxHeaderStr = headerMatcher.group(1);
+          maxHeaderBytes =
+              maxHeaderStr != null ? Integer.parseInt(maxHeaderStr) : Integer.MAX_VALUE;
+          maxMsgBytes = 0;
+        } else if ((msgMatcher = msgRe.matcher(logConfig)).matches()) {
+          maxHeaderBytes = 0;
+          String maxMsgStr = msgMatcher.group(1);
+          maxMsgBytes = maxMsgStr != null ? Integer.parseInt(maxMsgStr) : Integer.MAX_VALUE;
+        } else if ((bothMatcher = bothRe.matcher(logConfig)).matches()) {
+          String maxHeaderStr = bothMatcher.group(1);
+          String maxMsgStr = bothMatcher.group(2);
+          maxHeaderBytes =
+              maxHeaderStr != null ? Integer.parseInt(maxHeaderStr) : Integer.MAX_VALUE;
+          maxMsgBytes = maxMsgStr != null ? Integer.parseInt(maxMsgStr) : Integer.MAX_VALUE;
+        } else {
+          logger.log(Level.SEVERE, "Illegal log config pattern: " + logConfig);
+          return null;
+        }
+        return new BinaryLog(maxHeaderBytes, maxMsgBytes);
+      } catch (NumberFormatException e) {
+        logger.log(Level.SEVERE, "Illegal log config pattern: " + logConfig);
+        return null;
+      }
+    }
+
+    /**
+     * Returns true if the input string is a glob of the form: {@code <package-service>/*}.
+     */
+    static boolean isServiceGlob(String input) {
+      return input.endsWith("/*");
+    }
+  }
+}

core/src/main/java/io/grpc/internal/ClientCallImpl.java

@@ -82,6 +82,7 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT> {
   private boolean fullStreamDecompression;
   private DecompressorRegistry decompressorRegistry = DecompressorRegistry.getDefaultInstance();
   private CompressorRegistry compressorRegistry = CompressorRegistry.getDefaultInstance();
+  private final BinaryLog binlog;
 
   ClientCallImpl(
       MethodDescriptor<ReqT, RespT> method, Executor executor, CallOptions callOptions,
@@ -101,6 +102,7 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT> {
     this.callOptions = callOptions;
     this.clientTransportProvider = clientTransportProvider;
     this.deadlineCancellationExecutor = deadlineCancellationExecutor;
+    this.binlog = BinaryLog.getLog(method.getFullMethodName());
   }
 
   private final class ContextCancellationListener implements CancellationListener {

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -54,6 +54,7 @@ final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
   private final byte[] messageAcceptEncoding;
   private final DecompressorRegistry decompressorRegistry;
   private final CompressorRegistry compressorRegistry;
+  private final BinaryLog binlog;
 
   // state
   private volatile boolean cancelled;
@@ -71,6 +72,7 @@ final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
     this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);
     this.decompressorRegistry = decompressorRegistry;
     this.compressorRegistry = compressorRegistry;
+    binlog = BinaryLog.getLog(method.getFullMethodName());
   }
 
   @Override

core/src/test/java/io/grpc/internal/BinaryLogTest.java

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2017, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import io.grpc.internal.BinaryLog.FactoryImpl;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link BinaryLog}. */
+@RunWith(JUnit4.class)
+public final class BinaryLogTest {
+  private static final BinaryLog NONE = new Builder().build();
+  private static final BinaryLog HEADER_FULL = new Builder().header(Integer.MAX_VALUE).build();
+  private static final BinaryLog HEADER_256 = new Builder().header(256).build();
+  private static final BinaryLog MSG_FULL = new Builder().msg(Integer.MAX_VALUE).build();
+  private static final BinaryLog MSG_256 = new Builder().msg(256).build();
+  private static final BinaryLog BOTH_256 = new Builder().header(256).msg(256).build();
+  private static final BinaryLog BOTH_FULL =
+      new Builder().header(Integer.MAX_VALUE).msg(Integer.MAX_VALUE).build();
+  
+
+  @Test
+  public void configBinLog_global() throws Exception {
+    assertEquals(BOTH_FULL, new FactoryImpl("*").getLog("p.s/m"));
+    assertEquals(BOTH_FULL, new FactoryImpl("*{h;m}").getLog("p.s/m"));
+    assertEquals(HEADER_FULL, new FactoryImpl("*{h}").getLog("p.s/m"));
+    assertEquals(MSG_FULL, new FactoryImpl("*{m}").getLog("p.s/m"));
+    assertEquals(HEADER_256, new FactoryImpl("*{h:256}").getLog("p.s/m"));
+    assertEquals(MSG_256, new FactoryImpl("*{m:256}").getLog("p.s/m"));
+    assertEquals(BOTH_256, new FactoryImpl("*{h:256;m:256}").getLog("p.s/m"));
+    assertEquals(
+        new Builder().header(Integer.MAX_VALUE).msg(256).build(),
+        new FactoryImpl("*{h;m:256}").getLog("p.s/m"));
+    assertEquals(
+        new Builder().header(256).msg(Integer.MAX_VALUE).build(),
+        new FactoryImpl("*{h:256;m}").getLog("p.s/m"));
+  }
+
+  @Test
+  public void configBinLog_method() throws Exception {
+    assertEquals(BOTH_FULL, new FactoryImpl("p.s/m").getLog("p.s/m"));
+    assertEquals(BOTH_FULL, new FactoryImpl("p.s/m{h;m}").getLog("p.s/m"));
+    assertEquals(HEADER_FULL, new FactoryImpl("p.s/m{h}").getLog("p.s/m"));
+    assertEquals(MSG_FULL, new FactoryImpl("p.s/m{m}").getLog("p.s/m"));
+    assertEquals(HEADER_256, new FactoryImpl("p.s/m{h:256}").getLog("p.s/m"));
+    assertEquals(MSG_256, new FactoryImpl("p.s/m{m:256}").getLog("p.s/m"));
+    assertEquals(BOTH_256, new FactoryImpl("p.s/m{h:256;m:256}").getLog("p.s/m"));
+    assertEquals(
+        new Builder().header(Integer.MAX_VALUE).msg(256).build(),
+        new FactoryImpl("p.s/m{h;m:256}").getLog("p.s/m"));
+    assertEquals(
+        new Builder().header(256).msg(Integer.MAX_VALUE).build(),
+        new FactoryImpl("p.s/m{h:256;m}").getLog("p.s/m"));
+  }
+
+  @Test
+  public void configBinLog_method_absent() throws Exception {
+    assertEquals(NONE, new FactoryImpl("p.s/m").getLog("p.s/absent"));
+  }
+
+  @Test
+  public void configBinLog_service() throws Exception {
+    assertEquals(BOTH_FULL, new FactoryImpl("p.s/*").getLog("p.s/m"));
+    assertEquals(BOTH_FULL, new FactoryImpl("p.s/*{h;m}").getLog("p.s/m"));
+    assertEquals(HEADER_FULL, new FactoryImpl("p.s/*{h}").getLog("p.s/m"));
+    assertEquals(MSG_FULL, new FactoryImpl("p.s/*{m}").getLog("p.s/m"));
+    assertEquals(HEADER_256, new FactoryImpl("p.s/*{h:256}").getLog("p.s/m"));
+    assertEquals(MSG_256, new FactoryImpl("p.s/*{m:256}").getLog("p.s/m"));
+    assertEquals(BOTH_256, new FactoryImpl("p.s/*{h:256;m:256}").getLog("p.s/m"));
+    assertEquals(
+        new Builder().header(Integer.MAX_VALUE).msg(256).build(),
+        new FactoryImpl("p.s/*{h;m:256}").getLog("p.s/m"));
+    assertEquals(
+        new Builder().header(256).msg(Integer.MAX_VALUE).build(),
+        new FactoryImpl("p.s/*{h:256;m}").getLog("p.s/m"));
+  }
+
+  @Test
+  public void configBinLog_service_absent() throws Exception {
+    assertEquals(NONE, new FactoryImpl("p.s/*").getLog("p.other/m"));
+  }
+
+  @Test
+  public void createLogFromOptionString() throws Exception {
+    assertEquals(BOTH_FULL, FactoryImpl.createBinaryLog(/*logConfig=*/ null));
+    assertEquals(HEADER_FULL, FactoryImpl.createBinaryLog("{h}"));
+    assertEquals(MSG_FULL, FactoryImpl.createBinaryLog("{m}"));
+    assertEquals(HEADER_256, FactoryImpl.createBinaryLog("{h:256}"));
+    assertEquals(MSG_256, FactoryImpl.createBinaryLog("{m:256}"));
+    assertEquals(BOTH_256, FactoryImpl.createBinaryLog("{h:256;m:256}"));
+    assertEquals(
+        new Builder().header(Integer.MAX_VALUE).msg(256).build(),
+        FactoryImpl.createBinaryLog("{h;m:256}"));
+    assertEquals(
+        new Builder().header(256).msg(Integer.MAX_VALUE).build(),
+        FactoryImpl.createBinaryLog("{h:256;m}"));
+  }
+
+  @Test
+  public void createLogFromOptionString_malformed() throws Exception {
+    assertNull(FactoryImpl.createBinaryLog("bad"));
+    assertNull(FactoryImpl.createBinaryLog("{bad}"));
+    assertNull(FactoryImpl.createBinaryLog("{x;y}"));
+    assertNull(FactoryImpl.createBinaryLog("{h:abc}"));
+    assertNull(FactoryImpl.createBinaryLog("{2}"));
+    assertNull(FactoryImpl.createBinaryLog("{2;2}"));
+    // The grammar specifies that if both h and m are present, h comes before m
+    assertNull(FactoryImpl.createBinaryLog("{m:123;h:123}"));
+    // NumberFormatException
+    assertNull(FactoryImpl.createBinaryLog("{h:99999999999999}"));
+  }
+
+  @Test
+  public void configBinLog_multiConfig_withGlobal() throws Exception {
+    FactoryImpl factory = new FactoryImpl(
+        "*{h},"
+        + "package.both256/*{h:256;m:256},"
+        + "package.service1/both128{h:128;m:128},"
+        + "package.service2/method_messageOnly{m}");
+    assertEquals(HEADER_FULL, factory.getLog("otherpackage.service/method"));
+
+    assertEquals(BOTH_256, factory.getLog("package.both256/method1"));
+    assertEquals(BOTH_256, factory.getLog("package.both256/method2"));
+    assertEquals(BOTH_256, factory.getLog("package.both256/method3"));
+
+    assertEquals(
+        new Builder().header(128).msg(128).build(), factory.getLog("package.service1/both128"));
+    // the global config is in effect
+    assertEquals(HEADER_FULL, factory.getLog("package.service1/absent"));
+
+    assertEquals(MSG_FULL, factory.getLog("package.service2/method_messageOnly"));
+    // the global config is in effect
+    assertEquals(HEADER_FULL, factory.getLog("package.service2/absent"));
+  }
+
+  @Test
+  public void configBinLog_multiConfig_noGlobal() throws Exception {
+    FactoryImpl factory = new FactoryImpl(
+        "package.both256/*{h:256;m:256},"
+        + "package.service1/both128{h:128;m:128},"
+        + "package.service2/method_messageOnly{m}");
+    assertEquals(NONE, factory.getLog("otherpackage.service/method"));
+
+    assertEquals(BOTH_256, factory.getLog("package.both256/method1"));
+    assertEquals(BOTH_256, factory.getLog("package.both256/method2"));
+    assertEquals(BOTH_256, factory.getLog("package.both256/method3"));
+
+    assertEquals(
+        new Builder().header(128).msg(128).build(), factory.getLog("package.service1/both128"));
+    // no global config in effect
+    assertEquals(NONE, factory.getLog("package.service1/absent"));
+
+    assertEquals(MSG_FULL, factory.getLog("package.service2/method_messageOnly"));
+    // no global config in effect
+    assertEquals(NONE, factory.getLog("package.service2/absent"));
+  }
+
+  @Test
+  public void configBinLog_ignoreDuplicates_global() throws Exception {
+    FactoryImpl factory = new FactoryImpl("*{h},p.s/m,*{h:256}");
+    // The duplicate
+    assertEquals(HEADER_FULL, factory.getLog("p.other1/m"));
+    assertEquals(HEADER_FULL, factory.getLog("p.other2/m"));
+    // Other
+    assertEquals(BOTH_FULL, factory.getLog("p.s/m"));
+  }
+
+  @Test
+  public void configBinLog_ignoreDuplicates_service() throws Exception {
+    FactoryImpl factory = new FactoryImpl("p.s/*,*{h:256},p.s/*{h}");
+    // The duplicate
+    assertEquals(BOTH_FULL, factory.getLog("p.s/m1"));
+    assertEquals(BOTH_FULL, factory.getLog("p.s/m2"));
+    // Other
+    assertEquals(HEADER_256, factory.getLog("p.other1/m"));
+    assertEquals(HEADER_256, factory.getLog("p.other2/m"));
+  }
+
+  @Test
+  public void configBinLog_ignoreDuplicates_method() throws Exception {
+    FactoryImpl factory = new FactoryImpl("p.s/m,*{h:256},p.s/m{h}");
+    // The duplicate
+    assertEquals(BOTH_FULL, factory.getLog("p.s/m"));
+    // Other
+    assertEquals(HEADER_256, factory.getLog("p.other1/m"));
+    assertEquals(HEADER_256, factory.getLog("p.other2/m"));
+  }
+
+  /** A builder class to make unit test code more readable. */
+  private static final class Builder {
+    int maxHeaderBytes = 0;
+    int maxMessageBytes = 0;
+
+    Builder header(int bytes) {
+      maxHeaderBytes = bytes;
+      return this;
+    }
+
+    Builder msg(int bytes) {
+      maxMessageBytes = bytes;
+      return this;
+    }
+
+    BinaryLog build() {
+      return new BinaryLog(maxHeaderBytes, maxMessageBytes);
+    }
+  }
+}