Revert "Use Micrometer rather than Dropwizard for a few metrics"

This reverts commit 2bc91c1.

Author: jkt-signal

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -21,7 +21,6 @@
 import io.dropwizard.core.setup.Bootstrap;
 import io.dropwizard.core.setup.Environment;
 import io.dropwizard.jetty.HttpsConnectorFactory;
-import io.dropwizard.lifecycle.setup.LifecycleEnvironment;
 import io.grpc.ServerBuilder;
 import io.lettuce.core.metrics.MicrometerCommandLatencyRecorder;
 import io.lettuce.core.metrics.MicrometerOptions;
@@ -365,8 +364,8 @@ public void run(WhisperServerConfiguration config, Environment environment) thro
 
     UncaughtExceptionHandler.register();
 
-    ScheduledExecutorService dynamicConfigurationExecutor = ScheduledExecutorServiceBuilder.of(environment, "dynamicConfiguration")
-        .threads(1).build();
+    ScheduledExecutorService dynamicConfigurationExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "dynamicConfiguration-%d")).threads(1).build();
 
     DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager =
         new DynamicConfigurationManager<>(
@@ -398,7 +397,8 @@ public void run(WhisperServerConfiguration config, Environment environment) thro
 
     environment.lifecycle().manage(new ManagedAwsCrt());
 
-    final ExecutorService awsSdkMetricsExecutor = virtualExecutorService(environment, "awsSdkMetrics");
+    final ExecutorService awsSdkMetricsExecutor = environment.lifecycle()
+        .virtualExecutorService(name(getClass(), "awsSdkMetrics-%d"));
 
     final DynamoDbAsyncClient dynamoDbAsyncClient = config.getDynamoDbClientConfiguration()
         .buildAsyncClient(awsCredentialsProvider, new MicrometerAwsSdkMetricPublisher(awsSdkMetricsExecutor, "dynamoDbAsync"));
@@ -415,7 +415,8 @@ public void run(WhisperServerConfiguration config, Environment environment) thro
     BlockingQueue<Runnable> messageDeletionQueue = new LinkedBlockingQueue<>();
     Metrics.gaugeCollectionSize(name(getClass(), "messageDeletionQueueSize"), Collections.emptyList(),
         messageDeletionQueue);
-    ExecutorService messageDeletionAsyncExecutor = ExecutorServiceBuilder.of(environment, "messageDeletionAsyncExecutor")
+    ExecutorService messageDeletionAsyncExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "messageDeletionAsyncExecutor-%d"))
         .minThreads(2)
         .maxThreads(2)
         .allowCoreThreadTimeOut(true)
@@ -502,79 +503,98 @@ public void run(WhisperServerConfiguration config, Environment environment) thro
     Metrics.gaugeCollectionSize(MetricsUtil.name(getClass(), "messageDeliveryQueue"), Collections.emptyList(),
         messageDeliveryQueue);
 
-    ScheduledExecutorService recurringJobExecutor = ScheduledExecutorServiceBuilder.of(environment, "recurringJob").threads(6).build();
-    ScheduledExecutorService websocketScheduledExecutor = ScheduledExecutorServiceBuilder.of(environment, "websocket").threads(8).build();
-    ExecutorService apnSenderExecutor = ExecutorServiceBuilder.of(environment, "apnSender")
+    ScheduledExecutorService recurringJobExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "recurringJob-%d")).threads(6).build();
+    ScheduledExecutorService websocketScheduledExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "websocket-%d")).threads(8).build();
+    ExecutorService apnSenderExecutor = environment.lifecycle().executorService(name(getClass(), "apnSender-%d"))
         .maxThreads(1).minThreads(1).build();
-    ExecutorService fcmSenderExecutor = ExecutorServiceBuilder.of(environment, "fcmSender")
+    ExecutorService fcmSenderExecutor = environment.lifecycle().executorService(name(getClass(), "fcmSender-%d"))
         .maxThreads(32).minThreads(32).workQueue(fcmSenderQueue).build();
