[test-only] simulations use tritium metric registry

.gitattributes

@@ -1 +1 @@
-*.png filter=lfs diff=lfs merge=lfs -text
+*.png filter=lfs diff=lfs merge=lfs

build.gradle

@@ -44,10 +44,13 @@ plugins {
 apply plugin: 'com.palantir.baseline'
 apply plugin: 'com.palantir.git-version'
 
+version System.env.CIRCLE_TAG ?: gitVersion()
+
 allprojects {
     apply plugin: 'com.palantir.java-format'
-    version System.env.CIRCLE_TAG ?: gitVersion()
+
     group 'com.palantir.dialogue'
+    version rootProject.version
 
     repositories {
         jcenter()

gradle.properties

@@ -1,3 +1,2 @@
-systemProp.org.gradle.internal.http.socketTimeout=600000
-systemProp.org.gradle.internal.http.connectionTimeout=600000
 org.gradle.parallel=true
+org.gradle.caching=true

settings.gradle

@@ -1,4 +1,5 @@
 rootProject.name = 'dialogue'
+
 include 'dialogue-client-test-lib'
 include 'dialogue-core'
 include 'dialogue-example'
@@ -7,3 +8,10 @@ include 'dialogue-okhttp-client'
 include 'dialogue-serde'
 include 'dialogue-target'
 include 'simulation'
+
+boolean isCiServer = System.getenv().containsKey('CI')
+buildCache {
+    local {
+        enabled = !isCiServer
+    }
+}

simulation/src/main/java/com/palantir/dialogue/core/Benchmark.java

@@ -27,6 +27,7 @@
 import com.palantir.dialogue.Request;
 import com.palantir.dialogue.Response;
 import com.palantir.logsafe.Preconditions;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Random;
@@ -152,6 +153,11 @@ public ListenableFuture<BenchmarkResult> schedule() {
             FutureCallback<Response> accumulateStatusCodes = new FutureCallback<Response>() {
                 @Override
                 public void onSuccess(Response response) {
+                    try {
+                        response.body().close(); // just being a good citizen
+                    } catch (IOException e) {
+                        log.warn("Failed to close body", e);
+                    }
                     statusCodes.compute(Integer.toString(response.code()), (c, num) -> num == null ? 1 : num + 1);
                 }
 
@@ -184,19 +190,27 @@ public void onFailure(Throwable throwable) {
                     TimeUnit.NANOSECONDS);
         });
 
-        benchmarkFinished.getFuture().addListener(simulation.metrics()::report, MoreExecutors.directExecutor());
+        benchmarkFinished.getFuture().addListener(simulation.metricsReporter()::report, MoreExecutors.directExecutor());
 
         return Futures.transform(
                 benchmarkFinished.getFuture(),
-                v -> ImmutableBenchmarkResult.builder()
-                        .clientHistogram(histogramChannel.getHistogram().getSnapshot())
-                        .endTime(Duration.ofNanos(simulation.clock().read()))
-                        .statusCodes(statusCodes)
-                        .successPercentage(
-                                Math.round(statusCodes.getOrDefault("200", 0) * 1000d / requestsStarted[0]) / 10d)
-                        .numSent(requestsStarted[0])
-                        .numReceived(responsesReceived[0])
-                        .build(),
+                v -> {
+                    long numGlobalResponses = MetricNames.globalResponses(simulation.taggedMetrics())
+                            .getCount();
+                    long leaked = numGlobalResponses
+                            - MetricNames.bodyClose(simulation.taggedMetrics()).getCount();
+                    return ImmutableBenchmarkResult.builder()
+                            .clientHistogram(histogramChannel.getHistogram().getSnapshot())
+                            .endTime(Duration.ofNanos(simulation.clock().read()))
+                            .statusCodes(statusCodes)
+                            .successPercentage(
+                                    Math.round(statusCodes.getOrDefault("200", 0) * 1000d / requestsStarted[0]) / 10d)
+                            .numSent(requestsStarted[0])
+                            .numReceived(responsesReceived[0])
+                            .numGlobalResponses(numGlobalResponses)
+                            .bodiesLeaked(leaked)
+                            .build();
+                },
                 MoreExecutors.directExecutor());
     }
 
@@ -218,11 +232,20 @@ interface BenchmarkResult {
 
         Map<String, Integer> statusCodes();
 
+        /** What proportion of responses were 200s. */
         double successPercentage();
 
+        /** How many requests did we fire off in the benchmark. */
         long numSent();
 
+        /** How many responses were returned to the user. */
         long numReceived();
+
+        /** How many responses were issued in total across all servers. */
+        long numGlobalResponses();
+
+        /** How many response bodies were never closed. */
+        long bodiesLeaked();
     }
 
     /**

simulation/src/main/java/com/palantir/dialogue/core/HistogramChannel.java

@@ -37,7 +37,7 @@ final class HistogramChannel implements Channel {
         histogram = new Histogram(new SlidingTimeWindowArrayReservoir(1, TimeUnit.DAYS, simulation.codahaleClock()));
     }
 
-    /** Unit is nanos. */
+    /** Unit is micros. */
     public Histogram getHistogram() {
         return histogram;
     }
@@ -46,7 +46,13 @@ public Histogram getHistogram() {
     public ListenableFuture<Response> execute(Endpoint endpoint, Request request) {
         long start = simulation.clock().read();
         ListenableFuture<Response> future = channel.execute(endpoint, request);
-        future.addListener(() -> histogram.update(simulation.clock().read() - start), MoreExecutors.directExecutor());
+        future.addListener(
+                () -> {
+                    long micros =
+                            TimeUnit.MICROSECONDS.convert(simulation.clock().read() - start, TimeUnit.NANOSECONDS);
+                    histogram.update(micros);
+                },
+                MoreExecutors.directExecutor());
         return future;
     }
 }

simulation/src/main/java/com/palantir/dialogue/core/MetricNames.java

@@ -0,0 +1,59 @@
+/*
+ * (c) Copyright 2020 Palantir Technologies Inc. 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 com.palantir.dialogue.core;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Meter;
+import com.palantir.dialogue.Endpoint;
+import com.palantir.tritium.metrics.registry.MetricName;
+import com.palantir.tritium.metrics.registry.TaggedMetricRegistry;
+
+final class MetricNames {
+    /** Counter incremented every time a {@code Response} body is closed. */
+    static Counter bodyClose(TaggedMetricRegistry reg) {
+        return reg.counter(MetricName.builder().safeName("bodyClose").build());
+    }
+
+    /** Counter for how many responses are issued across all servers. */
+    static Counter globalResponses(TaggedMetricRegistry registry) {
+        return registry.counter(MetricName.builder().safeName("globalResponses").build());
+    }
+
+    /** Counter for how long servers spend processing requests. */
+    static Counter globalServerTimeNanos(TaggedMetricRegistry registry) {
+        return registry.counter(
+                MetricName.builder().safeName("globalServerTime").build());
+    }
+
+    static Counter activeRequests(TaggedMetricRegistry reg, String serverName) {
+        return reg.counter(MetricName.builder()
+                .safeName("activeRequests")
+                .putSafeTags("server", serverName)
+                .build());
+    }
+
+    /** Marked every time a server received a request. */
+    static Meter requestMeter(TaggedMetricRegistry reg, String serverName, Endpoint endpoint) {
+        return reg.meter(MetricName.builder()
+                .safeName("request")
+                .putSafeTags("server", serverName)
+                .putSafeTags("endpoint", endpoint.endpointName())
+                .build());
+    }
+
+    private MetricNames() {}
+}

simulation/src/main/java/com/palantir/dialogue/core/Simulation.java

@@ -35,10 +35,12 @@ final class Simulation {
     private final DeterministicScheduler deterministicExecutor = new DeterministicScheduler();
     private final ListeningScheduledExecutorService listenableExecutor =
             MoreExecutors.listeningDecorator(deterministicExecutor);
+
     private final TestCaffeineTicker ticker = new TestCaffeineTicker();
-    private final SimulationMetrics metrics = new SimulationMetrics(this);
+    private final SimulationMetricsReporter metrics = new SimulationMetricsReporter(this);
     private final CodahaleClock codahaleClock = new CodahaleClock(ticker);
     private final EventMarkers eventMarkers = new EventMarkers(ticker);
+    private final TaggedMetrics taggedMetrics = new TaggedMetrics(codahaleClock);
 
     Simulation() {
         Thread.currentThread().setUncaughtExceptionHandler((t, e) -> log.error("Uncaught throwable", e));
@@ -82,7 +84,11 @@ public CodahaleClock codahaleClock() {
         return codahaleClock;
     }
 
-    public SimulationMetrics metrics() {
+    public TaggedMetrics taggedMetrics() {
+        return taggedMetrics;
+    }
+
+    public SimulationMetricsReporter metricsReporter() {
         return metrics;
     }