RFC: logging: experimental batching writer

google-cloud-logging/pom.xml

@@ -46,6 +46,11 @@
       <groupId>io.grpc</groupId>
       <artifactId>grpc-auth</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.truth</groupId>
+      <artifactId>truth</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>google-cloud-core</artifactId>

google-cloud-logging/src/main/java/com/google/cloud/logging/BatchingWriter.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * 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 com.google.cloud.logging;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutureCallback;
+import com.google.api.core.ApiFutures;
+import com.google.common.base.Preconditions;
+import com.google.logging.v2.WriteLogEntriesRequest;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+class BatchingWriter {
+
+  interface Rpc {
+    ApiFuture<Void> call(WriteLogEntriesRequest request);
+  }
+
+  private final Rpc rpc;
+  private final int batchSize;
+  private final Semaphore pending;
+  private final WriteLogEntriesRequest requestPrototype;
+  private final ScheduledThreadPoolExecutor executor;
+
+  // Use Boolean, since concurrent maps don't support nulls.
+  private final ConcurrentHashMap<ApiFuture<Void>, Boolean> pendingWrites =
+      new ConcurrentHashMap<>();
+
+  private final ArrayList<com.google.logging.v2.LogEntry> currentBatch;
+
+  private Future<?> flushJob = null;
+
+  BatchingWriter(
+      Rpc rpc,
+      int batchSize,
+      int maxPending,
+      WriteLogEntriesRequest requestPrototype,
+      ScheduledThreadPoolExecutor executor) {
+    Preconditions.checkArgument(batchSize > 0, "batchSize must be positive");
+    this.batchSize = batchSize;
+    Preconditions.checkArgument(maxPending > 0, "maxPending must be positive");
+    this.pending = new Semaphore(maxPending);
+
+    this.rpc = Preconditions.checkNotNull(rpc);
+    this.requestPrototype = Preconditions.checkNotNull(requestPrototype);
+    this.executor = Preconditions.checkNotNull(executor);
+    this.currentBatch = new ArrayList<>(batchSize);
+  }
+
+  synchronized void startJob() {
+    Preconditions.checkArgument(flushJob == null, "job already started");
+    flushJob =
+        executor.scheduleWithFixedDelay(
+            new Runnable() {
+              @Override
+              public void run() {
+                initFlush();
+              }
+            },
+            100,
+            100,
+            TimeUnit.MILLISECONDS);
+  }
+
+  synchronized void stopJob() {
+    Preconditions.checkArgument(flushJob != null, "job not started");
+    flushJob.cancel(false);
+  }
+
+  void add(com.google.logging.v2.LogEntry entry) throws InterruptedException {
+    pending.acquire(1);
+    synchronized (currentBatch) {
+      currentBatch.add(entry);
+      if (currentBatch.size() == batchSize) {
+        final WriteLogEntriesRequest request =
+            requestPrototype.toBuilder().addAllEntries(currentBatch).build();
+        currentBatch.clear();
+
+        // Whoever calls send serializes the proto; so we do it off-thread.
+        // This gives better CPU utilization if there are few producer threads
+        // on a many-core machine.
+        executor.execute(
+            new Runnable() {
+              @Override
+              public void run() {
+                send(request);
+              }
+            });
+      }
+    }
+  }
+
+  void initFlush() {
+    WriteLogEntriesRequest request;
+    synchronized (currentBatch) {
+      request = requestPrototype.toBuilder().addAllEntries(currentBatch).build();
+      currentBatch.clear();
+    }
+    send(request);
+  }
+
+  private void send(WriteLogEntriesRequest request) {
+    final int count = request.getEntriesCount();
+    if (count == 0) {
+      return;
+    }
+
+    final ApiFuture<Void> writeFuture = rpc.call(request);
+    pendingWrites.put(writeFuture, Boolean.TRUE);
+    ApiFutures.addCallback(
+        writeFuture,
+        new ApiFutureCallback<Void>() {
+          private void onBoth() {
+            pendingWrites.remove(writeFuture);
+            pending.release(count);
+          }
+
+          @Override
+          public void onSuccess(Void v) {
+            onBoth();
+          }
+
+          @Override
+          public void onFailure(Throwable t) {
+            // Report failure.
+            onBoth();
+          }
+        });
+  }
+
+  List<ApiFuture<Void>> pendingRpcs() {
+    return new ArrayList<>(pendingWrites.keySet());
+  }
+}

google-cloud-logging/src/test/java/com/google/cloud/logging/BatchingWriterTest.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * 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 com.google.cloud.logging;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.SettableApiFuture;
+import com.google.logging.v2.WriteLogEntriesRequest;
+import java.util.List;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class BatchingWriterTest {
+
+  private static final WriteLogEntriesRequest EMPTY_REQUEST =
+      WriteLogEntriesRequest.newBuilder().build();
+  private static final com.google.logging.v2.LogEntry EMPTY_ENTRY =
+      com.google.logging.v2.LogEntry.newBuilder().build();
+  private static ScheduledThreadPoolExecutor EXECUTOR;
+
+  @BeforeClass
+  public static void beforeClass() {
+    // Executor is never used.
+    EXECUTOR = new ScheduledThreadPoolExecutor(1);
+    EXECUTOR.shutdownNow();
+  }
+
+  @Test
+  public void testEmptyWrite() {
+    BatchingWriter.Rpc rpc =
+        new BatchingWriter.Rpc() {
+          @Override
+          public ApiFuture<Void> call(WriteLogEntriesRequest request) {
+            throw new UnsupportedOperationException("should never be called");
+          }
+        };
+
+    BatchingWriter writer = new BatchingWriter(rpc, 10, 10, EMPTY_REQUEST, EXECUTOR);
+    writer.initFlush();
+
+    // If there's no message, there's no RPC.
+    assertThat(writer.pendingRpcs()).isEmpty();
+  }
+
+  @Test
+  public void testFlush() throws Exception {
+    final SettableApiFuture<Void> fakeFuture = SettableApiFuture.create();
+    BatchingWriter.Rpc rpc =
+        new BatchingWriter.Rpc() {
+          @Override
+          public ApiFuture<Void> call(WriteLogEntriesRequest request) {
+            return fakeFuture;
+          }
+        };
+
+    BatchingWriter writer = new BatchingWriter(rpc, 10, 10, EMPTY_REQUEST, EXECUTOR);
+    writer.add(EMPTY_ENTRY);
+    writer.initFlush();
+
+    List<ApiFuture<Void>> futures = writer.pendingRpcs();
+    assertThat(futures).hasSize(1);
+    assertThat(futures.get(0).isDone()).isFalse();
+
+    fakeFuture.set(null);
+    assertThat(futures.get(0).isDone()).isTrue();
+    assertThat(writer.pendingRpcs()).isEmpty();
+  }
+}