Metrics (#9)

* metrics

* fixed imports

* fixed imports

* make the names prometheus friendly

* cleanup, and refactoring

* metrics

* fixed imports

* fixed imports

* make the names prometheus friendly

* cleanup, and refactoring

Author: robertroeser

build.gradle

@@ -83,6 +83,22 @@ project(':metrics-idl') {
     apply from: file('../gradle/java.gradle')
 }
 
+project(':metrics-micrometer') {
+    description = 'Netifi Proteus Micrometer Support'
+    ext.artifactName = 'metrics-prometheus'
+
+    apply from: file('../gradle/release.gradle')
+    apply from: file('../gradle/java.gradle')
+}
+
+project(':metrics-prometheus') {
+    description = 'Netifi Proteus Prometheus Integration'
+    ext.artifactName = 'metrics-prometheus'
+
+    apply from: file('../gradle/release.gradle')
+    apply from: file('../gradle/java.gradle')
+}
+
 project(':protobuf-rpc') {
     description = 'Netifi Proteus RPC'
     ext.artifactName = 'protobuf-rpc'
@@ -92,7 +108,7 @@ project(':protobuf-rpc') {
 
 
 project(':tracing-openzipkin') {
-    description = 'Netifi Openzipkin Integration'
+    description = 'Netifi Proteus Openzipkin Integration'
     ext.artifactName = 'tracing'
 
     apply from: file('../gradle/release.gradle')
@@ -101,7 +117,7 @@ project(':tracing-openzipkin') {
 
 
 project(':tracing-idl') {
-    description = 'Netifi Tracing IDL'
+    description = 'Netifi Proteus Tracing IDL'
     ext.artifactName = 'tracing-idl'
 
     apply from: file('../gradle/release.gradle')

client/build.gradle

@@ -15,7 +15,7 @@ dependencies {
     compile 'com.google.guava:guava:22.0'
 
     testProtobuf project (':core-idl')
-
+    testProtobuf project (':protobuf-rpc')
     testCompile 'io.micrometer:micrometer-registry-atlas:1.0.3'
 }
 

client/src/main/java/io/netifi/proteus/DefaultProteusBrokerService.java

@@ -26,9 +26,11 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.MonoProcessor;
+import reactor.retry.Retry;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.time.Duration;
 import java.util.*;
 import java.util.function.Function;
 import java.util.function.Supplier;

client/src/main/java/io/netifi/proteus/Proteus.java

@@ -1,10 +1,6 @@
 package io.netifi.proteus;
 
 import io.micrometer.core.instrument.MeterRegistry;
-import io.netifi.proteus.metrics.ProteusMetricsExporter;
-import io.netifi.proteus.metrics.ProteusOperatingSystemMetrics;
-import io.netifi.proteus.metrics.om.MetricsSnapshotHandler;
-import io.netifi.proteus.metrics.om.MetricsSnapshotHandlerClient;
 import io.netifi.proteus.rsocket.ProteusSocket;
 import io.netifi.proteus.rsocket.RequestHandlingRSocket;
 import io.netty.buffer.ByteBuf;
@@ -20,7 +16,6 @@
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
-import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -37,6 +32,7 @@ public class Proteus implements Closeable {
     java.security.Security.setProperty("networkaddress.cache.ttl", "60");
   }
 
+  private final long accesskey;
   private final String fromGroup;
   private final DestinationNameFactory destinationNameFactory;
   private final ProteusBrokerService brokerService;
@@ -56,6 +52,7 @@ private Proteus(
       Function<SocketAddress, ClientTransport> clientTransportFactory,
       int poolSize,
       Supplier<Tracer> tracerSupplier) {
+    this.accesskey = accessKey;
     this.onClose = MonoProcessor.create();
     this.fromGroup = group;
     this.requestHandlingRSocket = new RequestHandlingRSocket();
@@ -109,15 +106,19 @@ public ProteusSocket destination(String destination, String group) {
   public ProteusSocket group(String group) {
     return brokerService.group(group);
   }
-  
+
+  public long getAccesskey() {
+    return accesskey;
+  }
+
   public String getGroupName() {
     return fromGroup;
   }
-  
+
   public String getDestination() {
     return destinationNameFactory.rootName();
   }
-  
+
   public static class Builder {
     private String host = DefaultBuilderConfig.getHost();
     private Integer port = DefaultBuilderConfig.getPort();
@@ -134,10 +135,7 @@ public static class Builder {
     private DestinationNameFactory destinationNameFactory;
 
     private MeterRegistry registry = null;
-    private String metricHandlerGroup = DefaultBuilderConfig.getMetricHandlerGroup();
     private int batchSize = DefaultBuilderConfig.getBatchSize();
-    private long exportFrequencySeconds = DefaultBuilderConfig.getExportFrequencySeconds();
-    private boolean exportSystemMetrics = DefaultBuilderConfig.getExportSystemMetrics();
     private Function<SocketAddress, ClientTransport> clientTransportFactory =
         address -> TcpClientTransport.create((InetSocketAddress) address);
     private int poolSize = Runtime.getRuntime().availableProcessors();
@@ -154,31 +152,11 @@ public Builder poolSize(int poolSize) {
       return this;
     }
 
-    public Builder meterRegistry(MeterRegistry registry) {
-      this.registry = registry;
-      return this;
-    }
-
-    public Builder metricHandlerGroup(String metricHandlerGroup) {
-      this.metricHandlerGroup = metricHandlerGroup;
-      return this;
-    }
-
     public Builder metricBatchSize(int batchSize) {
       this.batchSize = batchSize;
       return this;
     }
 
-    public Builder metricExportFrequencySeconds(long exportFrequencySeconds) {
-      this.exportFrequencySeconds = exportFrequencySeconds;
-      return this;
-    }
-
-    public Builder exportSystemMetrics(boolean exportSystemMetrics) {
-      this.exportSystemMetrics = exportSystemMetrics;
-      return this;
-    }
-
     public Builder keepalive(boolean useKeepAlive) {
       this.keepalive = keepalive;
       return this;
@@ -208,12 +186,12 @@ public Builder port(int port) {
       this.port = port;
       return this;
     }
-  
+
     public Builder tracerSupplier(Supplier<Tracer> tracerSupplier) {
       this.tracerSupplier = tracerSupplier;
       return this;
     }
-  
+
     public Builder seedAddresses(Collection<SocketAddress> addresses) {
       if (addresses instanceof List) {
         this.seedAddresses = (List<SocketAddress>) addresses;
@@ -309,35 +287,6 @@ public Proteus build() {
                     tracerSupplier);
             proteus.onClose.doFinally(s -> PROTEUS.remove(proteusKey)).subscribe();
 
-            if (registry != null) {
-              registry
-                  .config()
-                  .commonTags(
-                      "accessKey",
-                      String.valueOf(accessKey),
-                      "group",
-                      group,
-                      "destination",
-                      destination);
-
-              ProteusOperatingSystemMetrics systemMetrics =
-                  new ProteusOperatingSystemMetrics(registry);
-
-              ProteusSocket socket = proteus.group(metricHandlerGroup);
-              MetricsSnapshotHandler handler = new MetricsSnapshotHandlerClient(socket);
-              ProteusMetricsExporter exporter =
-                  new ProteusMetricsExporter(
-                      null, registry, Duration.ofSeconds(exportFrequencySeconds), batchSize);
-              exporter.run();
-              proteus
-                  .onClose
-                  .doFinally(
-                      s -> {
-                        exporter.dispose();
-                      })
-                  .subscribe();
-            }
-
             return proteus;
           });
     }

client/src/main/java/io/netifi/proteus/presence/BrokerInfoPresenceNotifier.java

@@ -11,8 +11,13 @@
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.FluxProcessor;
 import reactor.core.publisher.Mono;
+import reactor.retry.Retry;
 
-import java.util.*;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 

client/src/main/java/io/netifi/proteus/rsocket/WeightedReconnectingRSocket.java

@@ -15,6 +15,8 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.*;
+import reactor.retry.Jitter;
+import reactor.retry.Retry;
 
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
@@ -257,7 +259,8 @@ void connect() {
                       .doOnNext(
                           _rSocket -> {
                             availability = 1.0;
-                            ErrorOnDisconnectRSocket rSocket = new ErrorOnDisconnectRSocket(_rSocket);
+                            ErrorOnDisconnectRSocket rSocket =
+                                new ErrorOnDisconnectRSocket(_rSocket);
                             _rSocket
                                 .onClose()
                                 .doFinally(
@@ -277,9 +280,9 @@ void connect() {
                                       synchronized (WeightedReconnectingRSocket.this) {
                                         connecting = false;
                                       }
-                                      
+
                                       rSocket.dispose();
-                                      
+
                                       connect();
                                     })
                                 .subscribe();

core/build.gradle

@@ -6,14 +6,16 @@ sourceCompatibility = 1.8
 targetCompatibility = 1.8
 
 dependencies {
-    protobuf project (':metrics-idl')
     compile project (':frames')
     compile 'io.opentracing:opentracing-api:0.31.0'
     compile 'javax.inject:javax.inject:1'
     compile 'com.google.protobuf:protobuf-java:3.6.0'
     compile 'io.rsocket:rsocket-core:0.11.3'
     compile 'io.rsocket:rsocket-transport-netty:0.11.3'
     compile 'io.micrometer:micrometer-core:1.0.3'
+
+    protobuf project(':metrics-idl')
+
     testCompile 'io.opentracing.brave:brave-opentracing:0.31.2'
     testCompile 'io.zipkin.reporter2:zipkin-sender-okhttp3:2.7.6'
 }

core/src/main/java/io/netifi/proteus/metrics/ProteusMetrics.java

@@ -27,7 +27,7 @@ public class ProteusMetrics {
             .tags(tags)
             .register(registry);
     Timer timer =
-        Timer.builder(name + ".latency").publishPercentiles(0.5, 0.9, 0.95, 0.99).tags(tags).register(registry);
+        Timer.builder(name + ".latency").publishPercentiles(0.5, 0.9, 0.95, 0.99).tags(tags).register( registry);
     return Operators.lift(
         (scannable, subscriber) ->
             new ProteusMetricsSubscriber<>(subscriber, next, complete, error, cancelled, timer));