Netty metrics (#745)

gradle/libs.versions.toml

@@ -65,7 +65,7 @@ kotlin-reflect = { module = "org.jetbrains.kotlin:kotlin-reflect", version.ref =
 
 lettuce-core = "io.lettuce:lettuce-core:5.2.2.RELEASE"
 
-micrometer-registry-prometheus = "io.micrometer:micrometer-registry-prometheus:1.5.6"
+micrometer-registry-prometheus = "io.micrometer:micrometer-registry-prometheus:1.10.0"
 
 mockserver-netty = "org.mock-server:mockserver-netty:5.11.2"
 

src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfig.kt

@@ -18,21 +18,32 @@ package io.emeraldpay.dshackle.config
 class MonitoringConfig(
     val enabled: Boolean,
     val prometheus: PrometheusConfig,
+    val nettyMetricsConfig: NettyMetricsConfig,
 ) {
 
     companion object {
         fun default(): MonitoringConfig {
-            return MonitoringConfig(true, PrometheusConfig.default())
+            return MonitoringConfig(true, PrometheusConfig.default(), NettyMetricsConfig.default())
         }
 
         fun disabled(): MonitoringConfig {
-            return MonitoringConfig(false, PrometheusConfig.disabled())
+            return MonitoringConfig(false, PrometheusConfig.disabled(), NettyMetricsConfig.default())
         }
     }
 
     var enableJvm: Boolean = true
     var enableExtended: Boolean = false
 
+    data class NettyMetricsConfig(
+        val enabled: Boolean,
+    ) {
+        companion object {
+            fun default(): NettyMetricsConfig {
+                return NettyMetricsConfig(false)
+            }
+        }
+    }
+
     data class PrometheusConfig(
         val enabled: Boolean,
         val path: String,

src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfigReader.kt

@@ -32,13 +32,22 @@ class MonitoringConfigReader : YamlConfigReader<MonitoringConfig>() {
             return MonitoringConfig.disabled()
         }
         val prometheus = readPrometheus(getMapping(input, "prometheus"))
-        return MonitoringConfig(enabled, prometheus).also { conf ->
+        val netty = readNettyCfg(getMapping(input, "netty-metrics"))
+        return MonitoringConfig(enabled, prometheus, netty).also { conf ->
             getValueAsBool(input, "JVM")?.let { conf.enableJvm = it }
             getValueAsBool(input, "jvm")?.let { conf.enableJvm = it }
             getValueAsBool(input, "extended")?.let { conf.enableExtended = it }
         }
     }
 
+    private fun readNettyCfg(input: MappingNode?): MonitoringConfig.NettyMetricsConfig {
+        if (input == null) {
+            return MonitoringConfig.NettyMetricsConfig.default()
+        }
+        val enabled = getValueAsBool(input, "enabled") ?: false
+        return MonitoringConfig.NettyMetricsConfig(enabled)
+    }
+
     private fun readPrometheus(input: MappingNode?): MonitoringConfig.PrometheusConfig {
         if (input == null) {
             return MonitoringConfig.PrometheusConfig.default()

src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt

@@ -86,7 +86,7 @@ open class SchedulersConfig {
             size * threadsMultiplier,
             60L,
             TimeUnit.SECONDS,
-            LinkedBlockingQueue(1000),
+            LinkedBlockingQueue(10000),
             CustomizableThreadFactory("$name-"),
         )
 

src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.startup.configure
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.MonitoringConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.BasicHttpFactory
 import io.emeraldpay.dshackle.upstream.BlockValidator
@@ -24,6 +25,7 @@ open class GenericConnectorFactoryCreator(
     private val headScheduler: Scheduler,
     private val wsScheduler: Scheduler,
     private val headLivenessScheduler: Scheduler,
+    private val monitoringCfg: MonitoringConfig,
 ) : ConnectorFactoryCreator {
     protected val log = LoggerFactory.getLogger(this::class.java)
 
@@ -66,7 +68,14 @@ open class GenericConnectorFactoryCreator(
                 }
             }
             urls?.add(endpoint.url)
-            BasicHttpFactory(endpoint.url.toString(), endpoint.maxConnections, endpoint.queueSize, conn.basicAuth, tls)
+            BasicHttpFactory(
+                endpoint.url.toString(),
+                endpoint.maxConnections,
+                endpoint.queueSize,
+                conn.basicAuth,
+                tls,
+                monitoringCfg.nettyMetricsConfig.enabled,
+            )
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/startup/configure/RestConnectorFactoryCreator.kt

@@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.MonitoringConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.TonCompoundHttpFactory
@@ -20,12 +21,14 @@ class RestConnectorFactoryCreator(
     fileResolver: FileResolver,
     private val headScheduler: Scheduler,
     private val headLivenessScheduler: Scheduler,
+    monitoringCfg: MonitoringConfig,
 ) : GenericConnectorFactoryCreator(
     fileResolver,
     Schedulers.single(),
     headScheduler,
     Schedulers.single(),
     headLivenessScheduler,
+    monitoringCfg,
 ) {
     override fun createConnectorFactory(
         id: String,

src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt

@@ -17,6 +17,7 @@ class BasicHttpFactory(
     private val queueSize: Int,
     private val basicAuth: AuthConfig.ClientBasicAuth?,
     private val tls: ByteArray?,
+    private val nettyMetricsEnabled: Boolean,
 ) : HttpFactory {
     private val log = LoggerFactory.getLogger(this::class.java)
 
@@ -39,6 +40,7 @@ class BasicHttpFactory(
                 .description("Number of failures of HTTP JSON RPC requests")
                 .tags(metricsTags)
                 .register(Metrics.globalRegistry),
+            nettyMetricsEnabled,
         )
 
         if (chain.type.apiType == ApiType.REST) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt

@@ -38,12 +38,21 @@ abstract class HttpReader(
             .maxConnections(maxConnections)
             .pendingAcquireMaxCount(queueSize)
             .pendingAcquireTimeout(Duration.ofSeconds(10))
+            .apply {
+                if (metrics?.nettyMetricsEnabled ?: false) {
+                    metrics(true)
+                }
+            }
             .build()
 
         var build = HttpClient.create(connectionProvider)
             .compress(true)
             .resolver(DefaultAddressResolverGroup.INSTANCE)
 
+        if (metrics?.nettyMetricsEnabled ?: false) {
+            build = build.metrics(true) { s -> s }
+        }
+
         build = build.headers { h ->
             h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestMetrics.kt

@@ -21,4 +21,5 @@ import io.micrometer.core.instrument.Timer
 class RequestMetrics(
     val timer: Timer,
     val fails: Counter,
+    val nettyMetricsEnabled: Boolean,
 )

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt

@@ -40,6 +40,7 @@ open class WsConnectionFactory(
                 .description("Number of failures of WebSocket JSON RPC requests")
                 .tags(metricsTags)
                 .register(Metrics.globalRegistry),
+            false,
         )
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt

@@ -295,6 +295,7 @@ class GrpcUpstreams(
                 .description("Number of failures of Dshackle/gRPC requests")
                 .tags(metricsTags)
                 .register(Metrics.globalRegistry),
+            false,
         )
     }
 

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy

@@ -37,7 +37,8 @@ class JsonRpcHttpReaderSpec extends Specification {
     int port = 19332
     RequestMetrics metrics = new RequestMetrics(
             Timer.builder("test1").register(TestingCommons.meterRegistry),
-            Counter.builder("test2").register(TestingCommons.meterRegistry)
+            Counter.builder("test2").register(TestingCommons.meterRegistry),
+            false,
     )
 
     def setup() {

src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactoryCreatorTest.kt

@@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.ChainsConfigReader
+import io.emeraldpay.dshackle.config.MonitoringConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.foundation.ChainOptionsReader
 import io.emeraldpay.dshackle.startup.configure.GenericConnectorFactoryCreator
@@ -37,6 +38,7 @@ class GenericConnectorFactoryCreatorTest {
             immediate(),
             immediate(),
             immediate(),
+            MonitoringConfig.default(),
         )
         var args: List<*>? = null