Improve monitor startup performance (#2395)

* Improve monitor startup performance by moving publish and subscribe flow initialization off main thread * Increase node validation timeout to 30s * Adjust monitor probe delay to reflect startup improvement Signed-off-by: Steven Sheehy <steven.sheehy@hedera.com>
hashgraph · Aug 11, 2021 · 44a3c9a · 44a3c9a
1 parent 1e6ef96
commit 44a3c9a
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 6 deletions.
diff --git a/charts/hedera-mirror-monitor/values.yaml b/charts/hedera-mirror-monitor/values.yaml
@@ -115,10 +115,11 @@ ingress:
 labels: {}
 
 livenessProbe:
+  failureThreshold: 5
   httpGet:
     path: /actuator/health/liveness
     port: http
-  initialDelaySeconds: 150
+  initialDelaySeconds: 60
   periodSeconds: 10
   timeoutSeconds: 2
 
@@ -266,10 +267,11 @@ rbac:
   enabled: true
 
 readinessProbe:
+  failureThreshold: 5
   httpGet:
     path: /actuator/health/readiness
     port: http
-  initialDelaySeconds: 100
+  initialDelaySeconds: 60
   timeoutSeconds: 2
 
 replicas: 1

diff --git a/...a-mirror-monitor/src/main/java/com/hedera/mirror/monitor/config/MonitorConfiguration.java b/...a-mirror-monitor/src/main/java/com/hedera/mirror/monitor/config/MonitorConfiguration.java
@@ -84,7 +84,6 @@ Disposable publish(PublishMetrics publishMetrics) {
                 .retry()
                 .name("generate")
                 .metrics()
-                .subscribeOn(Schedulers.single())
                 .parallel(publishProperties.getClients())
                 .runOn(Schedulers.newParallel("publisher", publishProperties.getClients()))
                 .map(transactionPublisher::publish)
@@ -98,6 +97,7 @@ Disposable publish(PublishMetrics publishMetrics) {
                 .onErrorContinue((t, r) -> log.error("Unexpected error during publish flow: ", t))
                 .doFinally(s -> log.warn("Stopped after {} signal", s))
                 .doOnSubscribe(s -> log.info("Starting publisher flow"))
+                .subscribeOn(Schedulers.single())
                 .subscribe(publishMetrics::onSuccess);
     }
 
@@ -117,6 +117,7 @@ Disposable subscribe(SubscribeMetrics subscribeMetrics) {
                 .onErrorContinue((t, r) -> log.error("Unexpected error during subscribe: ", t))
                 .doFinally(s -> log.warn("Stopped after {} signal", s))
                 .doOnSubscribe(s -> log.info("Starting subscribe flow"))
+                .subscribeOn(Schedulers.parallel())
                 .subscribe(subscribeMetrics::onNext);
     }
 }
diff --git a/...-mirror-monitor/src/main/java/com/hedera/mirror/monitor/publish/TransactionPublisher.java b/...-mirror-monitor/src/main/java/com/hedera/mirror/monitor/publish/TransactionPublisher.java
@@ -56,8 +56,8 @@ public class TransactionPublisher implements AutoCloseable {
 
     private final MonitorProperties monitorProperties;
     private final PublishProperties publishProperties;
-    private final Flux<Client> clients = Flux.defer(this::getClients).cache();
     private final List<AccountId> nodeAccountIds = new CopyOnWriteArrayList<>();
+    private final Flux<Client> clients = Flux.defer(this::getClients).cache();
     private final SecureRandom secureRandom = new SecureRandom();
 
     @Override
@@ -186,8 +186,8 @@ private boolean validateNode(Client client, NodeProperties node) {
             new TransferTransaction()
                     .addHbarTransfer(nodeAccountId, hbar)
                     .addHbarTransfer(client.getOperatorAccountId(), hbar.negated())
-                    .execute(client, Duration.ofSeconds(10L))
-                    .getReceipt(client, Duration.ofSeconds(10L));
+                    .execute(client, Duration.ofSeconds(30L))
+                    .getReceipt(client, Duration.ofSeconds(30L));
             log.info("Validated node: {}", node);
             valid = true;
         } catch (TimeoutException e) {