-    ExecutorService secureValueRecoveryServiceExecutor = ExecutorServiceBuilder.of(environment, "secureValueRecoveryService")
-        .maxThreads(1).minThreads(1).build();
-    ExecutorService storageServiceExecutor = ExecutorServiceBuilder.of(environment, "storageService")
-        .maxThreads(1).minThreads(1).build();
-    ExecutorService virtualThreadEventLoggerExecutor = ExecutorServiceBuilder.of(environment, "virtualThreadEventLogger")
-        .minThreads(1).maxThreads(1).build();
-    ExecutorService asyncOperationQueueingExecutor = ExecutorServiceBuilder.of(environment, "asyncOperationQueueing")
-        .minThreads(1).maxThreads(1).build();
-    ScheduledExecutorService secureValueRecoveryServiceRetryExecutor =
-      ScheduledExecutorServiceBuilder.of(environment, "secureValueRecoveryServiceRetry").threads(1).build();
-    ScheduledExecutorService storageServiceRetryExecutor =
-      ScheduledExecutorServiceBuilder.of(environment, "storageServiceRetry").threads(1).build();
-    ScheduledExecutorService remoteStorageRetryExecutor =
-      ScheduledExecutorServiceBuilder.of(environment, "remoteStorageRetry").threads(1).build();
-    ScheduledExecutorService registrationIdentityTokenRefreshExecutor =
-      ScheduledExecutorServiceBuilder.of(environment, "registrationIdentityTokenRefresh").threads(1).build();
+    ExecutorService secureValueRecoveryServiceExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "secureValueRecoveryService-%d")).maxThreads(1).minThreads(1).build();
+    ExecutorService storageServiceExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "storageService-%d")).maxThreads(1).minThreads(1).build();
+    ExecutorService virtualThreadEventLoggerExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "virtualThreadEventLogger-%d")).minThreads(1).maxThreads(1).build();
+    ExecutorService asyncOperationQueueingExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "asyncOperationQueueing-%d")).minThreads(1).maxThreads(1).build();
+    ScheduledExecutorService secureValueRecoveryServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "secureValueRecoveryServiceRetry-%d")).threads(1).build();
+    ScheduledExecutorService storageServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "storageServiceRetry-%d")).threads(1).build();
+    ScheduledExecutorService remoteStorageRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "remoteStorageRetry-%d")).threads(1).build();
+    ScheduledExecutorService registrationIdentityTokenRefreshExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "registrationIdentityTokenRefresh-%d")).threads(1).build();
 
     Scheduler messageDeliveryScheduler = Schedulers.fromExecutorService(
-        ExecutorServiceBuilder.of(environment, "messageDelivery")
-            .minThreads(20)
-            .maxThreads(20)
-            .workQueue(messageDeliveryQueue)
-            .build(),
+        ExecutorServiceMetrics.monitor(Metrics.globalRegistry,
+            environment.lifecycle().executorService(name(getClass(), "messageDelivery-%d"))
+                .minThreads(20)
+                .maxThreads(20)
+                .workQueue(messageDeliveryQueue)
+                .build(),
+            MetricsUtil.name(getClass(), "messageDeliveryExecutor"), MetricsUtil.PREFIX),
         "messageDelivery");
 
     // TODO: generally speaking this is a DynamoDB I/O executor for the accounts table; we should eventually have a general executor for speaking to the accounts table, but most of the server is still synchronous so this isn't widely useful yet
-    ExecutorService batchIdentityCheckExecutor = ExecutorServiceBuilder.of(environment, "batchIdentityCheck").minThreads(32).maxThreads(32).build();
-    ExecutorService subscriptionProcessorExecutor = ExecutorServiceBuilder.of(environment, "subscriptionProcessor")
+    ExecutorService batchIdentityCheckExecutor = environment.lifecycle().executorService(name(getClass(), "batchIdentityCheck-%d")).minThreads(32).maxThreads(32).build();
+    ExecutorService subscriptionProcessorExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "subscriptionProcessor-%d"))
         .maxThreads(availableProcessors)  // mostly this is IO bound so tying to number of processors is tenuous at best
         .minThreads(availableProcessors)  // mostly this is IO bound so tying to number of processors is tenuous at best
         .allowCoreThreadTimeOut(true).
         build();
-    ExecutorService receiptSenderExecutor = ExecutorServiceBuilder.of(environment, "receiptSender")
+    ExecutorService receiptSenderExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "receiptSender-%d"))
         .maxThreads(2)
         .minThreads(2)
         .workQueue(receiptSenderQueue)
         .rejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy())
         .build();
-    ExecutorService registrationCallbackExecutor = ExecutorServiceBuilder.of(environment, "registration")
+    ExecutorService registrationCallbackExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "registration-%d"))
         .maxThreads(2)
         .minThreads(2)
         .build();
-    ExecutorService accountLockExecutor = ExecutorServiceBuilder.of(environment, "accountLock")
+    ExecutorService accountLockExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "accountLock-%d"))
         .minThreads(8)
         .maxThreads(8)
         .build();
     // unbounded executor (same as cachedThreadPool)
-    ExecutorService remoteStorageHttpExecutor = ExecutorServiceBuilder.of(environment, "remoteStorage")
+    ExecutorService remoteStorageHttpExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "remoteStorage-%d"))
         .minThreads(0)
         .maxThreads(Integer.MAX_VALUE)
         .workQueue(new SynchronousQueue<>())
         .keepAliveTime(io.dropwizard.util.Duration.seconds(60L))
         .build();
