Get rid of metrics-transmitter (#20)

Author: artem-v

metrics-examples/src/main/java/io/scalecube/metrics/mimir/MimirAllInOneAeron.java

@@ -30,7 +30,6 @@
 import io.scalecube.metrics.HistogramMetric;
 import io.scalecube.metrics.MetricsReaderAgent;
 import io.scalecube.metrics.MetricsRecorder;
-import io.scalecube.metrics.MetricsTransmitter;
 import io.scalecube.metrics.TpsMetric;
 import io.scalecube.metrics.aeron.CncCountersReaderAgent;
 import java.time.Duration;
@@ -82,8 +81,6 @@ public static void main(String[] args) throws Exception {
     System.out.println("Grafana is available at: " + grafanaUrl);
 
     final var metricsRecorder = MetricsRecorder.launch();
-    final var metricsTransmitter = MetricsTransmitter.launch();
-
     final var countersRegistry = CountersRegistry.create();
     final var countersManager = countersRegistry.countersManager();
     final var sessionCounter = countersManager.newCounter("session_count");
@@ -159,7 +156,8 @@ public static void main(String[] args) throws Exception {
                 new CountersMimirHandler(null, mimirPublisher.proxy())),
             new MetricsReaderAgent(
                 "MetricsReaderAgent",
-                metricsTransmitter.context().broadcastBuffer(),
+                metricsRecorder.context().metricsDir(),
+                true,
                 SystemEpochClock.INSTANCE,
                 Duration.ofSeconds(3),
                 new MetricsMimirHandler(null, mimirPublisher.proxy())));

metrics-examples/src/main/java/io/scalecube/metrics/mimir/MimirAllInOneHistogram.java

@@ -6,7 +6,6 @@
 import io.scalecube.metrics.MetricsHandler;
 import io.scalecube.metrics.MetricsReaderAgent;
 import io.scalecube.metrics.MetricsRecorder;
-import io.scalecube.metrics.MetricsTransmitter;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
@@ -68,7 +67,6 @@ public static void main(String[] args) {
     System.out.println("Grafana is available at: " + grafanaUrl);
 
     final var metricsRecorder = MetricsRecorder.launch();
-    final var metricsTransmitter = MetricsTransmitter.launch();
 
     AgentRunner.startOnThread(
         new AgentRunner(
@@ -77,7 +75,8 @@ public static void main(String[] args) {
             null,
             new MetricsReaderAgent(
                 "MetricsReaderAgent",
-                metricsTransmitter.context().broadcastBuffer(),
+                metricsRecorder.context().metricsDir(),
+                true,
                 SystemEpochClock.INSTANCE,
                 Duration.ofSeconds(3),
                 new MimirHistogramHandler(pushUrl))));

metrics-examples/src/main/java/io/scalecube/metrics/mimir/MimirAllInOnePercentiles.java

@@ -4,7 +4,6 @@
 
 import io.scalecube.metrics.MetricsReaderAgent;
 import io.scalecube.metrics.MetricsRecorder;
-import io.scalecube.metrics.MetricsTransmitter;
 import java.time.Duration;
 import org.agrona.concurrent.AgentRunner;
 import org.agrona.concurrent.BackoffIdleStrategy;
@@ -49,7 +48,6 @@ public static void main(String[] args) {
     System.out.println("Grafana is available at: " + grafanaUrl);
 
     final var metricsRecorder = MetricsRecorder.launch();
-    final var metricsTransmitter = MetricsTransmitter.launch();
     final var mimirPublisher = MimirPublisher.launch(new MimirPublisher.Context().url(pushUrl));
 
     // Metrics
@@ -61,7 +59,8 @@ public static void main(String[] args) {
             null,
             new MetricsReaderAgent(
                 "MetricsReaderAgent",
-                metricsTransmitter.context().broadcastBuffer(),
+                metricsRecorder.context().metricsDir(),
+                true,
                 SystemEpochClock.INSTANCE,
                 Duration.ofSeconds(3),
                 new MetricsMimirHandler(null, mimirPublisher.proxy()))));

metrics-examples/src/main/java/io/scalecube/metrics/prometheus/PrometheusAllInOneAeron.java

@@ -30,7 +30,6 @@
 import io.scalecube.metrics.HistogramMetric;
 import io.scalecube.metrics.MetricsReaderAgent;
 import io.scalecube.metrics.MetricsRecorder;
-import io.scalecube.metrics.MetricsTransmitter;
 import io.scalecube.metrics.TpsMetric;
 import io.scalecube.metrics.aeron.CncCountersReaderAgent;
 import java.io.IOException;
@@ -80,7 +79,6 @@ public static void main(String[] args) throws IOException, InterruptedException
     System.out.println("Grafana: " + grafanaUrl);
 
     final var metricsRecorder = MetricsRecorder.launch();
-    final var metricsTransmitter = MetricsTransmitter.launch();
 
     final var highestTrackableValue = (long) 1e9;
     final var conversionFactor = 1e-3;
@@ -185,7 +183,8 @@ public static void main(String[] args) throws IOException, InterruptedException
                 cncCountersAdapter),
             new MetricsReaderAgent(
                 "MetricsReaderAgent",
-                metricsTransmitter.context().broadcastBuffer(),
+                metricsRecorder.context().metricsDir(),
+                true,
                 SystemEpochClock.INSTANCE,
                 Duration.ofSeconds(3),
                 metricsAdapter));

metrics-examples/src/main/java/io/scalecube/metrics/prometheus/PrometheusNativeHistogram.java

@@ -8,7 +8,6 @@
 import io.scalecube.metrics.MetricsHandler;
 import io.scalecube.metrics.MetricsReaderAgent;
 import io.scalecube.metrics.MetricsRecorder;
-import io.scalecube.metrics.MetricsTransmitter;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
@@ -69,7 +68,6 @@ public static void main(String[] args) throws IOException {
     System.out.println(Instant.now() + " | Server started on " + socketAddress);
 
     final var metricsRecorder = MetricsRecorder.launch();
-    final var metricsTransmitter = MetricsTransmitter.launch();
 
     AgentRunner.startOnThread(
         new AgentRunner(
@@ -78,7 +76,8 @@ public static void main(String[] args) throws IOException {
             null,
             new MetricsReaderAgent(
                 "MetricsReaderAgent",
-                metricsTransmitter.context().broadcastBuffer(),
+                metricsRecorder.context().metricsDir(),
+                true,
                 SystemEpochClock.INSTANCE,
                 Duration.ofSeconds(3),
                 metricsHandlerAdapter)));
@@ -88,6 +87,7 @@ public static void main(String[] args) throws IOException {
     final var highestTrackableValue = (long) 1e9;
     final var conversionFactor = 1e-3;
     final var resolutionMs = 1000;
+
     final var latencyMetric =
         metricsRecorder.newHistogram(
             keyFlyweight -> keyFlyweight.tagsCount(1).stringValue("name", "hft_latency"),

metrics-mimir/src/main/java/io/scalecube/metrics/mimir/MimirPublisher.java

@@ -150,7 +150,7 @@ private void conclude() {
       }
 
       if (errorHandler == null) {
-        errorHandler = ex -> LOGGER.error("Exception occurred: ", ex);
+        errorHandler = ex -> LOGGER.error("Exception occurred", ex);
       }
 
       if (idleStrategy == null) {

metrics/src/main/java/io/scalecube/metrics/CountersRegistry.java

@@ -170,14 +170,10 @@ private void concludeCountersBuffers() {
         throw new IllegalArgumentException("countersValuesBufferLength must be power of 2");
       }
 
-      final var countersMetaDataBufferLength =
-          LayoutDescriptor.countersMetaDataBufferLength(countersValuesBufferLength);
-      final var countersFileLength =
-          LayoutDescriptor.HEADER_LENGTH
-              + countersMetaDataBufferLength
-              + countersValuesBufferLength;
-
-      mappedByteBuffer = mapNewFile(new File(countersDir, COUNTERS_FILE), countersFileLength);
+      mappedByteBuffer =
+          mapNewFile(
+              new File(countersDir, COUNTERS_FILE),
+              LayoutDescriptor.countersFileLength(countersValuesBufferLength));
 
       final var headerBuffer = LayoutDescriptor.createHeaderBuffer(mappedByteBuffer);
       final var startTimestamp = ManagementFactory.getRuntimeMXBean().getStartTime();
@@ -274,6 +270,12 @@ public static class LayoutDescriptor {
 
     private LayoutDescriptor() {}
 
+    public static int countersFileLength(int countersValuesBufferLength) {
+      return LayoutDescriptor.HEADER_LENGTH
+          + LayoutDescriptor.countersMetaDataBufferLength(countersValuesBufferLength)
+          + countersValuesBufferLength;
+    }
+
     public static int countersMetaDataBufferLength(int countersValuesBufferLength) {
       return countersValuesBufferLength * (METADATA_LENGTH / COUNTER_LENGTH);
     }

metrics/src/main/java/io/scalecube/metrics/HistogramAggregate.java

@@ -2,9 +2,9 @@
 
 import static io.scalecube.metrics.HistogramRecorder.NUMBER_OF_SIGNIFICANT_VALUE_DIGITS;
 
-import io.scalecube.metrics.MetricsRecorder.MetricsPublication;
 import org.HdrHistogram.Histogram;
 import org.agrona.DirectBuffer;
+import org.agrona.concurrent.broadcast.BroadcastTransmitter;
 
 class HistogramAggregate {
 
@@ -13,7 +13,7 @@ class HistogramAggregate {
   private final double conversionFactor;
   private final long resolutionMs;
   private final MetricsEncoder encoder;
-  private final MetricsPublication metricsPublication;
+  private final BroadcastTransmitter metricsTransmitter;
 
   private final Histogram accumulated;
   private final Histogram distinct;
@@ -24,13 +24,13 @@ class HistogramAggregate {
       double conversionFactor,
       long resolutionMs,
       MetricsEncoder encoder,
-      MetricsPublication metricsPublication) {
+      BroadcastTransmitter metricsTransmitter) {
     this.keyBuffer = keyBuffer;
     this.highestTrackableValue = highestTrackableValue;
     this.conversionFactor = conversionFactor;
     this.resolutionMs = resolutionMs;
     this.encoder = encoder;
-    this.metricsPublication = metricsPublication;
+    this.metricsTransmitter = metricsTransmitter;
     accumulated = new Histogram(highestTrackableValue, NUMBER_OF_SIGNIFICANT_VALUE_DIGITS);
     distinct = new Histogram(highestTrackableValue, NUMBER_OF_SIGNIFICANT_VALUE_DIGITS);
   }
@@ -49,7 +49,7 @@ void publish(long timestamp) {
       final var length =
           encoder.encodeHistogram(
               timestamp, keyBuffer, accumulated, distinct, highestTrackableValue, conversionFactor);
-      metricsPublication.publish(encoder.buffer(), 0, length);
+      metricsTransmitter.transmit(1, encoder.buffer(), 0, length);
     } finally {
       distinct.reset();
     }