Pool reader buffers

pom.xml

@@ -6,7 +6,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.pinterest.singer</groupId>
     <artifactId>singer-package</artifactId>
-    <version>0.8.0.86</version>
+    <version>0.8.0.87</version>
     <packaging>pom</packaging>
     <description>Singer Logging Agent modules</description>
     <inceptionYear>2013</inceptionYear>

singer-commons/pom.xml

@@ -19,7 +19,7 @@
   <parent>
     <groupId>com.pinterest.singer</groupId>
     <artifactId>singer-package</artifactId>
-    <version>0.8.0.86</version>
+    <version>0.8.0.87</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
   <developers>

singer-commons/src/main/thrift/config.thrift

@@ -439,4 +439,9 @@ struct SingerConfig {
    */
   28: optional string hostnamePrefixRegex = "-";
 
+  /**
+  * Initialize LogStreamReaders with pooled buffers
+  */
+  29: optional bool enablePooledReaderBuffers = false;
+
 }

singer/pom.xml

@@ -4,10 +4,13 @@
     <packaging>jar</packaging>
     <description>Logging Agent</description>
     <inceptionYear>2013</inceptionYear>
+    <properties>
+        <netty.version>4.1.75.Final</netty.version>
+    </properties>
     <parent>
         <groupId>com.pinterest.singer</groupId>
         <artifactId>singer-package</artifactId>
-        <version>0.8.0.86</version>
+        <version>0.8.0.87</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <licenses>
@@ -87,6 +90,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>${netty.version}</version>
+        </dependency>
         <dependency>
             <groupId>commons-cli</groupId>
             <artifactId>commons-cli</artifactId>

singer/src/main/java/com/pinterest/singer/monitor/DefaultLogMonitor.java

@@ -58,6 +58,8 @@
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.twitter.ostrich.stats.Stats;
+import io.netty.buffer.PoolArenaMetric;
+import io.netty.buffer.PooledByteBufAllocator;
 import org.apache.commons.configuration.ConfigurationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -559,6 +561,15 @@ public void reportStats() {
     Map<String, Integer> perLogStuck = Maps.newHashMap();
     DefaultLogStreamProcessor processor;
 
+    OpenTsdbMetricConverter.gauge("singer.netty_heap_buffer.memory.used",
+        PooledByteBufAllocator.DEFAULT.metric().usedHeapMemory());
+    OpenTsdbMetricConverter.gauge("singer.netty_heap_buffer.active.allocation.bytes", PooledByteBufAllocator.DEFAULT.metric().heapArenas().stream()
+        .mapToLong(PoolArenaMetric::numActiveBytes).sum());
+    OpenTsdbMetricConverter.gauge("singer.netty_direct_buffer.memory.used",
+        PooledByteBufAllocator.DEFAULT.metric().usedDirectMemory());
+    OpenTsdbMetricConverter.gauge("singer.netty_direct_buffer.active.allocation.bytes", PooledByteBufAllocator.DEFAULT.metric().directArenas().stream()
+        .mapToLong(PoolArenaMetric::numActiveBytes).sum());
+
     for (LogStream logStream : processedLogStreams.keySet()) {
       String logName = logStream.getSingerLog().getSingerLogConfig().getName();
       if (!perLogLatency.containsKey(logName)) {

singer/src/main/java/com/pinterest/singer/reader/ByteOffsetInputStream.java

@@ -26,7 +26,7 @@
  * The byteOffset will be from the beginning of the file.
  * Note that the file can be appended as it is being read.
  */
-public final class ByteOffsetInputStream extends BufferedInputStream {
+public final class ByteOffsetInputStream extends BufferedInputStream implements OffsetInputStream {
 
   private final RandomAccessFile randomAccessFile;
   // Byte offset of current read position.

singer/src/main/java/com/pinterest/singer/reader/OffsetInputStream.java

@@ -0,0 +1,13 @@
+package com.pinterest.singer.reader;
+
+import java.io.IOException;
+
+/**
+ * Interface for InputStream that can track and set the byte offset for the next read.
+ * byteOffset will start from the beginning of the file.
+ */
+public interface OffsetInputStream {
+  long getByteOffset();
+  void setByteOffset(long byteOffset) throws IOException;
+  boolean isEOF() throws IOException;
+}

singer/src/main/java/com/pinterest/singer/reader/RandomAccessFileInputStream.java

@@ -28,7 +28,7 @@
  * This class will allow us to set the file pointer when we need to go back in the stream. It
  * also allow us to track the byte offset of the next thrift message by reading the file pointer.
  */
-final class RandomAccessFileInputStream extends InputStream {
+public final class RandomAccessFileInputStream extends InputStream {
 
   private final RandomAccessFile file;
 

singer/src/main/java/com/pinterest/singer/reader/ThriftReader.java

@@ -15,8 +15,12 @@
  */
 package com.pinterest.singer.reader;
 
+import com.pinterest.singer.common.SingerSettings;
+import com.pinterest.singer.reader.pooled.PooledByteBufInputStream;
+
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
+import io.netty.buffer.PooledByteBufAllocator;
 import org.apache.thrift.TBase;
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TProtocol;
@@ -26,6 +30,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.RandomAccessFile;
 
 /**
@@ -58,7 +63,7 @@ public static interface TProtocolFactory {
   private final TBaseFactory<T> baseFactory;
 
   // The ByteOffsetInputStream to read from.
-  private final ByteOffsetInputStream byteOffsetInputStream;
+  private final OffsetInputStream byteOffsetInputStream;
 
   // The framed framedTransport.
   private final TFramedTransport framedTransport;
@@ -75,9 +80,14 @@ public ThriftReader(
     Preconditions.checkArgument(!Strings.isNullOrEmpty(path));
     Preconditions.checkNotNull(protocolFactory);
 
-    this.byteOffsetInputStream = new ByteOffsetInputStream(
-        new RandomAccessFile(path, "r"), readBufferSize);
-    this.framedTransport = new TFramedTransport(new TIOStreamTransport(this
+    if (SingerSettings.getSingerConfig() != null && SingerSettings.getSingerConfig().isEnablePooledReaderBuffers()) {
+      byteOffsetInputStream = new PooledByteBufInputStream(new RandomAccessFile(path, "r"),
+          PooledByteBufAllocator.DEFAULT.directBuffer(readBufferSize, readBufferSize));
+    } else {
+      byteOffsetInputStream =
+          new ByteOffsetInputStream(new RandomAccessFile(path, "r"), readBufferSize);
+    }
+    this.framedTransport = new TFramedTransport(new TIOStreamTransport((InputStream) this
         .byteOffsetInputStream), maxMessageSize);
     this.baseFactory = Preconditions.checkNotNull(baseFactory);
     this.protocol = protocolFactory.get(this.framedTransport);