Fix GRPC high CPU usage and long running queries (#524)

* Fix GRPC high CPU usage and long running queries

Signed-off-by: Steven Sheehy <steven.sheehy@hedera.com>

* Set r2dbc timeout values & bump scheduler pool

Signed-off-by: Steven Sheehy <steven.sheehy@hedera.com>

* More fixes and bump versions

Signed-off-by: Steven Sheehy <steven.sheehy@hedera.com>

* Rollback some changes

Signed-off-by: Steven Sheehy <steven.sheehy@hedera.com>

hedera-mirror-coverage/pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <dependencies>

hedera-mirror-datagenerator/pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <dependencies>

hedera-mirror-grpc/pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <dependencies>
@@ -37,10 +37,13 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-elastic</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.r2dbc</groupId>
+            <artifactId>r2dbc-pool</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.r2dbc</groupId>
             <artifactId>r2dbc-postgresql</artifactId>
-            <version>0.8.1.BUILD-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>javax.inject</groupId>
@@ -217,12 +220,4 @@
             </plugin>
         </plugins>
     </build>
-
-    <repositories>
-        <repository>
-            <id>spring-libs-snapshot</id>
-            <name>Spring Snapshot Repository</name>
-            <url>https://repo.spring.io/libs-snapshot</url>
-        </repository>
-    </repositories>
 </project>

hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/GrpcProperties.java

@@ -0,0 +1,40 @@
+package com.hedera.mirror.grpc;
+
+/*-
+ * ‌
+ * Hedera Mirror Node
+ * ​
+ * Copyright (C) 2019 Hedera Hashgraph, LLC
+ * ​
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ‍
+ */
+
+import java.util.HashMap;
+import java.util.Map;
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+@Data
+@Validated
+@ConfigurationProperties("hedera.mirror.grpc")
+public class GrpcProperties {
+    @Min(32)
+    private int maxPageSize = 1000;
+
+    @NotNull
+    private Map<String, String> connectionOptions = new HashMap<>();
+}

hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/MirrorGrpcApplication.java

@@ -22,11 +22,12 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.data.r2dbc.R2dbcTransactionManagerAutoConfiguration;
 import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
 import reactor.core.scheduler.Schedulers;
 
 @ConfigurationPropertiesScan
-@SpringBootApplication
+@SpringBootApplication(exclude = R2dbcTransactionManagerAutoConfiguration.class)
 public class MirrorGrpcApplication {
 
     public static void main(String[] args) {

hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/config/DatabaseConfiguration.java

@@ -20,10 +20,12 @@
  * ‍
  */
 
+import io.r2dbc.postgresql.PostgresqlConnectionFactoryProvider;
 import io.r2dbc.spi.ConnectionFactory;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryOptionsBuilderCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.convert.converter.Converter;
@@ -32,6 +34,8 @@
 import org.springframework.data.r2dbc.dialect.DialectResolver;
 import org.springframework.data.r2dbc.dialect.R2dbcDialect;
 
+import com.hedera.mirror.grpc.GrpcProperties;
+
 @Configuration
 public class DatabaseConfiguration {
 
@@ -45,4 +49,10 @@ R2dbcCustomConversions customConversions(ConnectionFactory connectionFactory,
         return new R2dbcCustomConversions(
                 CustomConversions.StoreConversions.of(dialect.getSimpleTypeHolder(), converters), customConverters);
     }
+
+    @Bean
+    ConnectionFactoryOptionsBuilderCustomizer connectionFactoryCustomizer(GrpcProperties grpcProperties) {
+        return builder -> builder
+                .option(PostgresqlConnectionFactoryProvider.OPTIONS, grpcProperties.getConnectionOptions());
+    }
 }

hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/listener/PollingTopicListener.java

@@ -70,7 +70,6 @@ private Flux<TopicMessage> poll(PollingContext context) {
                 .topicNum(filter.getTopicNum())
                 .build();
 
-        log.debug("Polling for messages: {}", newFilter);
         return topicMessageRepository.findByFilter(newFilter)
                 .doOnSubscribe(s -> context.setRunning(true))
                 .doOnCancel(() -> context.setRunning(false))

hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/repository/TopicMessageRepositoryCustomImpl.java

@@ -23,11 +23,14 @@
 import javax.inject.Named;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.core.DatabaseClient;
 import org.springframework.data.r2dbc.query.Criteria;
 import reactor.core.publisher.Flux;
 
+import com.hedera.mirror.grpc.GrpcProperties;
 import com.hedera.mirror.grpc.converter.InstantToLongConverter;
 import com.hedera.mirror.grpc.domain.TopicMessage;
 import com.hedera.mirror.grpc.domain.TopicMessageFilter;
@@ -38,6 +41,7 @@
 public class TopicMessageRepositoryCustomImpl implements TopicMessageRepositoryCustom {
 
     private final DatabaseClient databaseClient;
+    private final GrpcProperties grpcProperties;
     private final InstantToLongConverter instantToLongConverter;
 
     @Override
@@ -54,15 +58,22 @@ public Flux<TopicMessage> findByFilter(TopicMessageFilter filter) {
                     .lessThan(instantToLongConverter.convert(filter.getEndTime()));
         }
 
+        int limit = filter.hasLimit() ? (int) filter.getLimit() : Integer.MAX_VALUE;
+        int pageSize = Math.min(limit, grpcProperties.getMaxPageSize());
+        Pageable pageable = PageRequest.of(0, pageSize);
+
         return databaseClient.select()
                 .from(TopicMessage.class)
                 .matching(whereClause)
                 .orderBy(Sort.by("consensus_timestamp"))
+                .page(pageable)
                 .fetch()
                 .all()
-                .as(t -> filter.hasLimit() ? t.limitRequest(filter.getLimit()) : t)
                 .name("findByFilter")
                 .metrics()
-                .doOnSubscribe(s -> log.debug("Executing query: {}", filter));
+                .doOnSubscribe(s -> log.debug("Executing query: {}", filter))
+                .doOnCancel(() -> log.debug("[{}] Cancelled query", filter.getSubscriberId()))
+                .doOnComplete(() -> log.debug("[{}] Completed query", filter.getSubscriberId()))
+                .doOnNext(t -> log.trace("[{}] Next message: {}", filter.getSubscriberId(), t));
     }
 }

hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/service/TopicMessageServiceImpl.java

@@ -22,6 +22,7 @@
 
 import com.google.common.base.Stopwatch;
 import java.time.Instant;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import javax.inject.Named;
 import lombok.Data;
@@ -157,8 +158,10 @@ boolean isNext(TopicMessage topicMessage) {
         }
 
         void onComplete() {
-            log.info("[{}] Topic {} {} complete with {} messages in {}", filter
-                    .getSubscriberId(), topicId, mode, count, stopwatch);
+            var elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
+            var rate = elapsed > 0 ? (int) (1000.0 * count.get() / elapsed) : 0;
+            log.info("[{}] Topic {} {} complete with {} messages in {} ({}/s)", filter
+                    .getSubscriberId(), topicId, mode, count, stopwatch, rate);
             mode = mode.next();
         }
     }

hedera-mirror-grpc/src/main/resources/application.yml

@@ -7,6 +7,11 @@ hedera:
         password: mirror_grpc_pass
         port: 5432
         username: mirror_grpc
+      connectionOptions:
+        default_transaction_read_only: "on"
+        idle_in_transaction_session_timeout: "30s"
+        lock_timeout: "30s"
+        statement_timeout: "120s"
       port: 5600
 grpc:
   server:

hedera-mirror-grpc/src/test/java/com/hedera/mirror/grpc/GrpcIntegrationTest.java

@@ -20,13 +20,17 @@
  * ‍
  */
 
+import io.r2dbc.postgresql.PostgresqlConnectionFactoryProvider;
+import java.util.Map;
 import javax.annotation.PreDestroy;
 import lombok.extern.log4j.Log4j2;
+import org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryOptionsBuilderCustomizer;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.boot.test.util.TestPropertyValues;
 import org.springframework.context.ApplicationContextInitializer;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.ContextConfiguration;
 import org.testcontainers.containers.PostgreSQLContainer;
 
@@ -52,6 +56,12 @@ static class TestDatabaseConfiguration implements ApplicationContextInitializer<
             System.setProperty("testcontainers.windowsprovider.timeout", "1");
         }
 
+        @Bean
+        ConnectionFactoryOptionsBuilderCustomizer testConnectionFactoryCustomizer() {
+            Map<String, String> options = Map.of("default_transaction_read_only", "off");
+            return builder -> builder.option(PostgresqlConnectionFactoryProvider.OPTIONS, options);
+        }
+
         @Override
         public void initialize(ConfigurableApplicationContext applicationContext) {
             try {

hedera-mirror-importer/pom.xml

@@ -13,7 +13,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <properties>

hedera-mirror-protobuf/pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <properties>

hedera-mirror-rest/package.json

@@ -1,6 +1,6 @@
 {
   "name": "hedera-mirror-rest",
-  "version": "0.5.3",
+  "version": "0.5.4",
   "description": "Hedera Mirror Node REST API",
   "main": "server.js",
   "scripts": {

hedera-mirror-rest/pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.hedera</groupId>
         <artifactId>hedera-mirror-node</artifactId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <build>

hedera-mirror-test/pom.xml

@@ -12,7 +12,7 @@
     <parent>
         <artifactId>hedera-mirror-node</artifactId>
         <groupId>com.hedera</groupId>
-        <version>0.5.3</version>
+        <version>0.5.4</version>
     </parent>
 
     <properties>

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>com.hedera</groupId>
     <artifactId>hedera-mirror-node</artifactId>
-    <version>0.5.3</version>
+    <version>0.5.4</version>
     <description>Hedera Mirror Node mirrors data from Hedera nodes and serves it via an API</description>
     <inceptionYear>2019</inceptionYear>
     <modelVersion>4.0.0</modelVersion>