-    ExecutorService cloudflareTurnHttpExecutor = ExecutorServiceBuilder.of(environment, "cloudflareTurn")
+    ExecutorService cloudflareTurnHttpExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "cloudflareTurn-%d"))
         .maxThreads(2)
         .minThreads(2)
         .build();
-    ExecutorService googlePlayBillingExecutor = virtualExecutorService(environment, "googlePlayBilling");
-    ExecutorService appleAppStoreExecutor = virtualExecutorService(environment, "appleAppStore");
-    ExecutorService clientEventExecutor = virtualExecutorService(environment, "clientEvent");
-    ExecutorService disconnectionRequestListenerExecutor = virtualExecutorService(environment, "disconnectionRequest");
-
-    ScheduledExecutorService appleAppStoreRetryExecutor = ScheduledExecutorServiceBuilder.of(environment, "appleAppStoreRetry").threads(1).build();
-    ScheduledExecutorService subscriptionProcessorRetryExecutor = ScheduledExecutorServiceBuilder.of(environment, "subscriptionProcessorRetry").threads(1).build();
-    ScheduledExecutorService cloudflareTurnRetryExecutor = ScheduledExecutorServiceBuilder.of(environment, "cloudflareTurnRetry").threads(1).build();
-    ScheduledExecutorService messagePollExecutor = ScheduledExecutorServiceBuilder.of(environment, "messagePollExecutor").threads(1).build();
-    ScheduledExecutorService provisioningWebsocketTimeoutExecutor = ScheduledExecutorServiceBuilder.of(environment, "provisioningWebsocketTimeout").threads(1).build();
+    ExecutorService googlePlayBillingExecutor = environment.lifecycle()
+        .virtualExecutorService(name(getClass(), "googlePlayBilling-%d"));
+    ExecutorService appleAppStoreExecutor = environment.lifecycle()
+        .virtualExecutorService(name(getClass(), "appleAppStore-%d"));
+    ExecutorService clientEventExecutor = environment.lifecycle()
+        .virtualExecutorService(name(getClass(), "clientEvent-%d"));
+    ExecutorService disconnectionRequestListenerExecutor = environment.lifecycle()
+        .virtualExecutorService(name(getClass(), "disconnectionRequest-%d"));
+
+    ScheduledExecutorService appleAppStoreRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "appleAppStoreRetry-%d")).threads(1).build();
+    ScheduledExecutorService subscriptionProcessorRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "subscriptionProcessorRetry-%d")).threads(1).build();
+    ScheduledExecutorService cloudflareTurnRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "cloudflareTurnRetry-%d")).threads(1).build();
+    ScheduledExecutorService messagePollExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "messagePollExecutor-%d")).threads(1).build();
+    ScheduledExecutorService provisioningWebsocketTimeoutExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "provisioningWebsocketTimeout-%d")).threads(1).build();
 
     final ManagedNioEventLoopGroup dnsResolutionEventLoopGroup = new ManagedNioEventLoopGroup();
     final DnsNameResolver cloudflareDnsResolver = new DnsNameResolverBuilder(dnsResolutionEventLoopGroup.next())
@@ -902,7 +922,8 @@ protected void configureServer(final ServerBuilder<?> serverBuilder) {
       noiseWebSocketTlsPrivateKey = null;
     }
 
-    final ExecutorService noiseWebSocketDelegatedTaskExecutor = ExecutorServiceBuilder.of(environment, "noiseWebsocketDelegatedTask")
+    final ExecutorService noiseWebSocketDelegatedTaskExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "noiseWebsocketDelegatedTask-%d"))
         .minThreads(8)
         .maxThreads(8)
         .allowCoreThreadTimeOut(false)
@@ -1206,47 +1227,6 @@ private void registerExceptionMappers(Environment environment,
     });
   }
 
-  private static class ExecutorServiceBuilder extends io.dropwizard.lifecycle.setup.ExecutorServiceBuilder {
-    private final String baseName;
-
-    public ExecutorServiceBuilder(final LifecycleEnvironment environment, final String baseName) {
-      super(environment, name(WhisperServerService.class, baseName) + "-%d");
-      this.baseName = baseName;
-    }
-
-    @Override
-    public ExecutorService build() {
-      return ExecutorServiceMetrics.monitor(Metrics.globalRegistry, super.build(), baseName, MetricsUtil.PREFIX);
-    }
-
-    public static ExecutorServiceBuilder of(final Environment environment, final String name) {
-      return new ExecutorServiceBuilder(environment.lifecycle(), name);
-    }
-  }
-
-  private static class ScheduledExecutorServiceBuilder extends io.dropwizard.lifecycle.setup.ScheduledExecutorServiceBuilder {
-    private final String baseName;
-
-    public ScheduledExecutorServiceBuilder(final LifecycleEnvironment environment, final String baseName) {
-      super(environment, name(WhisperServerService.class, baseName) + "-%d", false);
-      this.baseName = baseName;
-    }
-
-    @Override
-    public ScheduledExecutorService build() {
-      return ExecutorServiceMetrics.monitor(Metrics.globalRegistry, super.build(), baseName, MetricsUtil.PREFIX);
-    }
-
-    public static ScheduledExecutorServiceBuilder of(final Environment environment, final String name) {
-      return new ScheduledExecutorServiceBuilder(environment.lifecycle(), name);
-    }
-  }
-
-  private ExecutorService virtualExecutorService(final Environment environment, final String name) {
-    return ExecutorServiceMetrics.monitor(
-      Metrics.globalRegistry, environment.lifecycle().virtualExecutorService(name(getClass(), name) + "-%d"), name, MetricsUtil.PREFIX);
-  }
-
   public static void main(String[] args) throws Exception {
     new WhisperServerService().run(args);
   }

