solution: extended monitoring option; monitoring for upstream selection filter

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

@@ -31,6 +31,8 @@ class Global {
 
     companion object {
 
+        var metricsExtended = false
+
         @JvmStatic
         val objectMapper: ObjectMapper = createObjectMapper()
 

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

@@ -30,6 +30,7 @@ class MonitoringConfig(
     }
 
     var enableJvm: Boolean = true
+    var enableExtended: Boolean = false
 
     data class PrometheusConfig(
             val enabled: Boolean,

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

@@ -44,7 +44,9 @@ class MonitoringConfigReader: YamlConfigReader(), ConfigReader<MonitoringConfig>
         }
         val prometheus = readPrometheus(getMapping(input, "prometheus"))
         return MonitoringConfig(enabled, prometheus).also { conf ->
-            getValueAsBool(input, "enableJVM")?.let { conf.enableJvm = it }
+            getValueAsBool(input, "JVM")?.let { conf.enableJvm = it }
+            getValueAsBool(input, "jvm")?.let { conf.enableJvm = it }
+            getValueAsBool(input, "extended")?.let { conf.enableExtended = it }
         }
     }
 

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

@@ -16,7 +16,7 @@
 package io.emeraldpay.dshackle.monitoring
 
 import com.sun.net.httpserver.HttpServer
-import io.emeraldpay.dshackle.config.MainConfig
+import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.MonitoringConfig
 import io.micrometer.core.instrument.Meter
 import io.micrometer.core.instrument.Metrics
@@ -32,9 +32,7 @@ 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
 
@@ -69,6 +67,9 @@ class MonitoringSetup(
             ProcessorMetrics().bindTo(Metrics.globalRegistry)
             JvmThreadMetrics().bindTo(Metrics.globalRegistry)
         }
+        if (monitoringConfig.enableExtended) {
+            Global.metricsExtended = true
+        }
 
         if (monitoringConfig.prometheus.enabled) {
             // use standard JVM server with a single thread blocking processing

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

@@ -16,11 +16,20 @@
  */
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.grpc.Chain
+import io.micrometer.core.instrument.DistributionSummary
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.Tag
 import org.reactivestreams.Subscriber
 import reactor.core.publisher.EmitterProcessor
 import reactor.core.publisher.Flux
 import java.time.Duration
+import java.util.*
+import java.util.concurrent.locks.Lock
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
 import kotlin.math.max
 import kotlin.math.min
 import kotlin.math.pow
@@ -28,9 +37,10 @@ import kotlin.math.roundToLong
 import kotlin.random.Random
 
 class FilteredApis(
+        val chain: Chain,
         private val allUpstreams: List<Upstream>,
         private val matcher: Selector.Matcher,
-        pos: Int,
+        private val pos: Int,
         /**
          * Limit of retries
          */
@@ -42,6 +52,8 @@ class FilteredApis(
         private const val DEFAULT_DELAY_STEP = 100
         private const val MAX_WAIT_MILLIS = 5000L
 
+        private const val metricsCode = "select"
+
         @JvmStatic
         fun <T> startFrom(upstreams: List<T>, pos: Int): List<T> {
             return if (upstreams.size <= 1 || pos == 0) {
@@ -51,14 +63,19 @@ class FilteredApis(
                 upstreams.subList(safePosition, upstreams.size) + upstreams.subList(0, safePosition)
             }
         }
+
+        private val metrics = EnumMap<Chain, Monitoring>(Chain::class.java)
+        private val metricsSetup: Lock = ReentrantLock()
     }
 
-    constructor(allUpstreams: List<Upstream>,
+    constructor(chain: Chain,
+                allUpstreams: List<Upstream>,
                 matcher: Selector.Matcher,
-                pos: Int) : this(allUpstreams, matcher, pos, 10, 7)
+                pos: Int) : this(chain, allUpstreams, matcher, pos, 10, 7)
 
-    constructor(allUpstreams: List<Upstream>,
+    constructor(chain: Chain,
-                matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10)
+                allUpstreams: List<Upstream>,
+                matcher: Selector.Matcher) : this(chain, allUpstreams, matcher, 0, 10, 10)
 
     private val delay: Int
     private val standardUpstreams: List<Upstream>
@@ -87,6 +104,30 @@ class FilteredApis(
                 .plus(standardUpstreams)
                 .plus(fallbackUpstreams)
 
+        if (Global.metricsExtended) {
+            getMetrics(chain).let { monitoring ->
+                monitoring.countStd.record(standardUpstreams.size.toDouble())
+                monitoring.countFallback.record(fallbackUpstreams.size.toDouble())
+            }
+        }
+    }
+
+    private fun getMetrics(chain: Chain): Monitoring {
+        val existing = metrics[chain]
+        return if (existing == null) {
+            metricsSetup.withLock {
+                val existingDoubleCheck = metrics[chain]
+                if (existingDoubleCheck != null) {
+                    existingDoubleCheck
+                } else {
+                    val created = Monitoring(chain)
+                    metrics[chain] = created
+                    created
+                }
+            }
+        } else {
+            existing
+        }
     }
 
     fun waitDuration(rawn: Long): Duration {
@@ -110,9 +151,16 @@ class FilteredApis(
                     .delaySubscription(waitDuration(r + 1))
         }.let { Flux.concat(it) }
 
-        Flux.concat(first, retries)
+        var result = Flux.concat(first, retries)
-                .filter(Upstream::isAvailable)
+
-                .filter(matcher::matches)
+        if (Global.metricsExtended) {
+            var count = 0
+            result = result
+                    .doOnNext { count++ }
+                    .doFinally { metrics[chain]?.tried?.record(count.toDouble()) }
+        }
+
+        result.filter { up -> up.isAvailable() && matcher.matches(up) }
                 .zipWith(control)
                 .map { it.t1 }
                 .subscribe(subscriber)
@@ -132,4 +180,19 @@ class FilteredApis(
     override fun toString(): String {
         return "Filter API: ${allUpstreams.size} upstreams with $matcher"
     }
+
+    class Monitoring(chain: Chain) {
+        val countStd: DistributionSummary = DistributionSummary.builder("$metricsCode.exist")
+                .description("Count of available upstreams to select")
+                .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "std")))
+                .register(Metrics.globalRegistry)
+        val countFallback: DistributionSummary = DistributionSummary.builder("$metricsCode.exist")
+                .description("Count of available fallback upstreams to select")
+                .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "fallback")))
+                .register(Metrics.globalRegistry)
+        val tried: DistributionSummary = DistributionSummary.builder("$metricsCode.tried")
+                .description("How many upstreams were checked")
+                .tags(listOf(Tag.of("chain", chain.chainCode)))
+                .register(Metrics.globalRegistry)
+    }
 }

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

