[test-only] simulations use tritium metric registry (#358)

Simulation metrics are now tracked in a tritium TaggedMetricRegistry implementation, which is hooked up to the simulation clock.

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

@@ -142,7 +142,8 @@ public BenchmarkResult run() {
 
     @SuppressWarnings("FutureReturnValueIgnored")
     public ListenableFuture<BenchmarkResult> schedule() {
-        HistogramChannel histogramChannel = new HistogramChannel(simulation, channelUnderTest);
+        DialogueClientMetrics clientMetrics = DialogueClientMetrics.of(simulation.taggedMetrics());
+        InstrumentedChannel channel = new InstrumentedChannel(channelUnderTest, clientMetrics);
 
         long[] requestsStarted = {0};
         long[] responsesReceived = {0};
@@ -152,6 +153,7 @@ public ListenableFuture<BenchmarkResult> schedule() {
             FutureCallback<Response> accumulateStatusCodes = new FutureCallback<Response>() {
                 @Override
                 public void onSuccess(Response response) {
+                    response.close(); // just being a good citizen
                     statusCodes.compute(Integer.toString(response.code()), (c, num) -> num == null ? 1 : num + 1);
                 }
 
@@ -166,7 +168,7 @@ public void onFailure(Throwable throwable) {
                     () -> {
                         log.debug(
                                 "time={} starting num={} {}", simulation.clock().read(), req.number(), req);
-                        ListenableFuture<Response> future = histogramChannel.execute(req.endpoint(), req.request());
+                        ListenableFuture<Response> future = channel.execute(req.endpoint(), req.request());
                         requestsStarted[0] += 1;
 
                         Futures.addCallback(future, accumulateStatusCodes, MoreExecutors.directExecutor());
@@ -184,19 +186,30 @@ 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.responseClose(simulation.taggedMetrics())
+                                    .getCount();
+                    return ImmutableBenchmarkResult.builder()
+                            .clientHistogram(clientMetrics
+                                    .response(SimulationUtils.SERVICE_NAME)
+                                    .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)
+                            .responsesLeaked(leaked)
+                            .build();
+                },
                 MoreExecutors.directExecutor());
     }
 
@@ -218,11 +231,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 responses were never closed. */
+        long responsesLeaked();
     }
 
     /**

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} is closed. */
+    static Counter responseClose(TaggedMetricRegistry reg) {
+        return reg.counter(MetricName.builder().safeName("responseClose").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;
     }