services: update the binlog message truncation flag (#4508)

The truncation flag tells us whether the msg or meta was truncated.

2 files changed, 170 insertions, 153 deletions

diff --git a/services/src/main/java/io/grpc/services/BinlogHelper.java b/services/src/main/java/io/grpc/services/BinlogHelper.java
index 727208453..ec9c36f24 100644
--- a/services/src/main/java/io/grpc/services/BinlogHelper.java
+++ b/services/src/main/java/io/grpc/services/BinlogHelper.java
@@ -16,10 +16,11 @@
 
 package io.grpc.services;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
 import static io.grpc.services.BinaryLogProvider.BYTEARRAY_MARSHALLER;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import com.google.protobuf.ByteString;
 import com.google.re2j.Matcher;
@@ -48,7 +49,6 @@ import io.grpc.binarylog.GrpcLogEntry;
 import io.grpc.binarylog.GrpcLogEntry.Type;
 import io.grpc.binarylog.Message;
 import io.grpc.binarylog.Metadata.Builder;
-import io.grpc.binarylog.MetadataEntry;
 import io.grpc.binarylog.Peer;
 import io.grpc.binarylog.Peer.PeerType;
 import io.grpc.binarylog.Uint128;
@@ -103,40 +103,34 @@ final class BinlogHelper {
 
     @Override
     void logSendInitialMetadata(Metadata metadata, boolean isServer, CallId callId) {
-      GrpcLogEntry entry = GrpcLogEntry
-          .newBuilder()
+      GrpcLogEntry.Builder entryBuilder = GrpcLogEntry.newBuilder()
           .setType(Type.SEND_INITIAL_METADATA)
           .setLogger(isServer ? GrpcLogEntry.Logger.SERVER : GrpcLogEntry.Logger.CLIENT)
-          .setCallId(callIdToProto(callId))
-          .setMetadata(metadataToProto(metadata, maxHeaderBytes))
-          .build();
-      sink.write(entry);
+          .setCallId(callIdToProto(callId));
+      addMetadataToProto(entryBuilder, metadata, maxHeaderBytes);
+      sink.write(entryBuilder.build());
     }
 
     @Override
     void logRecvInitialMetadata(
         Metadata metadata, boolean isServer, CallId callId, SocketAddress peerSocket) {
-      GrpcLogEntry entry = GrpcLogEntry
-          .newBuilder()
+      GrpcLogEntry.Builder entryBuilder = GrpcLogEntry.newBuilder()
           .setType(Type.RECV_INITIAL_METADATA)
           .setLogger(isServer ? GrpcLogEntry.Logger.SERVER : GrpcLogEntry.Logger.CLIENT)
           .setCallId(callIdToProto(callId))
-          .setPeer(socketToProto(peerSocket))
-          .setMetadata(metadataToProto(metadata, maxHeaderBytes))
-          .build();
-      sink.write(entry);
+          .setPeer(socketToProto(peerSocket));
+      addMetadataToProto(entryBuilder, metadata, maxHeaderBytes);
+      sink.write(entryBuilder.build());
     }
 
     @Override
     void logTrailingMetadata(Metadata metadata, boolean isServer, CallId callId) {
-      GrpcLogEntry entry = GrpcLogEntry
-          .newBuilder()
+      GrpcLogEntry.Builder entryBuilder = GrpcLogEntry.newBuilder()
           .setType(isServer ? Type.SEND_TRAILING_METADATA : Type.RECV_TRAILING_METADATA)
           .setLogger(isServer ? GrpcLogEntry.Logger.SERVER : GrpcLogEntry.Logger.CLIENT)
-          .setCallId(callIdToProto(callId))
-          .setMetadata(metadataToProto(metadata, maxHeaderBytes))
-          .build();
-      sink.write(entry);
+          .setCallId(callIdToProto(callId));
+      addMetadataToProto(entryBuilder, metadata, maxHeaderBytes);
+      sink.write(entryBuilder.build());
     }
 
     @Override
@@ -149,15 +143,12 @@ final class BinlogHelper {
       if (marshaller != BYTEARRAY_MARSHALLER) {
         throw new IllegalStateException("Expected the BinaryLog's ByteArrayMarshaller");
       }
-      byte[] bytes = (byte[]) message;
-      GrpcLogEntry entry = GrpcLogEntry
-          .newBuilder()
+      GrpcLogEntry.Builder entryBuilder = GrpcLogEntry.newBuilder()
           .setType(Type.SEND_MESSAGE)
           .setLogger(isServer ? GrpcLogEntry.Logger.SERVER : GrpcLogEntry.Logger.CLIENT)
-          .setCallId(callIdToProto(callId))
-          .setMessage(messageToProto(bytes, compressed, maxMessageBytes))
-          .build();
-      sink.write(entry);
+          .setCallId(callIdToProto(callId));
+      messageToProto(entryBuilder, (byte[]) message, compressed, maxMessageBytes);
+      sink.write(entryBuilder.build());
     }
 
     @Override
@@ -170,15 +161,13 @@ final class BinlogHelper {
       if (marshaller != BYTEARRAY_MARSHALLER) {
         throw new IllegalStateException("Expected the BinaryLog's ByteArrayMarshaller");
       }
-      byte[] bytes = (byte[]) message;
-      GrpcLogEntry entry = GrpcLogEntry
-          .newBuilder()
+      GrpcLogEntry.Builder entryBuilder = GrpcLogEntry.newBuilder()
           .setType(Type.RECV_MESSAGE)
           .setLogger(isServer ? GrpcLogEntry.Logger.SERVER : GrpcLogEntry.Logger.CLIENT)
-          .setCallId(callIdToProto(callId))
-          .setMessage(messageToProto(bytes, compressed, maxMessageBytes))
-          .build();
-      sink.write(entry);
+          .setCallId(callIdToProto(callId));
+
+      messageToProto(entryBuilder, (byte[]) message, compressed, maxMessageBytes);
+      sink.write(entryBuilder.build());
     }
 
     @Override
@@ -381,7 +370,7 @@ final class BinlogHelper {
      */
     @VisibleForTesting
     FactoryImpl(BinaryLogSink sink, String configurationString) {
-      Preconditions.checkNotNull(sink);
+      checkNotNull(sink, "sink");
       BinlogHelper globalLog = null;
       Map<String, BinlogHelper> perServiceLogs = new HashMap<String, BinlogHelper>();
       Map<String, BinlogHelper> perMethodLogs = new HashMap<String, BinlogHelper>();
@@ -511,7 +500,7 @@ final class BinlogHelper {
    * Returns a {@link Uint128} from a CallId.
    */
   static Uint128 callIdToProto(CallId callId) {
-    Preconditions.checkNotNull(callId);
+    checkNotNull(callId, "callId");
     return Uint128
         .newBuilder()
         .setHigh(callId.hi)
@@ -521,7 +510,7 @@ final class BinlogHelper {
 
   @VisibleForTesting
   static Peer socketToProto(SocketAddress address) {
-    Preconditions.checkNotNull(address);
+    checkNotNull(address, "address");
 
     Peer.Builder builder = Peer.newBuilder();
     if (address instanceof InetSocketAddress) {
@@ -548,44 +537,49 @@ final class BinlogHelper {
   }
 
   @VisibleForTesting
-  static io.grpc.binarylog.Metadata metadataToProto(Metadata metadata, int maxHeaderBytes) {
-    Preconditions.checkNotNull(metadata);
-    Preconditions.checkState(maxHeaderBytes >= 0);
-    Builder builder = io.grpc.binarylog.Metadata.newBuilder();
+  static void addMetadataToProto(
+      GrpcLogEntry.Builder entryBuilder, Metadata metadata, int maxHeaderBytes) {
+    checkNotNull(entryBuilder, "entryBuilder");
+    checkNotNull(metadata, "metadata");
+    checkArgument(maxHeaderBytes >= 0, "maxHeaderBytes must be non negative");
+    Builder metaBuilder = io.grpc.binarylog.Metadata.newBuilder();
     // This code is tightly coupled with Metadata's implementation
-    byte[][] serialized;
+    byte[][] serialized = null;
     if (maxHeaderBytes > 0 && (serialized = InternalMetadata.serialize(metadata)) != null) {
       int written = 0;
       for (int i = 0; i < serialized.length && written < maxHeaderBytes; i += 2) {
         byte[] key = serialized[i];
         byte[] value = serialized[i + 1];
         if (written + key.length + value.length <= maxHeaderBytes) {
-          builder.addEntry(
-              MetadataEntry
-                  .newBuilder()
+          metaBuilder.addEntryBuilder()
                   .setKey(ByteString.copyFrom(key))
-                  .setValue(ByteString.copyFrom(value))
-                  .build());
+                  .setValue(ByteString.copyFrom(value));
           written += key.length;
           written += value.length;
         }
       }
     }
-    return builder.build();
+    // This check must be updated when we add filtering
+    entryBuilder.setTruncated(maxHeaderBytes == 0
+        || (serialized != null && metaBuilder.getEntryCount() < (serialized.length / 2)));
+    entryBuilder.setMetadata(metaBuilder);
   }
 
   @VisibleForTesting
-  static Message messageToProto(byte[] message, boolean compressed, int maxMessageBytes) {
-    Preconditions.checkNotNull(message);
-    Message.Builder builder = Message
+  static void messageToProto(
+      GrpcLogEntry.Builder entryBuilder, byte[] message, boolean compressed, int maxMessageBytes) {
+    checkNotNull(message, "message");
+    checkArgument(maxMessageBytes >= 0, "maxMessageBytes must be non negative");
+    Message.Builder msgBuilder = Message
         .newBuilder()
         .setFlags(flagsForMessage(compressed))
         .setLength(message.length);
     if (maxMessageBytes > 0) {
       int desiredBytes = Math.min(maxMessageBytes, message.length);
-      builder.setData(ByteString.copyFrom(message, 0, desiredBytes));
+      msgBuilder.setData(ByteString.copyFrom(message, 0, desiredBytes));
     }
-    return builder.build();
+    entryBuilder.setMessage(msgBuilder);
+    entryBuilder.setTruncated(maxMessageBytes < message.length);
   }
 
   /**
diff --git a/services/src/test/java/io/grpc/services/BinlogHelperTest.java b/services/src/test/java/io/grpc/services/BinlogHelperTest.java
index c19d0182a..2b400a249 100644
--- a/services/src/test/java/io/grpc/services/BinlogHelperTest.java
+++ b/services/src/test/java/io/grpc/services/BinlogHelperTest.java
@@ -371,83 +371,116 @@ public final class BinlogHelperTest {
   @Test
   public void metadataToProto_empty() throws Exception {
     assertEquals(
-        io.grpc.binarylog.Metadata.getDefaultInstance(),
-        BinlogHelper.metadataToProto(new Metadata(), Integer.MAX_VALUE));
+        GrpcLogEntry.newBuilder()
+            .setMetadata(io.grpc.binarylog.Metadata.getDefaultInstance())
+            .build(),
+        metadataToProtoTestHelper(new Metadata(), Integer.MAX_VALUE));
   }
 
   @Test
   public void metadataToProto() throws Exception {
     assertEquals(
-        io.grpc.binarylog.Metadata
-            .newBuilder()
-            .addEntry(ENTRY_A)
-            .addEntry(ENTRY_B)
-            .addEntry(ENTRY_C)
+        GrpcLogEntry.newBuilder()
+            .setMetadata(
+                io.grpc.binarylog.Metadata
+                    .newBuilder()
+                    .addEntry(ENTRY_A)
+                    .addEntry(ENTRY_B)
+                    .addEntry(ENTRY_C)
+                    .build())
             .build(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, Integer.MAX_VALUE));
+        metadataToProtoTestHelper(nonEmptyMetadata, Integer.MAX_VALUE));
   }
 
   @Test
   public void metadataToProto_truncated() throws Exception {
     // 0 byte limit not enough for any metadata
     assertEquals(
-        io.grpc.binarylog.Metadata.getDefaultInstance(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 0));
+        GrpcLogEntry.newBuilder()
+            .setMetadata(io.grpc.binarylog.Metadata.getDefaultInstance())
+            .setTruncated(true)
+            .build(),
+        metadataToProtoTestHelper(nonEmptyMetadata, 0));
     // not enough bytes for first key value
     assertEquals(
-        io.grpc.binarylog.Metadata.getDefaultInstance(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 9));
+        GrpcLogEntry.newBuilder()
+            .setMetadata(io.grpc.binarylog.Metadata.getDefaultInstance())
+            .setTruncated(true)
+            .build(),
+        metadataToProtoTestHelper(nonEmptyMetadata, 9));
     // enough for first key value
     assertEquals(
-        io.grpc.binarylog.Metadata
-            .newBuilder()
-            .addEntry(ENTRY_A)
+        GrpcLogEntry.newBuilder()
+            .setMetadata(
+                io.grpc.binarylog.Metadata
+                    .newBuilder()
+                    .addEntry(ENTRY_A)
+                    .build())
+            .setTruncated(true)
             .build(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 10));
+        metadataToProtoTestHelper(nonEmptyMetadata, 10));
     // Test edge cases for >= 2 key values
     assertEquals(
-        io.grpc.binarylog.Metadata
-            .newBuilder()
-            .addEntry(ENTRY_A)
+        GrpcLogEntry.newBuilder()
+            .setMetadata(
+                io.grpc.binarylog.Metadata
+                    .newBuilder()
+                    .addEntry(ENTRY_A)
+                    .build())
+            .setTruncated(true)
             .build(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 19));
+        metadataToProtoTestHelper(nonEmptyMetadata, 19));
     assertEquals(
-        io.grpc.binarylog.Metadata
-            .newBuilder()
-            .addEntry(ENTRY_A)
-            .addEntry(ENTRY_B)
+        GrpcLogEntry.newBuilder()
+            .setMetadata(
+                io.grpc.binarylog.Metadata
+                    .newBuilder()
+                    .addEntry(ENTRY_A)
+                    .addEntry(ENTRY_B)
+                    .build())
+            .setTruncated(true)
             .build(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 20));
+        metadataToProtoTestHelper(nonEmptyMetadata, 20));
     assertEquals(
-        io.grpc.binarylog.Metadata
-            .newBuilder()
-            .addEntry(ENTRY_A)
-            .addEntry(ENTRY_B)
+        GrpcLogEntry.newBuilder()
+            .setMetadata(
+                io.grpc.binarylog.Metadata
+                    .newBuilder()
+                    .addEntry(ENTRY_A)
+                    .addEntry(ENTRY_B)
+                    .build())
+            .setTruncated(true)
             .build(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 29));
+        metadataToProtoTestHelper(nonEmptyMetadata, 29));
+    // not truncated: enough for all keys
     assertEquals(
-        io.grpc.binarylog.Metadata
-            .newBuilder()
-            .addEntry(ENTRY_A)
-            .addEntry(ENTRY_B)
-            .addEntry(ENTRY_C)
+        GrpcLogEntry.newBuilder()
+            .setMetadata(
+                io.grpc.binarylog.Metadata
+                    .newBuilder()
+                    .addEntry(ENTRY_A)
+                    .addEntry(ENTRY_B)
+                    .addEntry(ENTRY_C)
+                    .build())
             .build(),
-        BinlogHelper.metadataToProto(nonEmptyMetadata, 30));
+        metadataToProtoTestHelper(nonEmptyMetadata, 30));
   }
 
   @Test
   public void messageToProto() throws Exception {
     byte[] bytes
         = "this is a long message: AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".getBytes(US_ASCII);
-    Message message = BinlogHelper.messageToProto(bytes, false, Integer.MAX_VALUE);
     assertEquals(
-        Message
-            .newBuilder()
-            .setData(ByteString.copyFrom(bytes))
-            .setFlags(0)
-            .setLength(bytes.length)
+        GrpcLogEntry.newBuilder()
+            .setMessage(
+                Message
+                    .newBuilder()
+                    .setData(ByteString.copyFrom(bytes))
+                    .setFlags(0)
+                    .setLength(bytes.length)
+                    .build())
             .build(),
-        message);
+        messageToProtoTestHelper(bytes, false, Integer.MAX_VALUE));
   }
 
   @Test
@@ -455,23 +488,31 @@ public final class BinlogHelperTest {
     byte[] bytes
         = "this is a long message: AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".getBytes(US_ASCII);
     assertEquals(
-        Message
-            .newBuilder()
-            .setFlags(0)
-            .setLength(bytes.length)
+        GrpcLogEntry.newBuilder()
+            .setMessage(
+                Message
+                    .newBuilder()
+                    .setFlags(0)
+                    .setLength(bytes.length)
+                    .build())
+            .setTruncated(true)
             .build(),
-        BinlogHelper.messageToProto(bytes, false, 0));
+        messageToProtoTestHelper(bytes, false, 0));
 
     int limit = 10;
     String truncatedMessage = "this is a ";
     assertEquals(
-        Message
-            .newBuilder()
-            .setData(ByteString.copyFrom(truncatedMessage.getBytes(US_ASCII)))
-            .setFlags(0)
-            .setLength(bytes.length)
+        GrpcLogEntry.newBuilder()
+            .setMessage(
+                Message
+                    .newBuilder()
+                    .setData(ByteString.copyFrom(truncatedMessage.getBytes(US_ASCII)))
+                    .setFlags(0)
+                    .setLength(bytes.length)
+                    .build())
+            .setTruncated(true)
             .build(),
-        BinlogHelper.messageToProto(bytes, false, limit));
+        messageToProtoTestHelper(bytes, false, limit));
   }
 
   @Test
@@ -484,12 +525,10 @@ public final class BinlogHelperTest {
   public void logSendInitialMetadata_server() throws Exception {
     sinkWriterImpl.logSendInitialMetadata(nonEmptyMetadata, IS_SERVER, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        metadataToProtoTestHelper(nonEmptyMetadata, 10).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_INITIAL_METADATA)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMetadata(BinlogHelper.metadataToProto(nonEmptyMetadata, 10))
             .build());
   }
 
@@ -497,12 +536,10 @@ public final class BinlogHelperTest {
   public void logSendInitialMetadata_client() throws Exception {
     sinkWriterImpl.logSendInitialMetadata(nonEmptyMetadata, IS_CLIENT, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        metadataToProtoTestHelper(nonEmptyMetadata, 10).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_INITIAL_METADATA)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMetadata(BinlogHelper.metadataToProto(nonEmptyMetadata, 10))
             .build());
   }
 
@@ -513,13 +550,11 @@ public final class BinlogHelperTest {
     InetSocketAddress socketAddress = new InetSocketAddress(address, port);
     sinkWriterImpl.logRecvInitialMetadata(nonEmptyMetadata, IS_SERVER, CALL_ID, socketAddress);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        metadataToProtoTestHelper(nonEmptyMetadata, 10).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_INITIAL_METADATA)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
             .setPeer(BinlogHelper.socketToProto(socketAddress))
-            .setMetadata(BinlogHelper.metadataToProto(nonEmptyMetadata, 10))
             .build());
   }
 
@@ -530,13 +565,11 @@ public final class BinlogHelperTest {
     InetSocketAddress socketAddress = new InetSocketAddress(address, port);
     sinkWriterImpl.logRecvInitialMetadata(nonEmptyMetadata, IS_CLIENT, CALL_ID, socketAddress);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        metadataToProtoTestHelper(nonEmptyMetadata, 10).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_INITIAL_METADATA)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
             .setPeer(BinlogHelper.socketToProto(socketAddress))
-            .setMetadata(BinlogHelper.metadataToProto(nonEmptyMetadata, 10))
             .build());
   }
 
@@ -544,12 +577,10 @@ public final class BinlogHelperTest {
   public void logTrailingMetadata_server() throws Exception {
     sinkWriterImpl.logTrailingMetadata(nonEmptyMetadata, IS_SERVER, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        metadataToProtoTestHelper(nonEmptyMetadata, 10).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_TRAILING_METADATA)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMetadata(BinlogHelper.metadataToProto(nonEmptyMetadata, 10))
             .build());
   }
 
@@ -557,12 +588,10 @@ public final class BinlogHelperTest {
   public void logTrailingMetadata_client() throws Exception {
     sinkWriterImpl.logTrailingMetadata(nonEmptyMetadata, IS_CLIENT, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        metadataToProtoTestHelper(nonEmptyMetadata, 10).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_TRAILING_METADATA)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMetadata(BinlogHelper.metadataToProto(nonEmptyMetadata, 10))
             .build());
   }
 
@@ -571,24 +600,19 @@ public final class BinlogHelperTest {
     sinkWriterImpl.logOutboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_COMPRESSED, IS_SERVER, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_COMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(BinlogHelper.messageToProto(message, IS_COMPRESSED, MESSAGE_LIMIT))
             .build());
 
     sinkWriterImpl.logOutboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_UNCOMPRESSED, IS_SERVER, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_UNCOMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(
-                BinlogHelper.messageToProto(message, IS_UNCOMPRESSED, MESSAGE_LIMIT))
             .build());
     verifyNoMoreInteractions(sink);
   }
@@ -598,24 +622,19 @@ public final class BinlogHelperTest {
     sinkWriterImpl.logOutboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_COMPRESSED, IS_CLIENT, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_COMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(BinlogHelper.messageToProto(message, IS_COMPRESSED, MESSAGE_LIMIT))
             .build());
 
     sinkWriterImpl.logOutboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_UNCOMPRESSED, IS_CLIENT, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_UNCOMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.SEND_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(
-                BinlogHelper.messageToProto(message, IS_UNCOMPRESSED, MESSAGE_LIMIT))
             .build());
     verifyNoMoreInteractions(sink);
   }
@@ -625,24 +644,19 @@ public final class BinlogHelperTest {
     sinkWriterImpl.logInboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_COMPRESSED, IS_SERVER, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_COMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(BinlogHelper.messageToProto(message, IS_COMPRESSED, MESSAGE_LIMIT))
             .build());
 
     sinkWriterImpl.logInboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_UNCOMPRESSED, IS_SERVER, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_UNCOMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.SERVER)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(
-                BinlogHelper.messageToProto(message, IS_UNCOMPRESSED, MESSAGE_LIMIT))
             .build());
     verifyNoMoreInteractions(sink);
   }
@@ -652,24 +666,19 @@ public final class BinlogHelperTest {
     sinkWriterImpl.logInboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_COMPRESSED, IS_CLIENT, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_COMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(BinlogHelper.messageToProto(message, IS_COMPRESSED, MESSAGE_LIMIT))
             .build());
 
     sinkWriterImpl.logInboundMessage(
         BYTEARRAY_MARSHALLER, message, IS_UNCOMPRESSED, IS_CLIENT, CALL_ID);
     verify(sink).write(
-        GrpcLogEntry
-            .newBuilder()
+        messageToProtoTestHelper(message, IS_UNCOMPRESSED, MESSAGE_LIMIT).toBuilder()
             .setType(GrpcLogEntry.Type.RECV_MESSAGE)
             .setLogger(GrpcLogEntry.Logger.CLIENT)
             .setCallId(BinlogHelper.callIdToProto(CALL_ID))
-            .setMessage(
-                BinlogHelper.messageToProto(message, IS_UNCOMPRESSED, MESSAGE_LIMIT))
             .build());
     verifyNoMoreInteractions(sink);
   }
@@ -968,4 +977,18 @@ public final class BinlogHelperTest {
   private BinlogHelper makeLog(String logConfigStr) {
     return FactoryImpl.createBinaryLog(sink, logConfigStr);
   }
+
+  private static GrpcLogEntry metadataToProtoTestHelper(
+      Metadata metadata, int maxHeaderBytes) {
+    GrpcLogEntry.Builder builder = GrpcLogEntry.newBuilder();
+    BinlogHelper.addMetadataToProto(builder, metadata, maxHeaderBytes);
+    return builder.build();
+  }
+
+  private static GrpcLogEntry messageToProtoTestHelper(
+      byte[] message, boolean compressed, int maxMessageBytes) {
+    GrpcLogEntry.Builder builder = GrpcLogEntry.newBuilder();
+    BinlogHelper.messageToProto(builder, message, compressed, maxMessageBytes);
+    return builder.build();
+  }
 }