Record exception name in onErrorResume (proxy), polish

Author: Jon Schneider

client/src/test/java/io/micrometer/prometheus/rsocket/PrometheusRSocketClientTests.java

@@ -39,10 +39,12 @@ class PrometheusRSocketClientTests {
 
   PrometheusMeterRegistry meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
 
-  /** See https://github.com/micrometer-metrics/prometheus-rsocket-proxy/issues/3 */
+  /**
+   * See https://github.com/micrometer-metrics/prometheus-rsocket-proxy/issues/3
+   */
   @Test
   void dyingScrapeAfterNormalScrape() throws NoSuchAlgorithmException, InterruptedException {
-    CountDownLatch dyingScapeLatch = new CountDownLatch(1);
+    CountDownLatch dyingScrapeLatch = new CountDownLatch(1);
     Payload payload = DefaultPayload.create(KeyPairGenerator.getInstance("RSA").generateKeyPair().getPublic().getEncoded());
     RSocketFactory.receive()
         .frameDecoder(PayloadDecoder.ZERO_COPY)
@@ -52,7 +54,7 @@ void dyingScrapeAfterNormalScrape() throws NoSuchAlgorithmException, Interrupted
           return Mono.just(new AbstractRSocket() {
             @Override
             public Mono<Void> fireAndForget(Payload payload) {
-              dyingScapeLatch.countDown();
+              dyingScrapeLatch.countDown();
               return Mono.empty();
             }
           });
@@ -65,7 +67,7 @@ public Mono<Void> fireAndForget(Payload payload) {
 
     // trigger dying scrape
     client.pushAndClose();
-    assertThat(dyingScapeLatch.await(1, TimeUnit.SECONDS))
+    assertThat(dyingScrapeLatch.await(1, TimeUnit.SECONDS))
         .as("Dying scrape (fire-and-forget) should be successfully called")
         .isTrue();
   }

proxy/build.gradle

@@ -24,7 +24,6 @@ publishing {
 dependencies {
   implementation 'org.springframework.boot:spring-boot-starter-actuator'
   implementation 'org.springframework.boot:spring-boot-starter-webflux'
-  implementation 'org.pcollections:pcollections:latest.release'
   implementation 'io.micrometer:micrometer-registry-prometheus'
   implementation 'io.rsocket:rsocket-micrometer'
   implementation 'io.rsocket:rsocket-transport-netty'

proxy/src/main/java/io/micrometer/prometheus/rsocket/PrometheusController.java

@@ -15,25 +15,9 @@
  */
 package io.micrometer.prometheus.rsocket;
 
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.nio.channels.ClosedChannelException;
-import java.security.KeyFactory;
-import java.security.KeyPair;
-import java.security.KeyPairGenerator;
-import java.security.NoSuchAlgorithmException;
-import java.security.PrivateKey;
-import java.security.spec.PKCS8EncodedKeySpec;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
-
-import javax.annotation.PostConstruct;
-import javax.crypto.Cipher;
-import javax.crypto.SecretKey;
-import javax.crypto.spec.SecretKeySpec;
-
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.Tags;
 import io.micrometer.core.instrument.Timer;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.netty.buffer.ByteBuf;
@@ -47,15 +31,25 @@
 import io.rsocket.transport.netty.server.TcpServerTransport;
 import io.rsocket.transport.netty.server.WebsocketServerTransport;
 import io.rsocket.util.DefaultPayload;
-import org.pcollections.HashTreePMap;
-import org.pcollections.PMap;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
 import org.xerial.snappy.Snappy;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RestController;
+import javax.annotation.PostConstruct;
+import javax.crypto.Cipher;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.channels.ClosedChannelException;
+import java.security.*;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 /**
  * A {@link Controller} for endpoints to be scraped by Prometheus.
@@ -69,25 +63,23 @@ class PrometheusController {
   private final Timer scrapeTimerSuccess;
   private final Timer scrapeTimerClosed;
   private final Counter scrapeSocketsClosed;
-  private final Timer scrapeTimerError;
   private final DistributionSummary scrapePayload;
   private final MicrometerRSocketInterceptor metricsInterceptor;
   private PrometheusControllerProperties properties;
-  private AtomicReference<PMap<RSocket, ConnectionState>> scrapableApps = new AtomicReference<>(HashTreePMap.empty());
+  private Map<RSocket, ConnectionState> scrapableApps = new ConcurrentHashMap<>();
 
   PrometheusController(PrometheusMeterRegistry meterRegistry, PrometheusControllerProperties properties) {
     this.meterRegistry = meterRegistry;
     this.metricsInterceptor = new MicrometerRSocketInterceptor(meterRegistry);
     this.properties = properties;
-    meterRegistry.gauge("prometheus.proxy.scrape.active.connections", scrapableApps, apps -> apps.get().size());
+    meterRegistry.gaugeMapSize("prometheus.proxy.scrape.active.connections", Tags.empty(), scrapableApps);
 
     this.scrapeTimerSuccess = Timer.builder("prometheus.proxy.scrape")
       .tag("outcome", "success")
       .publishPercentileHistogram()
       .register(meterRegistry);
 
     this.scrapeTimerClosed = meterRegistry.timer("prometheus.proxy.scrape", "outcome", "closed");
-    this.scrapeTimerError = meterRegistry.timer("prometheus.proxy.scrape", "outcome", "error");
     this.scrapePayload = DistributionSummary.builder("prometheus.proxy.scrape.payload")
       .publishPercentileHistogram()
       .baseUnit("bytes")
@@ -119,7 +111,7 @@ private Mono<RSocket> acceptRSocket(KeyPairGenerator generator, RSocket sendingS
     // respond with Mono.error(..) to
 
     ConnectionState connectionState = new ConnectionState(generator.generateKeyPair());
-    scrapableApps.getAndUpdate(apps -> apps.plus(metricsInterceptor.apply(sendingSocket), connectionState));
+    scrapableApps.put(metricsInterceptor.apply(sendingSocket), connectionState);
 
     // for use by the client to push metrics as it's dying if this happens before the first scrape
     sendingSocket.fireAndForget(connectionState.createKeyPayload())
@@ -147,7 +139,7 @@ public Mono<String> proxyMetrics() {
   @GetMapping(value = "/metrics/connected", produces = "text/plain")
   public Mono<String> prometheus() {
     return Flux
-      .fromIterable(scrapableApps.get().entrySet())
+      .fromIterable(scrapableApps.entrySet())
       .flatMap(socketAndState -> {
         ConnectionState connectionState = socketAndState.getValue();
         RSocket rsocket = socketAndState.getKey();
@@ -158,11 +150,14 @@ public Mono<String> prometheus() {
           .onErrorResume(throwable -> {
             if (throwable instanceof ClosedChannelException) {
               scrapeSocketsClosed.increment();
-              scrapableApps.getAndUpdate(apps -> apps.minus(rsocket));
+              scrapableApps.remove(rsocket);
               sample.stop(scrapeTimerClosed);
               return connectionState.getDyingPush();
             }
-            sample.stop(scrapeTimerError);
+
+            sample.stop(meterRegistry.timer("prometheus.proxy.scrape", "outcome", "error",
+                "exception", throwable.getClass().getName()));
+
             return Mono.empty();
           });
       })
@@ -171,6 +166,7 @@ public Mono<String> prometheus() {
 
   class ConnectionState {
     private final KeyPair keyPair;
+
     // the last metrics of a dying application instance
     private String dyingPush;