service/src/main/java/org/whispersystems/textsecuregcm/controllers/MessageController.java

@@ -6,6 +6,7 @@
 
 import static com.codahale.metrics.MetricRegistry.name;
 
+import com.codahale.metrics.annotation.Timed;
 import com.google.common.net.HttpHeaders;
 import io.dropwizard.auth.Auth;
 import io.micrometer.core.instrument.Metrics;
@@ -468,6 +469,7 @@ private void sendIndividualMessage(final Account destination,
     }
   }
 
+  @Timed
   @Path("/multi_recipient")
   @PUT
   @Consumes(MultiRecipientMessageProvider.MEDIA_TYPE)
@@ -754,6 +756,7 @@ private void checkAccessKeys(
     }
   }
 
+  @Timed
   @GET
   @Produces(MediaType.APPLICATION_JSON)
   public CompletableFuture<OutgoingMessageEntityList> getPendingMessages(@Auth AuthenticatedDevice auth,
@@ -832,6 +835,7 @@ private static long estimateMessageListSizeBytes(final OutgoingMessageEntityList
     return size;
   }
 
+  @Timed
   @DELETE
   @Path("/uuid/{uuid}")
   public CompletableFuture<Response> removePendingMessage(@Auth AuthenticatedDevice auth, @PathParam("uuid") UUID uuid) {
@@ -870,6 +874,7 @@ public CompletableFuture<Response> removePendingMessage(@Auth AuthenticatedDevic
         .thenApply(Util.ASYNC_EMPTY_RESPONSE);
   }
 
+  @Timed
   @POST
   @Consumes(MediaType.APPLICATION_JSON)
   @Path("/report/{source}/{messageGuid}")

service/src/main/java/org/whispersystems/textsecuregcm/jetty/JettyHttpConfigurationCustomizer.java

@@ -14,8 +14,6 @@
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.binder.jetty.JettyConnectionMetrics;
 
 /**
  * Uses {@link Container.Listener} to update {@link org.eclipse.jetty.server.HttpConfiguration}
@@ -27,6 +25,7 @@ public class JettyHttpConfigurationCustomizer implements Container.Listener, Lif
   @Override
   public void beanAdded(final Container parent, final Object child) {
     if (child instanceof Connector c) {
+
       for (ConnectionFactory cf : c.getConnectionFactories()) {
         final HttpConfiguration httpConfiguration = switch (cf) {
           case HTTP2ServerConnectionFactory h2cf -> h2cf.getHttpConfiguration();
@@ -40,8 +39,6 @@ public void beanAdded(final Container parent, final Object child) {
           httpConfiguration.setNotifyRemoteAsyncErrors(false);
         }
       }
-
-      c.addBean(new JettyConnectionMetrics(Metrics.globalRegistry));
     }
   }
 

service/src/main/java/org/whispersystems/textsecuregcm/push/ReceiptSender.java

@@ -5,6 +5,8 @@
 
 package org.whispersystems.textsecuregcm.push;
 
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
@@ -14,6 +16,7 @@
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
 import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier;
 import org.whispersystems.textsecuregcm.identity.ServiceIdentifier;
+import org.whispersystems.textsecuregcm.metrics.MetricsUtil;
 import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
@@ -30,7 +33,9 @@ public ReceiptSender(final AccountsManager accountManager, final MessageSender m
       final ExecutorService executor) {
     this.accountManager = accountManager;
     this.messageSender = messageSender;
-    this.executor = executor;
+    this.executor = ExecutorServiceMetrics.monitor(
+        Metrics.globalRegistry, executor, MetricsUtil.name(ReceiptSender.class, "executor"), MetricsUtil.PREFIX)
+    ;
   }
 
   public void sendReceipt(ServiceIdentifier sourceIdentifier, byte sourceDeviceId, AciServiceIdentifier destinationIdentifier, long messageId) {