Introduce client-side micrometer, start with exposing watcher revision metrics

client/java-armeria-legacy/src/main/java/com/linecorp/centraldogma/client/armeria/legacy/LegacyCentralDogma.java

@@ -90,12 +90,15 @@
 import com.linecorp.centraldogma.internal.thrift.WatchFileResult;
 import com.linecorp.centraldogma.internal.thrift.WatchRepositoryResult;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 final class LegacyCentralDogma extends AbstractCentralDogma {
 
     private final CentralDogmaService.AsyncIface client;
 
-    LegacyCentralDogma(ScheduledExecutorService blockingTaskExecutor, CentralDogmaService.AsyncIface client) {
-        super(blockingTaskExecutor);
+    LegacyCentralDogma(ScheduledExecutorService blockingTaskExecutor, CentralDogmaService.AsyncIface client,
+                       MeterRegistry meterRegistry) {
+        super(blockingTaskExecutor, meterRegistry);
         this.client = requireNonNull(client, "client");
     }
 

client/java-armeria-legacy/src/main/java/com/linecorp/centraldogma/client/armeria/legacy/LegacyCentralDogmaBuilder.java

@@ -18,6 +18,9 @@
 import java.net.UnknownHostException;
 import java.util.concurrent.ScheduledExecutorService;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.linecorp.armeria.client.ClientBuilder;
 import com.linecorp.armeria.client.ClientRequestContext;
 import com.linecorp.armeria.client.encoding.DecodingClient;
@@ -30,13 +33,17 @@
 import com.linecorp.centraldogma.internal.client.ReplicationLagTolerantCentralDogma;
 import com.linecorp.centraldogma.internal.thrift.CentralDogmaService.AsyncIface;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 /**
  * Builds a legacy {@link CentralDogma} client based on Thrift.
  *
  * @deprecated Use {@link ArmeriaCentralDogmaBuilder}.
  */
 @Deprecated
 public class LegacyCentralDogmaBuilder extends AbstractArmeriaCentralDogmaBuilder<LegacyCentralDogmaBuilder> {
+    private static final Logger logger = LoggerFactory.getLogger(LegacyCentralDogmaBuilder.class);
+
     /**
      * Returns a newly-created {@link CentralDogma} instance.
      *
@@ -67,8 +74,14 @@ public CentralDogma build() throws UnknownHostException {
         final ScheduledExecutorService blockingTaskExecutor = blockingTaskExecutor();
 
         final int maxRetriesOnReplicationLag = maxNumRetriesOnReplicationLag();
+
+        final MeterRegistry meterRegistry = meterRegistry().orElse(clientFactory().meterRegistry());
+        if (meterRegistry != clientFactory().meterRegistry()) {
+            logger.warn("The specified meterRegistry differs from the meterRegistry from clientFactory.");
+        }
+
         final CentralDogma dogma = new LegacyCentralDogma(blockingTaskExecutor,
-                                                          builder.build(AsyncIface.class));
+                                                          builder.build(AsyncIface.class), meterRegistry);
         if (maxRetriesOnReplicationLag <= 0) {
             return dogma;
         } else {
@@ -80,7 +93,7 @@ public CentralDogma build() throws UnknownHostException {
                         //                 in Armeria: https://github.com/line/armeria/issues/760
                         final ClientRequestContext ctx = ClientRequestContext.currentOrNull();
                         return ctx != null ? ctx.remoteAddress() : null;
-                    });
+                    }, meterRegistry);
         }
     }
 }

client/java-armeria-legacy/src/test/java/com/linecorp/centraldogma/client/armeria/legacy/LegacyCentralDogmaTest.java

@@ -37,6 +37,7 @@
 import com.google.common.collect.ImmutableSet;
 
 import com.linecorp.armeria.common.CommonPools;
+import com.linecorp.armeria.common.metric.NoopMeterRegistry;
 import com.linecorp.centraldogma.client.CentralDogma;
 import com.linecorp.centraldogma.client.RepositoryInfo;
 import com.linecorp.centraldogma.client.armeria.legacy.ThriftTypes.TAuthor;
@@ -80,7 +81,7 @@ class LegacyCentralDogmaTest {
 
     @BeforeEach
     void setUp() {
-        client = new LegacyCentralDogma(CommonPools.blockingTaskExecutor(), iface);
+        client = new LegacyCentralDogma(CommonPools.blockingTaskExecutor(), iface, NoopMeterRegistry.get());
     }
 
     @Test

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/ArmeriaCentralDogma.java

@@ -106,6 +106,8 @@
 import com.linecorp.centraldogma.internal.Util;
 import com.linecorp.centraldogma.internal.api.v1.WatchTimeout;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 final class ArmeriaCentralDogma extends AbstractCentralDogma {
 
     private static final MediaType JSON_PATCH_UTF8 = MediaType.JSON_PATCH.withCharset(StandardCharsets.UTF_8);
@@ -134,8 +136,9 @@ final class ArmeriaCentralDogma extends AbstractCentralDogma {
     private final WebClient client;
     private final String authorization;
 
-    ArmeriaCentralDogma(ScheduledExecutorService blockingTaskExecutor, WebClient client, String accessToken) {
-        super(blockingTaskExecutor);
+    ArmeriaCentralDogma(ScheduledExecutorService blockingTaskExecutor, WebClient client, String accessToken,
+                        MeterRegistry meterRegistry) {
+        super(blockingTaskExecutor, meterRegistry);
         this.client = requireNonNull(client, "client");
         authorization = "Bearer " + requireNonNull(accessToken, "accessToken");
     }

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/ArmeriaCentralDogmaBuilder.java

@@ -18,6 +18,9 @@
 import java.net.UnknownHostException;
 import java.util.concurrent.ScheduledExecutorService;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.linecorp.armeria.client.ClientBuilder;
 import com.linecorp.armeria.client.ClientRequestContext;
 import com.linecorp.armeria.client.WebClient;
@@ -26,12 +29,16 @@
 import com.linecorp.centraldogma.client.CentralDogma;
 import com.linecorp.centraldogma.internal.client.ReplicationLagTolerantCentralDogma;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 /**
  * Builds a {@link CentralDogma} client based on an <a href="https://line.github.io/armeria/">Armeria</a>
  * HTTP client.
  */
 public final class ArmeriaCentralDogmaBuilder
         extends AbstractArmeriaCentralDogmaBuilder<ArmeriaCentralDogmaBuilder> {
+    private static final Logger logger = LoggerFactory.getLogger(ArmeriaCentralDogmaBuilder.class);
+
     /**
      * Returns a newly-created {@link CentralDogma} instance.
      *
@@ -44,13 +51,17 @@ public CentralDogma build() throws UnknownHostException {
                 newClientBuilder(scheme, endpointGroup, cb -> cb.decorator(DecodingClient.newDecorator()), "/");
         final int maxRetriesOnReplicationLag = maxNumRetriesOnReplicationLag();
 
+        final MeterRegistry meterRegistry = meterRegistry().orElse(clientFactory().meterRegistry());
+        if (meterRegistry != clientFactory().meterRegistry()) {
+            logger.warn("The specified meterRegistry differs from the meterRegistry from clientFactory.");
+        }
         // TODO(ikhoon): Apply ExecutorServiceMetrics for the 'blockingTaskExecutor' once
         //               https://github.com/line/centraldogma/pull/542 is merged.
         final ScheduledExecutorService blockingTaskExecutor = blockingTaskExecutor();
 
         final CentralDogma dogma = new ArmeriaCentralDogma(blockingTaskExecutor,
                                                            builder.build(WebClient.class),
-                                                           accessToken());
+                                                           accessToken(), meterRegistry);
         if (maxRetriesOnReplicationLag <= 0) {
             return dogma;
         } else {
@@ -62,7 +73,7 @@ public CentralDogma build() throws UnknownHostException {
                         //                 in Armeria: https://github.com/line/armeria/issues/760
                         final ClientRequestContext ctx = ClientRequestContext.currentOrNull();
                         return ctx != null ? ctx.remoteAddress() : null;
-                    });
+                    }, meterRegistry);
         }
     }
 }

client/java-spring-boot-autoconfigure/src/main/java/com/linecorp/centraldogma/client/spring/CentralDogmaClientAutoConfiguration.java

@@ -36,6 +36,8 @@
 import com.linecorp.centraldogma.client.armeria.ArmeriaClientConfigurator;
 import com.linecorp.centraldogma.client.armeria.DnsAddressEndpointGroupConfigurator;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 /**
  * Spring bean configuration for {@link CentralDogma} client.
  */
@@ -55,8 +57,8 @@ public CentralDogma dogmaClient(
             CentralDogmaSettings settings,
             Optional<List<CentralDogmaClientFactoryConfigurator>> factoryConfigurators,
             Optional<ArmeriaClientConfigurator> armeriaClientConfigurator,
-            Optional<DnsAddressEndpointGroupConfigurator> dnsAddressEndpointGroupConfigurator)
-            throws UnknownHostException {
+            Optional<DnsAddressEndpointGroupConfigurator> dnsAddressEndpointGroupConfigurator,
+            Optional<MeterRegistry> meterRegistry) throws UnknownHostException {
 
         final ArmeriaCentralDogmaBuilder builder = new ArmeriaCentralDogmaBuilder();
 
@@ -125,6 +127,10 @@ public CentralDogma dogmaClient(
             builder.retryIntervalOnReplicationLagMillis(retryIntervalOnReplicationLagMillis);
         }
 
+        if (meterRegistry.isPresent()) {
+            builder.meterRegistry(meterRegistry.get());
+        }
+
         return builder.build();
     }
 }

client/java/build.gradle

@@ -1,3 +1,4 @@
 dependencies {
     implementation 'org.javassist:javassist'
+    implementation 'io.micrometer:micrometer-core'
 }

client/java/src/main/java/com/linecorp/centraldogma/client/AbstractCentralDogma.java

@@ -37,12 +37,16 @@
 import com.linecorp.centraldogma.internal.client.FileWatcher;
 import com.linecorp.centraldogma.internal.client.RepositoryWatcher;
 
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+
 /**
  * A skeletal {@link CentralDogma} implementation.
  */
 public abstract class AbstractCentralDogma implements CentralDogma {
 
     private final ScheduledExecutorService blockingTaskExecutor;
+    private final MeterRegistry meterRegistry;
 
     /**
      * Creates a new instance.
@@ -52,7 +56,22 @@ public abstract class AbstractCentralDogma implements CentralDogma {
      *                             watched changes.
      */
     protected AbstractCentralDogma(ScheduledExecutorService blockingTaskExecutor) {
+        this(blockingTaskExecutor, Metrics.globalRegistry);
+    }
+
+    /**
+     * Creates a new instance.
+     *
+     * @param blockingTaskExecutor the {@link ScheduledExecutorService} which will be used for scheduling the
+     *                             tasks related with automatic retries and invoking the callbacks for
+     *                             watched changes.
+     * @param meterRegistry the {@link MeterRegistry} which collects metrics for
+     *                      this {@link CentralDogma} instance.
+     */
+    protected AbstractCentralDogma(ScheduledExecutorService blockingTaskExecutor,
+                                   MeterRegistry meterRegistry) {
         this.blockingTaskExecutor = requireNonNull(blockingTaskExecutor, "blockingTaskExecutor");
+        this.meterRegistry = requireNonNull(meterRegistry, "meterRegistry");
     }
 
     /**
@@ -215,4 +234,9 @@ protected final CompletableFuture<Revision> maybeNormalizeRevision(
             return CompletableFuture.completedFuture(revision);
         }
     }
+
+    @Override
+    public final MeterRegistry meterRegistry() {
+        return meterRegistry;
+    }
 }

client/java/src/main/java/com/linecorp/centraldogma/client/AbstractCentralDogmaBuilder.java

@@ -31,6 +31,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -46,6 +47,8 @@
 import com.linecorp.centraldogma.common.RevisionNotFoundException;
 import com.linecorp.centraldogma.internal.CsrfToken;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 /**
  * Builds a {@link CentralDogma} client.
  */
@@ -70,6 +73,7 @@ public abstract class AbstractCentralDogmaBuilder<B extends AbstractCentralDogma
     private int maxNumRetriesOnReplicationLag = DEFAULT_MAX_NUM_RETRIES_ON_REPLICATION_LAG;
     private long retryIntervalOnReplicationLagMillis =
             TimeUnit.SECONDS.toMillis(DEFAULT_RETRY_INTERVAL_ON_REPLICATION_LAG_SECONDS);
+    private Optional<MeterRegistry> meterRegistry = Optional.empty();
 
     /**
      * Returns {@code this}.
@@ -415,4 +419,16 @@ public final B retryIntervalOnReplicationLagMillis(long retryIntervalOnReplicati
     protected long retryIntervalOnReplicationLagMillis() {
         return retryIntervalOnReplicationLagMillis;
     }
+
+    /**
+     * Sets the {@link MeterRegistry} used to collect metrics.
+     */
+    public final B meterRegistry(MeterRegistry meterRegistry) {
+        this.meterRegistry = Optional.of(requireNonNull(meterRegistry, "meterRegistry"));
+        return self();
+    }
+
+    protected Optional<MeterRegistry> meterRegistry() {
+        return meterRegistry;
+    }
 }

client/java/src/main/java/com/linecorp/centraldogma/client/CentralDogma.java

@@ -42,6 +42,8 @@
 import com.linecorp.centraldogma.common.Revision;
 import com.linecorp.centraldogma.common.RevisionNotFoundException;
 
+import io.micrometer.core.instrument.MeterRegistry;
+
 /**
  * Central Dogma client.
  */
@@ -591,4 +593,11 @@ <T> Watcher<T> repositoryWatcher(String projectName, String repositoryName, Stri
      */
     <T> Watcher<T> repositoryWatcher(String projectName, String repositoryName, String pathPattern,
                                      Function<Revision, ? extends T> function, Executor executor);
+
+    /**
+     * The {@link MeterRegistry} which collects metrics for this {@link CentralDogma} instance.
+     *
+     * @return the {@link MeterRegistry}
+     */
+    MeterRegistry meterRegistry();
 }

client/java/src/main/java/com/linecorp/centraldogma/internal/client/AbstractWatcher.java

@@ -20,6 +20,7 @@
 import static java.util.Objects.requireNonNull;
 
 import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
@@ -39,6 +40,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.util.concurrent.AtomicDouble;
+
 import com.linecorp.centraldogma.client.CentralDogma;
 import com.linecorp.centraldogma.client.Latest;
 import com.linecorp.centraldogma.client.Watcher;
@@ -49,6 +52,8 @@
 import com.linecorp.centraldogma.common.Revision;
 import com.linecorp.centraldogma.common.ShuttingDownException;
 
+import io.micrometer.core.instrument.Gauge;
+
 abstract class AbstractWatcher<T> implements Watcher<T> {
 
     private static final Logger logger = LoggerFactory.getLogger(AbstractWatcher.class);
@@ -114,6 +119,7 @@ private enum State {
     private final List<Map.Entry<BiConsumer<? super Revision, ? super T>, Executor>> updateListeners;
     private final AtomicReference<State> state;
     private final CompletableFuture<Latest<T>> initialValueFuture;
+    private final AtomicDouble latestNotifiedRevision = new AtomicDouble();
 
     private volatile Latest<T> latest;
     private volatile ScheduledFuture<?> currentScheduleFuture;
@@ -127,6 +133,13 @@ protected AbstractWatcher(CentralDogma client, ScheduledExecutorService watchSch
         this.repositoryName = requireNonNull(repositoryName, "repositoryName");
         this.pathPattern = requireNonNull(pathPattern, "pathPattern");
 
+        Gauge.builder("centraldogma.client.watcher.revision",
+                      this, watcher -> watcher.latestNotifiedRevision.get())
+             .tag("project", projectName)
+             .tag("repository", repositoryName)
+             .tag("path", pathPattern)
+             .register(client.meterRegistry());
+
         updateListeners = new CopyOnWriteArrayList<>();
         state = new AtomicReference<>(State.INIT);
         initialValueFuture = new CompletableFuture<>();
@@ -296,18 +309,28 @@ private void notifyListeners() {
         }
 
         final Latest<T> latest = this.latest;
+        final List<CompletableFuture<Boolean>> futures = new ArrayList<>();
         for (Map.Entry<BiConsumer<? super Revision, ? super T>, Executor> entry : updateListeners) {
             final BiConsumer<? super Revision, ? super T> listener = entry.getKey();
             final Executor executor = entry.getValue();
-            executor.execute(() -> {
+            final CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() -> {
                 try {
                     listener.accept(latest.revision(), latest.value());
+                    return true;
                 } catch (Exception e) {
                     logger.warn("Exception thrown for watcher ({}/{}{}): rev={}",
                                 projectName, repositoryName, pathPattern, latest.revision(), e);
+                    return false;
                 }
-            });
+            }, executor);
+            futures.add(future);
         }
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).thenAccept(ignored -> {
+            final boolean result = futures.stream().allMatch(CompletableFuture::join);
+            if (result) {
+                latestNotifiedRevision.set(latest.revision().major());
+            }
+        });
     }
 
     private void handleExecutorShutdown(Executor executor, RejectedExecutionException e) {