solution: base code for prometheus monitoring

build.gradle

@@ -87,7 +87,7 @@ dependencies {
     implementation 'io.projectreactor.addons:reactor-extra:3.3.3.RELEASE'
     implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE'
     implementation "com.salesforce.servicelibs:reactor-grpc-stub:$reactiveGrpcVersion"
-
+    implementation 'io.micrometer:micrometer-registry-prometheus:1.5.6'
     implementation 'io.lettuce:lettuce-core:5.2.2.RELEASE'
 
     implementation "io.infinitape:etherjar-domain:$etherjarVersion"

src/main/kotlin/io/emeraldpay/dshackle/Config.kt

@@ -105,4 +105,9 @@ open class Config(
         return mainConfig.tokens ?: TokensConfig(emptyList())
     }
 
+    @Bean
+    open fun monitoringConfig(@Autowired mainConfig: MainConfig): MonitoringConfig {
+        return mainConfig.monitoring
+    }
+
 }

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

@@ -23,5 +23,6 @@ class MainConfig {
     var proxy: ProxyConfig? = null
     var upstreams: UpstreamsConfig? = null
     var tokens: TokensConfig? = null
+    var monitoring: MonitoringConfig = MonitoringConfig.default()
 
 }

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

@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * 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.
+ */
+package io.emeraldpay.dshackle.config
+
+class MonitoringConfig(
+        val enabled: Boolean,
+        val prometheus: PrometheusConfig
+) {
+
+    companion object {
+        fun default(): MonitoringConfig {
+            return MonitoringConfig(true, PrometheusConfig.default())
+        }
+    }
+
+
+    data class PrometheusConfig(
+            val enabled: Boolean,
+            val path: String,
+            val port: Int
+    ) {
+        companion object {
+            fun default(): PrometheusConfig {
+                return PrometheusConfig(true, "/prometheus", 8081)
+            }
+        }
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt

@@ -0,0 +1,74 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * 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.
+ */
+package io.emeraldpay.dshackle.monitoring
+
+import com.sun.net.httpserver.HttpServer
+import io.emeraldpay.dshackle.config.MainConfig
+import io.emeraldpay.dshackle.config.MonitoringConfig
+import io.micrometer.core.instrument.Meter
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.config.MeterFilter
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Service
+import io.micrometer.prometheus.PrometheusConfig
+
+import io.micrometer.prometheus.PrometheusMeterRegistry
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Value
+import java.io.IOException
+import java.io.OutputStream
+import java.net.InetSocketAddress
+import javax.annotation.PostConstruct
+
+
+@Service
+class MonitoringSetup(
+        @Autowired private val monitoringConfig: MonitoringConfig
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(MonitoringSetup::class.java)
+    }
+
+    @PostConstruct
+    fun setup() {
+        val prometheusRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
+        Metrics.globalRegistry.add(prometheusRegistry)
+        Metrics.globalRegistry.config().meterFilter(object: MeterFilter {
+            override fun map(id: Meter.Id): Meter.Id {
+                return id.withName("dshackle." + id.name)
+            }
+        })
+
+        if (monitoringConfig.prometheus.enabled) {
+            // use standard JVM server with a single thread blocking processing
+            // prometheus is a single thread periodic call, no reason to setup anything complex
+            try {
+                val server = HttpServer.create(InetSocketAddress(monitoringConfig.prometheus.port), 0);
+                server.createContext(monitoringConfig.prometheus.path) { httpExchange ->
+                    val response = prometheusRegistry.scrape()
+                    httpExchange.sendResponseHeaders(200, response.toByteArray().size.toLong());
+                    httpExchange.responseBody.use { os ->
+                        os.write(response.toByteArray())
+                    }
+                }
+                Thread(server::start).start();
+            } catch (e: IOException) {
+                log.error("Failed to start Prometheus Server", e)
+            }
+        }
+    }
+}

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

@@ -24,6 +24,8 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.Tag
 import org.apache.commons.collections4.Factory
 import org.apache.commons.collections4.FunctorException
 import org.slf4j.LoggerFactory
@@ -50,6 +52,7 @@ abstract class Multistream(
 
     companion object {
         private val log = LoggerFactory.getLogger(Multistream::class.java)
+        private const val metrics = "upstreams"
     }
 
     private var cacheSubscription: Disposable? = null
@@ -63,6 +66,24 @@ abstract class Multistream(
     private var subscription: Disposable? = null
     private var capabilities: Set<Capability> = emptySet()
 
+    init {
+        UpstreamAvailability.values().forEach { status ->
+            Metrics.gauge("$metrics.availability",
+                    listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.toLowerCase()))
+                    , this) {
+                upstreams.count { it.getStatus() == status }.toDouble()
+            }
+        }
+
+        upstreams.forEach { up ->
+            Metrics.gauge("$metrics.lag",
+                    listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", up.getId()))
+                    , this) {
+                up.getLag().toDouble()
+            }
+        }
+    }
+
     open fun init() {
         onUpstreamsUpdated()
     }