Replace GzipSink with DeflateOutputStream

- Add Reporter#getReported
- Fix benachmarking with JDBC mock (still not enabled by default)
- Remove Moshi JSON benchmarks

Author: felixbarny

apm-agent-benchmarks/pom.xml

@@ -60,16 +60,6 @@
             <version>1.4.22.Final</version>
         </dependency>
 
-        <dependency>
-            <groupId>com.squareup.moshi</groupId>
-            <artifactId>moshi</artifactId>
-            <version>1.5.0</version>
-        </dependency>
-        <dependency>
-            <groupId>com.squareup.moshi</groupId>
-            <artifactId>moshi-adapters</artifactId>
-            <version>1.5.0</version>
-        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>

apm-agent-benchmarks/src/main/java/co/elastic/apm/benchmark/ElasticApmContinuousBenchmark.java

@@ -103,6 +103,7 @@ public abstract class ElasticApmContinuousBenchmark extends AbstractBenchmark {
     protected HttpServlet httpServlet;
     private Undertow server;
     private ElasticApmTracer tracer;
+    private long receivedPayloads = 0;
 
     public ElasticApmContinuousBenchmark(boolean apmEnabled) {
         this.apmEnabled = apmEnabled;
@@ -112,7 +113,10 @@ public ElasticApmContinuousBenchmark(boolean apmEnabled) {
     public void setUp(Blackhole blackhole) throws SQLException {
         server = Undertow.builder()
             .addHttpListener(0, "127.0.0.1")
-            .setHandler(exchange -> exchange.setStatusCode(200).endExchange()).build();
+            .setHandler(exchange -> {
+                receivedPayloads++;
+                exchange.setStatusCode(200).endExchange();
+            }).build();
         server.start();
         int port = ((InetSocketAddress) server.getListenerInfo().get(0).getAddress()).getPort();
         tracer = new ElasticApmTracerBuilder()
@@ -143,6 +147,9 @@ public void tearDown() {
         server.stop();
         tracer.stop();
         ElasticApmAgent.reset();
+        System.out.println("Reported: " + tracer.getReporter().getReported());
+        System.out.println("Dropped: " + tracer.getReporter().getDropped());
+        System.out.println("receivedPayloads = " + receivedPayloads);
     }
 
     private MockHttpServletRequest createRequest() {
@@ -236,7 +243,7 @@ public void doGet(HttpServletRequest request, HttpServletResponse response) thro
                 // makes sure the jdbc query and the reporting can't be eliminated by JIT
                 // setting it as the http status code so that there are no allocations necessary
                 // for example converting to string
-                response.setStatus(reporter.getDropped());
+                response.setStatus((int) reporter.getDropped());
             } catch (Exception e) {
                 throw new ServletException(e);
             }

apm-agent-benchmarks/src/main/java/co/elastic/apm/benchmark/reporter/AbstractHttpJacksonReporterBenchmark.java

@@ -24,7 +24,6 @@
 import co.elastic.apm.report.serialize.PayloadSerializer;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.module.afterburner.AfterburnerModule;
-import okio.Buffer;
 import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.profile.GCProfiler;
 import org.openjdk.jmh.runner.Runner;
@@ -51,9 +50,6 @@ public static void main(String[] args) throws RunnerException {
     @Setup
     public void setUp() throws Exception {
         super.setUp();
-        Buffer buffer = new Buffer();
-        getPayloadSerializer().serializePayload(buffer, payload);
-        System.out.println("Size of payload in bytes: " + buffer.readByteString().toByteArray().length);
     }
 
     @Override

apm-agent-benchmarks/src/main/java/co/elastic/apm/benchmark/reporter/AbstractHttpReporterBenchmark.java

@@ -25,11 +25,6 @@
 import co.elastic.apm.report.serialize.PayloadSerializer;
 import io.undertow.Undertow;
 import okhttp3.OkHttpClient;
-import okio.Buffer;
-import okio.BufferedSink;
-import okio.ByteString;
-import okio.Source;
-import okio.Timeout;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.TearDown;
@@ -39,18 +34,17 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
-import java.nio.charset.Charset;
 
 public abstract class AbstractHttpReporterBenchmark extends AbstractReporterBenchmark {
     private Undertow server;
     private int port;
     private PayloadSerializer payloadSerializer;
-    private NoopBufferedSink noopBufferedSink;
+    private BlackholeOutputStream blackholeOutputStream;
 
     @Setup
     public void setUp(Blackhole blackhole) throws Exception {
         super.setUp();
-        noopBufferedSink = new NoopBufferedSink(blackhole);
+        blackholeOutputStream = new BlackholeOutputStream(blackhole);
         server = Undertow.builder()
             .addHttpListener(0, "127.0.0.1")
             .setHandler(exchange -> exchange.setStatusCode(200).endExchange()).build();
@@ -80,185 +74,22 @@ public void tearDown() {
     @Benchmark
     @Threads(1)
     public void testSerialization() throws IOException {
-        payloadSerializer.serializePayload(noopBufferedSink, payload);
+        payloadSerializer.serializePayload(blackholeOutputStream, payload);
     }
 
-    @SuppressWarnings("ConstantConditions")
-    private static class NoopBufferedSink implements BufferedSink {
+    private static class BlackholeOutputStream extends OutputStream {
 
         private final Blackhole blackhole;
 
-        private NoopBufferedSink(Blackhole blackhole) {
+        private BlackholeOutputStream(Blackhole blackhole) {
             this.blackhole = blackhole;
         }
 
         @Override
-        public Buffer buffer() {
-            return null;
-        }
-
-        @Override
-        public BufferedSink write(ByteString byteString) throws IOException {
-            blackhole.consume(byteString);
-            return this;
-        }
-
-        @Override
-        public BufferedSink write(byte[] source) throws IOException {
-            blackhole.consume(source);
-            return this;
-        }
-
-        @Override
-        public BufferedSink write(byte[] source, int offset, int byteCount) throws IOException {
-            blackhole.consume(source);
-            return this;
-        }
-
-        @Override
-        public long writeAll(Source source) throws IOException {
-            blackhole.consume(source);
-            return 0;
-        }
-
-        @Override
-        public BufferedSink write(Source source, long byteCount) throws IOException {
-            blackhole.consume(source);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeUtf8(String string) throws IOException {
-            blackhole.consume(string);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeUtf8(String string, int beginIndex, int endIndex) throws IOException {
-            blackhole.consume(string);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeUtf8CodePoint(int codePoint) throws IOException {
-            blackhole.consume(codePoint);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeString(String string, Charset charset) throws IOException {
-            blackhole.consume(string);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeString(String string, int beginIndex, int endIndex, Charset charset) throws IOException {
-            blackhole.consume(string);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeByte(int b) throws IOException {
+        public void write(int b) {
             blackhole.consume(b);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeShort(int s) throws IOException {
-            blackhole.consume(s);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeShortLe(int s) throws IOException {
-            blackhole.consume(s);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeInt(int i) throws IOException {
-            blackhole.consume(i);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeIntLe(int i) throws IOException {
-            blackhole.consume(i);
-            return this;
         }
 
-        @Override
-        public BufferedSink writeLong(long v) throws IOException {
-            blackhole.consume(v);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeLongLe(long v) throws IOException {
-            blackhole.consume(v);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeDecimalLong(long v) throws IOException {
-            blackhole.consume(v);
-            return this;
-        }
-
-        @Override
-        public BufferedSink writeHexadecimalUnsignedLong(long v) throws IOException {
-            blackhole.consume(v);
-            return this;
-        }
-
-        @Override
-        public void flush() throws IOException {
-
-        }
-
-        @Override
-        public BufferedSink emit() throws IOException {
-            return this;
-        }
-
-        @Override
-        public BufferedSink emitCompleteSegments() throws IOException {
-            return this;
-        }
-
-        @Override
-        public OutputStream outputStream() {
-            return new OutputStream() {
-                @Override
-                public void write(int b) throws IOException {
-                    blackhole.consume(b);
-                }
-
-                @Override
-                public void write(byte[] b) throws IOException {
-                    blackhole.consume(b);
-                }
-
-                @Override
-                public void write(byte[] b, int off, int len) throws IOException {
-                    blackhole.consume(b);
-                }
-            };
-        }
-
-        @Override
-        public void write(Buffer source, long byteCount) throws IOException {
-            blackhole.consume(source);
-        }
-
-        @Override
-        public Timeout timeout() {
-            return null;
-        }
-
-        @Override
-        public void close() throws IOException {
-
-        }
     }
+
 }

apm-agent-benchmarks/src/main/java/co/elastic/apm/benchmark/reporter/HttpMoshiReporterBenchmark.java

@@ -36,9 +36,9 @@ public static void main(String[] args) throws RunnerException {
 
     @Override
     protected PayloadSerializer getPayloadSerializer() {
-        return (sink, payload) -> {
-            sink.writeByte('{');
-            sink.writeByte('}');
+        return (os, payload) -> {
+            os.write('{');
+            os.write('}');
         };
     }
 }

apm-agent-benchmarks/src/main/java/co/elastic/apm/benchmark/reporter/HttpNoopJsonReporterBenchmark.java

@@ -40,6 +40,21 @@ public static void main(String[] args) throws RunnerException {
     }
 
     protected PayloadSender getPayloadSender() {
-        return Payload::recycle;
+        return new PayloadSender() {
+            @Override
+            public void sendPayload(Payload payload) {
+                payload.recycle();
+            }
+
+            @Override
+            public long getReported() {
+                return 0;
+            }
+
+            @Override
+            public long getDropped() {
+                return 0;
+            }
+        };
     }
 }