@@ -69,18 +69,25 @@ abstract class Multistream(
     init {
         UpstreamAvailability.values().forEach { status ->
             Metrics.gauge("$metrics.availability",
-                    listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.toLowerCase()))
+                    listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.toLowerCase())), this) {
-                    , this) {
                 upstreams.count { it.getStatus() == status }.toDouble()
             }
         }
 
+        Metrics.gauge("$metrics.connected",
+                listOf(Tag.of("chain", chain.chainCode)), this) {
+            upstreams.size.toDouble()
+        }
+
         upstreams.forEach { up ->
-            Metrics.gauge("$metrics.lag",
+            monitorUpstream(up)
-                    listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", up.getId()))
+        }
-                    , this) {
+    }
-                up.getLag().toDouble()
+
-            }
+    private fun monitorUpstream(upstream: Upstream) {
+        Metrics.gauge("$metrics.lag",
+                listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", upstream.getId())), upstream) {
+            it.getLag().toDouble()
         }
     }
 
@@ -102,6 +109,7 @@ abstract class Multistream(
         upstreams.add(upstream)
         onUpstreamsUpdated()
         setHead(updateHead())
+        monitorUpstream(upstream)
     }
 
     fun removeUpstream(id: String) {
@@ -119,7 +127,7 @@ abstract class Multistream(
         if (seq >= Int.MAX_VALUE / 2) {
             seq = 0
         }
-        return FilteredApis(upstreams, matcher, i)
+        return FilteredApis(chain, upstreams, matcher, i)
     }
 
     /**